laantungir/routstr-core
1
0
Fork 0
mirror of https://github.com/Routstr/routstr-core.git synced 2026-10-05 20:28:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

better retry logic

Browse Source
This commit is contained in:
9qeklajc
2026-07-12 14:12:29 +02:00
parent aea24d23da
commit 07c15de106
6 changed files with 387 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
migrations/versions/d7e8f9a0b1c2_unique_token_type_cashu_transactions.py
+75 -14
View File
@@ -6,8 +6,11 @@ Revises: c6d7e8f9a0b1
from __future__ import annotations
from itertools import groupby
import sqlalchemy as sa
from alembic import op
from sqlalchemy.engine import Connection, RowMapping
revision = "d7e8f9a0b1c2"
down_revision = "c6d7e8f9a0b1"
@@ -15,24 +18,82 @@ branch_labels = None
depends_on = None
_INDEX_NAME = "uq_cashu_transactions_token_type"
_NULLABLE_METADATA = ("request_id", "mint_url", "api_key_hashed_key")
def _merge_duplicate_transactions(connection: Connection) -> None:
transactions = sa.Table(
"cashu_transactions",
sa.MetaData(),
autoload_with=connection,
)
duplicate_keys = (
sa.select(transactions.c.token, transactions.c.type)
.group_by(transactions.c.token, transactions.c.type)
.having(sa.func.count() > 1)
.subquery()
)
rows = (
connection.execute(
sa.select(transactions)
.join(
duplicate_keys,
sa.and_(
transactions.c.token == duplicate_keys.c.token,
transactions.c.type == duplicate_keys.c.type,
),
)
.order_by(
transactions.c.token,
transactions.c.type,
transactions.c.created_at,
transactions.c.id,
)
)
.mappings()
.all()
)
def transaction_key(row: RowMapping) -> tuple[str, str]:
return row["token"], row["type"]
for _, grouped_rows in groupby(rows, key=transaction_key):
duplicates = list(grouped_rows)
if len(duplicates) < 2:
continue
keeper = duplicates[0]
updates: dict[str, object] = {
"collected": any(row["collected"] for row in duplicates),
"swept": any(row["swept"] for row in duplicates),
}
for column in _NULLABLE_METADATA:
if not keeper[column]:
updates[column] = next(
(row[column] for row in duplicates if row[column]),
keeper[column],
)
if not keeper["source"]:
updates["source"] = next(
(row["source"] for row in duplicates if row["source"]),
keeper["source"],
)
connection.execute(
transactions.update()
.where(transactions.c.id == keeper["id"])
.values(**updates)
)
connection.execute(
transactions.delete().where(
transactions.c.id.in_(row["id"] for row in duplicates[1:])
)
)
def upgrade() -> None:
connection = op.get_bind()
connection.execute(
sa.text(
"DELETE FROM cashu_transactions "
"WHERE id IN ("
" SELECT id FROM ("
" SELECT id, ROW_NUMBER() OVER ("
" PARTITION BY token, type "
" ORDER BY created_at ASC, id ASC"
" ) AS row_number "
" FROM cashu_transactions"
" ) WHERE row_number > 1"
")"
)
)
_merge_duplicate_transactions(connection)
op.create_index(
_INDEX_NAME,
"cashu_transactions",
routstr/core/db.py
+27 -9
View File
@@ -321,22 +321,41 @@ async def store_cashu_transaction(
api_key_hashed_key=api_key_hashed_key,
)
last_error_type: str | None = None
attempts_performed = 0
for attempt in range(1, max_attempts + 1):
attempts_performed = attempt
retry_error: OperationalError | None = None
try:
await _insert_cashu_transaction(transaction)
return True
except IntegrityError:
if await _cashu_transaction_exists(token, typ):
return True
break
except IntegrityError as error:
last_error_type = type(error).__name__
try:
if await _cashu_transaction_exists(token, typ):
return True
except OperationalError as lookup_error:
retry_error = lookup_error
except Exception as lookup_error:
last_error_type = type(lookup_error).__name__
break
else:
break
except OperationalError as error:
retry_error = error
except Exception as error:
last_error_type = type(error).__name__
break
if retry_error is not None:
last_error_type = type(retry_error.orig).__name__
if attempt == max_attempts:
break
delay = 0.25 * (2 ** (attempt - 1))
logger.warning(
"Transient database failure storing Cashu transaction; retrying",
extra={
"error": str(error),
"error_type": last_error_type,
"type": typ,
"request_id": request_id,
"attempt": attempt,
@@ -345,18 +364,17 @@ async def store_cashu_transaction(
},
)
await asyncio.sleep(delay)
except Exception:
break
logger.critical(
"Cashu transaction could not be stored after bounded retries",
"Cashu transaction could not be stored",
extra={
"error_type": last_error_type,
"type": typ,
"request_id": request_id,
"amount": amount,
"unit": unit,
"mint_url": mint_url,
"token": token,
"attempts_performed": attempts_performed,
"max_attempts": max_attempts,
},
)
routstr/wallet.py
+9 -3
View File
@@ -733,8 +733,9 @@ async def credit_balance(
extra={"new_balance": key.balance},
)
transaction_stored = False
try:
await store_cashu_transaction(
transaction_stored = await store_cashu_transaction(
token=cashu_token,
amount=original_amount,
unit=original_unit,
@@ -747,8 +748,13 @@ async def credit_balance(
pass
logger.debug(
"Cashu token successfully redeemed and stored",
extra={"amount": amount, "unit": unit, "mint_url": mint_url},
"Cashu token successfully redeemed",
extra={
"amount": amount,
"unit": unit,
"mint_url": mint_url,
"transaction_stored": transaction_stored,
},
)
return amount
except Exception as e:
tests/unit/test_cashu_transaction_storage.py
+110 -2
View File
@@ -49,8 +49,20 @@ async def test_store_cashu_transaction_stops_after_bounded_retries(
assert stored is False
assert insert.await_count == 3
assert sleep.await_count == 2
critical.assert_called_once()
assert critical.call_args.kwargs["extra"]["token"] == "cashuAfailed"
critical.assert_called_once_with(
"Cashu transaction could not be stored",
extra={
"error_type": "Exception",
"type": "out",
"request_id": None,
"amount": 100,
"unit": "sat",
"mint_url": None,
"attempts_performed": 3,
"max_attempts": 3,
},
)
assert "cashuAfailed" not in repr(critical.call_args)
@pytest.mark.asyncio
@@ -71,3 +83,99 @@ async def test_store_cashu_transaction_treats_duplicate_as_success(
assert stored is True
insert.assert_awaited_once()
exists.assert_awaited_once_with("cashuAduplicate", "out")
@pytest.mark.asyncio
async def test_store_cashu_transaction_retries_duplicate_lookup_failure(
monkeypatch: pytest.MonkeyPatch,
) -> None:
insert = AsyncMock(
side_effect=IntegrityError("insert", {}, Exception("unique constraint"))
)
exists = AsyncMock(
side_effect=[
OperationalError("select", {}, Exception("database is locked")),
True,
]
)
sleep = AsyncMock()
warning = AsyncMock()
monkeypatch.setattr(db, "_insert_cashu_transaction", insert)
monkeypatch.setattr(db, "_cashu_transaction_exists", exists)
monkeypatch.setattr(db.asyncio, "sleep", sleep)
monkeypatch.setattr(db.logger, "warning", warning)
stored = await db.store_cashu_transaction(
token="cashuAlookup-retry", amount=100, unit="sat", typ="out"
)
assert stored is True
assert insert.await_count == 2
assert exists.await_count == 2
sleep.assert_awaited_once_with(0.25)
assert "cashuAlookup-retry" not in repr(warning.call_args)
assert warning.call_args.kwargs["extra"]["error_type"] == "Exception"
@pytest.mark.asyncio
async def test_store_cashu_transaction_bounds_duplicate_lookup_failures(
monkeypatch: pytest.MonkeyPatch,
) -> None:
insert = AsyncMock(
side_effect=IntegrityError("insert", {}, Exception("unique constraint"))
)
exists = AsyncMock(
side_effect=OperationalError("select", {}, Exception("database is locked"))
)
sleep = AsyncMock()
critical = AsyncMock()
monkeypatch.setattr(db, "_insert_cashu_transaction", insert)
monkeypatch.setattr(db, "_cashu_transaction_exists", exists)
monkeypatch.setattr(db.asyncio, "sleep", sleep)
monkeypatch.setattr(db.logger, "critical", critical)
stored = await db.store_cashu_transaction(
token="cashuAlookup-failed", amount=100, unit="sat", typ="out"
)
assert stored is False
assert insert.await_count == 3
assert exists.await_count == 3
assert sleep.await_count == 2
critical.assert_called_once()
assert "cashuAlookup-failed" not in repr(critical.call_args)
@pytest.mark.asyncio
async def test_store_cashu_transaction_contains_non_transient_lookup_failure(
monkeypatch: pytest.MonkeyPatch,
) -> None:
insert = AsyncMock(
side_effect=IntegrityError("insert", {}, Exception("unique constraint"))
)
exists = AsyncMock(side_effect=RuntimeError("lookup failed"))
critical = AsyncMock()
monkeypatch.setattr(db, "_insert_cashu_transaction", insert)
monkeypatch.setattr(db, "_cashu_transaction_exists", exists)
monkeypatch.setattr(db.logger, "critical", critical)
stored = await db.store_cashu_transaction(
token="cashuAlookup-error", amount=100, unit="sat", typ="out"
)
assert stored is False
insert.assert_awaited_once()
exists.assert_awaited_once()
critical.assert_called_once_with(
"Cashu transaction could not be stored",
extra={
"error_type": "RuntimeError",
"type": "out",
"request_id": None,
"amount": 100,
"unit": "sat",
"mint_url": None,
"attempts_performed": 1,
"max_attempts": 3,
},
)
tests/unit/test_cashu_transaction_uniqueness_migration.py
+110
View File
@@ -0,0 +1,110 @@
from __future__ import annotations
import importlib.util
from pathlib import Path
import sqlalchemy as sa
_MIGRATION_PATH = (
Path(__file__).resolve().parents[2]
/ "migrations"
/ "versions"
/ "d7e8f9a0b1c2_unique_token_type_cashu_transactions.py"
)
_spec = importlib.util.spec_from_file_location(
"cashu_transaction_uniqueness_migration", _MIGRATION_PATH
)
assert _spec is not None and _spec.loader is not None
migration = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(migration)
def test_duplicate_merge_preserves_state_and_missing_linkage() -> None:
engine = sa.create_engine("sqlite:///:memory:")
metadata = sa.MetaData()
transactions = sa.Table(
"cashu_transactions",
metadata,
sa.Column("id", sa.String, primary_key=True),
sa.Column("token", sa.String, nullable=False),
sa.Column("amount", sa.Integer, nullable=False),
sa.Column("unit", sa.String, nullable=False),
sa.Column("mint_url", sa.String),
sa.Column("type", sa.String, nullable=False),
sa.Column("request_id", sa.String),
sa.Column("created_at", sa.Integer, nullable=False),
sa.Column("collected", sa.Boolean, nullable=False),
sa.Column("swept", sa.Boolean, nullable=False),
sa.Column("source", sa.String, nullable=False),
sa.Column("api_key_hashed_key", sa.String),
)
metadata.create_all(engine)
with engine.begin() as connection:
connection.execute(
transactions.insert(),
[
{
"id": "oldest",
"token": "cashuAduplicate",
"amount": 100,
"unit": "sat",
"mint_url": "",
"type": "out",
"request_id": "",
"created_at": 1,
"collected": False,
"swept": False,
"source": "",
"api_key_hashed_key": "",
},
{
"id": "newer",
"token": "cashuAduplicate",
"amount": 100,
"unit": "sat",
"mint_url": "https://mint.example",
"type": "out",
"request_id": "request-newer",
"created_at": 2,
"collected": True,
"swept": False,
"source": "apikey",
"api_key_hashed_key": "hashed-key",
},
{
"id": "newest",
"token": "cashuAduplicate",
"amount": 100,
"unit": "sat",
"mint_url": "https://other.example",
"type": "out",
"request_id": "request-newest",
"created_at": 3,
"collected": False,
"swept": True,
"source": "x-cashu",
"api_key_hashed_key": "other-key",
},
],
)
migration._merge_duplicate_transactions(connection)
rows = connection.execute(sa.select(transactions)).mappings().all()
assert rows == [
{
"id": "oldest",
"token": "cashuAduplicate",
"amount": 100,
"unit": "sat",
"mint_url": "https://mint.example",
"type": "out",
"request_id": "request-newer",
"created_at": 1,
"collected": True,
"swept": True,
"source": "apikey",
"api_key_hashed_key": "hashed-key",
}
]
tests/unit/test_wallet.py
+56 -16
View File
@@ -1060,31 +1060,71 @@ async def test_credit_balance_msat_unit_not_converted() -> None:
@pytest.mark.asyncio
async def test_credit_balance_survives_audit_store_failure() -> None:
"""A failure writing the CashuTransaction history record must not undo the
already-committed balance credit. (The silent swallow is a known
audit-trail gap slated for its own fix — this test pins the financial
invariant that the user keeps their credit, not the swallow itself.)"""
mock_key = Mock()
mock_key.balance = 0
mock_key.hashed_key = "test_hash"
async def test_credit_balance_reports_audit_store_false() -> None:
mock_key = Mock(balance=0, hashed_key="test_hash")
mock_session = AsyncMock()
debug = Mock()
from routstr.core.settings import settings
with patch.object(settings, "cashu_mints", ["http://mint:3338"]):
with patch(
with (
patch.object(settings, "cashu_mints", ["http://mint:3338"]),
patch(
"routstr.wallet.recieve_token",
return_value=(1000, "sat", "http://mint:3338"),
):
with patch(
"routstr.wallet.store_cashu_transaction",
side_effect=Exception("history table locked"),
):
amount = await credit_balance("cashuAtest", mock_key, mock_session)
),
patch("routstr.wallet.store_cashu_transaction", return_value=False),
patch("routstr.wallet.logger.debug", debug),
):
amount = await credit_balance("cashuAtest", mock_key, mock_session)
assert amount == 1_000_000
assert mock_session.commit.called
debug.assert_called_once_with(
"Cashu token successfully redeemed",
extra={
"amount": 1_000_000,
"unit": "sat",
"mint_url": "http://mint:3338",
"transaction_stored": False,
},
)
@pytest.mark.asyncio
async def test_credit_balance_survives_audit_store_failure() -> None:
"""A failed CashuTransaction history write must not undo balance credit."""
mock_key = Mock(balance=0, hashed_key="test_hash")
mock_session = AsyncMock()
debug = Mock()
from routstr.core.settings import settings
with (
patch.object(settings, "cashu_mints", ["http://mint:3338"]),
patch(
"routstr.wallet.recieve_token",
return_value=(1000, "sat", "http://mint:3338"),
),
patch(
"routstr.wallet.store_cashu_transaction",
side_effect=Exception("history table locked"),
),
patch("routstr.wallet.logger.debug", debug),
):
amount = await credit_balance("cashuAtest", mock_key, mock_session)
assert amount == 1_000_000
assert mock_session.commit.called
debug.assert_called_once_with(
"Cashu token successfully redeemed",
extra={
"amount": 1_000_000,
"unit": "sat",
"mint_url": "http://mint:3338",
"transaction_stored": False,
},
)
@pytest.mark.asyncio