From 91fa2786efdd3117ce2c1a875ede7431767d7e87 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Wed, 16 Sep 2026 00:27:52 +0000 Subject: [PATCH 1/2] fix(reporting): scope configuration generations by account --- docs/reporting-ledger-migration.md | 100 ++++ src/adcp/reporting/ledger/__init__.py | 2 + src/adcp/reporting/ledger/consumer_status.py | 21 +- src/adcp/reporting/ledger/models.py | 42 +- src/adcp/reporting/ledger/pg.py | 211 ++++--- src/adcp/reporting/ledger/producer.py | 7 +- .../reporting/ledger/reporting_ledger.sql | 15 +- .../reporting_ledger_account_generations.sql | 52 ++ src/adcp/reporting/ledger/status.py | 27 +- src/adcp/reporting/ledger/store.py | 60 +- .../reporting/_generation_support.py | 155 ++++++ .../test_reporting_generation_identity.py | 521 ++++++++++++++++++ .../test_reporting_generation_migration.py | 278 ++++++++++ tests/fixtures/reporting_ledger_beta15.sql | 284 ++++++++++ .../fixtures/reporting_ledger_beta15_data.sql | 26 + .../type_checks/reporting_generation_keys.py | 56 ++ 16 files changed, 1721 insertions(+), 136 deletions(-) create mode 100644 docs/reporting-ledger-migration.md create mode 100644 src/adcp/reporting/ledger/reporting_ledger_account_generations.sql create mode 100644 tests/conformance/reporting/_generation_support.py create mode 100644 tests/conformance/reporting/test_reporting_generation_identity.py create mode 100644 tests/conformance/reporting/test_reporting_generation_migration.py create mode 100644 tests/fixtures/reporting_ledger_beta15.sql create mode 100644 tests/fixtures/reporting_ledger_beta15_data.sql create mode 100644 tests/type_checks/reporting_generation_keys.py diff --git a/docs/reporting-ledger-migration.md b/docs/reporting-ledger-migration.md new file mode 100644 index 000000000..6690eb113 --- /dev/null +++ b/docs/reporting-ledger-migration.md @@ -0,0 +1,100 @@ +# Account-qualified reporting generations + +The fix for [#1169](https://github.com/adcontextprotocol/adcp-client-python/issues/1169) +changes a reporting configuration generation's identity to +`(account_id, delivery_config_id, delivery_config_version)`. Two accounts can +each accept `daily@1`, with different immutable contents, in one ledger. + +Use the frozen, hashable public value for maps and joins: + +```python +from adcp.reporting.ledger import ReportingConfigurationGenerationKey + +key = ReportingConfigurationGenerationKey( + account_id="account-a", + delivery_config_id="daily", + delivery_config_version=1, +) +generations[key] = configuration +``` + +`ReportingConfiguration.generation_key` now returns this value. Code that +unpacked or indexed the beta.15 two-tuple must use the named attributes instead. +`ReportingObligationRecord`, `ConsumerStatusRecord`, and `LeasedConfiguration` +also expose `generation_key`. The existing constructors and store method +arguments, including `find_obligation` and the lease/release calls, remain +compatible. Consumer-status chain tuples, retained hashes, and derived +obligation/issue identifiers keep their existing serialization. + +The worker lease API still selects work across the store's accounts. Resolve +work using the returned lease's account and generation. Releases match its +account, config ID, version, worker ID, and expiry; a release of an expired +handle cannot clear a newer lease held under the same worker ID. Callers must +retain the returned expiry when persisting or reconstructing a lease handle. +Passing a configuration from a different account or generation to +`ReportingProducer.acquire_obligation` now raises +`CONFIGURATION_GENERATION_MISMATCH` before touching the source. + +Seller-issued obligation IDs remain globally unique. A low-level write that +reuses an ID for a different logical period now raises +`OBLIGATION_IDENTITY_CONFLICT` in both stores; it cannot overwrite another +account's obligation in memory. When filtering consumer statements by +obligation IDs, the named obligations must exist in the requested account. + +## Upgrading PostgreSQL from 8.0.0-beta.15 + +1. Stop and drain all older reporting workers and configuration writers that + use this ledger. Keep them stopped throughout the upgrade. Older code + still selects and releases generations without an account predicate, so + mixing old and new workers is unsafe once accounts reuse a config ID. +2. With the upgraded SDK, run `await store.create_schema()` before starting + reporting work. It creates missing tables and applies the bundled + `reporting_ledger_account_generations.sql` migration in one transaction. +3. Restart reporting work with the upgraded SDK on every instance. + +For deployments managed by a migration tool, the standalone migration is +[`reporting_ledger_account_generations.sql`](../src/adcp/reporting/ledger/reporting_ledger_account_generations.sql). +It upgrades an existing beta.15 ledger by itself, including in autocommit mode. +For a combined bootstrap and upgrade, run both bundled files in one transaction: + +```sh +psql "$REPORTING_DATABASE_URL" --set=ON_ERROR_STOP=1 --single-transaction \ + -f src/adcp/reporting/ledger/reporting_ledger.sql \ + -f src/adcp/reporting/ledger/reporting_ledger_account_generations.sql +``` + +Use the ledger's existing `search_path` and a role that owns its tables. Both +SQL files are also included as resources in the installed SDK's +`adcp.reporting.ledger` package. Running only `CREATE TABLE IF NOT EXISTS` +leaves the old primary key in place and does not perform this upgrade. + +The migration inspects the primary-key columns under an exclusive table lock +and replaces the beta.15 global primary key with the account-qualified key. +It preserves the constraint's name, including a renamed beta.15 key. Existing +configuration rows, leases, obligations, revisions and frozen rows, +adjustments, consumer statements, issue lifecycle records, change-feed +sequences, hashes, and unrelated constraints/indexes remain intact. + +Bootstrap and migration share a transaction-scoped advisory lock, so concurrent +upgraded processes serialize their DDL. Repeated runs recognize the new primary +key and leave its index intact. A failed upgrade rolls back; the migration +does not use `CASCADE` or discard records. An unexpected primary key or an +adopter-added foreign key referencing the old key requires an explicit +adopter migration. The released beta.15 schema has no foreign keys referencing +`reporting_configurations`. + +The key's index rebuild holds an `ACCESS EXCLUSIVE` lock on +`reporting_configurations`, temporarily blocking its reads, writes, and leases. +Allow a maintenance window appropriate to the number of retained generations; +large obligation/revision tables are not rewritten. Configure deployment +lock/statement timeouts to match that window and retry a rolled-back migration +after resolving the blocking condition. These lock semantics follow +[PostgreSQL's ALTER TABLE documentation](https://www.postgresql.org/docs/16/sql-altertable.html). + +A rollback to beta.15 is unsafe once multiple accounts share a config ID and +version. Do not recreate the global key or delete conflicting account rows to +make an old worker start. Preserve the ledger and roll forward with a corrected +account-qualified implementation. + +The in-memory store needs no schema migration. Restart it with the upgraded +SDK and reload accepted configurations from the adopter's source of truth. diff --git a/src/adcp/reporting/ledger/__init__.py b/src/adcp/reporting/ledger/__init__.py index 7c514fca4..bb785ff5f 100644 --- a/src/adcp/reporting/ledger/__init__.py +++ b/src/adcp/reporting/ledger/__init__.py @@ -84,6 +84,7 @@ LedgerSnapshot, ReportingAdjustmentRecord, ReportingConfiguration, + ReportingConfigurationGenerationKey, ReportingDefinitionBinding, ReportingDeliveryEscalation, ReportingFinality, @@ -140,6 +141,7 @@ "ProducerOfferings", "ReportingAdjustmentRecord", "ReportingConfiguration", + "ReportingConfigurationGenerationKey", "ReportingDefinitionBinding", "ReportingDeliveryEscalation", "ReportingFinality", diff --git a/src/adcp/reporting/ledger/consumer_status.py b/src/adcp/reporting/ledger/consumer_status.py index 368dc38c1..7deeb79be 100644 --- a/src/adcp/reporting/ledger/consumer_status.py +++ b/src/adcp/reporting/ledger/consumer_status.py @@ -72,6 +72,7 @@ from adcp.reporting.ledger.models import ( ConsumerStatusRecord, ConsumerStatusValue, + ReportingConfigurationGenerationKey, ReportingDeliveryEscalation, ReportingHealth, ReportingIssue, @@ -283,11 +284,7 @@ async def _validate_against_configuration(self, record: ConsumerStatusRecord) -> account_id=record.account_id, delivery_config_ids=[record.delivery_config_id] ) generation = next( - ( - item - for item in configurations - if item.delivery_config_version == record.delivery_config_version - ), + (item for item in configurations if item.generation_key == record.generation_key), None, ) if generation is None: @@ -330,8 +327,7 @@ async def _resolve_named_records(self, record: ConsumerStatusRecord) -> None: "resolve for this caller and account", ) if ( - obligation.delivery_config_id != record.delivery_config_id - or obligation.delivery_config_version != record.delivery_config_version + obligation.generation_key != record.generation_key or obligation.report_definition_id != record.report_definition_id or _utc(obligation.period.start) != _utc(record.period_start) or _utc(obligation.period.end) != _utc(record.period_end) @@ -528,13 +524,18 @@ def consumer_mismatch_issue_key( ``obligation_missing`` attaches to the repaired obligation later, and the spec requires that chain never be lost, forked, or reset. """ + generation = ReportingConfigurationGenerationKey( + account_id=account_id, + delivery_config_id=delivery_config_id, + delivery_config_version=delivery_config_version, + ) payload = canonical_json_utf8_v1( [ "core-consumer-status-mismatch-v1", - account_id, + generation.account_id, consumer_id, - delivery_config_id, - delivery_config_version, + generation.delivery_config_id, + generation.delivery_config_version, report_definition_id, _utc(period_start).isoformat(), _utc(period_end).isoformat(), diff --git a/src/adcp/reporting/ledger/models.py b/src/adcp/reporting/ledger/models.py index 7539627cc..75fa3ee39 100644 --- a/src/adcp/reporting/ledger/models.py +++ b/src/adcp/reporting/ledger/models.py @@ -36,6 +36,7 @@ "LedgerSnapshot", "ReportingAdjustmentRecord", "ReportingConfiguration", + "ReportingConfigurationGenerationKey", "ReportingFinality", "ReportingHealth", "ReportingDeliveryEscalation", @@ -287,6 +288,20 @@ def first_ordinal_after( ordinal += 1 +@dataclass(frozen=True) +class ReportingConfigurationGenerationKey: + """The account-qualified identity of one accepted configuration generation. + + ``delivery_config_id`` is caller-selected and may be reused by another + account. Use this value for lookups, joins, and leases rather than a tuple + that could omit the account. It is immutable and hashable for use in maps. + """ + + account_id: str + delivery_config_id: str + delivery_config_version: int + + @dataclass(frozen=True) class ReportingConfiguration: """One accepted reporting configuration generation. @@ -320,8 +335,12 @@ class ReportingConfiguration: authoritative_party: Literal["seller", "consumer"] = "seller" @property - def generation_key(self) -> tuple[str, int]: - return (self.delivery_config_id, self.delivery_config_version) + def generation_key(self) -> ReportingConfigurationGenerationKey: + return ReportingConfigurationGenerationKey( + account_id=self.account_id, + delivery_config_id=self.delivery_config_id, + delivery_config_version=self.delivery_config_version, + ) @dataclass(frozen=True) @@ -352,6 +371,14 @@ class ReportingObligationRecord: definition: ReportingDefinitionBinding | None = None created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) + @property + def generation_key(self) -> ReportingConfigurationGenerationKey: + return ReportingConfigurationGenerationKey( + account_id=self.account_id, + delivery_config_id=self.delivery_config_id, + delivery_config_version=self.delivery_config_version, + ) + def __post_init__(self) -> None: if _utc(self.scope_resolved_at) != _utc(self.period.end): raise ValueError( @@ -478,6 +505,14 @@ class ConsumerStatusRecord: seller_ledger_as_of: datetime | None = None superseded: bool = False + @property + def generation_key(self) -> ReportingConfigurationGenerationKey: + return ReportingConfigurationGenerationKey( + account_id=self.account_id, + delivery_config_id=self.delivery_config_id, + delivery_config_version=self.delivery_config_version, + ) + @property def chain_key(self) -> tuple[str, str, str, int, str, str, str]: """The logical chain this statement belongs to. @@ -485,6 +520,9 @@ def chain_key(self) -> tuple[str, str, str, int, str, str, str]: Deliberately keyed *without* the seller's obligation id. Requiring it would make the first missing report invisible again, which is the exact failure this loop exists to surface. + + The flat shape is retained for compatibility with persisted statement + digests. Use ``generation_key`` when joining configuration generations. """ return ( self.account_id, diff --git a/src/adcp/reporting/ledger/pg.py b/src/adcp/reporting/ledger/pg.py index bf403668f..efd53b996 100644 --- a/src/adcp/reporting/ledger/pg.py +++ b/src/adcp/reporting/ledger/pg.py @@ -23,9 +23,12 @@ Schema bootstrap ---------------- -:meth:`create_schema` is idempotent. The equivalent raw DDL ships at -:file:`src/adcp/reporting/ledger/reporting_ledger.sql` for adopters using -Alembic, Flyway, or psql. +:meth:`create_schema` creates or upgrades the schema transactionally, including +the account-qualified configuration primary key for beta.15 installations. +The raw DDL ships in :file:`reporting_ledger.sql` followed by +:file:`reporting_ledger_account_generations.sql`; run both in one transaction +when using Alembic, Flyway, or psql. See :file:`docs/reporting-ledger-migration.md` +for deployment and compatibility notes. Where the invariants actually live ---------------------------------- @@ -81,6 +84,7 @@ LedgerSnapshot, ReportingAdjustmentRecord, ReportingConfiguration, + ReportingConfigurationGenerationKey, ReportingDefinitionBinding, ReportingIssueLifecycle, ReportingObligationRecord, @@ -116,6 +120,7 @@ ) _DDL_PATH = Path(__file__).parent / "reporting_ledger.sql" +_ACCOUNT_GENERATIONS_DDL_PATH = Path(__file__).parent / "reporting_ledger_account_generations.sql" __all__ = ["PG_AVAILABLE", "PgReportingLedgerStore"] @@ -155,9 +160,16 @@ def __init__( self._clock = clock async def create_schema(self) -> None: - """Create every ledger table and index. Idempotent; safe on every boot.""" + """Create or upgrade the ledger atomically, serializing concurrent boots. + + The bootstrap takes a transaction-scoped schema lock before any DDL. + Keep the migration in that transaction, including with an autocommit + pool, so a second process cannot observe a partially upgraded schema. + """ async with self._pool.connection() as connection: - await connection.execute(_DDL_PATH.read_text()) + async with connection.transaction(): + await connection.execute(_DDL_PATH.read_text()) + await connection.execute(_ACCOUNT_GENERATIONS_DDL_PATH.read_text()) # -- change feed ------------------------------------------------------ @@ -182,26 +194,10 @@ async def _lock_account(connection: Any, account_id: str) -> None: async def put_configuration(self, configuration: ReportingConfiguration) -> None: reject_reserved_authoritative_party(configuration) + key = configuration.generation_key payload = _configuration_payload(configuration) digest = _fingerprint(payload) async with self._pool.connection() as connection: - row = await ( - await connection.execute( - "SELECT content_sha256 FROM reporting_configurations" - " WHERE delivery_config_id = %s AND delivery_config_version = %s", - (configuration.delivery_config_id, configuration.delivery_config_version), - ) - ).fetchone() - if row is not None: - if row[0] != digest: - raise LedgerConflictError( - "CONFIGURATION_GENERATION_IMMUTABLE", - f"configuration {configuration.delivery_config_id}" - f"@{configuration.delivery_config_version} already exists with " - "different content; publish a new version instead of editing a " - "retained generation", - ) - return await connection.execute( "INSERT INTO reporting_configurations" " (delivery_config_id, delivery_config_version, account_id," @@ -211,11 +207,11 @@ async def put_configuration(self, configuration: ReportingConfiguration) -> None " authoritative_party, content_sha256)" " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, %s::jsonb, %s, %s, %s, %s," " %s::jsonb, %s, %s)" - " ON CONFLICT (delivery_config_id, delivery_config_version) DO NOTHING", + " ON CONFLICT (account_id, delivery_config_id, delivery_config_version) DO NOTHING", ( - configuration.delivery_config_id, - configuration.delivery_config_version, - configuration.account_id, + key.delivery_config_id, + key.delivery_config_version, + key.account_id, configuration.report_definition_id, configuration.reporting_profile, configuration.feed_purpose, @@ -232,6 +228,25 @@ async def put_configuration(self, configuration: ReportingConfiguration) -> None digest, ), ) + # Check *after* the insert. ON CONFLICT waits for a concurrent + # winner; a fresh READ COMMITTED statement sees its retained + # content. A pre-insert check followed by DO NOTHING could silently + # accept a different immutable generation from a losing writer. + row = await ( + await connection.execute( + "SELECT content_sha256 FROM reporting_configurations" + " WHERE account_id = %s AND delivery_config_id = %s" + " AND delivery_config_version = %s", + (key.account_id, key.delivery_config_id, key.delivery_config_version), + ) + ).fetchone() + if row is None or row[0] != digest: + raise LedgerConflictError( + "CONFIGURATION_GENERATION_IMMUTABLE", + f"configuration {key.delivery_config_id}@{key.delivery_config_version} " + "already exists with different content for this account; publish a new " + "version instead of editing a retained generation", + ) async def list_configurations( self, *, account_id: str, delivery_config_ids: Sequence[str] | None = None @@ -261,50 +276,55 @@ async def list_configurations( async def commit_obligation( self, obligation: ReportingObligationRecord ) -> ReportingObligationRecord: + key = obligation.generation_key async with self._pool.connection() as connection: - await self._lock_account(connection, obligation.account_id) - inserted = await ( - await connection.execute( - "INSERT INTO reporting_obligations" - " (reporting_obligation_id, account_id, delivery_config_id," - " delivery_config_version, report_definition_id, reporting_profile," - " feed_purpose, period_key, period_start, period_end, source_timezone," - " expected_at, scope_resolved_at, automated_recovery_deadline_at," - " required_finality, coverage_status, media_buy_ids, package_ids," - " schedule, definition, created_at)" - " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s," - " %s::jsonb, %s::jsonb, %s::jsonb, %s::jsonb, %s)" - " ON CONFLICT DO NOTHING" - " RETURNING reporting_obligation_id", - ( - obligation.reporting_obligation_id, - obligation.account_id, - obligation.delivery_config_id, - obligation.delivery_config_version, - obligation.report_definition_id, - obligation.reporting_profile, - obligation.feed_purpose, - obligation.period.period_key, - obligation.period.start, - obligation.period.end, - obligation.period.source_timezone, - obligation.period.expected_at, - obligation.scope_resolved_at, - obligation.automated_recovery_deadline_at, - obligation.required_finality, - obligation.coverage_status, - _json(sorted(obligation.media_buy_ids)), - _json(sorted(obligation.package_ids)), - _json(_schedule_payload(obligation.schedule)), + await self._lock_account(connection, key.account_id) + try: + inserted = await ( + await connection.execute( + "INSERT INTO reporting_obligations" + " (reporting_obligation_id, account_id, delivery_config_id," + " delivery_config_version, report_definition_id, reporting_profile," + " feed_purpose, period_key, period_start, period_end, source_timezone," + " expected_at, scope_resolved_at, automated_recovery_deadline_at," + " required_finality, coverage_status, media_buy_ids, package_ids," + " schedule, definition, created_at)" + " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s," + " %s::jsonb, %s::jsonb, %s::jsonb, %s::jsonb, %s)" + " ON CONFLICT (account_id, delivery_config_id, delivery_config_version," + " period_start, period_end) DO NOTHING" + " RETURNING reporting_obligation_id", ( - _json(_definition_payload(obligation.definition)) - if obligation.definition - else None + obligation.reporting_obligation_id, + key.account_id, + key.delivery_config_id, + key.delivery_config_version, + obligation.report_definition_id, + obligation.reporting_profile, + obligation.feed_purpose, + obligation.period.period_key, + obligation.period.start, + obligation.period.end, + obligation.period.source_timezone, + obligation.period.expected_at, + obligation.scope_resolved_at, + obligation.automated_recovery_deadline_at, + obligation.required_finality, + obligation.coverage_status, + _json(sorted(obligation.media_buy_ids)), + _json(sorted(obligation.package_ids)), + _json(_schedule_payload(obligation.schedule)), + ( + _json(_definition_payload(obligation.definition)) + if obligation.definition + else None + ), + obligation.created_at, ), - obligation.created_at, - ), - ) - ).fetchone() + ) + ).fetchone() + except Exception as error: + raise _translate_integrity_error(error) from error if inserted is not None: await self._append_change( connection, @@ -349,6 +369,11 @@ async def find_obligation( period_start: datetime, period_end: datetime, ) -> ReportingObligationRecord | None: + key = ReportingConfigurationGenerationKey( + account_id=account_id, + delivery_config_id=delivery_config_id, + delivery_config_version=delivery_config_version, + ) async with self._pool.connection() as connection: row = await ( await connection.execute( @@ -356,9 +381,9 @@ async def find_obligation( " WHERE account_id = %s AND delivery_config_id = %s" " AND delivery_config_version = %s AND period_start = %s AND period_end = %s", ( - account_id, - delivery_config_id, - delivery_config_version, + key.account_id, + key.delivery_config_id, + key.delivery_config_version, period_start, period_end, ), @@ -471,7 +496,8 @@ async def _require_current_leaf(connection: Any, revision: ReportingRevisionReco await connection.execute( "SELECT r.reporting_revision_id," " EXISTS (SELECT 1 FROM reporting_revisions s" - " WHERE s.supersedes_reporting_revision_id = r.reporting_revision_id)" + " WHERE s.account_id = r.account_id" + " AND s.supersedes_reporting_revision_id = r.reporting_revision_id)" " FROM reporting_revisions r" " WHERE r.reporting_revision_id = %s AND r.account_id = %s" " AND r.reporting_obligation_id = %s", @@ -649,6 +675,7 @@ async def list_adjustments( async def record_consumer_status( self, status: ConsumerStatusRecord ) -> tuple[ConsumerStatusRecord, bool]: + key = status.generation_key digest = _fingerprint(_consumer_status_payload(status)) async with self._pool.connection() as connection: replay = await self._replay(connection, status, digest) @@ -663,10 +690,10 @@ async def record_consumer_status( " AND delivery_config_version = %s AND report_definition_id = %s" " AND period_start = %s AND period_end = %s AND superseded = FALSE", ( - status.account_id, + key.account_id, status.consumer_id, - status.delivery_config_id, - status.delivery_config_version, + key.delivery_config_id, + key.delivery_config_version, status.report_definition_id, status.period_start, status.period_end, @@ -682,8 +709,8 @@ async def record_consumer_status( ) await connection.execute( "UPDATE reporting_consumer_statuses SET superseded = TRUE" - " WHERE reporting_status_id = %s", - (status.supersedes_reporting_status_id,), + " WHERE account_id = %s AND consumer_id = %s AND reporting_status_id = %s", + (key.account_id, status.consumer_id, status.supersedes_reporting_status_id), ) elif leaf is not None: raise LedgerConflictError( @@ -793,17 +820,18 @@ async def list_consumer_statuses( params: list[Any] = [account_id, consumer_id] if reporting_obligation_ids is not None: clause = ( - " AND (s.reporting_obligation_id = ANY(%s) OR EXISTS (" + " AND EXISTS (" " SELECT 1 FROM reporting_obligations o" " WHERE o.reporting_obligation_id = ANY(%s)" " AND o.account_id = s.account_id" - " AND o.delivery_config_id = s.delivery_config_id" + " AND (o.reporting_obligation_id = s.reporting_obligation_id OR (" + " o.delivery_config_id = s.delivery_config_id" " AND o.delivery_config_version = s.delivery_config_version" " AND o.report_definition_id = s.report_definition_id" " AND o.period_start = s.period_start" - " AND o.period_end = s.period_end))" + " AND o.period_end = s.period_end)))" ) - params.extend([list(reporting_obligation_ids), list(reporting_obligation_ids)]) + params.append(list(reporting_obligation_ids)) async with self._pool.connection() as connection: rows = await ( await connection.execute( @@ -1089,8 +1117,8 @@ async def lease_period_close( await connection.execute( "UPDATE reporting_configurations SET lease_worker_id = %s," " lease_expires_at = %s" - " WHERE (delivery_config_id, delivery_config_version) = (" - " SELECT delivery_config_id, delivery_config_version" + " WHERE (account_id, delivery_config_id, delivery_config_version) = (" + " SELECT account_id, delivery_config_id, delivery_config_version" " FROM reporting_configurations" " WHERE lease_expires_at IS NULL OR lease_expires_at <= %s" " ORDER BY lease_expires_at NULLS FIRST" @@ -1110,13 +1138,21 @@ async def lease_period_close( ) async def release_period_close(self, lease: LeasedConfiguration, *, worker_id: str) -> None: + key = lease.generation_key async with self._pool.connection() as connection: await connection.execute( "UPDATE reporting_configurations SET lease_worker_id = NULL," " lease_expires_at = NULL" - " WHERE delivery_config_id = %s AND delivery_config_version = %s" - " AND lease_worker_id = %s", - (lease.delivery_config_id, lease.delivery_config_version, worker_id), + " WHERE account_id = %s AND delivery_config_id = %s" + " AND delivery_config_version = %s AND lease_worker_id = %s" + " AND lease_expires_at = %s", + ( + key.account_id, + key.delivery_config_id, + key.delivery_config_version, + worker_id, + lease.lease_expires_at, + ), ) @@ -1157,6 +1193,11 @@ def _translate_integrity_error(error: Exception) -> LedgerConflictError: return LedgerConflictError( "OBLIGATION_EXISTS", "an obligation already exists for this logical period" ) + if "reporting_obligations_pkey" in text: + return LedgerConflictError( + "OBLIGATION_IDENTITY_CONFLICT", + "the obligation identifier already belongs to a different logical period", + ) return LedgerConflictError("LEDGER_WRITE_FAILED", "the ledger write violated an integrity rule") diff --git a/src/adcp/reporting/ledger/producer.py b/src/adcp/reporting/ledger/producer.py index 8cfec5933..3801b8926 100644 --- a/src/adcp/reporting/ledger/producer.py +++ b/src/adcp/reporting/ledger/producer.py @@ -275,7 +275,7 @@ async def run_worker(self) -> WorkerTurn: account_id=leased.account_id, delivery_config_ids=[leased.delivery_config_id], ) - if candidate.delivery_config_version == leased.delivery_config_version + if candidate.generation_key == leased.generation_key ), None, ) @@ -439,6 +439,11 @@ async def acquire_obligation( because re-reading a settled period on every worker turn would burn upstream quota to republish bytes nobody asked for. """ + if configuration.generation_key != obligation.generation_key: + raise LedgerConflictError( + "CONFIGURATION_GENERATION_MISMATCH", + "the source configuration must belong to the obligation's account and generation", + ) turn = turn or WorkerTurn() now = now or self._clock() revisions = await self._store.list_revisions( diff --git a/src/adcp/reporting/ledger/reporting_ledger.sql b/src/adcp/reporting/ledger/reporting_ledger.sql index d62b0a471..839408c2f 100644 --- a/src/adcp/reporting/ledger/reporting_ledger.sql +++ b/src/adcp/reporting/ledger/reporting_ledger.sql @@ -1,8 +1,10 @@ -- AdCP Reliable Reporting ledger — durable obligations, revisions, and status. -- --- Run this once per deployment, or call --- PgReportingLedgerStore.create_schema() from application code, which runs the --- equivalent DDL idempotently on boot. +-- Run this followed by reporting_ledger_account_generations.sql in ONE +-- transaction (psql --single-transaction -f ... -f ...), or call +-- PgReportingLedgerStore.create_schema(). CREATE TABLE IF NOT EXISTS alone +-- does not upgrade the global configuration primary key from 8.0.0-beta.15. +-- See docs/reporting-ledger-migration.md before upgrading live workers. -- -- COLLATE "C" on identifier columns avoids locale-dependent case folding — on -- some locales "Buy-A" and "buy-a" compare equal, which would collapse two @@ -14,6 +16,11 @@ -- persists a checkpoint and replays from it cannot miss a record or see the -- same record twice under a different identity. +-- Serialize bootstrap/upgrade before touching catalog objects: IF NOT EXISTS +-- by itself can still race another CREATE TABLE on an empty schema. The +-- standalone account-generations migration uses the same advisory lock. +SELECT pg_advisory_xact_lock(hashtext('adcp.reporting.schema'), hashtext(current_schema())); + CREATE TABLE IF NOT EXISTS reporting_configurations ( delivery_config_id TEXT COLLATE "C" NOT NULL, delivery_config_version INTEGER NOT NULL, @@ -40,7 +47,7 @@ CREATE TABLE IF NOT EXISTS reporting_configurations ( -- expiry instead of wedging the period forever. lease_worker_id TEXT COLLATE "C", lease_expires_at TIMESTAMPTZ, - PRIMARY KEY (delivery_config_id, delivery_config_version) + PRIMARY KEY (account_id, delivery_config_id, delivery_config_version) ); CREATE INDEX IF NOT EXISTS reporting_configurations_account_idx diff --git a/src/adcp/reporting/ledger/reporting_ledger_account_generations.sql b/src/adcp/reporting/ledger/reporting_ledger_account_generations.sql new file mode 100644 index 000000000..13e0e4566 --- /dev/null +++ b/src/adcp/reporting/ledger/reporting_ledger_account_generations.sql @@ -0,0 +1,52 @@ +-- Upgrade the 8.0.0-beta.15 configuration identity without rewriting evidence. +-- May also run on the current schema. This single DO statement is atomic even +-- in psql autocommit mode; create_schema() runs it with the bootstrap in one +-- transaction. Stop all older reporting workers before applying the upgrade. +-- +-- Only the known old primary key is replaced. Its name may have been changed +-- by an adopter. No CASCADE: unexpected keys or dependent foreign keys fail +-- and roll back, leaving the original schema and rows intact for inspection. +-- The SDK's beta.15 schema has no foreign keys to reporting_configurations. + +DO $account_generations$ +DECLARE + primary_key_name TEXT; + primary_key_columns TEXT[]; +BEGIN + PERFORM pg_advisory_xact_lock(hashtext('adcp.reporting.schema'), hashtext(current_schema())); + -- Lock before inspecting the key, so concurrent migrations cannot both + -- decide to replace it. This also excludes configuration writes/leases + -- while the unique index is rebuilt. Other ledger rows are untouched. + LOCK TABLE reporting_configurations IN ACCESS EXCLUSIVE MODE; + + SELECT pk.conname, ARRAY( + SELECT attribute.attname::TEXT + FROM unnest(pk.conkey) WITH ORDINALITY AS key_column(attnum, position) + JOIN pg_attribute attribute + ON attribute.attrelid = pk.conrelid AND attribute.attnum = key_column.attnum + ORDER BY key_column.position + ) + INTO primary_key_name, primary_key_columns + FROM pg_constraint pk + WHERE pk.conrelid = 'reporting_configurations'::regclass AND pk.contype = 'p'; + + IF primary_key_columns = ARRAY[ + 'account_id', 'delivery_config_id', 'delivery_config_version' + ] THEN + RETURN; + END IF; + + IF primary_key_columns IS DISTINCT FROM ARRAY[ + 'delivery_config_id', 'delivery_config_version' + ] THEN + RAISE EXCEPTION 'Unexpected reporting_configurations primary key: %', primary_key_columns + USING HINT = 'Expected the beta.15 or account-qualified key; inspect the schema before migrating.'; + END IF; + + EXECUTE format( + 'ALTER TABLE reporting_configurations DROP CONSTRAINT %I, ' + 'ADD CONSTRAINT %I PRIMARY KEY (account_id, delivery_config_id, delivery_config_version)', + primary_key_name, primary_key_name + ); +END +$account_generations$; diff --git a/src/adcp/reporting/ledger/status.py b/src/adcp/reporting/ledger/status.py index aab579367..11b251a09 100644 --- a/src/adcp/reporting/ledger/status.py +++ b/src/adcp/reporting/ledger/status.py @@ -257,9 +257,7 @@ async def _periods_view( projection=projection, revisions=revisions, statuses=statuses, - generation=generations.get( - (obligation.delivery_config_id, obligation.delivery_config_version) - ), + generation=generations.get(obligation.generation_key), caller=caller, snapshot=snapshot, ) @@ -436,9 +434,7 @@ async def _project_scope( projection=projection, revisions=revisions, statuses=statuses, - generation=generations.get( - (obligation.delivery_config_id, obligation.delivery_config_version) - ), + generation=generations.get(obligation.generation_key), caller=caller, snapshot=snapshot, ) @@ -692,6 +688,7 @@ def _scope_to_wire( ) horizon_start = _parse(requested.get("start")) or retained_from horizon_end = _parse(requested.get("end")) or _utc(ledger_as_of) + generations = {item.generation_key: item for item in configurations} return { "period_start": _iso(horizon_start), "period_end": _iso(horizon_end), @@ -702,15 +699,17 @@ def _scope_to_wire( "all_accessible_media_buys": not request.get("media_buy_ids"), "delivery_config_generations": [ { - "delivery_config_id": config_id, - "delivery_config_version": version, - "feed_purpose": feed_purpose, + "delivery_config_id": generation.delivery_config_id, + "delivery_config_version": generation.delivery_config_version, + "feed_purpose": generation.feed_purpose, } - for config_id, version, feed_purpose in sorted( - { - (item.delivery_config_id, item.delivery_config_version, item.feed_purpose) - for item in configurations - } + for generation in sorted( + generations.values(), + key=lambda item: ( + item.account_id, + item.delivery_config_id, + item.delivery_config_version, + ), ) ], "feed_purposes": sorted({item.feed_purpose for item in configurations}), diff --git a/src/adcp/reporting/ledger/store.py b/src/adcp/reporting/ledger/store.py index 53075799b..6724a5652 100644 --- a/src/adcp/reporting/ledger/store.py +++ b/src/adcp/reporting/ledger/store.py @@ -48,6 +48,7 @@ LedgerSnapshot, ReportingAdjustmentRecord, ReportingConfiguration, + ReportingConfigurationGenerationKey, ReportingIssueLifecycle, ReportingObligationRecord, ReportingRevisionRecord, @@ -94,6 +95,14 @@ class LeasedConfiguration: delivery_config_version: int lease_expires_at: datetime + @property + def generation_key(self) -> ReportingConfigurationGenerationKey: + return ReportingConfigurationGenerationKey( + account_id=self.account_id, + delivery_config_id=self.delivery_config_id, + delivery_config_version=self.delivery_config_version, + ) + @dataclass(frozen=True) class LedgerPage: @@ -160,7 +169,7 @@ class ReportingLedgerStore(Protocol): """Durable home for obligations, revisions, adjustments, and statuses.""" async def create_schema(self) -> None: - """Idempotently create whatever this store needs. Safe on every boot.""" + """Idempotently create or upgrade this store's schema. Safe on every boot.""" ... # -- configurations -------------------------------------------------- @@ -523,9 +532,11 @@ class InMemoryReportingLedgerStore: def __init__(self, *, clock: Callable[[], datetime] | None = None) -> None: self._clock = clock or (lambda: datetime.now(timezone.utc)) self._lock = asyncio.Lock() - self._configurations: dict[tuple[str, int], ReportingConfiguration] = {} + self._configurations: dict[ReportingConfigurationGenerationKey, ReportingConfiguration] = {} self._obligations: dict[str, ReportingObligationRecord] = {} - self._obligation_by_period: dict[tuple[str, str, int, str, str], str] = {} + self._obligation_by_period: dict[ + tuple[ReportingConfigurationGenerationKey, str, str], str + ] = {} self._revisions: dict[str, ReportingRevisionRecord] = {} self._revision_identity: dict[str, str] = {} self._rows: dict[str, tuple[dict[str, Any], ...]] = {} @@ -534,7 +545,7 @@ def __init__(self, *, clock: Callable[[], datetime] | None = None) -> None: self._status_identity: dict[str, str] = {} self._changes: list[tuple[int, str, LedgerRecordKind, str, datetime]] = [] self._sequence = 0 - self._leases: dict[tuple[str, int], tuple[str, datetime]] = {} + self._leases: dict[ReportingConfigurationGenerationKey, tuple[str, datetime]] = {} # Live occurrence per (account, issue_key), plus the retired generation # high-water mark so a recurrence never reuses an id. self._issues: dict[tuple[str, str], ReportingIssueLifecycle] = {} @@ -561,7 +572,8 @@ async def put_configuration(self, configuration: ReportingConfiguration) -> None ): raise LedgerConflictError( "CONFIGURATION_GENERATION_IMMUTABLE", - f"configuration {key[0]}@{key[1]} already exists with different content; " + f"configuration {key.delivery_config_id}@{key.delivery_config_version} " + "already exists with different content for this account; " "publish a new version instead of editing a retained generation", ) self._configurations[key] = configuration @@ -584,15 +596,18 @@ async def commit_obligation( ) -> ReportingObligationRecord: async with self._lock: key = ( - obligation.account_id, - obligation.delivery_config_id, - obligation.delivery_config_version, + obligation.generation_key, _utc(obligation.period.start).isoformat(), _utc(obligation.period.end).isoformat(), ) existing_id = self._obligation_by_period.get(key) if existing_id is not None: return self._obligations[existing_id] + if obligation.reporting_obligation_id in self._obligations: + raise LedgerConflictError( + "OBLIGATION_IDENTITY_CONFLICT", + "the obligation identifier already belongs to a different logical period", + ) self._obligations[obligation.reporting_obligation_id] = obligation self._obligation_by_period[key] = obligation.reporting_obligation_id self._append(obligation.account_id, "obligation", obligation.reporting_obligation_id) @@ -614,14 +629,20 @@ async def find_obligation( period_end: datetime, ) -> ReportingObligationRecord | None: key = ( - account_id, - delivery_config_id, - delivery_config_version, + ReportingConfigurationGenerationKey( + account_id=account_id, + delivery_config_id=delivery_config_id, + delivery_config_version=delivery_config_version, + ), _utc(period_start).isoformat(), _utc(period_end).isoformat(), ) found = self._obligation_by_period.get(key) - return self._obligations.get(found) if found else None + return ( + await self.get_obligation(account_id=account_id, reporting_obligation_id=found) + if found + else None + ) # -- revisions ------------------------------------------------------- @@ -862,15 +883,14 @@ def _matches_obligations(self, status: ConsumerStatusRecord, wanted: set[str]) - when the seller later creates that obligation the existing chain must attach to it rather than being lost, forked, or reset. """ - if status.reporting_obligation_id in wanted: - return True for obligation_id in wanted: obligation = self._obligations.get(obligation_id) - if obligation is None: + if obligation is None or obligation.account_id != status.account_id: continue + if status.reporting_obligation_id == obligation_id: + return True if ( - obligation.delivery_config_id == status.delivery_config_id - and obligation.delivery_config_version == status.delivery_config_version + obligation.generation_key == status.generation_key and obligation.report_definition_id == status.report_definition_id and _utc(obligation.period.start) == _utc(status.period_start) and _utc(obligation.period.end) == _utc(status.period_end) @@ -1011,7 +1031,7 @@ async def read_page( records: list[tuple[int, LedgerRecordKind, Any]] = [] for sequence, _account, kind, record_id, _committed in sorted(selected): record = self._resolve(kind, record_id) - if record is None: + if record is None or record.account_id != snapshot.account_id: continue if not self._in_scope(kind, record, config_filter, media_buy_filter, consumer_id): continue @@ -1100,9 +1120,9 @@ async def lease_period_close( async def release_period_close(self, lease: LeasedConfiguration, *, worker_id: str) -> None: async with self._lock: - key = (lease.delivery_config_id, lease.delivery_config_version) + key = lease.generation_key held = self._leases.get(key) - if held is not None and held[0] == worker_id: + if held == (worker_id, _utc(lease.lease_expires_at)): del self._leases[key] diff --git a/tests/conformance/reporting/_generation_support.py b/tests/conformance/reporting/_generation_support.py new file mode 100644 index 000000000..8bcd1059a --- /dev/null +++ b/tests/conformance/reporting/_generation_support.py @@ -0,0 +1,155 @@ +"""Shared account-isolation scenarios and an isolated, real PostgreSQL schema.""" + +from __future__ import annotations + +import asyncio +import os +import secrets +from collections.abc import AsyncIterator, Callable +from contextlib import asynccontextmanager +from datetime import datetime, timedelta, timezone +from typing import TYPE_CHECKING, Any + +import pytest + +from adcp.reporting.ledger import ( + ReportingConfiguration, + ReportingDefinitionBinding, + ReportingObligationRecord, + ReportingRevisionRecord, + ReportingScheduleSpec, + derive_period, + revision_content_sha256, +) +from adcp.reporting.source import ( + ReportingSourceCapabilitiesV1, + ReportingSourceExecutorResult, + ReportingSourceSliceRequestV1, +) + +if TYPE_CHECKING: + from psycopg_pool import AsyncConnectionPool + +START = datetime(2026, 9, 1, tzinfo=timezone.utc) +END = START + timedelta(hours=1) +NOW = START + timedelta(hours=3) + + +@asynccontextmanager +async def isolated_reporting_pool( + *, autocommit: bool = False +) -> AsyncIterator[AsyncConnectionPool]: + """Never touch an adopter's tables; each invocation owns one random schema.""" + url = os.environ.get("ADCP_PG_TEST_URL") + if not url: + pytest.skip("ADCP_PG_TEST_URL not set — requires real PostgreSQL") + pytest.importorskip("psycopg") + pytest.importorskip("psycopg_pool") + from psycopg import AsyncConnection, sql + from psycopg_pool import AsyncConnectionPool + + schema = f"adcp_reporting_identity_{secrets.token_hex(6)}" + async with await AsyncConnection.connect(url, autocommit=True) as admin: + await admin.execute(sql.SQL("CREATE SCHEMA {}").format(sql.Identifier(schema))) + try: + async with AsyncConnectionPool( + url, + kwargs={ + "options": f"-csearch_path={schema} -cstatement_timeout=15000", + "autocommit": autocommit, + }, + min_size=2, + max_size=8, + open=False, + ) as pool: + await pool.wait(timeout=10) + yield pool + finally: + await admin.execute(sql.SQL("DROP SCHEMA {} CASCADE").format(sql.Identifier(schema))) + + +def configuration(account_id: str = "acct_a") -> ReportingConfiguration: + return ReportingConfiguration( + delivery_config_id="daily", + delivery_config_version=1, + account_id=account_id, + report_definition_id="hourly_delivery", + reporting_profile="paid_media_delivery", + feed_purpose="analytics", + schedule=ReportingScheduleSpec("PT1H", "PT1H", period_anchor=START), + required_finality="snapshot", + activated_at=START, + deactivated_at=END, + media_buy_ids=(f"mb_{account_id}",), + definition=ReportingDefinitionBinding( + report_definition_uri=f"https://contracts.example.test/{account_id}/hourly", + report_definition_sha256="a" * 64, + schema_version="1.0.0", + schema_uri=f"https://contracts.example.test/{account_id}/rows.json", + schema_sha256="b" * 64, + ), + ) + + +def obligation_for(config: ReportingConfiguration) -> ReportingObligationRecord: + period = derive_period(config.schedule, account_timezone=config.account_timezone, ordinal=0) + return ReportingObligationRecord( + reporting_obligation_id=f"rpo_{config.account_id}", + account_id=config.account_id, + delivery_config_id=config.delivery_config_id, + delivery_config_version=config.delivery_config_version, + report_definition_id=config.report_definition_id, + reporting_profile=config.reporting_profile, + feed_purpose=config.feed_purpose, + period=period, + scope_resolved_at=period.end, + media_buy_ids=config.media_buy_ids, + required_finality=config.required_finality, + automated_recovery_deadline_at=period.expected_at + config.automated_recovery_window, + schedule=config.schedule, + definition=config.definition, + created_at=END, + ) + + +def revision_for( + obligation: ReportingObligationRecord, *, suffix: str = "first" +) -> tuple[ReportingRevisionRecord, list[dict[str, Any]]]: + rows: list[dict[str, Any]] = [{"media_buy_id": obligation.media_buy_ids[0], "impressions": 5}] + revision_id = f"rpr_{obligation.account_id}_{suffix}" + totals = (("impressions", "5"),) + return ( + ReportingRevisionRecord( + reporting_revision_id=revision_id, + account_id=obligation.account_id, + reporting_obligation_id=obligation.reporting_obligation_id, + finality="snapshot", + revision_content_sha256=revision_content_sha256( + reporting_revision_id=revision_id, + row_count=1, + control_totals=totals, + reporting_rows=rows, + ), + row_count=1, + control_totals=totals, + observed_at=END, + data_through=END, + created_at=END, + ), + rows, + ) + + +class UncalledSource: + @property + def capabilities(self) -> ReportingSourceCapabilitiesV1: + raise AssertionError("period close must not acquire source data") + + async def execute( + self, + request: ReportingSourceSliceRequestV1, + *, + cancel: asyncio.Event, + heartbeat: Callable[[], None] | None = None, + ) -> ReportingSourceExecutorResult: + raise AssertionError("period close must not acquire source data") diff --git a/tests/conformance/reporting/test_reporting_generation_identity.py b/tests/conformance/reporting/test_reporting_generation_identity.py new file mode 100644 index 000000000..5e5d4031e --- /dev/null +++ b/tests/conformance/reporting/test_reporting_generation_identity.py @@ -0,0 +1,521 @@ +"""Run identical tenant-isolation promises against memory and real PostgreSQL.""" + +from __future__ import annotations + +import asyncio +from collections.abc import AsyncIterator +from dataclasses import FrozenInstanceError, replace +from datetime import timedelta + +import pytest + +from adcp.reporting.fixtures import redacted_capabilities +from adcp.reporting.inline_source import InlineReportingSource +from adcp.reporting.ledger import ( + ConsumerStatusIngest, + ConsumerStatusRecord, + InMemoryReportingLedgerStore, + LeasedConfiguration, + LedgerConflictError, + ProducerOfferings, + ReportingConfigurationGenerationKey, + ReportingLedgerStore, + ReportingProducer, + ReportingStatusCaller, + ReportingStatusHandler, + consumer_mismatch_issue_key, +) +from adcp.reporting.ledger.pg import PgReportingLedgerStore +from adcp.reporting.source import ReportingSourceSliceRequestV1 +from tests.conformance.reporting._generation_support import ( + END, + NOW, + START, + UncalledSource, + configuration, + isolated_reporting_pool, + obligation_for, + revision_for, +) + + +@pytest.fixture(params=["memory", "postgres"]) +async def store(request: pytest.FixtureRequest) -> AsyncIterator[ReportingLedgerStore]: + if request.param == "memory": + yield InMemoryReportingLedgerStore(clock=lambda: NOW) + else: + async with isolated_reporting_pool() as pool: + ledger = PgReportingLedgerStore(pool=pool, clock=lambda: NOW) + await ledger.create_schema() + yield ledger + + +def test_generation_keys_are_public_frozen_and_account_qualified() -> None: + from adcp.reporting.ledger.models import ReportingConfigurationGenerationKey as ModelKey + + config = configuration() + key = ReportingConfigurationGenerationKey("acct_a", "daily", 1) + assert ModelKey is ReportingConfigurationGenerationKey + assert config.generation_key == obligation_for(config).generation_key == key + lease = LeasedConfiguration("acct_a", "daily", 1, NOW) + assert lease.generation_key == key + assert ( + len({key, configuration("acct_b").generation_key, replace(key, delivery_config_version=2)}) + == 3 + ) + with pytest.raises(FrozenInstanceError): + setattr(key, "account_id", "acct_b") + + +@pytest.mark.parametrize("accounts", [("acct_a", "acct_b"), ("Account", "account")]) +async def test_same_name_generations_keep_independent_content( + store: ReportingLedgerStore, accounts: tuple[str, str] +) -> None: + first, second = (configuration(account) for account in accounts) + second = replace( + second, + feed_purpose="billing", + required_finality="official", + schedule=replace(second.schedule, delivery_sla="PT4H"), + automated_recovery_window=timedelta(hours=12), + ) + await asyncio.gather(store.put_configuration(first), store.put_configuration(second)) + for config in (first, second): + assert await store.list_configurations(account_id=config.account_id) == (config,) + assert await store.list_configurations( + account_id=config.account_id, delivery_config_ids=["daily"] + ) == (config,) + assert ( + await store.list_configurations( + account_id=config.account_id, delivery_config_ids=["absent"] + ) + == () + ) + await asyncio.gather(*(store.put_configuration(config) for _ in range(4))) + with pytest.raises(LedgerConflictError) as caught: + await store.put_configuration(replace(config, media_buy_ids=("changed",))) + assert caught.value.code == "CONFIGURATION_GENERATION_IMMUTABLE" + assert await store.list_configurations(account_id="unavailable") == () + + new_version = replace(first, delivery_config_version=2, media_buy_ids=("new_buy",)) + await store.put_configuration(new_version) + assert set(await store.list_configurations(account_id=first.account_id)) == { + first, + new_version, + } + assert await store.list_configurations(account_id=second.account_id) == (second,) + + +async def test_concurrent_changed_writes_cannot_silently_succeed( + store: ReportingLedgerStore, +) -> None: + candidates = [replace(configuration(), media_buy_ids=(f"mb_{index}",)) for index in range(8)] + results = await asyncio.gather( + *(store.put_configuration(config) for config in candidates), return_exceptions=True + ) + assert sum(result is None for result in results) == 1 + for result in results: + if result is not None: + assert isinstance(result, LedgerConflictError) + assert result.code == "CONFIGURATION_GENERATION_IMMUTABLE" + winner = candidates[results.index(None)] + assert await store.list_configurations(account_id=winner.account_id) == (winner,) + + +async def test_concurrent_leases_and_releases_keep_accounts_separate( + store: ReportingLedgerStore, +) -> None: + configs = (configuration(), configuration("acct_b")) + await asyncio.gather(*(store.put_configuration(config) for config in configs)) + # Default producer ids may be shared. Releasing one tenant must not clear + # every same-name generation held by that worker. + leases = await asyncio.gather( + *(store.lease_period_close(worker_id="shared", now=NOW, lease_seconds=60) for _ in range(6)) + ) + held = [lease for lease in leases if lease is not None] + assert len(held) == 2 + assert {lease.generation_key for lease in held} == {config.generation_key for config in configs} + first, second = held + await asyncio.gather( + store.release_period_close(first, worker_id="shared"), + store.release_period_close(first, worker_id="shared"), + store.release_period_close(second, worker_id="wrong-worker"), + store.release_period_close(replace(second, account_id="unavailable"), worker_id="shared"), + ) + replacement = await store.lease_period_close(worker_id="next", now=NOW, lease_seconds=60) + assert replacement is not None and replacement.generation_key == first.generation_key + assert await store.lease_period_close(worker_id="extra", now=NOW, lease_seconds=60) is None + await asyncio.gather( + store.release_period_close(second, worker_id="shared"), + store.release_period_close(replacement, worker_id="next"), + ) + again = await asyncio.gather( + *(store.lease_period_close(worker_id="again", now=NOW, lease_seconds=60) for _ in range(2)) + ) + assert {lease.generation_key for lease in again if lease is not None} == { + config.generation_key for config in configs + } + + +async def test_an_expired_lease_cannot_release_a_replacement_with_the_same_worker_id( + store: ReportingLedgerStore, +) -> None: + await store.put_configuration(configuration()) + await store.put_configuration(configuration("acct_b")) + expired = await store.lease_period_close(worker_id="shared", now=NOW, lease_seconds=1) + other = await store.lease_period_close(worker_id="shared", now=NOW, lease_seconds=60) + assert expired is not None and other is not None + later = NOW + timedelta(seconds=5) + replacement = await store.lease_period_close(worker_id="shared", now=later, lease_seconds=60) + assert replacement is not None and replacement.generation_key == expired.generation_key + await store.release_period_close(expired, worker_id="shared") + assert await store.lease_period_close(worker_id="extra", now=later, lease_seconds=60) is None + await store.release_period_close(replacement, worker_id="shared") + reclaimed = await store.lease_period_close(worker_id="new", now=later, lease_seconds=60) + assert reclaimed is not None and reclaimed.generation_key == expired.generation_key + assert await store.lease_period_close(worker_id="extra", now=later, lease_seconds=60) is None + + +async def test_concurrent_period_closes_converge_within_each_account( + store: ReportingLedgerStore, +) -> None: + configs = (configuration(), configuration("acct_b")) + await asyncio.gather(*(store.put_configuration(config) for config in configs)) + producer = ReportingProducer( + source=UncalledSource(), offerings=ProducerOfferings(), store=store + ) + await asyncio.gather( + *(producer.close_elapsed_periods(config, now=NOW) for config in configs for _ in range(4)) + ) + ids = set() + for config in configs: + found = await store.find_obligation( + account_id=config.account_id, + delivery_config_id="daily", + delivery_config_version=1, + period_start=START, + period_end=END, + ) + assert found is not None + assert found.generation_key == config.generation_key + assert found.media_buy_ids == config.media_buy_ids + assert found.definition == config.definition + ids.add(found.reporting_obligation_id) + snapshot = await store.open_snapshot(account_id=config.account_id, filters_fingerprint="") + page = await store.read_page( + snapshot=snapshot, + consumer_id=None, + delivery_config_ids=["daily"], + media_buy_ids=None, + offset=0, + limit=10, + changes_after_sequence=None, + ) + assert page.total_count == 1 + assert page.obligations == (found,) + assert page.revisions == () + assert len(ids) == 2 + + +async def test_an_obligation_id_cannot_overwrite_another_accounts_period( + store: ReportingLedgerStore, +) -> None: + first, second = configuration(), configuration("acct_b") + await asyncio.gather(store.put_configuration(first), store.put_configuration(second)) + original = await store.commit_obligation(obligation_for(first)) + with pytest.raises(LedgerConflictError) as caught: + await store.commit_obligation( + replace( + obligation_for(second), reporting_obligation_id=original.reporting_obligation_id + ) + ) + assert caught.value.code == "OBLIGATION_IDENTITY_CONFLICT" + assert ( + await store.find_obligation( + account_id=first.account_id, + delivery_config_id="daily", + delivery_config_version=1, + period_start=START, + period_end=END, + ) + == original + ) + assert ( + await store.find_obligation( + account_id=second.account_id, + delivery_config_id="daily", + delivery_config_version=1, + period_start=START, + period_end=END, + ) + is None + ) + + +async def test_concurrent_workers_use_their_leased_generation( + store: ReportingLedgerStore, +) -> None: + configs = (configuration(), configuration("acct_b")) + await asyncio.gather(*(store.put_configuration(config) for config in configs)) + entered = {config.account_id: asyncio.Event() for config in configs} + finish = {config.account_id: asyncio.Event() for config in configs} + calls: list[ReportingSourceSliceRequestV1] = [] + + async def fetch(request: ReportingSourceSliceRequestV1) -> None: + calls.append(request) + entered[request.identity.account_id].set() + await finish[request.identity.account_id].wait() + + source = InlineReportingSource( + capabilities=redacted_capabilities(), fetch=fetch, clock=lambda: NOW + ) + producer = ReportingProducer( + source=source, + offerings=ProducerOfferings( + snapshot_offering_id="FIXTURE_PULSE_V1", + requested_dimensions=("campaign_id",), + source_scope=dict(source.capabilities.source_scope), + ), + store=store, + clock=lambda: NOW, + ) + tasks = [asyncio.create_task(producer.run_worker()) for _ in range(2)] + try: + await asyncio.wait_for(asyncio.gather(*(event.wait() for event in entered.values())), 10) + finish["acct_a"].set() + done, pending = await asyncio.wait(tasks, timeout=10, return_when=asyncio.FIRST_COMPLETED) + assert len(done) == len(pending) == 1 + finished = next(iter(done)).result() + assert finished.leased is not None and finished.leased.account_id == "acct_a" + probe = await store.lease_period_close(worker_id="probe", now=NOW, lease_seconds=60) + assert probe is not None and probe.generation_key == configs[0].generation_key + assert await store.lease_period_close(worker_id="extra", now=NOW, lease_seconds=60) is None + await store.release_period_close(probe, worker_id="probe") + finish["acct_b"].set() + turns = await asyncio.wait_for(asyncio.gather(*tasks), 10) + assert {turn.leased.generation_key for turn in turns if turn.leased} == { + config.generation_key for config in configs + } + for turn in turns: + assert len(turn.obligations_committed) == 1 + assert turn.slices_failed == turn.obligations_committed + assert len({call.identity.reporting_obligation_id for call in calls}) == 2 + for call in calls: + expected = next( + config for config in configs if config.account_id == call.identity.account_id + ) + assert call.identity.delivery_config_id == expected.delivery_config_id + assert call.identity.delivery_config_version == expected.delivery_config_version + assert [item.constituent_id for item in call.coverage.constituents] == list( + expected.media_buy_ids + ) + finally: + for task in tasks: + if not task.done(): + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + + +async def test_source_acquisition_rejects_a_different_account_or_generation( + store: ReportingLedgerStore, +) -> None: + config = configuration() + obligation = obligation_for(config) + producer = ReportingProducer( + source=UncalledSource(), offerings=ProducerOfferings(), store=store + ) + for foreign in (configuration("acct_b"), replace(config, delivery_config_version=2)): + with pytest.raises(LedgerConflictError) as caught: + await producer.acquire_obligation(foreign, obligation) + assert caught.value.code == "CONFIGURATION_GENERATION_MISMATCH" + + +@pytest.mark.parametrize("names_foreign_obligation", [False, True]) +async def test_missing_obligation_statements_attach_only_within_their_account( + store: ReportingLedgerStore, + names_foreign_obligation: bool, +) -> None: + first, second = configuration(), configuration("acct_b") + await asyncio.gather(store.put_configuration(first), store.put_configuration(second)) + foreign = await store.commit_obligation(obligation_for(second)) + statement = ConsumerStatusRecord( + reporting_status_id="rps_missing_a", + account_id=first.account_id, + consumer_id="shared-buyer", + delivery_config_id="daily", + delivery_config_version=1, + report_definition_id=first.report_definition_id, + period_start=START, + period_end=END, + period_source_timezone="UTC", + consumer_status="obligation_missing", + status_as_of=NOW, + recorded_at=NOW, + reporting_obligation_id=( + foreign.reporting_obligation_id if names_foreign_obligation else None + ), + ) + assert statement.generation_key == first.generation_key + await store.record_consumer_status(statement) + assert ( + await store.list_consumer_statuses( + account_id=first.account_id, + consumer_id=statement.consumer_id, + reporting_obligation_ids=[foreign.reporting_obligation_id], + ) + == () + ) + own = await store.commit_obligation(obligation_for(first)) + assert await store.list_consumer_statuses( + account_id=first.account_id, + consumer_id=statement.consumer_id, + reporting_obligation_ids=[own.reporting_obligation_id], + ) == (statement,) + assert ( + await store.list_consumer_statuses( + account_id=second.account_id, consumer_id=statement.consumer_id + ) + == () + ) + + +async def test_status_and_issue_projection_use_each_accounts_own_generation( + store: ReportingLedgerStore, +) -> None: + first, second = configuration(), configuration("acct_b") + first = replace(first, schedule=replace(first.schedule, delivery_sla="PT30M")) + second = replace(second, schedule=replace(second.schedule, delivery_sla="PT4H")) + await asyncio.gather(store.put_configuration(first), store.put_configuration(second)) + for config in (first, second): + obligation = await store.commit_obligation(obligation_for(config)) + original, rows = revision_for(obligation) + await store.commit_revision(original, rows) + await store.record_consumer_status( + ConsumerStatusRecord( + reporting_status_id=f"rps_{config.account_id}", + account_id=config.account_id, + consumer_id="shared-buyer", + delivery_config_id="daily", + delivery_config_version=1, + report_definition_id=config.report_definition_id, + period_start=START, + period_end=END, + period_source_timezone="UTC", + consumer_status="received", + status_as_of=END, + recorded_at=END, + reporting_obligation_id=obligation.reporting_obligation_id, + reporting_revision_id=original.reporting_revision_id, + observed_revision_content_sha256=original.revision_content_sha256, + ) + ) + restatement, rows = revision_for(obligation, suffix="restated") + await store.commit_revision( + replace( + restatement, + supersedes_reporting_revision_id=original.reporting_revision_id, + created_at=END + timedelta(minutes=10), + ), + rows, + ) + + handler = ReportingStatusHandler(store, consumer_status_enabled=True) + callers = [ + ReportingStatusCaller(config.account_id, "shared-buyer") for config in (first, second) + ] + summaries = await asyncio.gather( + *(handler.handle({"view": "summary"}, caller=caller) for caller in callers) + ) + assert [summary["health"] for summary in summaries] == ["action_required", "delayed"] + assert summaries[0]["issues"][0]["issue_id"] != summaries[1]["issues"][0]["issue_id"] + for config, caller, summary in zip((first, second), callers, summaries): + assert summary["account_id"] == config.account_id + assert summary["coverage"]["media_buy_ids"] == list(config.media_buy_ids) + assert summary["obligation_counts"]["total"] == 1 + periods = await handler.handle( + {"view": "periods", "delivery_config_ids": ["daily"]}, caller=caller + ) + assert periods["pagination"]["total_count"] == 4 + assert periods["periods"][0]["health"] == summary["health"] + assert periods["periods"][0]["account_id"] == config.account_id + assert {revision["account_id"] for revision in periods["revisions"]} == {config.account_id} + assert [status["reporting_status_id"] for status in periods["consumer_statuses"]] == [ + f"rps_{config.account_id}" + ] + assert config.definition is not None + for revision in periods["revisions"]: + assert revision["report_definition_uri"] == config.definition.report_definition_uri + own_revision_id = f"rpr_{config.account_id}_restated" + exact = await handler.handle( + {"view": "revision", "reporting_revision_id": own_revision_id}, caller=caller + ) + assert exact["revision"]["media_buy_ids"] == list(config.media_buy_ids) + assert ( + await store.get_revision( + account_id="unavailable", reporting_revision_id=own_revision_id + ) + is None + ) + + first_key = consumer_mismatch_issue_key( + account_id=first.account_id, + consumer_id="shared-buyer", + delivery_config_id="daily", + delivery_config_version=1, + report_definition_id=first.report_definition_id, + period_start=START, + period_end=END, + ) + assert await store.get_issue(issue_key=first_key, account_id=second.account_id) is None + await store.set_issue_state( + issue_key=first_key, + account_id=first.account_id, + state="waived", + at=NOW, + external_ref="private-case-a", + ) + after = await asyncio.gather( + *(handler.handle({"view": "summary"}, caller=caller) for caller in callers) + ) + assert after[0]["issues"] == [] + assert after[1]["issues"] == summaries[1]["issues"] + for unavailable_id in ("rpr_acct_b_restated", "nonexistent"): + with pytest.raises(LedgerConflictError) as caught: + await handler.handle( + {"view": "revision", "reporting_revision_id": unavailable_id}, caller=callers[0] + ) + assert caught.value.code == "LOOKUP_UNAVAILABLE" + + +async def test_ingest_resolves_the_accounts_own_configuration( + store: ReportingLedgerStore, +) -> None: + configs = ( + configuration(), + replace(configuration("acct_b"), report_definition_id="other_definition"), + ) + await asyncio.gather(*(store.put_configuration(config) for config in configs)) + ingest = ConsumerStatusIngest(store, enabled=True, clock=lambda: NOW) + for config in configs: + result = await ingest.handle( + { + "statuses": [ + { + "reporting_status_id": f"rps_{config.account_id}_missing", + "delivery_config_id": "daily", + "delivery_config_version": 1, + "report_definition_id": config.report_definition_id, + "period": { + "start": START.isoformat(), + "end": END.isoformat(), + "source_timezone": "UTC", + }, + "consumer_status": "obligation_missing", + "status_as_of": NOW.isoformat(), + } + ] + }, + account_id=config.account_id, + consumer_id="shared-buyer", + ) + assert result["results"][0]["result"] == "recorded" diff --git a/tests/conformance/reporting/test_reporting_generation_migration.py b/tests/conformance/reporting/test_reporting_generation_migration.py new file mode 100644 index 000000000..888b5225a --- /dev/null +++ b/tests/conformance/reporting/test_reporting_generation_migration.py @@ -0,0 +1,278 @@ +"""Upgrade literal beta.15 tables and retained evidence on real PostgreSQL.""" + +from __future__ import annotations + +import asyncio +import hashlib +from dataclasses import replace +from datetime import timedelta +from importlib.resources import files +from pathlib import Path +from typing import TYPE_CHECKING, Any + +import pytest + +from adcp.reporting.ledger import LeasedConfiguration, LedgerConflictError +from adcp.reporting.ledger.pg import PgReportingLedgerStore +from tests.conformance.reporting._generation_support import ( + NOW, + configuration, + isolated_reporting_pool, + obligation_for, +) + +if TYPE_CHECKING: + from psycopg_pool import AsyncConnectionPool + +_FIXTURES = Path(__file__).resolve().parents[2] / "fixtures" +_BETA15_SCHEMA = _FIXTURES / "reporting_ledger_beta15.sql" +_BETA15_DATA = _FIXTURES / "reporting_ledger_beta15_data.sql" +_MIGRATION = files("adcp.reporting.ledger").joinpath("reporting_ledger_account_generations.sql") +_TABLES = ( + "reporting_configurations", + "reporting_obligations", + "reporting_revisions", + "reporting_revision_rows", + "reporting_adjustments", + "reporting_consumer_statuses", + "reporting_issue_lifecycle", + "reporting_ledger_changes", +) + + +def test_upgrade_fixture_is_the_literal_beta15_schema() -> None: + # v8.0.0-beta.15:src/adcp/reporting/ledger/reporting_ledger.sql, byte for + # byte. Do not synthesize an "old" schema by editing the current DDL: that + # would allow future upgrades to pass without ever seeing released tables. + assert hashlib.sha256(_BETA15_SCHEMA.read_bytes()).hexdigest() == ( + "00b3dd643cf338a2a8ffb1ac98b3acd49ca6666b82a76b8f5c551f557c968371" + ) + + +async def _load_beta15(pool: AsyncConnectionPool) -> None: + async with pool.connection() as connection: + await connection.execute(_BETA15_SCHEMA.read_text()) + await connection.execute(_BETA15_DATA.read_text()) + + +async def _raw_upgrade(pool: AsyncConnectionPool) -> None: + async with pool.connection() as connection: + await connection.execute(_MIGRATION.read_text()) + + +async def _retained_rows(pool: AsyncConnectionPool) -> dict[str, list[Any]]: + from psycopg import sql + + result = {} + async with pool.connection() as connection: + for table in _TABLES: + rows = await ( + await connection.execute( + sql.SQL("SELECT to_jsonb(t) FROM {} t ORDER BY to_jsonb(t)::text").format( + sql.Identifier(table) + ) + ) + ).fetchall() + result[table] = [row[0] for row in rows] + return result + + +async def _primary_key(pool: AsyncConnectionPool) -> tuple[Any, ...]: + async with pool.connection() as connection: + row = await ( + await connection.execute( + "SELECT conname, oid, conindid, pg_get_constraintdef(oid) FROM pg_constraint" + " WHERE conrelid = 'reporting_configurations'::regclass AND contype = 'p'" + ) + ).fetchone() + assert row is not None + return row + + +async def _other_constraints(pool: AsyncConnectionPool) -> list[Any]: + """Keep every unrelated constraint/index, including its physical identity.""" + async with pool.connection() as connection: + constraints = await ( + await connection.execute( + "SELECT oid, conname, pg_get_constraintdef(oid) FROM pg_constraint" + " WHERE connamespace = current_schema()::regnamespace" + " AND NOT (conrelid = 'reporting_configurations'::regclass AND contype = 'p')" + " ORDER BY oid" + ) + ).fetchall() + indexes = await ( + await connection.execute( + "SELECT indexrelid, pg_get_indexdef(indexrelid) FROM pg_index" + " JOIN pg_class ON pg_class.oid = indrelid" + " WHERE relnamespace = current_schema()::regnamespace" + " AND NOT (indrelid = 'reporting_configurations'::regclass AND indisprimary)" + " ORDER BY indexrelid" + ) + ).fetchall() + return [constraints, indexes] + + +async def test_beta15_upgrade_preserves_all_evidence_and_survives_concurrent_boots() -> None: + async with isolated_reporting_pool() as pool: + await _load_beta15(pool) + before = await _retained_rows(pool) + assert all(before.values()), "The fixture must contain evidence in every beta.15 table" + constraints = await _other_constraints(pool) + assert (await _primary_key(pool))[ + 3 + ] == "PRIMARY KEY (delivery_config_id, delivery_config_version)" + # An application boot and a deployment migration may race. Both entry + # points must share a lock and converge without rebuilding the key twice. + await asyncio.gather( + *(PgReportingLedgerStore(pool=pool).create_schema() for _ in range(4)), + *(_raw_upgrade(pool) for _ in range(4)), + ) + upgraded_key = await _primary_key(pool) + assert ( + upgraded_key[3] + == "PRIMARY KEY (account_id, delivery_config_id, delivery_config_version)" + ) + assert await _retained_rows(pool) == before + assert await _other_constraints(pool) == constraints + await asyncio.gather(_raw_upgrade(pool), PgReportingLedgerStore(pool=pool).create_schema()) + assert await _primary_key(pool) == upgraded_key + + store = PgReportingLedgerStore(pool=pool, clock=lambda: NOW) + old = replace(configuration(), required_finality="official") + assert await store.list_configurations(account_id=old.account_id) == (old,) + await store.put_configuration(old) # Retained beta.15 configuration digest still replays. + with pytest.raises(LedgerConflictError) as caught: + await store.put_configuration(replace(old, media_buy_ids=("changed",))) + assert caught.value.code == "CONFIGURATION_GENERATION_IMMUTABLE" + revision = await store.get_revision( + account_id=old.account_id, reporting_revision_id="rpr_acct_a_official" + ) + assert revision is not None and revision.finality == "official" + rows = await store.read_revision_rows( + account_id=old.account_id, reporting_revision_id=revision.reporting_revision_id + ) + assert rows.rows == ({"media_buy_id": "mb_acct_a", "impressions": 5},) + assert await store.commit_revision(revision, rows.rows) == revision + statuses = await store.list_consumer_statuses( + account_id=old.account_id, consumer_id="shared-buyer" + ) + assert len(statuses) == 1 and statuses[0].mismatch_code == "metric_missing" + assert await store.record_consumer_status(statuses[0]) == (statuses[0], False) + assert await _retained_rows(pool) == before # Replays append no new feed entries. + + # Existing leases survive, and releasing an old handle must not release + # a new tenant's same-name generation even if the worker id is reused. + assert await store.lease_period_close(worker_id="extra", now=NOW, lease_seconds=60) is None + other = configuration("acct_b") + await store.put_configuration(other) + other_lease = await store.lease_period_close( + worker_id="beta15-worker", now=NOW, lease_seconds=3600 + ) + assert other_lease is not None and other_lease.generation_key == other.generation_key + old_lease = LeasedConfiguration("acct_a", "daily", 1, NOW + timedelta(hours=1)) + await store.release_period_close(old_lease, worker_id="beta15-worker") + reclaimed = await store.lease_period_close( + worker_id="new-worker", now=NOW, lease_seconds=60 + ) + assert reclaimed is not None and reclaimed.generation_key == old.generation_key + assert await store.lease_period_close(worker_id="extra", now=NOW, lease_seconds=60) is None + + # The retained feed and its sequence continue; no history is renumbered. + checkpoint = await store.open_snapshot(account_id=old.account_id, filters_fingerprint="") + assert checkpoint.max_sequence == 4 + await store.commit_obligation(obligation_for(other)) + new_checkpoint = await store.open_snapshot( + account_id=other.account_id, filters_fingerprint="" + ) + assert new_checkpoint.max_sequence == checkpoint.max_sequence + 1 + assert await store.list_configurations(account_id=old.account_id) == (old,) + + +@pytest.mark.parametrize("autocommit", [False, True]) +async def test_concurrent_bootstrap_creates_the_account_key(autocommit: bool) -> None: + async with isolated_reporting_pool(autocommit=autocommit) as pool: + await asyncio.gather(*(PgReportingLedgerStore(pool=pool).create_schema() for _ in range(8))) + primary = await _primary_key(pool) + assert primary[3] == "PRIMARY KEY (account_id, delivery_config_id, delivery_config_version)" + await PgReportingLedgerStore(pool=pool).create_schema() + assert await _primary_key(pool) == primary + + +async def test_standalone_migration_is_atomic_in_autocommit_mode() -> None: + async with isolated_reporting_pool(autocommit=True) as pool: + await _load_beta15(pool) + before = await _retained_rows(pool) + await asyncio.gather(*(_raw_upgrade(pool) for _ in range(8))) + assert (await _primary_key(pool))[ + 3 + ] == "PRIMARY KEY (account_id, delivery_config_id, delivery_config_version)" + assert await _retained_rows(pool) == before + + +async def test_upgrade_recognizes_a_renamed_beta15_primary_key() -> None: + async with isolated_reporting_pool() as pool: + await _load_beta15(pool) + async with pool.connection() as connection: + await connection.execute( + "ALTER TABLE reporting_configurations" + ' RENAME CONSTRAINT reporting_configurations_pkey TO "adopter key"' + ) + before = await _retained_rows(pool) + await PgReportingLedgerStore(pool=pool).create_schema() + primary = await _primary_key(pool) + assert primary[0] == "adopter key" + assert primary[3] == "PRIMARY KEY (account_id, delivery_config_id, delivery_config_version)" + assert await _retained_rows(pool) == before + await _raw_upgrade(pool) + assert await _primary_key(pool) == primary + + +async def test_upgrade_refuses_an_unexpected_primary_key_without_changing_rows() -> None: + async with isolated_reporting_pool() as pool: + from psycopg.errors import RaiseException + + await _load_beta15(pool) + async with pool.connection() as connection: + await connection.execute( + "ALTER TABLE reporting_configurations" + " DROP CONSTRAINT reporting_configurations_pkey," + " ADD PRIMARY KEY (account_id, delivery_config_id)" + ) + primary = await _primary_key(pool) + before = await _retained_rows(pool) + with pytest.raises(RaiseException, match="Unexpected reporting_configurations primary key"): + await PgReportingLedgerStore(pool=pool).create_schema() + assert await _primary_key(pool) == primary + assert await _retained_rows(pool) == before + + +async def test_upgrade_preserves_adopter_foreign_keys_and_rolls_back_on_failure() -> None: + async with isolated_reporting_pool() as pool: + from psycopg.errors import DependentObjectsStillExist + + await _load_beta15(pool) + async with pool.connection() as connection: + await connection.execute( + "CREATE TABLE adopter_reference (config_id TEXT, version INTEGER," + " FOREIGN KEY (config_id, version) REFERENCES reporting_configurations" + " (delivery_config_id, delivery_config_version))" + ) + await connection.execute("INSERT INTO adopter_reference VALUES ('daily', 1)") + primary = await _primary_key(pool) + constraints = await _other_constraints(pool) + before = await _retained_rows(pool) + with pytest.raises(DependentObjectsStillExist): + await PgReportingLedgerStore(pool=pool).create_schema() + assert await _primary_key(pool) == primary + assert await _other_constraints(pool) == constraints + assert await _retained_rows(pool) == before + async with pool.connection() as connection: + assert await ( + await connection.execute("SELECT * FROM adopter_reference") + ).fetchall() == [("daily", 1)] + # Adopter-owned remediation; the SDK must never do this by CASCADE. + await connection.execute("DROP TABLE adopter_reference") + await PgReportingLedgerStore(pool=pool).create_schema() + assert (await _primary_key(pool))[ + 3 + ] == "PRIMARY KEY (account_id, delivery_config_id, delivery_config_version)" diff --git a/tests/fixtures/reporting_ledger_beta15.sql b/tests/fixtures/reporting_ledger_beta15.sql new file mode 100644 index 000000000..d62b0a471 --- /dev/null +++ b/tests/fixtures/reporting_ledger_beta15.sql @@ -0,0 +1,284 @@ +-- AdCP Reliable Reporting ledger — durable obligations, revisions, and status. +-- +-- Run this once per deployment, or call +-- PgReportingLedgerStore.create_schema() from application code, which runs the +-- equivalent DDL idempotently on boot. +-- +-- COLLATE "C" on identifier columns avoids locale-dependent case folding — on +-- some locales "Buy-A" and "buy-a" compare equal, which would collapse two +-- distinct accounts or revisions into one. "C" is the byte-for-byte comparison +-- reporting evidence actually requires. +-- +-- Every immutable write also appends to reporting_ledger_changes in the same +-- transaction. That feed is what makes `changes_after` exact: a consumer that +-- persists a checkpoint and replays from it cannot miss a record or see the +-- same record twice under a different identity. + +CREATE TABLE IF NOT EXISTS reporting_configurations ( + delivery_config_id TEXT COLLATE "C" NOT NULL, + delivery_config_version INTEGER NOT NULL, + account_id TEXT COLLATE "C" NOT NULL, + report_definition_id TEXT COLLATE "C" NOT NULL, + reporting_profile TEXT NOT NULL, + feed_purpose TEXT NOT NULL, + required_finality TEXT NOT NULL, + account_timezone TEXT NOT NULL DEFAULT 'UTC', + schedule JSONB NOT NULL, + media_buy_ids JSONB NOT NULL DEFAULT '[]'::jsonb, + activated_at TIMESTAMPTZ, + deactivated_at TIMESTAMPTZ, + automated_recovery_seconds DOUBLE PRECISION NOT NULL DEFAULT 21600, + status_retention_days INTEGER NOT NULL DEFAULT 400, + -- Content-addressed report definition and row schema (URI + digest). Core + -- wire records are self-describing; without this a retained revision + -- cannot name what it was produced under. + definition JSONB, + -- Binds the whole generation so a re-put with changed content is a + -- detectable conflict rather than a silent edit of retained evidence. + content_sha256 TEXT COLLATE "C" NOT NULL, + -- Period-close leasing. A worker that dies mid-close releases its work by + -- expiry instead of wedging the period forever. + lease_worker_id TEXT COLLATE "C", + lease_expires_at TIMESTAMPTZ, + PRIMARY KEY (delivery_config_id, delivery_config_version) +); + +CREATE INDEX IF NOT EXISTS reporting_configurations_account_idx + ON reporting_configurations (account_id); + +-- Supports "lease the least recently worked generation" without a full scan. +CREATE INDEX IF NOT EXISTS reporting_configurations_lease_idx + ON reporting_configurations (lease_expires_at NULLS FIRST); + +CREATE TABLE IF NOT EXISTS reporting_obligations ( + reporting_obligation_id TEXT COLLATE "C" NOT NULL PRIMARY KEY, + account_id TEXT COLLATE "C" NOT NULL, + delivery_config_id TEXT COLLATE "C" NOT NULL, + delivery_config_version INTEGER NOT NULL, + report_definition_id TEXT COLLATE "C" NOT NULL, + reporting_profile TEXT NOT NULL, + feed_purpose TEXT NOT NULL, + period_key TEXT NOT NULL, + period_start TIMESTAMPTZ NOT NULL, + period_end TIMESTAMPTZ NOT NULL, + source_timezone TEXT NOT NULL, + expected_at TIMESTAMPTZ NOT NULL, + scope_resolved_at TIMESTAMPTZ NOT NULL, + automated_recovery_deadline_at TIMESTAMPTZ NOT NULL, + required_finality TEXT NOT NULL, + coverage_status TEXT NOT NULL DEFAULT 'full', + media_buy_ids JSONB NOT NULL DEFAULT '[]'::jsonb, + package_ids JSONB NOT NULL DEFAULT '[]'::jsonb, + schedule JSONB NOT NULL, + -- Frozen with the obligation, not read from the live configuration: a + -- definition that changes later must not retroactively re-describe a + -- period that already closed. + definition JSONB, + created_at TIMESTAMPTZ NOT NULL +); + +-- One obligation per logical period. Without this, two workers racing a period +-- close could commit two obligations and a seller could quietly publish twice +-- and pick a winner. +CREATE UNIQUE INDEX IF NOT EXISTS reporting_obligations_period_key + ON reporting_obligations + (account_id, delivery_config_id, delivery_config_version, period_start, period_end); + +CREATE INDEX IF NOT EXISTS reporting_obligations_account_idx + ON reporting_obligations (account_id, period_end DESC); + +CREATE TABLE IF NOT EXISTS reporting_revisions ( + reporting_revision_id TEXT COLLATE "C" NOT NULL PRIMARY KEY, + account_id TEXT COLLATE "C" NOT NULL, + reporting_obligation_id TEXT COLLATE "C" NOT NULL + REFERENCES reporting_obligations (reporting_obligation_id), + finality TEXT NOT NULL, + revision_content_sha256 TEXT COLLATE "C" NOT NULL, + row_count BIGINT NOT NULL, + control_totals JSONB NOT NULL DEFAULT '[]'::jsonb, + observed_at TIMESTAMPTZ NOT NULL, + data_through TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL, + supersedes_reporting_revision_id TEXT COLLATE "C", + finality_basis TEXT, + finality_policy_id TEXT, + finalized_at TIMESTAMPTZ, + -- Core promises a committed revision stays readable for + -- status_retention_days. `readable` records reality; `readable_at_commit` + -- remembers that it once was, so the health projection can name the + -- revision that opened the gap rather than an arbitrary one. + readable BOOLEAN NOT NULL DEFAULT TRUE, + readable_at_commit BOOLEAN NOT NULL DEFAULT TRUE, + source_publication_id TEXT COLLATE "C", + source_manifest_sha256 TEXT COLLATE "C", + -- Binds what was published, excluding mutable readability. + content_sha256 TEXT COLLATE "C" NOT NULL +); + +-- An official revision is terminal: at most one per obligation. A later source +-- correction is an adjustment, never a second official close. +CREATE UNIQUE INDEX IF NOT EXISTS reporting_revisions_one_official + ON reporting_revisions (reporting_obligation_id) + WHERE finality = 'official'; + +-- Supersession must not fork: a given revision may be superseded at most once. +CREATE UNIQUE INDEX IF NOT EXISTS reporting_revisions_one_successor + ON reporting_revisions (supersedes_reporting_revision_id) + WHERE supersedes_reporting_revision_id IS NOT NULL; + +CREATE INDEX IF NOT EXISTS reporting_revisions_obligation_idx + ON reporting_revisions (account_id, reporting_obligation_id); + +CREATE TABLE IF NOT EXISTS reporting_revision_rows ( + reporting_revision_id TEXT COLLATE "C" NOT NULL + REFERENCES reporting_revisions (reporting_revision_id), + ordinal BIGINT NOT NULL, + row_payload JSONB NOT NULL, + PRIMARY KEY (reporting_revision_id, ordinal) +); + +CREATE TABLE IF NOT EXISTS reporting_adjustments ( + reporting_adjustment_id TEXT COLLATE "C" NOT NULL PRIMARY KEY, + account_id TEXT COLLATE "C" NOT NULL, + adjusts_reporting_revision_id TEXT COLLATE "C" NOT NULL + REFERENCES reporting_revisions (reporting_revision_id), + reason_code TEXT NOT NULL, + reason_detail TEXT, + accounting_period_start TIMESTAMPTZ NOT NULL, + accounting_period_end TIMESTAMPTZ NOT NULL, + control_total_deltas JSONB NOT NULL DEFAULT '[]'::jsonb, + correction_observed_at TIMESTAMPTZ NOT NULL, + created_at TIMESTAMPTZ NOT NULL +); + +CREATE INDEX IF NOT EXISTS reporting_adjustments_revision_idx + ON reporting_adjustments (account_id, adjusts_reporting_revision_id); + +-- PREVIEW: sync_reporting_status. Created unconditionally because DDL is +-- cheap and a migration mid-rollout is not; the ingest that writes here is off +-- until an adopter enables it. See adcp/reporting/ledger/consumer_status.py. +CREATE TABLE IF NOT EXISTS reporting_consumer_statuses ( + reporting_status_id TEXT COLLATE "C" NOT NULL PRIMARY KEY, + account_id TEXT COLLATE "C" NOT NULL, + -- Derived from authenticated transport, never from the request body. + consumer_id TEXT COLLATE "C" NOT NULL, + delivery_config_id TEXT COLLATE "C" NOT NULL, + delivery_config_version INTEGER NOT NULL, + report_definition_id TEXT COLLATE "C" NOT NULL, + period_start TIMESTAMPTZ NOT NULL, + period_end TIMESTAMPTZ NOT NULL, + period_source_timezone TEXT NOT NULL, + consumer_status TEXT NOT NULL, + status_as_of TIMESTAMPTZ NOT NULL, + recorded_at TIMESTAMPTZ NOT NULL, + supersedes_reporting_status_id TEXT COLLATE "C", + -- Nullable by design: obligation_missing is filed precisely when the + -- seller's ledger omitted the period, so requiring a seller-issued + -- obligation id would make the first missing report invisible again. + reporting_obligation_id TEXT COLLATE "C", + reporting_revision_id TEXT COLLATE "C", + observed_revision_content_sha256 TEXT COLLATE "C", + failure_code TEXT, + consumer_commit_ref TEXT, + seller_ledger_snapshot_id TEXT, + seller_ledger_as_of TIMESTAMPTZ, + superseded BOOLEAN NOT NULL DEFAULT FALSE, + content_sha256 TEXT COLLATE "C" NOT NULL +); + +-- Exactly one unsuperseded leaf per logical chain. This is what makes +-- supersession atomic: a concurrent update naming a stale leaf hits this +-- constraint instead of forking the chain, so a successful retry cannot erase +-- a recorded outage. +CREATE UNIQUE INDEX IF NOT EXISTS reporting_consumer_statuses_one_leaf + ON reporting_consumer_statuses + (account_id, consumer_id, delivery_config_id, delivery_config_version, + report_definition_id, period_start, period_end) + WHERE superseded = FALSE; + +CREATE UNIQUE INDEX IF NOT EXISTS reporting_consumer_statuses_one_successor + ON reporting_consumer_statuses (supersedes_reporting_status_id) + WHERE supersedes_reporting_status_id IS NOT NULL; + +CREATE INDEX IF NOT EXISTS reporting_consumer_statuses_chain_idx + ON reporting_consumer_statuses (account_id, consumer_id, period_end DESC); + +-- AdCP 3.2.0-rc.3 additions to tables created by an earlier SDK. ADD COLUMN +-- IF NOT EXISTS keeps create_schema() an upgrade path, not just a bootstrap: +-- an adopter that installed the rc.2 schema gets these on the next boot +-- without a hand-written migration, and a fresh install is unaffected. +ALTER TABLE reporting_consumer_statuses + ADD COLUMN IF NOT EXISTS mismatch_code TEXT; + +ALTER TABLE reporting_configurations + ADD COLUMN IF NOT EXISTS authoritative_party TEXT NOT NULL DEFAULT 'seller'; + +-- Durable issue lifecycle. Every other issue in this ledger has a *derived* +-- identity, because the conditions they name are monotone for one immutable +-- obligation: once a qualifying revision is associated, REPORT_OVERDUE cannot +-- recur. A consumer mismatch is not monotone -- the buyer can supersede, the +-- seller can restate, the disagreement can clear and come back -- and rc.3 +-- requires opened_at to survive every re-emission because it anchors the +-- escalation clock. A derived timestamp would reset on each poll and an +-- unattended mismatch would never escalate. +-- +-- issue_key identifies the condition; issue_id identifies one occurrence of +-- it. Retirement bumps generation, so a recurrence after resolved/waived gets +-- a new issue_id and a new opened_at, while an unresolved condition keeps both +-- across a severity change from delayed to action_required. +CREATE TABLE IF NOT EXISTS reporting_issue_lifecycle ( + issue_key TEXT COLLATE "C" NOT NULL, + account_id TEXT COLLATE "C" NOT NULL, + generation INTEGER NOT NULL, + issue_id TEXT COLLATE "C" NOT NULL, + -- Caller-scoped issues (every consumer mismatch) carry the consumer whose + -- statement caused them; NULL is a seller-wide condition. + consumer_id TEXT COLLATE "C", + opened_at TIMESTAMPTZ NOT NULL, + issue_state TEXT NOT NULL DEFAULT 'open', + -- Inert correlation text for the party's own tracker. Never dereferenced. + external_ref TEXT, + retired_at TIMESTAMPTZ, + PRIMARY KEY (account_id, issue_key, generation) +); + +-- At most one live occurrence per condition. 'waived' counts as live: waiving +-- records an agreement to stop *acting*, not a finding that the reporting is +-- fine, so the occurrence must keep blocking a new one -- otherwise the next +-- poll would open a fresh occurrence and republish the very issue the parties +-- agreed to stop acting on. Only 'resolved' frees the condition to recur. +-- +-- This is also what makes +-- ensure_issue_opened() safe under concurrent reads: two readers of the same +-- condition collide on this index and converge on one row instead of opening +-- two occurrences with two different opened_at values -- which would give the +-- same disagreement two escalation clocks. +CREATE UNIQUE INDEX IF NOT EXISTS reporting_issue_lifecycle_one_live + ON reporting_issue_lifecycle (account_id, issue_key) + WHERE issue_state IN ('open', 'acknowledged', 'waived'); + +CREATE UNIQUE INDEX IF NOT EXISTS reporting_issue_lifecycle_issue_id + ON reporting_issue_lifecycle (issue_id); + +-- The per-account change feed. `seq` orders every immutable record across +-- kinds so `changes_after` is exact. +-- +-- Appends take a transaction-scoped advisory lock on the account, which makes +-- sequence order equal commit order *within an account*. Without it, a +-- transaction that grabbed a low sequence but committed late would be invisible +-- to a consumer that had already checkpointed past it — a silently lost record, +-- which is the one failure a reporting ledger must not have. Contention is +-- per-account and the appends are short. +CREATE TABLE IF NOT EXISTS reporting_ledger_changes ( + seq BIGSERIAL NOT NULL PRIMARY KEY, + account_id TEXT COLLATE "C" NOT NULL, + record_kind TEXT NOT NULL, + record_id TEXT COLLATE "C" NOT NULL, + committed_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE UNIQUE INDEX IF NOT EXISTS reporting_ledger_changes_record + ON reporting_ledger_changes (account_id, record_kind, record_id); + +CREATE INDEX IF NOT EXISTS reporting_ledger_changes_feed_idx + ON reporting_ledger_changes (account_id, seq); diff --git a/tests/fixtures/reporting_ledger_beta15_data.sql b/tests/fixtures/reporting_ledger_beta15_data.sql new file mode 100644 index 000000000..af9142d30 --- /dev/null +++ b/tests/fixtures/reporting_ledger_beta15_data.sql @@ -0,0 +1,26 @@ +-- Literal rows written by the PgReportingLedgerStore from v8.0.0-beta.15. +-- Captured against reporting_ledger_beta15.sql; keep hashes and evidence unchanged. + +INSERT INTO reporting_configurations (delivery_config_id, delivery_config_version, account_id, report_definition_id, reporting_profile, feed_purpose, required_finality, account_timezone, schedule, media_buy_ids, activated_at, deactivated_at, automated_recovery_seconds, status_retention_days, definition, content_sha256, lease_worker_id, lease_expires_at, authoritative_party) VALUES ('daily', 1, 'acct_a', 'hourly_delivery', 'paid_media_delivery', 'analytics', 'official', 'UTC', '{"alignment": "utc", "delivery_sla": "PT1H", "period_anchor": "2026-09-01T00:00:00+00:00", "period_duration": "PT1H", "period_timezone": null}', '["mb_acct_a"]', '2026-09-01 00:00:00+00', '2026-09-01 01:00:00+00', 21600, 400, '{"schema_uri": "https://contracts.example.test/acct_a/rows.json", "schema_sha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", "schema_dialect": "https://json-schema.org/draft/2020-12/schema", "schema_version": "1.0.0", "schema_ref_policy": "local_fragment_only", "report_definition_uri": "https://contracts.example.test/acct_a/hourly", "report_definition_sha256": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"}', '6c6905c52eb6468c25b1968896a549196390c4d4ebafbe0505c4d07771e51f75', 'beta15-worker', '2026-09-01 04:00:00+00', 'seller'); + +INSERT INTO reporting_obligations (reporting_obligation_id, account_id, delivery_config_id, delivery_config_version, report_definition_id, reporting_profile, feed_purpose, period_key, period_start, period_end, source_timezone, expected_at, scope_resolved_at, automated_recovery_deadline_at, required_finality, coverage_status, media_buy_ids, package_ids, schedule, definition, created_at) VALUES ('rpo_acct_a', 'acct_a', 'daily', 1, 'hourly_delivery', 'paid_media_delivery', 'analytics', '2026-09-01T00:00:00Z_PT1H', '2026-09-01 00:00:00+00', '2026-09-01 01:00:00+00', 'UTC', '2026-09-01 02:00:00+00', '2026-09-01 01:00:00+00', '2026-09-01 08:00:00+00', 'official', 'full', '["mb_acct_a"]', '[]', '{"alignment": "utc", "delivery_sla": "PT1H", "period_anchor": "2026-09-01T00:00:00+00:00", "period_duration": "PT1H", "period_timezone": null}', '{"schema_uri": "https://contracts.example.test/acct_a/rows.json", "schema_sha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", "schema_dialect": "https://json-schema.org/draft/2020-12/schema", "schema_version": "1.0.0", "schema_ref_policy": "local_fragment_only", "report_definition_uri": "https://contracts.example.test/acct_a/hourly", "report_definition_sha256": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"}', '2026-09-01 01:00:00+00'); + +INSERT INTO reporting_revisions (reporting_revision_id, account_id, reporting_obligation_id, finality, revision_content_sha256, row_count, control_totals, observed_at, data_through, created_at, supersedes_reporting_revision_id, finality_basis, finality_policy_id, finalized_at, readable, readable_at_commit, source_publication_id, source_manifest_sha256, content_sha256) VALUES ('rpr_acct_a_official', 'acct_a', 'rpo_acct_a', 'official', 'eef1a4929f5c984afc191b71ccbae9374a0e88ba63c0b0e97dc46604ada6c555', 1, '[["impressions", "5"]]', '2026-09-01 01:00:00+00', '2026-09-01 01:00:00+00', '2026-09-01 01:00:00+00', NULL, 'source_final', 'policy_beta15', '2026-09-01 01:00:00+00', true, true, NULL, NULL, '0c9bd628cd43fae9bee3bd12bdf8a27374786392c8372a167bd73870efd0df6b'); + +INSERT INTO reporting_revision_rows (reporting_revision_id, ordinal, row_payload) VALUES ('rpr_acct_a_official', 0, '{"impressions": 5, "media_buy_id": "mb_acct_a"}'); + +INSERT INTO reporting_adjustments (reporting_adjustment_id, account_id, adjusts_reporting_revision_id, reason_code, reason_detail, accounting_period_start, accounting_period_end, control_total_deltas, correction_observed_at, created_at) VALUES ('rpa_beta15', 'acct_a', 'rpr_acct_a_official', 'source_correction', 'retained beta.15 correction', '2026-09-01 00:00:00+00', '2026-09-01 01:00:00+00', '[["impressions", "1"]]', '2026-09-01 03:00:00+00', '2026-09-01 03:00:00+00'); + +INSERT INTO reporting_consumer_statuses (reporting_status_id, account_id, consumer_id, delivery_config_id, delivery_config_version, report_definition_id, period_start, period_end, period_source_timezone, consumer_status, status_as_of, recorded_at, supersedes_reporting_status_id, reporting_obligation_id, reporting_revision_id, observed_revision_content_sha256, failure_code, consumer_commit_ref, seller_ledger_snapshot_id, seller_ledger_as_of, superseded, content_sha256, mismatch_code) VALUES ('rps_beta15', 'acct_a', 'shared-buyer', 'daily', 1, 'hourly_delivery', '2026-09-01 00:00:00+00', '2026-09-01 01:00:00+00', 'UTC', 'content_mismatch', '2026-09-01 03:00:00+00', '2026-09-01 03:00:00+00', NULL, 'rpo_acct_a', 'rpr_acct_a_official', 'eef1a4929f5c984afc191b71ccbae9374a0e88ba63c0b0e97dc46604ada6c555', NULL, 'beta15-import', NULL, NULL, false, 'fe2c6360c22375e1643b8ef422d186c52033f5591d6877925f5708c40e2c9d91', 'metric_missing'); + +INSERT INTO reporting_issue_lifecycle (issue_key, account_id, generation, issue_id, consumer_id, opened_at, issue_state, external_ref, retired_at) VALUES ('rpik_e144c72e2fd9305cace514d0ab00bf7cfa8591dd', 'acct_a', 1, 'rpti_R5cyZqx0bqNHZ7A954F3r1Wsxgks6eRc', 'shared-buyer', '2026-09-01 03:00:00+00', 'acknowledged', 'beta15-ticket', NULL); + +INSERT INTO reporting_ledger_changes (seq, account_id, record_kind, record_id, committed_at) VALUES (1, 'acct_a', 'obligation', 'rpo_acct_a', '2026-09-16 00:03:09.149226+00'); + +INSERT INTO reporting_ledger_changes (seq, account_id, record_kind, record_id, committed_at) VALUES (2, 'acct_a', 'revision', 'rpr_acct_a_official', '2026-09-16 00:03:09.151596+00'); + +INSERT INTO reporting_ledger_changes (seq, account_id, record_kind, record_id, committed_at) VALUES (3, 'acct_a', 'adjustment', 'rpa_beta15', '2026-09-16 00:03:09.15454+00'); + +INSERT INTO reporting_ledger_changes (seq, account_id, record_kind, record_id, committed_at) VALUES (4, 'acct_a', 'consumer_status', 'rps_beta15', '2026-09-16 00:03:09.156439+00'); + +SELECT pg_catalog.setval('reporting_ledger_changes_seq_seq', 4, true); diff --git a/tests/type_checks/reporting_generation_keys.py b/tests/type_checks/reporting_generation_keys.py new file mode 100644 index 000000000..a3020218b --- /dev/null +++ b/tests/type_checks/reporting_generation_keys.py @@ -0,0 +1,56 @@ +"""Account-qualified reporting identity without changing low-level adopter calls.""" + +from __future__ import annotations + +from datetime import datetime + +from typing_extensions import assert_type + +from adcp.reporting.ledger import ( + ConsumerStatusRecord, + LeasedConfiguration, + ReportingConfiguration, + ReportingConfigurationGenerationKey, + ReportingLedgerStore, + ReportingObligationRecord, +) + + +async def inspect_generation( + store: ReportingLedgerStore, + configuration: ReportingConfiguration, + obligation: ReportingObligationRecord, + status: ConsumerStatusRecord, + lease: LeasedConfiguration, + now: datetime, +) -> None: + key = assert_type(configuration.generation_key, ReportingConfigurationGenerationKey) + assert_type(obligation.generation_key, ReportingConfigurationGenerationKey) + assert_type(status.generation_key, ReportingConfigurationGenerationKey) + assert_type(lease.generation_key, ReportingConfigurationGenerationKey) + assert_type(key.account_id, str) + assert_type(key.delivery_config_id, str) + assert_type(key.delivery_config_version, int) + generations: dict[ReportingConfigurationGenerationKey, ReportingConfiguration] = { + key: configuration + } + assert_type(generations.get(obligation.generation_key), ReportingConfiguration | None) + assert_type( + await store.find_obligation( + account_id=key.account_id, + delivery_config_id=key.delivery_config_id, + delivery_config_version=key.delivery_config_version, + period_start=obligation.period.start, + period_end=obligation.period.end, + ), + ReportingObligationRecord | None, + ) + assert_type( + await store.lease_period_close(worker_id="worker", now=now, lease_seconds=60), + LeasedConfiguration | None, + ) + # The beta.15 lease constructor and release call remain valid. + retained_handle = LeasedConfiguration( + key.account_id, key.delivery_config_id, key.delivery_config_version, lease.lease_expires_at + ) + await store.release_period_close(retained_handle, worker_id="worker") From 6c74a458b8ee2a34c47c154186b950faed50f4fa Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Wed, 16 Sep 2026 01:00:23 +0000 Subject: [PATCH 2/2] fix(reporting): rotate in-memory period-close leases across accounts InMemoryReportingLedgerStore.lease_period_close handed back the first leasable generation in insertion order, and release_period_close puts that generation straight back into the free pool. ReportingProducer.run_worker releases in a finally, so a worker loop re-leased the same generation on every turn and never closed a period for any other account. The SQL store orders by lease_expires_at NULLS FIRST, and an update to that indexed column sends a released row to the back of the queue, so the two stores only diverged once two accounts could hold the same delivery_config_id -- which is what this branch enables. Rank leasable generations the way the SQL store does, unheld before expired and oldest expiry first, then break the tie by whichever generation went longest without a turn. The shared memory/PostgreSQL suite now asserts that a worker releasing every turn reaches all three accounts; it failed on memory and passed on PostgreSQL before this change. Co-Authored-By: Claude Opus 5 (1M context) --- src/adcp/reporting/ledger/store.py | 49 ++++++++++++++----- .../test_reporting_generation_identity.py | 23 +++++++++ 2 files changed, 60 insertions(+), 12 deletions(-) diff --git a/src/adcp/reporting/ledger/store.py b/src/adcp/reporting/ledger/store.py index 6724a5652..b190f879d 100644 --- a/src/adcp/reporting/ledger/store.py +++ b/src/adcp/reporting/ledger/store.py @@ -546,6 +546,11 @@ def __init__(self, *, clock: Callable[[], datetime] | None = None) -> None: self._changes: list[tuple[int, str, LedgerRecordKind, str, datetime]] = [] self._sequence = 0 self._leases: dict[ReportingConfigurationGenerationKey, tuple[str, datetime]] = {} + # When each generation was last handed to a worker, so releasing a + # lease sends that generation to the back of the queue instead of + # letting it win every turn. + self._lease_turns: dict[ReportingConfigurationGenerationKey, int] = {} + self._lease_turn = 0 # Live occurrence per (account, issue_key), plus the retired generation # high-water mark so a recurrence never reuses an id. self._issues: dict[tuple[str, str], ReportingIssueLifecycle] = {} @@ -1104,19 +1109,39 @@ async def lease_period_close( from datetime import timedelta async with self._lock: - for key, configuration in self._configurations.items(): + moment = _utc(now) + # Rank leasable generations the way the SQL store's + # `ORDER BY lease_expires_at NULLS FIRST` does -- unheld before + # expired, oldest expiry first -- then break the tie by whichever + # generation went longest without a turn. Without that last term a + # worker that releases at the end of every turn re-leases the same + # generation forever, and every other account's periods are never + # closed: starvation that only appears once two accounts can hold + # the same delivery_config_id. + ranked: list[tuple[tuple[int, float, int], ReportingConfigurationGenerationKey]] = [] + for key in self._configurations: + turn = self._lease_turns.get(key, 0) held = self._leases.get(key) - if held is not None and _utc(held[1]) > _utc(now): - continue - expires = _utc(now) + timedelta(seconds=lease_seconds) - self._leases[key] = (worker_id, expires) - return LeasedConfiguration( - account_id=configuration.account_id, - delivery_config_id=configuration.delivery_config_id, - delivery_config_version=configuration.delivery_config_version, - lease_expires_at=expires, - ) - return None + if held is None: + ranked.append(((0, 0.0, turn), key)) + elif _utc(held[1]) <= moment: + ranked.append(((1, _utc(held[1]).timestamp(), turn), key)) + if not ranked: + return None + # `min` keeps the first of equal ranks, so generations that have + # never been leased are handed out in the order they were accepted. + key = min(ranked, key=lambda item: item[0])[1] + configuration = self._configurations[key] + expires = moment + timedelta(seconds=lease_seconds) + self._lease_turn += 1 + self._lease_turns[key] = self._lease_turn + self._leases[key] = (worker_id, expires) + return LeasedConfiguration( + account_id=configuration.account_id, + delivery_config_id=configuration.delivery_config_id, + delivery_config_version=configuration.delivery_config_version, + lease_expires_at=expires, + ) async def release_period_close(self, lease: LeasedConfiguration, *, worker_id: str) -> None: async with self._lock: diff --git a/tests/conformance/reporting/test_reporting_generation_identity.py b/tests/conformance/reporting/test_reporting_generation_identity.py index 5e5d4031e..644bcffbe 100644 --- a/tests/conformance/reporting/test_reporting_generation_identity.py +++ b/tests/conformance/reporting/test_reporting_generation_identity.py @@ -176,6 +176,29 @@ async def test_an_expired_lease_cannot_release_a_replacement_with_the_same_worke assert await store.lease_period_close(worker_id="extra", now=later, lease_seconds=60) is None +async def test_a_worker_that_releases_each_turn_reaches_every_accounts_generation( + store: ReportingLedgerStore, +) -> None: + """A single worker loop must not starve the accounts it did not lease first. + + ``ReportingProducer.run_worker`` releases in a ``finally``, so a store that + always hands back the first leasable generation would close periods for one + account forever and never reach the others -- invisible until two accounts + share a ``delivery_config_id``, which is exactly what this change allows. + """ + accounts = ("acct_a", "acct_b", "acct_c") + await asyncio.gather(*(store.put_configuration(configuration(name)) for name in accounts)) + worked: list[str] = [] + for _ in range(len(accounts) * 3): + lease = await store.lease_period_close(worker_id="solo", now=NOW, lease_seconds=60) + assert lease is not None + assert lease.generation_key == configuration(lease.account_id).generation_key + worked.append(lease.account_id) + await store.release_period_close(lease, worker_id="solo") + assert set(worked) == set(accounts) + assert set(worked[: len(accounts)]) == set(accounts) + + async def test_concurrent_period_closes_converge_within_each_account( store: ReportingLedgerStore, ) -> None: