diff --git a/CHANGELOG.md b/CHANGELOG.md index b2f801dd..c85b28b0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,7 @@ All notable product and architecture changes are recorded here. Releases use imm - PostgreSQL persistence for immutable version-pinned scoring-request identity, with exact replay and fail-closed rebinding of snapshot or measurement-version evidence. - PostgreSQL 18 purpose-specific consent persistence for one participant-bound ledger plus append-only grant/revoke events, with exact replay, fail-closed conflicting event identity, independent research-scope shape, and `READ COMMITTED` replay classification. +- Atomic consent-change persistence with a tenant-, participant-, and insertion-order durable-ledger-tail-bound outbox event in the same caller-owned transaction. The durable event set must appear in the submitted ledger, and an ambiguous physical tail (shared `occurred_at_unix_ms` and `created_at`) fails closed instead of using opaque event identity as a correctness key, so a later stored revocation cannot be committed beside stale grant propagation. - PostgreSQL 18 inbox-consumption persistence so an accepted inbox receipt is not side-effect completion: pending identity, fenced processing claims that cannot be stolen, expire-and-reclaim of a crashed claim without transferring the fence, local or claimed completion, and poison quarantine, with exact replay and `READ COMMITTED` fail-closed classification. - Participant-owned consent authorization: `ResourceKind::ConsentLedger` and `ProductPermission::ManageOwnConsent` reuse the fail-closed tenant, resource-kind, and owner checks so consent cannot be treated as an untyped generic concern. - PostgreSQL expired-lease recovery for scoring jobs: an expired leased row becomes a due retry or quarantine with typed `lease_expired` evidence, and a later claim issues the next fencing token instead of inheriting the crashed worker's lease. diff --git a/docs/TRACEABILITY.md b/docs/TRACEABILITY.md index 72bc73c2..55f638e0 100644 --- a/docs/TRACEABILITY.md +++ b/docs/TRACEABILITY.md @@ -34,7 +34,7 @@ An active PR, architecture document, conversation decision, or scheduler plan is | Instrument publication requires intended-use scientific/right/locale evidence | PRD §6, §9, §10 | Measurement Governance; publication evidence gate | ADR-0004, ADR-0013, ADR-0019 | **Implemented** policy gate and immutable evidence provenance in `src/instrument.rs`; each real instrument still requires its own rights/locale/scientific evidence artifacts before publication | | Optional Keyverse account linking | PRD §3.1, §9.7 | TRD §10; UML identity-link lifecycle | ADR-0003, ADR-0020 | **Partially implemented**: issuer-scoped first-link fail-closed domain primitive in `src/participant.rs`; append-only unlink/relink/recovery history, persistence, audit, and transport remain Target | | Cross-cutting tenant/task authorization | PRD §7, §9 | TRD §11; Security/Data | ADR-0001, ADR-0003 | **Implemented** fail-closed domain gate in `src/authorization.rs` binds consent operations to participant-owned `ConsentLedger` / `ManageOwnConsent`; persistence/policy-adapter/public-transport integration remains Target | -| Purpose-specific consent | PRD §5, §9.6 | TRD §12 | ADR-0006 | **Implemented** domain contract in `src/consent.rs` plus `migrations/0005_consent_lifecycle.sql` / `src/postgres_consent.rs` purpose-specific ledgers; HTTP transport remains Target | +| Purpose-specific consent | PRD §5, §9.6 | TRD §12 | ADR-0006 | **Implemented** domain contract in `src/consent.rs` plus `migrations/0005_consent_lifecycle.sql` / `src/postgres_consent.rs` purpose-specific ledgers; HTTP transport remains Target. Active PR #142 composes that ledger with a bound outbox after locking the insertion-order durable tail and failing closed on an ambiguous physical tail | | Explicit research contribution + withdrawal | PRD §5 | TRD §12, §14–15 | ADR-0006, ADR-0007 | **Implemented** product-domain lifecycle in `src/consent.rs`; dataset snapshot/release integration is Target | | Participant export/deletion | PRD §3.1, §9, §11 | TRD §13 | ADR-0006 | **Implemented** domain lifecycle in `src/data_rights.rs` plus `migrations/0003_data_rights_propagation.sql` and `src/postgres_data_rights.rs`; dependent-system execution remains Target | | Research identity separation | PRD §5, §11 | TRD §14; ERD restricted linkage | ADR-0003, ADR-0006, ADR-0007, ADR-0020 | Partially implemented via research-contribution identity separation; restricted linkage persistence is Target | @@ -75,7 +75,7 @@ An active PR, architecture document, conversation decision, or scheduler plan is | No cross-service DB access | TRD §1–2; ADR-0015 | architecture policy only | deployment credential/fitness-function test | | Initial physical persistence target is upstream PostgreSQL 18.x | ADR-0015; Deployment/Operations | **Implemented subset** in `migrations/0001_integration_delivery.sql`, `migrations/0002_scoring_job_state.sql`, `migrations/0003_data_rights_propagation.sql`, `migrations/0005_consent_lifecycle.sql`, `migrations/0006_instrument_release.sql`, `migrations/0011_scoring_request.sql`, `migrations/0012_integration_consumption.sql`, matching adapters, and PostgreSQL operational-store readiness | remaining product aggregates, crash/restart restore acceptance | | No default tenant for writes | TRD §11; Security/Data | authorization-domain primitive exists; persistence remains Target | persistence/API tenant negative tests | -| Tenant-bound transactional outbox/inbox | TRD §19–20; ADR-0014/0015 | `src/integration.rs` domain envelope/inbox/retry contracts plus PostgreSQL tenant/source-scoped integration evidence, delivery-attempt persistence, and inbox consumption | durable side-effect processing completion, poison-message/crash recovery, broader aggregate transaction integration | +| Tenant-bound transactional outbox/inbox | TRD §19–20; ADR-0014/0015 | `src/integration.rs` domain envelope/inbox/retry contracts plus PostgreSQL tenant/source-scoped integration evidence, delivery-attempt persistence, and inbox consumption | durable side-effect processing completion, poison-message/crash recovery, broader aggregate transaction integration. Active PR #142 binds one consent change to one outbox row after locking the insertion-order durable ledger tail and rejecting an ambiguous `(occurred_at_unix_ms, created_at)` tail | | Inbox receipt is not side-effect completion | ADR-0014/0015; UML integration sequence | `src/integration.rs` states/retry semantics; PostgreSQL inbox consumption persists pending/processing/completed and expire-and-reclaim | live adapter crash/retry tests | | Liveness is distinct from operation readiness | Operability §3–4; ADR-0017 | **Implemented** in `src/health.rs` and `src/postgres_health.rs`: liveness is modeled independently from operation-scoped readiness and PostgreSQL write-readiness | live transport probes, metrics, and deployment-profile acceptance | | Optional capability outage does not fail unrelated work | Operability §3–4; ADR-0011/0017 | **Implemented** in `src/health.rs` and `src/postgres_health.rs`: readiness evaluates only capabilities required by the selected operation and maps PostgreSQL evidence onto that contract | degraded-mode transport/integration tests | @@ -134,6 +134,8 @@ Still-Target logical modules/adapters include remaining product aggregate persis **Active PR** #76 data-rights processing-start persistence is not protected-main truth until an unchanged reviewed/check-clean head is integrated. Identity-verified requests persist an immutable operation identity and processing-start time under `FOR UPDATE` so later lifecycle composition cannot race the classified row. Dependent-system execution remains outside this slice. +**Active PR** #142 consent/outbox composition is not protected-main truth until an unchanged reviewed/check-clean head is integrated. `src/postgres_consent_propagation.rs` persists one purpose-specific ledger snapshot and one `psychometrics_commons` outbox envelope in the caller-owned transaction, then locks `consent_ledger`, requires every durable `consent_event` identity to appear in the submitted ledger, and requires a unique durable tail—ordered by occurrence time, then `created_at`—to match `causation_ref` and occurrence time. An ambiguous physical tail fails closed. Do not merge #70 at `3180620`, #112 at `040bcf7`, #120 at `3f72446`, #123 at `6643041`, or #134 at `729a2bd`. HTTP consent transport and live outbox dispatch remain outside this slice. + ## 5. ADR traceability by concern | Concern | Governing ADR(s) | diff --git a/docs/adr/0006-consent-data-rights-and-research-separation.md b/docs/adr/0006-consent-data-rights-and-research-separation.md index 9dbcf296..15fc6ee4 100644 --- a/docs/adr/0006-consent-data-rights-and-research-separation.md +++ b/docs/adr/0006-consent-data-rights-and-research-separation.md @@ -64,10 +64,12 @@ Construct-relevant personal data remains available to authorized workflows when 4. Public research data contains no Keyverse subject or operational participant reference. 5. Service denial cannot be based on refusal of optional research contribution. 6. Data-rights operations are tenant-scoped and identity-verified. +7. Two durable consent events that share occurrence time and physical insertion time fail closed; opaque event identity is not a correctness tie-breaker. +8. A later consent-receipt or notice interchange, when added, must map onto ISO/IEC TS 27560:2023 consent-record structure and ISO/IEC 29184:2020 notice/consent presentation. This decision does not emit those receipts yet. ## Failure behavior -If consent verification is unavailable, optional research processing fails closed while core assessment processing may continue under its valid service basis. An ambiguous withdrawal or identity conflict enters manual adjudication; it is not auto-resolved by email or display name. +If consent verification is unavailable, optional research processing fails closed while core assessment processing may continue under its valid service basis. An ambiguous withdrawal, identity conflict, or physically tied consent tail enters fail-closed rejection or manual adjudication; it is not auto-resolved by email, display name, or opaque event-reference order. ## Validation @@ -86,3 +88,15 @@ If consent verification is unavailable, optional research processing fails close ## Reversal conditions Revisit individual retention or withdrawal mechanics when a deployment's law or ethics approval imposes stricter requirements. The separation of service and research purposes remains mandatory. + +## References + +European Parliament & Council of the European Union. (2016). Regulation (EU) 2016/679 of the European Parliament and of the Council of 27 April 2016 on the protection of natural persons with regard to the processing of personal data and on the free movement of such data (General Data Protection Regulation). *Official Journal of the European Union, L 119*, 1–88. https://eur-lex.europa.eu/eli/reg/2016/679/oj + +International Organization for Standardization. (2020). *Information technology — Online privacy notices and consent* (ISO/IEC 29184:2020). https://www.iso.org/standard/70331.html + +International Organization for Standardization. (2023). *Privacy technologies — Consent record information structure* (ISO/IEC TS 27560:2023). https://www.iso.org/standard/80392.html + +National Institute of Standards and Technology. (2020). *NIST Privacy Framework: A tool for improving privacy through enterprise risk management, version 1.0* (NIST Cybersecurity White Paper). https://doi.org/10.6028/NIST.CSWP.01162020 + +World Medical Association. (2024). World Medical Association Declaration of Helsinki: Ethical principles for medical research involving human participants. *JAMA, 333*(1), 71–74. https://doi.org/10.1001/jama.2024.21972 diff --git a/docs/adr/0015-persistence-and-transaction-boundaries.md b/docs/adr/0015-persistence-and-transaction-boundaries.md index c6643000..3078c0bf 100644 --- a/docs/adr/0015-persistence-and-transaction-boundaries.md +++ b/docs/adr/0015-persistence-and-transaction-boundaries.md @@ -114,6 +114,8 @@ An inbox row that merely proves receipt is never marked `completed` before the r Consent decisions and data-rights lifecycle events are append-only evidence. External propagation of deletion/export/research changes is asynchronous and reconciled; local state never claims an external effect completed until evidence exists. +A consent-change write that must propagate shares the caller-owned transaction with its bound outbox row. After the submitted ledger snapshot is persisted, the same transaction locks `consent_ledger`, requires every durable `consent_event` identity to appear in the submitted ledger, and requires a unique durable tail—ordered by `occurred_at_unix_ms`, then physical `created_at`—to equal the envelope `causation_ref` and occurrence time. Equal server timestamps therefore keep a later-inserted revocation ahead of an earlier grant. Two rows that also share `created_at` fail closed; opaque `event_ref` is not a correctness tie-breaker. Callers roll back when either adapter fails so newly accepted consent evidence cannot survive without its outbox record. + ## Concurrency and idempotency Physical constraints must enforce equivalents of: @@ -303,6 +305,12 @@ The physical database technology or decomposition may change if scale, residency ## References +European Parliament & Council of the European Union. (2016). Regulation (EU) 2016/679 of the European Parliament and of the Council of 27 April 2016 on the protection of natural persons with regard to the processing of personal data and on the free movement of such data (General Data Protection Regulation). *Official Journal of the European Union, L 119*, 1–88. https://eur-lex.europa.eu/eli/reg/2016/679/oj + +Hohpe, G., & Woolf, B. (2003). *Enterprise integration patterns: Designing, building, and deploying messaging solutions*. Addison-Wesley. + PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation*. +PostgreSQL Global Development Group. (2026). *Date/time functions and operators*. https://www.postgresql.org/docs/18/functions-datetime.html + PostgreSQL Global Development Group. (2026). *PostgreSQL versioning policy*. diff --git a/docs/architecture/AS_BUILT_SCHEMA.md b/docs/architecture/AS_BUILT_SCHEMA.md index 8c2c3510..d43ce4b7 100644 --- a/docs/architecture/AS_BUILT_SCHEMA.md +++ b/docs/architecture/AS_BUILT_SCHEMA.md @@ -58,6 +58,10 @@ The protected-main slice persists: The slice does **not** persist publication-event history, bound scientific evidence records, HTTP publication transport, or session-creation integration. Those remain Target unless separately evidenced on protected main. +## Active PR consent/outbox composition + +Active PR #142 (`src/postgres_consent_propagation.rs`) adds no new physical objects. It composes the existing `consent_ledger` / `consent_event` slice with `integration_outbox` in one caller-owned `READ COMMITTED` transaction. After the submitted ledger snapshot is persisted, the adapter locks the participant `consent_ledger` row, requires every durable `consent_event` identity to appear in the submitted ledger, and requires a unique durable tail—ordered by `occurred_at_unix_ms`, then `created_at`—to match the outbox `causation_ref` and occurrence time. A grant-only in-memory snapshot therefore cannot enqueue grant propagation after a later stored revocation, and two rows that share occurrence time and `created_at` fail closed instead of using opaque `event_ref` as a correctness key. This slice is **Active PR**, not protected-main truth. Do not merge #70, #112, #120, #123, or #134. + ## Logical-to-physical mapping rule A logical entity is classified as physical only when all of the following exist on the named protected-main baseline: diff --git a/docs/architecture/ERD.md b/docs/architecture/ERD.md index 8f21954a..ccd483ec 100644 --- a/docs/architecture/ERD.md +++ b/docs/architecture/ERD.md @@ -425,7 +425,7 @@ The target ERD deliberately includes several logical entities that are not yet p - `instrument_release` is the locale-specific publication identity already owned by `src/instrument.rs`. Physical `migrations/0006_instrument_release.sql` persists that one-row aggregate (immutable manifest columns plus `publication_state`); HTTP publication transport remains Target. - `data_rights_request` and `data_rights_propagation_state` are the first durable export/deletion slice. Physical `migrations/0003_data_rights_propagation.sql` stores requested-state identity plus one local outbox event per dependent system; verification, processing, completion, and dependent-system execution remain Target. - `item_delivery_event` reflects the already-merged `src/item_delivery.rs` domain primitive; durable persistence/API orchestration is still Target. -- `consent_ledger` and `consent_event` persist the already-merged `src/consent.rs` append-only ledger. Physical persistence is carried by Active PR #49 (`migrations/0005_consent_lifecycle.sql`); HTTP consent transport and derived snapshot tables remain Target. +- `consent_ledger` and `consent_event` persist the already-merged `src/consent.rs` append-only ledger through `migrations/0005_consent_lifecycle.sql`. Active PR #142 composes those existing rows with `integration_outbox` in one caller-owned transaction after locking the participant ledger, requiring the submitted ledger to contain every durable event identity, and selecting a unique durable tail by occurrence time then `created_at`; it adds no new physical objects. HTTP consent transport and derived snapshot tables remain Target. - `participant_identity_link` is the persistence target accepted by ADR-0020. The current `src/participant.rs` `keyverse_subject_ref` field is an application-domain first-link projection, not the future mutable persistence source of truth. - `longitudinal_enrollment`, `longitudinal_observation_record`, and `temporal_analysis_submission` make the ADR-0008 Commons-owned Gyeot/TEPP orchestration boundary explicit. No TEPP analytical kernel is duplicated here. - `integration_outbox`, `integration_delivery_attempt`, `integration_inbox`, and `integration_consumption` reflect `src/integration.rs` domain semantics. Outbox/inbox/delivery-attempt tables are on protected main; `integration_consumption` pending/processing/completed/quarantined persistence and expire-and-reclaim of a crashed processing claim exist only on this Active PR until merged. diff --git a/docs/architecture/UML.md b/docs/architecture/UML.md index 13988939..91e4e31c 100644 --- a/docs/architecture/UML.md +++ b/docs/architecture/UML.md @@ -451,7 +451,8 @@ sequenceDiagram participant S as semantic-data-portal P->>A: explicit research opt-in for versioned scope - A->>DB: append consent evidence + research contribution + A->>DB: append consent evidence + bound outbox + research contribution + Note over A,DB: Active PR #142 commits the consent ledger change with its causally bound outbox row after locking the insertion-order durable ledger tail and failing closed on an ambiguous physical tail DB-->>A: contribution_ref A->>L: create/reuse scoped research pseudonym L-->>A: research_participant_ref diff --git a/src/lib.rs b/src/lib.rs index 8b586a68..e6e5c1b2 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -20,6 +20,7 @@ pub mod item_delivery; pub mod narrative; pub mod participant; pub mod postgres_consent; +pub mod postgres_consent_propagation; pub mod postgres_data_rights; pub mod postgres_data_rights_processing; pub mod postgres_health; diff --git a/src/postgres_consent_propagation.rs b/src/postgres_consent_propagation.rs new file mode 100644 index 00000000..e325ec34 --- /dev/null +++ b/src/postgres_consent_propagation.rs @@ -0,0 +1,712 @@ +//! Atomic `PostgreSQL` composition for consent evidence and propagation outbox events. +//! +//! Consent remains purpose-specific, append-only product evidence. This module composes the +//! existing consent adapter with the existing transactional outbox and verifies that the emitted +//! event is bound to the authorized tenant, same participant, and latest accepted consent event +//! before any durable write. After the ledger snapshot is persisted, the same transaction locks +//! the participant ledger and requires a unique durable event tail—ordered by occurrence time, then +//! insertion time—to match the envelope, so a grant-only in-memory snapshot cannot pair a later +//! stored revocation with stale grant propagation even when both events share one server timestamp. +//! An ambiguous physical tail—two rows with the same occurrence time and `created_at`—fails closed +//! instead of using opaque event identity as a correctness key. + +use crate::consent::{ConsentEvent, ConsentLedger}; +use crate::integration::IntegrationEvent; +use crate::postgres_consent::{ + persist_consent_ledger, ConsentPersistenceDisposition, ConsentPersistenceError, +}; +use crate::postgres_integration::{enqueue_outbox_event, PersistenceDisposition, PersistenceError}; +use crate::reference::normalized_reference; +use postgres::Transaction; +use std::collections::HashSet; +use std::error::Error; +use std::fmt::{Display, Formatter}; +use std::time::SystemTime; + +const SOURCE_REF: &str = "psychometrics_commons"; + +/// Durable dispositions produced by one atomic consent/outbox operation. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct ConsentOutboxPersistence { + consent: ConsentPersistenceDisposition, + outbox: PersistenceDisposition, +} + +impl ConsentOutboxPersistence { + /// Return whether consent evidence was newly inserted or exactly replayed. + #[must_use] + pub const fn consent(self) -> ConsentPersistenceDisposition { + self.consent + } + + /// Return whether immutable outbox evidence was newly inserted or exactly replayed. + #[must_use] + pub const fn outbox(self) -> PersistenceDisposition { + self.outbox + } +} + +/// Fail-closed error for atomic consent and outbox persistence. +#[derive(Debug)] +#[non_exhaustive] +pub enum ConsentOutboxPersistenceError { + /// The propagation envelope does not identify this tenant, participant, and latest ledger event. + InvalidPropagationEnvelope, + /// Durable consent evidence failed validation or persistence. + Consent(ConsentPersistenceError), + /// Durable integration outbox evidence failed validation or persistence. + Outbox(PersistenceError), +} + +impl Display for ConsentOutboxPersistenceError { + fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result { + formatter.write_str(match self { + Self::InvalidPropagationEnvelope => { + "consent propagation event must bind the authorized tenant, participant, and latest consent event" + } + Self::Consent(_) => "consent propagation consent persistence failed", + Self::Outbox(_) => "consent propagation outbox persistence failed", + }) + } +} + +impl Error for ConsentOutboxPersistenceError { + fn source(&self) -> Option<&(dyn Error + 'static)> { + match self { + Self::InvalidPropagationEnvelope => None, + Self::Consent(error) => Some(error), + Self::Outbox(error) => Some(error), + } + } +} + +/// Persist one consent ledger and its latest causally bound outbox event in the same transaction. +/// +/// `authorized_tenant_ref` is the product authorization context resolved by the caller. It must +/// already be in canonical opaque-reference spelling; this boundary never normalizes aliases. The +/// integration event must use that exact tenant, be emitted by `psychometrics_commons`, use the +/// consent ledger's participant as its subject, identify the latest accepted consent event through +/// `causation_ref`, and use that consent event's server-authoritative occurrence time. Binding +/// propagation to tenant, participant, and ledger tail prevents cross-tenant dispatch and prevents +/// a later revocation or grant from being durably paired with stale historical propagation evidence. +/// In-memory envelope checks run before persistence. After the submitted snapshot is written, the +/// same transaction locks `consent_ledger`, requires every durable `consent_event` identity to +/// appear in the submitted ledger, and requires a unique durable tail—ordered by occurrence time, +/// then physical insertion time—to equal that causation reference and time. Equal server timestamps +/// therefore keep a later-inserted revocation ahead of an earlier grant. Two rows that also share +/// `created_at` fail closed instead of letting opaque `event_ref` pick a winner. A caller that +/// omits a later stored revocation or grant therefore fails closed before an outbox row is created. +/// Callers should persist one new consent event per composition so an earlier purpose change is not +/// hidden behind a later event that becomes the only bound outbox. Event type, correlation +/// reference, schema version, and payload digest remain owned by the caller's versioned +/// integration contract. +/// +/// The caller owns the `READ COMMITTED` transaction and final commit/rollback decision. If either +/// durable adapter fails, callers must roll the transaction back so newly accepted consent evidence +/// cannot survive without its outbox record. Exact replay remains idempotent at both adapters. +/// +/// # Errors +/// +/// Returns [`ConsentOutboxPersistenceError::InvalidPropagationEnvelope`] before writes for an +/// invalid, non-canonical, or mismatched authorized tenant, unrelated source or participant, stale +/// or missing causation reference, or timestamp, and after ledger persistence when the durable +/// tail or durable event set does not match the envelope, including when the physical tail is +/// ambiguous. Consent and outbox failures are preserved in typed error variants. +pub fn persist_consent_ledger_with_outbox( + transaction: &mut Transaction<'_>, + authorized_tenant_ref: &str, + ledger: &ConsentLedger, + propagation_event: &IntegrationEvent, + outbox_max_attempts: usize, +) -> Result { + validate_propagation_envelope(authorized_tenant_ref, ledger, propagation_event)?; + let consent = persist_consent_ledger(transaction, ledger) + .map_err(ConsentOutboxPersistenceError::Consent)?; + require_durable_ledger_tail(transaction, ledger, propagation_event)?; + let outbox = enqueue_outbox_event(transaction, propagation_event, outbox_max_attempts) + .map_err(ConsentOutboxPersistenceError::Outbox)?; + Ok(ConsentOutboxPersistence { consent, outbox }) +} + +fn validate_propagation_envelope( + authorized_tenant_ref: &str, + ledger: &ConsentLedger, + propagation_event: &IntegrationEvent, +) -> Result<(), ConsentOutboxPersistenceError> { + let Some(normalized_tenant_ref) = normalized_reference(authorized_tenant_ref) else { + return Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope); + }; + if normalized_tenant_ref != authorized_tenant_ref + || propagation_event.tenant_ref() != authorized_tenant_ref + || propagation_event.source() != SOURCE_REF + || propagation_event.subject_ref() != ledger.participant_ref() + { + return Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope); + } + let Some(causation_ref) = propagation_event.causation_ref() else { + return Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope); + }; + let Some(consent_event) = ledger + .events() + .last() + .filter(|event| event.event_ref() == causation_ref) + else { + return Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope); + }; + if consent_event.occurred_at_unix_ms() != propagation_event.occurred_at_unix_ms() { + return Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope); + } + Ok(()) +} + +fn require_durable_ledger_tail( + transaction: &mut Transaction<'_>, + ledger: &ConsentLedger, + propagation_event: &IntegrationEvent, +) -> Result<(), ConsentOutboxPersistenceError> { + let Some(causation_ref) = propagation_event.causation_ref() else { + return Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope); + }; + if transaction + .query_opt( + "SELECT participant_ref FROM consent_ledger WHERE participant_ref = $1 FOR UPDATE", + &[&ledger.participant_ref()], + ) + .map_err(|error| ConsentOutboxPersistenceError::Consent(error.into()))? + .is_none() + { + return Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope); + } + let rows = transaction + .query( + "SELECT event_ref, occurred_at_unix_ms, created_at \ + FROM consent_event \ + WHERE participant_ref = $1 \ + ORDER BY occurred_at_unix_ms DESC, created_at DESC", + &[&ledger.participant_ref()], + ) + .map_err(|error| ConsentOutboxPersistenceError::Consent(error.into()))?; + let Some(tail) = rows.first() else { + return Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope); + }; + let submitted: HashSet<&str> = ledger + .events() + .iter() + .map(ConsentEvent::event_ref) + .collect(); + for stored in &rows { + let stored_event_ref: String = stored.get(0); + if !submitted.contains(stored_event_ref.as_str()) { + return Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope); + } + } + let tail_occurred: i64 = tail.get(1); + let tail_created_at: SystemTime = tail.get(2); + let tied = rows + .iter() + .filter(|row| { + let occurred: i64 = row.get(1); + let created_at: SystemTime = row.get(2); + occurred == tail_occurred && created_at == tail_created_at + }) + .count(); + if tied != 1 { + return Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope); + } + let stored_event_ref: String = tail.get(0); + let Ok(expected_occurred) = i64::try_from(propagation_event.occurred_at_unix_ms()) else { + return Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope); + }; + if stored_event_ref != causation_ref { + return Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope); + } + if tail_occurred != expected_occurred { + return Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope); + } + Ok(()) +} + +#[cfg(test)] +mod envelope_tests { + use super::{validate_propagation_envelope, ConsentOutboxPersistenceError}; + use crate::consent::{ConsentDecision, ConsentEventInput, ConsentLedger, ConsentPurpose}; + use crate::integration::IntegrationEvent; + + const DIGEST: &str = "sha256:dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd"; + const TENANT_REF: &str = "tenant_consent_envelope_unit"; + + fn ledger() -> ConsentLedger { + let mut ledger = ConsentLedger::new("participant_consent_envelope_unit").unwrap(); + ledger + .record(ConsentEventInput { + event_ref: "consent_event_envelope_unit", + purpose: ConsentPurpose::ResearchContribution, + decision: ConsentDecision::Granted, + consent_form_version_ref: "consent_form_envelope_unit", + research_scope_ref: Some("research_scope_envelope_unit"), + occurred_at_unix_ms: 10_000, + }) + .unwrap(); + ledger + } + + fn event( + source: &str, + tenant_ref: &str, + subject: &str, + causation: Option<&str>, + occurred_at_unix_ms: u64, + ) -> IntegrationEvent { + IntegrationEvent::new( + "event_consent_envelope_unit", + "consent.changed", + "v1", + source, + tenant_ref, + subject, + occurred_at_unix_ms, + "correlation_consent_envelope_unit", + causation, + DIGEST, + ) + .unwrap() + } + + #[test] + fn every_envelope_binding_boundary_fails_closed() { + let ledger = ledger(); + let valid = event( + "psychometrics_commons", + TENANT_REF, + ledger.participant_ref(), + Some("consent_event_envelope_unit"), + 10_000, + ); + assert!(validate_propagation_envelope(TENANT_REF, &ledger, &valid).is_ok()); + + let invalid = [ + ( + "tenant_consent_envelope_other", + event( + "psychometrics_commons", + TENANT_REF, + ledger.participant_ref(), + Some("consent_event_envelope_unit"), + 10_000, + ), + ), + ( + TENANT_REF, + event( + "psychometrics_commons", + "tenant_consent_envelope_other", + ledger.participant_ref(), + Some("consent_event_envelope_unit"), + 10_000, + ), + ), + ( + TENANT_REF, + event( + "other_source", + TENANT_REF, + ledger.participant_ref(), + Some("consent_event_envelope_unit"), + 10_000, + ), + ), + ( + TENANT_REF, + event( + "psychometrics_commons", + TENANT_REF, + "participant_other", + Some("consent_event_envelope_unit"), + 10_000, + ), + ), + ( + TENANT_REF, + event( + "psychometrics_commons", + TENANT_REF, + ledger.participant_ref(), + None, + 10_000, + ), + ), + ( + TENANT_REF, + event( + "psychometrics_commons", + TENANT_REF, + ledger.participant_ref(), + Some("consent_event_unknown"), + 10_000, + ), + ), + ( + TENANT_REF, + event( + "psychometrics_commons", + TENANT_REF, + ledger.participant_ref(), + Some("consent_event_envelope_unit"), + 10_001, + ), + ), + ]; + for (authorized_tenant_ref, candidate) in invalid { + assert!(matches!( + validate_propagation_envelope(authorized_tenant_ref, &ledger, &candidate), + Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope) + )); + } + + for invalid_tenant_ref in [ + " ", + "42", + " tenant_consent_envelope_unit", + "tenant_consent_envelope_unit ", + ] { + assert!(matches!( + validate_propagation_envelope(invalid_tenant_ref, &ledger, &valid), + Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope) + )); + } + } + + #[test] + fn empty_ledger_cannot_bind_a_propagation_envelope() { + let empty = ConsentLedger::new("participant_consent_envelope_unit").unwrap(); + let candidate = event( + "psychometrics_commons", + TENANT_REF, + empty.participant_ref(), + Some("consent_event_envelope_unit"), + 10_000, + ); + assert!(matches!( + validate_propagation_envelope(TENANT_REF, &empty, &candidate), + Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope) + )); + } +} + +#[cfg(test)] +mod durable_tail_boundary_tests { + use super::{require_durable_ledger_tail, ConsentOutboxPersistenceError}; + use crate::consent::{ConsentDecision, ConsentEventInput, ConsentLedger, ConsentPurpose}; + use crate::integration::IntegrationEvent; + use crate::postgres_consent::apply_consent_migration; + use postgres::{Client, NoTls}; + + const DIGEST: &str = "sha256:ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff"; + const SCHEMA: &str = "consent_outbox_durable_tail_unit"; + const AMBIGUOUS_SCHEMA: &str = "consent_outbox_durable_tail_ambiguous_unit"; + + fn ready_client() -> Client { + ready_schema_client(SCHEMA) + } + + fn ready_schema_client(schema: &str) -> Client { + let connection = std::env::var("TEST_DATABASE_URL") + .expect("TEST_DATABASE_URL must identify the isolated CI PostgreSQL database"); + let mut client = Client::connect(&connection, NoTls) + .expect("isolated CI PostgreSQL database must be reachable"); + client + .batch_execute(&format!( + "DROP SCHEMA IF EXISTS {schema} CASCADE; + CREATE SCHEMA {schema}; + SET search_path TO {schema};" + )) + .unwrap(); + apply_consent_migration(&mut client).unwrap(); + client + } + + fn grant_ledger() -> ConsentLedger { + let mut ledger = ConsentLedger::new("participant_consent_durable_tail").unwrap(); + ledger + .record(ConsentEventInput { + event_ref: "consent_event_durable_tail_grant", + purpose: ConsentPurpose::ResearchContribution, + decision: ConsentDecision::Granted, + consent_form_version_ref: "consent_form_durable_tail", + research_scope_ref: Some("research_scope_durable_tail"), + occurred_at_unix_ms: 40_000, + }) + .unwrap(); + ledger + } + + fn research_ledger(event_ref: &str, decision: ConsentDecision) -> ConsentLedger { + let mut ledger = ConsentLedger::new("participant_consent_durable_tail").unwrap(); + ledger + .record(ConsentEventInput { + event_ref, + purpose: ConsentPurpose::ResearchContribution, + decision, + consent_form_version_ref: "consent_form_durable_tail", + research_scope_ref: Some("research_scope_durable_tail"), + occurred_at_unix_ms: 40_000, + }) + .unwrap(); + ledger + } + + fn insert_tied_grant_and_revoke(client: &mut Client) { + let research_scope_ref = Some("research_scope_durable_tail"); + client + .execute( + "INSERT INTO consent_ledger (participant_ref) VALUES ($1)", + &[&"participant_consent_durable_tail"], + ) + .unwrap(); + for (event_ref, decision) in [ + ("consent_event_zzz_tied_grant", "granted"), + ("consent_event_aaa_tied_revoke", "revoked"), + ] { + client + .execute( + "INSERT INTO consent_event (\ + participant_ref, event_ref, consent_purpose, consent_decision, \ + consent_form_version_ref, research_scope_ref, occurred_at_unix_ms, created_at\ + ) VALUES ($1, $2, 'research_contribution', $3, $4, $5, 40000, \ + TIMESTAMPTZ '2026-08-16 15:00:00+00')", + &[ + &"participant_consent_durable_tail", + &event_ref, + &decision, + &"consent_form_durable_tail", + &research_scope_ref, + ], + ) + .unwrap(); + } + } + + fn event(causation: Option<&str>, occurred_at_unix_ms: u64) -> IntegrationEvent { + IntegrationEvent::new( + "event_consent_durable_tail", + "consent.changed", + "v1", + "psychometrics_commons", + "tenant_consent_durable_tail", + "participant_consent_durable_tail", + occurred_at_unix_ms, + "correlation_consent_durable_tail", + causation, + DIGEST, + ) + .unwrap() + } + + #[test] + fn durable_tail_boundaries_fail_closed_without_enqueue() { + let mut client = ready_client(); + let ledger = grant_ledger(); + let bound = event(Some("consent_event_durable_tail_grant"), 40_000); + + let mut missing_ledger = client.transaction().unwrap(); + assert!(matches!( + require_durable_ledger_tail(&mut missing_ledger, &ledger, &bound), + Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope) + )); + assert!(matches!( + require_durable_ledger_tail(&mut missing_ledger, &ledger, &event(None, 40_000)), + Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope) + )); + missing_ledger.rollback().unwrap(); + + client + .execute( + "INSERT INTO consent_ledger (participant_ref) VALUES ($1)", + &[&ledger.participant_ref()], + ) + .unwrap(); + let mut missing_events = client.transaction().unwrap(); + assert!(matches!( + require_durable_ledger_tail(&mut missing_events, &ledger, &bound), + Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope) + )); + missing_events.rollback().unwrap(); + + let research_scope_ref = Some("research_scope_durable_tail"); + client + .execute( + "INSERT INTO consent_event (\ + participant_ref, event_ref, consent_purpose, consent_decision, \ + consent_form_version_ref, research_scope_ref, occurred_at_unix_ms\ + ) VALUES ($1, $2, 'research_contribution', 'granted', $3, $4, 40000)", + &[ + &ledger.participant_ref(), + &"consent_event_durable_tail_grant", + &"consent_form_durable_tail", + &research_scope_ref, + ], + ) + .unwrap(); + + let mut mismatch = client.transaction().unwrap(); + assert!(matches!( + require_durable_ledger_tail( + &mut mismatch, + &ledger, + &event(Some("consent_event_durable_tail_grant"), 40_001), + ), + Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope) + )); + assert!(matches!( + require_durable_ledger_tail( + &mut mismatch, + &ledger, + &event(Some("consent_event_durable_tail_other"), 40_000), + ), + Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope) + )); + require_durable_ledger_tail(&mut mismatch, &ledger, &bound) + .expect("exact durable tail should accept the bound envelope"); + mismatch.rollback().unwrap(); + + let mut overflow_transaction = client.transaction().unwrap(); + assert!(matches!( + require_durable_ledger_tail( + &mut overflow_transaction, + &ledger, + &event(Some("consent_event_durable_tail_grant"), u64::MAX), + ), + Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope) + )); + overflow_transaction.rollback().unwrap(); + + client.batch_execute("DROP TABLE consent_event;").unwrap(); + let mut missing_event_relation = client.transaction().unwrap(); + assert!(matches!( + require_durable_ledger_tail(&mut missing_event_relation, &ledger, &bound), + Err(ConsentOutboxPersistenceError::Consent(_)) + )); + missing_event_relation.rollback().unwrap(); + client.batch_execute("DROP TABLE consent_ledger;").unwrap(); + let mut missing_ledger_relation = client.transaction().unwrap(); + assert!(matches!( + require_durable_ledger_tail(&mut missing_ledger_relation, &ledger, &bound), + Err(ConsentOutboxPersistenceError::Consent(_)) + )); + missing_ledger_relation.rollback().unwrap(); + client + .batch_execute(&format!("DROP SCHEMA IF EXISTS {SCHEMA} CASCADE;")) + .unwrap(); + } + + #[test] + fn same_millisecond_revoke_is_the_durable_tail() { + let mut client = ready_client(); + let ledger = grant_ledger(); + let research_scope_ref = Some("research_scope_durable_tail"); + client + .execute( + "INSERT INTO consent_ledger (participant_ref) VALUES ($1)", + &[&ledger.participant_ref()], + ) + .unwrap(); + client + .execute( + "INSERT INTO consent_event (\ + participant_ref, event_ref, consent_purpose, consent_decision, \ + consent_form_version_ref, research_scope_ref, occurred_at_unix_ms\ + ) VALUES ($1, $2, 'research_contribution', 'granted', $3, $4, 40000)", + &[ + &ledger.participant_ref(), + &"consent_event_durable_tail_grant", + &"consent_form_durable_tail", + &research_scope_ref, + ], + ) + .unwrap(); + client + .execute( + "INSERT INTO consent_event (\ + participant_ref, event_ref, consent_purpose, consent_decision, \ + consent_form_version_ref, research_scope_ref, occurred_at_unix_ms\ + ) VALUES ($1, $2, 'research_contribution', 'revoked', $3, $4, 40000)", + &[ + &ledger.participant_ref(), + &"consent_event_aaa_durable_tail_revoke", + &"consent_form_durable_tail", + &research_scope_ref, + ], + ) + .unwrap(); + + let mut complete = ledger.clone(); + complete + .record(ConsentEventInput { + event_ref: "consent_event_aaa_durable_tail_revoke", + purpose: ConsentPurpose::ResearchContribution, + decision: ConsentDecision::Revoked, + consent_form_version_ref: "consent_form_durable_tail", + research_scope_ref: Some("research_scope_durable_tail"), + occurred_at_unix_ms: 40_000, + }) + .unwrap(); + let mut transaction = client.transaction().unwrap(); + assert!(matches!( + require_durable_ledger_tail( + &mut transaction, + &ledger, + &event(Some("consent_event_durable_tail_grant"), 40_000), + ), + Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope) + )); + require_durable_ledger_tail( + &mut transaction, + &complete, + &event(Some("consent_event_aaa_durable_tail_revoke"), 40_000), + ) + .expect("later-inserted same-millisecond revoke should be the durable tail"); + transaction.rollback().unwrap(); + client + .batch_execute(&format!("DROP SCHEMA IF EXISTS {SCHEMA} CASCADE;")) + .unwrap(); + } + + #[test] + fn equal_created_at_tail_is_ambiguous_and_fails_closed() { + let mut client = ready_schema_client(AMBIGUOUS_SCHEMA); + insert_tied_grant_and_revoke(&mut client); + let grant_only = research_ledger("consent_event_zzz_tied_grant", ConsentDecision::Granted); + let mut complete = grant_only.clone(); + complete + .record(ConsentEventInput { + event_ref: "consent_event_aaa_tied_revoke", + purpose: ConsentPurpose::ResearchContribution, + decision: ConsentDecision::Revoked, + consent_form_version_ref: "consent_form_durable_tail", + research_scope_ref: Some("research_scope_durable_tail"), + occurred_at_unix_ms: 40_000, + }) + .unwrap(); + let mut transaction = client.transaction().unwrap(); + for (ledger, causation) in [ + (&grant_only, "consent_event_zzz_tied_grant"), + (&complete, "consent_event_zzz_tied_grant"), + (&complete, "consent_event_aaa_tied_revoke"), + ] { + assert!(matches!( + require_durable_ledger_tail( + &mut transaction, + ledger, + &event(Some(causation), 40_000), + ), + Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope) + )); + } + transaction.rollback().unwrap(); + client + .batch_execute(&format!( + "DROP SCHEMA IF EXISTS {AMBIGUOUS_SCHEMA} CASCADE;" + )) + .unwrap(); + } +} diff --git a/tests/postgres_consent_outbox_error_boundary.rs b/tests/postgres_consent_outbox_error_boundary.rs new file mode 100644 index 00000000..f7c17a5e --- /dev/null +++ b/tests/postgres_consent_outbox_error_boundary.rs @@ -0,0 +1,104 @@ +//! Failure-boundary coverage for consent/outbox transactional composition. + +use postgres::{Client, IsolationLevel, NoTls}; +use psychometrics_commons_runtime::consent::{ + ConsentDecision, ConsentEventInput, ConsentLedger, ConsentPurpose, +}; +use psychometrics_commons_runtime::integration::IntegrationEvent; +use psychometrics_commons_runtime::postgres_consent::{ + apply_consent_migration, ConsentPersistenceError, +}; +use psychometrics_commons_runtime::postgres_consent_propagation::{ + persist_consent_ledger_with_outbox, ConsentOutboxPersistenceError, +}; +use psychometrics_commons_runtime::postgres_integration::apply_integration_migration; +use std::sync::{Mutex, MutexGuard}; + +const DIGEST: &str = "sha256:eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee"; +const TENANT_REF: &str = "tenant_consent_boundary_alpha"; +static CONSENT_BOUNDARY_TEST_LOCK: Mutex<()> = Mutex::new(()); + +fn boundary_guard() -> MutexGuard<'static, ()> { + CONSENT_BOUNDARY_TEST_LOCK + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} + +fn test_client() -> Client { + let connection = std::env::var("TEST_DATABASE_URL") + .expect("TEST_DATABASE_URL must identify the isolated CI PostgreSQL database"); + let mut client = Client::connect(&connection, NoTls) + .expect("isolated CI PostgreSQL database must be reachable"); + client + .batch_execute( + "CREATE SCHEMA IF NOT EXISTS consent_outbox_boundary_test;\ + SET search_path TO consent_outbox_boundary_test;\ + DROP TABLE IF EXISTS integration_delivery_attempt;\ + DROP TABLE IF EXISTS integration_inbox;\ + DROP TABLE IF EXISTS integration_outbox;\ + DROP TABLE IF EXISTS consent_event;\ + DROP TABLE IF EXISTS consent_ledger;", + ) + .unwrap(); + apply_integration_migration(&mut client).unwrap(); + apply_consent_migration(&mut client).unwrap(); + client +} + +fn ledger_and_event() -> (ConsentLedger, IntegrationEvent) { + let mut ledger = ConsentLedger::new("participant_consent_boundary_alpha").unwrap(); + ledger + .record(ConsentEventInput { + event_ref: "consent_event_boundary_alpha", + purpose: ConsentPurpose::ResearchContribution, + decision: ConsentDecision::Granted, + consent_form_version_ref: "consent_form_boundary_v1", + research_scope_ref: Some("research_scope_boundary_v1"), + occurred_at_unix_ms: 30_000, + }) + .unwrap(); + let event = IntegrationEvent::new( + "event_consent_boundary_alpha", + "consent.research.changed", + "v1", + "psychometrics_commons", + TENANT_REF, + ledger.participant_ref(), + 30_000, + "correlation_consent_boundary_alpha", + Some("consent_event_boundary_alpha"), + DIGEST, + ) + .unwrap(); + (ledger, event) +} + +#[test] +fn unsupported_consent_isolation_is_typed_and_enqueues_nothing() { + let _guard = boundary_guard(); + let mut client = test_client(); + let (ledger, event) = ledger_and_event(); + let mut transaction = client + .build_transaction() + .isolation_level(IsolationLevel::Serializable) + .start() + .unwrap(); + assert!(matches!( + persist_consent_ledger_with_outbox(&mut transaction, TENANT_REF, &ledger, &event, 3), + Err(ConsentOutboxPersistenceError::Consent( + ConsentPersistenceError::UnsupportedIsolationLevel + )) + )); + transaction.rollback().unwrap(); + + let consent_count: i64 = client + .query_one("SELECT count(*) FROM consent_event", &[]) + .unwrap() + .get(0); + let outbox_count: i64 = client + .query_one("SELECT count(*) FROM integration_outbox", &[]) + .unwrap() + .get(0); + assert_eq!(consent_count, 0); + assert_eq!(outbox_count, 0); +} diff --git a/tests/postgres_consent_outbox_latest_event.rs b/tests/postgres_consent_outbox_latest_event.rs new file mode 100644 index 00000000..62345f4d --- /dev/null +++ b/tests/postgres_consent_outbox_latest_event.rs @@ -0,0 +1,409 @@ +//! Consent propagation must be bound to the latest accepted consent change, not stale history. + +use postgres::{Client, NoTls}; +use psychometrics_commons_runtime::consent::{ + ConsentDecision, ConsentEventInput, ConsentLedger, ConsentPurpose, +}; +use psychometrics_commons_runtime::integration::IntegrationEvent; +use psychometrics_commons_runtime::postgres_consent::{ + apply_consent_migration, persist_consent_ledger, +}; +use psychometrics_commons_runtime::postgres_consent_propagation::{ + persist_consent_ledger_with_outbox, ConsentOutboxPersistenceError, +}; +use psychometrics_commons_runtime::postgres_integration::apply_integration_migration; + +const SCHEMA: &str = "consent_outbox_latest_event_test"; +const DATABASE_TEST_LOCK_KEY: i64 = 0x434F_4E53_4C41_5445; +const DIGEST: &str = "sha256:eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee"; +const TENANT_REF: &str = "tenant_consent_latest_alpha"; + +fn ready_client() -> Client { + let connection = std::env::var("TEST_DATABASE_URL") + .expect("TEST_DATABASE_URL must identify the isolated CI PostgreSQL database"); + let mut client = Client::connect(&connection, NoTls) + .expect("isolated CI PostgreSQL database must be reachable"); + client + .query_one("SELECT pg_advisory_lock($1)", &[&DATABASE_TEST_LOCK_KEY]) + .expect("shared PostgreSQL consent-latest-event lock should be acquired"); + client + .batch_execute(&format!( + "DROP SCHEMA IF EXISTS {SCHEMA} CASCADE; + CREATE SCHEMA {SCHEMA}; + SET search_path TO {SCHEMA};" + )) + .expect("isolated consent-latest-event schema should be reset"); + apply_integration_migration(&mut client).expect("integration migration should apply"); + apply_consent_migration(&mut client).expect("consent migration should apply"); + client +} + +fn grant_event_input() -> ConsentEventInput<'static> { + ConsentEventInput { + event_ref: "consent_event_grant_alpha", + purpose: ConsentPurpose::ResearchContribution, + decision: ConsentDecision::Granted, + consent_form_version_ref: "consent_form_latest_v1", + research_scope_ref: Some("research_scope_latest_alpha"), + occurred_at_unix_ms: 30_000, + } +} + +fn ledger_with_grant_only() -> ConsentLedger { + let mut ledger = ConsentLedger::new("participant_consent_latest_alpha").unwrap(); + ledger.record(grant_event_input()).unwrap(); + ledger +} + +fn ledger_with_revocation() -> ConsentLedger { + let mut ledger = ledger_with_grant_only(); + ledger + .record(ConsentEventInput { + event_ref: "consent_event_revoke_alpha", + purpose: ConsentPurpose::ResearchContribution, + decision: ConsentDecision::Revoked, + consent_form_version_ref: "consent_form_latest_v1", + research_scope_ref: Some("research_scope_latest_alpha"), + occurred_at_unix_ms: 31_000, + }) + .unwrap(); + ledger +} + +fn propagation_event( + event_ref: &str, + causation_ref: &str, + occurred_at_unix_ms: u64, +) -> IntegrationEvent { + IntegrationEvent::new( + event_ref, + "consent.research.changed", + "v1", + "psychometrics_commons", + TENANT_REF, + "participant_consent_latest_alpha", + occurred_at_unix_ms, + "correlation_consent_latest_alpha", + Some(causation_ref), + DIGEST, + ) + .unwrap() +} + +#[test] +fn stale_grant_cannot_be_propagated_after_a_later_revocation() { + let mut client = ready_client(); + let ledger = ledger_with_revocation(); + let stale_event = propagation_event( + "event_consent_stale_grant", + "consent_event_grant_alpha", + 30_000, + ); + + let mut transaction = client.transaction().unwrap(); + assert!(matches!( + persist_consent_ledger_with_outbox(&mut transaction, TENANT_REF, &ledger, &stale_event, 3,), + Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope) + )); + transaction.rollback().unwrap(); + + let consent_count: i64 = client + .query_one("SELECT count(*) FROM consent_event", &[]) + .unwrap() + .get(0); + let outbox_count: i64 = client + .query_one("SELECT count(*) FROM integration_outbox", &[]) + .unwrap() + .get(0); + assert_eq!(consent_count, 0); + assert_eq!(outbox_count, 0); + + client + .batch_execute(&format!( + "SET search_path TO public; + DROP SCHEMA IF EXISTS {SCHEMA} CASCADE;" + )) + .unwrap(); +} + +#[test] +fn durable_revoke_rejects_grant_only_snapshot_propagation() { + let mut client = ready_client(); + let mut persist_transaction = client.transaction().unwrap(); + persist_consent_ledger(&mut persist_transaction, &ledger_with_revocation()) + .expect("grant then revoke should persist as append-only consent evidence"); + persist_transaction.commit().unwrap(); + + let stale_event = propagation_event( + "event_consent_stale_grant_after_durable_revoke", + "consent_event_grant_alpha", + 30_000, + ); + let mut transaction = client.transaction().unwrap(); + assert!(matches!( + persist_consent_ledger_with_outbox( + &mut transaction, + TENANT_REF, + &ledger_with_grant_only(), + &stale_event, + 3, + ), + Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope) + )); + transaction.rollback().unwrap(); + + let consent_count: i64 = client + .query_one("SELECT count(*) FROM consent_event", &[]) + .unwrap() + .get(0); + let outbox_count: i64 = client + .query_one("SELECT count(*) FROM integration_outbox", &[]) + .unwrap() + .get(0); + assert_eq!(consent_count, 2); + assert_eq!(outbox_count, 0); + + client + .batch_execute(&format!( + "SET search_path TO public; + DROP SCHEMA IF EXISTS {SCHEMA} CASCADE;" + )) + .unwrap(); +} + +#[test] +fn same_millisecond_revoke_beats_lexicographic_grant_tail() { + let mut client = ready_client(); + let mut grant_ledger = ConsentLedger::new("participant_consent_latest_alpha").unwrap(); + grant_ledger + .record(ConsentEventInput { + event_ref: "consent_event_zzz_same_ms_grant", + purpose: ConsentPurpose::ResearchContribution, + decision: ConsentDecision::Granted, + consent_form_version_ref: "consent_form_latest_v1", + research_scope_ref: Some("research_scope_latest_alpha"), + occurred_at_unix_ms: 32_000, + }) + .unwrap(); + let mut persist_grant = client.transaction().unwrap(); + persist_consent_ledger(&mut persist_grant, &grant_ledger) + .expect("same-millisecond grant should persist first"); + persist_grant.commit().unwrap(); + + let mut revoked_ledger = grant_ledger.clone(); + revoked_ledger + .record(ConsentEventInput { + event_ref: "consent_event_aaa_same_ms_revoke", + purpose: ConsentPurpose::ResearchContribution, + decision: ConsentDecision::Revoked, + consent_form_version_ref: "consent_form_latest_v1", + research_scope_ref: Some("research_scope_latest_alpha"), + occurred_at_unix_ms: 32_000, + }) + .unwrap(); + let mut persist_revoke = client.transaction().unwrap(); + persist_consent_ledger(&mut persist_revoke, &revoked_ledger) + .expect("same-millisecond revoke should append after the grant"); + persist_revoke.commit().unwrap(); + + let stale_grant = propagation_event( + "event_consent_stale_same_ms_grant", + "consent_event_zzz_same_ms_grant", + 32_000, + ); + let mut stale_transaction = client.transaction().unwrap(); + assert!(matches!( + persist_consent_ledger_with_outbox( + &mut stale_transaction, + TENANT_REF, + &grant_ledger, + &stale_grant, + 3, + ), + Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope) + )); + stale_transaction.rollback().unwrap(); + + let revoke_event = propagation_event( + "event_consent_same_ms_revoke", + "consent_event_aaa_same_ms_revoke", + 32_000, + ); + let mut revoke_transaction = client.transaction().unwrap(); + persist_consent_ledger_with_outbox( + &mut revoke_transaction, + TENANT_REF, + &revoked_ledger, + &revoke_event, + 3, + ) + .expect("later-inserted same-millisecond revoke should bind the outbox"); + revoke_transaction.commit().unwrap(); + + let consent_count: i64 = client + .query_one("SELECT count(*) FROM consent_event", &[]) + .unwrap() + .get(0); + let outbox = client + .query_one( + "SELECT causation_ref, occurred_at_unix_ms FROM integration_outbox", + &[], + ) + .unwrap(); + let causation_ref: Option = outbox.get(0); + let occurred_at_unix_ms: i64 = outbox.get(1); + assert_eq!(consent_count, 2); + assert_eq!( + causation_ref.as_deref(), + Some("consent_event_aaa_same_ms_revoke") + ); + assert_eq!(occurred_at_unix_ms, 32_000); + + client + .batch_execute(&format!( + "SET search_path TO public; + DROP SCHEMA IF EXISTS {SCHEMA} CASCADE;" + )) + .unwrap(); +} + +#[test] +fn equal_created_at_cannot_bind_either_tied_envelope() { + let mut client = ready_client(); + let research_scope_ref = Some("research_scope_latest_alpha"); + client + .execute( + "INSERT INTO consent_ledger (participant_ref) VALUES ($1)", + &[&"participant_consent_latest_alpha"], + ) + .unwrap(); + client + .execute( + "INSERT INTO consent_event (\ + participant_ref, event_ref, consent_purpose, consent_decision, \ + consent_form_version_ref, research_scope_ref, occurred_at_unix_ms, created_at\ + ) VALUES ($1, $2, 'research_contribution', 'granted', $3, $4, 33000, \ + TIMESTAMPTZ '2026-08-16 15:00:00+00')", + &[ + &"participant_consent_latest_alpha", + &"consent_event_zzz_tied_grant", + &"consent_form_latest_v1", + &research_scope_ref, + ], + ) + .unwrap(); + client + .execute( + "INSERT INTO consent_event (\ + participant_ref, event_ref, consent_purpose, consent_decision, \ + consent_form_version_ref, research_scope_ref, occurred_at_unix_ms, created_at\ + ) VALUES ($1, $2, 'research_contribution', 'revoked', $3, $4, 33000, \ + TIMESTAMPTZ '2026-08-16 15:00:00+00')", + &[ + &"participant_consent_latest_alpha", + &"consent_event_aaa_tied_revoke", + &"consent_form_latest_v1", + &research_scope_ref, + ], + ) + .unwrap(); + + let mut complete = ConsentLedger::new("participant_consent_latest_alpha").unwrap(); + complete + .record(ConsentEventInput { + event_ref: "consent_event_zzz_tied_grant", + purpose: ConsentPurpose::ResearchContribution, + decision: ConsentDecision::Granted, + consent_form_version_ref: "consent_form_latest_v1", + research_scope_ref: Some("research_scope_latest_alpha"), + occurred_at_unix_ms: 33_000, + }) + .unwrap(); + complete + .record(ConsentEventInput { + event_ref: "consent_event_aaa_tied_revoke", + purpose: ConsentPurpose::ResearchContribution, + decision: ConsentDecision::Revoked, + consent_form_version_ref: "consent_form_latest_v1", + research_scope_ref: Some("research_scope_latest_alpha"), + occurred_at_unix_ms: 33_000, + }) + .unwrap(); + let stale_grant = propagation_event( + "event_consent_tied_grant", + "consent_event_zzz_tied_grant", + 33_000, + ); + let tied_revoke = propagation_event( + "event_consent_tied_revoke", + "consent_event_aaa_tied_revoke", + 33_000, + ); + let mut transaction = client.transaction().unwrap(); + for event in [&stale_grant, &tied_revoke] { + assert!(matches!( + persist_consent_ledger_with_outbox(&mut transaction, TENANT_REF, &complete, event, 3,), + Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope) + )); + } + transaction.rollback().unwrap(); + + let outbox_count: i64 = client + .query_one("SELECT count(*) FROM integration_outbox", &[]) + .unwrap() + .get(0); + assert_eq!(outbox_count, 0); + + client + .batch_execute(&format!( + "SET search_path TO public; + DROP SCHEMA IF EXISTS {SCHEMA} CASCADE;" + )) + .unwrap(); +} + +#[test] +fn latest_revocation_can_be_persisted_with_its_propagation_event() { + let mut client = ready_client(); + let ledger = ledger_with_revocation(); + let latest_event = propagation_event( + "event_consent_latest_revocation", + "consent_event_revoke_alpha", + 31_000, + ); + + let mut transaction = client.transaction().unwrap(); + persist_consent_ledger_with_outbox(&mut transaction, TENANT_REF, &ledger, &latest_event, 3) + .expect("latest consent change should persist with its bound outbox event"); + transaction.commit().unwrap(); + + let consent_count: i64 = client + .query_one("SELECT count(*) FROM consent_event", &[]) + .unwrap() + .get(0); + let outbox = client + .query_one( + "SELECT source_ref, tenant_ref, subject_ref, causation_ref, occurred_at_unix_ms \ + FROM integration_outbox", + &[], + ) + .unwrap(); + let source_ref: String = outbox.get(0); + let tenant_ref: String = outbox.get(1); + let subject_ref: String = outbox.get(2); + let causation_ref: Option = outbox.get(3); + let occurred_at_unix_ms: i64 = outbox.get(4); + assert_eq!(consent_count, 2); + assert_eq!(source_ref, "psychometrics_commons"); + assert_eq!(tenant_ref, TENANT_REF); + assert_eq!(subject_ref, "participant_consent_latest_alpha"); + assert_eq!(causation_ref.as_deref(), Some("consent_event_revoke_alpha")); + assert_eq!(occurred_at_unix_ms, 31_000); + + client + .batch_execute(&format!( + "SET search_path TO public; + DROP SCHEMA IF EXISTS {SCHEMA} CASCADE;" + )) + .unwrap(); +} diff --git a/tests/postgres_consent_outbox_transaction.rs b/tests/postgres_consent_outbox_transaction.rs new file mode 100644 index 00000000..5ec79369 --- /dev/null +++ b/tests/postgres_consent_outbox_transaction.rs @@ -0,0 +1,317 @@ +//! Real `PostgreSQL` contract for consent persistence with transactional outbox evidence. + +use postgres::{Client, NoTls}; +use psychometrics_commons_runtime::consent::{ + ConsentDecision, ConsentEventInput, ConsentLedger, ConsentPurpose, +}; +use psychometrics_commons_runtime::integration::IntegrationEvent; +use psychometrics_commons_runtime::postgres_consent::{ + apply_consent_migration, persist_consent_ledger, ConsentPersistenceDisposition, + ConsentPersistenceError, +}; +use psychometrics_commons_runtime::postgres_consent_propagation::{ + persist_consent_ledger_with_outbox, ConsentOutboxPersistenceError, +}; +use psychometrics_commons_runtime::postgres_integration::{ + apply_integration_migration, enqueue_outbox_event, PersistenceDisposition, PersistenceError, +}; +use std::error::Error; +use std::sync::{Mutex, MutexGuard}; + +const DIGEST_A: &str = "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; +const DIGEST_B: &str = "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"; +const TENANT_REF: &str = "tenant_consent_outbox_alpha"; + +static CONSENT_OUTBOX_TEST_LOCK: Mutex<()> = Mutex::new(()); + +fn consent_outbox_guard() -> MutexGuard<'static, ()> { + CONSENT_OUTBOX_TEST_LOCK + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} + +fn test_client() -> Client { + let connection = std::env::var("TEST_DATABASE_URL") + .expect("TEST_DATABASE_URL must identify the isolated CI PostgreSQL database"); + let mut client = Client::connect(&connection, NoTls) + .expect("isolated CI PostgreSQL database must be reachable"); + client + .batch_execute( + "CREATE SCHEMA IF NOT EXISTS consent_outbox_transaction_test;\ + SET search_path TO consent_outbox_transaction_test;", + ) + .unwrap(); + client +} + +fn reset_and_migrate(client: &mut Client) { + client + .batch_execute( + "DROP TABLE IF EXISTS integration_delivery_attempt;\ + DROP TABLE IF EXISTS integration_inbox;\ + DROP TABLE IF EXISTS integration_outbox;\ + DROP TABLE IF EXISTS consent_event;\ + DROP TABLE IF EXISTS consent_ledger;", + ) + .unwrap(); + apply_integration_migration(client).unwrap(); + apply_consent_migration(client).unwrap(); +} + +fn research_ledger(event_ref: &str, decision: ConsentDecision) -> ConsentLedger { + let mut ledger = ConsentLedger::new("participant_consent_outbox_alpha").unwrap(); + ledger + .record(ConsentEventInput { + event_ref, + purpose: ConsentPurpose::ResearchContribution, + decision, + consent_form_version_ref: "consent_form_research_v1", + research_scope_ref: Some("research_scope_publication_v1"), + occurred_at_unix_ms: 10_000, + }) + .unwrap(); + ledger +} + +fn propagation_event( + event_ref: &str, + consent_event_ref: &str, + tenant_ref: &str, + subject_ref: &str, + digest: &str, +) -> IntegrationEvent { + IntegrationEvent::new( + event_ref, + "consent.research.changed", + "v1", + "psychometrics_commons", + tenant_ref, + subject_ref, + 10_000, + "correlation_consent_outbox_alpha", + Some(consent_event_ref), + digest, + ) + .unwrap() +} + +#[test] +fn consent_and_outbox_commit_and_replay_together() { + let _guard = consent_outbox_guard(); + let mut client = test_client(); + reset_and_migrate(&mut client); + let ledger = research_ledger("consent_event_research_grant", ConsentDecision::Granted); + let event = propagation_event( + "event_consent_research_grant", + "consent_event_research_grant", + TENANT_REF, + ledger.participant_ref(), + DIGEST_A, + ); + + let mut transaction = client.transaction().unwrap(); + let inserted = + persist_consent_ledger_with_outbox(&mut transaction, TENANT_REF, &ledger, &event, 3) + .unwrap(); + assert_eq!(inserted.consent(), ConsentPersistenceDisposition::Inserted); + assert_eq!(inserted.outbox(), PersistenceDisposition::Inserted); + transaction.commit().unwrap(); + + let mut transaction = client.transaction().unwrap(); + let duplicate = + persist_consent_ledger_with_outbox(&mut transaction, TENANT_REF, &ledger, &event, 3) + .unwrap(); + assert_eq!( + duplicate.consent(), + ConsentPersistenceDisposition::Duplicate + ); + assert_eq!(duplicate.outbox(), PersistenceDisposition::Duplicate); + transaction.commit().unwrap(); + + let consent_count: i64 = client + .query_one("SELECT count(*) FROM consent_event", &[]) + .unwrap() + .get(0); + let outbox_count: i64 = client + .query_one("SELECT count(*) FROM integration_outbox", &[]) + .unwrap() + .get(0); + assert_eq!(consent_count, 1); + assert_eq!(outbox_count, 1); +} + +#[test] +fn existing_consent_can_enqueue_a_new_bound_outbox() { + let _guard = consent_outbox_guard(); + let mut client = test_client(); + reset_and_migrate(&mut client); + let ledger = research_ledger("consent_event_legacy_grant", ConsentDecision::Granted); + let mut persist_transaction = client.transaction().unwrap(); + assert_eq!( + persist_consent_ledger(&mut persist_transaction, &ledger).unwrap(), + ConsentPersistenceDisposition::Inserted + ); + persist_transaction.commit().unwrap(); + + let event = propagation_event( + "event_consent_legacy_grant", + "consent_event_legacy_grant", + TENANT_REF, + ledger.participant_ref(), + DIGEST_A, + ); + let mut transaction = client.transaction().unwrap(); + let mixed = + persist_consent_ledger_with_outbox(&mut transaction, TENANT_REF, &ledger, &event, 3) + .unwrap(); + assert_eq!(mixed.consent(), ConsentPersistenceDisposition::Duplicate); + assert_eq!(mixed.outbox(), PersistenceDisposition::Inserted); + transaction.commit().unwrap(); + + let consent_count: i64 = client + .query_one("SELECT count(*) FROM consent_event", &[]) + .unwrap() + .get(0); + let outbox_count: i64 = client + .query_one("SELECT count(*) FROM integration_outbox", &[]) + .unwrap() + .get(0); + assert_eq!(consent_count, 1); + assert_eq!(outbox_count, 1); +} + +#[test] +fn propagation_envelope_must_bind_tenant_participant_and_exact_consent_event() { + let _guard = consent_outbox_guard(); + let mut client = test_client(); + reset_and_migrate(&mut client); + let ledger = research_ledger("consent_event_envelope_alpha", ConsentDecision::Revoked); + let wrong_tenant = propagation_event( + "event_consent_wrong_tenant", + "consent_event_envelope_alpha", + "tenant_consent_outbox_other", + ledger.participant_ref(), + DIGEST_A, + ); + + let mut transaction = client.transaction().unwrap(); + assert!(matches!( + persist_consent_ledger_with_outbox(&mut transaction, TENANT_REF, &ledger, &wrong_tenant, 3,), + Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope) + )); + transaction.rollback().unwrap(); + + let wrong_subject = propagation_event( + "event_consent_wrong_subject", + "consent_event_envelope_alpha", + TENANT_REF, + "participant_other", + DIGEST_A, + ); + + let mut transaction = client.transaction().unwrap(); + assert!(matches!( + persist_consent_ledger_with_outbox( + &mut transaction, + TENANT_REF, + &ledger, + &wrong_subject, + 3, + ), + Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope) + )); + transaction.rollback().unwrap(); + + let wrong_causation = propagation_event( + "event_consent_wrong_causation", + "consent_event_other", + TENANT_REF, + ledger.participant_ref(), + DIGEST_A, + ); + let mut transaction = client.transaction().unwrap(); + assert!(matches!( + persist_consent_ledger_with_outbox( + &mut transaction, + TENANT_REF, + &ledger, + &wrong_causation, + 3, + ), + Err(ConsentOutboxPersistenceError::InvalidPropagationEnvelope) + )); + transaction.rollback().unwrap(); + + let consent_count: i64 = client + .query_one("SELECT count(*) FROM consent_event", &[]) + .unwrap() + .get(0); + assert_eq!(consent_count, 0); +} + +#[test] +fn late_outbox_conflict_rolls_back_new_consent_evidence() { + let _guard = consent_outbox_guard(); + let mut client = test_client(); + reset_and_migrate(&mut client); + let ledger = research_ledger("consent_event_conflict_alpha", ConsentDecision::Granted); + let existing_event = propagation_event( + "event_consent_conflict_alpha", + "consent_event_conflict_alpha", + TENANT_REF, + ledger.participant_ref(), + DIGEST_A, + ); + assert_eq!( + enqueue_outbox_event(&mut client, &existing_event, 3).unwrap(), + PersistenceDisposition::Inserted + ); + let conflicting_event = propagation_event( + "event_consent_conflict_alpha", + "consent_event_conflict_alpha", + TENANT_REF, + ledger.participant_ref(), + DIGEST_B, + ); + + let mut transaction = client.transaction().unwrap(); + assert!(matches!( + persist_consent_ledger_with_outbox( + &mut transaction, + TENANT_REF, + &ledger, + &conflicting_event, + 3, + ), + Err(ConsentOutboxPersistenceError::Outbox( + PersistenceError::ConflictingReplay + )) + )); + transaction.rollback().unwrap(); + + let consent_count: i64 = client + .query_one("SELECT count(*) FROM consent_event", &[]) + .unwrap() + .get(0); + let outbox_count: i64 = client + .query_one("SELECT count(*) FROM integration_outbox", &[]) + .unwrap() + .get(0); + assert_eq!(consent_count, 0); + assert_eq!(outbox_count, 1); +} + +#[test] +fn consent_outbox_errors_retain_typed_sources() { + let errors = [ + ConsentOutboxPersistenceError::Consent(ConsentPersistenceError::InvalidReference), + ConsentOutboxPersistenceError::Outbox(PersistenceError::InvalidReference), + ]; + for error in errors { + assert!(!error.to_string().is_empty()); + assert!(error.source().is_some()); + } + let envelope = ConsentOutboxPersistenceError::InvalidPropagationEnvelope; + assert!(!envelope.to_string().is_empty()); + assert!(envelope.source().is_none()); +} diff --git a/tests/postgres_recovery_invariants.rs b/tests/postgres_recovery_invariants.rs index e8af1d61..b97d15f8 100644 --- a/tests/postgres_recovery_invariants.rs +++ b/tests/postgres_recovery_invariants.rs @@ -84,11 +84,11 @@ fn seed_recovery_critical_state(client: &mut Client) { INSERT INTO {SOURCE_SCHEMA}.integration_consumption ( consumer_ref, source_ref, tenant_ref, source_event_ref, consumption_ref, side_effect_ref, consumption_state, fencing_token, latest_event_at_unix_ms, - claim_expires_at_unix_ms, completion_evidence_ref, cause_code + claim_expires_at_unix_ms, claim_deadline_at, completion_evidence_ref, cause_code ) VALUES ( 'consumer_recovery_alpha', 'dependency_recovery_alpha', 'tenant_recovery_alpha', 'event_dependency_alpha', 'consumption_recovery_alpha', 'effect_recovery_alpha', - 'processing', 7, 12000, 13000, NULL, NULL + 'processing', 7, 12000, 13000, clock_timestamp() + INTERVAL '1 hour', NULL, NULL ); INSERT INTO {SOURCE_SCHEMA}.response_snapshot ( snapshot_ref, session_ref, event_count, last_sequence @@ -150,7 +150,8 @@ fn assert_restored_evidence(client: &mut Client) { let restored_consumption = client .query_one( &format!( - "SELECT consumption_state, fencing_token, claim_expires_at_unix_ms + "SELECT consumption_state, fencing_token, claim_expires_at_unix_ms, + claim_deadline_at IS NOT NULL FROM {RESTORED_SCHEMA}.integration_consumption WHERE consumption_ref = 'consumption_recovery_alpha'" ), @@ -160,6 +161,29 @@ fn assert_restored_evidence(client: &mut Client) { assert_eq!(restored_consumption.get::<_, String>(0), "processing"); assert_eq!(restored_consumption.get::<_, i64>(1), 7); assert_eq!(restored_consumption.get::<_, Option>(2), Some(13000)); + assert!(restored_consumption.get::<_, bool>(3)); + + let claim_deadline_matches_source: bool = client + .query_one( + &format!( + "SELECT source_row.claim_deadline_at = restored_row.claim_deadline_at + FROM {SOURCE_SCHEMA}.integration_consumption AS source_row + JOIN {RESTORED_SCHEMA}.integration_consumption AS restored_row + ON restored_row.consumer_ref = source_row.consumer_ref + AND restored_row.source_ref = source_row.source_ref + AND restored_row.tenant_ref = source_row.tenant_ref + AND restored_row.source_event_ref = source_row.source_event_ref + AND restored_row.consumption_ref = source_row.consumption_ref + WHERE source_row.consumption_ref = 'consumption_recovery_alpha'" + ), + &[], + ) + .expect("restored claim-deadline evidence should remain comparable") + .get(0); + assert!( + claim_deadline_matches_source, + "restore must preserve the exact database-authoritative processing claim deadline" + ); let restored_snapshot = client .query_one(