From f6d1a417281be8052beaee65ce702c26119fc138 Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Sat, 18 Jul 2026 14:08:41 +0200 Subject: [PATCH] fix: checkpoint fee payouts before sending --- .../d7e8f9a0b1c2_add_fee_payout_checkpoint.py | 37 +++++ routstr/core/db.py | 30 +++- routstr/wallet.py | 47 ++++++- tests/unit/test_fee_payout_crash_safety.py | 128 ++++++++++++++++++ 4 files changed, 236 insertions(+), 6 deletions(-) create mode 100644 migrations/versions/d7e8f9a0b1c2_add_fee_payout_checkpoint.py create mode 100644 tests/unit/test_fee_payout_crash_safety.py diff --git a/migrations/versions/d7e8f9a0b1c2_add_fee_payout_checkpoint.py b/migrations/versions/d7e8f9a0b1c2_add_fee_payout_checkpoint.py new file mode 100644 index 0000000..301a0b5 --- /dev/null +++ b/migrations/versions/d7e8f9a0b1c2_add_fee_payout_checkpoint.py @@ -0,0 +1,37 @@ +"""add fee payout checkpoint + +Revision ID: d7e8f9a0b1c2 +Revises: c6d7e8f9a0b1 +Create Date: 2026-07-18 00:00:00.000000 +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op + +revision = "d7e8f9a0b1c2" +down_revision = "c6d7e8f9a0b1" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column( + "routstr_fees", + sa.Column( + "payout_in_progress_msats", + sa.Integer(), + nullable=False, + server_default="0", + ), + ) + op.add_column( + "routstr_fees", + sa.Column("payout_started_at", sa.Integer(), nullable=True), + ) + + +def downgrade() -> None: + op.drop_column("routstr_fees", "payout_started_at") + op.drop_column("routstr_fees", "payout_in_progress_msats") diff --git a/routstr/core/db.py b/routstr/core/db.py index 9ebb87e..5654a5c 100644 --- a/routstr/core/db.py +++ b/routstr/core/db.py @@ -354,6 +354,8 @@ class RoutstrFee(SQLModel, table=True): # type: ignore accumulated_msats: int = Field(default=0) total_paid_msats: int = Field(default=0) last_paid_at: int | None = Field(default=None) + payout_in_progress_msats: int = Field(default=0) + payout_started_at: int | None = Field(default=None) class CliToken(SQLModel, table=True): # type: ignore @@ -394,18 +396,42 @@ async def get_routstr_fee(session: AsyncSession) -> RoutstrFee: return fee -async def reset_routstr_fee(session: AsyncSession, paid_msats: int) -> None: +async def reset_routstr_fee(session: AsyncSession, paid_msats: int) -> bool: + """Checkpoint a fee payout before making the external payment.""" stmt = ( update(RoutstrFee) .where(col(RoutstrFee.id) == 1) + .where(col(RoutstrFee.payout_in_progress_msats) == 0) + .where(col(RoutstrFee.accumulated_msats) >= paid_msats) .values( accumulated_msats=RoutstrFee.accumulated_msats - paid_msats, + payout_in_progress_msats=paid_msats, + payout_started_at=int(time.time()), + ) + ) + result = await session.exec(stmt) # type: ignore[call-overload] + await session.commit() + return result.rowcount == 1 + + +async def complete_routstr_fee_payout( + session: AsyncSession, paid_msats: int +) -> bool: + """Mark a checkpointed payout complete after the external payment succeeds.""" + stmt = ( + update(RoutstrFee) + .where(col(RoutstrFee.id) == 1) + .where(col(RoutstrFee.payout_in_progress_msats) == paid_msats) + .values( + payout_in_progress_msats=0, + payout_started_at=None, total_paid_msats=RoutstrFee.total_paid_msats + paid_msats, last_paid_at=int(time.time()), ) ) - await session.exec(stmt) # type: ignore[call-overload] + result = await session.exec(stmt) # type: ignore[call-overload] await session.commit() + return result.rowcount == 1 async def balances_for_mint_and_unit( diff --git a/routstr/wallet.py b/routstr/wallet.py index 68f93cc..03d45fb 100644 --- a/routstr/wallet.py +++ b/routstr/wallet.py @@ -1067,17 +1067,56 @@ async def periodic_routstr_fee_payout() -> None: try: async with db.create_session() as session: fee = await db.get_routstr_fee(session) + if fee.payout_in_progress_msats: + logger.critical( + "Routstr fee payout requires manual reconciliation", + extra={ + "payout_in_progress_msats": fee.payout_in_progress_msats, + "payout_started_at": fee.payout_started_at, + }, + ) + continue + accumulated_sats = fee.accumulated_msats // 1000 if accumulated_sats >= ROUTSTR_FEE_DEFAULT_PAYOUT: wallet = await get_wallet(settings.primary_mint, "sat") proofs = get_proofs_per_mint_and_unit( wallet, settings.primary_mint, "sat", not_reserved=True ) - amount_received = await raw_send_to_lnurl( - wallet, proofs, ROUTSTR_LN_ADDRESS, "sat", amount=accumulated_sats - ) paid_msats = accumulated_sats * 1000 - await db.reset_routstr_fee(session, paid_msats) + payout_checkpointed = await db.reset_routstr_fee( + session, paid_msats + ) + if not payout_checkpointed: + logger.warning("Routstr fee payout was already claimed") + continue + + try: + amount_received = await raw_send_to_lnurl( + wallet, + proofs, + ROUTSTR_LN_ADDRESS, + "sat", + amount=accumulated_sats, + ) + except Exception: + logger.critical( + "Routstr fee payout outcome is unknown; manual reconciliation required", + extra={"payout_in_progress_msats": paid_msats}, + exc_info=True, + ) + continue + + payout_completed = await db.complete_routstr_fee_payout( + session, paid_msats + ) + if not payout_completed: + logger.critical( + "Routstr fee payout sent but checkpoint was not completed", + extra={"payout_in_progress_msats": paid_msats}, + ) + continue + logger.info( "Routstr fee payout sent", extra={ diff --git a/tests/unit/test_fee_payout_crash_safety.py b/tests/unit/test_fee_payout_crash_safety.py new file mode 100644 index 0000000..04206b7 --- /dev/null +++ b/tests/unit/test_fee_payout_crash_safety.py @@ -0,0 +1,128 @@ +import asyncio +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock, patch + +import pytest +from sqlalchemy.ext.asyncio import create_async_engine +from sqlmodel import SQLModel +from sqlmodel.ext.asyncio.session import AsyncSession + +from routstr import wallet +from routstr.core import db + + +@asynccontextmanager +async def _session_context(session: Mock) -> AsyncIterator[Mock]: + yield session + + +@pytest.mark.asyncio +async def test_fee_payout_checkpoint_is_atomic_and_durable() -> None: + engine = create_async_engine("sqlite+aiosqlite://") + async with engine.begin() as connection: + await connection.run_sync(SQLModel.metadata.create_all) + + async with AsyncSession(engine) as session: + session.add(db.RoutstrFee(id=1, accumulated_msats=5_000)) + await session.commit() + + assert await db.reset_routstr_fee(session, 5_000) is True + assert await db.reset_routstr_fee(session, 5_000) is False + + fee = await db.get_routstr_fee(session) + await session.refresh(fee) + assert fee.accumulated_msats == 0 + assert fee.payout_in_progress_msats == 5_000 + assert fee.total_paid_msats == 0 + + assert await db.complete_routstr_fee_payout(session, 5_000) is True + await session.refresh(fee) + assert fee.payout_in_progress_msats == 0 + assert fee.total_paid_msats == 5_000 + assert fee.last_paid_at is not None + + await engine.dispose() + + +@pytest.mark.asyncio +async def test_fee_payout_checkpoints_before_sending() -> None: + session = Mock() + fee = SimpleNamespace( + accumulated_msats=5_000, + payout_in_progress_msats=0, + payout_started_at=None, + ) + payout_wallet = Mock() + events: list[str] = [] + + async def checkpoint(*_args: object) -> bool: + events.append("checkpoint") + return True + + async def send(*_args: object, **_kwargs: object) -> int: + events.append("send") + return 5 + + async def complete(*_args: object) -> bool: + events.append("complete") + return True + + with ( + patch("routstr.auth.ROUTSTR_FEE_DEFAULT_PAYOUT", 1), + patch("routstr.auth.ROUTSTR_FEE_PAYOUT_INTERVAL_SECONDS", 1), + patch("routstr.auth.ROUTSTR_LN_ADDRESS", "fees@example.com"), + patch( + "routstr.wallet.asyncio.sleep", + AsyncMock(side_effect=[None, asyncio.CancelledError()]), + ), + patch("routstr.wallet.db.create_session", return_value=_session_context(session)), + patch("routstr.wallet.db.get_routstr_fee", AsyncMock(return_value=fee)), + patch("routstr.wallet.db.reset_routstr_fee", side_effect=checkpoint), + patch("routstr.wallet.db.complete_routstr_fee_payout", side_effect=complete), + patch("routstr.wallet.get_wallet", AsyncMock(return_value=payout_wallet)), + patch("routstr.wallet.get_proofs_per_mint_and_unit", return_value=[]), + patch("routstr.wallet.raw_send_to_lnurl", side_effect=send), + ): + with pytest.raises(asyncio.CancelledError): + await wallet.periodic_routstr_fee_payout() + + assert events == ["checkpoint", "send", "complete"] + + +@pytest.mark.asyncio +async def test_fee_payout_keeps_checkpoint_when_send_outcome_is_unknown() -> None: + session = Mock() + fee = SimpleNamespace( + accumulated_msats=5_000, + payout_in_progress_msats=0, + payout_started_at=None, + ) + complete = AsyncMock() + + with ( + patch("routstr.auth.ROUTSTR_FEE_DEFAULT_PAYOUT", 1), + patch("routstr.auth.ROUTSTR_FEE_PAYOUT_INTERVAL_SECONDS", 1), + patch("routstr.auth.ROUTSTR_LN_ADDRESS", "fees@example.com"), + patch( + "routstr.wallet.asyncio.sleep", + AsyncMock(side_effect=[None, asyncio.CancelledError()]), + ), + patch("routstr.wallet.db.create_session", return_value=_session_context(session)), + patch("routstr.wallet.db.get_routstr_fee", AsyncMock(return_value=fee)), + patch("routstr.wallet.db.reset_routstr_fee", AsyncMock(return_value=True)), + patch("routstr.wallet.db.complete_routstr_fee_payout", complete), + patch("routstr.wallet.get_wallet", AsyncMock(return_value=Mock())), + patch("routstr.wallet.get_proofs_per_mint_and_unit", return_value=[]), + patch( + "routstr.wallet.raw_send_to_lnurl", + AsyncMock(side_effect=TimeoutError("unknown outcome")), + ), + patch("routstr.wallet.logger.critical") as critical, + ): + with pytest.raises(asyncio.CancelledError): + await wallet.periodic_routstr_fee_payout() + + complete.assert_not_awaited() + critical.assert_called_once()