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..b190f879d 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,12 @@ 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]] = {} + # 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] = {} @@ -561,7 +577,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 +601,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 +634,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 +888,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 +1036,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 @@ -1084,25 +1109,45 @@ 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: - 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..644bcffbe --- /dev/null +++ b/tests/conformance/reporting/test_reporting_generation_identity.py @@ -0,0 +1,544 @@ +"""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_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: + 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")