fix: checkpoint fee payouts before sending

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")

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(

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={

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()