""" Mutable contribution receipt lifecycle control plane. The contribution data path has two very different storage needs: * pending review rows must live in a mutable store so a participant can delete them before promotion and reviewers can atomically claim exactly one promotion; * promoted rows may be copied to append-only/versioned providers, but the receipt lifecycle must remain mutable so a later withdrawal can be represented truthfully without pretending that Git history was physically erased. This module therefore stores only the *control plane*. Pending canonical rows are kept only until promotion/deletion/expiry. After promotion the raw rows are removed from the ledger and only bounded lifecycle metadata, deduplication keys, a digest of the participant delete/withdraw capability, and provider record-path metadata remain. Two backends are bundled: ``memory`` Compatibility/development backend. Process-local and intentionally not durable. ``sqlite`` Local transactional durable backend using the Python standard library. It survives process restarts when its file lives on durable storage and prevents duplicate promotion with transactional state transitions. It is **not** a shared multi-replica database; operators must not represent it as one. """ from __future__ import annotations import asyncio import hashlib import hmac import json import secrets import sqlite3 import time from pathlib import Path from typing import Any from ._redis_security import RedisSecurityError, redis_connection_kwargs _TERMINAL = {"deleted", "expired"} _ACTIVE_PENDING = {"quarantined", "promoting", "promotion_uncertain", "withdrawing"} _MANAGED = { "quarantined", "promoting", "promotion_uncertain", "eligible", "withdrawing", "withdrawal_uncertain", "withdrawn", } class ContributionLedgerError(RuntimeError): """Stable, non-sensitive control-plane error.""" def __init__(self, code: str) -> None: super().__init__(code) self.code = code def _copy_entry(entry: dict[str, Any] | None) -> dict[str, Any] | None: if entry is None: return None # JSON round-trip prevents callers from mutating nested records/storage maps. return json.loads(json.dumps(entry, ensure_ascii=False)) def _now() -> float: return time.time() class MemoryContributionLedger: """Bounded process-local compatibility ledger.""" backend = "memory" durability = "process_local" durable = False shared = False def __init__( self, *, max_pending_entries: int, max_pending_bytes: int, max_receipts: int, terminal_retention_seconds: int = 86_400, ) -> None: self.max_pending_entries = max_pending_entries self.max_pending_bytes = max_pending_bytes self.max_receipts = max_receipts self.terminal_retention_seconds = max(60, int(terminal_retention_seconds)) self.entries: dict[str, dict[str, Any]] = {} self._lock = asyncio.Lock() async def initialize(self) -> None: return None async def close(self) -> None: return None def manifest(self) -> dict[str, Any]: return { "backend": self.backend, "durability": self.durability, "durable": self.durable, "shared": self.shared, } def _sweep_locked(self, now: float) -> None: retire_before = now - self.terminal_retention_seconds retired: list[str] = [] for receipt_id, entry in self.entries.items(): if ( entry.get("state") == "quarantined" and float(entry.get("expiresAt") or 0) <= now ): entry["state"] = "expired" entry["records"] = [] entry["bytes"] = 0 entry["updatedAt"] = now if ( entry.get("state") in _TERMINAL | {"withdrawn"} and float(entry.get("updatedAt") or 0) <= retire_before ): retired.append(receipt_id) for receipt_id in retired: self.entries.pop(receipt_id, None) def _pending_counts_locked(self) -> tuple[int, int]: pending = [ e for e in self.entries.values() if e.get("state") in _ACTIVE_PENDING ] return len(pending), sum(int(e.get("bytes") or 0) for e in pending) async def create(self, entry: dict[str, Any]) -> None: async with self._lock: now = _now() self._sweep_locked(now) if entry["receiptId"] in self.entries: raise ContributionLedgerError("DUPLICATE_RECEIPT") if len(self.entries) >= self.max_receipts: raise ContributionLedgerError("RECEIPT_CAPACITY") count, total = self._pending_counts_locked() if count >= self.max_pending_entries: raise ContributionLedgerError("PENDING_CAPACITY") if total + int(entry.get("bytes") or 0) > self.max_pending_bytes: raise ContributionLedgerError("PENDING_BYTE_CAPACITY") self.entries[entry["receiptId"]] = _copy_entry(entry) or {} async def get(self, receipt_id: str) -> dict[str, Any] | None: async with self._lock: self._sweep_locked(_now()) return _copy_entry(self.entries.get(receipt_id)) async def replace_pending_payload( self, receipt_id: str, *, records: list[dict[str, Any]], byte_count: int, dedup_keys: list[str], payload_digest: str, row_count: int, storage: dict[str, Any] | None = None, ) -> dict[str, Any]: """Replace one quarantined payload without changing receipt authority.""" async with self._lock: now = _now() self._sweep_locked(now) entry = self.entries.get(receipt_id) if entry is None: raise ContributionLedgerError("NOT_FOUND") if entry.get("state") == "expired": raise ContributionLedgerError("EXPIRED") if entry.get("state") != "quarantined": raise ContributionLedgerError("NOT_PENDING") _, total = self._pending_counts_locked() old_bytes = int(entry.get("bytes") or 0) if total - old_bytes + int(byte_count) > self.max_pending_bytes: raise ContributionLedgerError("PENDING_BYTE_CAPACITY") operation = dict(entry.get("operation") or {}) operation["payloadDigest"] = str(payload_digest) operation["reviewRevision"] = int(operation.get("reviewRevision") or 1) + 1 entry["records"] = _copy_entry({"records": records})["records"] entry["bytes"] = int(byte_count) entry["dedupKeys"] = list(dedup_keys) entry["operation"] = operation entry["rowCount"] = int(row_count) if storage is not None: entry["storage"] = _copy_entry(storage) or {} entry["lastError"] = "" entry["updatedAt"] = now return _copy_entry(entry) or {} async def set_pending_storage( self, receipt_id: str, *, storage: dict[str, Any] ) -> dict[str, Any]: """Attach provider-review metadata while the receipt is quarantined.""" async with self._lock: now = _now() self._sweep_locked(now) entry = self.entries.get(receipt_id) if entry is None: raise ContributionLedgerError("NOT_FOUND") if entry.get("state") == "expired": raise ContributionLedgerError("EXPIRED") if entry.get("state") != "quarantined": raise ContributionLedgerError("NOT_PENDING") entry["storage"] = _copy_entry(storage) or {} entry["updatedAt"] = now return _copy_entry(entry) or {} async def begin_promotion(self, receipt_id: str) -> dict[str, Any]: async with self._lock: now = _now() self._sweep_locked(now) entry = self.entries.get(receipt_id) if entry is None: raise ContributionLedgerError("NOT_FOUND") state = entry.get("state") if state == "expired": raise ContributionLedgerError("EXPIRED") if state == "promoting": raise ContributionLedgerError("PROMOTION_IN_PROGRESS") if state != "quarantined": raise ContributionLedgerError("NOT_PENDING") entry["state"] = "promoting" entry["updatedAt"] = now return _copy_entry(entry) or {} async def promotion_failed( self, receipt_id: str, code: str, *, claim_token: str | None = None ) -> None: async with self._lock: entry = self.entries.get(receipt_id) if not entry or entry.get("state") != "promoting": return now = _now() if float(entry.get("expiresAt") or 0) <= now: entry["state"] = "expired" entry["records"] = [] entry["bytes"] = 0 else: entry["state"] = "quarantined" entry["lastError"] = str(code or "PROMOTION_FAILED")[:64] entry["updatedAt"] = now async def mark_promotion_uncertain( self, receipt_id: str, code: str, *, claim_token: str | None = None ) -> dict[str, Any]: async with self._lock: entry = self.entries.get(receipt_id) if entry is None or entry.get("state") != "promoting": raise ContributionLedgerError("PROMOTION_STATE") entry["state"] = "promotion_uncertain" entry["lastError"] = str(code or "PROMOTION_OUTCOME_UNCERTAIN")[:64] entry["updatedAt"] = _now() return _copy_entry(entry) or {} async def mark_promoted( self, receipt_id: str, *, storage: dict[str, Any], claim_token: str | None = None, ) -> dict[str, Any]: async with self._lock: entry = self.entries.get(receipt_id) if entry is None or entry.get("state") != "promoting": raise ContributionLedgerError("PROMOTION_STATE") now = _now() entry["state"] = "eligible" entry["promotedAt"] = now entry["storage"] = _copy_entry(storage) or {} entry["records"] = [] entry["bytes"] = 0 entry["lastError"] = "" entry["updatedAt"] = now return _copy_entry(entry) or {} async def delete_pending(self, receipt_id: str) -> dict[str, Any]: async with self._lock: now = _now() self._sweep_locked(now) entry = self.entries.get(receipt_id) if entry is None: raise ContributionLedgerError("NOT_FOUND") state = entry.get("state") if state == "expired": raise ContributionLedgerError("EXPIRED") if state in {"promoting", "withdrawing"}: raise ContributionLedgerError("BUSY") if state != "quarantined": raise ContributionLedgerError("NOT_PENDING") entry["state"] = "deleted" entry["records"] = [] entry["bytes"] = 0 entry["deletedAt"] = now entry["updatedAt"] = now return _copy_entry(entry) or {} async def begin_withdrawal(self, receipt_id: str) -> dict[str, Any]: async with self._lock: entry = self.entries.get(receipt_id) if entry is None: raise ContributionLedgerError("NOT_FOUND") state = entry.get("state") if state == "withdrawn": return _copy_entry(entry) or {} if state == "withdrawing": raise ContributionLedgerError("WITHDRAWAL_IN_PROGRESS") if state not in {"eligible", "promotion_uncertain", "withdrawal_uncertain"}: raise ContributionLedgerError("NOT_ELIGIBLE") entry["state"] = "withdrawing" entry["updatedAt"] = _now() return _copy_entry(entry) or {} async def withdrawal_failed( self, receipt_id: str, code: str, *, claim_token: str | None = None ) -> None: async with self._lock: entry = self.entries.get(receipt_id) if not entry or entry.get("state") != "withdrawing": return entry["state"] = ( "eligible" if entry.get("promotedAt") else "promotion_uncertain" ) entry["lastError"] = str(code or "WITHDRAWAL_FAILED")[:64] entry["updatedAt"] = _now() async def mark_withdrawn( self, receipt_id: str, *, withdrawal_storage: dict[str, Any], current_view_removal: dict[str, str], claim_token: str | None = None, ) -> dict[str, Any]: async with self._lock: entry = self.entries.get(receipt_id) if entry is None or entry.get("state") != "withdrawing": raise ContributionLedgerError("WITHDRAWAL_STATE") now = _now() entry["state"] = "withdrawn" entry["records"] = [] entry["bytes"] = 0 entry["withdrawnAt"] = now entry["withdrawalStorage"] = _copy_entry(withdrawal_storage) or {} entry["currentViewRemoval"] = dict(current_view_removal) entry["lastError"] = "" entry["updatedAt"] = now return _copy_entry(entry) or {} def clear_for_tests(self) -> None: self.entries.clear() class SQLiteContributionLedger: """ Local ACID receipt ledger backed by SQLite. SQLite is transactional and process-restart durable when the configured file resides on durable storage. It is deliberately advertised as *local*, not shared/distributed, so a multi-replica deployment cannot accidentally claim one authoritative review ledger. """ backend = "sqlite" durability = "local_transactional" durable = True shared = False def __init__( self, path: str, *, max_pending_entries: int, max_pending_bytes: int, max_receipts: int, terminal_retention_seconds: int = 86_400, ) -> None: self.path = str(Path(path).expanduser()) self.max_pending_entries = max_pending_entries self.max_pending_bytes = max_pending_bytes self.max_receipts = max_receipts self.terminal_retention_seconds = max(60, int(terminal_retention_seconds)) self._lock = asyncio.Lock() def manifest(self) -> dict[str, Any]: # Do not expose the filesystem path in public discovery/logs. return { "backend": self.backend, "durability": self.durability, "durable": self.durable, "shared": self.shared, } def _connect(self) -> sqlite3.Connection: conn = sqlite3.connect(self.path, timeout=5.0) conn.row_factory = sqlite3.Row conn.execute("PRAGMA busy_timeout=5000") # Defense in depth for sensitive pending rows. This reduces forensic # remnants in ordinary SQLite table pages; it is not a global erasure # guarantee because WAL/filesystem snapshots/backups may exist. conn.execute("PRAGMA secure_delete=ON") conn.execute("PRAGMA journal_size_limit=0") return conn def _init_sync(self) -> None: parent = Path(self.path).parent parent.mkdir(parents=True, exist_ok=True) conn = self._connect() try: conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA synchronous=FULL") conn.execute(""" CREATE TABLE IF NOT EXISTS contribution_receipts ( receipt_id TEXT PRIMARY KEY, state TEXT NOT NULL, records_json TEXT NOT NULL DEFAULT '[]', bytes INTEGER NOT NULL DEFAULT 0, delete_token_hash TEXT NOT NULL, expires_at REAL NOT NULL, received_at REAL NOT NULL, promoted_at REAL, withdrawn_at REAL, deleted_at REAL, dedup_keys_json TEXT NOT NULL DEFAULT '[]', storage_json TEXT NOT NULL DEFAULT '{}', withdrawal_storage_json TEXT NOT NULL DEFAULT '{}', current_view_removal_json TEXT NOT NULL DEFAULT '{}', last_error TEXT NOT NULL DEFAULT '', operation_json TEXT NOT NULL DEFAULT '{}', row_count INTEGER NOT NULL DEFAULT 0, updated_at REAL NOT NULL ) """) # Additive migration for pre-Run-18 ledgers. Existing receipt # lifecycle state is preserved; only recovery metadata is new. _columns = { row[1] for row in conn.execute( "PRAGMA table_info(contribution_receipts)" ).fetchall() } if "operation_json" not in _columns: conn.execute( "ALTER TABLE contribution_receipts ADD COLUMN operation_json TEXT NOT NULL DEFAULT '{}'" ) if "row_count" not in _columns: conn.execute( "ALTER TABLE contribution_receipts ADD COLUMN row_count INTEGER NOT NULL DEFAULT 0" ) # Any transient operation state present during startup belongs to a # previous process. Promotion is replay-safe because the app writes # the reviewed payload to a receipt-stable provider path derived from # receivedAt. Withdrawal tombstones are also idempotent under dataset # last-write-wins, so both states can be reclaimed rather than left # permanently BUSY after a crash. now = _now() conn.execute( """UPDATE contribution_receipts SET state=CASE WHEN expires_at <= ? THEN 'expired' ELSE 'quarantined' END, records_json=CASE WHEN expires_at <= ? THEN '[]' ELSE records_json END, bytes=CASE WHEN expires_at <= ? THEN 0 ELSE bytes END, last_error='RECOVERED_AFTER_RESTART',updated_at=? WHERE state='promoting'""", (now, now, now, now), ) conn.execute( """UPDATE contribution_receipts SET state=CASE WHEN promoted_at IS NULL THEN 'promotion_uncertain' ELSE 'eligible' END,last_error='RECOVERED_AFTER_RESTART',updated_at=? WHERE state='withdrawing'""", (now,), ) self._sweep_sync(conn, now) conn.commit() self._checkpoint_sensitive(conn) finally: conn.close() async def initialize(self) -> None: await asyncio.to_thread(self._init_sync) async def close(self) -> None: return None @staticmethod def _row_to_entry(row: sqlite3.Row | None) -> dict[str, Any] | None: if row is None: return None return { "receiptId": row["receipt_id"], "state": row["state"], "records": json.loads(row["records_json"] or "[]"), "bytes": int(row["bytes"] or 0), "deleteTokenHash": row["delete_token_hash"], "expiresAt": float(row["expires_at"] or 0), "receivedAt": float(row["received_at"] or 0), "promotedAt": row["promoted_at"], "withdrawnAt": row["withdrawn_at"], "deletedAt": row["deleted_at"], "dedupKeys": json.loads(row["dedup_keys_json"] or "[]"), "storage": json.loads(row["storage_json"] or "{}"), "withdrawalStorage": json.loads(row["withdrawal_storage_json"] or "{}"), "currentViewRemoval": json.loads(row["current_view_removal_json"] or "{}"), "lastError": row["last_error"] or "", "operation": json.loads(row["operation_json"] or "{}"), "rowCount": int(row["row_count"] or 0), "updatedAt": float(row["updated_at"] or 0), } def _sweep_sync(self, conn: sqlite3.Connection, now: float) -> None: conn.execute( """UPDATE contribution_receipts SET state='expired', records_json='[]', bytes=0, updated_at=? WHERE state='quarantined' AND expires_at <= ?""", (now, now), ) # Terminal lifecycle tombstones are useful for a bounded status window, # but keeping them forever turns max_receipts into a permanent denial of # future intake. Eligible receipts are intentionally retained until the # participant withdraws or an external control-plane policy supersedes # this single-instance backend. conn.execute( """DELETE FROM contribution_receipts WHERE state IN ('deleted','expired','withdrawn') AND updated_at <= ?""", (now - self.terminal_retention_seconds,), ) @staticmethod def _checkpoint_sensitive(conn: sqlite3.Connection) -> None: """Best-effort truncate WAL after content-clearing lifecycle writes.""" try: # ruff: ignore[suppressible-exception] conn.execute("PRAGMA wal_checkpoint(TRUNCATE)") except sqlite3.DatabaseError: # Checkpoint may be busy when another process/connection is active. # The lifecycle transaction is already committed; never reinterpret # a checkpoint limitation as proof that the user content was erased. pass async def create(self, entry: dict[str, Any]) -> None: async with self._lock: def _op() -> None: conn = self._connect() try: conn.execute("BEGIN IMMEDIATE") now = _now() self._sweep_sync(conn, now) total_rows = int( conn.execute( "SELECT COUNT(*) FROM contribution_receipts" ).fetchone()[0] ) if total_rows >= self.max_receipts: raise ContributionLedgerError("RECEIPT_CAPACITY") pending_count, pending_bytes = conn.execute( "SELECT COUNT(*), COALESCE(SUM(bytes), 0) FROM contribution_receipts WHERE state IN ('quarantined','promoting','promotion_uncertain','withdrawing')" ).fetchone() if int(pending_count) >= self.max_pending_entries: raise ContributionLedgerError("PENDING_CAPACITY") if ( int(pending_bytes) + int(entry.get("bytes") or 0) > self.max_pending_bytes ): raise ContributionLedgerError("PENDING_BYTE_CAPACITY") conn.execute( """INSERT INTO contribution_receipts (receipt_id,state,records_json,bytes,delete_token_hash,expires_at,received_at, dedup_keys_json,operation_json,row_count,updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?)""", ( entry["receiptId"], entry["state"], json.dumps( entry.get("records") or [], ensure_ascii=False, separators=(",", ":"), ), int(entry.get("bytes") or 0), entry["deleteTokenHash"], float(entry["expiresAt"]), float(entry["receivedAt"]), json.dumps( entry.get("dedupKeys") or [], separators=(",", ":") ), json.dumps( entry.get("operation") or {}, separators=(",", ":") ), int(entry.get("rowCount") or 0), now, ), ) conn.commit() except sqlite3.IntegrityError as exc: conn.rollback() raise ContributionLedgerError("DUPLICATE_RECEIPT") from exc except Exception: conn.rollback() raise finally: conn.close() await asyncio.to_thread(_op) async def get(self, receipt_id: str) -> dict[str, Any] | None: async with self._lock: def _op() -> dict[str, Any] | None: conn = self._connect() try: conn.execute("BEGIN IMMEDIATE") self._sweep_sync(conn, _now()) row = conn.execute( "SELECT * FROM contribution_receipts WHERE receipt_id=?", (receipt_id,), ).fetchone() conn.commit() self._checkpoint_sensitive(conn) return self._row_to_entry(row) finally: conn.close() return await asyncio.to_thread(_op) async def _transition( self, receipt_id: str, *, allowed: set[str], to_state: str, busy_code: str | None = None, ) -> dict[str, Any]: async with self._lock: def _op() -> dict[str, Any]: conn = self._connect() try: conn.execute("BEGIN IMMEDIATE") now = _now() self._sweep_sync(conn, now) row = conn.execute( "SELECT * FROM contribution_receipts WHERE receipt_id=?", (receipt_id,), ).fetchone() entry = self._row_to_entry(row) if entry is None: raise ContributionLedgerError("NOT_FOUND") state = str(entry["state"]) if state == "expired": raise ContributionLedgerError("EXPIRED") if state not in allowed: if busy_code and state in {"promoting", "withdrawing"}: raise ContributionLedgerError(busy_code) raise ContributionLedgerError( "NOT_PENDING" if to_state == "promoting" else "NOT_ELIGIBLE" ) conn.execute( "UPDATE contribution_receipts SET state=?, updated_at=? WHERE receipt_id=?", (to_state, now, receipt_id), ) row = conn.execute( "SELECT * FROM contribution_receipts WHERE receipt_id=?", (receipt_id,), ).fetchone() conn.commit() return self._row_to_entry(row) or {} except Exception: conn.rollback() raise finally: conn.close() return await asyncio.to_thread(_op) async def replace_pending_payload( self, receipt_id: str, *, records: list[dict[str, Any]], byte_count: int, dedup_keys: list[str], payload_digest: str, row_count: int, storage: dict[str, Any] | None = None, ) -> dict[str, Any]: async with self._lock: def _op() -> dict[str, Any]: conn = self._connect() try: conn.execute("BEGIN IMMEDIATE") now = _now() self._sweep_sync(conn, now) row = conn.execute( "SELECT * FROM contribution_receipts WHERE receipt_id=?", (receipt_id,), ).fetchone() entry = self._row_to_entry(row) if entry is None: raise ContributionLedgerError("NOT_FOUND") if entry.get("state") == "expired": raise ContributionLedgerError("EXPIRED") if entry.get("state") != "quarantined": raise ContributionLedgerError("NOT_PENDING") pending_bytes = int( conn.execute( "SELECT COALESCE(SUM(bytes), 0) FROM contribution_receipts WHERE state IN ('quarantined','promoting','promotion_uncertain','withdrawing')" ).fetchone()[0] ) if ( pending_bytes - int(entry.get("bytes") or 0) + int(byte_count) > self.max_pending_bytes ): raise ContributionLedgerError("PENDING_BYTE_CAPACITY") operation = dict(entry.get("operation") or {}) operation["payloadDigest"] = str(payload_digest) operation["reviewRevision"] = ( int(operation.get("reviewRevision") or 1) + 1 ) storage_json = ( json.dumps(storage, separators=(",", ":")) if storage is not None else json.dumps( entry.get("storage") or {}, separators=(",", ":") ) ) conn.execute( """UPDATE contribution_receipts SET records_json=?,bytes=?,dedup_keys_json=?,operation_json=?,row_count=?,storage_json=?,last_error='',updated_at=? WHERE receipt_id=?""", ( json.dumps( records, ensure_ascii=False, separators=(",", ":") ), int(byte_count), json.dumps(dedup_keys, separators=(",", ":")), json.dumps(operation, separators=(",", ":")), int(row_count), storage_json, now, receipt_id, ), ) row = conn.execute( "SELECT * FROM contribution_receipts WHERE receipt_id=?", (receipt_id,), ).fetchone() conn.commit() return self._row_to_entry(row) or {} except Exception: conn.rollback() raise finally: conn.close() return await asyncio.to_thread(_op) async def set_pending_storage( self, receipt_id: str, *, storage: dict[str, Any] ) -> dict[str, Any]: """Persist provider-review metadata without changing lifecycle state.""" async with self._lock: def _op() -> dict[str, Any]: conn = self._connect() try: conn.execute("BEGIN IMMEDIATE") self._sweep_sync(conn, _now()) row = conn.execute( "SELECT state FROM contribution_receipts WHERE receipt_id=?", (receipt_id,), ).fetchone() if row is None: raise ContributionLedgerError("NOT_FOUND") if row["state"] == "expired": raise ContributionLedgerError("EXPIRED") if row["state"] != "quarantined": raise ContributionLedgerError("NOT_PENDING") now = _now() conn.execute( "UPDATE contribution_receipts SET storage_json=?,updated_at=? WHERE receipt_id=?", (json.dumps(storage, separators=(",", ":")), now, receipt_id), ) row = conn.execute( "SELECT * FROM contribution_receipts WHERE receipt_id=?", (receipt_id,), ).fetchone() conn.commit() self._checkpoint_sensitive(conn) return self._row_to_entry(row) or {} except Exception: conn.rollback() raise finally: conn.close() return await asyncio.to_thread(_op) async def begin_promotion(self, receipt_id: str) -> dict[str, Any]: return await self._transition( receipt_id, allowed={"quarantined"}, to_state="promoting", busy_code="PROMOTION_IN_PROGRESS", ) async def promotion_failed( self, receipt_id: str, code: str, *, claim_token: str | None = None ) -> None: async with self._lock: def _op() -> None: conn = self._connect() try: conn.execute("BEGIN IMMEDIATE") row = conn.execute( "SELECT state,expires_at FROM contribution_receipts WHERE receipt_id=?", (receipt_id,), ).fetchone() if row and row["state"] == "promoting": now = _now() if float(row["expires_at"] or 0) <= now: conn.execute( "UPDATE contribution_receipts SET state='expired',records_json='[]',bytes=0,last_error=?,updated_at=? WHERE receipt_id=?", (str(code or "PROMOTION_FAILED")[:64], now, receipt_id), ) else: conn.execute( "UPDATE contribution_receipts SET state='quarantined',last_error=?,updated_at=? WHERE receipt_id=?", (str(code or "PROMOTION_FAILED")[:64], now, receipt_id), ) conn.commit() self._checkpoint_sensitive(conn) finally: conn.close() await asyncio.to_thread(_op) async def mark_promotion_uncertain( self, receipt_id: str, code: str, *, claim_token: str | None = None ) -> dict[str, Any]: async with self._lock: def _op() -> dict[str, Any]: conn = self._connect() try: conn.execute("BEGIN IMMEDIATE") row = conn.execute( "SELECT state FROM contribution_receipts WHERE receipt_id=?", (receipt_id,), ).fetchone() if row is None or row["state"] != "promoting": raise ContributionLedgerError("PROMOTION_STATE") now = _now() conn.execute( "UPDATE contribution_receipts SET state='promotion_uncertain',last_error=?,updated_at=? WHERE receipt_id=?", ( str(code or "PROMOTION_OUTCOME_UNCERTAIN")[:64], now, receipt_id, ), ) row = conn.execute( "SELECT * FROM contribution_receipts WHERE receipt_id=?", (receipt_id,), ).fetchone() conn.commit() return self._row_to_entry(row) or {} except Exception: conn.rollback() raise finally: conn.close() return await asyncio.to_thread(_op) async def mark_promoted( self, receipt_id: str, *, storage: dict[str, Any], claim_token: str | None = None, ) -> dict[str, Any]: async with self._lock: def _op() -> dict[str, Any]: conn = self._connect() try: conn.execute("BEGIN IMMEDIATE") row = conn.execute( "SELECT state FROM contribution_receipts WHERE receipt_id=?", (receipt_id,), ).fetchone() if row is None or row["state"] != "promoting": raise ContributionLedgerError("PROMOTION_STATE") now = _now() conn.execute( """UPDATE contribution_receipts SET state='eligible',records_json='[]',bytes=0,promoted_at=?,storage_json=?,last_error='',updated_at=? WHERE receipt_id=?""", ( now, json.dumps(storage, separators=(",", ":")), now, receipt_id, ), ) row = conn.execute( "SELECT * FROM contribution_receipts WHERE receipt_id=?", (receipt_id,), ).fetchone() conn.commit() self._checkpoint_sensitive(conn) return self._row_to_entry(row) or {} except Exception: conn.rollback() raise finally: conn.close() return await asyncio.to_thread(_op) async def delete_pending(self, receipt_id: str) -> dict[str, Any]: async with self._lock: def _op() -> dict[str, Any]: conn = self._connect() try: conn.execute("BEGIN IMMEDIATE") now = _now() self._sweep_sync(conn, now) row = conn.execute( "SELECT * FROM contribution_receipts WHERE receipt_id=?", (receipt_id,), ).fetchone() entry = self._row_to_entry(row) if entry is None: raise ContributionLedgerError("NOT_FOUND") state = entry["state"] if state == "expired": raise ContributionLedgerError("EXPIRED") if state in {"promoting", "withdrawing"}: raise ContributionLedgerError("BUSY") if state != "quarantined": raise ContributionLedgerError("NOT_PENDING") conn.execute( """UPDATE contribution_receipts SET state='deleted',records_json='[]',bytes=0,deleted_at=?,updated_at=? WHERE receipt_id=?""", (now, now, receipt_id), ) row = conn.execute( "SELECT * FROM contribution_receipts WHERE receipt_id=?", (receipt_id,), ).fetchone() conn.commit() self._checkpoint_sensitive(conn) return self._row_to_entry(row) or {} except Exception: conn.rollback() raise finally: conn.close() return await asyncio.to_thread(_op) async def begin_withdrawal(self, receipt_id: str) -> dict[str, Any]: current = await self.get(receipt_id) if current and current.get("state") == "withdrawn": return current return await self._transition( receipt_id, allowed={"eligible", "promotion_uncertain", "withdrawal_uncertain"}, to_state="withdrawing", busy_code="WITHDRAWAL_IN_PROGRESS", ) async def withdrawal_failed( self, receipt_id: str, code: str, *, claim_token: str | None = None ) -> None: async with self._lock: def _op() -> None: conn = self._connect() try: now = _now() conn.execute( """UPDATE contribution_receipts SET state=CASE WHEN promoted_at IS NULL THEN 'promotion_uncertain' ELSE 'eligible' END,last_error=?,updated_at=? WHERE receipt_id=? AND state='withdrawing'""", (str(code or "WITHDRAWAL_FAILED")[:64], now, receipt_id), ) conn.commit() finally: conn.close() await asyncio.to_thread(_op) async def mark_withdrawn( self, receipt_id: str, *, withdrawal_storage: dict[str, Any], current_view_removal: dict[str, str], claim_token: str | None = None, ) -> dict[str, Any]: async with self._lock: def _op() -> dict[str, Any]: conn = self._connect() try: conn.execute("BEGIN IMMEDIATE") row = conn.execute( "SELECT state FROM contribution_receipts WHERE receipt_id=?", (receipt_id,), ).fetchone() if row is None or row["state"] != "withdrawing": raise ContributionLedgerError("WITHDRAWAL_STATE") now = _now() conn.execute( """UPDATE contribution_receipts SET state='withdrawn',records_json='[]',bytes=0,withdrawn_at=?,withdrawal_storage_json=?,current_view_removal_json=?,last_error='',updated_at=? WHERE receipt_id=?""", ( now, json.dumps(withdrawal_storage, separators=(",", ":")), json.dumps(current_view_removal, separators=(",", ":")), now, receipt_id, ), ) row = conn.execute( "SELECT * FROM contribution_receipts WHERE receipt_id=?", (receipt_id,), ).fetchone() conn.commit() return self._row_to_entry(row) or {} except Exception: conn.rollback() raise finally: conn.close() return await asyncio.to_thread(_op) # Redis scripts intentionally keep all index keys in one ``{contribution}`` # hash slot. This makes the lifecycle operations compatible with a Redis # Cluster consistency domain without scattering one receipt transition across # slots. Receipt identifiers are HMACed before becoming Redis key material. _REDIS_CREATE_LUA = r""" local now = tonumber(ARGV[1]) local member = ARGV[2] local payload = ARGV[3] local expires_at = tonumber(ARGV[4]) local live_until = tonumber(ARGV[5]) local max_receipts = tonumber(ARGV[6]) local max_pending = tonumber(ARGV[7]) local max_bytes = tonumber(ARGV[8]) local bytes = tonumber(ARGV[9]) local ttl = tonumber(ARGV[10]) redis.call('ZREMRANGEBYSCORE', KEYS[1], '-inf', now) local expired = redis.call('ZRANGEBYSCORE', KEYS[2], '-inf', now) if #expired > 0 then redis.call('ZREM', KEYS[2], unpack(expired)) redis.call('HDEL', KEYS[3], unpack(expired)) end if redis.call('EXISTS', KEYS[4]) == 1 then return {0, 'DUPLICATE_RECEIPT'} end if redis.call('ZCARD', KEYS[1]) >= max_receipts then return {0, 'RECEIPT_CAPACITY'} end if redis.call('ZCARD', KEYS[2]) >= max_pending then return {0, 'PENDING_CAPACITY'} end local values = redis.call('HVALS', KEYS[3]) local pending_bytes = 0 for _, value in ipairs(values) do pending_bytes = pending_bytes + tonumber(value) end if pending_bytes + bytes > max_bytes then return {0, 'PENDING_BYTE_CAPACITY'} end local created = redis.call('SET', KEYS[4], payload, 'EX', ttl, 'NX') if not created then return {0, 'DUPLICATE_RECEIPT'} end redis.call('ZADD', KEYS[1], live_until, member) redis.call('ZADD', KEYS[2], expires_at, member) redis.call('HSET', KEYS[3], member, bytes) return {1, payload} """.strip() _REDIS_GET_LUA = r""" local now = tonumber(ARGV[1]) local member = ARGV[2] local terminal_retention = tonumber(ARGV[3]) local immortal = tonumber(ARGV[4]) local raw = redis.call('GET', KEYS[4]) if not raw then redis.call('ZREM', KEYS[1], member) redis.call('ZREM', KEYS[2], member) redis.call('HDEL', KEYS[3], member) return {1, ''} end local entry = cjson.decode(raw) local state = tostring(entry.state or '') local expires_at = tonumber(entry.expiresAt or 0) local lease_until = tonumber(entry.operationLeaseUntil or 0) if state == 'quarantined' and expires_at <= now then entry.state = 'expired'; entry.records = {}; entry.bytes = 0 entry.lastError = ''; entry.updatedAt = now entry.operationClaimHash = ''; entry.operationLeaseUntil = 0 raw = cjson.encode(entry) redis.call('SET', KEYS[4], raw, 'EX', terminal_retention) redis.call('ZREM', KEYS[2], member); redis.call('HDEL', KEYS[3], member) redis.call('ZADD', KEYS[1], now + terminal_retention, member) elseif state == 'promoting' and lease_until > 0 and lease_until <= now then entry.state = 'promotion_uncertain'; entry.lastError = 'CLAIM_EXPIRED_RECONCILIATION_REQUIRED'; entry.updatedAt = now entry.operationClaimHash = ''; entry.operationLeaseUntil = 0 raw = cjson.encode(entry) redis.call('SET', KEYS[4], raw); redis.call('PERSIST', KEYS[4]) redis.call('ZADD', KEYS[1], immortal, member); redis.call('ZADD', KEYS[2], immortal, member) redis.call('HSET', KEYS[3], member, tonumber(entry.bytes or 0)) elseif state == 'withdrawing' and lease_until > 0 and lease_until <= now then entry.state = 'withdrawal_uncertain'; entry.lastError = 'CLAIM_EXPIRED_RECONCILIATION_REQUIRED'; entry.updatedAt = now entry.operationClaimHash = ''; entry.operationLeaseUntil = 0 raw = cjson.encode(entry) redis.call('SET', KEYS[4], raw); redis.call('PERSIST', KEYS[4]); redis.call('ZADD', KEYS[1], immortal, member) end return {1, raw} """.strip() _REDIS_REPLACE_PENDING_LUA = r""" local now=tonumber(ARGV[1]); local member=ARGV[2]; local records_json=ARGV[3] local new_bytes=tonumber(ARGV[4]); local dedup_json=ARGV[5]; local payload_digest=ARGV[6] local row_count=tonumber(ARGV[7]); local max_bytes=tonumber(ARGV[8]); local terminal_retention=tonumber(ARGV[9]); local storage_json=ARGV[10] local raw=redis.call('GET', KEYS[4]); if not raw then return {0, 'NOT_FOUND'} end local entry=cjson.decode(raw); local state=tostring(entry.state or ''); local expires_at=tonumber(entry.expiresAt or 0) if state == 'quarantined' and expires_at <= now then entry.state='expired'; entry.records={}; entry.bytes=0; entry.updatedAt=now raw=cjson.encode(entry); redis.call('SET', KEYS[4], raw, 'EX', terminal_retention) redis.call('ZREM', KEYS[2], member); redis.call('HDEL', KEYS[3], member); redis.call('ZADD', KEYS[1], now+terminal_retention, member) return {0, 'EXPIRED'} end if state ~= 'quarantined' then return {0, 'NOT_PENDING'} end local values=redis.call('HVALS', KEYS[3]); local total=0 for _, value in ipairs(values) do total=total+tonumber(value) end local old_bytes=tonumber(entry.bytes or 0) if total-old_bytes+new_bytes > max_bytes then return {0, 'PENDING_BYTE_CAPACITY'} end entry.records=cjson.decode(records_json); entry.bytes=new_bytes; entry.dedupKeys=cjson.decode(dedup_json); entry.rowCount=row_count local op=entry.operation or {}; op.payloadDigest=payload_digest; op.reviewRevision=tonumber(op.reviewRevision or 1)+1; entry.operation=op if storage_json ~= '' then entry.storage=cjson.decode(storage_json) end entry.lastError=''; entry.updatedAt=now raw=cjson.encode(entry); redis.call('SET', KEYS[4], raw, 'KEEPTTL'); redis.call('HSET', KEYS[3], member, new_bytes) return {1, raw} """.strip() _REDIS_SET_PENDING_STORAGE_LUA = r""" local now=tonumber(ARGV[1]); local member=ARGV[2]; local storage_json=ARGV[3]; local terminal_retention=tonumber(ARGV[4]) local raw=redis.call('GET', KEYS[4]); if not raw then return {0, 'NOT_FOUND'} end local entry=cjson.decode(raw); local state=tostring(entry.state or ''); local expires_at=tonumber(entry.expiresAt or 0) if state == 'quarantined' and expires_at <= now then entry.state='expired'; entry.records={}; entry.bytes=0; entry.updatedAt=now raw=cjson.encode(entry); redis.call('SET', KEYS[4], raw, 'EX', terminal_retention) redis.call('ZREM', KEYS[2], member); redis.call('HDEL', KEYS[3], member); redis.call('ZADD', KEYS[1], now+terminal_retention, member) return {0, 'EXPIRED'} end if state ~= 'quarantined' then return {0, 'NOT_PENDING'} end entry.storage=cjson.decode(storage_json); entry.updatedAt=now raw=cjson.encode(entry); redis.call('SET', KEYS[4], raw, 'KEEPTTL') return {1, raw} """.strip() _REDIS_BEGIN_PROMOTION_LUA = r""" local now = tonumber(ARGV[1]); local member = ARGV[2]; local claim_hash = ARGV[3] local lease_until = tonumber(ARGV[4]); local terminal_retention = tonumber(ARGV[5]) local raw = redis.call('GET', KEYS[4]); if not raw then return {0, 'NOT_FOUND'} end local entry = cjson.decode(raw); local state = tostring(entry.state or '') local expires_at = tonumber(entry.expiresAt or 0); local old_lease = tonumber(entry.operationLeaseUntil or 0) if state == 'promoting' and old_lease > now then return {0, 'PROMOTION_IN_PROGRESS'} end if state == 'promoting' and old_lease <= now then entry.state='promotion_uncertain'; entry.lastError='CLAIM_EXPIRED_RECONCILIATION_REQUIRED' entry.operationClaimHash=''; entry.operationLeaseUntil=0; entry.updatedAt=now raw=cjson.encode(entry); redis.call('SET', KEYS[4], raw); redis.call('PERSIST', KEYS[4]) redis.call('ZADD', KEYS[1], 253402300799, member); redis.call('ZADD', KEYS[2], 253402300799, member) redis.call('HSET', KEYS[3], member, tonumber(entry.bytes or 0)) return {0, 'RECONCILIATION_REQUIRED'} end if state == 'quarantined' and expires_at <= now then entry.state='expired'; entry.records={}; entry.bytes=0; entry.updatedAt=now entry.operationClaimHash=''; entry.operationLeaseUntil=0 raw=cjson.encode(entry); redis.call('SET', KEYS[4], raw, 'EX', terminal_retention) redis.call('ZREM', KEYS[2], member); redis.call('HDEL', KEYS[3], member) redis.call('ZADD', KEYS[1], now + terminal_retention, member) return {0, 'EXPIRED'} end if state ~= 'quarantined' then return {0, 'NOT_PENDING'} end entry.state='promoting'; entry.operationClaimHash=claim_hash; entry.operationLeaseUntil=lease_until entry.lastError=''; entry.updatedAt=now raw=cjson.encode(entry); redis.call('SET', KEYS[4], raw, 'KEEPTTL') local pending_until=expires_at; if lease_until > pending_until then pending_until=lease_until end redis.call('ZADD', KEYS[2], pending_until, member); redis.call('HSET', KEYS[3], member, tonumber(entry.bytes or 0)) return {1, raw} """.strip() _REDIS_PROMOTION_FAILED_LUA = r""" local now=tonumber(ARGV[1]); local member=ARGV[2]; local claim_hash=ARGV[3]; local code=ARGV[4] local terminal_retention=tonumber(ARGV[5]) local raw=redis.call('GET', KEYS[4]); if not raw then return {1, ''} end local entry=cjson.decode(raw) if tostring(entry.state or '') ~= 'promoting' or tostring(entry.operationClaimHash or '') ~= claim_hash then return {1, raw} end local expires_at=tonumber(entry.expiresAt or 0) entry.operationClaimHash=''; entry.operationLeaseUntil=0; entry.lastError=string.sub(code,1,64); entry.updatedAt=now if expires_at <= now then entry.state='expired'; entry.records={}; entry.bytes=0 raw=cjson.encode(entry); redis.call('SET', KEYS[4], raw, 'EX', terminal_retention) redis.call('ZREM', KEYS[2], member); redis.call('HDEL', KEYS[3], member) redis.call('ZADD', KEYS[1], now + terminal_retention, member) else entry.state='quarantined'; raw=cjson.encode(entry); redis.call('SET', KEYS[4], raw, 'KEEPTTL') redis.call('ZADD', KEYS[2], expires_at, member); redis.call('HSET', KEYS[3], member, tonumber(entry.bytes or 0)) end return {1, raw} """.strip() _REDIS_MARK_PROMOTION_UNCERTAIN_LUA = r""" local now=tonumber(ARGV[1]); local member=ARGV[2]; local claim_hash=ARGV[3]; local code=ARGV[4] local immortal=tonumber(ARGV[5]); local raw=redis.call('GET', KEYS[4]); if not raw then return {0, 'NOT_FOUND'} end local entry=cjson.decode(raw) if tostring(entry.state or '') ~= 'promoting' then return {0, 'PROMOTION_STATE'} end if tostring(entry.operationClaimHash or '') ~= claim_hash then return {0, 'STALE_CLAIM'} end entry.state='promotion_uncertain'; entry.lastError=string.sub(code,1,64); entry.updatedAt=now entry.operationClaimHash=''; entry.operationLeaseUntil=0 raw=cjson.encode(entry); redis.call('SET', KEYS[4], raw); redis.call('PERSIST', KEYS[4]) redis.call('ZADD', KEYS[1], immortal, member); redis.call('ZADD', KEYS[2], immortal, member) redis.call('HSET', KEYS[3], member, tonumber(entry.bytes or 0)) return {1, raw} """.strip() _REDIS_MARK_PROMOTED_LUA = r""" local now=tonumber(ARGV[1]); local member=ARGV[2]; local claim_hash=ARGV[3]; local storage_json=ARGV[4] local immortal=tonumber(ARGV[5]); local raw=redis.call('GET', KEYS[4]); if not raw then return {0, 'NOT_FOUND'} end local entry=cjson.decode(raw) if tostring(entry.state or '') ~= 'promoting' then return {0, 'PROMOTION_STATE'} end if tostring(entry.operationClaimHash or '') ~= claim_hash then return {0, 'STALE_CLAIM'} end entry.state='eligible'; entry.records={}; entry.bytes=0; entry.promotedAt=now; entry.storage=cjson.decode(storage_json) entry.lastError=''; entry.updatedAt=now; entry.operationClaimHash=''; entry.operationLeaseUntil=0 raw=cjson.encode(entry); redis.call('SET', KEYS[4], raw); redis.call('PERSIST', KEYS[4]) redis.call('ZREM', KEYS[2], member); redis.call('HDEL', KEYS[3], member); redis.call('ZADD', KEYS[1], immortal, member) return {1, raw} """.strip() _REDIS_DELETE_PENDING_LUA = r""" local now=tonumber(ARGV[1]); local member=ARGV[2]; local terminal_retention=tonumber(ARGV[3]) local raw=redis.call('GET', KEYS[4]); if not raw then return {0, 'NOT_FOUND'} end local entry=cjson.decode(raw); local state=tostring(entry.state or '') local expires_at=tonumber(entry.expiresAt or 0); local lease_until=tonumber(entry.operationLeaseUntil or 0) if state == 'promoting' and lease_until > now then return {0, 'BUSY'} end if state == 'promoting' and lease_until <= now then entry.state='promotion_uncertain'; entry.lastError='CLAIM_EXPIRED_RECONCILIATION_REQUIRED'; entry.updatedAt=now entry.operationClaimHash=''; entry.operationLeaseUntil=0 raw=cjson.encode(entry); redis.call('SET', KEYS[4], raw) return {0, 'RECONCILIATION_REQUIRED'} end if state == 'quarantined' and expires_at <= now then entry.state='expired'; entry.records={}; entry.bytes=0; entry.updatedAt=now entry.operationClaimHash=''; entry.operationLeaseUntil=0 raw=cjson.encode(entry); redis.call('SET', KEYS[4], raw, 'EX', terminal_retention) redis.call('ZREM', KEYS[2], member); redis.call('HDEL', KEYS[3], member); redis.call('ZADD', KEYS[1], now+terminal_retention, member) return {0, 'EXPIRED'} end if state == 'withdrawing' then return {0, 'BUSY'} end if state ~= 'quarantined' then return {0, 'NOT_PENDING'} end entry.state='deleted'; entry.records={}; entry.bytes=0; entry.deletedAt=now; entry.updatedAt=now entry.operationClaimHash=''; entry.operationLeaseUntil=0 raw=cjson.encode(entry); redis.call('SET', KEYS[4], raw, 'EX', terminal_retention) redis.call('ZREM', KEYS[2], member); redis.call('HDEL', KEYS[3], member); redis.call('ZADD', KEYS[1], now+terminal_retention, member) return {1, raw} """.strip() _REDIS_BEGIN_WITHDRAWAL_LUA = r""" local now=tonumber(ARGV[1]); local member=ARGV[2]; local claim_hash=ARGV[3]; local lease_until=tonumber(ARGV[4]) local immortal=tonumber(ARGV[5]); local raw=redis.call('GET', KEYS[4]); if not raw then return {0, 'NOT_FOUND'} end local entry=cjson.decode(raw); local state=tostring(entry.state or ''); local old_lease=tonumber(entry.operationLeaseUntil or 0) if state == 'withdrawn' then return {1, raw} end if state == 'withdrawing' and old_lease > now then return {0, 'WITHDRAWAL_IN_PROGRESS'} end if state == 'withdrawing' and old_lease <= now then state='withdrawal_uncertain' end if state ~= 'eligible' and state ~= 'promotion_uncertain' and state ~= 'withdrawal_uncertain' then return {0, 'NOT_ELIGIBLE'} end entry.operationPriorState=state entry.state='withdrawing'; entry.operationClaimHash=claim_hash; entry.operationLeaseUntil=lease_until entry.lastError=''; entry.updatedAt=now raw=cjson.encode(entry); redis.call('SET', KEYS[4], raw); redis.call('PERSIST', KEYS[4]); redis.call('ZADD', KEYS[1], immortal, member) return {1, raw} """.strip() _REDIS_WITHDRAWAL_FAILED_LUA = r""" local now=tonumber(ARGV[1]); local claim_hash=ARGV[2]; local code=ARGV[3] local raw=redis.call('GET', KEYS[1]); if not raw then return {1, ''} end local entry=cjson.decode(raw) if tostring(entry.state or '') ~= 'withdrawing' or tostring(entry.operationClaimHash or '') ~= claim_hash then return {1, raw} end local prior=tostring(entry.operationPriorState or 'eligible') if prior == 'promotion_uncertain' or prior == 'withdrawal_uncertain' then entry.state=prior else entry.state='eligible' end entry.operationClaimHash=''; entry.operationLeaseUntil=0; entry.operationPriorState='' entry.lastError=string.sub(code,1,64); entry.updatedAt=now raw=cjson.encode(entry); redis.call('SET', KEYS[1], raw); redis.call('PERSIST', KEYS[1]); return {1, raw} """.strip() _REDIS_MARK_WITHDRAWN_LUA = r""" local now=tonumber(ARGV[1]); local member=ARGV[2]; local claim_hash=ARGV[3] local withdrawal_json=ARGV[4]; local removal_json=ARGV[5]; local terminal_retention=tonumber(ARGV[6]) local raw=redis.call('GET', KEYS[2]); if not raw then return {0, 'NOT_FOUND'} end local entry=cjson.decode(raw) if tostring(entry.state or '') ~= 'withdrawing' then return {0, 'WITHDRAWAL_STATE'} end if tostring(entry.operationClaimHash or '') ~= claim_hash then return {0, 'STALE_CLAIM'} end entry.state='withdrawn'; entry.records={}; entry.bytes=0; entry.withdrawnAt=now; entry.withdrawalStorage=cjson.decode(withdrawal_json) entry.currentViewRemoval=cjson.decode(removal_json); entry.lastError=''; entry.updatedAt=now entry.operationClaimHash=''; entry.operationLeaseUntil=0; entry.operationPriorState='' raw=cjson.encode(entry); redis.call('SET', KEYS[4], raw, 'EX', terminal_retention); redis.call('ZADD', KEYS[1], now+terminal_retention, member) redis.call('ZREM', KEYS[2], member); redis.call('HDEL', KEYS[3], member) return {1, raw} """.strip() class RedisContributionLedger: """ Shared transactional receipt authority backed by one Redis domain. The Redis backend closes the *multi-replica coordination* gap: create, promotion claims, pending delete, withdrawal claims, and terminal transitions are atomic server-side operations. It does **not** infer the operator's Redis persistence/backup policy; therefore ``durable`` remains false and ``CONTRIBUTION_REQUIRE_DURABLE`` must be satisfied separately when crash/power loss durability is a deployment requirement. """ backend = "redis" durability = "shared_transactional_external" durable = False shared = True authoritative = True consistency_scope = "single_redis_consistency_domain" _IMMORTAL_SCORE = 253402300799.0 def __init__( self, url: str, *, key_secret: str, key_prefix: str, max_pending_entries: int, max_pending_bytes: int, max_receipts: int, terminal_retention_seconds: int = 86_400, operation_lease_seconds: int = 120, socket_timeout_seconds: float = 2.0, client: Any | None = None, require_tls: bool = False, ) -> None: if not str(url or "").strip(): raise ContributionLedgerError("REDIS_URL_REQUIRED") if len(str(key_secret or "").encode("utf-8")) < ( 32 # ruff: ignore[magic-value-comparison] ): raise ContributionLedgerError("REDIS_KEY_SECRET_TOO_SHORT") self.url = str(url).strip() self.require_tls = bool(require_tls) try: self._transport, self._connection_kwargs = redis_connection_kwargs( self.url, require_tls=self.require_tls, socket_timeout_seconds=socket_timeout_seconds, ) except RedisSecurityError as exc: raise ContributionLedgerError(exc.code) from exc self._secret = str(key_secret).encode("utf-8") safe_prefix = "".join( ch for ch in str(key_prefix or "sphinx-ai-assistant").lower() if ch.isalnum() or ch in "_-:" ) self.key_prefix = safe_prefix[:64] or "sphinx-ai-assistant" self.max_pending_entries = int(max_pending_entries) self.max_pending_bytes = int(max_pending_bytes) self.max_receipts = int(max_receipts) self.terminal_retention_seconds = max(60, int(terminal_retention_seconds)) self.operation_lease_seconds = max(30, min(int(operation_lease_seconds), 900)) self.socket_timeout_seconds = max( 0.25, min(float(socket_timeout_seconds), 10.0) ) self._client = client self._owns_client = client is None self._init_lock = asyncio.Lock() tag = f"{self.key_prefix}:{{contribution}}" self._all_key = f"{tag}:all" self._pending_key = f"{tag}:pending" self._pending_bytes_key = f"{tag}:pending-bytes" self._receipt_prefix = f"{tag}:receipt:" def manifest(self) -> dict[str, Any]: return { "backend": self.backend, "durability": self.durability, "durable": self.durable, "shared": self.shared, "authoritative": self.authoritative, "consistency_scope": self.consistency_scope, "receipt_id_externalized": "hmac_sha256", "operation_claims": "leased_sha256", **self._transport.manifest(), } async def initialize(self) -> None: async with self._init_lock: if self._client is None: try: import redis.asyncio as redis_async # type: ignore[import-not-found] # ruff: ignore[import-outside-top-level] except Exception as exc: # pragma: no cover - deployment dependency raise ContributionLedgerError( "REDIS_DEPENDENCY_UNAVAILABLE" ) from exc self._client = redis_async.from_url(self.url, **self._connection_kwargs) try: await self._client.ping() except Exception as exc: raise ContributionLedgerError("REDIS_UNAVAILABLE") from exc async def close(self) -> None: if self._client is None or not self._owns_client: return closer = getattr(self._client, "aclose", None) or getattr( self._client, "close", None ) if closer is not None: result = closer() if hasattr(result, "__await__"): await result self._client = None def _member(self, receipt_id: str) -> str: return hmac.new( self._secret, str(receipt_id).encode("utf-8"), hashlib.sha256 ).hexdigest() def _receipt_key(self, member: str) -> str: return f"{self._receipt_prefix}{member}" @staticmethod def _claim_hash(claim: str) -> str: return hashlib.sha256(str(claim).encode("utf-8")).hexdigest() @staticmethod def _encode(entry: dict[str, Any]) -> str: private = { k: v for k, v in entry.items() if k not in {"receiptId", "operationClaim"} } private.setdefault("operationClaimHash", "") private.setdefault("operationLeaseUntil", 0) return json.dumps(private, ensure_ascii=False, separators=(",", ":")) @staticmethod def _decode(raw: Any, receipt_id: str) -> dict[str, Any] | None: if raw in {None, b"", ""}: return None if isinstance(raw, bytes): raw = raw.decode("utf-8") entry = json.loads(str(raw)) entry.pop("operationClaimHash", None) entry.pop("operationLeaseUntil", None) entry.pop("operationPriorState", None) entry["receiptId"] = receipt_id return entry @staticmethod def _result_parts(result: Any) -> tuple[int, Any]: if not isinstance(result, (list, tuple)) or len(result) < ( 2 # ruff: ignore[magic-value-comparison] ): raise ContributionLedgerError("REDIS_PROTOCOL_ERROR") ok = int(result[0]) value = result[1] if isinstance(value, bytes): value = value.decode("utf-8") return ok, value async def _eval( self, script: str, keys: list[str], args: list[Any] ) -> tuple[int, Any]: if self._client is None: raise ContributionLedgerError("REDIS_NOT_INITIALIZED") try: result = await self._client.eval(script, len(keys), *keys, *args) except ContributionLedgerError: raise except Exception as exc: raise ContributionLedgerError("REDIS_OPERATION_FAILED") from exc return self._result_parts(result) def _keys(self, receipt_id: str) -> tuple[str, str]: member = self._member(receipt_id) return member, self._receipt_key(member) async def create(self, entry: dict[str, Any]) -> None: now = _now() member, receipt_key = self._keys(entry["receiptId"]) expires_at = float(entry["expiresAt"]) live_until = expires_at + self.terminal_retention_seconds ttl = max(1, int(live_until - now + 0.999)) ok, value = await self._eval( _REDIS_CREATE_LUA, [self._all_key, self._pending_key, self._pending_bytes_key, receipt_key], [ now, member, self._encode(entry), expires_at, live_until, self.max_receipts, self.max_pending_entries, self.max_pending_bytes, int(entry.get("bytes") or 0), ttl, ], ) if not ok: raise ContributionLedgerError(str(value)) async def get(self, receipt_id: str) -> dict[str, Any] | None: member, receipt_key = self._keys(receipt_id) ok, value = await self._eval( _REDIS_GET_LUA, [self._all_key, self._pending_key, self._pending_bytes_key, receipt_key], [_now(), member, self.terminal_retention_seconds, self._IMMORTAL_SCORE], ) if not ok: raise ContributionLedgerError(str(value)) return self._decode(value, receipt_id) async def replace_pending_payload( self, receipt_id: str, *, records: list[dict[str, Any]], byte_count: int, dedup_keys: list[str], payload_digest: str, row_count: int, storage: dict[str, Any] | None = None, ) -> dict[str, Any]: member, receipt_key = self._keys(receipt_id) ok, value = await self._eval( _REDIS_REPLACE_PENDING_LUA, [self._all_key, self._pending_key, self._pending_bytes_key, receipt_key], [ _now(), member, json.dumps(records, ensure_ascii=False, separators=(",", ":")), int(byte_count), json.dumps(dedup_keys, separators=(",", ":")), str(payload_digest), int(row_count), self.max_pending_bytes, self.terminal_retention_seconds, ( json.dumps(storage, ensure_ascii=False, separators=(",", ":")) if storage is not None else "" ), ], ) if not ok: raise ContributionLedgerError(str(value)) return self._decode(value, receipt_id) or {} async def set_pending_storage( self, receipt_id: str, *, storage: dict[str, Any] ) -> dict[str, Any]: member, receipt_key = self._keys(receipt_id) ok, value = await self._eval( _REDIS_SET_PENDING_STORAGE_LUA, [self._all_key, self._pending_key, self._pending_bytes_key, receipt_key], [ _now(), member, json.dumps(storage, ensure_ascii=False, separators=(",", ":")), self.terminal_retention_seconds, ], ) if not ok: raise ContributionLedgerError(str(value)) return self._decode(value, receipt_id) or {} async def begin_promotion(self, receipt_id: str) -> dict[str, Any]: member, receipt_key = self._keys(receipt_id) now = _now() claim = secrets.token_urlsafe(24) claim_hash = self._claim_hash(claim) ok, value = await self._eval( _REDIS_BEGIN_PROMOTION_LUA, [self._all_key, self._pending_key, self._pending_bytes_key, receipt_key], [ now, member, claim_hash, now + self.operation_lease_seconds, self.terminal_retention_seconds, ], ) if not ok: raise ContributionLedgerError(str(value)) entry = self._decode(value, receipt_id) or {} entry["operationClaim"] = claim return entry async def promotion_failed( self, receipt_id: str, code: str, *, claim_token: str | None = None ) -> None: member, receipt_key = self._keys(receipt_id) claim_hash = self._claim_hash(claim_token or "") await self._eval( _REDIS_PROMOTION_FAILED_LUA, [self._all_key, self._pending_key, self._pending_bytes_key, receipt_key], [ _now(), member, claim_hash, str(code or "PROMOTION_FAILED")[:64], self.terminal_retention_seconds, ], ) async def mark_promotion_uncertain( self, receipt_id: str, code: str, *, claim_token: str | None = None ) -> dict[str, Any]: member, receipt_key = self._keys(receipt_id) ok, value = await self._eval( _REDIS_MARK_PROMOTION_UNCERTAIN_LUA, [self._all_key, self._pending_key, self._pending_bytes_key, receipt_key], [ _now(), member, self._claim_hash(claim_token or ""), str(code or "PROMOTION_OUTCOME_UNCERTAIN")[:64], self._IMMORTAL_SCORE, ], ) if not ok: raise ContributionLedgerError(str(value)) return self._decode(value, receipt_id) or {} async def mark_promoted( self, receipt_id: str, *, storage: dict[str, Any], claim_token: str | None = None, ) -> dict[str, Any]: member, receipt_key = self._keys(receipt_id) ok, value = await self._eval( _REDIS_MARK_PROMOTED_LUA, [self._all_key, self._pending_key, self._pending_bytes_key, receipt_key], [ _now(), member, self._claim_hash(claim_token or ""), json.dumps(storage, separators=(",", ":")), self._IMMORTAL_SCORE, ], ) if not ok: raise ContributionLedgerError(str(value)) return self._decode(value, receipt_id) or {} async def delete_pending(self, receipt_id: str) -> dict[str, Any]: member, receipt_key = self._keys(receipt_id) ok, value = await self._eval( _REDIS_DELETE_PENDING_LUA, [self._all_key, self._pending_key, self._pending_bytes_key, receipt_key], [_now(), member, self.terminal_retention_seconds], ) if not ok: raise ContributionLedgerError(str(value)) return self._decode(value, receipt_id) or {} async def begin_withdrawal(self, receipt_id: str) -> dict[str, Any]: member, receipt_key = self._keys(receipt_id) now = _now() claim = secrets.token_urlsafe(24) claim_hash = self._claim_hash(claim) ok, value = await self._eval( _REDIS_BEGIN_WITHDRAWAL_LUA, [self._all_key, self._pending_key, self._pending_bytes_key, receipt_key], [ now, member, claim_hash, now + self.operation_lease_seconds, self._IMMORTAL_SCORE, ], ) if not ok: raise ContributionLedgerError(str(value)) entry = self._decode(value, receipt_id) or {} if entry.get("state") != "withdrawn": entry["operationClaim"] = claim return entry async def withdrawal_failed( self, receipt_id: str, code: str, *, claim_token: str | None = None ) -> None: _member, receipt_key = self._keys(receipt_id) await self._eval( _REDIS_WITHDRAWAL_FAILED_LUA, [receipt_key], [ _now(), self._claim_hash(claim_token or ""), str(code or "WITHDRAWAL_FAILED")[:64], ], ) async def mark_withdrawn( self, receipt_id: str, *, withdrawal_storage: dict[str, Any], current_view_removal: dict[str, str], claim_token: str | None = None, ) -> dict[str, Any]: member, receipt_key = self._keys(receipt_id) ok, value = await self._eval( _REDIS_MARK_WITHDRAWN_LUA, [self._all_key, self._pending_key, self._pending_bytes_key, receipt_key], [ _now(), member, self._claim_hash(claim_token or ""), json.dumps(withdrawal_storage, separators=(",", ":")), json.dumps(current_view_removal, separators=(",", ":")), self.terminal_retention_seconds, ], ) if not ok: raise ContributionLedgerError(str(value)) return self._decode(value, receipt_id) or {} def build_contribution_ledger( backend: str, *, sqlite_path: str, redis_url: str = "", redis_key_secret: str = "", redis_key_prefix: str = "sphinx-ai-assistant", redis_timeout_seconds: float = 2.0, operation_lease_seconds: int = 120, max_pending_entries: int, max_pending_bytes: int, max_receipts: int, terminal_retention_seconds: int = 86_400, require_redis_tls: bool = False, ): """Construct the configured receipt ledger without reading any credentials.""" mode = str(backend or "memory").strip().lower() if mode == "redis": return RedisContributionLedger( redis_url, key_secret=redis_key_secret, key_prefix=redis_key_prefix, max_pending_entries=max_pending_entries, max_pending_bytes=max_pending_bytes, max_receipts=max_receipts, terminal_retention_seconds=terminal_retention_seconds, operation_lease_seconds=operation_lease_seconds, socket_timeout_seconds=redis_timeout_seconds, require_tls=require_redis_tls, ) if mode == "sqlite": return SQLiteContributionLedger( sqlite_path, max_pending_entries=max_pending_entries, max_pending_bytes=max_pending_bytes, max_receipts=max_receipts, terminal_retention_seconds=terminal_retention_seconds, ) if mode != "memory": raise ContributionLedgerError("UNSUPPORTED_BACKEND") return MemoryContributionLedger( max_pending_entries=max_pending_entries, max_pending_bytes=max_pending_bytes, max_receipts=max_receipts, terminal_retention_seconds=terminal_retention_seconds, ) __all__ = [ "_REDIS_BEGIN_PROMOTION_LUA", "_REDIS_CREATE_LUA", "_REDIS_MARK_PROMOTED_LUA", "_REDIS_MARK_PROMOTION_UNCERTAIN_LUA", "ContributionLedgerError", "MemoryContributionLedger", "RedisContributionLedger", "SQLiteContributionLedger", "build_contribution_ledger", ]