aiolabs/libra
9
0
Fork 1
Code Issues 36 Pull requests 7 Projects Releases 6 Packages Wiki Activity Actions

fix(payments): record-payment fails closed and shares the claim gate

Browse source 
Two fixes to POST /api/v1/record-payment:

- The Fava duplicate check caught every exception and proceeded to
  write, so a transient Fava blip produced double entries. It now
  fails closed: transport errors return 503 and the client retries.
  While here: the check queried {base_url}/api/journal, but base_url
  already ends in /api — the doubled path 404'd, meaning the
  duplicate check has silently never run.

- The endpoint now goes through the same processed_payments claim
  gate as the background invoice listener, so the webhook+poller pair
  can't both record the same payment_hash: a 'done' claim replays as
  "already recorded", an in-flight claim returns 409.

Addresses CODE-REVIEW-2026-06 finding #10.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Padreug 2026-07-12 12:41:42 +02:00
commit 44e10caac7
2 changed files with 386 additions and 65 deletions
Download patch file Download diff file Expand all files Collapse all files

281
tests/test_payment_idempotency.py Normal file
View file

@ -0,0 +1,281 @@
"""Lightning payment idempotency — the `processed_payments` claim gate.
The background invoice listener (`tasks.on_invoice_paid`) and the
client-driven `POST /record-payment` endpoint can both fire for the
same `payment_hash` (queue redelivery after restart, webhook + poller).
The Fava-side duplicate checks are read-then-write races; the local
`processed_payments` primary key makes exactly one claimant win.
These tests bypass invoice generation (blocked by libra/issues/40) by
delivering synthetic paid `Payment` objects straight to
`on_invoice_paid` and by inserting paid payment rows via the LNbits
core crud for the endpoint tests.
"""
import asyncio
import importlib
from uuid import uuid4
import pytest
from lnbits.core.crud.payments import create_payment
from lnbits.core.models.payments import CreatePayment, Payment, PaymentState
from .helpers import list_user_entries, post_receivable
pytestmark = pytest.mark.anyio
def _module(name: str):
"""Import a libra submodule under whichever path the active LNbits layout
uses (default `lnbits.extensions.libra` or bare `libra`)."""
for prefix in ("lnbits.extensions.libra", "libra"):
try:
return importlib.import_module(f"{prefix}.{name}")
except ModuleNotFoundError:
continue
raise ModuleNotFoundError(f"libra.{name}: tried both import paths")
tasks = _module("tasks")
libra_crud = _module("crud")
def _paid_payment(
wallet_id: str,
user_id: str,
*,
fiat_amount: str = "100.00",
fiat_currency: str = "EUR",
sats: int = 100_000,
) -> Payment:
payment_hash = uuid4().hex + uuid4().hex[:32]
return Payment(
checking_id=payment_hash,
payment_hash=payment_hash,
wallet_id=wallet_id,
amount=sats * 1000,
fee=0,
bolt11="lnbcfake",
status=PaymentState.SUCCESS,
extra={
"tag": "libra",
"user_id": user_id,
"fiat_currency": fiat_currency,
"fiat_amount": fiat_amount,
},
)
async def _setup_receivable(
client, super_user_headers, configured_user, standard_accounts,
amount: str = "100.00",
):
user, wallet = configured_user
await post_receivable(
client,
super_user_headers=super_user_headers,
user_id=user.id,
amount=amount,
description=f"Idempotency setup {uuid4().hex[:6]}",
revenue_account=standard_accounts["revenue_rent"]["name"],
)
# Force a Fava reload before downstream balance reads (see #37).
await list_user_entries(client, wallet_inkey=wallet.inkey)
return user, wallet
async def _entries_with_link(client, wallet_inkey: str, link: str) -> list:
payload = await list_user_entries(client, wallet_inkey=wallet_inkey)
return [
e for e in payload["entries"] if link in (e.get("links") or [])
]
# ---------------------------------------------------------------------------
# on_invoice_paid — the background listener path
# ---------------------------------------------------------------------------
async def test_double_delivery_records_exactly_once(
client, super_user_headers, configured_user, standard_accounts
):
"""Same payment delivered twice (queue redelivery) → one ledger entry."""
user, wallet = await _setup_receivable(
client, super_user_headers, configured_user, standard_accounts
)
payment = _paid_payment(wallet.id, user.id)
await tasks.on_invoice_paid(payment)
await tasks.on_invoice_paid(payment)
link = f"ln-{payment.payment_hash[:16]}"
assert len(await _entries_with_link(client, wallet.inkey, link)) == 1
row = await libra_crud.get_processed_payment(payment.payment_hash)
assert row is not None and row["status"] == "done"
async def test_failed_recording_releases_claim_and_retry_succeeds(
client, super_user_headers, configured_user, standard_accounts, monkeypatch
):
"""A Fava failure mid-write must not permanently block the payment."""
user, wallet = await _setup_receivable(
client, super_user_headers, configured_user, standard_accounts
)
payment = _paid_payment(wallet.id, user.id)
fava_client = _module("fava_client")
fava = fava_client.get_fava_client()
async def _boom(*args, **kwargs):
raise RuntimeError("fava down")
monkeypatch.setattr(fava, "add_entry_idempotent", _boom)
with pytest.raises(RuntimeError):
await tasks.on_invoice_paid(payment)
monkeypatch.undo()
# Claim released → nothing recorded, retry allowed.
assert await libra_crud.get_processed_payment(payment.payment_hash) is None
await tasks.on_invoice_paid(payment)
row = await libra_crud.get_processed_payment(payment.payment_hash)
assert row is not None and row["status"] == "done"
link = f"ln-{payment.payment_hash[:16]}"
assert len(await _entries_with_link(client, wallet.inkey, link)) == 1
async def test_listener_survives_poison_payment_and_clears_stale_claims(
client, super_user_headers, configured_user, standard_accounts, monkeypatch
):
"""One bad payment must not kill the listener; stale 'processing'
claims from a previous process life are cleared at startup."""
user, wallet = await _setup_receivable(
client, super_user_headers, configured_user, standard_accounts
)
# A claim left behind by a "crashed" previous run.
stale_hash = uuid4().hex + uuid4().hex[:32]
assert await libra_crud.claim_payment(stale_hash)
captured: dict = {}
monkeypatch.setattr(
tasks,
"register_invoice_listener",
lambda queue, name: captured.update(queue=queue),
)
listener = asyncio.create_task(tasks.wait_for_paid_invoices())
try:
for _ in range(50):
if "queue" in captured:
break
await asyncio.sleep(0.05)
assert "queue" in captured, "listener never registered its queue"
poison = _paid_payment(wallet.id, user.id, fiat_amount="not-a-number")
good = _paid_payment(wallet.id, user.id)
captured["queue"].put_nowait(poison)
captured["queue"].put_nowait(good)
row = None
for _ in range(100):
row = await libra_crud.get_processed_payment(good.payment_hash)
if row and row["status"] == "done":
break
await asyncio.sleep(0.1)
assert row is not None and row["status"] == "done", (
"good payment was not recorded after the poison payment"
)
finally:
listener.cancel()
# Startup cleared the stale claim; the poison payment's claim was
# released on failure so redelivery could retry it.
assert await libra_crud.get_processed_payment(stale_hash) is None
assert await libra_crud.get_processed_payment(poison.payment_hash) is None
# ---------------------------------------------------------------------------
# POST /record-payment — the client-driven path
# ---------------------------------------------------------------------------
async def _insert_paid_payment_row(wallet_id: str, user_id: str) -> str:
payment_hash = uuid4().hex + uuid4().hex[:32]
await create_payment(
checking_id=payment_hash,
data=CreatePayment(
wallet_id=wallet_id,
payment_hash=payment_hash,
bolt11="lnbcfake",
amount_msat=100_000_000,
memo="idempotency test",
extra={
"tag": "libra",
"user_id": user_id,
"fiat_currency": "EUR",
"fiat_amount": "100.00",
},
),
status=PaymentState.SUCCESS,
)
return payment_hash
async def test_record_payment_conflicts_while_in_flight(
client, super_user_headers, configured_user, standard_accounts
):
user, wallet = await _setup_receivable(
client, super_user_headers, configured_user, standard_accounts
)
payment_hash = await _insert_paid_payment_row(wallet.id, user.id)
# Another claimant (e.g. the background listener) is mid-recording.
assert await libra_crud.claim_payment(payment_hash)
r = await client.post(
"/libra/api/v1/record-payment",
headers={"X-Api-Key": wallet.inkey},
json={"payment_hash": payment_hash},
)
assert r.status_code == 409, r.text
# Once that claimant finishes, a replay reports "already recorded"
# instead of writing a second entry.
await libra_crud.mark_payment_done(payment_hash, f"ln-{payment_hash[:16]}")
r = await client.post(
"/libra/api/v1/record-payment",
headers={"X-Api-Key": wallet.inkey},
json={"payment_hash": payment_hash},
)
assert r.status_code == 200, r.text
assert "already recorded" in r.json()["message"].lower()
async def test_record_payment_records_once_then_replays_safely(
client, super_user_headers, configured_user, standard_accounts
):
user, wallet = await _setup_receivable(
client, super_user_headers, configured_user, standard_accounts
)
payment_hash = await _insert_paid_payment_row(wallet.id, user.id)
r = await client.post(
"/libra/api/v1/record-payment",
headers={"X-Api-Key": wallet.inkey},
json={"payment_hash": payment_hash},
)
assert r.status_code == 200, r.text
assert r.json()["message"] == "Payment recorded successfully"
r = await client.post(
"/libra/api/v1/record-payment",
headers={"X-Api-Key": wallet.inkey},
json={"payment_hash": payment_hash},
)
assert r.status_code == 200, r.text
assert "already recorded" in r.json()["message"].lower()
link = f"ln-{payment_hash[:16]}"
assert len(await _entries_with_link(client, wallet.inkey, link)) == 1

176
views_api.py
View file

@ -1850,90 +1850,130 @@ async def api_record_payment(
try:
async with httpx.AsyncClient(timeout=5.0) as client:
# Get recent entries from Fava's journal endpoint
# Get recent entries from Fava's journal endpoint. base_url
# already ends in /api — the previous "/api/journal" path
# 404'd, so this duplicate check silently never ran.
response = await client.get(
f"{fava.base_url}/api/journal",
f"{fava.base_url}/journal",
params={"time": ""} # Get all entries
)
response.raise_for_status()
response_data = response.json()
entries = response_data.get('entries', [])
if response.status_code == 200:
response_data = response.json()
entries = response_data.get('entries', [])
# Check if any entry has our payment link
for entry in entries:
entry_links = entry.get('links', [])
if link_to_find in entry_links:
# Payment already recorded, return existing entry
balance_data = await fava.get_user_balance_bql(target_user_id)
return {
"journal_entry_id": f"fava-exists-{data.payment_hash[:16]}",
"new_balance": balance_data["balance"],
"message": "Payment already recorded",
}
except Exception as e:
# Check if any entry has our payment link
for entry in entries:
entry_links = entry.get('links', [])
if link_to_find in entry_links:
# Payment already recorded, return existing entry
balance_data = await fava.get_user_balance_bql(target_user_id)
return {
"journal_entry_id": f"fava-exists-{data.payment_hash[:16]}",
"new_balance": balance_data["balance"],
"message": "Payment already recorded",
}
except httpx.HTTPError as e:
# Fail CLOSED: if Fava can't confirm the payment isn't already
# recorded, refuse to write — proceeding on a transient blip is
# how double entries happen. The client can simply retry.
logger.warning(f"Could not check Fava for duplicate payment: {e}")
# Continue anyway - Fava/Beancount will catch duplicate if it exists
# Convert amount from millisatoshis to satoshis
amount_sats = payment.amount // 1000
# Extract fiat metadata from invoice (if present)
fiat_currency = None
fiat_amount = None
if payment.extra and isinstance(payment.extra, dict):
logger.info(f"Payment.extra contents: {payment.extra}")
fiat_currency = payment.extra.get("fiat_currency")
fiat_amount_str = payment.extra.get("fiat_amount")
if fiat_amount_str:
from decimal import Decimal
fiat_amount = Decimal(str(fiat_amount_str))
logger.info(f"Extracted fiat metadata - currency: {fiat_currency}, amount: {fiat_amount}")
# Get user's receivable account (what user owes)
user_receivable = await get_or_create_user_account(
target_user_id, AccountType.ASSET, "Accounts Receivable"
)
# Get lightning account
lightning_account = await get_account_by_name("Assets:Bitcoin:Lightning")
if not lightning_account:
raise HTTPException(
status_code=HTTPStatus.NOT_FOUND, detail="Lightning account not found"
status_code=HTTPStatus.SERVICE_UNAVAILABLE,
detail="Cannot verify payment duplicate status; try again shortly",
)
# Get unsettled receivable entries to link to this settlement
unsettled = await fava.get_unsettled_entries_bql(target_user_id, "receivable")
settled_links = [e["link"] for e in unsettled if e.get("link")]
# Format payment entry and submit to Fava
entry = format_payment_entry(
user_id=target_user_id,
payment_account=lightning_account.name,
payable_or_receivable_account=user_receivable.name,
amount_sats=amount_sats,
description=f"Lightning payment from user {target_user_id[:8]}",
entry_date=datetime.now().date(),
is_payable=False, # User paying libra (receivable settlement)
fiat_currency=fiat_currency,
fiat_amount=fiat_amount,
payment_hash=data.payment_hash,
reference=data.payment_hash,
settled_entry_links=settled_links
# Local idempotency gate shared with the background invoice listener
# (tasks.on_invoice_paid): exactly one claimant records a payment_hash.
from .crud import (
claim_payment,
get_processed_payment,
mark_payment_done,
release_payment_claim,
)
logger.info(f"Formatted payment entry: {entry}")
if not await claim_payment(data.payment_hash):
existing = await get_processed_payment(data.payment_hash)
if existing and existing["status"] == "done":
balance_data = await fava.get_user_balance_bql(target_user_id)
return {
"journal_entry_id": existing.get("entry_id")
or f"fava-exists-{data.payment_hash[:16]}",
"new_balance": balance_data["balance"],
"message": "Payment already recorded",
}
raise HTTPException(
status_code=HTTPStatus.CONFLICT,
detail="Payment is being recorded; check balance shortly",
)
# Submit to Fava
result = await fava.add_entry(entry)
logger.info(f"Payment entry submitted to Fava: {result.get('data', 'Unknown')}")
# Convert amount from millisatoshis to satoshis
try:
amount_sats = payment.amount // 1000
# Extract fiat metadata from invoice (if present)
fiat_currency = None
fiat_amount = None
if payment.extra and isinstance(payment.extra, dict):
logger.info(f"Payment.extra contents: {payment.extra}")
fiat_currency = payment.extra.get("fiat_currency")
fiat_amount_str = payment.extra.get("fiat_amount")
if fiat_amount_str:
from decimal import Decimal
fiat_amount = Decimal(str(fiat_amount_str))
logger.info(f"Extracted fiat metadata - currency: {fiat_currency}, amount: {fiat_amount}")
# Get user's receivable account (what user owes)
user_receivable = await get_or_create_user_account(
target_user_id, AccountType.ASSET, "Accounts Receivable"
)
# Get lightning account
lightning_account = await get_account_by_name("Assets:Bitcoin:Lightning")
if not lightning_account:
raise HTTPException(
status_code=HTTPStatus.NOT_FOUND, detail="Lightning account not found"
)
# Get unsettled receivable entries to link to this settlement
unsettled = await fava.get_unsettled_entries_bql(target_user_id, "receivable")
settled_links = [e["link"] for e in unsettled if e.get("link")]
# Format payment entry and submit to Fava
entry = format_payment_entry(
user_id=target_user_id,
payment_account=lightning_account.name,
payable_or_receivable_account=user_receivable.name,
amount_sats=amount_sats,
description=f"Lightning payment from user {target_user_id[:8]}",
entry_date=datetime.now().date(),
is_payable=False, # User paying libra (receivable settlement)
fiat_currency=fiat_currency,
fiat_amount=fiat_amount,
payment_hash=data.payment_hash,
reference=data.payment_hash,
settled_entry_links=settled_links
)
logger.info(f"Formatted payment entry: {entry}")
# Submit to Fava
result = await fava.add_entry(entry)
logger.info(f"Payment entry submitted to Fava: {result.get('data', 'Unknown')}")
entry_id = f"ln-{data.payment_hash[:16]}"
await mark_payment_done(data.payment_hash, entry_id)
except BaseException:
# Release the claim so a retry (client or background listener)
# can record this payment.
await release_payment_claim(data.payment_hash)
raise
# Get updated balance from Fava
balance_data = await fava.get_user_balance_bql(target_user_id)
return {
"journal_entry_id": f"fava-{datetime.now().timestamp()}",
"journal_entry_id": entry_id,
"new_balance": balance_data["balance"],
"message": "Payment recorded successfully",
}