diff --git a/docs/decisions/0043-separate-file-acquisition-progress-from-publication-progress.md b/docs/decisions/0043-separate-file-acquisition-progress-from-publication-progress.md new file mode 100644 index 00000000..ba2e2ee2 --- /dev/null +++ b/docs/decisions/0043-separate-file-acquisition-progress-from-publication-progress.md @@ -0,0 +1,98 @@ +--- +name: adr-0043-separate-file-acquisition-progress-from-publication-progress +version: "1.0.0" +description: > + Persist source-scoped accepted-change checkpoints separately from contiguous + Runtime-visibility publication watermarks. +--- + +# 0043. Separate File acquisition progress from publication progress + +- Status: accepted +- Date: 2026-07-23 +- Refines: ADR-0037, ADR-0039, ADR-0040, ADR-0041, ADR-0042 + +## Context + +A durable File import request proves that a source change was accepted, but it +does not prove that compilation, indexing, or publication completed. Treating +the import job state as one generic checkpoint makes operators unable to tell a +recoverable publication pause from completed Runtime visibility. It also makes +an out-of-order later completion capable of hiding an earlier failed change if +the source reports only its greatest completed ordinal. + +Issue #29 needs two source-level progress signals without turning either signal +into Runtime authorization or claiming the deferred provider `checkpoint` +operation from the standard ProviderPort. + +## Decision + +Each accepted File import job and trusted File tombstone receives one immutable, +database-ordered `file_source_acquisition_checkpoint`, scoped by Organization +and Source. The import checkpoint is appended in the same transaction that +creates its acquisition/job; the tombstone checkpoint is appended in the same +transaction as tombstone visibility and cleanup intent. Opaque references are +deterministic SHA-256-derived tokens, while positive sequence numbers remain +database-owned ordering facts. Idempotent replay does not append another row. + +Each unchanged result, active Revision event, and tombstone appends one distinct +immutable `file_source_publish_watermark` row at its existing visibility +transaction boundary. Prepared, indexed, claimed, interrupted, or failed state +cannot create that row. Exact foreign keys bind each completion to its accepted +checkpoint, durable job or cleanup intent, Resource, and Revision. + +The read-only `ContextControl.read_file_source_progress` seam returns the latest +accepted checkpoint and the highest *contiguous* completed sequence. If change +3 completes while change 2 is paused, completion 3 remains durably recorded but +the exposed watermark remains 1. When recovery completes change 2, the exposed +watermark becomes 3 without rewriting either append-only stream. Calls are +operation-bound and Organization-scoped through a SECURITY DEFINER function; +all application roles are denied direct table access. + +Runtime receives no table privilege or dependency on this status. Current +Resource state, active Revision, policy, audience, and AuthorizationKernel +checks remain the only delivery authority. The File capability manifest also +continues to report the standard provider cursor/checkpoint operations as +unavailable: this issue activates operational progress for the manual File +carrier, not provider `readChanges` acknowledgement. + +## Rationale + +Acceptance and visibility have different transaction boundaries and different +failure meanings, so one mutable cursor cannot truthfully represent both. Two +append-only streams retain the causal history needed to diagnose a pause, +while exposing only the contiguous publication prefix prevents an +out-of-order completion from overstating Runtime freshness. Database-owned +ordering, deterministic opaque references, and exact durable-lineage foreign +keys make each accepted ordering fact stable without promoting progress +metadata into delivery authority. Existing durable lineage is ordered once +during the initial upgrade; after any progress exists, downgrade is refused so +concurrent online lock order can never be reconstructed differently or accept +new changes through an older migration boundary. + +Keeping the operator read seam in ContextControl also preserves the existing +ProviderPort contract: File progress can be observed now without pretending a +provider-native change cursor has been acknowledged or making Runtime depend +on an operational status surface. + +## Consequences + +- An accepted change remains observable through compilation, indexing, + interruption, failure, and recovery without falsely claiming visibility. +- First publication, replacement, unchanged classification, and tombstone each + advance publication progress only inside their committed visibility outcome. +- Concurrent or out-of-order completion cannot skip an earlier gap; recovery + closes the gap and deterministically catches the watermark up. +- Both streams are append-only, forced-RLS tenant tables with no source content. + Downgrade is allowed only while both streams are empty; after any accepted + change, a forward fix is required so sequence numbers and opaque references + cannot be renumbered from historical timestamps. +- Cross-source aggregation, UI, provider cursor formats, resync, and watermark- + based authorization remain inactive. + +## Revisit trigger + +Revisit before provider-native `readChanges`/checkpoint activation, source +offboarding, resync, progress compaction, cross-source aggregation, or deletion +recovery. Any change must preserve separate signals, contiguous watermarks, +exact tenant lineage, replay idempotency, and Runtime independence. diff --git a/docs/decisions/README.md b/docs/decisions/README.md index 3862cd3f..65a6ec2b 100644 --- a/docs/decisions/README.md +++ b/docs/decisions/README.md @@ -41,6 +41,7 @@ kernel, capability separation, and publication visibility model. | Structural Markdown compiler | [0038 — Structural Markdown units](0038-compile-and-publish-structural-markdown.md) | Explicit v2 compilation publishes one coherent Fragment per heading, paragraph, list, fenced code block, or table with exact provenance and same-Fragment heading ancestry | Silent v1 reinterpretation, item/cell fragmentation, post-authorization parent expansion, or unbudgeted context | | Unchanged File acquisition | [0039 — File acquisition no-op](0039-deduplicate-unchanged-file-acquisitions.md) | Tenant/source/resource-scoped canonical identity and one PostgreSQL guard lock classify a complete active artifact before publication; each observation retains an immutable digest-only outcome | Cross-tenant/global deduplication, process-local locking, partial-artifact reuse, silent version reuse, or no-op by bypassing publication validation | | File Resource deletion | [0042 — Tombstone before cleanup](0042-tombstone-file-resources-before-cleanup.md) | One trusted Control transaction tombstones the active File Resource, advances its Organization Policy Epoch, and records immutable pending cleanup lineage before any physical deletion | Cleanup-defined visibility, caller-authored tenant/epoch/cleanup identity, index deletion as authorization, restore, or native watcher claims | +| File Source progress | [0043 — Separate acquisition and publication progress](0043-separate-file-acquisition-progress-from-publication-progress.md) | Append accepted changes separately from contiguous Runtime-visibility completion and expose them through an Organization/Source-scoped Control read | One ambiguous checkpoint, skipped publication gaps, Runtime authorization from watermarks, or false standard ProviderPort capability claims | Each baseline ADR is `accepted` and contains Context, Decision, Rationale, Consequences, and Revisit trigger sections. A revisit trigger permits review; it @@ -130,3 +131,4 @@ touched: - [0040 — Atomic File replacement](0040-stage-and-atomically-activate-file-replacements.md) - [0041 — Durable File publication recovery](0041-recover-file-publication-by-durable-boundary.md) - [0042 — Tombstone File Resources before cleanup](0042-tombstone-file-resources-before-cleanup.md) +- [0043 — Separate File acquisition and publication progress](0043-separate-file-acquisition-progress-from-publication-progress.md) diff --git a/docs/design/2026-07-18-context-engine-implementation-design.md b/docs/design/2026-07-18-context-engine-implementation-design.md index 57e5ccf2..ad57e275 100644 --- a/docs/design/2026-07-18-context-engine-implementation-design.md +++ b/docs/design/2026-07-18-context-engine-implementation-design.md @@ -594,6 +594,17 @@ Duplicate and older deletion observations return the original result without another epoch bump. Physical cleanup, restore, native watcher detection, source offboarding, and cleanup-job execution remain inactive. +ADR-0043 separates operational File progress into two append-only, +Organization/Source-scoped streams. Import-job or tombstone acceptance appends +an acquisition checkpoint at the durable acceptance boundary. Unchanged +classification, active Revision publication/replacement, or tombstone commit +appends a distinct publish completion at the existing visibility boundary. The +Control status seam exposes the latest accepted sequence and only the greatest +contiguous completed sequence: a later publication cannot jump over an earlier +interrupted job, while recovery closes the gap without rewriting history. +Runtime has no privilege on or dependency upon either signal, and the deferred +standard ProviderPort cursor/checkpoint operations remain unavailable. + Because Runtime resolves through multiple SQL statements at `READ COMMITTED`, each UserActor transaction takes an Organization-scoped shared publication barrier and activation takes the matching exclusive transaction barrier around diff --git a/engine/control/__init__.py b/engine/control/__init__.py index b3070842..f5a97ce9 100644 --- a/engine/control/__init__.py +++ b/engine/control/__init__.py @@ -38,6 +38,13 @@ PreparedFileImport, PrepareFileImport, ) +from engine.control.file_source_progress import ( + FileSourceAcquisitionCheckpoint, + FileSourceChangeKind, + FileSourceProgress, + FileSourcePublishOutcome, + FileSourcePublishWatermark, +) from engine.control.module import ContextControl, ControlStorePort __all__ = [ @@ -56,6 +63,11 @@ "FileImportPath", "FileImportReceiver", "FileRootRef", + "FileSourceAcquisitionCheckpoint", + "FileSourceChangeKind", + "FileSourceProgress", + "FileSourcePublishOutcome", + "FileSourcePublishWatermark", "FileResourceTombstone", "RegisterFileSource", "PrepareFileImport", diff --git a/engine/control/authority.py b/engine/control/authority.py index 2021fda4..7a91c64f 100644 --- a/engine/control/authority.py +++ b/engine/control/authority.py @@ -22,6 +22,7 @@ class ControlOperation(StrEnum): IMPORT_FILE = "import_file" REGISTER_SOURCE = "register_source" READ_SOURCE = "read_source" + READ_SOURCE_PROGRESS = "read_source_progress" TOMBSTONE_FILE_RESOURCE = "tombstone_file_resource" diff --git a/engine/control/file_source_progress.py b/engine/control/file_source_progress.py new file mode 100644 index 00000000..779c1212 --- /dev/null +++ b/engine/control/file_source_progress.py @@ -0,0 +1,208 @@ +"""Read-only File Source acquisition-checkpoint and publication-watermark contracts.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime +from enum import StrEnum +from uuid import UUID + +from engine.control.contracts import ( + SourceRef, + _require_bounded_text, + _require_token, + _require_utc, +) + +_MAX_BIGINT = 9_223_372_036_854_775_807 + + +class FileSourceChangeKind(StrEnum): + """Closed durable change carriers that participate in File progress.""" + + FILE_IMPORT = "file_import" + FILE_TOMBSTONE = "file_tombstone" + + +class FileSourcePublishOutcome(StrEnum): + """Closed visibility outcomes that may advance a publish watermark.""" + + PUBLISHED = "published" + REPLACED = "replaced" + UNCHANGED = "unchanged" + TOMBSTONED = "tombstoned" + + +def _require_sequence(name: str, value: object) -> int: + if type(value) is not int or not 1 <= value <= _MAX_BIGINT: + raise ValueError(f"{name} must fit a positive signed bigint") + return value + + +def _require_resource_ref(value: object) -> str: + return _require_bounded_text("File Source progress ResourceRef", value, 512) + + +def _require_progress_ref(name: str, value: object, prefix: str) -> str: + token = _require_token(name, value) + digest = token.removeprefix(prefix) + if not token.startswith(prefix) or len(digest) != 64 or any( + character not in "0123456789abcdef" for character in digest + ): + raise ValueError(f"{name} is not a recognized opaque reference") + return token + + +def _require_lineage( + *, + change_kind: FileSourceChangeKind, + acquisition_ref: object, + job_ref: object, + cleanup_intent_ref: object, + resource_ref: object, + revision_ref: object, + event_ref: object, + event_sequence: object, + allow_unresolved_import_resource: bool, +) -> None: + if type(change_kind) is not FileSourceChangeKind: + raise TypeError("File Source progress change_kind is invalid") + if change_kind is FileSourceChangeKind.FILE_IMPORT: + if type(acquisition_ref) is not UUID or type(job_ref) is not UUID: + raise TypeError("File import progress requires acquisition and job lineage") + if any( + value is not None + for value in (cleanup_intent_ref, event_ref, event_sequence) + ): + raise ValueError("File import progress cannot carry tombstone lineage") + if allow_unresolved_import_resource and resource_ref is revision_ref is None: + return + else: + if type(cleanup_intent_ref) is not UUID: + raise TypeError("File tombstone progress requires cleanup lineage") + if acquisition_ref is not None or job_ref is not None: + raise ValueError("File tombstone progress cannot carry import lineage") + _require_token("File tombstone progress event_ref", event_ref) + _require_sequence("File tombstone progress event_sequence", event_sequence) + _require_resource_ref(resource_ref) + if type(revision_ref) is not UUID: + raise TypeError("File Source progress revision_ref must be UUID") + + +@dataclass(frozen=True, slots=True) +class FileSourceAcquisitionCheckpoint: + """Latest source change durably accepted, irrespective of visibility.""" + + sequence: int + checkpoint_ref: str = field(repr=False) + change_kind: FileSourceChangeKind + acquisition_ref: UUID | None = field(repr=False) + job_ref: UUID | None = field(repr=False) + cleanup_intent_ref: UUID | None = field(repr=False) + resource_ref: str | None + revision_ref: UUID | None = field(repr=False) + event_ref: str | None = field(repr=False) + event_sequence: int | None + accepted_at: datetime + + def __post_init__(self) -> None: + _require_sequence("File Source acquisition sequence", self.sequence) + _require_progress_ref( + "File Source acquisition checkpoint_ref", self.checkpoint_ref, "facp_" + ) + _require_lineage( + change_kind=self.change_kind, + acquisition_ref=self.acquisition_ref, + job_ref=self.job_ref, + cleanup_intent_ref=self.cleanup_intent_ref, + resource_ref=self.resource_ref, + revision_ref=self.revision_ref, + event_ref=self.event_ref, + event_sequence=self.event_sequence, + allow_unresolved_import_resource=True, + ) + _require_utc("File Source acquisition accepted_at", self.accepted_at) + + +@dataclass(frozen=True, slots=True) +class FileSourcePublishWatermark: + """Latest contiguous accepted change fully reflected in Runtime visibility.""" + + sequence: int + watermark_ref: str = field(repr=False) + checkpoint_ref: str = field(repr=False) + change_kind: FileSourceChangeKind + outcome: FileSourcePublishOutcome + acquisition_ref: UUID | None = field(repr=False) + job_ref: UUID | None = field(repr=False) + cleanup_intent_ref: UUID | None = field(repr=False) + resource_ref: str + revision_ref: UUID = field(repr=False) + event_ref: str | None = field(repr=False) + event_sequence: int | None + published_at: datetime + + def __post_init__(self) -> None: + _require_sequence("File Source publish sequence", self.sequence) + _require_progress_ref( + "File Source publish watermark_ref", self.watermark_ref, "fpwm_" + ) + _require_progress_ref( + "File Source publish checkpoint_ref", self.checkpoint_ref, "facp_" + ) + if type(self.outcome) is not FileSourcePublishOutcome: + raise TypeError("File Source publish outcome is invalid") + _require_lineage( + change_kind=self.change_kind, + acquisition_ref=self.acquisition_ref, + job_ref=self.job_ref, + cleanup_intent_ref=self.cleanup_intent_ref, + resource_ref=self.resource_ref, + revision_ref=self.revision_ref, + event_ref=self.event_ref, + event_sequence=self.event_sequence, + allow_unresolved_import_resource=False, + ) + if ( + self.change_kind is FileSourceChangeKind.FILE_TOMBSTONE + ) is not (self.outcome is FileSourcePublishOutcome.TOMBSTONED): + raise ValueError("File Source publish outcome does not match its change") + _require_utc("File Source publish published_at", self.published_at) + + +@dataclass(frozen=True, slots=True) +class FileSourceProgress: + """Organization/Source-scoped read model for the two progress signals.""" + + organization_id: UUID = field(repr=False) + source_ref: SourceRef + acquisition_checkpoint: FileSourceAcquisitionCheckpoint | None + publish_watermark: FileSourcePublishWatermark | None + + def __post_init__(self) -> None: + if type(self.organization_id) is not UUID: + raise TypeError("File Source progress organization_id must be UUID") + if type(self.source_ref) is not SourceRef: + raise TypeError("File Source progress source_ref must be SourceRef") + if self.acquisition_checkpoint is not None and type( + self.acquisition_checkpoint + ) is not FileSourceAcquisitionCheckpoint: + raise TypeError("File Source acquisition checkpoint is invalid") + if self.publish_watermark is not None and type( + self.publish_watermark + ) is not FileSourcePublishWatermark: + raise TypeError("File Source publish watermark is invalid") + if self.publish_watermark is not None and ( + self.acquisition_checkpoint is None + or self.publish_watermark.sequence + > self.acquisition_checkpoint.sequence + ): + raise ValueError("File Source publish watermark cannot exceed checkpoint") + if ( + self.publish_watermark is not None + and self.acquisition_checkpoint is not None + and self.publish_watermark.sequence == self.acquisition_checkpoint.sequence + and self.publish_watermark.checkpoint_ref + != self.acquisition_checkpoint.checkpoint_ref + ): + raise ValueError("File Source publish watermark lineage is invalid") diff --git a/engine/control/module.py b/engine/control/module.py index 7aa6cce3..cee59047 100644 --- a/engine/control/module.py +++ b/engine/control/module.py @@ -25,6 +25,7 @@ TombstoneFileResource, ) from engine.control.file_imports import PreparedFileImport, PrepareFileImport +from engine.control.file_source_progress import FileSourceProgress class ControlStorePort(Protocol): @@ -54,6 +55,12 @@ def tombstone_file_resource( command: TombstoneFileResource, ) -> FileResourceTombstone: ... + def read_file_source_progress( + self, + call: TrustedControlCall, + source_ref: SourceRef, + ) -> FileSourceProgress: ... + class ContextControl: """Own trusted File enrollment, read-back, and import preparation.""" @@ -71,6 +78,7 @@ def __init__( "prepare_file_import", "register_file_source", "read_source", + "read_file_source_progress", "tombstone_file_resource", ): if not callable(getattr(store, method_name, None)): @@ -179,6 +187,41 @@ def prepare_file_import( "File import preparation is unavailable" ) from None + def read_file_source_progress( + self, + call: TrustedControlCall, + source_ref: SourceRef, + ) -> FileSourceProgress: + """Read separate durable acceptance and visibility progress signals.""" + + if type(source_ref) is not SourceRef: + raise TypeError("read_file_source_progress requires SourceRef") + try: + _validate_and_consume_control_call( + call, + authority=self._authority, + expected_operation=ControlOperation.READ_SOURCE_PROGRESS, + checked_at=self._clock(), + ) + progress = self._store.read_file_source_progress(call, source_ref) + if ( + type(progress) is not FileSourceProgress + or progress.organization_id != call.organization_id + or progress.source_ref != source_ref + ): + raise SourceControlUnavailable( + "source store returned mismatched File progress" + ) + return progress + except (ControlOperatorAuthenticationRejected, SourceNotAvailable): + raise SourceNotAvailable from None + except SourceControlUnavailable: + raise + except Exception: + raise SourceControlUnavailable( + "File Source progress read is unavailable" + ) from None + def tombstone_file_resource( self, call: TrustedControlCall, diff --git a/engine/persistence/control_sources.py b/engine/persistence/control_sources.py index 7167a44b..a57d7ea9 100644 --- a/engine/persistence/control_sources.py +++ b/engine/persistence/control_sources.py @@ -17,6 +17,11 @@ FILE_IMPORT_CAPABILITY_MANIFEST, FileResourceTombstone, FileRootRef, + FileSourceAcquisitionCheckpoint, + FileSourceChangeKind, + FileSourceProgress, + FileSourcePublishOutcome, + FileSourcePublishWatermark, RegisterFileSource, SourceControlUnavailable, SourceManifest, @@ -326,6 +331,96 @@ def prepare_file_import( "File import Control database authority is unavailable" ) from None + def read_file_source_progress( + self, + call: TrustedControlCall, + source_ref: SourceRef, + ) -> FileSourceProgress: + """Read the current contiguous progress signals and durable lineage.""" + + if type(call) is not TrustedControlCall or type(source_ref) is not SourceRef: + raise SourceNotAvailable + try: + with self._engine.begin() as connection: + assert_control_role(connection) + _set_organization_context(connection, call.organization_id) + row = connection.execute( + text( + """ + SELECT * + FROM public.context_control_read_file_source_progress( + :organization_id, :source_id + ) + """ + ), + { + "organization_id": call.organization_id, + "source_id": source_ref.value, + }, + ).mappings().one_or_none() + if row is None: + raise SourceNotAvailable + checkpoint = ( + None + if row["acquisition_sequence"] is None + else FileSourceAcquisitionCheckpoint( + sequence=row["acquisition_sequence"], + checkpoint_ref=row["acquisition_checkpoint_ref"], + change_kind=FileSourceChangeKind( + row["acquisition_change_kind"] + ), + acquisition_ref=row["acquisition_acquisition_id"], + job_ref=row["acquisition_job_id"], + cleanup_intent_ref=row[ + "acquisition_cleanup_intent_id" + ], + resource_ref=row["acquisition_resource_ref"], + revision_ref=row["acquisition_revision_id"], + event_ref=row["acquisition_event_ref"], + event_sequence=row["acquisition_event_sequence"], + accepted_at=row["acquisition_accepted_at"], + ) + ) + watermark = ( + None + if row["publish_sequence"] is None + else FileSourcePublishWatermark( + sequence=row["publish_sequence"], + watermark_ref=row["publish_watermark_ref"], + checkpoint_ref=row["publish_checkpoint_ref"], + change_kind=FileSourceChangeKind( + row["publish_change_kind"] + ), + outcome=FileSourcePublishOutcome(row["publish_outcome"]), + acquisition_ref=row["publish_acquisition_id"], + job_ref=row["publish_job_id"], + cleanup_intent_ref=row["publish_cleanup_intent_id"], + resource_ref=row["publish_resource_ref"], + revision_ref=row["publish_revision_id"], + event_ref=row["publish_event_ref"], + event_sequence=row["publish_event_sequence"], + published_at=row["publish_published_at"], + ) + ) + return FileSourceProgress( + organization_id=call.organization_id, + source_ref=source_ref, + acquisition_checkpoint=checkpoint, + publish_watermark=watermark, + ) + except SourceNotAvailable: + raise + except ( + DBAPIError, + SQLAlchemyError, + AssertionError, + TypeError, + ValueError, + ): + raise SourceControlUnavailable( + "File Source progress database authority is unavailable" + ) from None + def tombstone_file_resource( self, call: TrustedControlCall, diff --git a/engine/persistence/schema_security_manifest.yaml b/engine/persistence/schema_security_manifest.yaml index 3a594c68..bad70c37 100644 --- a/engine/persistence/schema_security_manifest.yaml +++ b/engine/persistence/schema_security_manifest.yaml @@ -1,5 +1,5 @@ { - "manifestVersion": "16.0.0", + "manifestVersion": "17.0.0", "controlOperations": [ { "name": "register_file_source", @@ -33,7 +33,19 @@ "databaseOwnedTime": true, "idempotencyBinding": ["organization_id", "source_id", "resource_ref"], "runtimeVisibilityBarrier": "Organization-scoped shared Runtime/exclusive publication transaction advisory lock", - "atomicWrites": ["context_resource", "organization_policy_epoch", "file_resource_cleanup_intent"] + "atomicWrites": ["context_resource", "organization_policy_epoch", "file_resource_cleanup_intent", "file_source_acquisition_checkpoint", "file_source_publish_watermark"] + }, + { + "name": "read_file_source_progress", + "databaseFunction": "context_control_read_file_source_progress", + "role": "context_engine_control", + "definerRole": "context_engine_worker_lease_definer", + "directTableMutationAllowed": false, + "trustedOrganizationSource": "TrustedControlCall", + "scope": ["organization_id", "source_id"], + "watermarkRule": "highest contiguous acquisition sequence with a committed visibility outcome", + "runtimeAuthorityAllowed": false, + "reads": ["context_source", "file_source_acquisition_checkpoint", "file_source_publish_watermark"] }, { "name": "issue_noop_worker_lease", @@ -74,7 +86,7 @@ "databaseOwnedTime": true, "filesystemAccessAllowed": false, "durableJobCreationAllowed": true, - "atomicWrites": ["source_version", "context_source", "file_acquisition", "file_import_job"] + "atomicWrites": ["source_version", "context_source", "file_acquisition", "file_import_job", "file_source_acquisition_checkpoint"] }, { "name": "issue_file_import_lease", @@ -125,7 +137,8 @@ "context_resource", "context_revision", "file_revision_snapshot", "context_fragment", "revision_publication_event", "exact_phrase_candidate", "resource_access_policy", - "membership_resource_field_right", "file_import_job" + "membership_resource_field_right", "file_import_job", + "file_source_publish_watermark" ] }, { @@ -149,11 +162,13 @@ "file_resource_ingestion_guard", "file_acquisition_result", "context_revision", "file_revision_snapshot", "context_fragment", "revision_publication_event", "exact_phrase_candidate", - "file_revision_replacement_plan", "file_import_job" + "file_revision_replacement_plan", "file_import_job", + "file_source_publish_watermark" ], "activationAtomicWrites": [ "context_resource", "revision_publication_event", - "file_revision_supersession", "file_import_job" + "file_revision_supersession", "file_import_job", + "file_source_publish_watermark" ] }, { @@ -175,10 +190,10 @@ "concurrentPublicationArbitration": "the Organization/Source/Resource ingestion guard serializes classification; a late identical job observes the winner as unchanged", "activationInvariant": "authorization and complete Fragment/index evidence are revalidated before one active-pointer transaction", "atomicWritesByBoundary": { - "acquired": ["file_resource_ingestion_guard", "file_acquisition_result", "file_publication_recovery", "file_import_job", "file_import_job_event"], + "acquired": ["file_resource_ingestion_guard", "file_acquisition_result", "file_source_publish_watermark", "file_publication_recovery", "file_import_job", "file_import_job_event"], "prepared": ["context_resource", "context_revision", "file_revision_snapshot", "context_fragment", "revision_publication_event", "resource_access_policy", "membership_resource_field_right", "file_publication_recovery", "file_import_job", "file_import_job_event"], "ready": ["exact_phrase_candidate", "revision_publication_event", "file_revision_replacement_plan", "file_publication_recovery", "file_import_job", "file_import_job_event"], - "completed": ["context_resource", "revision_publication_event", "file_revision_supersession", "file_publication_recovery", "file_import_job", "file_import_job_event"] + "completed": ["context_resource", "revision_publication_event", "file_source_publish_watermark", "file_revision_supersession", "file_publication_recovery", "file_import_job", "file_import_job_event"] } }, { @@ -2620,7 +2635,8 @@ "organizationColumn": "organization_id", "organizationInclusiveKeys": [ {"name": "pk_file_import_job", "kind": "primary_key", "columns": ["organization_id", "job_id"]}, - {"name": "uq_file_import_job_acquisition", "kind": "unique", "columns": ["organization_id", "acquisition_id"]} + {"name": "uq_file_import_job_acquisition", "kind": "unique", "columns": ["organization_id", "acquisition_id"]}, + {"name": "uq_file_import_job_progress_lineage", "kind": "unique", "columns": ["organization_id", "job_id", "acquisition_id", "source_id"]} ], "foreignKeys": [ { @@ -2928,7 +2944,8 @@ "organizationInclusiveKeys": [ {"name": "pk_file_resource_cleanup_intent", "kind": "primary_key", "columns": ["organization_id", "cleanup_intent_id"]}, {"name": "uq_file_resource_cleanup_intent_resource", "kind": "unique", "columns": ["organization_id", "resource_ref"]}, - {"name": "uq_file_resource_cleanup_intent_event", "kind": "unique", "columns": ["organization_id", "event_ref"]} + {"name": "uq_file_resource_cleanup_intent_event", "kind": "unique", "columns": ["organization_id", "event_ref"]}, + {"name": "uq_file_resource_cleanup_intent_progress_lineage", "kind": "unique", "columns": ["organization_id", "cleanup_intent_id", "source_id", "resource_ref", "revision_id"]} ], "foreignKeys": [ {"name": "fk_file_resource_cleanup_intent_source_same_organization", "columns": ["organization_id", "source_id"], "references": {"table": "context_source", "columns": ["organization_id", "source_id"]}}, @@ -2954,6 +2971,73 @@ "securityInvariantIds": ["TENANT-OWNERSHIP-001", "TENANT-FK-002", "RLS-FAIL-CLOSED-003", "INDEX-NOT-AUTHORITY-005", "REVOCATION-006"], "negativeTestIds": ["DB-001", "DB-004", "DB-008", "PG-FILE-TOMBSTONE-028"] }, + { + "name": "file_source_acquisition_checkpoint", + "classification": "tenant_owned", + "nonOwnerEvidence": {"evidenceId": "PG-FILE-PROGRESS-029", "selector": {"table": "file_source_acquisition_checkpoint"}}, + "purpose": "Immutable Organization/Source sequence for each durably accepted File import or tombstone change", + "organizationColumn": "organization_id", + "organizationInclusiveKeys": [ + {"name": "pk_file_source_acquisition_checkpoint", "kind": "primary_key", "columns": ["organization_id", "source_id", "sequence"]}, + {"name": "uq_file_source_acquisition_checkpoint_exact", "kind": "unique", "columns": ["organization_id", "source_id", "sequence", "checkpoint_ref", "change_kind"]}, + {"name": "uq_file_source_acquisition_checkpoint_ref", "kind": "unique", "columns": ["organization_id", "checkpoint_ref"]}, + {"name": "uq_file_source_acquisition_checkpoint_acquisition", "kind": "unique", "columns": ["organization_id", "acquisition_id"]}, + {"name": "uq_file_source_acquisition_checkpoint_cleanup", "kind": "unique", "columns": ["organization_id", "cleanup_intent_id"]} + ], + "foreignKeys": [ + {"name": "fk_file_source_acquisition_checkpoint_source_same_organization", "columns": ["organization_id", "source_id"], "references": {"table": "context_source", "columns": ["organization_id", "source_id"]}}, + {"name": "fk_file_source_acquisition_checkpoint_job_exact", "columns": ["organization_id", "job_id", "acquisition_id", "source_id"], "references": {"table": "file_import_job", "columns": ["organization_id", "job_id", "acquisition_id", "source_id"]}}, + {"name": "fk_file_source_acquisition_checkpoint_cleanup_exact", "columns": ["organization_id", "cleanup_intent_id", "source_id", "resource_ref", "revision_id"], "references": {"table": "file_resource_cleanup_intent", "columns": ["organization_id", "cleanup_intent_id", "source_id", "resource_ref", "revision_id"]}} + ], + "checkConstraints": [ + {"name": "ck_file_source_acquisition_checkpoint_sequence", "expression": "sequence BETWEEN 1 AND 9223372036854775807"}, + {"name": "ck_file_source_acquisition_checkpoint_ref", "expression": "checkpoint_ref is facp_ plus one SHA-256 hex digest"}, + {"name": "ck_file_source_acquisition_checkpoint_lineage", "expression": "exactly one complete file_import or file_tombstone durable lineage is present"} + ], + "rowLevelSecurity": {"enabled": true, "forced": true, "policies": [ + {"name": "file_source_acquisition_checkpoint_migrator_administration", "command": "ALL", "roles": ["context_engine_migrator"], "using": "true", "withCheck": "true"}, + {"name": "file_source_acquisition_checkpoint_file_progress_definer_select", "command": "SELECT", "roles": ["context_engine_worker_lease_definer"], "using": "organization_id = NULLIF(current_setting('app.organization_id', true), '')::uuid"}, + {"name": "file_source_acquisition_checkpoint_file_progress_definer_insert", "command": "INSERT", "roles": ["context_engine_worker_lease_definer"], "withCheck": "organization_id = NULLIF(current_setting('app.organization_id', true), '')::uuid"} + ]}, + "immutableRows": {"trigger": "file_source_acquisition_checkpoint_immutable", "function": "context_content_reject_mutation", "events": ["UPDATE", "DELETE"], "sqlstate": "55000"}, + "functionOnlyMutation": {"databaseFunctions": ["context_file_source_checkpoint_import_job", "context_file_source_checkpoint_tombstone"], "causalDatabaseFunctions": ["context_control_prepare_file_import", "context_control_tombstone_file_resource"], "definerRole": "context_engine_worker_lease_definer", "directTableMutationAllowed": false}, + "permittedOperations": {"context_engine_control": ["EXECUTE context_control_prepare_file_import", "EXECUTE context_control_read_file_source_progress", "EXECUTE context_control_tombstone_file_resource"], "context_engine_runtime": [], "context_engine_worker": [], "context_engine_worker_lease_definer": ["SELECT", "INSERT"]}, + "partitions": [], + "securityInvariantIds": ["TENANT-OWNERSHIP-001", "TENANT-FK-002", "RLS-FAIL-CLOSED-003", "WORKER-LEASE-007"], + "negativeTestIds": ["DB-001", "DB-004", "DB-008", "PG-FILE-PROGRESS-029"] + }, + { + "name": "file_source_publish_watermark", + "classification": "tenant_owned", + "nonOwnerEvidence": {"evidenceId": "PG-FILE-PROGRESS-029", "selector": {"table": "file_source_publish_watermark"}}, + "purpose": "Immutable per-change visibility completion; the operator watermark is its highest contiguous acquisition sequence", + "organizationColumn": "organization_id", + "organizationInclusiveKeys": [ + {"name": "pk_file_source_publish_watermark", "kind": "primary_key", "columns": ["organization_id", "source_id", "sequence"]}, + {"name": "uq_file_source_publish_watermark_ref", "kind": "unique", "columns": ["organization_id", "watermark_ref"]} + ], + "foreignKeys": [ + {"name": "fk_file_source_publish_watermark_checkpoint_exact", "columns": ["organization_id", "source_id", "sequence", "checkpoint_ref", "change_kind"], "references": {"table": "file_source_acquisition_checkpoint", "columns": ["organization_id", "source_id", "sequence", "checkpoint_ref", "change_kind"]}}, + {"name": "fk_file_source_publish_watermark_revision_same_organization", "columns": ["organization_id", "resource_ref", "revision_id"], "references": {"table": "context_revision", "columns": ["organization_id", "resource_ref", "revision_id"]}} + ], + "checkConstraints": [ + {"name": "ck_file_source_publish_watermark_sequence", "expression": "sequence BETWEEN 1 AND 9223372036854775807"}, + {"name": "ck_file_source_publish_watermark_ref", "expression": "watermark_ref is fpwm_ plus one SHA-256 hex digest"}, + {"name": "ck_file_source_publish_watermark_resource", "expression": "resource_ref is bounded nonblank Unicode without control characters"}, + {"name": "ck_file_source_publish_watermark_outcome", "expression": "import outcomes are published, replaced or unchanged; tombstone outcome is tombstoned"} + ], + "rowLevelSecurity": {"enabled": true, "forced": true, "policies": [ + {"name": "file_source_publish_watermark_migrator_administration", "command": "ALL", "roles": ["context_engine_migrator"], "using": "true", "withCheck": "true"}, + {"name": "file_source_publish_watermark_file_progress_definer_select", "command": "SELECT", "roles": ["context_engine_worker_lease_definer"], "using": "organization_id = NULLIF(current_setting('app.organization_id', true), '')::uuid"}, + {"name": "file_source_publish_watermark_file_progress_definer_insert", "command": "INSERT", "roles": ["context_engine_worker_lease_definer"], "withCheck": "organization_id = NULLIF(current_setting('app.organization_id', true), '')::uuid"} + ]}, + "immutableRows": {"trigger": "file_source_publish_watermark_immutable", "function": "context_content_reject_mutation", "events": ["UPDATE", "DELETE"], "sqlstate": "55000"}, + "functionOnlyMutation": {"databaseFunctions": ["context_file_source_append_publish_watermark", "context_file_source_publish_active_revision", "context_file_source_publish_unchanged_acquisition", "context_file_source_checkpoint_tombstone"], "causalDatabaseFunctions": ["context_worker_publish_file_import_v2", "context_worker_publish_structural_file_import_v2", "context_worker_stage_file_replacement", "context_worker_stage_structural_file_replacement", "context_worker_activate_file_replacement", "context_worker_acquire_file_publication", "context_worker_activate_recoverable_file_publication", "context_control_tombstone_file_resource"], "definerRole": "context_engine_worker_lease_definer", "directTableMutationAllowed": false}, + "permittedOperations": {"context_engine_control": ["EXECUTE context_control_read_file_source_progress", "EXECUTE context_control_tombstone_file_resource"], "context_engine_runtime": [], "context_engine_worker": ["EXECUTE context_worker_publish_file_import_v2", "EXECUTE context_worker_publish_structural_file_import_v2", "EXECUTE context_worker_stage_file_replacement", "EXECUTE context_worker_stage_structural_file_replacement", "EXECUTE context_worker_activate_file_replacement", "EXECUTE context_worker_acquire_file_publication", "EXECUTE context_worker_activate_recoverable_file_publication"], "context_engine_worker_lease_definer": ["SELECT", "INSERT"]}, + "partitions": [], + "securityInvariantIds": ["TENANT-OWNERSHIP-001", "TENANT-FK-002", "RLS-FAIL-CLOSED-003", "WORKER-LEASE-007"], + "negativeTestIds": ["DB-001", "DB-004", "DB-008", "PG-FILE-PROGRESS-029"] + }, { "name": "release_promotion_audit", "classification": "tenant_owned", diff --git a/eval/catalogs/m0-security-evidence.yaml b/eval/catalogs/m0-security-evidence.yaml index 59d34e47..16f5bcf3 100644 --- a/eval/catalogs/m0-security-evidence.yaml +++ b/eval/catalogs/m0-security-evidence.yaml @@ -53,6 +53,7 @@ {"id": "PG-FILE-REPLACEMENT-026", "layer": "postgres", "selector": "tests/integration/test_zz_file_revision_replacement.py::test_replacement_does_not_change_another_organization_resource"}, {"id": "PG-FILE-RECOVERY-027", "layer": "postgres", "selector": "tests/integration/test_zz_file_publication_recovery.py::test_recovery_tables_deny_direct_access_to_every_nonowner_role"}, {"id": "PG-FILE-TOMBSTONE-028", "layer": "postgres", "selector": "tests/integration/test_zz_file_resource_tombstone.py::test_tombstone_is_synchronously_empty_with_stale_physical_rows_and_index"}, + {"id": "PG-FILE-PROGRESS-029", "layer": "postgres", "selector": "tests/integration/test_zz_file_source_progress.py::test_progress_lineage_denies_direct_nonowner_access_and_cross_org_reads"}, {"id": "RUNTIME-TENANT-FK-002", "layer": "runtime", "selector": "tests/integration/test_runtime_authorized_evidence_integration.py::test_real_postgres_http_delivers_only_exact_authorized_evidence_bidirectionally"}, {"id": "PROP-RLS-FAIL-CLOSED-003", "layer": "property", "selector": "tests/unit/test_schema_security_manifest.py::test_tenant_owned_manifest_entry_preserves_every_security_property"}, {"id": "PG-RLS-FAIL-CLOSED-003", "layer": "postgres", "selector": "tests/integration/test_organization_isolation.py::test_missing_tenant_context_is_fail_closed_for_every_operation"}, @@ -108,13 +109,13 @@ {"id": "FIXTURE-ACCEPT-012", "layer": "runtime", "selector": "tests/unit/test_ticket_audience_separation.py::test_accept_012_context_read_ticket_cannot_create_an_action_effect"} ], "invariantMappings": [ - {"invariantRef": "TENANT-OWNERSHIP-001", "evidenceRefs": {"property": ["PROP-TENANT-OWNERSHIP-001"], "postgres": ["PG-FILE-SOURCE-RLS-021", "PG-RLS-ALL-TENANT-TABLES", "PG-FILE-IMPORT-023", "PG-FILE-REPLACEMENT-026", "PG-FILE-RECOVERY-027", "PG-FILE-TOMBSTONE-028"], "runtime": ["RUNTIME-TENANT-OWNERSHIP-001"]}}, - {"invariantRef": "TENANT-FK-002", "evidenceRefs": {"property": ["PROP-TENANT-FK-002"], "postgres": ["PG-TENANT-FK-002", "PG-FILE-SOURCE-FK-021", "PG-FILE-IMPORT-023", "PG-FILE-REPLACEMENT-026", "PG-FILE-TOMBSTONE-028"], "runtime": ["RUNTIME-TENANT-FK-002"]}}, - {"invariantRef": "RLS-FAIL-CLOSED-003", "evidenceRefs": {"property": ["PROP-RLS-FAIL-CLOSED-003"], "postgres": ["PG-RLS-FAIL-CLOSED-003", "PG-RLS-ALL-TENANT-TABLES", "PG-FILE-IMPORT-023", "PG-FILE-REPLACEMENT-026", "PG-FILE-TOMBSTONE-028"], "runtime": ["RUNTIME-RLS-FAIL-CLOSED-003"]}}, + {"invariantRef": "TENANT-OWNERSHIP-001", "evidenceRefs": {"property": ["PROP-TENANT-OWNERSHIP-001"], "postgres": ["PG-FILE-SOURCE-RLS-021", "PG-RLS-ALL-TENANT-TABLES", "PG-FILE-IMPORT-023", "PG-FILE-REPLACEMENT-026", "PG-FILE-RECOVERY-027", "PG-FILE-TOMBSTONE-028", "PG-FILE-PROGRESS-029"], "runtime": ["RUNTIME-TENANT-OWNERSHIP-001"]}}, + {"invariantRef": "TENANT-FK-002", "evidenceRefs": {"property": ["PROP-TENANT-FK-002"], "postgres": ["PG-TENANT-FK-002", "PG-FILE-SOURCE-FK-021", "PG-FILE-IMPORT-023", "PG-FILE-REPLACEMENT-026", "PG-FILE-TOMBSTONE-028", "PG-FILE-PROGRESS-029"], "runtime": ["RUNTIME-TENANT-FK-002"]}}, + {"invariantRef": "RLS-FAIL-CLOSED-003", "evidenceRefs": {"property": ["PROP-RLS-FAIL-CLOSED-003"], "postgres": ["PG-RLS-FAIL-CLOSED-003", "PG-RLS-ALL-TENANT-TABLES", "PG-FILE-IMPORT-023", "PG-FILE-REPLACEMENT-026", "PG-FILE-TOMBSTONE-028", "PG-FILE-PROGRESS-029"], "runtime": ["RUNTIME-RLS-FAIL-CLOSED-003"]}}, {"invariantRef": "SCOPE-INTERSECTION-004", "evidenceRefs": {"property": ["PROP-SCOPE-INTERSECTION-004"], "postgres": ["PG-SCOPE-INTERSECTION-004", "PG-FIELD-PROJECTION-RLS-048"], "runtime": ["RUNTIME-SCOPE-INTERSECTION-004"]}}, {"invariantRef": "INDEX-NOT-AUTHORITY-005", "evidenceRefs": {"property": ["PROP-INDEX-NOT-AUTHORITY-005"], "postgres": ["PG-INDEX-NOT-AUTHORITY-005", "PG-FILE-IMPORT-023", "PG-FILE-REPLACEMENT-026", "PG-FILE-TOMBSTONE-028"], "runtime": ["RUNTIME-INDEX-NOT-AUTHORITY-005"]}}, {"invariantRef": "REVOCATION-006", "evidenceRefs": {"property": ["PROP-REVOCATION-006"], "postgres": ["PG-REVOCATION-006", "PG-FILE-TOMBSTONE-028"], "runtime": ["RUNTIME-REVOCATION-006"]}}, - {"invariantRef": "WORKER-LEASE-007", "evidenceRefs": {"property": ["PROP-WORKER-LEASE-007"], "postgres": ["PG-WORKER-LEASE-007", "PG-FILE-IMPORT-023", "PG-FILE-REPLACEMENT-026", "PG-FILE-RECOVERY-027"], "runtime": ["RUNTIME-WORKER-LEASE-007"]}}, + {"invariantRef": "WORKER-LEASE-007", "evidenceRefs": {"property": ["PROP-WORKER-LEASE-007"], "postgres": ["PG-WORKER-LEASE-007", "PG-FILE-IMPORT-023", "PG-FILE-REPLACEMENT-026", "PG-FILE-RECOVERY-027", "PG-FILE-PROGRESS-029"], "runtime": ["RUNTIME-WORKER-LEASE-007"]}}, {"invariantRef": "TRANSPORT-UNTRUSTED-008", "evidenceRefs": {"property": ["PROP-TRANSPORT-UNTRUSTED-008"], "postgres": ["PG-TRANSPORT-UNTRUSTED-008"], "runtime": ["RUNTIME-TRANSPORT-UNTRUSTED-008"]}}, {"invariantRef": "NON-ENUMERATION-009", "evidenceRefs": {"property": ["PROP-NON-ENUMERATION-009"], "postgres": ["PG-NON-ENUMERATION-009"], "runtime": ["RUNTIME-NON-ENUMERATION-009"]}}, {"invariantRef": "CITATION-AUTH-010", "evidenceRefs": {"property": ["PROP-CITATION-AUTH-010"], "postgres": ["PG-CITATION-AUTH-010"], "runtime": ["RUNTIME-CITATION-AUTH-010"]}}, diff --git a/migrations/versions/20260723_0017_file_source_progress.py b/migrations/versions/20260723_0017_file_source_progress.py new file mode 100644 index 00000000..ebce8dc4 --- /dev/null +++ b/migrations/versions/20260723_0017_file_source_progress.py @@ -0,0 +1,833 @@ +"""Separate File acquisition checkpoints from publish watermarks. + +Revision ID: 20260723_0017 +Revises: 20260723_0016 +Create Date: 2026-07-23 +""" + +# ruff: noqa: E501 + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +revision: str = "20260723_0017" +down_revision: str | None = "20260723_0016" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +_MIGRATOR = "context_engine_migrator" +_CONTROL = "context_engine_control" +_RUNTIME = "context_engine_runtime" +_WORKER = "context_engine_worker" +_DEFINER = "context_engine_worker_lease_definer" +_CHECKPOINT = "file_source_acquisition_checkpoint" +_WATERMARK = "file_source_publish_watermark" +_READ_FUNCTION = "context_control_read_file_source_progress" +_READ_SIGNATURE = "(uuid, uuid)" +_MAX_BIGINT = 9223372036854775807 + + +def _tenant_append_only(table: str) -> None: + op.execute(f"REVOKE ALL ON TABLE {table} FROM PUBLIC") + for role in (_CONTROL, _RUNTIME, _WORKER): + op.execute(f"REVOKE ALL ON TABLE {table} FROM {role}") + op.execute(f"ALTER TABLE {table} ENABLE ROW LEVEL SECURITY") + op.execute(f"ALTER TABLE {table} FORCE ROW LEVEL SECURITY") + op.execute( + f"CREATE POLICY {table}_migrator_administration ON {table} " + f"FOR ALL TO {_MIGRATOR} USING (true) WITH CHECK (true)" + ) + tenant = ( + "organization_id = NULLIF(" + "current_setting('app.organization_id', true), ''" + ")::uuid" + ) + op.execute( + f"CREATE POLICY {table}_file_progress_definer_select ON {table} " + f"FOR SELECT TO {_DEFINER} USING ({tenant})" + ) + op.execute( + f"CREATE POLICY {table}_file_progress_definer_insert ON {table} " + f"FOR INSERT TO {_DEFINER} WITH CHECK ({tenant})" + ) + op.execute(f"GRANT SELECT, INSERT ON TABLE {table} TO {_DEFINER}") + op.execute( + f"CREATE TRIGGER {table}_immutable BEFORE UPDATE OR DELETE ON {table} " + "FOR EACH ROW EXECUTE FUNCTION public.context_content_reject_mutation()" + ) + + +def upgrade() -> None: + """Persist both progress signals at their exact durable boundaries.""" + + op.create_unique_constraint( + "uq_file_import_job_progress_lineage", + "file_import_job", + ["organization_id", "job_id", "acquisition_id", "source_id"], + ) + op.create_unique_constraint( + "uq_file_resource_cleanup_intent_progress_lineage", + "file_resource_cleanup_intent", + [ + "organization_id", + "cleanup_intent_id", + "source_id", + "resource_ref", + "revision_id", + ], + ) + op.create_table( + _CHECKPOINT, + sa.Column("organization_id", postgresql.UUID(as_uuid=True), nullable=False), + sa.Column("source_id", postgresql.UUID(as_uuid=True), nullable=False), + sa.Column("sequence", sa.BigInteger(), nullable=False), + sa.Column("checkpoint_ref", sa.Text(), nullable=False), + sa.Column("change_kind", sa.Text(), nullable=False), + sa.Column("acquisition_id", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("job_id", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("cleanup_intent_id", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("resource_ref", sa.Text(), nullable=True), + sa.Column("revision_id", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("event_ref", sa.Text(), nullable=True), + sa.Column("event_sequence", sa.BigInteger(), nullable=True), + sa.Column("accepted_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint( + "organization_id", + "source_id", + "sequence", + name="pk_file_source_acquisition_checkpoint", + ), + sa.UniqueConstraint( + "organization_id", + "source_id", + "sequence", + "checkpoint_ref", + "change_kind", + name="uq_file_source_acquisition_checkpoint_exact", + ), + sa.UniqueConstraint( + "organization_id", + "checkpoint_ref", + name="uq_file_source_acquisition_checkpoint_ref", + ), + sa.UniqueConstraint( + "organization_id", + "acquisition_id", + name="uq_file_source_acquisition_checkpoint_acquisition", + ), + sa.UniqueConstraint( + "organization_id", + "cleanup_intent_id", + name="uq_file_source_acquisition_checkpoint_cleanup", + ), + sa.ForeignKeyConstraint( + ["organization_id", "source_id"], + ["context_source.organization_id", "context_source.source_id"], + name="fk_file_source_acquisition_checkpoint_source_same_organization", + ), + sa.ForeignKeyConstraint( + ["organization_id", "job_id", "acquisition_id", "source_id"], + [ + "file_import_job.organization_id", + "file_import_job.job_id", + "file_import_job.acquisition_id", + "file_import_job.source_id", + ], + name="fk_file_source_acquisition_checkpoint_job_exact", + ), + sa.ForeignKeyConstraint( + [ + "organization_id", + "cleanup_intent_id", + "source_id", + "resource_ref", + "revision_id", + ], + [ + "file_resource_cleanup_intent.organization_id", + "file_resource_cleanup_intent.cleanup_intent_id", + "file_resource_cleanup_intent.source_id", + "file_resource_cleanup_intent.resource_ref", + "file_resource_cleanup_intent.revision_id", + ], + name="fk_file_source_acquisition_checkpoint_cleanup_exact", + ), + sa.CheckConstraint( + f"sequence BETWEEN 1 AND {_MAX_BIGINT}", + name="ck_file_source_acquisition_checkpoint_sequence", + ), + sa.CheckConstraint( + "checkpoint_ref ~ '^facp_[0-9a-f]{64}$'", + name="ck_file_source_acquisition_checkpoint_ref", + ), + sa.CheckConstraint( + "(change_kind = 'file_import' AND acquisition_id IS NOT NULL " + "AND job_id IS NOT NULL AND cleanup_intent_id IS NULL " + "AND resource_ref IS NULL AND revision_id IS NULL " + "AND event_ref IS NULL AND event_sequence IS NULL) OR " + "(change_kind = 'file_tombstone' AND acquisition_id IS NULL " + "AND job_id IS NULL AND cleanup_intent_id IS NOT NULL " + "AND resource_ref ~ '^resource:file:[0-9a-f]{64}$' " + "AND revision_id IS NOT NULL " + "AND event_ref ~ '^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$' " + f"AND event_sequence BETWEEN 1 AND {_MAX_BIGINT})", + name="ck_file_source_acquisition_checkpoint_lineage", + ), + ) + op.create_table( + _WATERMARK, + sa.Column("organization_id", postgresql.UUID(as_uuid=True), nullable=False), + sa.Column("source_id", postgresql.UUID(as_uuid=True), nullable=False), + sa.Column("sequence", sa.BigInteger(), nullable=False), + sa.Column("watermark_ref", sa.Text(), nullable=False), + sa.Column("checkpoint_ref", sa.Text(), nullable=False), + sa.Column("change_kind", sa.Text(), nullable=False), + sa.Column("outcome", sa.Text(), nullable=False), + sa.Column("resource_ref", sa.Text(), nullable=False), + sa.Column("revision_id", postgresql.UUID(as_uuid=True), nullable=False), + sa.Column("published_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint( + "organization_id", + "source_id", + "sequence", + name="pk_file_source_publish_watermark", + ), + sa.UniqueConstraint( + "organization_id", + "watermark_ref", + name="uq_file_source_publish_watermark_ref", + ), + sa.ForeignKeyConstraint( + [ + "organization_id", + "source_id", + "sequence", + "checkpoint_ref", + "change_kind", + ], + [ + f"{_CHECKPOINT}.organization_id", + f"{_CHECKPOINT}.source_id", + f"{_CHECKPOINT}.sequence", + f"{_CHECKPOINT}.checkpoint_ref", + f"{_CHECKPOINT}.change_kind", + ], + name="fk_file_source_publish_watermark_checkpoint_exact", + ), + sa.ForeignKeyConstraint( + ["organization_id", "resource_ref", "revision_id"], + [ + "context_revision.organization_id", + "context_revision.resource_ref", + "context_revision.revision_id", + ], + name="fk_file_source_publish_watermark_revision_same_organization", + ), + sa.CheckConstraint( + f"sequence BETWEEN 1 AND {_MAX_BIGINT}", + name="ck_file_source_publish_watermark_sequence", + ), + sa.CheckConstraint( + "watermark_ref ~ '^fpwm_[0-9a-f]{64}$'", + name="ck_file_source_publish_watermark_ref", + ), + sa.CheckConstraint( + "btrim(resource_ref) <> '' AND char_length(resource_ref) <= 512 " + "AND resource_ref !~ '[[:cntrl:]]'", + name="ck_file_source_publish_watermark_resource", + ), + sa.CheckConstraint( + "(change_kind = 'file_import' AND outcome IN " + "('published', 'replaced', 'unchanged')) OR " + "(change_kind = 'file_tombstone' AND outcome = 'tombstoned')", + name="ck_file_source_publish_watermark_outcome", + ), + ) + _tenant_append_only(_CHECKPOINT) + _tenant_append_only(_WATERMARK) + + _backfill_progress() + _create_progress_functions() + _create_progress_triggers() + + +def _backfill_progress() -> None: + op.execute( + f""" + WITH changes AS ( + SELECT acquisition.organization_id, + acquisition.source_id, + 'file_import'::text AS change_kind, + job.job_id AS carrier_id, + acquisition.created_at AS accepted_at, + acquisition.acquisition_id, + job.job_id, + NULL::uuid AS cleanup_intent_id, + NULL::text AS resource_ref, + NULL::uuid AS revision_id, + NULL::text AS event_ref, + NULL::bigint AS event_sequence + FROM public.file_acquisition AS acquisition + JOIN public.file_import_job AS job + ON job.organization_id = acquisition.organization_id + AND job.acquisition_id = acquisition.acquisition_id + AND job.source_id = acquisition.source_id + UNION ALL + SELECT intent.organization_id, + intent.source_id, + 'file_tombstone'::text, + intent.cleanup_intent_id, + intent.tombstoned_at, + NULL::uuid, + NULL::uuid, + intent.cleanup_intent_id, + intent.resource_ref, + intent.revision_id, + intent.event_ref, + intent.event_sequence + FROM public.file_resource_cleanup_intent AS intent + ), ranked AS ( + SELECT changes.*, + row_number() OVER ( + PARTITION BY organization_id, source_id + ORDER BY accepted_at, change_kind, carrier_id + ) AS sequence + FROM changes + ) + INSERT INTO public.{_CHECKPOINT} ( + organization_id, source_id, sequence, checkpoint_ref, + change_kind, acquisition_id, job_id, cleanup_intent_id, + resource_ref, revision_id, event_ref, event_sequence, accepted_at + ) + SELECT organization_id, source_id, sequence, + 'facp_' || encode(public.digest( + convert_to('context-engine.file-acquisition-checkpoint.v1', 'UTF8') + || decode('00', 'hex') || uuid_send(organization_id) + || uuid_send(source_id) || int8send(sequence) + || uuid_send(carrier_id), 'sha256' + ), 'hex'), + change_kind, acquisition_id, job_id, cleanup_intent_id, + resource_ref, revision_id, event_ref, event_sequence, accepted_at + FROM ranked + """ + ) + op.execute( + f""" + WITH completed AS ( + SELECT checkpoint.organization_id, checkpoint.source_id, + checkpoint.sequence, checkpoint.checkpoint_ref, + checkpoint.change_kind, 'unchanged'::text AS outcome, + result.resource_ref, + result.active_revision_id AS revision_id, + result.observed_at AS published_at + FROM public.{_CHECKPOINT} AS checkpoint + JOIN public.file_acquisition_result AS result + ON result.organization_id = checkpoint.organization_id + AND result.acquisition_id = checkpoint.acquisition_id + WHERE checkpoint.change_kind = 'file_import' + UNION ALL + SELECT checkpoint.organization_id, checkpoint.source_id, + checkpoint.sequence, checkpoint.checkpoint_ref, + checkpoint.change_kind, + CASE WHEN EXISTS ( + SELECT 1 + FROM public.file_revision_replacement_plan AS plan + WHERE plan.organization_id = checkpoint.organization_id + AND plan.acquisition_id = checkpoint.acquisition_id + AND plan.replacement_revision_id = snapshot.revision_id + ) THEN 'replaced'::text ELSE 'published'::text END, + snapshot.resource_ref, snapshot.revision_id, + event.recorded_at + FROM public.{_CHECKPOINT} AS checkpoint + JOIN public.file_revision_snapshot AS snapshot + ON snapshot.organization_id = checkpoint.organization_id + AND snapshot.acquisition_id = checkpoint.acquisition_id + JOIN public.revision_publication_event AS event + ON event.organization_id = snapshot.organization_id + AND event.resource_ref = snapshot.resource_ref + AND event.revision_id = snapshot.revision_id + AND event.state = 'active' + WHERE checkpoint.change_kind = 'file_import' + UNION ALL + SELECT checkpoint.organization_id, checkpoint.source_id, + checkpoint.sequence, checkpoint.checkpoint_ref, + checkpoint.change_kind, 'tombstoned'::text, + checkpoint.resource_ref, checkpoint.revision_id, + checkpoint.accepted_at + FROM public.{_CHECKPOINT} AS checkpoint + WHERE checkpoint.change_kind = 'file_tombstone' + ) + INSERT INTO public.{_WATERMARK} ( + organization_id, source_id, sequence, watermark_ref, + checkpoint_ref, change_kind, outcome, resource_ref, + revision_id, published_at + ) + SELECT organization_id, source_id, sequence, + 'fpwm_' || encode(public.digest( + convert_to('context-engine.file-publish-watermark.v1', 'UTF8') + || decode('00', 'hex') || uuid_send(organization_id) + || uuid_send(source_id) || int8send(sequence) + || convert_to(outcome, 'UTF8') || decode('00', 'hex') + || convert_to(resource_ref, 'UTF8') || uuid_send(revision_id), + 'sha256' + ), 'hex'), + checkpoint_ref, change_kind, outcome, resource_ref, + revision_id, published_at + FROM completed + ON CONFLICT (organization_id, source_id, sequence) DO NOTHING + """ + ) + + +def _create_progress_functions() -> None: + op.execute( + f""" + CREATE FUNCTION public.context_file_source_append_publish_watermark( + requested_organization_id uuid, + requested_source_id uuid, + requested_sequence bigint, + requested_change_kind text, + requested_outcome text, + requested_resource_ref text, + requested_revision_id uuid, + requested_published_at timestamptz + ) RETURNS void + LANGUAGE plpgsql + SECURITY DEFINER + SET search_path = pg_catalog, pg_temp + SET row_security = on + AS $function$ + DECLARE selected_checkpoint_ref text; + BEGIN + SELECT checkpoint.checkpoint_ref + INTO selected_checkpoint_ref + FROM public.{_CHECKPOINT} AS checkpoint + WHERE checkpoint.organization_id = requested_organization_id + AND checkpoint.source_id = requested_source_id + AND checkpoint.sequence = requested_sequence + AND checkpoint.change_kind = requested_change_kind; + IF selected_checkpoint_ref IS NULL THEN + RAISE EXCEPTION USING ERRCODE = '55000', + MESSAGE = 'File publish outcome has no accepted checkpoint'; + END IF; + INSERT INTO public.{_WATERMARK} ( + organization_id, source_id, sequence, watermark_ref, + checkpoint_ref, change_kind, outcome, resource_ref, + revision_id, published_at + ) VALUES ( + requested_organization_id, requested_source_id, + requested_sequence, + 'fpwm_' || encode(public.digest( + convert_to('context-engine.file-publish-watermark.v1', 'UTF8') + || decode('00', 'hex') + || uuid_send(requested_organization_id) + || uuid_send(requested_source_id) + || int8send(requested_sequence) + || convert_to(requested_outcome, 'UTF8') + || decode('00', 'hex') + || convert_to(requested_resource_ref, 'UTF8') + || uuid_send(requested_revision_id), 'sha256' + ), 'hex'), + selected_checkpoint_ref, requested_change_kind, + requested_outcome, requested_resource_ref, + requested_revision_id, requested_published_at + ) ON CONFLICT (organization_id, source_id, sequence) DO NOTHING; + END; + $function$ + """ + ) + op.execute( + f""" + CREATE FUNCTION public.{_READ_FUNCTION}( + requested_organization_id uuid, + requested_source_id uuid + ) RETURNS TABLE ( + acquisition_sequence bigint, + acquisition_checkpoint_ref text, + acquisition_change_kind text, + acquisition_acquisition_id uuid, + acquisition_job_id uuid, + acquisition_cleanup_intent_id uuid, + acquisition_resource_ref text, + acquisition_revision_id uuid, + acquisition_event_ref text, + acquisition_event_sequence bigint, + acquisition_accepted_at timestamptz, + publish_sequence bigint, + publish_watermark_ref text, + publish_checkpoint_ref text, + publish_change_kind text, + publish_outcome text, + publish_acquisition_id uuid, + publish_job_id uuid, + publish_cleanup_intent_id uuid, + publish_resource_ref text, + publish_revision_id uuid, + publish_event_ref text, + publish_event_sequence bigint, + publish_published_at timestamptz + ) + LANGUAGE plpgsql + SECURITY DEFINER + SET search_path = pg_catalog, pg_temp + SET row_security = on + AS $function$ + BEGIN + IF SESSION_USER <> '{_CONTROL}' + OR NULLIF(current_setting('app.organization_id', true), '')::uuid + IS DISTINCT FROM requested_organization_id + OR NOT EXISTS ( + SELECT 1 FROM public.context_source AS source + WHERE source.organization_id = requested_organization_id + AND source.source_id = requested_source_id + AND source.source_kind = 'file' + ) + THEN RETURN; END IF; + RETURN QUERY + WITH bounds AS ( + SELECT max(checkpoint.sequence) AS latest_sequence, + min(checkpoint.sequence) FILTER ( + WHERE watermark.sequence IS NULL + ) AS first_missing + FROM public.{_CHECKPOINT} AS checkpoint + LEFT JOIN public.{_WATERMARK} AS watermark + ON watermark.organization_id = checkpoint.organization_id + AND watermark.source_id = checkpoint.source_id + AND watermark.sequence = checkpoint.sequence + WHERE checkpoint.organization_id = requested_organization_id + AND checkpoint.source_id = requested_source_id + ), visible AS ( + SELECT bounds.latest_sequence, + CASE + WHEN bounds.latest_sequence IS NULL THEN NULL::bigint + WHEN bounds.first_missing IS NULL + THEN bounds.latest_sequence + ELSE bounds.first_missing - 1 + END AS visible_sequence + FROM bounds + ) + SELECT latest.sequence, latest.checkpoint_ref, + latest.change_kind, latest.acquisition_id, + latest.job_id, latest.cleanup_intent_id, + latest.resource_ref, latest.revision_id, + latest.event_ref, latest.event_sequence, + latest.accepted_at, + watermark.sequence, watermark.watermark_ref, + watermark.checkpoint_ref, watermark.change_kind, + watermark.outcome, published.acquisition_id, + published.job_id, published.cleanup_intent_id, + watermark.resource_ref, watermark.revision_id, + published.event_ref, published.event_sequence, + watermark.published_at + FROM visible + LEFT JOIN public.{_CHECKPOINT} AS latest + ON latest.organization_id = requested_organization_id + AND latest.source_id = requested_source_id + AND latest.sequence = visible.latest_sequence + LEFT JOIN public.{_WATERMARK} AS watermark + ON watermark.organization_id = requested_organization_id + AND watermark.source_id = requested_source_id + AND watermark.sequence = visible.visible_sequence + LEFT JOIN public.{_CHECKPOINT} AS published + ON published.organization_id = watermark.organization_id + AND published.source_id = watermark.source_id + AND published.sequence = watermark.sequence; + END; + $function$ + """ + ) + for function_name, signature in ( + ( + "context_file_source_append_publish_watermark", + "(uuid, uuid, bigint, text, text, text, uuid, timestamp with time zone)", + ), + (_READ_FUNCTION, _READ_SIGNATURE), + ): + op.execute( + f"REVOKE ALL ON FUNCTION public.{function_name}{signature} FROM PUBLIC" + ) + if function_name == _READ_FUNCTION: + op.execute( + f"GRANT EXECUTE ON FUNCTION public.{function_name}{signature} " + f"TO {_CONTROL}" + ) + op.execute(f"GRANT CREATE ON SCHEMA public TO {_DEFINER}") + op.execute( + f"ALTER FUNCTION public.{function_name}{signature} OWNER TO {_DEFINER}" + ) + op.execute(f"REVOKE CREATE ON SCHEMA public FROM {_DEFINER}") + + +def _create_progress_triggers() -> None: + op.execute( + f""" + CREATE FUNCTION public.context_file_source_checkpoint_import_job() + RETURNS trigger + LANGUAGE plpgsql SECURITY DEFINER + SET search_path = pg_catalog, pg_temp SET row_security = on + AS $function$ + DECLARE next_sequence bigint; + BEGIN + PERFORM pg_advisory_xact_lock(hashtextextended( + 'context-engine.file-source-progress:' + || NEW.organization_id::text || ':' || NEW.source_id::text, 0 + )); + SELECT COALESCE(max(checkpoint.sequence), 0) + 1 + INTO next_sequence + FROM public.{_CHECKPOINT} AS checkpoint + WHERE checkpoint.organization_id = NEW.organization_id + AND checkpoint.source_id = NEW.source_id; + IF next_sequence NOT BETWEEN 1 AND {_MAX_BIGINT} THEN + RAISE EXCEPTION USING ERRCODE = '22003', + MESSAGE = 'File acquisition checkpoint sequence exhausted'; + END IF; + INSERT INTO public.{_CHECKPOINT} ( + organization_id, source_id, sequence, checkpoint_ref, + change_kind, acquisition_id, job_id, accepted_at + ) + SELECT NEW.organization_id, NEW.source_id, next_sequence, + 'facp_' || encode(public.digest( + convert_to('context-engine.file-acquisition-checkpoint.v1', 'UTF8') + || decode('00', 'hex') + || uuid_send(NEW.organization_id) + || uuid_send(NEW.source_id) + || int8send(next_sequence) + || uuid_send(NEW.job_id), 'sha256' + ), 'hex'), + 'file_import', NEW.acquisition_id, NEW.job_id, + acquisition.created_at + FROM public.file_acquisition AS acquisition + WHERE acquisition.organization_id = NEW.organization_id + AND acquisition.acquisition_id = NEW.acquisition_id + AND acquisition.source_id = NEW.source_id; + IF NOT FOUND THEN + RAISE EXCEPTION USING ERRCODE = '55000', + MESSAGE = 'File import job has no accepted acquisition'; + END IF; + RETURN NULL; + END; + $function$ + """ + ) + op.execute( + f""" + CREATE FUNCTION public.context_file_source_publish_active_revision() + RETURNS trigger + LANGUAGE plpgsql SECURITY DEFINER + SET search_path = pg_catalog, pg_temp SET row_security = on + AS $function$ + DECLARE selected_source_id uuid; + selected_sequence bigint; + selected_outcome text; + BEGIN + SELECT checkpoint.source_id, checkpoint.sequence, + CASE WHEN EXISTS ( + SELECT 1 + FROM public.file_revision_replacement_plan AS plan + WHERE plan.organization_id = NEW.organization_id + AND plan.replacement_revision_id = NEW.revision_id + AND plan.resource_ref = NEW.resource_ref + ) THEN 'replaced'::text ELSE 'published'::text END + INTO selected_source_id, selected_sequence, selected_outcome + FROM public.file_revision_snapshot AS snapshot + JOIN public.{_CHECKPOINT} AS checkpoint + ON checkpoint.organization_id = snapshot.organization_id + AND checkpoint.acquisition_id = snapshot.acquisition_id + AND checkpoint.change_kind = 'file_import' + WHERE snapshot.organization_id = NEW.organization_id + AND snapshot.resource_ref = NEW.resource_ref + AND snapshot.revision_id = NEW.revision_id; + IF selected_sequence IS NULL THEN + RAISE EXCEPTION USING ERRCODE = '55000', + MESSAGE = 'Active File Revision has no accepted checkpoint'; + END IF; + PERFORM public.context_file_source_append_publish_watermark( + NEW.organization_id, selected_source_id, selected_sequence, + 'file_import', selected_outcome, NEW.resource_ref, + NEW.revision_id, NEW.recorded_at + ); + RETURN NULL; + END; + $function$ + """ + ) + op.execute( + f""" + CREATE FUNCTION public.context_file_source_publish_unchanged_acquisition() + RETURNS trigger + LANGUAGE plpgsql SECURITY DEFINER + SET search_path = pg_catalog, pg_temp SET row_security = on + AS $function$ + DECLARE selected_sequence bigint; + BEGIN + SELECT checkpoint.sequence INTO selected_sequence + FROM public.{_CHECKPOINT} AS checkpoint + WHERE checkpoint.organization_id = NEW.organization_id + AND checkpoint.source_id = NEW.source_id + AND checkpoint.acquisition_id = NEW.acquisition_id + AND checkpoint.change_kind = 'file_import'; + IF selected_sequence IS NULL THEN + RAISE EXCEPTION USING ERRCODE = '55000', + MESSAGE = 'Unchanged File acquisition has no checkpoint'; + END IF; + PERFORM public.context_file_source_append_publish_watermark( + NEW.organization_id, NEW.source_id, selected_sequence, + 'file_import', 'unchanged', NEW.resource_ref, + NEW.active_revision_id, NEW.observed_at + ); + RETURN NULL; + END; + $function$ + """ + ) + op.execute( + f""" + CREATE FUNCTION public.context_file_source_checkpoint_tombstone() + RETURNS trigger + LANGUAGE plpgsql SECURITY DEFINER + SET search_path = pg_catalog, pg_temp SET row_security = on + AS $function$ + DECLARE next_sequence bigint; + BEGIN + PERFORM pg_advisory_xact_lock(hashtextextended( + 'context-engine.file-source-progress:' + || NEW.organization_id::text || ':' || NEW.source_id::text, 0 + )); + SELECT COALESCE(max(checkpoint.sequence), 0) + 1 + INTO next_sequence + FROM public.{_CHECKPOINT} AS checkpoint + WHERE checkpoint.organization_id = NEW.organization_id + AND checkpoint.source_id = NEW.source_id; + IF next_sequence NOT BETWEEN 1 AND {_MAX_BIGINT} THEN + RAISE EXCEPTION USING ERRCODE = '22003', + MESSAGE = 'File acquisition checkpoint sequence exhausted'; + END IF; + INSERT INTO public.{_CHECKPOINT} ( + organization_id, source_id, sequence, checkpoint_ref, + change_kind, cleanup_intent_id, resource_ref, revision_id, + event_ref, event_sequence, accepted_at + ) VALUES ( + NEW.organization_id, NEW.source_id, next_sequence, + 'facp_' || encode(public.digest( + convert_to('context-engine.file-acquisition-checkpoint.v1', 'UTF8') + || decode('00', 'hex') + || uuid_send(NEW.organization_id) + || uuid_send(NEW.source_id) || int8send(next_sequence) + || uuid_send(NEW.cleanup_intent_id), 'sha256' + ), 'hex'), + 'file_tombstone', NEW.cleanup_intent_id, + NEW.resource_ref, NEW.revision_id, + NEW.event_ref, NEW.event_sequence, NEW.tombstoned_at + ); + PERFORM public.context_file_source_append_publish_watermark( + NEW.organization_id, NEW.source_id, next_sequence, + 'file_tombstone', 'tombstoned', NEW.resource_ref, + NEW.revision_id, NEW.tombstoned_at + ); + RETURN NULL; + END; + $function$ + """ + ) + trigger_functions = ( + "context_file_source_checkpoint_import_job", + "context_file_source_publish_active_revision", + "context_file_source_publish_unchanged_acquisition", + "context_file_source_checkpoint_tombstone", + ) + for function_name in trigger_functions: + op.execute(f"REVOKE ALL ON FUNCTION public.{function_name}() FROM PUBLIC") + op.execute( + "CREATE TRIGGER file_import_job_source_checkpoint " + "AFTER INSERT ON file_import_job FOR EACH ROW " + "EXECUTE FUNCTION public.context_file_source_checkpoint_import_job()" + ) + op.execute( + "CREATE TRIGGER revision_publication_event_source_watermark " + "AFTER INSERT ON revision_publication_event FOR EACH ROW " + "WHEN (NEW.state = 'active') " + "EXECUTE FUNCTION public.context_file_source_publish_active_revision()" + ) + op.execute( + "CREATE TRIGGER file_acquisition_result_source_watermark " + "AFTER INSERT ON file_acquisition_result FOR EACH ROW " + "EXECUTE FUNCTION public.context_file_source_publish_unchanged_acquisition()" + ) + op.execute( + "CREATE TRIGGER file_resource_cleanup_intent_source_progress " + "AFTER INSERT ON file_resource_cleanup_intent FOR EACH ROW " + "EXECUTE FUNCTION public.context_file_source_checkpoint_tombstone()" + ) + for function_name in trigger_functions: + op.execute(f"GRANT CREATE ON SCHEMA public TO {_DEFINER}") + op.execute(f"ALTER FUNCTION public.{function_name}() OWNER TO {_DEFINER}") + op.execute(f"REVOKE CREATE ON SCHEMA public FROM {_DEFINER}") + + +def downgrade() -> None: + """Remove only unused progress state; never renumber accepted changes.""" + + op.execute( + "LOCK TABLE file_source_publish_watermark, " + "file_source_acquisition_checkpoint, file_import_job, " + "file_resource_cleanup_intent, revision_publication_event, " + "file_acquisition_result IN ACCESS EXCLUSIVE MODE" + ) + has_progress = bool( + op.get_bind() + .execute( + sa.text( + "SELECT EXISTS (" + "SELECT 1 FROM file_source_acquisition_checkpoint " + "UNION ALL " + "SELECT 1 FROM file_source_publish_watermark" + ")" + ) + ) + .scalar_one() + ) + if has_progress: + raise RuntimeError( + "File source progress downgrade requires empty progress streams; " + "use a forward fix to preserve monotonic ordering and opaque refs" + ) + + op.execute( + "DROP TRIGGER file_resource_cleanup_intent_source_progress ON file_resource_cleanup_intent" + ) + op.execute( + "DROP TRIGGER file_acquisition_result_source_watermark ON file_acquisition_result" + ) + op.execute( + "DROP TRIGGER revision_publication_event_source_watermark ON revision_publication_event" + ) + op.execute("DROP TRIGGER file_import_job_source_checkpoint ON file_import_job") + op.execute(f"SET LOCAL ROLE {_DEFINER}") + for function_name in ( + "context_file_source_checkpoint_tombstone", + "context_file_source_publish_unchanged_acquisition", + "context_file_source_publish_active_revision", + "context_file_source_checkpoint_import_job", + ): + op.execute(f"DROP FUNCTION public.{function_name}()") + op.execute(f"DROP FUNCTION public.{_READ_FUNCTION}{_READ_SIGNATURE}") + op.execute( + "DROP FUNCTION public.context_file_source_append_publish_watermark" + "(uuid, uuid, bigint, text, text, text, uuid, timestamp with time zone)" + ) + op.execute("RESET ROLE") + op.drop_table(_WATERMARK) + op.drop_table(_CHECKPOINT) + op.drop_constraint( + "uq_file_resource_cleanup_intent_progress_lineage", + "file_resource_cleanup_intent", + type_="unique", + ) + op.drop_constraint( + "uq_file_import_job_progress_lineage", + "file_import_job", + type_="unique", + ) diff --git a/scripts/security_gate/rls.py b/scripts/security_gate/rls.py index 60d49071..6a49dfa9 100644 --- a/scripts/security_gate/rls.py +++ b/scripts/security_gate/rls.py @@ -29,6 +29,8 @@ "file_import_job_event": "PG-FILE-RECOVERY-027", "file_publication_recovery": "PG-FILE-RECOVERY-027", "file_resource_cleanup_intent": "PG-FILE-TOMBSTONE-028", + "file_source_acquisition_checkpoint": "PG-FILE-PROGRESS-029", + "file_source_publish_watermark": "PG-FILE-PROGRESS-029", "file_resource_ingestion_guard": "PG-FILE-IMPORT-023", "file_revision_snapshot": "PG-FILE-IMPORT-023", "file_revision_replacement_plan": "PG-FILE-REPLACEMENT-026", diff --git a/tests/integration/test_file_import_tracer.py b/tests/integration/test_file_import_tracer.py index 598d746d..78dadd8b 100644 --- a/tests/integration/test_file_import_tracer.py +++ b/tests/integration/test_file_import_tracer.py @@ -88,6 +88,7 @@ canonicalize_parsed_document, ) from tests.support.context_run_operator import exact_test_context_run_operator_read +from tests.support.file_source_progress import clear_file_source_progress_projection pytestmark = pytest.mark.integration NOW = datetime.now(UTC).replace(microsecond=0) @@ -147,6 +148,7 @@ def authenticate(self, opaque_credential: str) -> VerifiedControlOperatorIdentit { ControlOperation.REGISTER_SOURCE, ControlOperation.READ_SOURCE, + ControlOperation.READ_SOURCE_PROGRESS, ControlOperation.IMPORT_FILE, ControlOperation.TOMBSTONE_FILE_RESOURCE, } @@ -1410,6 +1412,7 @@ def _assert_structural_file_import_returns_coherent_authorized_units_over_http( assert "Handbook" not in denied.text assert "red-rocket" not in denied.text + clear_file_source_progress_projection(migration_configuration) with pytest.raises( RuntimeError, match="structural Markdown downgrade requires no v2 snapshots", @@ -1420,7 +1423,7 @@ def _assert_structural_file_import_returns_coherent_authorized_units_over_http( connection.execute( text("SELECT version_num FROM alembic_version") ).scalar_one() - == "20260723_0016" + == "20260723_0017" ) diff --git a/tests/integration/test_m0_security_gate_rls.py b/tests/integration/test_m0_security_gate_rls.py index 5b03fe32..4ff32a50 100644 --- a/tests/integration/test_m0_security_gate_rls.py +++ b/tests/integration/test_m0_security_gate_rls.py @@ -30,7 +30,7 @@ def _manifest() -> dict[str, object]: def test_all_manifest_tenant_tables_pass_live_non_owner_rls_audit( guarded_runtime_engine: Engine, ) -> None: - """PG-RLS-ALL-TENANT-TABLES: the live denominator is exactly 34/34.""" + """PG-RLS-ALL-TENANT-TABLES: the live denominator is exactly 36/36.""" with guarded_runtime_engine.connect() as connection: report = audit_live_rls( @@ -41,13 +41,13 @@ def test_all_manifest_tenant_tables_pass_live_non_owner_rls_audit( assert report["passed"] is True assert report["denominator"] == { - "allTables": 37, - "tenantOwned": 34, + "allTables": 39, + "tenantOwned": 36, "global": 3, } assert report["coverage"] == { - "numerator": 34, - "denominator": 34, + "numerator": 36, + "denominator": 36, "percent": 100.0, } assert report["failures"] == [] @@ -86,9 +86,9 @@ def test_no_force_row_level_security_mutation_fails_and_rolls_back( assert mutated["passed"] is False assert mutated["coverage"] == { - "numerator": 33, - "denominator": 34, - "percent": 97.06, + "numerator": 35, + "denominator": 36, + "percent": 97.22, } tenant_tables = cast(list[dict[str, Any]], mutated["tenantTables"]) organization_record = next( @@ -111,8 +111,8 @@ def test_no_force_row_level_security_mutation_fails_and_rolls_back( ) assert restored["passed"] is True assert restored["coverage"] == { - "numerator": 34, - "denominator": 34, + "numerator": 36, + "denominator": 36, "percent": 100.0, } diff --git a/tests/integration/test_migrations.py b/tests/integration/test_migrations.py index 179c690d..573d7e24 100644 --- a/tests/integration/test_migrations.py +++ b/tests/integration/test_migrations.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +from collections.abc import Iterator from concurrent.futures import ThreadPoolExecutor from dataclasses import replace from datetime import UTC, datetime @@ -37,6 +38,7 @@ ) from tests.integration.test_zz_file_resource_tombstone import _tombstone from tests.integration.test_zz_file_revision_replacement import NEW_MARKDOWN +from tests.support.file_source_progress import clear_file_source_progress_projection pytestmark = pytest.mark.integration ROOT = Path(__file__).parents[2] @@ -62,6 +64,8 @@ "file_revision_replacement_plan", "file_revision_snapshot", "file_revision_supersession", + "file_source_acquisition_checkpoint", + "file_source_publish_watermark", "membership", "membership_resource_field_right", "organization", @@ -81,6 +85,20 @@ ] +@pytest.fixture(autouse=True) +def isolated_migration_progress_projection( + migration_configuration: DatabaseConfiguration, +) -> Iterator[None]: + """Give destructive migration compatibility checks an empty projection.""" + + clear_file_source_progress_projection(migration_configuration) + try: + yield + finally: + if _revision_rows(migration_configuration) == ["20260723_0017"]: + clear_file_source_progress_projection(migration_configuration) + + def _delete_issue_27_upgrade_fixture( configuration: DatabaseConfiguration, organization_id: UUID, @@ -89,6 +107,14 @@ def _delete_issue_27_upgrade_fixture( engine = create_database_engine(configuration) immutable_tables = ( + ( + "file_source_publish_watermark", + "file_source_publish_watermark_immutable", + ), + ( + "file_source_acquisition_checkpoint", + "file_source_acquisition_checkpoint_immutable", + ), ("file_resource_cleanup_intent", "file_resource_cleanup_intent_immutable"), ("file_import_job_event", "file_import_job_event_immutable"), ("file_revision_supersession", "file_revision_supersession_immutable"), @@ -112,11 +138,14 @@ def _delete_issue_27_upgrade_fixture( try: with engine.begin() as connection: for table in ( + "file_source_publish_watermark", + "file_source_acquisition_checkpoint", "file_resource_cleanup_intent", "file_import_job_event", "file_publication_recovery", "file_revision_supersession", "file_revision_replacement_plan", + "file_acquisition_result", "exact_phrase_candidate", "revision_publication_event", "membership_resource_field_right", @@ -125,7 +154,6 @@ def _delete_issue_27_upgrade_fixture( "file_revision_snapshot", "context_revision", "context_resource", - "file_acquisition_result", "file_resource_ingestion_guard", "file_import_job", "file_acquisition", @@ -205,7 +233,7 @@ def test_empty_baseline_remains_a_reversible_historical_revision( assert _application_tables(migration_configuration) == ["alembic_version"] finally: command.upgrade(alembic_configuration, "head") - assert _revision_rows(migration_configuration) == ["20260723_0016"] + assert _revision_rows(migration_configuration) == ["20260723_0017"] def test_organization_isolation_revision_downgrades_and_reapplies_cleanly( @@ -220,7 +248,7 @@ def test_organization_isolation_revision_downgrades_and_reapplies_cleanly( finally: command.upgrade(alembic_configuration, "head") - assert _revision_rows(migration_configuration) == ["20260723_0016"] + assert _revision_rows(migration_configuration) == ["20260723_0017"] assert _application_tables(migration_configuration) == HEAD_TABLES @@ -240,7 +268,7 @@ def test_membership_revision_downgrades_to_issue_8_and_reapplies_cleanly( finally: command.upgrade(alembic_configuration, "head") - assert _revision_rows(migration_configuration) == ["20260723_0016"] + assert _revision_rows(migration_configuration) == ["20260723_0017"] def test_content_schema_revision_downgrades_to_membership_and_reapplies_cleanly( @@ -261,7 +289,7 @@ def test_content_schema_revision_downgrades_to_membership_and_reapplies_cleanly( finally: command.upgrade(alembic_configuration, "head") - assert _revision_rows(migration_configuration) == ["20260723_0016"] + assert _revision_rows(migration_configuration) == ["20260723_0017"] assert _application_tables(migration_configuration) == HEAD_TABLES @@ -288,7 +316,7 @@ def test_policy_epoch_revision_downgrades_to_content_and_reapplies_cleanly( finally: command.upgrade(alembic_configuration, "head") - assert _revision_rows(migration_configuration) == ["20260723_0016"] + assert _revision_rows(migration_configuration) == ["20260723_0017"] assert _application_tables(migration_configuration) == HEAD_TABLES @@ -317,7 +345,7 @@ def test_worker_lease_revision_downgrades_to_policy_epoch_and_reapplies_cleanly( finally: command.upgrade(alembic_configuration, "head") - assert _revision_rows(migration_configuration) == ["20260723_0016"] + assert _revision_rows(migration_configuration) == ["20260723_0017"] assert _application_tables(migration_configuration) == HEAD_TABLES @@ -339,7 +367,7 @@ def test_decision_lineage_revision_downgrades_to_worker_lease_and_reapplies_clea finally: command.upgrade(alembic_configuration, "head") - assert _revision_rows(migration_configuration) == ["20260723_0016"] + assert _revision_rows(migration_configuration) == ["20260723_0017"] assert _application_tables(migration_configuration) == HEAD_TABLES @@ -377,7 +405,7 @@ def test_field_projection_revision_downgrades_to_decision_lineage_and_reapplies_ finally: command.upgrade(alembic_configuration, "head") - assert _revision_rows(migration_configuration) == ["20260723_0016"] + assert _revision_rows(migration_configuration) == ["20260723_0017"] assert "context_fragment_field" in _application_tables(migration_configuration) assert "membership_resource_field_right" in _application_tables( migration_configuration @@ -400,7 +428,7 @@ def test_file_source_revision_downgrades_to_learning_release_and_reapplies_clean finally: command.upgrade(alembic_configuration, "head") - assert _revision_rows(migration_configuration) == ["20260723_0016"] + assert _revision_rows(migration_configuration) == ["20260723_0017"] tables = _application_tables(migration_configuration) assert "context_source" in tables assert "source_version" in tables @@ -451,7 +479,7 @@ def test_structural_markdown_revision_downgrades_and_reapplies_cleanly( command.upgrade(alembic_configuration, "head") engine.dispose() - assert _revision_rows(migration_configuration) == ["20260723_0016"] + assert _revision_rows(migration_configuration) == ["20260723_0017"] engine = create_database_engine(migration_configuration) try: with engine.connect() as connection: @@ -504,7 +532,7 @@ def test_file_noop_revision_downgrades_and_reapplies_cleanly( finally: command.upgrade(alembic_configuration, "head") - assert _revision_rows(migration_configuration) == ["20260723_0016"] + assert _revision_rows(migration_configuration) == ["20260723_0017"] tables = _application_tables(migration_configuration) assert "file_acquisition_result" in tables assert "file_resource_ingestion_guard" in tables @@ -525,7 +553,7 @@ def test_file_replacement_revision_downgrades_and_reapplies_cleanly( finally: command.upgrade(alembic_configuration, "head") - assert _revision_rows(migration_configuration) == ["20260723_0016"] + assert _revision_rows(migration_configuration) == ["20260723_0017"] tables = _application_tables(migration_configuration) assert "file_revision_replacement_plan" in tables assert "file_revision_supersession" in tables @@ -546,7 +574,7 @@ def test_file_recovery_revision_downgrades_and_reapplies_cleanly( finally: command.upgrade(alembic_configuration, "head") - assert _revision_rows(migration_configuration) == ["20260723_0016"] + assert _revision_rows(migration_configuration) == ["20260723_0017"] tables = _application_tables(migration_configuration) assert "file_publication_recovery" in tables assert "file_import_job_event" in tables @@ -567,12 +595,148 @@ def test_file_tombstone_revision_downgrades_and_reapplies_cleanly( finally: command.upgrade(alembic_configuration, "head") - assert _revision_rows(migration_configuration) == ["20260723_0016"] + assert _revision_rows(migration_configuration) == ["20260723_0017"] assert "file_resource_cleanup_intent" in _application_tables( migration_configuration ) +def test_file_progress_revision_downgrades_and_reapplies_cleanly( + migration_configuration: DatabaseConfiguration, +) -> None: + """Issue #29 progress is a deterministic projection of retained lineage.""" + + alembic_configuration = Config(ROOT / "alembic.ini") + try: + command.downgrade(alembic_configuration, "20260723_0016") + tables = _application_tables(migration_configuration) + assert _revision_rows(migration_configuration) == ["20260723_0016"] + assert "file_source_acquisition_checkpoint" not in tables + assert "file_source_publish_watermark" not in tables + finally: + command.upgrade(alembic_configuration, "head") + + assert _revision_rows(migration_configuration) == ["20260723_0017"] + assert "file_source_acquisition_checkpoint" in _application_tables( + migration_configuration + ) + assert "file_source_publish_watermark" in _application_tables( + migration_configuration + ) + + +def test_file_progress_refuses_downgrade_and_preserves_refs( + tmp_path: Path, + migration_configuration: DatabaseConfiguration, + guarded_control_engine: Engine, + guarded_worker_engine: Engine, +) -> None: + """Issue #29 never discards the database ordering behind opaque refs.""" + + scenario = _prepare_file_import_scenario( + tmp_path, + migration_configuration, + guarded_control_engine, + ) + assert scenario.token is not None + _run_file_import( + scenario, + scenario.prepared, + scenario.token, + guarded_worker_engine, + ) + repeated, repeated_token = _prepare_repeat_file_import( + scenario, + guarded_control_engine, + idempotency_key="progress-ref-rebuild-repeat", + ) + repeated_result = _run_file_import( + scenario, + repeated, + repeated_token, + guarded_worker_engine, + ) + _tombstone( + scenario, + guarded_control_engine, + resource_ref=repeated_result.candidate_ref.resource_ref, + event_ref="progress-ref-rebuild-delete", + event_sequence=1, + ) + migration_engine = create_database_engine(migration_configuration) + + def progress_refs() -> tuple[ + tuple[tuple[int, str, str], ...], + tuple[tuple[int, str, str], ...], + ]: + with migration_engine.connect() as connection: + checkpoint_rows = connection.execute( + text( + """ + SELECT sequence, checkpoint_ref, change_kind + FROM file_source_acquisition_checkpoint + WHERE organization_id = :organization_id + AND source_id = :source_id + ORDER BY sequence + """ + ), + { + "organization_id": scenario.organization_id, + "source_id": scenario.source_ref.value, + }, + ).all() + watermark_rows = connection.execute( + text( + """ + SELECT sequence, watermark_ref, outcome + FROM file_source_publish_watermark + WHERE organization_id = :organization_id + AND source_id = :source_id + ORDER BY sequence + """ + ), + { + "organization_id": scenario.organization_id, + "source_id": scenario.source_ref.value, + }, + ).all() + return ( + tuple( + (int(row.sequence), str(row.checkpoint_ref), str(row.change_kind)) + for row in checkpoint_rows + ), + tuple( + (int(row.sequence), str(row.watermark_ref), str(row.outcome)) + for row in watermark_rows + ), + ) + + before = progress_refs() + assert [row[0] for row in before[0]] == [1, 2, 3] + assert [row[2] for row in before[0]] == [ + "file_import", + "file_import", + "file_tombstone", + ] + assert [row[0] for row in before[1]] == [1, 2, 3] + alembic_configuration = Config(ROOT / "alembic.ini") + try: + with pytest.raises( + RuntimeError, + match="requires empty progress streams", + ): + command.downgrade(alembic_configuration, "20260723_0016") + assert _revision_rows(migration_configuration) == ["20260723_0017"] + assert progress_refs() == before + finally: + if _revision_rows(migration_configuration) != ["20260723_0017"]: + command.upgrade(alembic_configuration, "head") + migration_engine.dispose() + _delete_issue_27_upgrade_fixture( + migration_configuration, scenario.organization_id + ) + + def test_file_tombstone_revision_refuses_downgrade_with_committed_intent( tmp_path: Path, migration_configuration: DatabaseConfiguration, @@ -601,11 +765,12 @@ def test_file_tombstone_revision_refuses_downgrade_with_committed_intent( event_sequence=1, ) + clear_file_source_progress_projection(migration_configuration) alembic_configuration = Config(ROOT / "alembic.ini") with pytest.raises(SQLAlchemyError): command.downgrade(alembic_configuration, "20260723_0015") - assert _revision_rows(migration_configuration) == ["20260723_0016"] + assert _revision_rows(migration_configuration) == ["20260723_0017"] migration_engine = create_database_engine(migration_configuration) try: with migration_engine.connect() as connection: @@ -682,6 +847,7 @@ def test_recovery_upgrade_adopts_an_existing_ready_replacement( resource_ref = initial.candidate_ref.resource_ref alembic_configuration = Config(ROOT / "alembic.ini") try: + clear_file_source_progress_projection(migration_configuration) command.downgrade(alembic_configuration, "20260723_0014") with guarded_worker_engine.begin() as connection: redeemed = connection.execute( @@ -748,17 +914,20 @@ def test_recovery_upgrade_adopts_an_existing_ready_replacement( migration_engine = create_database_engine(migration_configuration) try: with migration_engine.connect() as connection: - assert connection.execute( - text( - "SELECT checkpoint FROM file_publication_recovery " - "WHERE organization_id = :organization_id " - "AND job_id = :job_id" - ), - { - "organization_id": claims.organization_id, - "job_id": claims.job_id, - }, - ).scalar_one() == "ready" + assert ( + connection.execute( + text( + "SELECT checkpoint FROM file_publication_recovery " + "WHERE organization_id = :organization_id " + "AND job_id = :job_id" + ), + { + "organization_id": claims.organization_id, + "job_id": claims.job_id, + }, + ).scalar_one() + == "ready" + ) finally: migration_engine.dispose() while datetime.now(UTC) <= claims.expires_at: @@ -1059,7 +1228,7 @@ def test_field_projection_downgrade_refuses_populated_content_atomically( ): command.downgrade(alembic_configuration, "20260722_0007") - assert _revision_rows(migration_configuration) == ["20260723_0016"] + assert _revision_rows(migration_configuration) == ["20260723_0017"] with engine.connect() as connection: assert ( connection.execute( @@ -1125,7 +1294,7 @@ def test_field_projection_downgrade_refuses_populated_content_atomically( ): connection.execute(text(statement), parameters) except SQLAlchemyError: - if _revision_rows(migration_configuration) != ["20260723_0016"]: + if _revision_rows(migration_configuration) != ["20260723_0017"]: command.upgrade(alembic_configuration, "head") raise finally: @@ -1279,7 +1448,7 @@ def test_field_projection_downgrade_serializes_with_concurrent_fragment_insert( == "concurrent-private-body" ) finally: - if _revision_rows(migration_configuration) != ["20260723_0016"]: + if _revision_rows(migration_configuration) != ["20260723_0017"]: command.upgrade(alembic_configuration, "head") with engine.begin() as connection: connection.execute( diff --git a/tests/integration/test_postgres_harness.py b/tests/integration/test_postgres_harness.py index c8e4e9f0..513a8676 100644 --- a/tests/integration/test_postgres_harness.py +++ b/tests/integration/test_postgres_harness.py @@ -33,6 +33,7 @@ RoleProvisioningContract, provision_security_roles, ) +from tests.support.file_source_progress import clear_file_source_progress_projection pytestmark = pytest.mark.integration ROOT = Path(__file__).parents[2] @@ -131,6 +132,7 @@ def test_all_login_roles_have_reviewed_capabilities( def test_post_init_role_provisioning_repairs_a_legacy_volume_idempotently( + migration_configuration: DatabaseConfiguration, guarded_control_engine: Engine, guarded_learning_engine: Engine, guarded_operator_engine: Engine, @@ -159,6 +161,7 @@ def test_post_init_role_provisioning_repairs_a_legacy_volume_idempotently( guarded_control_engine.dispose() guarded_learning_engine.dispose() guarded_operator_engine.dispose() + clear_file_source_progress_projection(migration_configuration) command.downgrade(alembic_configuration, "20260721_0004") with psycopg.connect( host="127.0.0.1", diff --git a/tests/integration/test_zz_file_content_noop.py b/tests/integration/test_zz_file_content_noop.py index b60cd2e5..a928c04f 100644 --- a/tests/integration/test_zz_file_content_noop.py +++ b/tests/integration/test_zz_file_content_noop.py @@ -38,6 +38,7 @@ _run_file_import, _RuntimeAuthenticator, ) +from tests.support.file_source_progress import clear_file_source_progress_projection pytestmark = pytest.mark.integration @@ -245,6 +246,7 @@ def test_repeated_canonically_identical_file_import_is_an_auditable_noop( assert job.effect_count == 0 assert job.revision_id == UUID(first.candidate_ref.revision_ref) + clear_file_source_progress_projection(migration_configuration) with pytest.raises( RuntimeError, match="File (?:recovery|replacement|no-op) downgrade requires", @@ -255,7 +257,7 @@ def test_repeated_canonically_identical_file_import_is_an_auditable_noop( connection.execute( text("SELECT version_num FROM alembic_version") ).scalar_one() - == "20260723_0016" + == "20260723_0017" ) diff --git a/tests/integration/test_zz_file_source_progress.py b/tests/integration/test_zz_file_source_progress.py new file mode 100644 index 00000000..4286ef17 --- /dev/null +++ b/tests/integration/test_zz_file_source_progress.py @@ -0,0 +1,484 @@ +from __future__ import annotations + +from dataclasses import replace +from datetime import timedelta +from pathlib import Path +from typing import cast +from uuid import UUID + +import pytest +from sqlalchemy import Engine, text +from sqlalchemy.exc import SQLAlchemyError + +from engine.control import ( + ContextControl, + ControlOperation, + ControlOperatorAuthority, + FileSourceChangeKind, + FileSourceProgress, + FileSourcePublishOutcome, + SourceNotAvailable, +) +from engine.persistence import ( + DatabaseConfiguration, + FileImportInterrupted, + FilePublicationBoundary, + PostgreSQLControlStore, + PostgreSQLWorkerLeaseIssuer, + create_database_engine, +) +from engine.runtime import QueryDigestKeyring +from engine.supply import FileImportPath +from tests.integration.test_file_import_tracer import ( + NOW, + _ControlAuthenticator, + _FileImportScenario, + _prepare_file_import_scenario, + _prepare_repeat_file_import, + _run_file_import, +) +from tests.integration.test_zz_file_publication_recovery import ( + _run, + _wait_for_expiry, + _worker, +) +from tests.integration.test_zz_file_resource_tombstone import _tombstone +from tests.integration.test_zz_file_revision_replacement import ( + NEW_MARKDOWN, + OLD_MARKDOWN, + _resolve, + _scenario_user_id, +) + +pytestmark = pytest.mark.integration + + +def _read_progress( + scenario: _FileImportScenario, + guarded_control_engine: Engine, + *, + request_id: str, +) -> FileSourceProgress: + authority = ControlOperatorAuthority( + _ControlAuthenticator(scenario.organization_id), + call_ttl=timedelta(minutes=5), + clock=lambda: NOW, + ) + control = ContextControl( + store=PostgreSQLControlStore(guarded_control_engine, clock=lambda: NOW), + authority=authority, + clock=lambda: NOW, + ) + with authority.authorize( + opaque_credential="control-secret", + operation=ControlOperation.READ_SOURCE_PROGRESS, + request_id=request_id, + ) as call: + return control.read_file_source_progress(call, scenario.source_ref) + + +def _visibility_state( + configuration: DatabaseConfiguration, + scenario: _FileImportScenario, + *, + resource_ref: str, + job_id: UUID, +) -> tuple[UUID, bool, str]: + database_engine = create_database_engine(configuration) + try: + with database_engine.connect() as connection: + row = connection.execute( + text( + """ + SELECT resource.active_revision_id, resource.tombstoned, + job.state + FROM context_resource AS resource + JOIN file_import_job AS job + ON job.organization_id = resource.organization_id + WHERE resource.organization_id = :organization_id + AND resource.resource_ref = :resource_ref + AND job.job_id = :job_id + """ + ), + { + "organization_id": scenario.organization_id, + "resource_ref": resource_ref, + "job_id": job_id, + }, + ).one() + return cast(tuple[UUID, bool, str], tuple(row)) + finally: + database_engine.dispose() + + +@pytest.mark.parametrize( + ("boundary", "interrupted_state"), + [ + (FilePublicationBoundary.ACQUIRED, "running"), + (FilePublicationBoundary.PREPARED, "prepared"), + (FilePublicationBoundary.INDEXED, "ready"), + ], +) +def test_each_publication_barrier_keeps_watermark_and_runtime_on_old_visibility( + tmp_path: Path, + migration_configuration: DatabaseConfiguration, + guarded_control_engine: Engine, + guarded_worker_engine: Engine, + guarded_runtime_engine: Engine, + query_digest_keyring: QueryDigestKeyring, + boundary: FilePublicationBoundary, + interrupted_state: str, +) -> None: + scenario = _prepare_file_import_scenario( + tmp_path, + migration_configuration, + guarded_control_engine, + payload=OLD_MARKDOWN, + lease_ttl_seconds=2, + ) + assert scenario.token is not None + initial = _run_file_import( + scenario, + scenario.prepared, + scenario.token, + guarded_worker_engine, + config_version="markdown-config-v2", + ) + initial_revision = UUID(initial.candidate_ref.revision_ref) + user_id = _scenario_user_id(scenario, migration_configuration) + + (scenario.root / "handbook.md").write_bytes(NEW_MARKDOWN) + changed, token = _prepare_repeat_file_import( + scenario, + guarded_control_engine, + idempotency_key=f"progress-barrier-{boundary.value}", + lease_ttl_seconds=2, + ) + interrupted_scenario = replace(scenario, prepared=changed, token=token) + with pytest.raises(FileImportInterrupted): + _run( + _worker( + interrupted_scenario, + guarded_worker_engine, + config_version="markdown-config-v2", + interrupt_after=boundary, + ), + interrupted_scenario, + token, + ) + + paused = _read_progress( + scenario, + guarded_control_engine, + request_id=f"progress-barrier-paused-{boundary.value}", + ) + assert paused.acquisition_checkpoint is not None + assert paused.publish_watermark is not None + assert paused.acquisition_checkpoint.sequence == 2 + assert paused.acquisition_checkpoint.job_ref == changed.job_id + assert paused.publish_watermark.sequence == 1 + assert _visibility_state( + migration_configuration, + scenario, + resource_ref=initial.candidate_ref.resource_ref, + job_id=changed.job_id, + ) == (initial_revision, False, interrupted_state) + old_package = _resolve( + scenario, + guarded_runtime_engine, + query_digest_keyring, + user_id=user_id, + query="OLD marker.", + request_id=f"progress-barrier-old-{boundary.value}", + ) + hidden_new = _resolve( + scenario, + guarded_runtime_engine, + query_digest_keyring, + user_id=user_id, + query="NEW marker.", + request_id=f"progress-barrier-new-hidden-{boundary.value}", + ) + assert [block["text"] for block in old_package["blocks"]] == [ + "# Handbook\n\nOLD marker." + ] + assert hidden_new["blocks"] == [] + assert hidden_new["evidence"] == [] + + _wait_for_expiry(migration_configuration) + recovered_token = PostgreSQLWorkerLeaseIssuer( + guarded_control_engine, scenario.codec + ).issue_file_import_lease(changed) + recovered = _run_file_import( + scenario, + changed, + recovered_token, + guarded_worker_engine, + config_version="markdown-config-v2", + ) + recovered_revision = UUID(recovered.candidate_ref.revision_ref) + caught_up = _read_progress( + scenario, + guarded_control_engine, + request_id=f"progress-barrier-recovered-{boundary.value}", + ) + assert caught_up.acquisition_checkpoint is not None + assert caught_up.publish_watermark is not None + assert caught_up.acquisition_checkpoint.sequence == 2 + assert caught_up.publish_watermark.sequence == 2 + assert _visibility_state( + migration_configuration, + scenario, + resource_ref=initial.candidate_ref.resource_ref, + job_id=changed.job_id, + ) == (recovered_revision, False, "completed") + visible_new = _resolve( + scenario, + guarded_runtime_engine, + query_digest_keyring, + user_id=user_id, + query="NEW marker.", + request_id=f"progress-barrier-new-visible-{boundary.value}", + ) + assert [block["text"] for block in visible_new["blocks"]] == [ + "# Handbook\n\nNEW marker." + ] + + _tombstone( + scenario, + guarded_control_engine, + resource_ref=initial.candidate_ref.resource_ref, + event_ref=f"progress-barrier-delete-{boundary.value}", + event_sequence=3, + ) + tombstoned = _read_progress( + scenario, + guarded_control_engine, + request_id=f"progress-barrier-tombstoned-{boundary.value}", + ) + assert tombstoned.acquisition_checkpoint is not None + assert tombstoned.publish_watermark is not None + assert tombstoned.acquisition_checkpoint.sequence == 3 + assert tombstoned.publish_watermark.sequence == 3 + assert _visibility_state( + migration_configuration, + scenario, + resource_ref=initial.candidate_ref.resource_ref, + job_id=changed.job_id, + ) == (recovered_revision, True, "completed") + hidden_deleted = _resolve( + scenario, + guarded_runtime_engine, + query_digest_keyring, + user_id=user_id, + query="NEW marker.", + request_id=f"progress-barrier-deleted-hidden-{boundary.value}", + ) + assert hidden_deleted["blocks"] == [] + assert hidden_deleted["evidence"] == [] + + +def test_checkpoint_can_lead_watermark_and_recovery_closes_only_contiguous_gap( + tmp_path: Path, + migration_configuration: DatabaseConfiguration, + guarded_control_engine: Engine, + guarded_worker_engine: Engine, +) -> None: + scenario = _prepare_file_import_scenario( + tmp_path, + migration_configuration, + guarded_control_engine, + lease_ttl_seconds=2, + ) + assert scenario.token is not None + + accepted = _read_progress( + scenario, guarded_control_engine, request_id="progress-initial-accepted" + ) + assert accepted.acquisition_checkpoint is not None + assert accepted.acquisition_checkpoint.sequence == 1 + assert accepted.publish_watermark is None + + initial = _run_file_import( + scenario, + scenario.prepared, + scenario.token, + guarded_worker_engine, + ) + first_visible = _read_progress( + scenario, guarded_control_engine, request_id="progress-initial-visible" + ) + assert first_visible.acquisition_checkpoint is not None + assert first_visible.publish_watermark is not None + assert first_visible.acquisition_checkpoint.sequence == 1 + assert first_visible.publish_watermark.sequence == 1 + assert first_visible.publish_watermark.outcome is FileSourcePublishOutcome.PUBLISHED + + (scenario.root / "handbook.md").write_bytes(NEW_MARKDOWN) + changed, changed_token = _prepare_repeat_file_import( + scenario, + guarded_control_engine, + idempotency_key="progress-changed-paused", + lease_ttl_seconds=2, + ) + changed_scenario = replace(scenario, prepared=changed, token=changed_token) + with pytest.raises(FileImportInterrupted): + _run( + _worker( + changed_scenario, + guarded_worker_engine, + config_version="markdown-config-v2", + interrupt_after=FilePublicationBoundary.INDEXED, + ), + changed_scenario, + changed_token, + ) + + paused = _read_progress( + scenario, guarded_control_engine, request_id="progress-changed-paused-read" + ) + assert paused.acquisition_checkpoint is not None + assert paused.publish_watermark is not None + assert paused.acquisition_checkpoint.sequence == 2 + assert paused.acquisition_checkpoint.job_ref == changed.job_id + assert paused.publish_watermark.sequence == 1 + + (scenario.root / "appendix.md").write_bytes(b"# Appendix\n\nLater change.\n") + later, later_token = _prepare_repeat_file_import( + scenario, + guarded_control_engine, + idempotency_key="progress-later-first", + path=FileImportPath("appendix.md"), + ) + later_published = _run_file_import( + scenario, later, later_token, guarded_worker_engine + ) + out_of_order = _read_progress( + scenario, guarded_control_engine, request_id="progress-out-of-order" + ) + assert out_of_order.acquisition_checkpoint is not None + assert out_of_order.publish_watermark is not None + assert out_of_order.acquisition_checkpoint.sequence == 3 + assert out_of_order.acquisition_checkpoint.job_ref == later.job_id + assert out_of_order.publish_watermark.sequence == 1 + + _wait_for_expiry(migration_configuration) + recovered_token = PostgreSQLWorkerLeaseIssuer( + guarded_control_engine, scenario.codec + ).issue_file_import_lease(changed) + recovered = _run_file_import( + scenario, + changed, + recovered_token, + guarded_worker_engine, + config_version="markdown-config-v2", + ) + assert recovered.outcome == "replaced" + caught_up = _read_progress( + scenario, guarded_control_engine, request_id="progress-recovered" + ) + assert caught_up.acquisition_checkpoint is not None + assert caught_up.publish_watermark is not None + assert caught_up.acquisition_checkpoint.sequence == 3 + assert caught_up.publish_watermark.sequence == 3 + assert caught_up.publish_watermark.resource_ref == ( + later_published.candidate_ref.resource_ref + ) + + deleted = _tombstone( + scenario, + guarded_control_engine, + resource_ref=initial.candidate_ref.resource_ref, + event_ref="progress-delete-4", + event_sequence=4, + ) + tombstoned = _read_progress( + scenario, guarded_control_engine, request_id="progress-tombstoned" + ) + assert tombstoned.acquisition_checkpoint is not None + assert tombstoned.publish_watermark is not None + assert tombstoned.acquisition_checkpoint.sequence == 4 + assert tombstoned.publish_watermark.sequence == 4 + assert ( + tombstoned.acquisition_checkpoint.change_kind + is FileSourceChangeKind.FILE_TOMBSTONE + ) + assert tombstoned.publish_watermark.outcome is FileSourcePublishOutcome.TOMBSTONED + assert tombstoned.publish_watermark.cleanup_intent_ref == ( + deleted.cleanup_intent_ref + ) + + replay = _tombstone( + scenario, + guarded_control_engine, + resource_ref=initial.candidate_ref.resource_ref, + event_ref="progress-delete-4", + event_sequence=4, + ) + assert replay == deleted + replayed = _read_progress( + scenario, guarded_control_engine, request_id="progress-replayed-delete" + ) + assert replayed.acquisition_checkpoint is not None + assert replayed.publish_watermark is not None + assert replayed.acquisition_checkpoint.sequence == 4 + assert replayed.publish_watermark.sequence == 4 + + +@pytest.mark.security_evidence(id="PG-FILE-PROGRESS-029", layer="postgres") +def test_progress_lineage_denies_direct_nonowner_access_and_cross_org_reads( + tmp_path: Path, + migration_configuration: DatabaseConfiguration, + guarded_control_engine: Engine, + guarded_runtime_engine: Engine, + guarded_worker_engine: Engine, +) -> None: + protected = _prepare_file_import_scenario( + tmp_path, migration_configuration, guarded_control_engine + ) + other = _prepare_file_import_scenario( + tmp_path, migration_configuration, guarded_control_engine + ) + authority = ControlOperatorAuthority( + _ControlAuthenticator(other.organization_id), + call_ttl=timedelta(minutes=5), + clock=lambda: NOW, + ) + control = ContextControl( + store=PostgreSQLControlStore(guarded_control_engine, clock=lambda: NOW), + authority=authority, + clock=lambda: NOW, + ) + with ( + authority.authorize( + opaque_credential="control-secret", + operation=ControlOperation.READ_SOURCE_PROGRESS, + request_id="cross-organization-progress-read", + ) as call, + pytest.raises(SourceNotAvailable), + ): + control.read_file_source_progress(call, protected.source_ref) + + statements = ( + "SELECT count(*) FROM {table}", + "INSERT INTO {table} DEFAULT VALUES", + "UPDATE {table} SET organization_id = organization_id WHERE false", + "DELETE FROM {table} WHERE false", + ) + for engine in ( + guarded_control_engine, + guarded_runtime_engine, + guarded_worker_engine, + ): + for table in ( + "file_source_acquisition_checkpoint", + "file_source_publish_watermark", + ): + for statement in statements: + with engine.connect() as connection: + transaction = connection.begin() + try: + with pytest.raises(SQLAlchemyError): + connection.execute(text(statement.format(table=table))) + finally: + transaction.rollback() diff --git a/tests/support/file_source_progress.py b/tests/support/file_source_progress.py new file mode 100644 index 00000000..725e5d0b --- /dev/null +++ b/tests/support/file_source_progress.py @@ -0,0 +1,57 @@ +from __future__ import annotations + +from sqlalchemy import text + +from engine.persistence import DatabaseConfiguration, create_database_engine + + +def clear_file_source_progress_projection( + configuration: DatabaseConfiguration, +) -> None: + """Clear only disposable Issue #29 projections for migration tests.""" + + engine = create_database_engine(configuration) + try: + with engine.begin() as connection: + tables_exist = connection.execute( + text( + """ + SELECT to_regclass( + 'public.file_source_acquisition_checkpoint' + ) IS NOT NULL + AND to_regclass( + 'public.file_source_publish_watermark' + ) IS NOT NULL + """ + ) + ).scalar_one() + if not tables_exist: + return + connection.execute( + text( + "ALTER TABLE file_source_publish_watermark " + "DISABLE TRIGGER file_source_publish_watermark_immutable" + ) + ) + connection.execute( + text( + "ALTER TABLE file_source_acquisition_checkpoint " + "DISABLE TRIGGER file_source_acquisition_checkpoint_immutable" + ) + ) + connection.execute(text("DELETE FROM file_source_publish_watermark")) + connection.execute(text("DELETE FROM file_source_acquisition_checkpoint")) + connection.execute( + text( + "ALTER TABLE file_source_acquisition_checkpoint " + "ENABLE TRIGGER file_source_acquisition_checkpoint_immutable" + ) + ) + connection.execute( + text( + "ALTER TABLE file_source_publish_watermark " + "ENABLE TRIGGER file_source_publish_watermark_immutable" + ) + ) + finally: + engine.dispose() diff --git a/tests/unit/test_context_control.py b/tests/unit/test_context_control.py index fcd0540a..e4ec4835 100644 --- a/tests/unit/test_context_control.py +++ b/tests/unit/test_context_control.py @@ -21,6 +21,7 @@ ControlStorePort, FileResourceTombstone, FileRootRef, + FileSourceProgress, RegisterFileSource, SourceControlUnavailable, SourceManifest, @@ -56,6 +57,7 @@ def authenticate(self, opaque_credential: str) -> VerifiedControlOperatorIdentit ControlOperation.IMPORT_FILE, ControlOperation.REGISTER_SOURCE, ControlOperation.READ_SOURCE, + ControlOperation.READ_SOURCE_PROGRESS, ControlOperation.TOMBSTONE_FILE_RESOURCE, } ), @@ -107,6 +109,18 @@ def prepare_file_import( service_principal_id=UUID("0f7bc78d-a76a-477c-b097-ce557b7844b9"), ) + def read_file_source_progress( + self, call: TrustedControlCall, source_ref: SourceRef + ) -> FileSourceProgress: + assert call.organization_id == ORGANIZATION_ID + assert call.operation is ControlOperation.READ_SOURCE_PROGRESS + return FileSourceProgress( + organization_id=ORGANIZATION_ID, + source_ref=source_ref, + acquisition_checkpoint=None, + publish_watermark=None, + ) + def tombstone_file_resource( self, call: TrustedControlCall, command: TombstoneFileResource ) -> FileResourceTombstone: diff --git a/tests/unit/test_file_source_progress.py b/tests/unit/test_file_source_progress.py new file mode 100644 index 00000000..8d5ebca0 --- /dev/null +++ b/tests/unit/test_file_source_progress.py @@ -0,0 +1,254 @@ +from __future__ import annotations + +from dataclasses import fields, replace +from datetime import UTC, datetime, timedelta +from uuid import UUID + +import pytest + +from engine.control import ( + ContextControl, + ControlOperation, + ControlOperatorAuthenticationRejected, + ControlOperatorAuthority, + FileResourceTombstone, + FileSourceAcquisitionCheckpoint, + FileSourceChangeKind, + FileSourceProgress, + FileSourcePublishOutcome, + FileSourcePublishWatermark, + RegisterFileSource, + SourceManifest, + SourceNotAvailable, + SourceRef, + TrustedControlCall, + VerifiedControlOperatorIdentity, +) +from engine.control.file_deletions import TombstoneFileResource +from engine.control.file_imports import PrepareFileImport +from engine.supply import PreparedFileImport + +ORGANIZATION_ID = UUID("f0381079-a64d-4984-977e-cd1654c049ed") +SOURCE_REF = SourceRef(UUID("e11d54e9-2ba6-4812-a215-794509bd1f4f")) +ACQUISITION_ID = UUID("61ca7538-e645-4d2a-a199-00dbf4960728") +JOB_ID = UUID("de2d89f2-042a-4fbe-b90b-6db335d02655") +REVISION_ID = UUID("270cd450-2b9f-4d66-997d-b47c85517031") +NOW = datetime(2026, 7, 23, 8, 0, tzinfo=UTC) +RESOURCE_REF = "resource:file:" + "a" * 64 + + +class _Authenticator: + def authenticate(self, opaque_credential: str) -> VerifiedControlOperatorIdentity: + if opaque_credential != "progress-reader": + raise ControlOperatorAuthenticationRejected + return VerifiedControlOperatorIdentity( + organization_id=ORGANIZATION_ID, + operator_ref="operator:progress-reader", + authentication_binding_ref="binding:progress-reader", + authority_ref="authority:source-progress", + allowed_operations=frozenset({ControlOperation.READ_SOURCE_PROGRESS}), + valid_from=NOW - timedelta(minutes=1), + expires_at=NOW + timedelta(minutes=10), + ) + + +class _Store: + def register_file_source( + self, call: TrustedControlCall, command: RegisterFileSource + ) -> SourceManifest: + raise AssertionError("unexpected registration") + + def read_source( + self, call: TrustedControlCall, source_ref: SourceRef + ) -> SourceManifest: + raise AssertionError("unexpected source read") + + def prepare_file_import( + self, call: TrustedControlCall, command: PrepareFileImport + ) -> PreparedFileImport: + raise AssertionError("unexpected import") + + def tombstone_file_resource( + self, call: TrustedControlCall, command: TombstoneFileResource + ) -> FileResourceTombstone: + raise AssertionError("unexpected tombstone") + + def read_file_source_progress( + self, + call: TrustedControlCall, + source_ref: SourceRef, + ) -> FileSourceProgress: + assert call.organization_id == ORGANIZATION_ID + assert call.operation is ControlOperation.READ_SOURCE_PROGRESS + if source_ref != SOURCE_REF: + raise SourceNotAvailable + checkpoint = FileSourceAcquisitionCheckpoint( + sequence=2, + checkpoint_ref="facp_" + "b" * 64, + change_kind=FileSourceChangeKind.FILE_IMPORT, + acquisition_ref=ACQUISITION_ID, + job_ref=JOB_ID, + cleanup_intent_ref=None, + resource_ref=None, + revision_ref=None, + event_ref=None, + event_sequence=None, + accepted_at=NOW, + ) + watermark = FileSourcePublishWatermark( + sequence=1, + watermark_ref="fpwm_" + "c" * 64, + checkpoint_ref="facp_" + "d" * 64, + change_kind=FileSourceChangeKind.FILE_IMPORT, + outcome=FileSourcePublishOutcome.PUBLISHED, + acquisition_ref=UUID("d9999beb-f185-454f-882b-6a2ca973d3ac"), + job_ref=UUID("ab87e202-42d8-49e9-87b0-a5317933bb07"), + cleanup_intent_ref=None, + resource_ref=RESOURCE_REF, + revision_ref=REVISION_ID, + event_ref=None, + event_sequence=None, + published_at=NOW - timedelta(seconds=1), + ) + return FileSourceProgress( + organization_id=ORGANIZATION_ID, + source_ref=source_ref, + acquisition_checkpoint=checkpoint, + publish_watermark=watermark, + ) + + +def test_control_reads_distinct_source_checkpoint_and_publish_watermark() -> None: + authority = ControlOperatorAuthority( + _Authenticator(), call_ttl=timedelta(minutes=5), clock=lambda: NOW + ) + control = ContextControl(store=_Store(), authority=authority, clock=lambda: NOW) + + with authority.authorize( + opaque_credential="progress-reader", + operation=ControlOperation.READ_SOURCE_PROGRESS, + request_id="read-source-progress", + ) as call: + progress = control.read_file_source_progress(call, SOURCE_REF) + + assert progress.organization_id == ORGANIZATION_ID + assert progress.source_ref == SOURCE_REF + assert progress.acquisition_checkpoint is not None + assert progress.publish_watermark is not None + assert progress.acquisition_checkpoint.sequence == 2 + assert progress.publish_watermark.sequence == 1 + assert progress.acquisition_checkpoint.checkpoint_ref.startswith("facp_") + assert progress.publish_watermark.watermark_ref.startswith("fpwm_") + + +def test_progress_contracts_keep_checkpoint_and_watermark_semantics_separate() -> None: + assert [field.name for field in fields(FileSourceProgress)] == [ + "organization_id", + "source_ref", + "acquisition_checkpoint", + "publish_watermark", + ] + assert FileSourceChangeKind.FILE_IMPORT.value == "file_import" + assert FileSourceChangeKind.FILE_TOMBSTONE.value == "file_tombstone" + assert FileSourcePublishOutcome.TOMBSTONED.value == "tombstoned" + + with pytest.raises(ValueError, match="cannot exceed"): + FileSourceProgress( + organization_id=ORGANIZATION_ID, + source_ref=SOURCE_REF, + acquisition_checkpoint=FileSourceAcquisitionCheckpoint( + sequence=1, + checkpoint_ref="facp_" + "a" * 64, + change_kind=FileSourceChangeKind.FILE_IMPORT, + acquisition_ref=ACQUISITION_ID, + job_ref=JOB_ID, + cleanup_intent_ref=None, + resource_ref=None, + revision_ref=None, + event_ref=None, + event_sequence=None, + accepted_at=NOW, + ), + publish_watermark=FileSourcePublishWatermark( + sequence=2, + watermark_ref="fpwm_" + "b" * 64, + checkpoint_ref="facp_" + "c" * 64, + change_kind=FileSourceChangeKind.FILE_IMPORT, + outcome=FileSourcePublishOutcome.PUBLISHED, + acquisition_ref=ACQUISITION_ID, + job_ref=JOB_ID, + cleanup_intent_ref=None, + resource_ref=RESOURCE_REF, + revision_ref=REVISION_ID, + event_ref=None, + event_sequence=None, + published_at=NOW, + ), + ) + + +def test_equal_progress_sequences_require_exact_checkpoint_lineage() -> None: + checkpoint_ref = "facp_" + "a" * 64 + checkpoint = FileSourceAcquisitionCheckpoint( + sequence=1, + checkpoint_ref=checkpoint_ref, + change_kind=FileSourceChangeKind.FILE_IMPORT, + acquisition_ref=ACQUISITION_ID, + job_ref=JOB_ID, + cleanup_intent_ref=None, + resource_ref=RESOURCE_REF, + revision_ref=REVISION_ID, + event_ref=None, + event_sequence=None, + accepted_at=NOW, + ) + watermark = FileSourcePublishWatermark( + sequence=1, + watermark_ref="fpwm_" + "b" * 64, + checkpoint_ref=checkpoint_ref, + change_kind=FileSourceChangeKind.FILE_IMPORT, + outcome=FileSourcePublishOutcome.PUBLISHED, + acquisition_ref=ACQUISITION_ID, + job_ref=JOB_ID, + cleanup_intent_ref=None, + resource_ref=RESOURCE_REF, + revision_ref=REVISION_ID, + event_ref=None, + event_sequence=None, + published_at=NOW, + ) + + progress = FileSourceProgress( + organization_id=ORGANIZATION_ID, + source_ref=SOURCE_REF, + acquisition_checkpoint=checkpoint, + publish_watermark=watermark, + ) + assert progress.publish_watermark == watermark + + with pytest.raises(ValueError, match="watermark lineage is invalid"): + FileSourceProgress( + organization_id=ORGANIZATION_ID, + source_ref=SOURCE_REF, + acquisition_checkpoint=checkpoint, + publish_watermark=replace( + watermark, + checkpoint_ref="facp_" + "c" * 64, + ), + ) + + +def test_progress_read_is_operation_bound_and_mismatch_fails_closed() -> None: + authority = ControlOperatorAuthority( + _Authenticator(), call_ttl=timedelta(minutes=5), clock=lambda: NOW + ) + control = ContextControl(store=_Store(), authority=authority, clock=lambda: NOW) + with authority.authorize( + opaque_credential="progress-reader", + operation=ControlOperation.READ_SOURCE_PROGRESS, + request_id="read-source-progress-mismatch", + ) as call, pytest.raises(SourceNotAvailable): + control.read_file_source_progress( + call, + SourceRef(UUID("0cd79e42-04b3-4146-929e-72c316171c99")), + ) diff --git a/tests/unit/test_m0_rls_inventory.py b/tests/unit/test_m0_rls_inventory.py index ea7f88bc..31d799dd 100644 --- a/tests/unit/test_m0_rls_inventory.py +++ b/tests/unit/test_m0_rls_inventory.py @@ -31,6 +31,8 @@ "file_publication_recovery", "file_resource_cleanup_intent", "file_resource_ingestion_guard", + "file_source_acquisition_checkpoint", + "file_source_publish_watermark", "file_revision_snapshot", "file_revision_replacement_plan", "file_revision_supersession", @@ -131,7 +133,7 @@ def test_manifest_declares_exact_live_table_denominator_and_rls_evidence() -> No assert global_tables == GLOBAL_TABLES assert tenant_tables == TENANT_TABLES - assert len(tables) == 37 + assert len(tables) == 39 for name in sorted(GLOBAL_TABLES): rationale = tables[name]["classificationRationale"] @@ -154,8 +156,8 @@ def test_rls_auditor_requires_every_live_control_and_non_owner_evidence() -> Non assert report["passed"] is True assert report["coverage"] == { - "numerator": 34, - "denominator": 34, + "numerator": 36, + "denominator": 36, "percent": 100.0, } inventory = cast(dict[str, object], report["inventory"]) @@ -180,7 +182,7 @@ def test_rls_auditor_does_not_count_force_rls_or_evidence_gaps() -> None: assert report["passed"] is False assert report["coverage"] == { "numerator": 0, - "denominator": 34, + "denominator": 36, "percent": 0.0, } tenant_reports = cast(list[dict[str, Any]], report["tenantTables"]) diff --git a/tests/unit/test_schema_security_manifest.py b/tests/unit/test_schema_security_manifest.py index 3b8c869d..6d1e2def 100644 --- a/tests/unit/test_schema_security_manifest.py +++ b/tests/unit/test_schema_security_manifest.py @@ -32,7 +32,7 @@ def test_manifest_classifies_the_exact_current_release_schema() -> None: document = manifest() tables = table_entries(document) - assert document["manifestVersion"] == "16.0.0" + assert document["manifestVersion"] == "17.0.0" assert set(tables) == { "active_release_manifest", "alembic_version", @@ -52,6 +52,8 @@ def test_manifest_classifies_the_exact_current_release_schema() -> None: "file_publication_recovery", "file_resource_cleanup_intent", "file_resource_ingestion_guard", + "file_source_acquisition_checkpoint", + "file_source_publish_watermark", "file_revision_snapshot", "file_revision_replacement_plan", "file_revision_supersession", @@ -107,6 +109,8 @@ def test_manifest_classifies_the_exact_current_release_schema() -> None: "file_revision_snapshot", "file_revision_replacement_plan", "file_revision_supersession", + "file_source_acquisition_checkpoint", + "file_source_publish_watermark", "revision_publication_event", ): assert tables[file_import_table]["classification"] == "tenant_owned" @@ -200,6 +204,7 @@ def test_issue_25_file_noop_contract_is_tenant_scoped_and_function_only() -> Non "unchangedReasonCode": "active-content-identity-match", "sourceContentRetainedInOutcome": False, } + assert "file_source_publish_watermark" in operation["atomicWrites"] guard = entries["file_resource_ingestion_guard"] result = entries["file_acquisition_result"] assert guard["organizationInclusiveKeys"][0]["columns"] == [ @@ -280,6 +285,8 @@ def test_issue_26_file_replacement_contract_is_staged_and_function_only() -> Non "file_resource_ingestion_guard", "file_acquisition_result", ] + assert "file_source_publish_watermark" in operation["stageAtomicWrites"] + assert "file_source_publish_watermark" in operation["activationAtomicWrites"] stage_functions = { "context_worker_stage_file_replacement", "context_worker_stage_structural_file_replacement", @@ -376,7 +383,8 @@ def test_issue_27_file_recovery_contract_is_generation_fenced_and_auditable() -> declared = function_only.get( "databaseFunctions", [function_only.get("databaseFunction")] ) - assert database_function in declared + causal = function_only.get("causalDatabaseFunctions", declared) + assert database_function in causal assert ( f"EXECUTE {database_function}" in entries[table_name]["permittedOperations"][ @@ -415,6 +423,8 @@ def test_issue_28_file_tombstone_contract_is_atomic_and_function_only() -> None: "context_resource", "organization_policy_epoch", "file_resource_cleanup_intent", + "file_source_acquisition_checkpoint", + "file_source_publish_watermark", ], } cleanup = entries["file_resource_cleanup_intent"] @@ -434,6 +444,17 @@ def test_issue_28_file_tombstone_contract_is_atomic_and_function_only() -> None: "kind": "unique", "columns": ["organization_id", "event_ref"], }, + { + "name": "uq_file_resource_cleanup_intent_progress_lineage", + "kind": "unique", + "columns": [ + "organization_id", + "cleanup_intent_id", + "source_id", + "resource_ref", + "revision_id", + ], + }, ] assert cleanup["rowLevelSecurity"]["enabled"] is True assert cleanup["rowLevelSecurity"]["forced"] is True