Compare commits

..

No commits in common. "fix/payment-idempotency" and "main" have entirely different histories.

6 changed files with 104 additions and 699 deletions

70
crud.py
View file

@ -1696,73 +1696,3 @@ async def check_user_has_role_permission(
return True return True
return False return False
# =============================================================================
# PROCESSED PAYMENTS (Lightning payment idempotency gate)
# =============================================================================
# The Fava-side duplicate checks are read-then-write races; this table's
# primary key on payment_hash makes exactly one claimant win. Shared by the
# background invoice listener (tasks.on_invoice_paid) and the client-driven
# /record-payment endpoint.
async def claim_payment(payment_hash: str) -> bool:
"""Atomically claim a Lightning payment for recording.
Returns True when this caller owns the claim; False when the payment
is already recorded or another coroutine is recording it right now.
"""
result = await db.execute(
"""
INSERT INTO processed_payments (payment_hash, status)
VALUES (:payment_hash, 'processing')
ON CONFLICT (payment_hash) DO NOTHING
""",
{"payment_hash": payment_hash},
)
return result.rowcount == 1
async def get_processed_payment(payment_hash: str) -> Optional[dict]:
row = await db.fetchone(
"SELECT payment_hash, status, entry_id FROM processed_payments"
" WHERE payment_hash = :payment_hash",
{"payment_hash": payment_hash},
)
return dict(row) if row else None
async def mark_payment_done(payment_hash: str, entry_id: Optional[str] = None) -> None:
await db.execute(
"""
UPDATE processed_payments SET status = 'done', entry_id = :entry_id
WHERE payment_hash = :payment_hash
""",
{"payment_hash": payment_hash, "entry_id": entry_id},
)
async def release_payment_claim(payment_hash: str) -> None:
"""Compensating delete after a failed recording, so redelivery retries.
Only removes an in-flight claim a 'done' row is permanent.
"""
await db.execute(
"DELETE FROM processed_payments"
" WHERE payment_hash = :payment_hash AND status = 'processing'",
{"payment_hash": payment_hash},
)
async def clear_stale_payment_claims() -> int:
"""Drop 'processing' claims left behind by a previous process life.
A live claim only exists inside a running coroutine, so anything
still 'processing' at listener startup belongs to a crashed or
restarted process and would otherwise block that payment forever.
"""
result = await db.execute(
"DELETE FROM processed_payments WHERE status = 'processing'"
)
return result.rowcount

View file

@ -34,33 +34,9 @@ Original migration sequence (Nov 2025):
- m014: Removed legacy equity accounts (MemberEquity, RetainedEarnings) - m014: Removed legacy equity accounts (MemberEquity, RetainedEarnings)
- m015: Converted entry_lines to single amount field - m015: Converted entry_lines to single amount field
- m016: Dropped journal_entries and entry_lines tables (Fava integration) - m016: Dropped journal_entries and entry_lines tables (Fava integration)
IDEMPOTENCY CONTRACT:
Every statement here must be a silent no-op on re-run. The migration
version bump lands in the core LNbits DB (`dbversions`) while the DDL
lands in `ext_libra` the two writes are not atomic. If the version
bump fails after the DDL commits, the whole migration re-runs on next
boot; a bare CREATE/ALTER/INSERT then crashes the extension until
manual dbversions surgery.
""" """
async def _alter_add_column_safe(db, sql: str) -> None:
"""ALTER TABLE ADD COLUMN that swallows duplicate-column errors.
Neither SQLite nor Postgres supports ADD COLUMN IF NOT EXISTS
portably, so re-runs are made no-ops by swallowing the error both
backends raise for an existing column.
"""
try:
await db.execute(sql)
except Exception as exc:
msg = str(exc).lower()
if "duplicate column" in msg or "already exists" in msg:
return
raise
async def m001_initial(db): async def m001_initial(db):
""" """
Initial Libra database schema (squashed from m001-m016). Initial Libra database schema (squashed from m001-m016).
@ -87,7 +63,7 @@ async def m001_initial(db):
await db.execute( await db.execute(
f""" f"""
CREATE TABLE IF NOT EXISTS accounts ( CREATE TABLE accounts (
id TEXT PRIMARY KEY, id TEXT PRIMARY KEY,
name TEXT NOT NULL UNIQUE, name TEXT NOT NULL UNIQUE,
account_type TEXT NOT NULL, account_type TEXT NOT NULL,
@ -100,13 +76,13 @@ async def m001_initial(db):
await db.execute( await db.execute(
""" """
CREATE INDEX IF NOT EXISTS idx_accounts_user_id ON accounts (user_id); CREATE INDEX idx_accounts_user_id ON accounts (user_id);
""" """
) )
await db.execute( await db.execute(
""" """
CREATE INDEX IF NOT EXISTS idx_accounts_type ON accounts (account_type); CREATE INDEX idx_accounts_type ON accounts (account_type);
""" """
) )
@ -117,7 +93,7 @@ async def m001_initial(db):
await db.execute( await db.execute(
f""" f"""
CREATE TABLE IF NOT EXISTS extension_settings ( CREATE TABLE extension_settings (
id TEXT NOT NULL PRIMARY KEY, id TEXT NOT NULL PRIMARY KEY,
libra_wallet_id TEXT, libra_wallet_id TEXT,
fava_url TEXT NOT NULL DEFAULT 'http://localhost:3333', fava_url TEXT NOT NULL DEFAULT 'http://localhost:3333',
@ -135,7 +111,7 @@ async def m001_initial(db):
await db.execute( await db.execute(
f""" f"""
CREATE TABLE IF NOT EXISTS user_wallet_settings ( CREATE TABLE user_wallet_settings (
id TEXT NOT NULL PRIMARY KEY, id TEXT NOT NULL PRIMARY KEY,
user_wallet_id TEXT, user_wallet_id TEXT,
updated_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now} updated_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now}
@ -150,7 +126,7 @@ async def m001_initial(db):
await db.execute( await db.execute(
f""" f"""
CREATE TABLE IF NOT EXISTS manual_payment_requests ( CREATE TABLE manual_payment_requests (
id TEXT PRIMARY KEY, id TEXT PRIMARY KEY,
user_id TEXT NOT NULL, user_id TEXT NOT NULL,
amount INTEGER NOT NULL, amount INTEGER NOT NULL,
@ -167,14 +143,14 @@ async def m001_initial(db):
await db.execute( await db.execute(
""" """
CREATE INDEX IF NOT EXISTS idx_manual_payment_requests_user_id CREATE INDEX idx_manual_payment_requests_user_id
ON manual_payment_requests (user_id); ON manual_payment_requests (user_id);
""" """
) )
await db.execute( await db.execute(
""" """
CREATE INDEX IF NOT EXISTS idx_manual_payment_requests_status CREATE INDEX idx_manual_payment_requests_status
ON manual_payment_requests (status); ON manual_payment_requests (status);
""" """
) )
@ -187,7 +163,7 @@ async def m001_initial(db):
await db.execute( await db.execute(
f""" f"""
CREATE TABLE IF NOT EXISTS balance_assertions ( CREATE TABLE balance_assertions (
id TEXT PRIMARY KEY, id TEXT PRIMARY KEY,
date TIMESTAMP NOT NULL, date TIMESTAMP NOT NULL,
account_id TEXT NOT NULL, account_id TEXT NOT NULL,
@ -212,21 +188,21 @@ async def m001_initial(db):
await db.execute( await db.execute(
""" """
CREATE INDEX IF NOT EXISTS idx_balance_assertions_account_id CREATE INDEX idx_balance_assertions_account_id
ON balance_assertions (account_id); ON balance_assertions (account_id);
""" """
) )
await db.execute( await db.execute(
""" """
CREATE INDEX IF NOT EXISTS idx_balance_assertions_status CREATE INDEX idx_balance_assertions_status
ON balance_assertions (status); ON balance_assertions (status);
""" """
) )
await db.execute( await db.execute(
""" """
CREATE INDEX IF NOT EXISTS idx_balance_assertions_date CREATE INDEX idx_balance_assertions_date
ON balance_assertions (date); ON balance_assertions (date);
""" """
) )
@ -240,7 +216,7 @@ async def m001_initial(db):
await db.execute( await db.execute(
f""" f"""
CREATE TABLE IF NOT EXISTS user_equity_status ( CREATE TABLE user_equity_status (
user_id TEXT PRIMARY KEY, user_id TEXT PRIMARY KEY,
is_equity_eligible BOOLEAN NOT NULL DEFAULT FALSE, is_equity_eligible BOOLEAN NOT NULL DEFAULT FALSE,
equity_account_name TEXT, equity_account_name TEXT,
@ -254,7 +230,7 @@ async def m001_initial(db):
await db.execute( await db.execute(
""" """
CREATE INDEX IF NOT EXISTS idx_user_equity_status_eligible CREATE INDEX idx_user_equity_status_eligible
ON user_equity_status (is_equity_eligible) ON user_equity_status (is_equity_eligible)
WHERE is_equity_eligible = TRUE; WHERE is_equity_eligible = TRUE;
""" """
@ -269,7 +245,7 @@ async def m001_initial(db):
await db.execute( await db.execute(
f""" f"""
CREATE TABLE IF NOT EXISTS account_permissions ( CREATE TABLE account_permissions (
id TEXT PRIMARY KEY, id TEXT PRIMARY KEY,
user_id TEXT NOT NULL, user_id TEXT NOT NULL,
account_id TEXT NOT NULL, account_id TEXT NOT NULL,
@ -286,7 +262,7 @@ async def m001_initial(db):
# Index for looking up permissions by user # Index for looking up permissions by user
await db.execute( await db.execute(
""" """
CREATE INDEX IF NOT EXISTS idx_account_permissions_user_id CREATE INDEX idx_account_permissions_user_id
ON account_permissions (user_id); ON account_permissions (user_id);
""" """
) )
@ -294,7 +270,7 @@ async def m001_initial(db):
# Index for looking up permissions by account # Index for looking up permissions by account
await db.execute( await db.execute(
""" """
CREATE INDEX IF NOT EXISTS idx_account_permissions_account_id CREATE INDEX idx_account_permissions_account_id
ON account_permissions (account_id); ON account_permissions (account_id);
""" """
) )
@ -302,7 +278,7 @@ async def m001_initial(db):
# Composite index for checking specific user+account permissions # Composite index for checking specific user+account permissions
await db.execute( await db.execute(
""" """
CREATE INDEX IF NOT EXISTS idx_account_permissions_user_account CREATE INDEX idx_account_permissions_user_account
ON account_permissions (user_id, account_id); ON account_permissions (user_id, account_id);
""" """
) )
@ -310,7 +286,7 @@ async def m001_initial(db):
# Index for finding permissions by type # Index for finding permissions by type
await db.execute( await db.execute(
""" """
CREATE INDEX IF NOT EXISTS idx_account_permissions_type CREATE INDEX idx_account_permissions_type
ON account_permissions (permission_type); ON account_permissions (permission_type);
""" """
) )
@ -318,7 +294,7 @@ async def m001_initial(db):
# Index for finding expired permissions # Index for finding expired permissions
await db.execute( await db.execute(
""" """
CREATE INDEX IF NOT EXISTS idx_account_permissions_expires CREATE INDEX idx_account_permissions_expires
ON account_permissions (expires_at) ON account_permissions (expires_at)
WHERE expires_at IS NOT NULL; WHERE expires_at IS NOT NULL;
""" """
@ -344,7 +320,6 @@ async def m001_initial(db):
f""" f"""
INSERT INTO accounts (id, name, account_type, description, created_at) INSERT INTO accounts (id, name, account_type, description, created_at)
VALUES (:id, :name, :type, :description, {db.timestamp_now}) VALUES (:id, :name, :type, :description, {db.timestamp_now})
ON CONFLICT (name) DO NOTHING
""", """,
{ {
"id": str(uuid.uuid4()), "id": str(uuid.uuid4()),
@ -367,18 +342,17 @@ async def m002_add_account_is_active(db):
Default: All existing accounts are marked as active (TRUE). Default: All existing accounts are marked as active (TRUE).
""" """
await _alter_add_column_safe( await db.execute(
db,
""" """
ALTER TABLE accounts ALTER TABLE accounts
ADD COLUMN is_active BOOLEAN NOT NULL DEFAULT TRUE ADD COLUMN is_active BOOLEAN NOT NULL DEFAULT TRUE
""", """
) )
# Create index for faster queries filtering by is_active # Create index for faster queries filtering by is_active
await db.execute( await db.execute(
""" """
CREATE INDEX IF NOT EXISTS idx_accounts_is_active ON accounts (is_active) CREATE INDEX idx_accounts_is_active ON accounts (is_active)
""" """
) )
@ -398,18 +372,17 @@ async def m003_add_account_is_virtual(db):
Default: All existing accounts are real (is_virtual = FALSE). Default: All existing accounts are real (is_virtual = FALSE).
""" """
await _alter_add_column_safe( await db.execute(
db,
""" """
ALTER TABLE accounts ALTER TABLE accounts
ADD COLUMN is_virtual BOOLEAN NOT NULL DEFAULT FALSE ADD COLUMN is_virtual BOOLEAN NOT NULL DEFAULT FALSE
""", """
) )
# Create index for faster queries filtering by is_virtual # Create index for faster queries filtering by is_virtual
await db.execute( await db.execute(
""" """
CREATE INDEX IF NOT EXISTS idx_accounts_is_virtual ON accounts (is_virtual) CREATE INDEX idx_accounts_is_virtual ON accounts (is_virtual)
""" """
) )
@ -429,7 +402,6 @@ async def m003_add_account_is_virtual(db):
f""" f"""
INSERT INTO accounts (id, name, account_type, description, is_active, is_virtual, created_at) INSERT INTO accounts (id, name, account_type, description, is_active, is_virtual, created_at)
VALUES (:id, :name, :type, :description, TRUE, TRUE, {db.timestamp_now}) VALUES (:id, :name, :type, :description, TRUE, TRUE, {db.timestamp_now})
ON CONFLICT (name) DO NOTHING
""", """,
{ {
"id": str(uuid.uuid4()), "id": str(uuid.uuid4()),
@ -466,7 +438,7 @@ async def m004_add_rbac_tables(db):
await db.execute( await db.execute(
f""" f"""
CREATE TABLE IF NOT EXISTS roles ( CREATE TABLE roles (
id TEXT PRIMARY KEY, id TEXT PRIMARY KEY,
name TEXT NOT NULL UNIQUE, name TEXT NOT NULL UNIQUE,
description TEXT, description TEXT,
@ -479,13 +451,13 @@ async def m004_add_rbac_tables(db):
await db.execute( await db.execute(
""" """
CREATE INDEX IF NOT EXISTS idx_roles_name ON roles (name); CREATE INDEX idx_roles_name ON roles (name);
""" """
) )
await db.execute( await db.execute(
""" """
CREATE INDEX IF NOT EXISTS idx_roles_is_default ON roles (is_default) CREATE INDEX idx_roles_is_default ON roles (is_default)
WHERE is_default = TRUE; WHERE is_default = TRUE;
""" """
) )
@ -497,7 +469,7 @@ async def m004_add_rbac_tables(db):
await db.execute( await db.execute(
f""" f"""
CREATE TABLE IF NOT EXISTS role_permissions ( CREATE TABLE role_permissions (
id TEXT PRIMARY KEY, id TEXT PRIMARY KEY,
role_id TEXT NOT NULL, role_id TEXT NOT NULL,
account_id TEXT NOT NULL, account_id TEXT NOT NULL,
@ -512,19 +484,19 @@ async def m004_add_rbac_tables(db):
await db.execute( await db.execute(
""" """
CREATE INDEX IF NOT EXISTS idx_role_permissions_role_id ON role_permissions (role_id); CREATE INDEX idx_role_permissions_role_id ON role_permissions (role_id);
""" """
) )
await db.execute( await db.execute(
""" """
CREATE INDEX IF NOT EXISTS idx_role_permissions_account_id ON role_permissions (account_id); CREATE INDEX idx_role_permissions_account_id ON role_permissions (account_id);
""" """
) )
await db.execute( await db.execute(
""" """
CREATE INDEX IF NOT EXISTS idx_role_permissions_type ON role_permissions (permission_type); CREATE INDEX idx_role_permissions_type ON role_permissions (permission_type);
""" """
) )
@ -535,7 +507,7 @@ async def m004_add_rbac_tables(db):
await db.execute( await db.execute(
f""" f"""
CREATE TABLE IF NOT EXISTS user_roles ( CREATE TABLE user_roles (
id TEXT PRIMARY KEY, id TEXT PRIMARY KEY,
user_id TEXT NOT NULL, user_id TEXT NOT NULL,
role_id TEXT NOT NULL, role_id TEXT NOT NULL,
@ -550,19 +522,19 @@ async def m004_add_rbac_tables(db):
await db.execute( await db.execute(
""" """
CREATE INDEX IF NOT EXISTS idx_user_roles_user_id ON user_roles (user_id); CREATE INDEX idx_user_roles_user_id ON user_roles (user_id);
""" """
) )
await db.execute( await db.execute(
""" """
CREATE INDEX IF NOT EXISTS idx_user_roles_role_id ON user_roles (role_id); CREATE INDEX idx_user_roles_role_id ON user_roles (role_id);
""" """
) )
await db.execute( await db.execute(
""" """
CREATE INDEX IF NOT EXISTS idx_user_roles_expires ON user_roles (expires_at) CREATE INDEX idx_user_roles_expires ON user_roles (expires_at)
WHERE expires_at IS NOT NULL; WHERE expires_at IS NOT NULL;
""" """
) )
@ -570,7 +542,7 @@ async def m004_add_rbac_tables(db):
# Composite index for checking specific user+role assignments # Composite index for checking specific user+role assignments
await db.execute( await db.execute(
""" """
CREATE INDEX IF NOT EXISTS idx_user_roles_user_role ON user_roles (user_id, role_id); CREATE INDEX idx_user_roles_user_role ON user_roles (user_id, role_id);
""" """
) )
@ -614,7 +586,6 @@ async def m004_add_rbac_tables(db):
f""" f"""
INSERT INTO roles (id, name, description, is_default, created_by, created_at) INSERT INTO roles (id, name, description, is_default, created_by, created_at)
VALUES (:id, :name, :description, :is_default, :created_by, {db.timestamp_now}) VALUES (:id, :name, :description, :is_default, :created_by, {db.timestamp_now})
ON CONFLICT (name) DO NOTHING
""", """,
{ {
"id": str(uuid.uuid4()), "id": str(uuid.uuid4()),
@ -624,30 +595,3 @@ async def m004_add_rbac_tables(db):
"created_by": "system", # System-created default roles "created_by": "system", # System-created default roles
}, },
) )
async def m005_add_processed_payments(db):
"""
Local idempotency gate for Lightning payment recording.
The Fava-side duplicate check (`add_entry_idempotent`, journal-link
scan) is a read-then-write race: the background invoice listener and
the client-driven /record-payment endpoint can both pass the "not
present" check for the same payment_hash and both insert. The
primary key on payment_hash makes exactly one claimant win.
status lifecycle: 'processing' (claimed, write in flight) 'done'
(entry recorded). Failed claims are deleted so redelivery retries;
'processing' rows from a crashed process are cleared at listener
startup.
"""
await db.execute(
f"""
CREATE TABLE IF NOT EXISTS processed_payments (
payment_hash TEXT PRIMARY KEY,
status TEXT NOT NULL DEFAULT 'processing',
entry_id TEXT,
created_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now}
);
"""
)

View file

@ -179,31 +179,12 @@ async def wait_for_paid_invoices():
This ensures payments are recorded even if the user closes their browser This ensures payments are recorded even if the user closes their browser
before the payment is detected by client-side polling. before the payment is detected by client-side polling.
""" """
from .crud import clear_stale_payment_claims
invoice_queue = Queue() invoice_queue = Queue()
register_invoice_listener(invoice_queue, "ext_libra") register_invoice_listener(invoice_queue, "ext_libra")
# Claims from a previous process life can't be live anymore — clear
# them so those payments aren't blocked forever.
cleared = await clear_stale_payment_claims()
if cleared:
logger.warning(
f"[LIBRA] Cleared {cleared} stale in-flight payment claim(s) "
"from a previous run"
)
while True: while True:
payment = await invoice_queue.get() payment = await invoice_queue.get()
try:
await on_invoice_paid(payment) await on_invoice_paid(payment)
except Exception:
# One bad payment must not kill the listener for the rest of
# the process lifetime; its claim was released, so redelivery
# can retry it.
logger.exception(
f"[LIBRA] Failed to record payment {payment.payment_hash}"
)
async def on_invoice_paid(payment: Payment) -> None: async def on_invoice_paid(payment: Payment) -> None:
@ -229,20 +210,8 @@ async def on_invoice_paid(payment: Payment) -> None:
logger.warning(f"Libra invoice {payment.payment_hash} missing user_id in metadata") logger.warning(f"Libra invoice {payment.payment_hash} missing user_id in metadata")
return return
from .crud import claim_payment, mark_payment_done, release_payment_claim
from .fava_client import get_fava_client from .fava_client import get_fava_client
# Local idempotency gate: exactly one claimant (this listener or the
# /record-payment endpoint) gets to record a given payment_hash. The
# Fava-side idempotent write below stays as a second layer for
# entries recorded before this table existed.
if not await claim_payment(payment.payment_hash):
logger.info(
f"Payment {payment.payment_hash} already recorded or being "
"recorded; skipping"
)
return
fava = get_fava_client() fava = get_fava_client()
# Use idempotency key based on payment hash - this ensures duplicate # Use idempotency key based on payment hash - this ensures duplicate
@ -276,7 +245,6 @@ async def on_invoice_paid(payment: Payment) -> None:
if not fiat_currency or not fiat_amount: if not fiat_currency or not fiat_amount:
logger.error(f"Payment {payment.payment_hash} missing fiat currency/amount metadata") logger.error(f"Payment {payment.payment_hash} missing fiat currency/amount metadata")
await release_payment_claim(payment.payment_hash)
return return
# Get user's current balance to determine receivables and payables # Get user's current balance to determine receivables and payables
@ -308,7 +276,6 @@ async def on_invoice_paid(payment: Payment) -> None:
lightning_account = await get_account_by_name("Assets:Bitcoin:Lightning") lightning_account = await get_account_by_name("Assets:Bitcoin:Lightning")
if not lightning_account: if not lightning_account:
logger.error("Lightning account 'Assets:Bitcoin:Lightning' not found") logger.error("Lightning account 'Assets:Bitcoin:Lightning' not found")
await release_payment_claim(payment.payment_hash)
return return
# Query for unsettled entries to link this settlement back to them # Query for unsettled entries to link this settlement back to them
@ -357,11 +324,6 @@ async def on_invoice_paid(payment: Payment) -> None:
f"{result.get('data', 'Unknown')}" f"{result.get('data', 'Unknown')}"
) )
await mark_payment_done(payment.payment_hash, idempotency_key)
except Exception as e: except Exception as e:
logger.error(f"Error recording Libra payment {payment.payment_hash}: {e}") logger.error(f"Error recording Libra payment {payment.payment_hash}: {e}")
# Release the claim so redelivery (or the /record-payment
# endpoint) can retry this payment.
await release_payment_claim(payment.payment_hash)
raise raise

View file

@ -1,110 +0,0 @@
"""Migration idempotency tests.
The migration version bump lands in the core LNbits DB (`dbversions`)
while the DDL lands in `ext_libra` the two writes are not atomic. If
the bump fails after the DDL commits, the whole migration re-runs on
next boot, so every statement must be a silent no-op on re-run instead
of crashing with `duplicate column` / `table exists` / UNIQUE
violations (which bricks the extension until manual dbversions
surgery).
These tests run the full migration chain twice against a fresh SQLite
database the second pass simulates the lost-version-bump re-run.
"""
import importlib
import re
from uuid import uuid4
import pytest
from lnbits.db import Database
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")
migrations = _module("migrations")
# Same discovery as lnbits.core.helpers.run_migration: m### prefix,
# module definition order.
_MIGRATION_RE = re.compile(r"^m(\d\d\d)_")
MIGRATION_FUNCTIONS = [
fn for name, fn in vars(migrations).items() if _MIGRATION_RE.match(name)
]
# Tables the chain must leave behind — one probe row read per table
# proves both existence and queryability after a double run.
EXPECTED_TABLES = [
"accounts",
"extension_settings",
"user_wallet_settings",
"manual_payment_requests",
"balance_assertions",
"user_equity_status",
"account_permissions",
"roles",
"role_permissions",
"user_roles",
"processed_payments",
]
async def _run_all_migrations(db: Database) -> None:
async with db.connect() as conn:
for migrate in MIGRATION_FUNCTIONS:
await migrate(conn)
async def _seed_counts(db: Database) -> dict:
async with db.connect() as conn:
accounts = await conn.fetchall("SELECT id, name FROM accounts")
roles = await conn.fetchall("SELECT id, name FROM roles")
return {
"account_names": sorted(r["name"] for r in accounts),
"account_ids": sorted(r["id"] for r in accounts),
"role_names": sorted(r["name"] for r in roles),
"role_ids": sorted(r["id"] for r in roles),
}
async def test_migrations_rerun_is_noop():
"""Full chain twice: second run must not raise and not re-seed."""
db = Database(f"ext_libra_migtest_{uuid4().hex[:8]}")
await _run_all_migrations(db)
first = await _seed_counts(db)
# Simulate the lost dbversions bump: everything runs again.
await _run_all_migrations(db)
second = await _seed_counts(db)
# Seeds must not duplicate (names) and must not be replaced (ids).
assert second == first
assert first["account_names"], "seed accounts missing after migration"
assert "Employee" in first["role_names"]
# Every table exists and is queryable after the double run.
async with db.connect() as conn:
for table in EXPECTED_TABLES:
await conn.fetchall(f"SELECT * FROM {table} LIMIT 1") # noqa: S608
async def test_single_migration_rerun_is_noop():
"""Each migration individually survives an immediate re-run (the
version bump fails right after that one migration committed)."""
db = Database(f"ext_libra_migtest_{uuid4().hex[:8]}")
async with db.connect() as conn:
for migrate in MIGRATION_FUNCTIONS:
await migrate(conn)
await migrate(conn) # re-run before "bumping" to the next

View file

@ -1,281 +0,0 @@
"""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

View file

@ -1850,14 +1850,13 @@ async def api_record_payment(
try: try:
async with httpx.AsyncClient(timeout=5.0) as client: async with httpx.AsyncClient(timeout=5.0) as client:
# Get recent entries from Fava's journal endpoint. base_url # Get recent entries from Fava's journal endpoint
# already ends in /api — the previous "/api/journal" path
# 404'd, so this duplicate check silently never ran.
response = await client.get( response = await client.get(
f"{fava.base_url}/journal", f"{fava.base_url}/api/journal",
params={"time": ""} # Get all entries params={"time": ""} # Get all entries
) )
response.raise_for_status()
if response.status_code == 200:
response_data = response.json() response_data = response.json()
entries = response_data.get('entries', []) entries = response_data.get('entries', [])
@ -1872,42 +1871,11 @@ async def api_record_payment(
"new_balance": balance_data["balance"], "new_balance": balance_data["balance"],
"message": "Payment already recorded", "message": "Payment already recorded",
} }
except httpx.HTTPError as e: except Exception 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}") logger.warning(f"Could not check Fava for duplicate payment: {e}")
raise HTTPException( # Continue anyway - Fava/Beancount will catch duplicate if it exists
status_code=HTTPStatus.SERVICE_UNAVAILABLE,
detail="Cannot verify payment duplicate status; try again shortly",
)
# 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,
)
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",
)
# Convert amount from millisatoshis to satoshis # Convert amount from millisatoshis to satoshis
try:
amount_sats = payment.amount // 1000 amount_sats = payment.amount // 1000
# Extract fiat metadata from invoice (if present) # Extract fiat metadata from invoice (if present)
@ -1961,19 +1929,11 @@ async def api_record_payment(
result = await fava.add_entry(entry) result = await fava.add_entry(entry)
logger.info(f"Payment entry submitted to Fava: {result.get('data', 'Unknown')}") 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 # Get updated balance from Fava
balance_data = await fava.get_user_balance_bql(target_user_id) balance_data = await fava.get_user_balance_bql(target_user_id)
return { return {
"journal_entry_id": entry_id, "journal_entry_id": f"fava-{datetime.now().timestamp()}",
"new_balance": balance_data["balance"], "new_balance": balance_data["balance"],
"message": "Payment recorded successfully", "message": "Payment recorded successfully",
} }