Files
davideandClaude Sonnet 5.5 f936d6660c Track every version of a bumped transaction and write the RBF replacement before broadcasting it (B-77, TX-12, TX-13, TX-19, TX-21, DAT-13, DAT-14, DAT-34, DAT-44, DAT-45, DAT-70, DAT-71)
The poller and the reconciler matched current_txid only. If an earlier version of
a bumped transaction confirmed it was never seen, and after 6h the row was abandoned:
a confirmed bet's participant deleted, a sent withdrawal marked failed, a paid payout
rebuilt. bump_fee also broadcast in phase 2 and persisted in phase 3 unconditionally,
so a crash or a `database is locked` in between left a transaction on the network that
the DB did not know, and an abandon or a confirmation in the meantime was overwritten.

- New table pending_transaction_txids (migration a3f5c2d9b7e1; txid unique, index on
  the owning row; existing rows backfilled with current_txid + raw and the txid their
  last bump replaced, raw_hex empty). Written with the row by place_bet,
  request_withdrawal and _trigger_payout; removed with a bet row that a verified
  refusal deletes (DAT-38).
- bump_fee in four phases: read, build and sign, commit the replacement as a candidate
  version, broadcast (verdict of B-76, fanned out when accepted; an explicit refusal
  raises RbfError and leaves the row), then switch the row over with
  `UPDATE ... WHERE status='pending' AND current_txid=<replaced>`. Zero rows matched is
  a logged conflict: nothing is overwritten and the candidate stays recorded. An
  uncertain broadcast adopts the replacement. A retry rebuilds the same candidate
  without tripping the unique txid.
- poll_once and reconcile_once look for every txid of a row. When the version found is
  not the current one, broadcast.adopt_version makes it current (raw tx included) and
  retargets spent_txid, bet_txid, withdrawals.txid and payout_txid as a bump does,
  before the confirmation handler runs. _release_inputs frees inputs claimed by any
  version of the row.
- RbfBumper takes the listener as `network`.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-14 14:46:43 +02:00

54 lines
2.6 KiB
Python

"""The versions of a pending transaction (B-77, DAT-70, DAT-71, TX-19).
A bumped transaction has several txids and any of them may be the one that confirms.
Everything that looks a transaction up on the chain asks for all of them."""
from sqlalchemy import delete, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.models import PendingTransaction, PendingTransactionTxid
async def record_version(session: AsyncSession, pending: PendingTransaction, txid: str, raw_hex: str) -> None:
"""Write `txid` as a version of `pending` in the caller's transaction. Idempotent for
the same pair (a bump retried after an uncertain broadcast rebuilds the same
transaction); a txid already held by another row is an error — it is unique across
the platform."""
if pending.id is None:
await session.flush()
existing = await session.scalar(select(PendingTransactionTxid).where(PendingTransactionTxid.txid == txid))
if existing is not None:
if existing.pending_transaction_id != pending.id:
raise ValueError(f"txid {txid} already belongs to pending_transaction {existing.pending_transaction_id}")
return
session.add(PendingTransactionTxid(pending_transaction_id=pending.id, txid=txid, raw_hex=raw_hex))
async def txids_of(session: AsyncSession, pending: PendingTransaction) -> set[str]:
"""Every txid of the row: all its recorded versions plus `current_txid` (a row
written before the table existed, or by a test, may have no version row for it)."""
rows = await session.scalars(
select(PendingTransactionTxid.txid).where(PendingTransactionTxid.pending_transaction_id == pending.id)
)
return {pending.current_txid, *rows.all()}
async def txids_by_row(session: AsyncSession, current: dict[int, str]) -> dict[int, set[str]]:
"""`txids_of` for many rows at once; `current` maps row id -> current_txid."""
result = {row_id: {txid} for row_id, txid in current.items()}
if not result:
return result
rows = await session.execute(
select(PendingTransactionTxid.pending_transaction_id, PendingTransactionTxid.txid).where(
PendingTransactionTxid.pending_transaction_id.in_(list(result))
)
)
for row_id, txid in rows.all():
result[row_id].add(txid)
return result
async def delete_versions(session: AsyncSession, pending_id: int) -> None:
"""The versions go with a `bet` row that a verified refusal deletes (DAT-38)."""
await session.execute(delete(PendingTransactionTxid).where(PendingTransactionTxid.pending_transaction_id == pending_id))