diff --git a/docs/legacy-purchase-continuations.md b/docs/legacy-purchase-continuations.md index 418d8dc0..6a604b17 100644 --- a/docs/legacy-purchase-continuations.md +++ b/docs/legacy-purchase-continuations.md @@ -116,13 +116,42 @@ presigned/signed query parameters rather than writing them to the ledger. Move such values to an application secret store and resolve them only inside the executor or pending poller. -`purge_resolved_before(cutoff)` removes only old succeeded/failed and -never-claimed continuations. It deliberately retains claimed, `in_flight`, -`pending`, and `ambiguous` operations regardless of age. Configure global and -per-principal record/logical-byte limits plus a per-payload limit for the -deployment; quota exhaustion fails closed and rolls back the attempted state -change. Before entering `in_flight` or `ambiguous`, the SQLite store durably -records a full result-payload reservation against both byte quotas. Every +`purge_resolved_before(cutoff)` removes payloads only for expired, old +succeeded/failed and never-claimed continuations. It never removes a +continuation while its bearer token is valid and deliberately retains claimed, +`in_flight`, `pending`, and `ambiguous` operations regardless of age. Cleanup +atomically replaces each removed authorization with a compact issuance +tombstone, so the same issuance identity cannot mint another redeemable token. +Tombstones are permanent replay fences and count against both global and +per-principal quotas. The SDK intentionally provides no tombstone-pruning path: +copying a fence elsewhere is safe only if every future issuance atomically +consults that durable archive, which requires an application-owned store rather +than this local SQLite implementation. +SQLite triggers make continuation deletion contingent on first recording its +tombstone and reject later reuse of a retired token or issuance fingerprint. +The tombstone schema and guards are installed in one locked migration. This +makes older workers fail closed during a rolling upgrade instead of silently +bypassing the new replay fence. Migrated pre-fingerprint rows cannot be +represented safely by a compact issuance tombstone and are therefore never +purged automatically; retain them until an operator completes migration and +replay-risk resolution. + +Configure global and per-principal record/logical-byte limits plus a per-payload +limit for the deployment. Both byte limits count the same persisted fields, +including identities and routing bindings; quota exhaustion fails closed and +rolls back the attempted state change. Size the per-principal record limit to at +least `peak live continuation/operation records + lifetime retired issuances`. +With the default SHA-256 hex identifiers, each modern tombstone consumes one +record and approximately `192 + UTF-8 bytes in principal_id` logical bytes +(token hash, issuance fingerprint, binding hash, and principal). Add the same +tombstones across principals for the global limits, monitor remaining capacity, +and include operational headroom. Deploy an application-owned durable +idempotency store from the start if the bounded SQLite horizon is insufficient; +exhaustion is an intentional availability failure rather than permission to +discard replay fences. + +Before entering `in_flight` or `ambiguous`, the SQLite store durably records a +full result-payload reservation against both byte quotas. Every worker uses that stored amount even if its local quota configuration differs. Pending and terminal writes consume the immutable reservation and therefore cannot be starved by another principal's later ledger use—or by a lower runtime @@ -172,12 +201,12 @@ The opaque token is derived with HMAC-SHA-256 from the application-held key, the principal-scoped issuance identity, and a canonical hash of every issuance binding; only its SHA-256 hash, key fingerprint, and binding hash are stored. The unique fingerprint makes exact projection retries return the same token, -while changed bindings produce a different token even after an old terminal -row has been purged. Reusing an unpurged issuance key with changed inputs fails -closed. A principal mismatch is reported as not found to avoid cross-tenant -token enumeration. Natural account comparison excludes mutable display -metadata such as `operator_unit.name` but includes the account's actual natural -key. +while reusing that issuance identity with changed bindings fails closed both +before and after payload cleanup. Start a genuinely new discovery with a new +issuance identity when the authorization changes. A principal mismatch is +reported as not found to avoid cross-tenant token enumeration. Natural account +comparison excludes mutable display metadata such as `operator_unit.name` but +includes the account's actual natural key. ## Claim and crash behavior diff --git a/src/adcp/compat/sqlite_continuation_store.py b/src/adcp/compat/sqlite_continuation_store.py index 5922ac28..7dcdb9ad 100644 --- a/src/adcp/compat/sqlite_continuation_store.py +++ b/src/adcp/compat/sqlite_continuation_store.py @@ -85,6 +85,89 @@ ); """ +_REPLAY_FENCE_SCHEMA = ( + """ + CREATE TABLE IF NOT EXISTS adcp_compat_issuance_tombstones ( + token_hash TEXT PRIMARY KEY, + principal_id TEXT NOT NULL, + issuance_fingerprint TEXT, + issuance_binding_hash TEXT, + retired_at TEXT NOT NULL + ) + """, + """ + CREATE UNIQUE INDEX IF NOT EXISTS adcp_compat_issuance_tombstones_issuance_idx + ON adcp_compat_issuance_tombstones (principal_id, issuance_fingerprint) + WHERE issuance_fingerprint IS NOT NULL + """, + """ + CREATE TRIGGER IF NOT EXISTS adcp_compat_continuations_retired_insert_guard + BEFORE INSERT ON adcp_compat_continuations + WHEN EXISTS ( + SELECT 1 FROM adcp_compat_issuance_tombstones AS tombstone + WHERE tombstone.token_hash = NEW.token_hash + OR ( + NEW.issuance_fingerprint IS NOT NULL + AND tombstone.principal_id = NEW.principal_id + AND tombstone.issuance_fingerprint = NEW.issuance_fingerprint + ) + ) + BEGIN + SELECT RAISE(ABORT, 'continuation issuance identity is retired'); + END + """, + """ + CREATE TRIGGER IF NOT EXISTS adcp_compat_continuations_retired_update_guard + BEFORE UPDATE OF token_hash, principal_id, issuance_fingerprint + ON adcp_compat_continuations + WHEN EXISTS ( + SELECT 1 FROM adcp_compat_issuance_tombstones AS tombstone + WHERE tombstone.token_hash = NEW.token_hash + OR ( + NEW.issuance_fingerprint IS NOT NULL + AND tombstone.principal_id = NEW.principal_id + AND tombstone.issuance_fingerprint = NEW.issuance_fingerprint + ) + ) + BEGIN + SELECT RAISE(ABORT, 'continuation issuance identity is retired'); + END + """, + """ + CREATE TRIGGER IF NOT EXISTS adcp_compat_continuations_delete_guard + BEFORE DELETE ON adcp_compat_continuations + WHEN NOT EXISTS ( + SELECT 1 FROM adcp_compat_issuance_tombstones AS tombstone + WHERE tombstone.token_hash = OLD.token_hash + ) + BEGIN + SELECT RAISE(ABORT, 'continuation deletion requires issuance tombstone'); + END + """, + """ + CREATE TRIGGER IF NOT EXISTS adcp_compat_issuance_tombstones_update_guard + BEFORE UPDATE ON adcp_compat_issuance_tombstones + BEGIN + SELECT RAISE(ABORT, 'continuation issuance tombstones are immutable'); + END + """, + """ + CREATE TRIGGER IF NOT EXISTS adcp_compat_issuance_tombstones_delete_guard + BEFORE DELETE ON adcp_compat_issuance_tombstones + BEGIN + SELECT RAISE(ABORT, 'continuation issuance tombstones are permanent'); + END + """, +) + +_REPLAY_FENCE_TRIGGER_NAMES = ( + "adcp_compat_continuations_retired_insert_guard", + "adcp_compat_continuations_retired_update_guard", + "adcp_compat_continuations_delete_guard", + "adcp_compat_issuance_tombstones_update_guard", + "adcp_compat_issuance_tombstones_delete_guard", +) + class SqliteCompatibilityContinuationStore: """Durable local continuation ledger backed by a SQLite file.""" @@ -214,6 +297,44 @@ def _ensure_timestamp_columns(self, conn: sqlite3.Connection) -> None: (self.max_payload_bytes,), ) self._audit_legacy_payloads(conn) + # The table, index, and guards share this migration's write transaction. + # Otherwise an older process could delete a continuation after the table + # became visible but before the delete guard committed. + self._migrate_replay_fence_schema(conn) + for statement in _REPLAY_FENCE_SCHEMA: + conn.execute(statement) + + @staticmethod + def _migrate_replay_fence_schema(conn: sqlite3.Connection) -> None: + """Remove the pre-release write-only equivalence column atomically.""" + + columns = { + row["name"] + for row in conn.execute("PRAGMA table_info(adcp_compat_issuance_tombstones)") + } + if "legacy_equivalence_hash" not in columns: + return + + for trigger_name in _REPLAY_FENCE_TRIGGER_NAMES: + conn.execute(f'DROP TRIGGER IF EXISTS "{trigger_name}"') + conn.execute( + "ALTER TABLE adcp_compat_issuance_tombstones " + "RENAME TO adcp_compat_issuance_tombstones_legacy" + ) + conn.execute(_REPLAY_FENCE_SCHEMA[0]) + conn.execute( + """ + INSERT INTO adcp_compat_issuance_tombstones ( + token_hash, principal_id, issuance_fingerprint, + issuance_binding_hash, retired_at + ) + SELECT + token_hash, principal_id, issuance_fingerprint, + issuance_binding_hash, retired_at + FROM adcp_compat_issuance_tombstones_legacy + """ + ) + conn.execute("DROP TABLE adcp_compat_issuance_tombstones_legacy") @staticmethod def _audit_legacy_payloads(conn: sqlite3.Connection) -> None: @@ -439,6 +560,18 @@ def _put_continuation(self, value: LegacyPurchaseContinuation) -> None: try: with closing(self._connect()) as conn, conn: conn.execute("BEGIN IMMEDIATE") + retired = conn.execute( + "SELECT 1 FROM adcp_compat_issuance_tombstones " + "WHERE token_hash = ? OR " + "(principal_id = ? AND issuance_fingerprint = ?)", + (value.token_hash, value.principal_id, value.issuance_fingerprint), + ).fetchone() + if retired is not None: + raise _error( + CompatibilityContinuationErrorCode.STORE_CONFLICT, + "continuation issuance identity was already retired", + "Start a genuinely new discovery with a new issuance identity.", + ) legacy = conn.execute( """ SELECT 1 FROM adcp_compat_continuations @@ -908,7 +1041,8 @@ def _enforce_ledger_quota(self, conn: sqlite3.Connection, *, principal_id: str) records = conn.execute( "SELECT " "(SELECT COUNT(*) FROM adcp_compat_continuations) + " - "(SELECT COUNT(*) FROM adcp_compat_operations)" + "(SELECT COUNT(*) FROM adcp_compat_operations) + " + "(SELECT COUNT(*) FROM adcp_compat_issuance_tombstones)" ).fetchone()[0] continuation_bytes = conn.execute( """ @@ -931,6 +1065,16 @@ def _enforce_ledger_quota(self, conn: sqlite3.Connection, *, principal_id: str) ), 0) FROM adcp_compat_continuations """ ).fetchone()[0] + tombstone_bytes = conn.execute( + """ + SELECT COALESCE(SUM( + length(CAST(token_hash AS BLOB)) + + length(CAST(principal_id AS BLOB)) + + length(CAST(COALESCE(issuance_fingerprint, '') AS BLOB)) + + length(CAST(COALESCE(issuance_binding_hash, '') AS BLOB)) + ), 0) FROM adcp_compat_issuance_tombstones + """ + ).fetchone()[0] operation_bytes = conn.execute( """ SELECT COALESCE(SUM( @@ -952,22 +1096,35 @@ def _enforce_ledger_quota(self, conn: sqlite3.Connection, *, principal_id: str) ), 0) FROM adcp_compat_operations """ ).fetchone()[0] - if records > self.max_records or continuation_bytes + operation_bytes > self.max_bytes: + if ( + records > self.max_records + or continuation_bytes + operation_bytes + tombstone_bytes > self.max_bytes + ): raise _quota_error(self.max_records, self.max_bytes) principal_records = conn.execute( "SELECT " "(SELECT COUNT(*) FROM adcp_compat_continuations WHERE principal_id = ?) + " - "(SELECT COUNT(*) FROM adcp_compat_operations WHERE principal_id = ?)", - (principal_id, principal_id), + "(SELECT COUNT(*) FROM adcp_compat_operations WHERE principal_id = ?) + " + "(SELECT COUNT(*) FROM adcp_compat_issuance_tombstones WHERE principal_id = ?)", + (principal_id, principal_id, principal_id), ).fetchone()[0] principal_continuation_bytes = conn.execute( """ SELECT COALESCE(SUM( + length(CAST(token_hash AS BLOB)) + + length(CAST(COALESCE(issuance_fingerprint, '') AS BLOB)) + + length(CAST(COALESCE(issuance_binding_hash, '') AS BLOB)) + + length(CAST(principal_id AS BLOB)) + + length(CAST(account_identity AS BLOB)) + + length(CAST(source_adcp_version AS BLOB)) + + length(CAST(expires_at AS BLOB)) + length(CAST(observed_request_json AS BLOB)) + length(CAST(observed_response_json AS BLOB)) + + length(CAST(observed_payload_hash AS BLOB)) + length(CAST(product_ids_json AS BLOB)) + length(CAST(COALESCE(projected_products_json, '') AS BLOB)) + length(CAST(losses_json AS BLOB)) + + length(CAST(target_binding AS BLOB)) + length(CAST(COALESCE(listed_purchase_context_json, '') AS BLOB)) ), 0) FROM adcp_compat_continuations WHERE principal_id = ? """, @@ -976,6 +1133,15 @@ def _enforce_ledger_quota(self, conn: sqlite3.Connection, *, principal_id: str) principal_operation_bytes = conn.execute( """ SELECT COALESCE(SUM( + length(CAST(operation_id AS BLOB)) + + length(CAST(principal_id AS BLOB)) + + length(CAST(idempotency_key AS BLOB)) + + length(CAST(token_hash AS BLOB)) + + length(CAST(payload_hash AS BLOB)) + + CASE + WHEN state = 'pending' THEN length(CAST('succeeded' AS BLOB)) + ELSE length(CAST(state AS BLOB)) + END + length(CAST(execution_input_json AS BLOB)) + CASE WHEN state IN ('in_flight', 'pending', 'ambiguous') @@ -986,9 +1152,20 @@ def _enforce_ledger_quota(self, conn: sqlite3.Connection, *, principal_id: str) """, (principal_id,), ).fetchone()[0] + principal_tombstone_bytes = conn.execute( + """ + SELECT COALESCE(SUM( + length(CAST(token_hash AS BLOB)) + + length(CAST(principal_id AS BLOB)) + + length(CAST(COALESCE(issuance_fingerprint, '') AS BLOB)) + + length(CAST(COALESCE(issuance_binding_hash, '') AS BLOB)) + ), 0) FROM adcp_compat_issuance_tombstones WHERE principal_id = ? + """, + (principal_id,), + ).fetchone()[0] if ( principal_records > self.max_records_per_principal - or principal_continuation_bytes + principal_operation_bytes + or principal_continuation_bytes + principal_operation_bytes + principal_tombstone_bytes > self.max_bytes_per_principal ): raise _quota_error(self.max_records_per_principal, self.max_bytes_per_principal) @@ -1005,11 +1182,13 @@ async def purge_resolved_before(self, cutoff: datetime) -> int: def _purge_resolved_before(self, cutoff: datetime) -> int: cutoff_utc = _as_utc(cutoff) + now_utc = _as_utc(self._clock()) with closing(self._connect()) as conn: rows = conn.execute( """ SELECT continuation.token_hash, + continuation.issuance_fingerprint, continuation.expires_at, continuation.updated_at AS continuation_updated_at, operation.operation_id, @@ -1018,8 +1197,11 @@ def _purge_resolved_before(self, cutoff: datetime) -> int: FROM adcp_compat_continuations AS continuation LEFT JOIN adcp_compat_operations AS operation ON operation.token_hash = continuation.token_hash - WHERE operation.state IN ('succeeded', 'failed') - OR operation.operation_id IS NULL + WHERE continuation.issuance_fingerprint IS NOT NULL + AND ( + operation.state IN ('succeeded', 'failed') + OR operation.operation_id IS NULL + ) """, ).fetchall() candidates = { @@ -1029,20 +1211,26 @@ def _purge_resolved_before(self, cutoff: datetime) -> int: row["operation_id"], row["state"], row["operation_updated_at"], + row["issuance_fingerprint"], ) for row in rows - if ( - row["state"] - in { - CompatibilityOperationState.SUCCEEDED.value, - CompatibilityOperationState.FAILED.value, - } - and _parse_datetime(row["operation_updated_at"]) < cutoff_utc - ) - or ( - row["state"] is None - and _parse_datetime(row["expires_at"]) < cutoff_utc - and _parse_datetime(row["continuation_updated_at"]) < cutoff_utc + if row["issuance_fingerprint"] is not None + and ( + ( + row["state"] + in { + CompatibilityOperationState.SUCCEEDED.value, + CompatibilityOperationState.FAILED.value, + } + and _parse_datetime(row["expires_at"]) <= now_utc + and _parse_datetime(row["operation_updated_at"]) < cutoff_utc + ) + or ( + row["state"] is None + and _parse_datetime(row["expires_at"]) <= now_utc + and _parse_datetime(row["expires_at"]) < cutoff_utc + and _parse_datetime(row["continuation_updated_at"]) < cutoff_utc + ) ) } if not candidates: @@ -1056,13 +1244,18 @@ def _purge_resolved_before(self, cutoff: datetime) -> int: token_hashes = list(candidates) for start in range(0, len(token_hashes), 200): batch = token_hashes[start : start + 200] - deleted += self._purge_candidate_batch(batch, candidates) + deleted += self._purge_candidate_batch(batch, candidates, now_utc=now_utc) return deleted def _purge_candidate_batch( self, token_hashes: list[str], - candidates: Mapping[str, tuple[str, str, str | None, str | None, str | None]], + candidates: Mapping[ + str, + tuple[str, str, str | None, str | None, str | None, str], + ], + *, + now_utc: datetime, ) -> int: placeholders = ",".join("?" for _ in token_hashes) with closing(self._connect()) as conn, conn: @@ -1071,6 +1264,9 @@ def _purge_candidate_batch( f""" SELECT continuation.token_hash, + continuation.principal_id, + continuation.issuance_fingerprint, + continuation.issuance_binding_hash, continuation.expires_at, continuation.updated_at AS continuation_updated_at, operation.operation_id, @@ -1083,8 +1279,8 @@ def _purge_candidate_batch( """, token_hashes, ).fetchall() - confirmed = [ - row["token_hash"] + confirmed_rows = [ + row for row in rows if candidates.get(row["token_hash"]) == ( @@ -1093,11 +1289,33 @@ def _purge_candidate_batch( row["operation_id"], row["state"], row["operation_updated_at"], + row["issuance_fingerprint"], ) + and row["issuance_fingerprint"] is not None + and _parse_datetime(row["expires_at"]) <= now_utc ] - if not confirmed: + if not confirmed_rows: return 0 + confirmed = [row["token_hash"] for row in confirmed_rows] confirmed_placeholders = ",".join("?" for _ in confirmed) + conn.executemany( + """ + INSERT INTO adcp_compat_issuance_tombstones ( + token_hash, principal_id, issuance_fingerprint, + issuance_binding_hash, retired_at + ) VALUES (?, ?, ?, ?, ?) + """, + [ + ( + row["token_hash"], + row["principal_id"], + row["issuance_fingerprint"], + row["issuance_binding_hash"], + _format_datetime(now_utc), + ) + for row in confirmed_rows + ], + ) conn.execute( f"DELETE FROM adcp_compat_operations " f"WHERE token_hash IN ({confirmed_placeholders})", diff --git a/tests/test_purchase_continuation.py b/tests/test_purchase_continuation.py index 50830f01..2ef9f407 100644 --- a/tests/test_purchase_continuation.py +++ b/tests/test_purchase_continuation.py @@ -1180,7 +1180,9 @@ async def test_identical_issuance_is_idempotent_across_sqlite_restart(tmp_path: async def test_pre_fingerprint_authorization_blocks_duplicate_reissuance(tmp_path: Path) -> None: case = copy.deepcopy(_cases()[2]) database = tmp_path / "continuations.sqlite3" - first = _coordinator(SqliteCompatibilityContinuationStore(database), lambda _ctx: {}) + mutable_now = _NOW + store = SqliteCompatibilityContinuationStore(database, clock=lambda: mutable_now) + first = _coordinator(store, lambda _ctx: {}) await _issue(first, case) with closing(sqlite3.connect(database)) as conn, conn: conn.execute( @@ -1196,6 +1198,13 @@ async def test_pre_fingerprint_authorization_blocks_duplicate_reissuance(tmp_pat with closing(sqlite3.connect(database)) as conn: assert conn.execute("SELECT COUNT(*) FROM adcp_compat_continuations").fetchone()[0] == 1 + # The old row has no stable issuance fingerprint from which to build a + # compact tombstone, so automatic cleanup must retain its full replay fence. + mutable_now = datetime(2099, 1, 2, tzinfo=timezone.utc) + assert await store.purge_resolved_before(mutable_now) == 0 + with closing(sqlite3.connect(database)) as conn: + assert conn.execute("SELECT COUNT(*) FROM adcp_compat_continuations").fetchone()[0] == 1 + @pytest.mark.asyncio async def test_reused_issuance_key_with_changed_discovery_conflicts(tmp_path: Path) -> None: @@ -1213,11 +1222,14 @@ async def test_reused_issuance_key_with_changed_discovery_conflicts(tmp_path: Pa @pytest.mark.asyncio -async def test_purged_issuance_key_with_changed_bindings_gets_different_token( +async def test_purged_issuance_key_with_changed_bindings_remains_retired( tmp_path: Path, ) -> None: case = copy.deepcopy(_cases()[2]) - store = SqliteCompatibilityContinuationStore(tmp_path / "continuations.sqlite3") + mutable_now = _NOW + store = SqliteCompatibilityContinuationStore( + tmp_path / "continuations.sqlite3", clock=lambda: mutable_now + ) coordinator = _coordinator(store, lambda ctx: _success_for(ctx, "mb-before-purge")) await _issue(coordinator, case) old_token = case["continuation_input"]["continuation_token"] @@ -1226,12 +1238,15 @@ async def test_purged_issuance_key_with_changed_bindings_gets_different_token( principal_id="principal-acme", target_binding="seller-session-acme", ) - assert await store.purge_resolved_before(_NOW) == 1 + mutable_now = datetime(2099, 1, 2, tzinfo=timezone.utc) + assert await store.purge_resolved_before(mutable_now) == 1 changed = copy.deepcopy(case) changed["legacy_request"]["brief"] = "A new discovery with changed authorization bindings." - await _issue(coordinator, changed) - assert changed["continuation_input"]["continuation_token"] != old_token + with pytest.raises(CompatibilityContinuationError) as retired: + await _issue(coordinator, changed) + assert retired.value.code == CompatibilityContinuationErrorCode.STORE_CONFLICT + assert changed["continuation_input"]["continuation_token"] == old_token @pytest.mark.asyncio @@ -1496,6 +1511,33 @@ async def test_sqlite_principal_quota_does_not_consume_other_principal_capacity( assert conn.execute("SELECT COUNT(*) FROM adcp_compat_continuations").fetchone()[0] == 2 +@pytest.mark.asyncio +async def test_sqlite_principal_byte_quota_counts_target_binding(tmp_path: Path) -> None: + database = tmp_path / "continuations.sqlite3" + store = SqliteCompatibilityContinuationStore( + database, + max_bytes=1_000_000, + max_bytes_per_principal=20_000, + ) + coordinator = _coordinator(store, lambda _ctx: {}) + oversized = copy.deepcopy(_cases()[2]) + + with pytest.raises(CompatibilityContinuationError) as exc: + await _issue( + coordinator, + oversized, + principal="principal-noisy", + target="x" * 50_000, + ) + assert exc.value.code == CompatibilityContinuationErrorCode.STORE_QUOTA_EXCEEDED + + ordinary = copy.deepcopy(_cases()[2]) + await _issue(coordinator, ordinary, principal="principal-other") + with sqlite3.connect(database) as conn: + principals = conn.execute("SELECT principal_id FROM adcp_compat_continuations").fetchall() + assert principals == [("principal-other",)] + + @pytest.mark.asyncio async def test_sqlite_payload_quota_rejects_large_discovery(tmp_path: Path) -> None: case = copy.deepcopy(_cases()[2]) @@ -1990,6 +2032,276 @@ async def test_sqlite_store_never_persists_raw_bearer_token(tmp_path: Path) -> N assert raw_token not in stored_bytes +@pytest.mark.asyncio +async def test_sqlite_cleanup_preserves_replay_fence_until_expiry_and_tombstones_it( + tmp_path: Path, +) -> None: + database = tmp_path / "continuations.sqlite3" + mutable_now = _NOW + calls = 0 + + def execute(ctx: Any) -> dict[str, Any]: + nonlocal calls + calls += 1 + return _success_for(ctx, f"mb-{calls}") + + store = SqliteCompatibilityContinuationStore(database, clock=lambda: mutable_now) + coordinator = _coordinator(store, execute) + case = copy.deepcopy(_cases()[2]) + await _issue(coordinator, case) + await coordinator.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + + mutable_now = _NOW + timedelta(days=2) + assert await store.purge_resolved_before(mutable_now) == 0 + + # An exact issuance retry still resolves to the claimed continuation, so a + # new purchase idempotency key cannot execute a second buy. + case["continuation_input"]["idempotency_key"] = "b15ac836-a49e-4e59-bb49-df24dc2cc339" + await _issue(coordinator, case) + with pytest.raises(CompatibilityContinuationError) as claimed: + await coordinator.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + assert claimed.value.code == CompatibilityContinuationErrorCode.ALREADY_CLAIMED + assert calls == 1 + + mutable_now = datetime(2099, 1, 2, tzinfo=timezone.utc) + assert await store.purge_resolved_before(mutable_now) == 1 + with sqlite3.connect(database) as conn: + assert conn.execute("SELECT COUNT(*) FROM adcp_compat_continuations").fetchone()[0] == 0 + assert conn.execute("SELECT COUNT(*) FROM adcp_compat_operations").fetchone()[0] == 0 + assert ( + conn.execute("SELECT COUNT(*) FROM adcp_compat_issuance_tombstones").fetchone()[0] == 1 + ) + + with pytest.raises(CompatibilityContinuationError) as retired: + await _issue(coordinator, case) + assert retired.value.code == CompatibilityContinuationErrorCode.STORE_CONFLICT + + +@pytest.mark.asyncio +async def test_sqlite_replay_fence_triggers_fail_closed_for_older_workers( + tmp_path: Path, +) -> None: + database = tmp_path / "continuations.sqlite3" + mutable_now = _NOW + store = SqliteCompatibilityContinuationStore(database, clock=lambda: mutable_now) + coordinator = _coordinator(store, lambda ctx: _success_for(ctx, "mb-trigger-fence")) + case = copy.deepcopy(_cases()[2]) + await _issue(coordinator, case) + await coordinator.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + + with sqlite3.connect(database) as conn: + conn.row_factory = sqlite3.Row + continuation = dict(conn.execute("SELECT * FROM adcp_compat_continuations").fetchone()) + + # Simulate the cleanup SQL from a process that predates tombstones. + conn.execute("BEGIN") + conn.execute("DELETE FROM adcp_compat_operations") + with pytest.raises(sqlite3.IntegrityError, match="requires issuance tombstone"): + conn.execute("DELETE FROM adcp_compat_continuations") + conn.rollback() + assert conn.execute("SELECT COUNT(*) FROM adcp_compat_operations").fetchone()[0] == 1 + assert conn.execute("SELECT COUNT(*) FROM adcp_compat_continuations").fetchone()[0] == 1 + + mutable_now = datetime(2099, 1, 2, tzinfo=timezone.utc) + assert await store.purge_resolved_before(mutable_now) == 1 + + columns = tuple(continuation) + placeholders = ", ".join("?" for _ in columns) + with sqlite3.connect(database) as conn: + with pytest.raises(sqlite3.IntegrityError, match="issuance identity is retired"): + conn.execute( + f"INSERT INTO adcp_compat_continuations ({', '.join(columns)}) " + f"VALUES ({placeholders})", + tuple(continuation[column] for column in columns), + ) + with pytest.raises(sqlite3.IntegrityError, match="immutable"): + conn.execute("UPDATE adcp_compat_issuance_tombstones SET retired_at = retired_at") + with pytest.raises(sqlite3.IntegrityError, match="permanent"): + conn.execute("DELETE FROM adcp_compat_issuance_tombstones") + assert conn.execute("SELECT COUNT(*) FROM adcp_compat_continuations").fetchone()[0] == 0 + assert ( + conn.execute("SELECT COUNT(*) FROM adcp_compat_issuance_tombstones").fetchone()[0] == 1 + ) + + +@pytest.mark.asyncio +async def test_sqlite_replay_fence_migration_is_atomic_against_older_cleanup( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + database = tmp_path / "continuations.sqlite3" + store = SqliteCompatibilityContinuationStore(database, clock=lambda: _NOW) + coordinator = _coordinator(store, lambda ctx: _success_for(ctx, "mb-before-upgrade")) + case = copy.deepcopy(_cases()[2]) + await _issue(coordinator, case) + await coordinator.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + + # Recreate the on-disk shape visible immediately before this migration. + with sqlite3.connect(database) as conn: + trigger_names = [ + row[0] + for row in conn.execute( + "SELECT name FROM sqlite_master " + "WHERE type = 'trigger' AND name LIKE 'adcp_compat_%_guard'" + ) + ] + for trigger_name in trigger_names: + conn.execute(f'DROP TRIGGER "{trigger_name}"') + conn.execute("DROP TABLE adcp_compat_issuance_tombstones") + + migration_has_write_lock = threading.Event() + allow_migration = threading.Event() + older_cleanup_started = threading.Event() + original_connect = SqliteCompatibilityContinuationStore._connect + + def connect_with_migration_pause( + self: SqliteCompatibilityContinuationStore, + ) -> sqlite3.Connection: + conn = original_connect(self) + + def trace(statement: str) -> None: + if "CREATE TABLE IF NOT EXISTS adcp_compat_issuance_tombstones" in statement: + migration_has_write_lock.set() + allow_migration.wait(timeout=10) + + conn.set_trace_callback(trace) + return conn + + monkeypatch.setattr( + SqliteCompatibilityContinuationStore, + "_connect", + connect_with_migration_pause, + ) + + def run_older_cleanup() -> str: + with sqlite3.connect(database, timeout=10) as conn: + conn.set_trace_callback( + lambda statement: ( + older_cleanup_started.set() + if statement.strip().upper() == "BEGIN IMMEDIATE" + else None + ) + ) + try: + conn.execute("BEGIN IMMEDIATE") + conn.execute("DELETE FROM adcp_compat_operations") + conn.execute("DELETE FROM adcp_compat_continuations") + conn.commit() + except sqlite3.IntegrityError as exc: + conn.rollback() + return str(exc) + return "committed" + + with ThreadPoolExecutor(max_workers=2) as executor: + startup = executor.submit(SqliteCompatibilityContinuationStore, database) + assert migration_has_write_lock.wait(timeout=5) + cleanup = executor.submit(run_older_cleanup) + assert older_cleanup_started.wait(timeout=5) + try: + assert not cleanup.done() + finally: + allow_migration.set() + startup.result(timeout=10) + assert "requires issuance tombstone" in cleanup.result(timeout=10) + + with sqlite3.connect(database) as conn: + assert conn.execute("SELECT COUNT(*) FROM adcp_compat_continuations").fetchone()[0] == 1 + assert conn.execute("SELECT COUNT(*) FROM adcp_compat_operations").fetchone()[0] == 1 + assert ( + conn.execute("SELECT COUNT(*) FROM adcp_compat_issuance_tombstones").fetchone()[0] == 0 + ) + + +@pytest.mark.asyncio +async def test_sqlite_migrates_pre_release_tombstone_schema_before_purge( + tmp_path: Path, +) -> None: + database = tmp_path / "continuations.sqlite3" + mutable_now = _NOW + store = SqliteCompatibilityContinuationStore(database, clock=lambda: mutable_now) + coordinator = _coordinator(store, lambda ctx: _success_for(ctx, "mb-schema-upgrade")) + case = copy.deepcopy(_cases()[2]) + await _issue(coordinator, case) + await coordinator.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + + with sqlite3.connect(database) as conn: + trigger_names = [ + row[0] + for row in conn.execute( + "SELECT name FROM sqlite_master " + "WHERE type = 'trigger' AND name LIKE 'adcp_compat_%_guard'" + ) + ] + for trigger_name in trigger_names: + conn.execute(f'DROP TRIGGER "{trigger_name}"') + conn.execute("DROP TABLE adcp_compat_issuance_tombstones") + conn.execute( + """ + CREATE TABLE adcp_compat_issuance_tombstones ( + token_hash TEXT PRIMARY KEY, + principal_id TEXT NOT NULL, + issuance_fingerprint TEXT, + issuance_binding_hash TEXT, + legacy_equivalence_hash TEXT NOT NULL, + retired_at TEXT NOT NULL + ) + """ + ) + conn.execute( + "CREATE UNIQUE INDEX adcp_compat_issuance_tombstones_issuance_idx " + "ON adcp_compat_issuance_tombstones (principal_id, issuance_fingerprint) " + "WHERE issuance_fingerprint IS NOT NULL" + ) + conn.execute( + "INSERT INTO adcp_compat_issuance_tombstones VALUES (?, ?, ?, ?, ?, ?)", + ( + "1" * 64, + "principal-existing", + "2" * 64, + "3" * 64, + "write-only-value", + "2098-01-01T00:00:00Z", + ), + ) + + mutable_now = datetime(2099, 1, 2, tzinfo=timezone.utc) + restarted = SqliteCompatibilityContinuationStore(database, clock=lambda: mutable_now) + with sqlite3.connect(database) as conn: + columns = { + row[1] for row in conn.execute("PRAGMA table_info(adcp_compat_issuance_tombstones)") + } + assert "legacy_equivalence_hash" not in columns + assert ( + conn.execute("SELECT COUNT(*) FROM adcp_compat_issuance_tombstones").fetchone()[0] == 1 + ) + + assert await restarted.purge_resolved_before(mutable_now) == 1 + with sqlite3.connect(database) as conn: + assert ( + conn.execute("SELECT COUNT(*) FROM adcp_compat_issuance_tombstones").fetchone()[0] == 2 + ) + + @pytest.mark.asyncio async def test_sqlite_cleanup_retains_unresolved_operations(tmp_path: Path) -> None: database = tmp_path / "continuations.sqlite3" @@ -2019,7 +2331,7 @@ async def uncertain(_ctx: Any) -> dict[str, Any]: target_binding="seller-session-acme", ) - mutable_now = _NOW + timedelta(days=2) + mutable_now = datetime(2099, 1, 2, tzinfo=timezone.utc) assert await store.purge_resolved_before(_NOW + timedelta(days=1)) == 1 with sqlite3.connect(database) as conn: states = [row[0] for row in conn.execute("SELECT state FROM adcp_compat_operations")] @@ -2032,8 +2344,9 @@ async def test_sqlite_cleanup_compares_fractional_timestamps_chronologically( ) -> None: case = copy.deepcopy(_cases()[1]) updated_at = _NOW + timedelta(microseconds=500_000) + mutable_now = updated_at store = SqliteCompatibilityContinuationStore( - tmp_path / "continuations.sqlite3", clock=lambda: updated_at + tmp_path / "continuations.sqlite3", clock=lambda: mutable_now ) coordinator = _coordinator(store, lambda ctx: _success_for(ctx, "mb-fractional")) await _issue(coordinator, case) @@ -2043,6 +2356,7 @@ async def test_sqlite_cleanup_compares_fractional_timestamps_chronologically( target_binding="seller-session-acme", ) + mutable_now = datetime(2099, 1, 2, tzinfo=timezone.utc) assert await store.purge_resolved_before(_NOW) == 0 assert await store.purge_resolved_before(_NOW + timedelta(seconds=1)) == 1 @@ -2052,8 +2366,9 @@ async def test_sqlite_cleanup_does_not_depend_on_sqlite_datetime_functions( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: case = copy.deepcopy(_cases()[1]) + mutable_now = _NOW store = SqliteCompatibilityContinuationStore( - tmp_path / "continuations.sqlite3", clock=lambda: _NOW + tmp_path / "continuations.sqlite3", clock=lambda: mutable_now ) coordinator = _coordinator(store, lambda ctx: _success_for(ctx, "mb-portable-cleanup")) await _issue(coordinator, case) @@ -2084,6 +2399,7 @@ def deny_datetime_functions( monkeypatch.setattr(store, "_connect", connect_without_datetime_functions) + mutable_now = datetime(2099, 1, 2, tzinfo=timezone.utc) assert await store.purge_resolved_before(_NOW + timedelta(seconds=1)) == 1