diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 19ddb7f93..dc83b1f3e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -76,13 +76,11 @@ jobs: # ruff/mypy/mypy --strict/the type-ignore contract before pytest starts, # then post-job cleanup after it ends. On an ubuntu-latest runner that # pre-test work is ~1m30s and the plain suite is ~16m, but 3.12 adds - # coverage tracing on ~9.4k tests: at 30 minutes that leg was cancelled - # mid-run at 99% with no failing test, which reports as a red matrix and - # hides real signal. Bound the suite itself below (so a hang fails one - # named step with its own message instead of silently taking the job), and - # leave the job enough room that step bound plus setup and cleanup still - # fit with margin for runner variance. - timeout-minutes: 60 + # coverage tracing on the full suite: even a 45-minute coverage step can + # time out near 99% on a loaded runner with no failing test. Bound the + # suite itself so a hang fails the named step. The 3.12 job keeps 15 + # minutes beyond that step for setup and cleanup; other jobs keep 60. + timeout-minutes: ${{ matrix.python-version == '3.12' && 70 || 60 }} strategy: matrix: python-version: ["3.10", "3.11", "3.12", "3.13"] @@ -118,10 +116,8 @@ jobs: if: matrix.python-version == '3.12' run: python scripts/check_type_ignore_contract.py - # Bounded well above the observed ~16m (plain) and ~30m (coverage) runs - # so ordinary variance never trips it, and well below the job ceiling so - # a genuinely stuck suite still fails *this* step with a timeout rather - # than being cancelled as a whole job. + # Bound the plain and coverage runs separately so a genuinely stuck + # suite fails its named step rather than cancelling the entire job. - name: Run tests if: matrix.python-version != '3.12' timeout-minutes: 45 @@ -129,19 +125,24 @@ jobs: - name: Run tests with coverage if: matrix.python-version == '3.12' - timeout-minutes: 45 + timeout-minutes: 55 run: python scripts/reporting_test_harness.py pytest tests/ -v -ra --cov=src/adcp --cov-report=term-missing pg-conformance: name: Postgres conformance tests (Postgres 16, ${{ matrix.lane }}) runs-on: ubuntu-latest - timeout-minutes: 15 + timeout-minutes: ${{ matrix.timeout_minutes }} permissions: contents: read strategy: fail-fast: false matrix: - lane: [core, process] + include: + # Core also builds and installs the reporting wheel/sdist fixtures. + - lane: core + timeout_minutes: 30 + - lane: process + timeout_minutes: 15 services: postgres: # CI-local ephemeral database. POSTGRES_HOST_AUTH_METHOD=trust @@ -184,8 +185,8 @@ jobs: ADCP_PG_TEST_URL: postgresql://postgres@localhost:5432/adcp_test PG_LANE: ${{ matrix.lane }} run: | - # Keep every case and its deadline. Separate process-crash controls - # so setup and teardown also fit inside each unchanged job budget. + # Keep every case and its per-case deadline. The core lane includes + # installed artifact coverage and has a larger overall job budget. case "$PG_LANE" in core) python scripts/reporting_test_harness.py pytest tests/conformance/signing/test_pg_replay_store.py \ @@ -827,7 +828,7 @@ jobs: runs-on: ubuntu-latest permissions: contents: read - timeout-minutes: 40 + timeout-minutes: 50 strategy: fail-fast: false matrix: @@ -857,7 +858,7 @@ jobs: run: pip install -e ".[dev,pg]" - name: Run installed source-equivalent production contract shell: bash - timeout-minutes: 35 + timeout-minutes: 45 env: ADCP_PG_TEST_URL: postgresql://postgres@localhost:5432/adcp_production_installed_test ADCP_PYTHON310: ${{ steps.production-python310.outputs.python-path }} diff --git a/docs/reporting-provisional-observations.md b/docs/reporting-provisional-observations.md new file mode 100644 index 000000000..8abfd5080 --- /dev/null +++ b/docs/reporting-provisional-observations.md @@ -0,0 +1,163 @@ +# Provisional observations and upgrade status + +The producer records every successful scheduled provisional read as a new +immutable revision, including a read whose rows are unchanged. A durable +reservation freezes the acquisition before source work starts. Its successful +observation, revision rows, ledger change, notifications and next checkpoint +commit together. A retry of the same reserved acquisition returns the existing +revision instead of appending another observation. + +This is a producer and persistence change. Complete service rollout qualification +is still pending, including activation of previously retired work, source-object +reuse, installed-package checks and independent integration review. Follow the +[production guide](reporting-production.md) and +[migration guide](reporting-ledger-migration.md) for the surrounding deployment gates. + +## Scheduling + +- An offering without `restatement_window` uses an SDK fallback of 72 elapsed + hours after the reporting period ends. An explicit offering window continues + to apply. +- The first reserved acquisition freezes the window and cadence. Cadence is the + explicit `restatement_cadence`, or the greater of the reporting-period duration + and `fastest_safe_cadence`. Later successful reads retain that policy. +- A successful read anchors the next due time at its checked/acquired instant + plus cadence, capped at the resolved provisional boundary. After downtime, + one due read runs; missed intervals do not create a burst of catch-up reads. +- The adapter may return typed `InlineFetchResult.provisional_until` evidence to + shorten or extend the boundary. It must be timezone-aware and at least the + source observation time. Invalid evidence commits no revision or checkpoint. + If evidence is absent, the frozen offering window or SDK fallback applies. +- Snapshot-only reporting retains one final inclusive read at the boundary, + including after downtime. Valid new source evidence may extend that boundary. + Expiry does not create an official revision. +- Existing explicitly configured official-close behavior remains in force. + `official_close_lag` still requires an explicitly declared `restatement_window` + and an official offering. The SDK fallback does not opt an adapter into + automatic official publication. + +The durable acquisition freezes the source request, predecessor revision and +observation ordinal. A retry renews only `deadline_at`, the execution budget for +that attempt. It retains the execution key, run ID, cutoff, scope and remaining +request fields even after the original deadline expires. Once reserved, a +successful retry completes that observation before a later acquisition starts. + +## Persistence and compatibility + +`PgReportingLedgerStore.create_schema()` installs two additive private tables: +`reporting_provisional_acquisitions` and `reporting_provisional_observations`. +They retain immutable reservation and observation metadata. The existing +restatement checkpoint remains the progress cursor. The in-memory store has the +same atomic operations. Custom producer stores must implement +`ProvisionalObservationStore` from `adcp.reporting.ledger.provisional` together +with `RestatementCheckpointStore`. + +The packaged `reporting/ledger/required_provisional_schema.json` describes all +27 objects added by this extension. Bootstrap and each reservation, observation +read, and atomic observation commit validate those objects, including immutable +triggers and their function definitions. Validation also runs with notifications +disabled. Missing, disabled or changed objects fail closed with +`PROVISIONAL_SCHEMA_UNREADY`; reinstall or repair the schema before resuming +production. Notification readiness alone does not establish observation readiness: +historical notification-only schemas remain valid for their original operations. + +Existing revision tables, exact-read row layout and pinned migration manifests +are unchanged. Each observation has its own revision ID and revision-specific +wire content hash. Unchanged source content retains its source content identity; +it does not mean that the revision ID or wire hash is reused. Historical readers +can still read rows directly from `reporting_revision_rows` for every revision. + +This change retains per-revision row payloads. Content-addressed reuse of staged +source objects is a separate pending change; complete payload deduplication has +not been accepted. Do not interpret immutable observation support as completion +of the full no-duplicate-payload requirement. + +Stop old producer writers before switching scheduling behavior. Old binaries do +not write the new observation metadata, even though they can still read the +existing revision rows. Historical-binary rolling tests and full service +qualification must pass before claiming an upgrade is supported. + +## Custom stores and publication decorators + +Declare the checkpoint and observation methods explicitly on the store class. +Dynamic `__getattr__` forwarding does not enable these optional capabilities. +The producer checks declarations consistently across supported Python versions; +an incomplete adapter raises `RESTATEMENT_CHECKPOINTS_NOT_SUPPORTED` or +`PROVISIONAL_OBSERVATIONS_NOT_SUPPORTED` before source acquisition. It does not +fall back to separate revision and checkpoint writes. + +A decorator that prepares revisions in `commit_revision` must also prepare them +in `commit_provisional_observation`. Explicitly compose all three observation +methods (`reserve_provisional_acquisition`, `get_provisional_observation`, +`commit_provisional_observation`) and both checkpoint methods +(`get_restatement_checkpoint`, `record_restatement_checkpoint`). Reuse the same +revision preparation, validation and fault boundaries around the delegated atomic +commit. Calling ordinary `commit_revision` first would break the observation's +transaction boundary; forwarding the new method without preparation would bypass +the decorator's publication rules. + +An `OBSERVATION_CONFLICT` fails only the affected slice for that worker turn. Its +pending work remains eligible for a later retry while sibling slices continue. +Schema or capability failures stop the turn so the operator can repair the store. + +## Pending scheduling qualification + +The production progress queue still selects bounded pending work before checking +its observation due time. During the new provisional window, future-due work can +therefore consume turn capacity that would otherwise acquire newly closed periods. +Durable due-time selection and its fairness tests remain a service activation +requirement; the default window is not a claim of completed production rollout. + +Successful observations retain their next cadence checkpoint. Failed or not-ready +attempts retain their acquisition identity, but do not yet persist an independent +retry time. Repeated worker turns can retry that acquisition before its source's +safe cadence. Restart-safe retry scheduling remains required before full rollout; +an in-process delay alone would not complete that requirement. + +## Previously retired work: activation requirement remains open + +A deployment that already processed a snapshot with no declared window may have +retired that obligation under the old one-shot policy. Installing this schema or +restarting a producer does **not** re-enroll it. Until the activation migration +below is implemented and verified, the new fallback applies to eligible pending +work; it is not a complete upgrade path for existing deployments. + +The affected progress APIs are `ReportingProducerProgress.next_producer_obligations`, +`finish_producer_acquisition`, `commit_producer_period` and +`producer_closed_through`. PostgreSQL retains retired rows as `state='settled'` +in `reporting_production_source_work`; `reporting_production_source_progress` +retains the generation's `closed_through` and acquisition turn. The in-memory +equivalents are `_production_source_work`, `_production_closed` and +`_production_source_turns`. Selection currently considers only pending work. + +Existing configuration enrollment installs generation/source/destination +bindings, without changing a retained work item's state. Repeated production +activation returns without re-enrollment. Creating a new configuration generation +would change the frozen reporting identity and is not a repair for an old +obligation. Rewinding `closed_through` is also unsupported. + +The activation owner must provide an explicit disposition for each affected +deployment. A candidate bounded migration starts with an admitted account and +configuration generation, selects settled snapshot obligations in a recorded +period-end range, excludes official publications, and resumes eligible work +without changing the obligation or its retained policy evidence. It must also +define the treatment of already-expired obligations and source overrides; a +simple three-day filter alone cannot establish eligibility for every legacy +record. Missing policy evidence needs an explicit upgrade decision. + +An indexed selection can use +`(account_id, delivery_config_id, delivery_config_version, period_end, +reporting_obligation_id) WHERE state='settled'`, a bounded period range and +keyset batches. The current pending-only index does not cover this selection. +Any index and re-enrollment operation belong in an additive migration, under the +existing account/generation lock, with idempotent progress and preserved +acquisition turns. Do not scan all history on each worker turn or change the +separate delivery queues to trigger acquisition. + +The required regression starts with an old writer completing a no-window +snapshot, leaving its work row settled. Upgrade while the intended provisional +boundary is still open, activate/configure using the same frozen generation, +advance one cadence, and require a second immutable snapshot with the original +revision still readable. Current activation does not satisfy that regression. +The completed migration must also cover official, expired, parked, foreign-account +and interrupted/resumed batches without changing unrelated work. diff --git a/pyproject.toml b/pyproject.toml index 28ea341fb..f869cdc9a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -194,6 +194,7 @@ adcp = [ "signing/pg/*.sql", "decisioning/pg/*.sql", "reporting/ledger/*.sql", + "reporting/ledger/*.json", "reporting/outbox/*.json", "reporting/materializer/assets/*.json", "reporting/materializer/*.json", diff --git a/src/adcp/reporting/ledger/pg.py b/src/adcp/reporting/ledger/pg.py index c23b0a11d..cf49a78cb 100644 --- a/src/adcp/reporting/ledger/pg.py +++ b/src/adcp/reporting/ledger/pg.py @@ -118,6 +118,7 @@ issue_evidence, validate_scope_refinement, ) +from adcp.reporting.ledger.provisional import ProvisionalAcquisition, ProvisionalObservation from adcp.reporting.ledger.store import ( LeasedConfiguration, LedgerConflictError, @@ -244,13 +245,20 @@ async def _create_schema_on(self, connection: Any) -> None: _RECONCILIATION_DDL_PATH, _NOTIFICATIONS_DDL_PATH, _ACTIVITY_DDL_PATH, + Path(__file__).with_name("reporting_provisional_observations.sql"), ): await connection.execute(path.read_text()) + await self._require_provisional_schema(connection) if self._notifications_enabled: from adcp.reporting.outbox._schema import validate_schema await validate_schema(connection) + async def _require_provisional_schema(self, connection: Any) -> None: + from adcp.reporting.outbox._schema import validate_provisional_schema + + await validate_provisional_schema(connection) + async def _notification_now(self, connection: Any) -> datetime: from adcp.reporting.outbox.pg import database_now @@ -882,6 +890,152 @@ async def list_revisions( ).fetchall() return tuple(_revision_from_row(row) for row in rows) + async def reserve_provisional_acquisition( + self, acquisition: ProvisionalAcquisition + ) -> ProvisionalAcquisition: + async with self.transaction(), self._connection() as connection: + await self._lock_account(connection, acquisition.account_id) + await self._require_provisional_schema(connection) + key = (acquisition.account_id, acquisition.obligation_id, acquisition.ordinal) + existing = await ( + await connection.execute( + "SELECT payload FROM reporting_provisional_acquisitions" + " WHERE account_id=%s AND reporting_obligation_id=%s AND ordinal=%s", + key, + ) + ).fetchone() + if existing is not None: + return ProvisionalAcquisition.from_wire(existing[0]) + obligation = await self.get_obligation( + account_id=acquisition.account_id, + reporting_obligation_id=acquisition.obligation_id, + ) + if obligation is None: + raise LedgerConflictError("OBLIGATION_NOT_FOUND", "unknown observation obligation") + if not acquisition.binds(obligation): + raise LedgerConflictError("OBSERVATION_CONFLICT", "acquisition generation differs") + duplicate = await ( + await connection.execute( + "SELECT 1 FROM reporting_provisional_acquisitions" + " WHERE account_id=%s AND source_execution_key=%s", + (acquisition.account_id, acquisition.execution_key), + ) + ).fetchone() + if duplicate is not None: + raise LedgerConflictError( + "OBSERVATION_CONFLICT", "execution key is already reserved" + ) + checkpoint = await self.get_restatement_checkpoint( + account_id=acquisition.account_id, + reporting_obligation_id=acquisition.obligation_id, + ) + expected = ( + checkpoint.next_observation + if checkpoint + else len( + await self.list_revisions( + account_id=acquisition.account_id, + reporting_obligation_id=acquisition.obligation_id, + ) + ) + ) + if acquisition.ordinal != expected: + raise LedgerConflictError("OBSERVATION_CONFLICT", "observation ordinal changed") + await connection.execute( + "INSERT INTO reporting_provisional_acquisitions" + " (account_id,reporting_obligation_id,ordinal,source_execution_key,payload)" + " VALUES (%s,%s,%s,%s,%s::jsonb)", + (*key, acquisition.execution_key, _json(acquisition.to_wire())), + ) + return ProvisionalAcquisition.from_wire(acquisition.to_wire()) + + async def get_provisional_observation( + self, *, account_id: str, reporting_obligation_id: str + ) -> ProvisionalObservation | None: + async with self._connection() as connection: + await self._require_provisional_schema(connection) + row = await ( + await connection.execute( + "SELECT payload FROM reporting_provisional_observations" + " WHERE account_id=%s AND reporting_obligation_id=%s" + " ORDER BY ordinal DESC LIMIT 1", + (account_id, reporting_obligation_id), + ) + ).fetchone() + return ProvisionalObservation.from_wire(row[0]) if row else None + + async def commit_provisional_observation( + self, + observation: ProvisionalObservation, + revision: ReportingRevisionRecord, + rows: Sequence[dict[str, Any]], + ) -> ReportingRevisionRecord: + acquisition = observation.acquisition + key = (acquisition.account_id, acquisition.obligation_id, acquisition.ordinal) + if ( + revision.account_id != acquisition.account_id + or revision.reporting_obligation_id != acquisition.obligation_id + or revision.reporting_revision_id != observation.revision_id + ): + raise LedgerConflictError("OBSERVATION_CONFLICT", "observation identity differs") + async with self.transaction(), self._connection() as connection: + await self._lock_account(connection, acquisition.account_id) + await self._require_provisional_schema(connection) + existing = await ( + await connection.execute( + "SELECT reporting_revision_id,payload FROM reporting_provisional_observations" + " WHERE account_id=%s AND reporting_obligation_id=%s AND ordinal=%s", + key, + ) + ).fetchone() + if existing is not None: + retained_observation = ProvisionalObservation.from_wire(existing[1]) + if ( + retained_observation.acquisition != acquisition + or existing[0] != observation.revision_id + ): + raise LedgerConflictError("OBSERVATION_CONFLICT", "observation replay differs") + retained_revision = await self.get_revision( + account_id=acquisition.account_id, reporting_revision_id=existing[0] + ) + if retained_revision is None: + raise LedgerConflictError( + "HISTORY_UNAVAILABLE", "observation revision is missing" + ) + return retained_revision + reserved = await ( + await connection.execute( + "SELECT payload FROM reporting_provisional_acquisitions" + " WHERE account_id=%s AND reporting_obligation_id=%s AND ordinal=%s", + key, + ) + ).fetchone() + if reserved is None or ProvisionalAcquisition.from_wire(reserved[0]) != acquisition: + raise LedgerConflictError("OBSERVATION_CONFLICT", "acquisition was not reserved") + checkpoint = await self.get_restatement_checkpoint( + account_id=acquisition.account_id, + reporting_obligation_id=acquisition.obligation_id, + ) + if checkpoint is not None and checkpoint.next_observation != acquisition.ordinal: + raise LedgerConflictError("OBSERVATION_CONFLICT", "observation ordinal changed") + committed = await self.commit_revision(revision, rows) + await self.record_restatement_checkpoint( + RestatementCheckpoint( + acquisition.account_id, + acquisition.obligation_id, + observation.checked_at, + acquisition.ordinal + 1, + observation.provisional_until, + ) + ) + await connection.execute( + "INSERT INTO reporting_provisional_observations" + " (account_id,reporting_obligation_id,ordinal,reporting_revision_id,payload)" + " VALUES (%s,%s,%s,%s,%s::jsonb)", + (*key, revision.reporting_revision_id, _json(observation.to_wire())), + ) + return committed + async def get_restatement_checkpoint( self, *, account_id: str, reporting_obligation_id: str ) -> RestatementCheckpoint | None: diff --git a/src/adcp/reporting/ledger/producer.py b/src/adcp/reporting/ledger/producer.py index fc54ea2b3..13f1d5119 100644 --- a/src/adcp/reporting/ledger/producer.py +++ b/src/adcp/reporting/ledger/producer.py @@ -52,11 +52,16 @@ ReportingRevisionRecord, iso_duration_to_timedelta, ) +from adcp.reporting.ledger.provisional import ( + ProvisionalAcquisition, + ProvisionalObservation, + ProvisionalObservationStore, + ProvisionalPolicy, +) from adcp.reporting.ledger.store import ( LeasedConfiguration, LedgerConflictError, ReportingLedgerStore, - RestatementCheckpoint, RestatementCheckpointStore, ) from adcp.reporting.revision_selection import select_reporting_revision @@ -560,12 +565,12 @@ async def _acquire_pending( turn=turn, now=now, ) - except ReportingCurrencyError as error: - # One obligation whose money cannot be interpreted -- a legacy - # period with no retained currency, or a source contradicting - # the frozen one -- is a stuck slice, not a broken worker. - # Raising here would starve every later period under this - # configuration on every turn, forever. + except (ReportingCurrencyError, LedgerConflictError) as error: + if isinstance(error, LedgerConflictError) and error.code != "OBSERVATION_CONFLICT": + raise + # Currency interpretation and competing observations are local + # slice failures. Let later periods make progress; a later turn + # can read the winning observation or retry the same reservation. logger.info( "reporting slice failed obligation=%s code=%s", obligation.reporting_obligation_id, @@ -604,13 +609,16 @@ async def _acquire_progress( if isinstance(error, LedgerConflictError) and error.code not in { "HISTORY_UNAVAILABLE", "EMPTY_DENOMINATOR", + "OBSERVATION_CONFLICT", }: raise turn.slices_failed.append(identifier) self._note_escalation(obligation, turn, now=now) # Retain the existing corrupt-history parking check. A transient - # currency failure during settling must not retire a readable leaf. - finished = policy is None or isinstance(error, LedgerConflictError) + # currency failure or observation conflict must not retire work. + finished = error.code != "OBSERVATION_CONFLICT" and ( + policy is None or isinstance(error, LedgerConflictError) + ) if finished: await progress.finish_producer_acquisition( configuration, reporting_obligation_id=identifier @@ -621,16 +629,13 @@ def _settling_policy( configuration: ReportingConfiguration, obligation: ReportingObligationRecord, ) -> _SettlingPolicy | None: - """Resolve the optional source-declared policy for a snapshot obligation.""" + """Resolve the source policy or SDK fallback for a snapshot obligation.""" offering_id = self._offerings.snapshot_offering_id if obligation.required_finality != "snapshot" or offering_id is None: return None offering = self._source.capabilities.offering(offering_id) if not isinstance(offering, ProvisionalSnapshotOfferingV1): return None - if offering.restatement_window is None: - return None - if offering.restatement_cadence is not None: cadence = timedelta( milliseconds=iso_duration_milliseconds_v1(offering.restatement_cadence) @@ -647,7 +652,7 @@ def _settling_policy( ) return _SettlingPolicy( restatement_window=timedelta( - milliseconds=iso_duration_milliseconds_v1(offering.restatement_window) + milliseconds=iso_duration_milliseconds_v1(offering.restatement_window or "P3D") ), restatement_cadence=cadence, official_close_lag=close_lag, @@ -668,6 +673,14 @@ async def _acquire_with_settling_policy( The progress store retains its rotating work item until the policy ends. """ checkpoint_store = self._restatement_store() + observation_store = self._observation_store() + latest = await observation_store.get_provisional_observation( + account_id=obligation.account_id, + reporting_obligation_id=obligation.reporting_obligation_id, + ) + if latest is not None: + frozen = latest.acquisition.policy + policy = _SettlingPolicy(frozen.window, frozen.cadence, frozen.official_close_lag) revisions = await self._store.list_revisions( account_id=obligation.account_id, reporting_obligation_id=obligation.reporting_obligation_id, @@ -695,13 +708,27 @@ async def _acquire_with_settling_policy( if checkpoint is not None and checkpoint.provisional_until is not None else declared_until ) - if _utc(now) < settles_at: - last_checked = ( - checkpoint.checked_at - if checkpoint is not None - else max(revisions, key=lambda item: _utc(item.created_at)).created_at - ) - if _utc(now) >= _utc(last_checked) + policy.restatement_cadence: + last_checked = ( + checkpoint.checked_at + if checkpoint is not None + else max(revisions, key=lambda item: _utc(item.created_at)).created_at + ) + explicit_close = ( + policy.official_close_lag is not None + and self._offerings.official_offering_id is not None + ) + next_due = ( + latest.next_due_at + if latest is not None + else min(_utc(last_checked) + policy.restatement_cadence, settles_at) + ) + # Preserve explicitly configured official-close boundary precedence. + # Snapshot-only policies retain their final inclusive due read through + # downtime; expiry alone is never an official publication. + if not explicit_close or _utc(now) < settles_at: + if next_due is None: + return not explicit_close + if _utc(now) >= _utc(next_due): await self.acquire_obligation( configuration, obligation, @@ -711,6 +738,12 @@ async def _acquire_with_settling_policy( target_finality="snapshot", track_settling=True, ) + published = await observation_store.get_provisional_observation( + account_id=obligation.account_id, + reporting_obligation_id=obligation.reporting_obligation_id, + ) + if not explicit_close and published is not None and published.next_due_at is None: + return True return False if policy.official_close_lag is None or self._offerings.official_offering_id is None: @@ -831,6 +864,35 @@ async def acquire_obligation( observation=observation, constituents=constituents, ) + acquisition = None + if track_settling: + policy = self._settling_policy(configuration, obligation) + assert policy is not None + latest = await self._observation_store().get_provisional_observation( + account_id=obligation.account_id, + reporting_obligation_id=obligation.reporting_obligation_id, + ) + frozen_policy = ( + latest.acquisition.policy + if latest is not None + else ProvisionalPolicy( + policy.restatement_window, policy.restatement_cadence, policy.official_close_lag + ) + ) + leaf = self._current_snapshot(revisions) + acquisition = await self._observation_store().reserve_provisional_acquisition( + ProvisionalAcquisition( + request.model_dump_json(), + observation, + frozen_policy, + leaf.reporting_revision_id if leaf is not None else None, + checkpoint.provisional_until if checkpoint is not None else None, + ) + ) + request = acquisition.request(deadline_at=_utc(now) + self._offerings.slice_timeout) + finality = ( + "snapshot" if request.publication_class == "PROVISIONAL_SNAPSHOT" else "official" + ) cancel = asyncio.Event() try: result = await asyncio.wait_for( @@ -859,24 +921,6 @@ async def acquire_obligation( manifest = self._verified_manifest(result) self._validate_manifest_currency(obligation, manifest) rows = await self._read_rows(request, manifest) - fingerprint = manifest.content_fingerprint.split(":", 1)[-1] - current_snapshot = self._current_snapshot(revisions) - if ( - track_settling - and finality == "snapshot" - and current_snapshot is not None - and current_snapshot.source_manifest_sha256 == fingerprint - ): - assert checkpoint_store is not None - await self._record_restatement_checkpoint( - checkpoint_store, - obligation, - manifest, - checked_at=now, - next_observation=observation + 1, - ) - return None - # ``now`` freezes dispatch/lease/cutoff decisions, not publication. # A conforming source can observe finality while acquisition is running. published_at = self._clock() @@ -885,51 +929,45 @@ async def acquire_obligation( "PUBLICATION_TIME_INVALID", "producer clock regressed during acquisition; correct the clock before retrying", ) - committed = await self.commit_revision_from_manifest( + return await self.commit_revision_from_manifest( obligation, manifest, rows=rows, finality=finality, now=published_at, turn=turn, + acquisition=acquisition, ) - if checkpoint_store is not None: - await self._record_restatement_checkpoint( - checkpoint_store, - obligation, - manifest, - checked_at=now, - next_observation=observation + 1, - ) - return committed def _restatement_store(self) -> RestatementCheckpointStore: - if not isinstance(self._store, RestatementCheckpointStore): + explicit = all( + inspect.getattr_static(self._store, name, None) is not None + for name in ("get_restatement_checkpoint", "record_restatement_checkpoint") + ) + if not explicit or not isinstance(self._store, RestatementCheckpointStore): raise LedgerConflictError( "RESTATEMENT_CHECKPOINTS_NOT_SUPPORTED", - "a source settling window requires a ledger store with durable restatement " - "checkpoints", + "a source settling window requires explicitly implemented durable " + "restatement checkpoint methods", ) return self._store - async def _record_restatement_checkpoint( - self, - store: RestatementCheckpointStore, - obligation: ReportingObligationRecord, - manifest: SourceBatchManifestV1, - *, - checked_at: datetime, - next_observation: int, - ) -> None: - await store.record_restatement_checkpoint( - RestatementCheckpoint( - account_id=obligation.account_id, - reporting_obligation_id=obligation.reporting_obligation_id, - checked_at=max(_utc(checked_at), _utc(manifest.acquired_at)), - next_observation=next_observation, - provisional_until=manifest.finality_evidence.provisional_until, + def _observation_store(self) -> ProvisionalObservationStore: + explicit = all( + inspect.getattr_static(self._store, name, None) is not None + for name in ( + "reserve_provisional_acquisition", + "get_provisional_observation", + "commit_provisional_observation", ) ) + if not explicit or not isinstance(self._store, ProvisionalObservationStore): + raise LedgerConflictError( + "PROVISIONAL_OBSERVATIONS_NOT_SUPPORTED", + "scheduled provisional reads require explicitly implemented atomic " + "observation methods, including any decorating publisher's preparation", + ) + return self._store def _note_escalation( self, obligation: ReportingObligationRecord, turn: WorkerTurn, *, now: datetime @@ -987,6 +1025,7 @@ async def commit_revision_from_manifest( finality: str, now: datetime | None = None, turn: WorkerTurn | None = None, + acquisition: ProvisionalAcquisition | None = None, ) -> ReportingRevisionRecord: """Project a verified manifest into an immutable ledger revision. @@ -1029,6 +1068,9 @@ async def commit_revision_from_manifest( else None ) + if acquisition is not None and finality == "snapshot": + supersedes = acquisition.predecessor_revision_id + control_totals = tuple((total.name, total.value) for total in manifest.control_totals) revision_id = f"rpr_{manifest.publication_id[4:44]}" prior = next((item for item in existing if item.reporting_revision_id == revision_id), None) @@ -1090,7 +1132,30 @@ async def commit_revision_from_manifest( from adcp.reporting.materializer.publication import verified_publication revision = verified_publication(self._revision_verifier, obligation, revision, rows) - committed = await self._store.commit_revision(revision, rows) + if acquisition is None: + committed = await self._store.commit_revision(revision, rows) + else: + checked_at = max(_utc(now), _utc(manifest.acquired_at)) + boundary = manifest.finality_evidence.provisional_until or ( + _utc(obligation.period.end) + acquisition.policy.window + ) + next_due = ( + min(checked_at + acquisition.policy.cadence, _utc(boundary)) + if finality == "snapshot" and checked_at < _utc(boundary) + else None + ) + committed = await self._observation_store().commit_provisional_observation( + ProvisionalObservation( + acquisition, + revision_id, + checked_at, + _utc(boundary), + next_due, + manifest.model_dump_json(), + ), + revision, + rows, + ) turn.revisions_committed.append(committed.reporting_revision_id) return committed diff --git a/src/adcp/reporting/ledger/provisional.py b/src/adcp/reporting/ledger/provisional.py new file mode 100644 index 000000000..9dc1fe14c --- /dev/null +++ b/src/adcp/reporting/ledger/provisional.py @@ -0,0 +1,197 @@ +"""Durable scheduling identities and immutable provisional observation metadata. + +These records are private persistence metadata, not new reporting wire fields. +A source execution's semantic request is frozen; its deadline is an attempt budget. +""" + +from __future__ import annotations + +import json +from collections.abc import Sequence +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable + +from adcp.reporting.source import ReportingSourceSliceRequestV1 + +if TYPE_CHECKING: + from adcp.reporting.ledger.models import ReportingObligationRecord, ReportingRevisionRecord + + +def utc(value: datetime) -> datetime: + if value.tzinfo is None or value.utcoffset() is None: + raise ValueError("provisional observation timestamps must be timezone-aware") + return value.astimezone(timezone.utc) + + +@dataclass(frozen=True) +class ProvisionalPolicy: + window: timedelta + cadence: timedelta + official_close_lag: timedelta | None = None + + def __post_init__(self) -> None: + if self.window < timedelta(0) or self.cadence <= timedelta(0): + raise ValueError("provisional window must be nonnegative and cadence positive") + if self.official_close_lag is not None and self.official_close_lag < timedelta(0): + raise ValueError("official close lag must be nonnegative") + + def to_wire(self) -> dict[str, int | None]: + return { + "window_us": self.window // timedelta(microseconds=1), + "cadence_us": self.cadence // timedelta(microseconds=1), + "official_close_lag_us": ( + self.official_close_lag // timedelta(microseconds=1) + if self.official_close_lag is not None + else None + ), + } + + @classmethod + def from_wire(cls, value: dict[str, Any]) -> ProvisionalPolicy: + lag = value["official_close_lag_us"] + return cls( + timedelta(microseconds=value["window_us"]), + timedelta(microseconds=value["cadence_us"]), + timedelta(microseconds=lag) if lag is not None else None, + ) + + +@dataclass(frozen=True) +class ProvisionalAcquisition: + request_json: str + ordinal: int + policy: ProvisionalPolicy + predecessor_revision_id: str | None = None + previous_boundary: datetime | None = None + + def __post_init__(self) -> None: + if type(self.ordinal) is not int or self.ordinal < 0: + raise ValueError("observation ordinal must be a nonnegative integer") + self.request() + if self.previous_boundary is not None: + utc(self.previous_boundary) + + def request(self, *, deadline_at: datetime | None = None) -> ReportingSourceSliceRequestV1: + # Parse on every access: callers cannot mutate the durable request's lists. + request = ReportingSourceSliceRequestV1.model_validate_json(self.request_json) + if deadline_at is not None: + request = request.model_copy(update={"deadline_at": utc(deadline_at)}) + return request + + @property + def account_id(self) -> str: + return self.request().identity.account_id + + @property + def obligation_id(self) -> str: + return self.request().identity.reporting_obligation_id + + @property + def execution_key(self) -> str: + return self.request().identity.source_execution_key + + def binds(self, obligation: ReportingObligationRecord) -> bool: + request = self.request() + identity = request.identity + return ( + identity.account_id == obligation.account_id + and identity.reporting_obligation_id == obligation.reporting_obligation_id + and identity.delivery_config_id == obligation.delivery_config_id + and identity.delivery_config_version == obligation.delivery_config_version + and identity.report_definition_id == obligation.report_definition_id + and request.period.start == obligation.period.start + and request.period.end == obligation.period.end + and request.currency == obligation.currency + ) + + def to_wire(self) -> dict[str, Any]: + return { + "request": json.loads(self.request_json), + "ordinal": self.ordinal, + "policy": self.policy.to_wire(), + "predecessor_revision_id": self.predecessor_revision_id, + "previous_boundary": ( + utc(self.previous_boundary).isoformat() + if self.previous_boundary is not None + else None + ), + } + + @classmethod + def from_wire(cls, value: dict[str, Any]) -> ProvisionalAcquisition: + return cls( + json.dumps(value["request"], sort_keys=True, separators=(",", ":")), + value["ordinal"], + ProvisionalPolicy.from_wire(value["policy"]), + value["predecessor_revision_id"], + ( + datetime.fromisoformat(value["previous_boundary"]) + if value["previous_boundary"] is not None + else None + ), + ) + + +@dataclass(frozen=True) +class ProvisionalObservation: + acquisition: ProvisionalAcquisition + revision_id: str + checked_at: datetime + provisional_until: datetime + next_due_at: datetime | None + manifest_json: str + + def __post_init__(self) -> None: + utc(self.checked_at) + utc(self.provisional_until) + if self.next_due_at is not None: + utc(self.next_due_at) + + def to_wire(self) -> dict[str, Any]: + return { + "acquisition": self.acquisition.to_wire(), + "revision_id": self.revision_id, + "checked_at": utc(self.checked_at).isoformat(), + "provisional_until": utc(self.provisional_until).isoformat(), + "next_due_at": ( + utc(self.next_due_at).isoformat() if self.next_due_at is not None else None + ), + "manifest": json.loads(self.manifest_json), + } + + @classmethod + def from_wire(cls, value: dict[str, Any]) -> ProvisionalObservation: + return cls( + ProvisionalAcquisition.from_wire(value["acquisition"]), + value["revision_id"], + datetime.fromisoformat(value["checked_at"]), + datetime.fromisoformat(value["provisional_until"]), + ( + datetime.fromisoformat(value["next_due_at"]) + if value["next_due_at"] is not None + else None + ), + json.dumps(value["manifest"], sort_keys=True, separators=(",", ":")), + ) + + +@runtime_checkable +class ProvisionalObservationStore(Protocol): + async def reserve_provisional_acquisition( + self, acquisition: ProvisionalAcquisition + ) -> ProvisionalAcquisition: + pass + + async def get_provisional_observation( + self, *, account_id: str, reporting_obligation_id: str + ) -> ProvisionalObservation | None: + pass + + async def commit_provisional_observation( + self, + observation: ProvisionalObservation, + revision: ReportingRevisionRecord, + rows: Sequence[dict[str, Any]], + ) -> ReportingRevisionRecord: + pass diff --git a/src/adcp/reporting/ledger/reporting_provisional_observations.sql b/src/adcp/reporting/ledger/reporting_provisional_observations.sql new file mode 100644 index 000000000..6622ea718 --- /dev/null +++ b/src/adcp/reporting/ledger/reporting_provisional_observations.sql @@ -0,0 +1,48 @@ +-- Additive private scheduling metadata. Original revision rows and manifests +-- remain readable by historical SDKs; no existing payload is rewritten. +CREATE TABLE IF NOT EXISTS reporting_provisional_acquisitions ( + account_id TEXT COLLATE "C" NOT NULL, + reporting_obligation_id TEXT COLLATE "C" NOT NULL, + ordinal BIGINT NOT NULL CHECK (ordinal >= 0), + source_execution_key TEXT COLLATE "C" NOT NULL, + payload JSONB NOT NULL, + PRIMARY KEY (account_id, reporting_obligation_id, ordinal), + CONSTRAINT provisional_acquisition_execution_key UNIQUE (account_id, source_execution_key), + CONSTRAINT provisional_acquisition_owner_fk + FOREIGN KEY (account_id, reporting_obligation_id) + REFERENCES reporting_obligations (account_id, reporting_obligation_id) +); + +CREATE TABLE IF NOT EXISTS reporting_provisional_observations ( + account_id TEXT COLLATE "C" NOT NULL, + reporting_obligation_id TEXT COLLATE "C" NOT NULL, + ordinal BIGINT NOT NULL, + reporting_revision_id TEXT COLLATE "C" NOT NULL UNIQUE, + payload JSONB NOT NULL, + PRIMARY KEY (account_id, reporting_obligation_id, ordinal), + CONSTRAINT provisional_observation_revision_fk + FOREIGN KEY (account_id, reporting_revision_id) + REFERENCES reporting_revisions (account_id, reporting_revision_id), + CONSTRAINT provisional_observation_acquisition_fk + FOREIGN KEY (account_id, reporting_obligation_id, ordinal) + REFERENCES reporting_provisional_acquisitions + (account_id, reporting_obligation_id, ordinal) +); + +CREATE OR REPLACE FUNCTION reporting_provisional_immutable() +RETURNS trigger LANGUAGE plpgsql AS $$ +BEGIN + RAISE EXCEPTION 'provisional observation metadata is immutable'; +END; +$$; + +DROP TRIGGER IF EXISTS reporting_provisional_acquisition_immutable + ON reporting_provisional_acquisitions; +CREATE TRIGGER reporting_provisional_acquisition_immutable + BEFORE UPDATE OR DELETE ON reporting_provisional_acquisitions + FOR EACH ROW EXECUTE FUNCTION reporting_provisional_immutable(); +DROP TRIGGER IF EXISTS reporting_provisional_observation_immutable + ON reporting_provisional_observations; +CREATE TRIGGER reporting_provisional_observation_immutable + BEFORE UPDATE OR DELETE ON reporting_provisional_observations + FOR EACH ROW EXECUTE FUNCTION reporting_provisional_immutable(); diff --git a/src/adcp/reporting/ledger/required_provisional_schema.json b/src/adcp/reporting/ledger/required_provisional_schema.json new file mode 100644 index 000000000..08e6eafe8 --- /dev/null +++ b/src/adcp/reporting/ledger/required_provisional_schema.json @@ -0,0 +1,110 @@ +{ + "column:reporting_provisional_acquisitions.account_id": { + "enabled": true, + "fingerprint": "fdc4b549138bf90a5af5ed00b87a86ff9a0cbea553e7f017b4d95510708f6f88" + }, + "column:reporting_provisional_acquisitions.ordinal": { + "enabled": true, + "fingerprint": "54a5f9962598caa8723435f8b55bd5e15823053af5dd63d531663997537dc6ba" + }, + "column:reporting_provisional_acquisitions.payload": { + "enabled": true, + "fingerprint": "ac355fc16c02b70cb0a24afee8214cdce5f5cbfdc7fd1630786d5101932ecfa4" + }, + "column:reporting_provisional_acquisitions.reporting_obligation_id": { + "enabled": true, + "fingerprint": "fdc4b549138bf90a5af5ed00b87a86ff9a0cbea553e7f017b4d95510708f6f88" + }, + "column:reporting_provisional_acquisitions.source_execution_key": { + "enabled": true, + "fingerprint": "fdc4b549138bf90a5af5ed00b87a86ff9a0cbea553e7f017b4d95510708f6f88" + }, + "column:reporting_provisional_observations.account_id": { + "enabled": true, + "fingerprint": "fdc4b549138bf90a5af5ed00b87a86ff9a0cbea553e7f017b4d95510708f6f88" + }, + "column:reporting_provisional_observations.ordinal": { + "enabled": true, + "fingerprint": "54a5f9962598caa8723435f8b55bd5e15823053af5dd63d531663997537dc6ba" + }, + "column:reporting_provisional_observations.payload": { + "enabled": true, + "fingerprint": "ac355fc16c02b70cb0a24afee8214cdce5f5cbfdc7fd1630786d5101932ecfa4" + }, + "column:reporting_provisional_observations.reporting_obligation_id": { + "enabled": true, + "fingerprint": "fdc4b549138bf90a5af5ed00b87a86ff9a0cbea553e7f017b4d95510708f6f88" + }, + "column:reporting_provisional_observations.reporting_revision_id": { + "enabled": true, + "fingerprint": "fdc4b549138bf90a5af5ed00b87a86ff9a0cbea553e7f017b4d95510708f6f88" + }, + "constraint:reporting_provisional_acquisitions.provisional_acquisition_execution_key": { + "enabled": true, + "fingerprint": "1276a011c989e1b6064812b2abab0c8ff1db5dc4e36f193d84b302d2f3152d04" + }, + "constraint:reporting_provisional_acquisitions.provisional_acquisition_owner_fk": { + "enabled": true, + "fingerprint": "a16c879766a899d572875db28fb19e13779063dfda05a1662c1919f38ed42ab5" + }, + "constraint:reporting_provisional_acquisitions.reporting_provisional_acquisitions_ordinal_check": { + "enabled": true, + "fingerprint": "c5100a7d23190ff8be6c1daab6ef9cfb45083283d78d77e319173e18ee7b8f39" + }, + "constraint:reporting_provisional_acquisitions.reporting_provisional_acquisitions_pkey": { + "enabled": true, + "fingerprint": "6e0009dfd0c8c76adc077b6f9e8bc866b476342946a107c8cba342ee3c6ec602" + }, + "constraint:reporting_provisional_observations.provisional_observation_acquisition_fk": { + "enabled": true, + "fingerprint": "bdfe1ed11c202a06b8f07b215f934e269628eda56a8f7922159ec056708a66b3" + }, + "constraint:reporting_provisional_observations.provisional_observation_revision_fk": { + "enabled": true, + "fingerprint": "6336638ff31659c0ea0a0fa6c67a7ed65e1b1bf792d17b5fbaec138fa6869c17" + }, + "constraint:reporting_provisional_observations.reporting_provisional_observations_pkey": { + "enabled": true, + "fingerprint": "6e0009dfd0c8c76adc077b6f9e8bc866b476342946a107c8cba342ee3c6ec602" + }, + "constraint:reporting_provisional_observations.reporting_provisional_observations_reporting_revision_id_key": { + "enabled": true, + "fingerprint": "dface19184e3b1196ac3f05865ce9fd2c1fde32593a604fd8e59e0e7b37525e7" + }, + "function:reporting_provisional_immutable()": { + "enabled": true, + "fingerprint": "5f46da14407d6b8acab20bf770940bcae96698b27a941eab2d1f4aa5c271484b" + }, + "index:reporting_provisional_acquisitions.provisional_acquisition_execution_key": { + "enabled": true, + "fingerprint": "ed532b97341cb02c8bdd5aa138e5a7b1748621c5d57907cc6ab3a1ea12bb6017" + }, + "index:reporting_provisional_acquisitions.reporting_provisional_acquisitions_pkey": { + "enabled": true, + "fingerprint": "fa257545f1b9106f2682d6baed4fdf51d062f7a29e106caca8b77734a613599f" + }, + "index:reporting_provisional_observations.reporting_provisional_observations_pkey": { + "enabled": true, + "fingerprint": "cc1a138288cf600afcffe140e7001d4e44d1a762ba2b6feeea6c5615c2e1c204" + }, + "index:reporting_provisional_observations.reporting_provisional_observations_reporting_revision_id_key": { + "enabled": true, + "fingerprint": "423b8e84d59c9f7c8a580f1b8f97058e964b191a9c4a2dfd0e1e95e00b1872c2" + }, + "table:reporting_provisional_acquisitions": { + "enabled": true, + "fingerprint": "1f824779ff80f110344420b019786663d8c9beaad230da90e0439795e734ccda" + }, + "table:reporting_provisional_observations": { + "enabled": true, + "fingerprint": "1f824779ff80f110344420b019786663d8c9beaad230da90e0439795e734ccda" + }, + "trigger:reporting_provisional_acquisitions.reporting_provisional_acquisition_immutable": { + "enabled": true, + "fingerprint": "0aedcd3a818d7899d2882f0543fbfc8446230a696a0f1d6dd757ea6894a4da41" + }, + "trigger:reporting_provisional_observations.reporting_provisional_observation_immutable": { + "enabled": true, + "fingerprint": "a93f79c3f18e6fbdee8c539ffcdd7e29f37018be6e9923104d2cecb95ad913e7" + } +} diff --git a/src/adcp/reporting/ledger/store.py b/src/adcp/reporting/ledger/store.py index f7773697a..8155a951f 100644 --- a/src/adcp/reporting/ledger/store.py +++ b/src/adcp/reporting/ledger/store.py @@ -74,6 +74,7 @@ issue_evidence, validate_scope_refinement, ) +from adcp.reporting.ledger.provisional import ProvisionalAcquisition, ProvisionalObservation if TYPE_CHECKING: from adcp.reporting.ledger.status_projection import ReportingStatusSnapshot @@ -696,6 +697,8 @@ def __init__( self._revision_identity: dict[str, str] = {} self._rows: dict[str, tuple[dict[str, Any], ...]] = {} self._restatement_checkpoints: dict[str, RestatementCheckpoint] = {} + self._provisional_acquisitions: dict[tuple[str, str, int], ProvisionalAcquisition] = {} + self._provisional_observations: dict[tuple[str, str, int], ProvisionalObservation] = {} self._adjustments: dict[str, ReportingAdjustmentRecord] = {} self._statuses: dict[tuple[str, str, str], ConsumerStatusRecord] = {} self._status_identity: dict[tuple[str, str, str], str] = {} @@ -1061,6 +1064,97 @@ async def list_revisions( and item.reporting_obligation_id == reporting_obligation_id ) + async def reserve_provisional_acquisition( + self, acquisition: ProvisionalAcquisition + ) -> ProvisionalAcquisition: + # Canonical round-trip also detaches all request collections. + acquisition = ProvisionalAcquisition.from_wire(acquisition.to_wire()) + key = (acquisition.account_id, acquisition.obligation_id, acquisition.ordinal) + async with self._mutation(): + existing = self._provisional_acquisitions.get(key) + if existing is not None: + return existing + obligation = self._obligations.get(acquisition.obligation_id) + if obligation is None or obligation.account_id != acquisition.account_id: + raise LedgerConflictError("OBLIGATION_NOT_FOUND", "unknown observation obligation") + if not acquisition.binds(obligation): + raise LedgerConflictError("OBSERVATION_CONFLICT", "acquisition generation differs") + if any( + item.account_id == acquisition.account_id + and item.execution_key == acquisition.execution_key + for item in self._provisional_acquisitions.values() + ): + raise LedgerConflictError( + "OBSERVATION_CONFLICT", "execution key is already reserved" + ) + checkpoint = self._restatement_checkpoints.get(acquisition.obligation_id) + expected = ( + checkpoint.next_observation + if checkpoint + else len( + await self.list_revisions( + account_id=acquisition.account_id, + reporting_obligation_id=acquisition.obligation_id, + ) + ) + ) + if acquisition.ordinal != expected: + raise LedgerConflictError("OBSERVATION_CONFLICT", "observation ordinal changed") + self._provisional_acquisitions[key] = acquisition + return acquisition + + async def get_provisional_observation( + self, *, account_id: str, reporting_obligation_id: str + ) -> ProvisionalObservation | None: + observations = [ + value + for (account, obligation, _), value in self._provisional_observations.items() + if account == account_id and obligation == reporting_obligation_id + ] + return max(observations, key=lambda item: item.acquisition.ordinal, default=None) + + async def commit_provisional_observation( + self, + observation: ProvisionalObservation, + revision: ReportingRevisionRecord, + rows: Sequence[dict[str, Any]], + ) -> ReportingRevisionRecord: + acquisition = observation.acquisition + key = (acquisition.account_id, acquisition.obligation_id, acquisition.ordinal) + if ( + revision.account_id != acquisition.account_id + or revision.reporting_obligation_id != acquisition.obligation_id + or revision.reporting_revision_id != observation.revision_id + ): + raise LedgerConflictError("OBSERVATION_CONFLICT", "observation identity differs") + async with self._mutation(): + existing = self._provisional_observations.get(key) + if existing is not None: + if ( + existing.acquisition != acquisition + or existing.revision_id != observation.revision_id + ): + raise LedgerConflictError("OBSERVATION_CONFLICT", "observation replay differs") + return self._revisions[existing.revision_id] + retained = self._provisional_acquisitions.get(key) + if retained != acquisition: + raise LedgerConflictError("OBSERVATION_CONFLICT", "acquisition was not reserved") + checkpoint = self._restatement_checkpoints.get(acquisition.obligation_id) + if checkpoint is not None and checkpoint.next_observation != acquisition.ordinal: + raise LedgerConflictError("OBSERVATION_CONFLICT", "observation ordinal changed") + committed = await self.commit_revision(revision, rows) + await self.record_restatement_checkpoint( + RestatementCheckpoint( + acquisition.account_id, + acquisition.obligation_id, + observation.checked_at, + acquisition.ordinal + 1, + observation.provisional_until, + ) + ) + self._provisional_observations[key] = observation + return committed + async def get_restatement_checkpoint( self, *, account_id: str, reporting_obligation_id: str ) -> RestatementCheckpoint | None: diff --git a/src/adcp/reporting/outbox/_schema.py b/src/adcp/reporting/outbox/_schema.py index 7fb1eb75c..07f05406d 100644 --- a/src/adcp/reporting/outbox/_schema.py +++ b/src/adcp/reporting/outbox/_schema.py @@ -18,6 +18,9 @@ REQUIRED_OBJECTS: dict[str, dict[str, Any]] = json.loads( files("adcp.reporting.outbox").joinpath("required_schema.json").read_text() ) +PROVISIONAL_REQUIRED_OBJECTS: dict[str, dict[str, Any]] = json.loads( + files("adcp.reporting.ledger").joinpath("required_provisional_schema.json").read_text() +) SCHEMA_CONTRACT: dict[str, str] = { key: value["fingerprint"] for key, value in REQUIRED_OBJECTS.items() } @@ -132,7 +135,13 @@ def _validate_schema_objects( """Apply the packaged B contract to an already captured catalog.""" if not REQUIRED_OBJECTS: raise ReportingNotificationError("notification_schema_unready:manifest_missing") - for key, expected in REQUIRED_OBJECTS.items(): + # A historical notification-only installation may have no provisional + # capability. Once any exact known extension object exists, its entire + # contract is required; unrelated adopter DDL remains unrelated. + required = REQUIRED_OBJECTS + if any(key in installed for key in PROVISIONAL_REQUIRED_OBJECTS): + required = {**required, **PROVISIONAL_REQUIRED_OBJECTS} + for key, expected in required.items(): if not activity and "reporting_webhook_" in key: continue actual = installed.get(key) @@ -145,3 +154,32 @@ def _validate_schema_objects( else: continue raise ReportingNotificationError(f"notification_schema_unready:{classification}:{key}") + + +async def validate_provisional_schema(connection: Any) -> None: + """Require the complete observation extension even when none of it exists.""" + from adcp.reporting.ledger.store import LedgerConflictError + + try: + installed = await schema_objects(connection) + except Exception: + raise LedgerConflictError( + "PROVISIONAL_SCHEMA_UNREADY", "provisional_schema_unready:catalog_unavailable" + ) from None + if not PROVISIONAL_REQUIRED_OBJECTS: + raise LedgerConflictError( + "PROVISIONAL_SCHEMA_UNREADY", "provisional_schema_unready:manifest_missing" + ) + for key, expected in PROVISIONAL_REQUIRED_OBJECTS.items(): + actual = installed.get(key) + if actual is None: + classification = "missing" + elif not actual["enabled"]: + classification = "disabled" + elif actual["fingerprint"] != expected["fingerprint"]: + classification = "changed" + else: + continue + raise LedgerConflictError( + "PROVISIONAL_SCHEMA_UNREADY", f"provisional_schema_unready:{classification}:{key}" + ) diff --git a/src/adcp/reporting/source.py b/src/adcp/reporting/source.py index 37f3071b9..eb0b89c70 100644 --- a/src/adcp/reporting/source.py +++ b/src/adcp/reporting/source.py @@ -664,12 +664,15 @@ class ProvisionalSnapshotOfferingV1(_OfferingBase): without upstream evidence that the account tolerates it is how a source gets itself rate-limited into ``action_required``. - ``restatement_window`` opts into automatic re-reads after period close. - ``restatement_cadence`` defaults to the configured reporting period (and - is never allowed to beat ``fastest_safe_cadence``). ``official_close_lag`` - asks a producer that also has an authoritative offering to publish a - terminal official revision after the source settles. Omitting the window - preserves the original one-shot behavior. + ``restatement_window`` overrides the SDK's default three-day provisional + re-read window after period close. ``restatement_cadence`` defaults to the + configured reporting period (and is never allowed to beat + ``fastest_safe_cadence``). Every successful scheduled read is a new immutable + observation, including unchanged content. A per-slice ``provisional_until`` + is authoritative over this offering default. ``official_close_lag`` still + requires an explicitly declared window and asks a producer with an + authoritative offering to publish an official revision after settlement; + the SDK fallback does not enable official close. """ publication_class: Literal["PROVISIONAL_SNAPSHOT"] = "PROVISIONAL_SNAPSHOT" diff --git a/tests/conformance/reporting/_production_packaging.py b/tests/conformance/reporting/_production_packaging.py index db9c0a2d5..4786f9889 100644 --- a/tests/conformance/reporting/_production_packaging.py +++ b/tests/conformance/reporting/_production_packaging.py @@ -338,7 +338,7 @@ def installed_production(root, python, wheel, source, *, label, driver_absent): label=label + "-installed-production", cwd=fixture_root, value=settings, - timeout=1800, + timeout=2400 if not driver_absent else 1800, progress=progress, ) ) diff --git a/tests/conformance/reporting/_provisional_catalog.py b/tests/conformance/reporting/_provisional_catalog.py new file mode 100644 index 000000000..7be0062d9 --- /dev/null +++ b/tests/conformance/reporting/_provisional_catalog.py @@ -0,0 +1,5 @@ +"""Exact additive observation catalog; historical manifests remain unchanged.""" + +from adcp.reporting.outbox._schema import PROVISIONAL_REQUIRED_OBJECTS as PROVISIONAL_OBJECTS + +assert len(PROVISIONAL_OBJECTS) == 27 diff --git a/tests/conformance/reporting/_reliable_support.py b/tests/conformance/reporting/_reliable_support.py index 248cb3e3e..5b5bb333a 100644 --- a/tests/conformance/reporting/_reliable_support.py +++ b/tests/conformance/reporting/_reliable_support.py @@ -85,6 +85,9 @@ if TYPE_CHECKING: from psycopg_pool import AsyncConnectionPool + from adcp.reporting.ledger.provisional import ProvisionalAcquisition, ProvisionalObservation + from adcp.reporting.ledger.store import RestatementCheckpoint + Store = InMemoryReportingReconciliationStore | PgReportingReconciliationStore Backend = Literal["memory", "postgres"] METRICS = ("impressions", "clicks", "spend") @@ -498,7 +501,7 @@ def __init__(self, harness: ReliableHarness) -> None: def __getattr__(self, name: str) -> Any: return getattr(self.harness.store, name) - async def commit_revision( + def _prepare_revision( self, revision: ReportingRevisionRecord, rows: Sequence[dict[str, Any]] ) -> ReportingRevisionRecord: fingerprint = f"sha256:{revision.source_manifest_sha256}" @@ -507,7 +510,7 @@ async def commit_revision( ReportingControlTotalRecord(total.name, total.value, total.value_type, total.unit) for total in manifest.control_totals ) - prepared = replace( + return replace( revision, managed_control_totals=totals, canonical_content_digest=ReportingCanonicalDigest( @@ -524,11 +527,56 @@ async def commit_revision( control_total_evidence=totals, ), ) + + async def commit_revision( + self, revision: ReportingRevisionRecord, rows: Sequence[dict[str, Any]] + ) -> ReportingRevisionRecord: + prepared = self._prepare_revision(revision, rows) await self.harness.failures.hit("revision.before") retained = await self.harness.store.commit_revision(prepared, rows) await self.harness.failures.hit("revision.after") return retained + # Optional persistence contracts are explicit across Python versions. A + # __getattr__ forwarder cannot compose this publisher's preparation/faults. + async def get_restatement_checkpoint( + self, *, account_id: str, reporting_obligation_id: str + ) -> RestatementCheckpoint | None: + return await self.harness.store.get_restatement_checkpoint( + account_id=account_id, reporting_obligation_id=reporting_obligation_id + ) + + async def record_restatement_checkpoint( + self, checkpoint: RestatementCheckpoint + ) -> RestatementCheckpoint: + return await self.harness.store.record_restatement_checkpoint(checkpoint) + + async def reserve_provisional_acquisition( + self, acquisition: ProvisionalAcquisition + ) -> ProvisionalAcquisition: + return await self.harness.store.reserve_provisional_acquisition(acquisition) + + async def get_provisional_observation( + self, *, account_id: str, reporting_obligation_id: str + ) -> ProvisionalObservation | None: + return await self.harness.store.get_provisional_observation( + account_id=account_id, reporting_obligation_id=reporting_obligation_id + ) + + async def commit_provisional_observation( + self, + observation: ProvisionalObservation, + revision: ReportingRevisionRecord, + rows: Sequence[dict[str, Any]], + ) -> ReportingRevisionRecord: + prepared = self._prepare_revision(revision, rows) + await self.harness.failures.hit("revision.before") + retained = await self.harness.store.commit_provisional_observation( + observation, prepared, rows + ) + await self.harness.failures.hit("revision.after") + return retained + @dataclass class ReliableHarness: diff --git a/tests/conformance/reporting/test_reporting_activity_migration.py b/tests/conformance/reporting/test_reporting_activity_migration.py index 52a2bafe4..c9598acbc 100644 --- a/tests/conformance/reporting/test_reporting_activity_migration.py +++ b/tests/conformance/reporting/test_reporting_activity_migration.py @@ -31,6 +31,7 @@ require_rolling_database, revision_for, ) +from ._provisional_catalog import PROVISIONAL_OBJECTS from ._reliable_support import ( Barrier, NotificationHarness, @@ -68,7 +69,7 @@ async def test_required_manifest_is_identical_across_random_schemas_and_repeated async with pool.connection() as conn: objects = await schema_objects(conn) assert len(WAIVER_OBJECTS) == 10 - assert objects == {**REQUIRED_OBJECTS, **WAIVER_OBJECTS} + assert objects == {**REQUIRED_OBJECTS, **WAIVER_OBJECTS, **PROVISIONAL_OBJECTS} # Regenerating unchanged SQL must have byte-for-byte zero # diff, including all function security/search_path flags. required = {key: objects[key] for key in REQUIRED_OBJECTS} diff --git a/tests/conformance/reporting/test_reporting_core_lifecycle.py b/tests/conformance/reporting/test_reporting_core_lifecycle.py index c1783f6b8..505e330a8 100644 --- a/tests/conformance/reporting/test_reporting_core_lifecycle.py +++ b/tests/conformance/reporting/test_reporting_core_lifecycle.py @@ -430,7 +430,14 @@ async def test_reporting_core_lifecycle(ledger: PgReportingLedgerStore) -> None: source.set(second.start, InlineFetchResult(rows=[], data_through=second.end)) turn = await _run_worker_at(ledger, source, now=second.end + timedelta(minutes=30)) assert len(turn.obligations_committed) == 1 - assert len(turn.revisions_committed) == 1 + # The source omits an explicit restatement window, so the SDK's default + # policy also schedules another read of the first period on this turn. + first_revisions = await ledger.list_revisions( + account_id=ACCOUNT, reporting_obligation_id=obligation.reporting_obligation_id + ) + assert len(first_revisions) == 2 + automatic = next(item for item in first_revisions if item.reporting_revision_id != revision_id) + assert automatic.supersedes_reporting_revision_id == revision_id zero_obligation = await ledger.find_obligation( account_id=ACCOUNT, @@ -444,6 +451,10 @@ async def test_reporting_core_lifecycle(ledger: PgReportingLedgerStore) -> None: account_id=ACCOUNT, reporting_obligation_id=zero_obligation.reporting_obligation_id ) assert [item.row_count for item in zero_revisions] == [0] + assert set(turn.revisions_committed) == { + automatic.reporting_revision_id, + zero_revisions[0].reporting_revision_id, + } settled = await _reconcile(ledger, expected=_expected_periods(2)) assert settled.definitive is True, settled.obligations @@ -495,8 +506,8 @@ async def test_reporting_core_lifecycle(ledger: PgReportingLedgerStore) -> None: # A buyer reconciling a scope containing that period is not definitive. assert (await _reconcile(ledger, expected=_expected_periods(3))).definitive is False - # 10. A restatement supersedes the first period's snapshot rather than - # editing it. Both revisions remain; only the new one is current. + # 10. A restatement supersedes the first period's current snapshot rather + # than editing it. All revisions remain; only the new one is current. source.set( first.start, InlineFetchResult( @@ -515,8 +526,8 @@ async def test_reporting_core_lifecycle(ledger: PgReportingLedgerStore) -> None: chain = await ledger.list_revisions( account_id=ACCOUNT, reporting_obligation_id=obligation.reporting_obligation_id ) - assert len(chain) == 2 - assert restated.supersedes_reporting_revision_id == revision_id + assert len(chain) == 3 + assert restated.supersedes_reporting_revision_id == automatic.reporting_revision_id # The superseded revision is still retained and still readable: a consumer # that already cited it must be able to fetch exactly what it cited. superseded = await ledger.read_revision_rows( diff --git a/tests/conformance/reporting/test_reporting_feed_migration.py b/tests/conformance/reporting/test_reporting_feed_migration.py index 886c997b1..caac9d39c 100644 --- a/tests/conformance/reporting/test_reporting_feed_migration.py +++ b/tests/conformance/reporting/test_reporting_feed_migration.py @@ -15,6 +15,7 @@ from ._durable_materializer_support import DurableHarness, durable_case from ._feed_support import feed_harness, feed_request, mixed_case, walk, without_feed from ._generation_support import isolated_reporting_pool +from ._provisional_catalog import PROVISIONAL_OBJECTS from ._receipt_support import receipt_case, request_for from ._reconciliation_support import Clock @@ -76,7 +77,8 @@ async def test_feed_migration_preserves_parent_catalog_receipts_pending_and_fair ) ) assert len(parent_manifest) == 763 - assert original == parent_manifest + assert {key: original[key] for key in parent_manifest} == parent_manifest + assert original == {**parent_manifest, **PROVISIONAL_OBJECTS} new = PgReportingFeedStore(pool=pool, notifications=notifications) with pytest.raises(ReportingFeedError) as error: await new.reporting_feed_ready() diff --git a/tests/conformance/reporting/test_reporting_feed_rolling.py b/tests/conformance/reporting/test_reporting_feed_rolling.py index b4bba4b57..390439112 100644 --- a/tests/conformance/reporting/test_reporting_feed_rolling.py +++ b/tests/conformance/reporting/test_reporting_feed_rolling.py @@ -6,6 +6,7 @@ import shutil import sys from dataclasses import replace +from importlib.resources import files from pathlib import Path import pytest @@ -26,6 +27,7 @@ from ._durable_materializer_support import DurableHarness, durable_case from ._feed_support import walk, without_feed from ._generation_support import isolated_reporting_pool +from ._provisional_catalog import PROVISIONAL_OBJECTS from ._receipt_support import adjustment_for from .test_reporting_feed_packaging import ROOT, run_step from .test_reporting_materializer_process import worker @@ -241,7 +243,12 @@ async def test_nine_actual_artifacts_preserve_ordinary_writes_and_frozen_b22_mou new_objects = await schema_objects(c) assert {k: new_objects[k] for k in old_objects} == old_objects added = new_objects.keys() - old_objects.keys() - assert len(added) == 33 and all("reporting_feed_" in k for k in added) + feed_objects = json.loads( + files("adcp.reporting.feed").joinpath("required_schema.json").read_text() + ) + assert len(feed_objects) == 33 + assert added == feed_objects.keys() | PROVISIONAL_OBJECTS.keys() + assert new_objects == {**old_objects, **feed_objects, **PROVISIONAL_OBJECTS} query = { "account": request["account"], "view": "periods", @@ -328,7 +335,8 @@ async def test_nine_actual_artifacts_preserve_ordinary_writes_and_frozen_b22_mou "b22_mounted_origins": replayed["origins"], "historical_wheel_sha256": installed_feed_history[3]["wheel_sha256"], "b22_wheel_sha256": approved_feed_b22[3]["wheel_sha256"], - "feed_objects": len(added), + "feed_objects": len(feed_objects), + "provisional_objects": len(PROVISIONAL_OBJECTS), "b24_additive_objects": len(production_added), "b24_activation": False, "page_count": len(expected[0]), diff --git a/tests/conformance/reporting/test_reporting_materializer_migration.py b/tests/conformance/reporting/test_reporting_materializer_migration.py index 10deb5a80..16872d0b8 100644 --- a/tests/conformance/reporting/test_reporting_materializer_migration.py +++ b/tests/conformance/reporting/test_reporting_materializer_migration.py @@ -14,6 +14,7 @@ from ._durable_materializer_support import durable_case, durable_harness from ._generation_support import isolated_reporting_pool, obligation_for +from ._provisional_catalog import PROVISIONAL_OBJECTS SQL = files("adcp.reporting.ledger").joinpath("reporting_materializer.sql").read_text() MANIFEST = json.loads( @@ -59,11 +60,13 @@ async def test_populated_repeated_and_concurrent_install_preserves_all_old_objec == { **REQUIRED_OBJECTS, **waiver_objects, + **PROVISIONAL_OBJECTS, } ) assert { key: value for key, value in actual.items() if "reporting_materializer_" in key } == MANIFEST + assert actual == {**original, **MANIFEST} assert ( await ( await c.execute( diff --git a/tests/conformance/reporting/test_reporting_materializer_rolling.py b/tests/conformance/reporting/test_reporting_materializer_rolling.py index 518d17698..6d5e55ab3 100644 --- a/tests/conformance/reporting/test_reporting_materializer_rolling.py +++ b/tests/conformance/reporting/test_reporting_materializer_rolling.py @@ -6,6 +6,7 @@ import shutil import sys import tarfile +from importlib.resources import files from pathlib import Path import pytest @@ -16,6 +17,7 @@ from ._durable_materializer_support import DurableHarness, durable_case from ._generation_support import isolated_reporting_pool, require_rolling_database +from ._provisional_catalog import PROVISIONAL_OBJECTS from .test_reporting_notification_packaging import ROOT, run_step ARTIFACTS = { @@ -193,7 +195,11 @@ async def test_installed_old_reader_writer_and_workers_on_populated_materializer async with pool.connection() as connection: objects = await schema_objects(connection) assert {key: objects[key] for key in old_objects} == old_objects - assert all("reporting_materializer_" in key for key in objects.keys() - old_objects.keys()) + materializer_objects = json.loads( + files("adcp.reporting.materializer").joinpath("required_schema.json").read_text() + ) + assert len(materializer_objects) == 187 + assert objects == {**old_objects, **materializer_objects, **PROVISIONAL_OBJECTS} case.store = sibling.store = store materializer_operation_1 = await case.service().run_once() assert (materializer_operation_1).state == "verified" diff --git a/tests/conformance/reporting/test_reporting_notification_migration.py b/tests/conformance/reporting/test_reporting_notification_migration.py index 19c7a48d8..b2734278d 100644 --- a/tests/conformance/reporting/test_reporting_notification_migration.py +++ b/tests/conformance/reporting/test_reporting_notification_migration.py @@ -15,6 +15,7 @@ from adcp.reporting.outbox.status_schema import REQUIRED_STATUS_OBJECTS from ._generation_support import NOW, isolated_reporting_pool, revision_for +from ._provisional_catalog import PROVISIONAL_OBJECTS from ._reconciliation_support import scenario from ._reliable_support import Barrier from .test_reporting_reconciliation_migration import FIXTURES @@ -164,6 +165,7 @@ async def test_direct_and_hopwise_historical_schema_chain(source, hopwise): assert await schema_contract(conn) == { **SCHEMA_CONTRACT, **{key: value["fingerprint"] for key, value in WAIVER_OBJECTS.items()}, + **{key: value["fingerprint"] for key, value in PROVISIONAL_OBJECTS.items()}, } assert await PgReportingOutbox(pool=pool).list_events(account_id="acct_a") == () diff --git a/tests/conformance/reporting/test_reporting_notification_packaging.py b/tests/conformance/reporting/test_reporting_notification_packaging.py index 593f2a937..65d46755b 100644 --- a/tests/conformance/reporting/test_reporting_notification_packaging.py +++ b/tests/conformance/reporting/test_reporting_notification_packaging.py @@ -25,6 +25,7 @@ obligation_for, revision_for, ) +from ._provisional_catalog import PROVISIONAL_OBJECTS from .test_reporting_notification_migration import CHAIN, WAIVER_OBJECTS ROOT = Path(__file__).resolve().parents[3] @@ -342,7 +343,8 @@ def installed_distribution(built_distribution, request): assert files("adcp.reporting.ledger").joinpath("reporting_status_selector_version.sql").is_file() assert files("adcp.reporting.outbox").joinpath("required_status_selector_schema.json").is_file() version = files("adcp").joinpath("ADCP_VERSION").read_text().strip() -assert get_named_validator("core/reporting-status-changed-webhook.json", version=version) is not None +assert get_named_validator( + "core/reporting-status-changed-webhook.json", version=version) is not None assert get_named_validator("core/webhook-activity-record.json", version=version) is not None assert ( get_named_validator("core/reporting-ledger-changed-webhook.json", version=version) @@ -410,6 +412,7 @@ async def test_installed_pg_extra_migrates_commits_and_restarts(installed_distri (ROOT / "src/adcp/reporting/outbox/required_schema.json").read_text() ), "waiver_objects": WAIVER_OBJECTS, + "provisional_objects": PROVISIONAL_OBJECTS, } for name, record in ( ("config", config), @@ -440,7 +443,12 @@ async def test_installed_pg_extra_migrates_commits_and_restarts(installed_distri values = json.load(sys.stdin) assert len(values["waiver_objects"]) == 10 -expected_objects = {**values["required_objects"], **values["waiver_objects"]} +assert len(values["provisional_objects"]) == 27 +assert json.loads(files("adcp.reporting.ledger").joinpath( + "required_provisional_schema.json").read_text()) == values["provisional_objects"] +expected_objects = { + **values["required_objects"], **values["waiver_objects"], **values["provisional_objects"] +} async def main(): @@ -491,7 +499,8 @@ async def get_active(self, *, account_id, subscriber_id, notification_type): clock=clock, ) projector = ReportingActivityProjector(outbox) - materializer_operation_1 = await ReportingActivitySupport(worker, store, projector).durable() + materializer_operation_1 = await ReportingActivitySupport( + worker, store, projector).durable() assert materializer_operation_1 events = await outbox.list_events(account_id="acct_a") expanded_1 = await worker.expand_one(account_id="acct_a") @@ -506,7 +515,8 @@ async def get_active(self, *, account_id, subscriber_id, notification_type): attempt, outcome=ActivityOutcome("failed", 503, 1), now=clock(), ) assert completed_attempt_1 - finished_delivery_1 = await outbox.finish_delivery(lease, now=clock(), state="pending", retry_at=clock()) + finished_delivery_1 = await outbox.finish_delivery( + lease, now=clock(), state="pending", retry_at=clock()) assert finished_delivery_1 async with AsyncConnectionPool( values["conninfo"], kwargs=values["kwargs"], open=False @@ -561,15 +571,19 @@ async def list_active(self, *, account_id, notification_type): async def get_active(self, *, account_id, subscriber_id, notification_type): return status_subscription if account_id == status_subscription.account_id else None status_worker = ReportingNotificationWorker(outbox=status.outbox, - subscriptions=StatusConfigurations(), cipher=cipher, clock=clock, activity=status.outbox) + subscriptions=StatusConfigurations(), cipher=cipher, clock=clock, + activity=status.outbox) status_operation_2 = await status_worker.expand_one(account_id="acct_a") assert status_operation_2 - status_lease = await status.outbox.claim_delivery(account_id="acct_a", now=clock(), lease_seconds=60) + status_lease = await status.outbox.claim_delivery( + account_id="acct_a", now=clock(), lease_seconds=60) status_body = cipher.open(status_lease.delivery).prepared assert json.loads(status_body.body)["notification_type"] == "reporting.status_changed" - status_operation_3 = await status.outbox.finish_delivery(status_lease, now=clock(), state="pending", retry_at=clock()) + status_operation_3 = await status.outbox.finish_delivery( + status_lease, now=clock(), state="pending", retry_at=clock()) assert status_operation_3 - async with AsyncConnectionPool(values["conninfo"], kwargs=values["kwargs"], open=False) as c_restart: + async with AsyncConnectionPool( + values["conninfo"], kwargs=values["kwargs"], open=False) as c_restart: ledger = PgReportingReconciliationStore(pool=c_restart, clock=clock, notifications=True) status = PgStatusNotificationStore(ledger) await status.create_schema() @@ -582,7 +596,8 @@ async def get_active(self, *, account_id, subscriber_id, notification_type): retry = cipher.open(status_lease.delivery).prepared assert retry.body == status_body.body assert retry.idempotency_key == status_body.idempotency_key - status_operation_5 = await status.outbox.finish_delivery(status_lease, now=clock(), state="complete") + status_operation_5 = await status.outbox.finish_delivery( + status_lease, now=clock(), state="complete") assert status_operation_5 status_operation_6 = await status.outbox.reemit(account_id="acct_a", consumer_namespace="", notification_id=status_events[0].notification_id, now=clock()) diff --git a/tests/conformance/reporting/test_reporting_prepared_observation_adapter.py b/tests/conformance/reporting/test_reporting_prepared_observation_adapter.py new file mode 100644 index 000000000..25aa31299 --- /dev/null +++ b/tests/conformance/reporting/test_reporting_prepared_observation_adapter.py @@ -0,0 +1,75 @@ +"""A decorating publisher composes the entire atomic observation contract.""" + +import asyncio + +import pytest + +from ._reliable_support import ScriptedSource, complete_fetch, configuration, reliable_factory +from .test_reporting_evidence_currency_integration import frozen_slice + + +@pytest.mark.parametrize("backend", ["memory", "postgres"]) +@pytest.mark.parametrize("failure_point", ["revision.before", "revision.after"]) +async def test_prepared_observation_faults_preserve_complete_atomic_state(backend, failure_point): + async with reliable_factory(backend) as h: + await h.store.put_configuration(configuration("eur")) + script = ScriptedSource(eur=[complete_fetch]) + h.failures.at(failure_point, OSError("publisher process loss")) + with pytest.raises(OSError, match="publisher process loss"): + await h.producer(h.source(script.async_fetch)).run_worker() + (request,) = script.requests + identity = { + "account_id": "eur", + "reporting_obligation_id": request.identity.reporting_obligation_id, + } + revisions = await h.store.list_revisions(**identity) + observation = await h.store.get_provisional_observation(**identity) + checkpoint = await h.store.get_restatement_checkpoint(**identity) + if failure_point == "revision.before": + assert revisions == () + assert observation is None and checkpoint is None + else: + assert len(revisions) == 1 + assert observation is not None and checkpoint is not None + assert observation.revision_id == revisions[0].reporting_revision_id + assert observation.acquisition.execution_key == request.identity.source_execution_key + assert checkpoint.next_observation == observation.acquisition.ordinal + 1 + assert checkpoint.checked_at == observation.checked_at + assert checkpoint.provisional_until == observation.provisional_until + assert revisions[0].managed_control_totals is not None + assert revisions[0].managed_control_totals[-1].unit == "EUR" + assert revisions[0].canonical_content_digest is not None + + await h.restart() + replacement = ScriptedSource(eur=[]) + await h.producer(h.source(replacement.async_fetch)).run_worker() + assert replacement.requests == [] + retained = await h.store.list_revisions(**identity) + current = await h.store.get_provisional_observation(**identity) + position = await h.store.get_restatement_checkpoint(**identity) + assert len(retained) == 1 and current is not None and position is not None + assert current.revision_id == retained[0].reporting_revision_id + assert current.acquisition.execution_key == request.identity.source_execution_key + assert position.next_observation == current.acquisition.ordinal + 1 + assert retained[0].managed_control_totals is not None + assert retained[0].canonical_content_digest is not None + if failure_point == "revision.after": + assert (retained, current, position) == (revisions, observation, checkpoint) + + +@pytest.mark.parametrize("backend", ["memory", "postgres"]) +async def test_prepared_ordinary_revision_keeps_the_existing_commit_seam(backend): + async with reliable_factory(backend) as h: + producer, obligation, request = await frozen_slice(h, partial=False) + result = await h.source(complete_fetch).execute(request, cancel=asyncio.Event()) + revision = await h.commit_slice(producer, obligation, request, result) + assert revision.managed_control_totals is not None + assert revision.canonical_content_digest is not None + assert h.failures.hits.count("revision.before") == 1 + assert h.failures.hits.count("revision.after") == 1 + identity = { + "account_id": obligation.account_id, + "reporting_obligation_id": obligation.reporting_obligation_id, + } + assert await h.store.get_provisional_observation(**identity) is None + assert await h.store.get_restatement_checkpoint(**identity) is None diff --git a/tests/conformance/reporting/test_reporting_production_late_accounts.py b/tests/conformance/reporting/test_reporting_production_late_accounts.py index bc9d1df1a..6a99c6240 100644 --- a/tests/conformance/reporting/test_reporting_production_late_accounts.py +++ b/tests/conformance/reporting/test_reporting_production_late_accounts.py @@ -333,6 +333,27 @@ async def ongoing_first_turns(pool, account): raise AssertionError({"continuous_first_work_not_established": observed}) +async def next_due_first_turn(pool, account, previous): + """Wait for another committed first-account turn while its work remains due.""" + deadline = time.monotonic() + 30 + last = None + while time.monotonic() < deadline: + async with pool.connection() as connection: + last = await ( + await connection.execute( + "SELECT served_at::text,due_at<=clock_timestamp()," + " (SELECT count(*) FROM reporting_materializer_candidates c" + " WHERE c.account_id=a.account_id AND c.due_at<=clock_timestamp())" + " FROM reporting_materializer_accounts a WHERE account_id=%s", + (account,), + ) + ).fetchone() + if last[0] != previous and last[1] and last[2] >= 2: + return last + await asyncio.sleep(0.02) + raise AssertionError({"first_account_did_not_advance_after_late_admission": last}) + + @pytest.mark.parametrize("first", ["usd", "eur"], ids=["late-sorts-first", "late-sorts-last"]) @pytest.mark.parametrize("notifications", [False, True]) async def test_late_account_progresses_via_typed_public_support_and_survives_restart( @@ -390,8 +411,7 @@ async def test_late_account_progresses_via_typed_public_support_and_survives_res period=results[0][account]["period"] if index else None, ) if index == 0 and account != first: - correction_condition_1 = (await served(pool))[first] != first_before - assert correction_condition_1 + await next_due_first_turn(pool, first, first_before) results.append(current) async with pool.connection() as connection: correction_condition_2 = await ( diff --git a/tests/conformance/reporting/test_reporting_production_progress.py b/tests/conformance/reporting/test_reporting_production_progress.py index 741076493..d050c62a2 100644 --- a/tests/conformance/reporting/test_reporting_production_progress.py +++ b/tests/conformance/reporting/test_reporting_production_progress.py @@ -424,6 +424,51 @@ def turn_document(turn, source): } +async def default_observation_states(h, snapshot, executions): + """Every bounded acquisition retains its identity, hourly due time and checkpoint.""" + states, retained_keys = {}, set() + revisions = {revision.reporting_revision_id: revision for revision in snapshot.revisions} + for obligation in snapshot.obligations: + identity = { + "account_id": obligation.account_id, + "reporting_obligation_id": obligation.reporting_obligation_id, + } + observation = await h.store.get_provisional_observation(**identity) + assert observation is not None + acquisition = observation.acquisition + seeded = obligation.reporting_obligation_id == h.item.obligation.reporting_obligation_id + assert acquisition.ordinal == (1 if seeded else 0) + assert acquisition.predecessor_revision_id == ( + h.item.revision.reporting_revision_id if seeded else None + ) + assert acquisition.binds(obligation) + assert acquisition.policy.window == timedelta(days=3) + assert acquisition.policy.cadence == timedelta(hours=1) + assert acquisition.policy.official_close_lag is None + assert observation.provisional_until == obligation.period.end + timedelta(days=3) + expected_due = ( + min(observation.checked_at + timedelta(hours=1), observation.provisional_until) + if observation.checked_at < observation.provisional_until + else None + ) + assert observation.next_due_at == expected_due + checkpoint = await h.store.get_restatement_checkpoint(**identity) + assert checkpoint is not None + assert checkpoint.checked_at == observation.checked_at + assert checkpoint.next_observation == acquisition.ordinal + 1 + assert checkpoint.provisional_until == observation.provisional_until + revision = revisions[observation.revision_id] + assert revision.finality == "snapshot" + assert revision.supersedes_reporting_revision_id == acquisition.predecessor_revision_id + assert acquisition.execution_key not in retained_keys + retained_keys.add(acquisition.execution_key) + states[obligation.reporting_obligation_id] = "pending" if expected_due else "settled" + assert retained_keys == set(executions) + assert list(states.values()).count("settled") == 60 + assert list(states.values()).count("pending") == 71 + return states + + @asynccontextmanager async def restarted_process(h, path, *, pause): from .test_reporting_materializer_process import Child @@ -497,11 +542,24 @@ async def test_bounded_producer_advances_past_processed_first_window(backend, tm ) first = await source_turn(support) assert not first.slices_failed - # The first period was already published before activation. The - # remaining first window is acquired once, using the real sealed source. + # The first period was published before activation. Its default-policy + # read retains that predecessor and adds an observation; all 64 periods + # still fit in the original bounded acquisition window. assert len(first.obligations_committed) == 63 - assert len(first.revisions_committed) == 63 - assert len(source.requests) == 63 + assert len(first.revisions_committed) == 64 + assert len(source.requests) == 64 + seeded_observation = await h.store.get_provisional_observation( + account_id=item.config.account_id, + reporting_obligation_id=item.obligation.reporting_obligation_id, + ) + assert seeded_observation is not None + assert seeded_observation.acquisition.ordinal == 1 + assert seeded_observation.acquisition.predecessor_revision_id == ( + item.revision.reporting_revision_id + ) + assert seeded_observation.checked_at == h.source_clock() + assert seeded_observation.next_due_at is None + assert seeded_observation.revision_id in first.revisions_committed first_document = turn_document(first, source) print( json.dumps({"progress_backend": backend, "phase": "first_window", "count": 64}), @@ -553,9 +611,18 @@ async def test_bounded_producer_advances_past_processed_first_window(backend, tm assert len(second["obligations"]) == len(second["revisions"]) == 64 assert len(third["obligations"]) == len(third["revisions"]) == 3 executions = first_document["executions"] + second["executions"] + third["executions"] - assert len(executions) == len(set(executions)) == 130 + assert len(executions) == len(set(executions)) == 131 snapshot = await h.store.read_status_snapshot(account_id=item.config.account_id) - assert len(snapshot.obligations) == len(snapshot.revisions) == 131 + assert len(snapshot.obligations) == 131 + assert len(snapshot.revisions) == 132 + assert ( + await h.store.get_provisional_observation( + account_id=item.config.account_id, + reporting_obligation_id=item.obligation.reporting_obligation_id, + ) + == seeded_observation + ) + expected_states = await default_observation_states(h, snapshot, executions) assert max(o.period.end for o in snapshot.obligations) == START + timedelta(hours=131) assert {o.generation_key for o in snapshot.obligations} == {item.config.generation_key} assert not (await h.store.read_status_snapshot(account_id="acct_b")).obligations @@ -567,12 +634,15 @@ async def test_bounded_producer_advances_past_processed_first_window(backend, tm == item.revision ) with sqlite3.connect(tmp_path / "source.seals") as connection: - assert connection.execute("SELECT count(*) FROM seals").fetchone() == (130,) + assert connection.execute("SELECT count(*) FROM seals").fetchone() == (131,) if h.pool is None: assert h.store._production_closed == { item.config.generation_key: START + timedelta(hours=131) } - assert {w.state for w in h.store._production_source_work.values()} == {"settled"} + assert { + identifier: work.state + for identifier, work in h.store._production_source_work.items() + } == expected_states assert not {c.generation_key for c in untouched} & set(h.store._production_closed) else: async with h.pool.connection() as c: @@ -590,18 +660,24 @@ async def test_bounded_producer_advances_past_processed_first_window(backend, tm START + timedelta(hours=131), ) ] - assert await ( - await c.execute( - "SELECT state,count(*) FROM reporting_production_source_work GROUP BY state" + assert ( + dict( + await ( + await c.execute( + "SELECT reporting_obligation_id,state" + " FROM reporting_production_source_work" + ) + ).fetchall() ) - ).fetchall() == [("settled", 131)] + == expected_states + ) print( json.dumps( { "progress_backend": backend, "periods": 131, "bound": 64, - "unique_acquisitions": 130, + "unique_acquisitions": 131, "restart": "SIGKILL" if h.pool else "new-service", } ), diff --git a/tests/conformance/reporting/test_reporting_production_settling.py b/tests/conformance/reporting/test_reporting_production_settling.py index 0daa23858..dcfe3f689 100644 --- a/tests/conformance/reporting/test_reporting_production_settling.py +++ b/tests/conformance/reporting/test_reporting_production_settling.py @@ -96,7 +96,7 @@ async def test_progress_retains_policy_until_terminal_publication( assert len(source.requests) == 1 h.source_clock.now = END + timedelta(hours=1) unchanged = await source_turn(h.production) - assert not unchanged.revisions_committed + assert len(unchanged.revisions_committed) == 1 assert len(source.requests) == 2 checkpoint = await h.store.get_restatement_checkpoint( account_id=h.item.config.account_id, @@ -111,12 +111,13 @@ async def test_progress_retains_policy_until_terminal_publication( changed = await source_turn(h.production) assert len(changed.revisions_committed) == 1 history = await revisions(h) - assert len(history) == 2 + assert len(history) == 3 assert history[1].supersedes_reporting_revision_id == history[0].reporting_revision_id + assert history[2].supersedes_reporting_revision_id == history[1].reporting_revision_id h.source_clock.now = END + timedelta(hours=3) await source_turn(h.production) - assert len(source.requests) == 3 + assert len(source.requests) == (3 if close_officially else 4) if not close_officially: production_operation_5 = await pending(h) assert production_operation_5 == () @@ -131,7 +132,12 @@ async def test_progress_retains_policy_until_terminal_publication( source.official_ready = True completed = await source_turn(h.production) assert len(completed.revisions_committed) == 1 - assert [r.finality for r in await revisions(h)] == ["snapshot", "snapshot", "official"] + assert [r.finality for r in await revisions(h)] == [ + "snapshot", + "snapshot", + "snapshot", + "official", + ] production_operation_4 = await pending(h) assert production_operation_4 == () count = len(source.requests) @@ -155,7 +161,7 @@ async def test_settling_checkpoint_and_pending_work_survive_fresh_service(backen assert len(initial.revisions_committed) == 1 h.source_clock.now = END + timedelta(hours=1) noop = await source_turn(h.production) - assert not noop.revisions_committed + assert len(noop.revisions_committed) == 1 original_source = h.production.offerings[0].producer._source assert len(original_source.requests) == 2 prior_keys = {r.identity.source_execution_key for r in original_source.requests} @@ -191,6 +197,6 @@ async def test_settling_checkpoint_and_pending_work_survive_fresh_service(backen reporting_obligation_id=fresh.item.obligation.reporting_obligation_id, ) assert checkpoint is not None and checkpoint.next_observation == 3 - assert len(await revisions(fresh)) == 2 + assert len(await revisions(fresh)) == 3 unfinished = await pending(fresh) assert unfinished == (fresh.item.obligation.reporting_obligation_id,) diff --git a/tests/conformance/reporting/test_reporting_provisional_catalog.py b/tests/conformance/reporting/test_reporting_provisional_catalog.py new file mode 100644 index 000000000..7fea4ed24 --- /dev/null +++ b/tests/conformance/reporting/test_reporting_provisional_catalog.py @@ -0,0 +1,171 @@ +"""The observation extension preserves the exact historical catalog and rows.""" + +import asyncio +import json +from importlib.resources import files + +import pytest + +from adcp.reporting.ledger import PgReportingLedgerStore +from adcp.reporting.outbox._schema import REQUIRED_OBJECTS, schema_objects +from adcp.reporting.outbox.status_schema import REQUIRED_STATUS_OBJECTS +from adcp.reporting.receipts import PgReportingReceiptStore, ReportingReceiptError + +from ._durable_materializer_support import DurableHarness, durable_case +from ._generation_support import ( + configuration, + isolated_reporting_pool, + obligation_for, + revision_for, +) +from ._provisional_catalog import PROVISIONAL_OBJECTS +from ._reconciliation_support import Clock +from .test_reporting_notification_migration import retained_physical_rows + + +@pytest.mark.parametrize("autocommit", [False, True]) +async def test_atomic_observation_bootstrap_preserves_exact_763_parent_objects( + autocommit, monkeypatch +): + # This is a source-level DDL control, not installed historical-binary proof. + # All eight pre-observation migration files and old manifests stay unchanged. + root = files("adcp.reporting.ledger") + chain = ( + "reporting_ledger.sql", + "reporting_ledger_account_generations.sql", + "reporting_ledger_obligation_currency.sql", + "reporting_ledger_reconciliation.sql", + "reporting_notification_outbox.sql", + "reporting_webhook_activity.sql", + "reporting_materializer.sql", + "reporting_receipt_ingestion.sql", + ) + expected_parent = { + **REQUIRED_OBJECTS, + **{ + key: value + for key, value in REQUIRED_STATUS_OBJECTS.items() + if "reporting_issue_waiver_bindings" in key + }, + } + for package in ("materializer", "receipts"): + expected_parent.update( + json.loads( + files("adcp.reporting." + package).joinpath("required_schema.json").read_text() + ) + ) + assert len(expected_parent) == 763 + async with isolated_reporting_pool(autocommit=autocommit) as pool: + async with pool.connection() as connection, connection.transaction(): + for name in chain: + await connection.execute(root.joinpath(name).read_text()) + store = PgReportingReceiptStore(pool=pool, clock=Clock()) + await durable_case(store) + h = DurableHarness(store, store._clock, pool) + rows = await h.image() + physical = await retained_physical_rows(pool) + async with pool.connection() as connection: + assert await schema_objects(connection) == expected_parent + original = store._create_schema_on + + async def fail_after_observation_ddl(connection): + await original(connection) + raise RuntimeError("injected observation bootstrap failure") + + with monkeypatch.context() as patch: + patch.setattr(store, "_create_schema_on", fail_after_observation_ddl) + with pytest.raises(ReportingReceiptError) as failure: + await store.create_schema() + assert failure.value.code == "RECEIPT_STORAGE_UNAVAILABLE" + async with pool.connection() as connection: + assert await schema_objects(connection) == expected_parent + assert await h.image() == rows + assert await retained_physical_rows(pool) == physical + await asyncio.gather(*(store.create_schema() for _ in range(3))) + async with pool.connection() as connection: + actual = await schema_objects(connection) + assert {key: actual[key] for key in expected_parent} == expected_parent + assert actual == {**expected_parent, **PROVISIONAL_OBJECTS} + assert await retained_physical_rows(pool) == physical + image = await h.image() + assert image == { + **rows, + "reporting_provisional_acquisitions": [], + "reporting_provisional_observations": [], + } + + +async def test_both_observation_tables_reject_update_and_delete(): + from tests.test_reporting_provisional_observations import latest + from tests.test_reporting_settling import _capabilities, _harness + + psycopg = pytest.importorskip("psycopg") + async with isolated_reporting_pool() as pool: + producer, store, _, _ = await _harness( + _capabilities(restatement_window=None), + store_factory=lambda clock: PgReportingLedgerStore( + pool=pool, clock=clock, notifications=True + ), + ) + await producer.run_worker() + before = await latest(store) + for table in ("reporting_provisional_acquisitions", "reporting_provisional_observations"): + for operation in (f"UPDATE {table} SET payload=payload", f"DELETE FROM {table}"): + async with pool.connection() as connection: + with pytest.raises(psycopg.errors.RaiseException, match="immutable"): + async with connection.transaction(): + await connection.execute(operation) + assert await latest(store) == before + + +async def test_observation_foreign_keys_bind_both_obligation_and_revision_accounts(): + psycopg = pytest.importorskip("psycopg") + async with isolated_reporting_pool() as pool: + store = PgReportingLedgerStore(pool=pool) + await store.create_schema() + records = [] + for account in ("owner-a", "owner-b"): + config = configuration(account) + obligation = obligation_for(config) + revision, rows = revision_for(obligation) + await store.put_configuration(config) + await store.commit_obligation(obligation) + await store.commit_revision(revision, rows) + records.append((obligation, revision)) + first, second = records + insert_acquisition = ( + "INSERT INTO reporting_provisional_acquisitions " + "(account_id, reporting_obligation_id, ordinal, source_execution_key, payload) " + "VALUES (%s, %s, 0, 'probe-execution', '{}')" + ) + async with pool.connection() as connection: + with pytest.raises(psycopg.errors.ForeignKeyViolation) as failure: + async with connection.transaction(): + await connection.execute( + insert_acquisition, (second[0].account_id, first[0].reporting_obligation_id) + ) + assert failure.value.diag.constraint_name == "provisional_acquisition_owner_fk" + await connection.execute( + insert_acquisition, (first[0].account_id, first[0].reporting_obligation_id) + ) + with pytest.raises(psycopg.errors.ForeignKeyViolation) as failure: + async with connection.transaction(): + await connection.execute( + "INSERT INTO reporting_provisional_observations " + "(account_id, reporting_obligation_id, ordinal, " + "reporting_revision_id, payload) " + "VALUES (%s, %s, 0, %s, '{}')", + ( + first[0].account_id, + first[0].reporting_obligation_id, + second[1].reporting_revision_id, + ), + ) + assert failure.value.diag.constraint_name == "provisional_observation_revision_fk" + assert ( + await ( + await connection.execute( + "SELECT count(*) FROM reporting_provisional_observations" + ) + ).fetchone() + )[0] == 0 diff --git a/tests/conformance/reporting/test_reporting_provisional_readiness.py b/tests/conformance/reporting/test_reporting_provisional_readiness.py new file mode 100644 index 000000000..5543027e7 --- /dev/null +++ b/tests/conformance/reporting/test_reporting_provisional_readiness.py @@ -0,0 +1,105 @@ +"""Provisional persistence cannot operate with a partial or unguarded catalog.""" + +import pytest + +from adcp.reporting.ledger import LedgerConflictError, PgReportingLedgerStore + +from ._generation_support import isolated_reporting_pool + +DAMAGE = { + "missing_trigger": ( + "DROP TRIGGER reporting_provisional_observation_immutable" + " ON reporting_provisional_observations" + ), + "disabled_trigger": ( + "ALTER TABLE reporting_provisional_observations" + " DISABLE TRIGGER reporting_provisional_observation_immutable" + ), + "changed_guard": ( + "CREATE OR REPLACE FUNCTION reporting_provisional_immutable()" + " RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RETURN OLD; END; $$" + ), + "partial_extension": "DROP TABLE reporting_provisional_acquisitions CASCADE", + "absent_extension": ( + "DROP TABLE reporting_provisional_observations,reporting_provisional_acquisitions;" + " DROP FUNCTION reporting_provisional_immutable()" + ), +} + + +@pytest.mark.parametrize("damage", DAMAGE) +@pytest.mark.parametrize("operation", ["read", "reserve", "commit"]) +@pytest.mark.parametrize("notifications", [False, True]) +async def test_provisional_operations_require_the_complete_immutable_extension( + damage, operation, notifications +): + from tests.test_reporting_provisional_observations import latest + from tests.test_reporting_settling import _capabilities, _harness + + async with isolated_reporting_pool(autocommit=True) as pool: + producer, store, _, _ = await _harness( + _capabilities(restatement_window=None), + store_factory=lambda clock: PgReportingLedgerStore( + pool=pool, clock=clock, notifications=notifications + ), + ) + await producer.run_worker() + observation = await latest(store) + identity = { + "account_id": observation.acquisition.account_id, + "reporting_obligation_id": observation.acquisition.obligation_id, + } + (revision,) = await store.list_revisions(**identity) + rows = await store.read_revision_rows( + account_id=revision.account_id, reporting_revision_id=revision.reporting_revision_id + ) + async with pool.connection() as connection: + await connection.execute(DAMAGE[damage]) + with pytest.raises(LedgerConflictError) as failure: + if operation == "read": + await store.get_provisional_observation(**identity) + elif operation == "reserve": + await store.reserve_provisional_acquisition(observation.acquisition) + else: + await store.commit_provisional_observation(observation, revision, rows.rows) + assert failure.value.code == "PROVISIONAL_SCHEMA_UNREADY" + assert await store.list_revisions(**identity) == (revision,) + assert ( + await store.read_revision_rows( + account_id=revision.account_id, + reporting_revision_id=revision.reporting_revision_id, + ) + == rows + ) + + +@pytest.mark.parametrize("damage", DAMAGE) +async def test_notification_readiness_checks_a_present_known_extension(damage): + from adcp.reporting.outbox import ReportingNotificationError + from adcp.reporting.outbox._schema import validate_schema + + async with isolated_reporting_pool(autocommit=True) as pool: + await PgReportingLedgerStore(pool=pool).create_schema() + async with pool.connection() as connection: + await connection.execute(DAMAGE[damage]) + if damage == "absent_extension": + # Ordinary legacy notification readiness does not declare + # provisional capability. Its original manifest remains valid. + await validate_schema(connection) + else: + with pytest.raises(ReportingNotificationError, match="notification_schema_unready"): + await validate_schema(connection) + + +@pytest.mark.parametrize("notifications", [False, True]) +async def test_bootstrap_rejects_an_existing_malformed_observation_column(notifications): + async with isolated_reporting_pool(autocommit=True) as pool: + store = PgReportingLedgerStore(pool=pool, notifications=notifications) + await store.create_schema() + async with pool.connection() as connection: + await connection.execute( + "ALTER TABLE reporting_provisional_acquisitions ALTER COLUMN payload DROP NOT NULL" + ) + with pytest.raises(LedgerConflictError) as failure: + await store.create_schema() + assert failure.value.code == "PROVISIONAL_SCHEMA_UNREADY" diff --git a/tests/conformance/reporting/test_reporting_provisional_worker_failures.py b/tests/conformance/reporting/test_reporting_provisional_worker_failures.py new file mode 100644 index 000000000..12750ce0b --- /dev/null +++ b/tests/conformance/reporting/test_reporting_provisional_worker_failures.py @@ -0,0 +1,107 @@ +"""Capability failure is explicit; one observation conflict cannot starve siblings.""" + +from datetime import timedelta + +import pytest + +from adcp.reporting.ledger import LedgerConflictError + +from ._production_support import production_harness +from .test_reporting_production_lock_order import source_turn + + +@pytest.mark.parametrize("backend", ["memory", "postgres"]) +async def test_one_observation_conflict_preserves_sibling_progress_and_retry(backend, tmp_path): + async with production_harness( + backend, tmp_path / "destination.sqlite", count=0, periods=3 + ) as h: + store, support = h.store, h.production + source = support.offerings[0].producer._source + h.source_clock.advance(timedelta(hours=3)) + await support.activate(account_id=h.item.config.account_id) + identifier = h.item.obligation.reporting_obligation_id + original = store.reserve_provisional_acquisition + + async def conflict(acquisition): + if acquisition.obligation_id == identifier: + raise LedgerConflictError("OBSERVATION_CONFLICT", "concurrent observation won") + return await original(acquisition) + + store.reserve_provisional_acquisition = conflict + first = await source_turn(support) + assert first.slices_failed == [identifier] + assert len(first.revisions_committed) == 2 + assert len(source.requests) == 2 + identity = {"account_id": h.item.config.account_id, "reporting_obligation_id": identifier} + assert await store.get_provisional_observation(**identity) is None + assert await store.get_restatement_checkpoint(**identity) is None + store.reserve_provisional_acquisition = original + second = await source_turn(support) + assert not second.slices_failed + assert len(second.revisions_committed) == 1 + assert len(source.requests) == 3 + assert len({request.identity.source_execution_key for request in source.requests}) == 3 + observation = await store.get_provisional_observation(**identity) + assert observation is not None and observation.acquisition.ordinal == 1 + assert ( + observation.acquisition.predecessor_revision_id == h.item.revision.reporting_revision_id + ) + + +async def test_basic_ledger_observation_conflict_does_not_abort_later_periods(): + from tests.test_reporting_settling import _capabilities, _harness + + producer, store, fetch, clock = await _harness(_capabilities(restatement_window=None)) + producer._max_periods_per_turn = 2 + clock[0] += timedelta(hours=1) + original = store.reserve_provisional_acquisition + conflicted = [] + + async def conflict(acquisition): + if not conflicted: + conflicted.append(acquisition.obligation_id) + raise LedgerConflictError("OBSERVATION_CONFLICT", "concurrent observation won") + return await original(acquisition) + + store.reserve_provisional_acquisition = conflict + first = await producer.run_worker() + assert first.slices_failed == conflicted + assert len(first.revisions_committed) == len(fetch.calls) == 1 + store.reserve_provisional_acquisition = original + second = await producer.run_worker() + assert not second.slices_failed + assert len(second.revisions_committed) == 1 + assert len(fetch.calls) == 2 + + +@pytest.mark.parametrize("explicit_checkpoints", [False, True]) +async def test_dynamic_protocol_forwarding_requires_explicit_publisher_composition( + explicit_checkpoints, +): + from tests.test_reporting_settling import _capabilities, _harness + + producer, store, fetch, _ = await _harness(_capabilities(restatement_window=None)) + + class Proxy: + def __getattr__(self, name): + return getattr(store, name) + + class CheckpointProxy(Proxy): + async def get_restatement_checkpoint(self, **identity): + return await store.get_restatement_checkpoint(**identity) + + async def record_restatement_checkpoint(self, checkpoint): + return await store.record_restatement_checkpoint(checkpoint) + + producer._store = CheckpointProxy() if explicit_checkpoints else Proxy() + with pytest.raises(LedgerConflictError) as failure: + await producer.run_worker() + assert failure.value.code == ( + "PROVISIONAL_OBSERVATIONS_NOT_SUPPORTED" + if explicit_checkpoints + else "RESTATEMENT_CHECKPOINTS_NOT_SUPPORTED" + ) + assert fetch.calls == [] + assert not store._provisional_acquisitions + assert not store._provisional_observations + assert not store._revisions diff --git a/tests/conformance/reporting/test_reporting_receipt_rolling.py b/tests/conformance/reporting/test_reporting_receipt_rolling.py index 0d8576b2a..5915e2dfa 100644 --- a/tests/conformance/reporting/test_reporting_receipt_rolling.py +++ b/tests/conformance/reporting/test_reporting_receipt_rolling.py @@ -5,6 +5,7 @@ import shutil import sys from datetime import datetime, timedelta, timezone +from importlib.resources import files from pathlib import Path import pytest @@ -21,6 +22,7 @@ from ._durable_materializer_support import DurableHarness, durable_case from ._generation_support import END, isolated_reporting_pool +from ._provisional_catalog import PROVISIONAL_OBJECTS from .test_reporting_materializer_process import worker from .test_reporting_materializer_rolling import ARTIFACTS, build_frozen, frozen_call @@ -137,10 +139,11 @@ async def test_actual_old_readers_and_writers_before_and_after_receipt_migration async with pool.connection() as c: new_objects = await schema_objects(c) assert {key: new_objects[key] for key in old_objects} == old_objects - assert all( - "reporting_receipt_ingestion_" in key - for key in new_objects.keys() - old_objects.keys() + receipt_objects = json.loads( + files("adcp.reporting.receipts").joinpath("required_schema.json").read_text() ) + assert len(receipt_objects) == 102 + assert new_objects == {**old_objects, **receipt_objects, **PROVISIONAL_OBJECTS} assert await immutable_parent_rows(pool) == saved adjustment = ReportingAdjustmentRecord( "frozen-adjustment", diff --git a/tests/conformance/reporting/test_reporting_status_migration.py b/tests/conformance/reporting/test_reporting_status_migration.py index d4d96e89f..0070f4be5 100644 --- a/tests/conformance/reporting/test_reporting_status_migration.py +++ b/tests/conformance/reporting/test_reporting_status_migration.py @@ -39,6 +39,7 @@ require_rolling_database, revision_for, ) +from ._provisional_catalog import PROVISIONAL_OBJECTS from ._reliable_support import ( Barrier, FailurePlan, @@ -212,6 +213,7 @@ async def test_populated_repeated_c_manifest_preserves_every_a_b_object_and_row( assert {k: v for k, v in objects.items() if k not in REQUIRED_OBJECTS} == { **REQUIRED_STATUS_OBJECTS, **REQUIRED_STATUS_SELECTOR_OBJECTS, + **PROVISIONAL_OBJECTS, } assert ( json.dumps(REQUIRED_STATUS_OBJECTS, sort_keys=True, indent=2) + "\n" diff --git a/tests/test_reporting_provisional_observations.py b/tests/test_reporting_provisional_observations.py new file mode 100644 index 000000000..eede7ae67 --- /dev/null +++ b/tests/test_reporting_provisional_observations.py @@ -0,0 +1,535 @@ +"""Successful scheduled reads are immutable observations in memory and PostgreSQL.""" + +from __future__ import annotations + +import asyncio +from dataclasses import replace +from datetime import timedelta, timezone + +import pytest +from pydantic import ValidationError + +from adcp.reporting.ledger import ( + InMemoryReportingLedgerStore, + PgReportingLedgerStore, + ReportingProducer, +) +from adcp.reporting.ledger.store import LedgerConflictError +from tests.conformance.reporting._generation_support import isolated_reporting_pool +from tests.test_reporting_settling import ( + ACCOUNT, + _capabilities, + _harness, + _only_obligation, + _revisions, +) + + +@pytest.fixture(params=["memory", "postgres"]) +async def make_harness(request): + if request.param == "memory": + + async def build(capabilities): + return await _harness( + capabilities, + store_factory=lambda clock: InMemoryReportingLedgerStore( + clock=clock, notifications=True + ), + ) + + yield build + else: + async with isolated_reporting_pool() as pool: + + async def build(capabilities): + return await _harness( + capabilities, + store_factory=lambda clock: PgReportingLedgerStore( + pool=pool, clock=clock, notifications=True + ), + ) + + yield build + + +async def latest(store): + obligation = await _only_obligation(store) + observation = await store.get_provisional_observation( + account_id=ACCOUNT, reporting_obligation_id=obligation.reporting_obligation_id + ) + assert observation is not None + return observation + + +async def test_adapter_without_window_rereads_with_sdk_default(make_harness): + producer, store, fetch, clock = await make_harness(_capabilities(restatement_window=None)) + first = await producer.run_worker() + assert len(first.revisions_committed) == 1 + initial = await latest(store) + obligation = await _only_obligation(store) + assert initial.provisional_until == obligation.period.end + timedelta(days=3) + + clock[0] += timedelta(hours=1) + refreshed = await producer.run_worker() + + assert fetch.calls == ["PROVISIONAL_SNAPSHOT", "PROVISIONAL_SNAPSHOT"] + assert len(refreshed.revisions_committed) == 1 + assert len(await _revisions(store)) == 2 + assert all(item.finality == "snapshot" for item in await _revisions(store)) + + +async def test_unchanged_scheduled_read_is_a_distinct_immutable_observation(make_harness): + producer, store, fetch, clock = await make_harness( + _capabilities(restatement_window="P3D", restatement_cadence="PT1H") + ) + await producer.run_worker() + original = (await _revisions(store))[0] + first = await latest(store) + original_rows = await store.read_revision_rows( + account_id=ACCOUNT, reporting_revision_id=original.reporting_revision_id + ) + clock[0] += timedelta(hours=1) + refreshed = await producer.run_worker() + + assert fetch.calls == ["PROVISIONAL_SNAPSHOT", "PROVISIONAL_SNAPSHOT"] + assert len(refreshed.revisions_committed) == 1 + history = await _revisions(store) + assert len(history) == 2 and history[0] == original + assert history[1].reporting_revision_id != original.reporting_revision_id + assert history[1].supersedes_reporting_revision_id == original.reporting_revision_id + assert history[1].observed_at > original.observed_at + assert history[1].source_manifest_sha256 == original.source_manifest_sha256 + assert history[1].revision_content_sha256 != original.revision_content_sha256 + second = await latest(store) + assert second.acquisition.ordinal == first.acquisition.ordinal + 1 + assert second.checked_at > first.checked_at + assert second.next_due_at == clock[0] + timedelta(hours=1) + assert second.revision_id == history[1].reporting_revision_id + current_rows = await store.read_revision_rows( + account_id=ACCOUNT, reporting_revision_id=second.revision_id + ) + assert current_rows.rows == original_rows.rows + # The legacy SQL layout retains every new revision's rows as well. + if isinstance(store, PgReportingLedgerStore): + async with store._pool.connection() as connection: + counts = await ( + await connection.execute( + "SELECT reporting_revision_id,count(*) FROM reporting_revision_rows" + " GROUP BY reporting_revision_id ORDER BY reporting_revision_id" + ) + ).fetchall() + assert sorted(count for _, count in counts) == [1, 1] + await producer.run_worker() + assert len(fetch.calls) == 2 + + +@pytest.mark.parametrize("offset", [timedelta(minutes=30), timedelta(days=4)]) +async def test_source_boundary_shortens_or_extends_fallback(make_harness, offset): + producer, store, fetch, clock = await make_harness(_capabilities(restatement_window=None)) + # Deliberately use a non-UTC representation of the same instant. + boundary = (clock[0] + offset).astimezone(timezone(timedelta(hours=-5))) + fetch.provisional_until = boundary + await producer.run_worker() + first = await latest(store) + assert first.provisional_until == boundary + assert first.next_due_at == min(clock[0] + timedelta(hours=1), boundary) + clock[0] = first.next_due_at + await producer.run_worker() + assert len(fetch.calls) == 2 + assert len(await _revisions(store)) == 2 + + +async def test_final_read_at_boundary_and_overdue_restart_never_promote(make_harness): + producer, store, fetch, clock = await make_harness(_capabilities(restatement_window=None)) + await producer.run_worker() + first = await latest(store) + clock[0] = first.provisional_until + timedelta(days=1) + await producer.run_worker() + final = await latest(store) + assert final.next_due_at is None + assert len(fetch.calls) == 2 + assert len(await _revisions(store)) == 2 + assert all(item.finality == "snapshot" for item in await _revisions(store)) + await producer.run_worker() + assert len(fetch.calls) == 2 + + +async def test_exact_boundary_gets_one_final_snapshot(make_harness): + producer, store, fetch, clock = await make_harness(_capabilities(restatement_window=None)) + fetch.provisional_until = clock[0] + timedelta(minutes=30) + await producer.run_worker() + clock[0] = fetch.provisional_until + await producer.run_worker() + assert len(fetch.calls) == 2 + assert (await latest(store)).next_due_at is None + await producer.run_worker() + assert len(fetch.calls) == 2 + + +async def test_new_source_override_can_extend_final_observation(make_harness): + producer, store, fetch, clock = await make_harness(_capabilities(restatement_window=None)) + fetch.provisional_until = clock[0] + timedelta(minutes=30) + await producer.run_worker() + clock[0] = fetch.provisional_until + fetch.provisional_until = clock[0] + timedelta(days=1) + await producer.run_worker() + second = await latest(store) + assert second.next_due_at == clock[0] + timedelta(hours=1) + clock[0] = second.next_due_at + await producer.run_worker() + assert len(fetch.calls) == 3 + + +async def test_changed_content_and_restart_keep_frozen_cadence(make_harness): + producer, store, fetch, clock = await make_harness( + _capabilities(restatement_window="P3D", restatement_cadence="PT1H") + ) + await producer.run_worker() + first = await latest(store) + # A fresh store instance, when PostgreSQL-backed, must restore all policy. + restored = ( + PgReportingLedgerStore(pool=store._pool, clock=lambda: clock[0], notifications=True) + if isinstance(store, PgReportingLedgerStore) + else store + ) + source = producer._source + source._capabilities = _capabilities(restatement_window="P1D", restatement_cadence="PT2H") + fresh = ReportingProducer( + source=source, + offerings=producer._offerings, + store=restored, + object_reader=producer._object_reader, + max_periods_per_turn=1, + clock=lambda: clock[0], + ) + clock[0] = first.next_due_at + fetch.impressions = 27 + await fresh.run_worker() + second = await latest(restored) + assert second.acquisition.policy == first.acquisition.policy + assert second.next_due_at == clock[0] + timedelta(hours=1) + assert len(await _revisions(restored)) == 2 + assert (await _revisions(restored))[0].source_manifest_sha256 != (await _revisions(restored))[ + 1 + ].source_manifest_sha256 + + +async def test_checkpoint_failure_rolls_back_observation_and_retry_keeps_identity( + make_harness, monkeypatch +): + producer, store, fetch, clock = await make_harness(_capabilities(restatement_window=None)) + original = store.record_restatement_checkpoint + + async def fail_after_checkpoint(checkpoint): + await original(checkpoint) + raise RuntimeError("checkpoint fault") + + monkeypatch.setattr(store, "record_restatement_checkpoint", fail_after_checkpoint) + with pytest.raises(RuntimeError, match="checkpoint fault"): + await producer.run_worker() + assert await _revisions(store) == () + obligation = await _only_obligation(store) + assert ( + await store.get_restatement_checkpoint( + account_id=ACCOUNT, reporting_obligation_id=obligation.reporting_obligation_id + ) + is None + ) + assert ( + await store.get_provisional_observation( + account_id=ACCOUNT, reporting_obligation_id=obligation.reporting_obligation_id + ) + is None + ) + monkeypatch.setattr(store, "record_restatement_checkpoint", original) + clock[0] += timedelta(hours=2) + await producer.run_worker() + # Replay after the original deadline consumes the retained source seal. + assert fetch.calls == ["PROVISIONAL_SNAPSHOT"] + assert len(await _revisions(store)) == 1 + assert (await latest(store)).acquisition.ordinal == 0 + + +async def test_concurrent_same_acquisition_commits_once(make_harness): + producer, store, fetch, clock = await make_harness(_capabilities(restatement_window=None)) + await producer.run_worker() + first = await latest(store) + clock[0] = first.next_due_at + config = (await store.list_configurations(account_id=ACCOUNT))[0] + obligation = await _only_obligation(store) + await asyncio.gather( + *[ + producer.acquire_obligation( + config, obligation, restate=True, now=clock[0], track_settling=True + ) + for _ in range(2) + ] + ) + assert len(await _revisions(store)) == 2 + assert (await latest(store)).acquisition.ordinal == 1 + checkpoint = await store.get_restatement_checkpoint( + account_id=ACCOUNT, reporting_obligation_id=obligation.reporting_obligation_id + ) + assert checkpoint.next_observation == 2 + + +async def test_account_cannot_read_or_publish_another_observation(make_harness): + producer, store, fetch, clock = await make_harness(_capabilities(restatement_window=None)) + await producer.run_worker() + observation = await latest(store) + assert ( + await store.get_provisional_observation( + account_id="other", reporting_obligation_id=observation.acquisition.obligation_id + ) + is None + ) + revision = (await _revisions(store))[0] + # Reserved identity cannot be rebound to a different account at commit. + with pytest.raises(LedgerConflictError): + forged = replace(observation, revision_id="forged") + await store.commit_provisional_observation( + forged, replace(revision, account_id="other", reporting_revision_id="forged"), [] + ) + + +def test_no_window_cannot_enable_official_close_lag(): + with pytest.raises(ValidationError, match="require restatement_window"): + _capabilities(restatement_window=None, official_close_lag="P3D") + + +async def test_unsealed_retry_refreshes_only_execution_deadline(make_harness): + producer, store, fetch, clock = await make_harness(_capabilities(restatement_window=None)) + source = producer._source + original_fetch = source._fetch + requests = [] + + def not_ready_once(request): + assert request.deadline_at > clock[0] + requests.append(request.model_dump(mode="json")) + return None if len(requests) == 1 else original_fetch(request) + + source._fetch = not_ready_once + await producer.run_worker() + assert await _revisions(store) == () + original_deadline = requests[0]["deadline_at"] + clock[0] += timedelta(hours=2) + await producer.run_worker() + assert len(requests) == 2 + assert requests[1]["deadline_at"] != original_deadline + assert {k: v for k, v in requests[0].items() if k != "deadline_at"} == { + k: v for k, v in requests[1].items() if k != "deadline_at" + } + assert len(await _revisions(store)) == 1 + assert (await latest(store)).acquisition.ordinal == 0 + + +@pytest.mark.parametrize("invalid_kind", ["past", "naive"]) +async def test_invalid_source_boundary_commits_nothing(make_harness, invalid_kind): + producer, store, fetch, clock = await make_harness(_capabilities(restatement_window=None)) + fetch.provisional_until = ( + clock[0] - timedelta(seconds=1) if invalid_kind == "past" else clock[0].replace(tzinfo=None) + ) + with pytest.raises(ValidationError, match="provisional_until"): + await producer.run_worker() + assert await _revisions(store) == () + obligation = await _only_obligation(store) + assert ( + await store.get_restatement_checkpoint( + account_id=ACCOUNT, reporting_obligation_id=obligation.reporting_obligation_id + ) + is None + ) + fetch.provisional_until = clock[0] + await producer.run_worker() + assert (await latest(store)).acquisition.ordinal == 0 + assert (await latest(store)).next_due_at is None + + +async def test_cancellation_after_checkpoint_rolls_back_publication(make_harness, monkeypatch): + producer, store, fetch, clock = await make_harness(_capabilities(restatement_window=None)) + original = store.record_restatement_checkpoint + + async def cancelled(checkpoint): + await original(checkpoint) + raise asyncio.CancelledError + + monkeypatch.setattr(store, "record_restatement_checkpoint", cancelled) + with pytest.raises(asyncio.CancelledError): + await producer.run_worker() + assert await _revisions(store) == () + monkeypatch.setattr(store, "record_restatement_checkpoint", original) + await producer.run_worker() + assert len(await _revisions(store)) == 1 + assert fetch.calls == ["PROVISIONAL_SNAPSHOT"] + + +async def test_watermark_and_history_advance_with_unchanged_rows(make_harness): + producer, store, fetch, clock = await make_harness(_capabilities(restatement_window=None)) + source = producer._source + original_fetch = source._fetch + complete = False + + def observed(request): + result = original_fetch(request) + return replace( + result, + data_through=request.period.end - (timedelta(0) if complete else timedelta(minutes=30)), + ) + + source._fetch = observed + await producer.run_worker() + first = (await _revisions(store))[0] + complete = True + clock[0] += timedelta(hours=1) + await producer.run_worker() + history = await _revisions(store) + assert len(history) == 2 + assert history[1].data_through > first.data_through + assert history[1].source_manifest_sha256 == first.source_manifest_sha256 + assert (await latest(store)).revision_id == history[1].reporting_revision_id + + +async def test_schema_reinstallation_preserves_observations_and_row_layout(make_harness): + producer, store, fetch, clock = await make_harness(_capabilities(restatement_window=None)) + await producer.run_worker() + prior = await latest(store) + prior_revisions = await _revisions(store) + await store.create_schema() + assert await latest(store) == prior + assert await _revisions(store) == prior_revisions + if isinstance(store, PgReportingLedgerStore): + async with store._pool.connection() as connection: + columns = await ( + await connection.execute( + "SELECT column_name FROM information_schema.columns" + " WHERE table_schema=current_schema() AND table_name='reporting_revision_rows'" + " ORDER BY ordinal_position" + ) + ).fetchall() + assert [row[0] for row in columns] == ["reporting_revision_id", "ordinal", "row_payload"] + import psycopg + + async with store._pool.connection() as connection: + with pytest.raises(psycopg.errors.RaiseException, match="immutable"): + async with connection.transaction(): + await connection.execute("DELETE FROM reporting_provisional_observations") + assert await latest(store) == prior + + +async def test_generation_and_replay_identity_cannot_be_rebound(make_harness): + producer, store, fetch, clock = await make_harness(_capabilities(restatement_window=None)) + await producer.run_worker() + observation = await latest(store) + acquisition = observation.acquisition + request = acquisition.request() + wrong_generation = request.model_copy( + update={ + "identity": request.identity.model_copy( + update={"delivery_config_version": request.identity.delivery_config_version + 1} + ) + } + ) + forged = replace(acquisition, ordinal=1, request_json=wrong_generation.model_dump_json()) + with pytest.raises(LedgerConflictError, match="generation differs"): + await store.reserve_provisional_acquisition(forged) + with pytest.raises(LedgerConflictError, match="already reserved"): + await store.reserve_provisional_acquisition(replace(acquisition, ordinal=1)) + with pytest.raises(LedgerConflictError, match="replay differs"): + await store.commit_provisional_observation( + replace(observation, acquisition=replace(forged, ordinal=0)), + (await _revisions(store))[0], + [], + ) + assert await latest(store) == observation + + +async def test_notification_failure_rolls_back_revision_checkpoint_and_history( + make_harness, monkeypatch +): + producer, store, fetch, clock = await make_harness(_capabilities(restatement_window=None)) + original = store._record_notification + if isinstance(store, PgReportingLedgerStore): + + async def fail(connection, event): + await original(connection, event) + raise RuntimeError("notification fault") + + else: + + def fail(event): + original(event) + raise RuntimeError("notification fault") + + monkeypatch.setattr(store, "_record_notification", fail) + with pytest.raises(RuntimeError, match="notification fault"): + await producer.run_worker() + assert await _revisions(store) == () + obligation = await _only_obligation(store) + assert ( + await store.get_restatement_checkpoint( + account_id=ACCOUNT, reporting_obligation_id=obligation.reporting_obligation_id + ) + is None + ) + if isinstance(store, PgReportingLedgerStore): + async with store._pool.connection() as connection: + counts = await ( + await connection.execute( + "SELECT (SELECT count(*) FROM reporting_notification_events)," + " (SELECT count(*) FROM reporting_ledger_changes WHERE record_kind='revision')," + " (SELECT count(*) FROM reporting_provisional_observations)" + ) + ).fetchone() + assert counts == (0, 0, 0) + else: + assert not store._notification_state.events + assert not any(change[2] == "revision" for change in store._changes) + assert not store._provisional_observations + monkeypatch.setattr(store, "_record_notification", original) + await producer.run_worker() + assert len(await _revisions(store)) == 1 + assert fetch.calls == ["PROVISIONAL_SNAPSHOT"] + + +async def test_reserved_snapshot_retry_survives_explicit_close_boundary(make_harness): + producer, store, fetch, clock = await make_harness( + _capabilities(restatement_window="PT3H", official_close_lag="PT4H") + ) + await producer.run_worker() + source = producer._source + original_fetch = source._fetch + requests = [] + + def not_ready_once(request): + requests.append(request) + return None if len(requests) == 1 else original_fetch(request) + + source._fetch = not_ready_once + clock[0] += timedelta(hours=1) + await producer.run_worker() + assert len(await _revisions(store)) == 1 + clock[0] += timedelta(hours=4) + await producer.run_worker() + assert requests[1].identity == requests[0].identity + assert requests[1].publication_class == "PROVISIONAL_SNAPSHOT" + assert [revision.finality for revision in await _revisions(store)] == ["snapshot", "snapshot"] + await producer.run_worker() + history = await _revisions(store) + assert sorted(revision.finality for revision in history) == ["official", "snapshot", "snapshot"] + official = next(revision for revision in history if revision.finality == "official") + assert (await latest(store)).revision_id == official.reporting_revision_id + assert (await latest(store)).acquisition.ordinal == 2 + + +async def test_empty_successful_reads_still_append_observations(make_harness): + producer, store, fetch, clock = await make_harness(_capabilities(restatement_window=None)) + source = producer._source + original_fetch = source._fetch + source._fetch = lambda request: replace(original_fetch(request), rows=[]) + await producer.run_worker() + clock[0] += timedelta(hours=1) + await producer.run_worker() + history = await _revisions(store) + assert len(history) == 2 + assert all(revision.row_count == 0 and revision.readable for revision in history) + assert history[1].supersedes_reporting_revision_id == history[0].reporting_revision_id diff --git a/tests/test_reporting_settling.py b/tests/test_reporting_settling.py index 9ddbf5436..54e23d5e0 100644 --- a/tests/test_reporting_settling.py +++ b/tests/test_reporting_settling.py @@ -98,6 +98,8 @@ def __call__(self, request: Any) -> InlineFetchResult: async def _harness( capabilities: ReportingSourceCapabilitiesV1, + *, + store_factory: Any = None, ) -> tuple[ ReportingProducer, InMemoryReportingLedgerStore, @@ -114,7 +116,12 @@ async def _harness( seals=InMemorySealStore(), clock=lambda: clock[0], ) - store = InMemoryReportingLedgerStore(clock=lambda: clock[0]) + store = ( + InMemoryReportingLedgerStore(clock=lambda: clock[0]) + if store_factory is None + else store_factory(lambda: clock[0]) + ) + await store.create_schema() configuration = ReportingConfiguration( delivery_config_id=CONFIG_ID, delivery_config_version=1, @@ -181,7 +188,7 @@ async def _revisions(store: InMemoryReportingLedgerStore): ) -async def test_unchanged_refresh_honors_cadence_without_committing_a_revision() -> None: +async def test_unchanged_refresh_honors_cadence_and_commits_an_observation() -> None: producer, store, fetch, clock = await _harness( _capabilities( restatement_window="P3D", @@ -198,11 +205,11 @@ async def test_unchanged_refresh_honors_cadence_without_committing_a_revision() clock[0] += timedelta(minutes=1) unchanged = await producer.run_worker() - assert unchanged.revisions_committed == [] + assert len(unchanged.revisions_committed) == 1 assert fetch.calls == ["PROVISIONAL_SNAPSHOT", "PROVISIONAL_SNAPSHOT"] - assert len(await _revisions(store)) == 1 + assert len(await _revisions(store)) == 2 - # The successful no-op advances both the cadence clock and execution + # The successful observation advances both the cadence clock and execution # ordinal, so an immediate worker turn neither polls nor replays it. await producer.run_worker() assert len(fetch.calls) == 2 @@ -288,21 +295,21 @@ async def test_slice_provisional_until_overrides_declared_window() -> None: assert (await _revisions(store))[-1].finality == "official" -async def test_source_without_a_window_keeps_one_shot_behavior() -> None: +async def test_source_without_a_window_retains_its_final_default_read() -> None: producer, store, fetch, clock = await _harness(_capabilities(restatement_window=None)) await producer.run_worker() obligation = await _only_obligation(store) clock[0] = obligation.period.end + timedelta(days=10) await producer.run_worker() - assert fetch.calls == ["PROVISIONAL_SNAPSHOT"] - assert len(await _revisions(store)) == 1 - assert ( - await store.get_restatement_checkpoint( - account_id=ACCOUNT, - reporting_obligation_id=obligation.reporting_obligation_id, - ) - is None + assert fetch.calls == ["PROVISIONAL_SNAPSHOT", "PROVISIONAL_SNAPSHOT"] + assert len(await _revisions(store)) == 2 + checkpoint = await store.get_restatement_checkpoint( + account_id=ACCOUNT, + reporting_obligation_id=obligation.reporting_obligation_id, ) + assert checkpoint is not None and checkpoint.next_observation == 2 + await producer.run_worker() + assert len(fetch.calls) == 2 def test_settling_options_require_a_declared_window_and_safe_cadence() -> None: