diff --git a/CHANGELOG.md b/CHANGELOG.md index b2f801dd..ec2a9479 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ All notable product and architecture changes are recorded here. Releases use imm ## Unreleased ### Added +- PostgreSQL operational-backlog probes classify aggregate outbox, inbox-consumption, data-rights, and scoring-job counts and ages against an explicit caller policy. They expose no payloads, tenant identities, worker identities, or restricted linkage values, and they invent no universal SLO defaults. `classify_postgres_operational_backlog` combines the three families so a healthy outbox cannot hide a stalled scoring queue. Call `apply_backlog_health_index_migration` after the integration, inbox-consumption, and data-rights migrations, and use `apply_scoring_job_migration` so scoring-job readiness indexes from `migrations/0021_scoring_job_health_indexes.sql` are created through the product apply path. - Scoring-job cancel and lease-expiry fallback classification lock the current row until the caller transaction ends, so concurrent workers cannot rewrite terminal or unleased evidence. - PostgreSQL operational-store readiness probe classifies the supported major version and write-readiness, and fails closed when a caller-declared required relation is missing. - PostgreSQL scoring-job cancellation: queued, leased, or retry-scheduled work becomes cancelled without transferring a fence, exact replay is idempotent, and completed or quarantined evidence cannot be rewritten. diff --git a/docs/OPERABILITY.md b/docs/OPERABILITY.md index b729f0c5..6955683d 100644 --- a/docs/OPERABILITY.md +++ b/docs/OPERABILITY.md @@ -39,7 +39,7 @@ The implementation must distinguish at least: - **liveness** — process can make progress / is not irrecoverably wedged; - **readiness** — mandatory dependencies for the selected profile are available enough to accept new work safely; - **capability health** — optional/independent capabilities such as authenticated linking, scoring, AI narrative, research registration, or temporal analysis; -- **backlog health** — durable work is within measured operating bounds and not silently stalled; +- **backlog health** — durable work is within measured operating bounds and not silently stalled. When a PostgreSQL operational-backlog probe is present, it classifies aggregate outbox, inbox-consumption, data-rights, and scoring-job counts/ages against caller-supplied policy only and must not expose payloads, tenant identities, worker identities, or restricted linkage values. Apply `migrations/0020_backlog_health_indexes.sql` through `apply_backlog_health_index_migration` after the owning table migrations, and apply scoring-job readiness indexes through `apply_scoring_job_migration`, so readiness probes stay bounded as terminal history grows; - **data integrity health** — migrations/schema/digests/reconciliation do not indicate incompatible or corrupt state. Readiness must not fail solely because an optional capability is unavailable if the selected operation can safely proceed without it. Conversely, a process can be live while not ready to accept new state-changing requests. @@ -161,6 +161,7 @@ The deployed profile must have executable or operator-tested runbooks for at lea - outbox publication backlog; - inbox processing backlog or poison message; - scoring dependency outage and job reconciliation; +- scoring-job queue, lease, retry, or quarantine backlog; - Keyverse/JWKS/federation outage; - account-link conflict/adjudication; - optional AI provider/orchestrator outage and deterministic fallback verification; @@ -229,6 +230,10 @@ Never collapse these maturity levels. SOC 2/CSAP readiness work may map evidence ## 15. References +Beyer, B., Jones, C., Petoff, J., & Murphy, N. R. (Eds.). (2016). *Site reliability engineering: How Google runs production systems*. O'Reilly Media. + International Organization for Standardization & International Electrotechnical Commission. (2023). *ISO/IEC 25010:2023 Systems and software engineering—Systems and software Quality Requirements and Evaluation (SQuaRE)—Product quality model*. +National Institute of Standards and Technology. (2020). *Security and privacy controls for information systems and organizations* (NIST SP 800-53 Rev. 5). https://doi.org/10.6028/NIST.SP.800-53r5 + National Institute of Standards and Technology. (2022). *Secure Software Development Framework (SSDF) Version 1.1: Recommendations for mitigating the risk of software vulnerabilities* (NIST SP 800-218). https://doi.org/10.6028/NIST.SP.800-218 diff --git a/docs/TRACEABILITY.md b/docs/TRACEABILITY.md index 72bc73c2..1faad4ac 100644 --- a/docs/TRACEABILITY.md +++ b/docs/TRACEABILITY.md @@ -40,7 +40,7 @@ An active PR, architecture document, conversation decision, or scheduler plan is | 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 | | Research release manifests | PRD §5 | TRD §15 | ADR-0007, ADR-0010 | Target; semantic-data-portal is External dependency | | Durable outbox/inbox delivery semantics | PRD §7, §9 | TRD §19–20 | ADR-0014, ADR-0015 | **Partially implemented**: domain contracts in `src/integration.rs`; PostgreSQL 18 outbox/inbox identity, delivery-attempt persistence, and inbox consumption distinct from receipt; live side-effect execution remains Target | -| Operation-scoped capability health | PRD §7, §13 | `docs/OPERABILITY.md` §3–4; Deployment/Operations | ADR-0011, ADR-0017 | **Implemented** domain health/readiness contract in `src/health.rs` plus `src/postgres_health.rs` PostgreSQL major/write-readiness and caller-declared relation presence; HTTP probes, measured thresholds, and deployment evidence remain Target | +| Operation-scoped capability health | PRD §7, §13 | `docs/OPERABILITY.md` §3–4; Deployment/Operations | ADR-0011, ADR-0017 | **Implemented** domain health/readiness contract in `src/health.rs` plus `src/postgres_health.rs` PostgreSQL major/write-readiness and caller-declared relation presence; **Active PR** #82 observes aggregate outbox, inbox-consumption, and data-rights backlog counts/ages and applies partial readiness indexes through `apply_backlog_health_index_migration`; **Active PR** #113 observes queued/leased/retry-scheduled/quarantined scoring-job counts and created-at age through `apply_scoring_job_migration`; HTTP probes remain #111; measured thresholds and deployment-profile evidence remain Target | | Korean/English exact locale versions | PRD §3.1, §9.9 | TRD §28; instrument release + locale governance | ADR-0013, ADR-0019 | **Partially implemented**: locale is pinned/validated by `src/instrument.rs`; actual English/Korean form content, rights, translation, invariance and serving are Target | | WCAG 2.2 AA supported reference client | PRD §9.10 | TRD §27; Quality Attributes | ADR-0002, ADR-0013 | Target; no reference client implementation on evaluated main | | EMA/ESM longitudinal flow | PRD §4 | TRD §16; UML longitudinal sequence; logical ERD extension | ADR-0008 | External Gyeot/TEPP dependencies + Target Commons enrollment/normalized-ingestion/orchestration adapter | @@ -79,7 +79,7 @@ An active PR, architecture document, conversation decision, or scheduler plan is | 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 | -| Unknown/stalled backlog or unknown/incompatible integrity blocks new state-changing work | Operability §3, §6, §8 | **Implemented** domain contract in `src/health.rs`; `src/postgres_health.rs` fails closed on unsupported/read-only PostgreSQL or a missing required relation | persistence/job backlog metrics, stronger schema probes, alerting, and failure-injection evidence | +| Unknown/stalled backlog or unknown/incompatible integrity blocks new state-changing work | Operability §3, §6, §8 | **Implemented** domain contract in `src/health.rs`; `src/postgres_health.rs` fails closed on unsupported/read-only PostgreSQL or a missing required relation; **Active PR** #82 classifies durable integration and data-rights backlog evidence against an explicit caller policy; **Active PR** #113 classifies queued/leased/retry-scheduled/quarantined scoring-job evidence the same way | HTTP probe wiring on #111, measured deployment-profile thresholds, alerting, and failure-injection evidence | | No operational IDs in public research release | TRD §14–15; Research Governance | architecture policy | release fixture/static/runtime leakage tests | | AI optional; deterministic core remains | PRD §9.5; TRD §17; AI Governance | architecture policy | narrative fallback end-to-end test | | AI cannot mutate numeric scientific result | AI Governance; ADR-0009, ADR-0018 | architecture policy | product adapter/adversarial mutation tests | @@ -104,7 +104,7 @@ src/lib.rs ├── participant.rs # stable participant identity + issuer-scoped optional Keyverse account link ├── postgres_consent.rs # PostgreSQL purpose-specific consent ledger persistence ├── postgres_data_rights.rs # PostgreSQL data-rights request and local propagation persistence -├── postgres_health.rs # PostgreSQL major/write-readiness and relation-integrity probe +├── postgres_health.rs # PostgreSQL major/write-readiness, relation-integrity, and Active PR operational-backlog probes including scoring-job queues ├── postgres_inbox_consumption.rs # PostgreSQL inbox consumption distinct from receipt ├── postgres_instrument_release.rs # PostgreSQL locale-specific instrument-release persistence ├── postgres_integration.rs # PostgreSQL integration evidence/delivery-attempt persistence adapter @@ -132,7 +132,7 @@ Still-Target logical modules/adapters include remaining product aggregate persis ### Active implementation work that is not protected-main truth -**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** #82 PostgreSQL operational-backlog observation is not protected-main truth until an unchanged reviewed/check-clean head is integrated. Aggregate outbox, inbox-consumption, and data-rights counts/ages are classified only against an explicit caller policy. Partial readiness indexes ship in `migrations/0020_backlog_health_indexes.sql` and must be applied through `apply_backlog_health_index_migration` after the owning table migrations; a private test `include_str!` is not a product apply path. **Active PR** #113 adds the same content-free observation for queued, leased, retry-scheduled, and quarantined scoring work, with readiness indexes applied through `apply_scoring_job_migration`. HTTP probes remain #111. Deployment-profile thresholds remain outside this slice. ## 5. ADR traceability by concern diff --git a/migrations/0020_backlog_health_indexes.sql b/migrations/0020_backlog_health_indexes.sql new file mode 100644 index 00000000..3ba6e210 --- /dev/null +++ b/migrations/0020_backlog_health_indexes.sql @@ -0,0 +1,21 @@ +-- Keep readiness probes bounded as durable terminal history grows. +-- data_rights_propagation_state already has the state/time index created by 0003. +CREATE INDEX IF NOT EXISTS integration_outbox_pending_health_idx + ON integration_outbox (latest_event_at_unix_ms) + WHERE current_state = 'pending'; + +CREATE INDEX IF NOT EXISTS integration_outbox_quarantined_health_idx + ON integration_outbox (latest_event_at_unix_ms) + WHERE current_state = 'quarantined'; + +CREATE INDEX IF NOT EXISTS integration_consumption_active_health_idx + ON integration_consumption (latest_event_at_unix_ms) + WHERE consumption_state IN ('pending', 'processing'); + +CREATE INDEX IF NOT EXISTS integration_consumption_quarantined_health_idx + ON integration_consumption (latest_event_at_unix_ms) + WHERE consumption_state = 'quarantined'; + +CREATE INDEX IF NOT EXISTS data_rights_request_active_health_idx + ON data_rights_request_state (requested_at_unix_ms) + WHERE current_state IN ('requested', 'identity_verified', 'processing'); diff --git a/migrations/0021_scoring_job_health_indexes.sql b/migrations/0021_scoring_job_health_indexes.sql new file mode 100644 index 00000000..bfbd44e3 --- /dev/null +++ b/migrations/0021_scoring_job_health_indexes.sql @@ -0,0 +1,8 @@ +-- Keep scoring-job readiness probes bounded as terminal history grows. +CREATE INDEX IF NOT EXISTS scoring_job_state_active_health_idx + ON scoring_job_state (created_at) + WHERE scoring_state IN ('queued', 'leased', 'retry_scheduled'); + +CREATE INDEX IF NOT EXISTS scoring_job_state_quarantined_health_idx + ON scoring_job_state (created_at) + WHERE scoring_state = 'quarantined'; diff --git a/src/health.rs b/src/health.rs index c2268606..546033ac 100644 --- a/src/health.rs +++ b/src/health.rs @@ -22,6 +22,23 @@ pub enum BacklogHealth { } impl BacklogHealth { + /// Combine two independently measured durable-work signals for one readiness decision. + /// + /// A known stalled backlog takes precedence because accepting more state-changing work is + /// already unsafe. Otherwise an unknown signal takes precedence over a healthy signal so a + /// missing observation cannot be masked. The result is within bounds only when both signals + /// are within their caller-supplied operating bounds. + #[must_use] + pub const fn combine(self, other: Self) -> Self { + if matches!(self, Self::Stalled) || matches!(other, Self::Stalled) { + Self::Stalled + } else if matches!(self, Self::Unknown) || matches!(other, Self::Unknown) { + Self::Unknown + } else { + Self::WithinBounds + } + } + const fn accepts_new_work(self) -> bool { matches!(self, Self::WithinBounds) } diff --git a/src/postgres_health.rs b/src/postgres_health.rs index 0abe56c9..e241799d 100644 --- a/src/postgres_health.rs +++ b/src/postgres_health.rs @@ -5,12 +5,19 @@ //! can safely accept product-owned state changes. It does not own credentials, //! connection pooling, migrations, backup, or recovery. -use crate::health::{CapabilityHealth, CapabilityState, DataIntegrityHealth, HealthContractError}; +use crate::health::{ + BacklogHealth, CapabilityHealth, CapabilityState, DataIntegrityHealth, HealthContractError, +}; use postgres::GenericClient; +use std::error::Error; +use std::fmt::{Display, Formatter}; /// Initial supported `PostgreSQL` server major version from ADR-0015. pub const SUPPORTED_POSTGRES_MAJOR: i32 = 18; +const BACKLOG_HEALTH_INDEX_MIGRATION: &str = + include_str!("../migrations/0020_backlog_health_indexes.sql"); + /// Capability identity used by operation-scoped runtime readiness. pub const POSTGRES_OPERATIONAL_STORE_CAPABILITY_REF: &str = "postgres_operational_store"; @@ -78,6 +85,217 @@ impl PostgresRuntimeHealth { } } +/// Operator-supplied bounds used to classify durable integration backlog evidence. +/// +/// The repository deliberately provides no universal defaults. Hosted, community, +/// and enterprise operators must derive these values from their measured workload, +/// topology, alert policy, and recovery evidence rather than treating architecture +/// prose as an SLO commitment. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct IntegrationBacklogPolicy { + /// Largest pending outbox population still accepted for new state-changing work. + pub max_pending_outbox_count: u64, + /// Largest age in milliseconds of the oldest pending outbox event. + pub max_pending_outbox_age_ms: u64, + /// Largest quarantined outbox population still accepted under this operator policy. + pub max_quarantined_outbox_count: u64, + /// Largest pending-or-processing inbox-consumption population still accepted. + pub max_active_consumption_count: u64, + /// Largest age in milliseconds of the oldest pending-or-processing consumption. + pub max_active_consumption_age_ms: u64, + /// Largest quarantined inbox-consumption population still accepted. + pub max_quarantined_consumption_count: u64, +} + +/// Aggregate, content-free evidence read from the durable integration tables. +/// +/// Only counts and server-authoritative event times are exposed. The probe never +/// returns assessment responses, event payloads, tenant identities, subjects, or +/// restricted research linkage values. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct PostgresIntegrationBacklogEvidence { + pending_outbox_count: u64, + quarantined_outbox_count: u64, + active_consumption_count: u64, + quarantined_consumption_count: u64, + oldest_pending_outbox_event_at_unix_ms: Option, + oldest_active_consumption_event_at_unix_ms: Option, +} + +impl PostgresIntegrationBacklogEvidence { + /// Return the number of durable outbox events still awaiting delivery. + #[must_use] + pub const fn pending_outbox_count(self) -> u64 { + self.pending_outbox_count + } + + /// Return the number of outbox events quarantined for operator attention. + #[must_use] + pub const fn quarantined_outbox_count(self) -> u64 { + self.quarantined_outbox_count + } + + /// Return pending plus processing inbox side-effect consumptions. + #[must_use] + pub const fn active_consumption_count(self) -> u64 { + self.active_consumption_count + } + + /// Return the number of quarantined inbox side-effect consumptions. + #[must_use] + pub const fn quarantined_consumption_count(self) -> u64 { + self.quarantined_consumption_count + } + + /// Return the oldest pending outbox event time, or `None` when no event is pending. + #[must_use] + pub const fn oldest_pending_outbox_event_at_unix_ms(self) -> Option { + self.oldest_pending_outbox_event_at_unix_ms + } + + /// Return the oldest pending/processing inbox-consumption event time, if any. + #[must_use] + pub const fn oldest_active_consumption_event_at_unix_ms(self) -> Option { + self.oldest_active_consumption_event_at_unix_ms + } +} + +/// Operator-supplied bounds for participant data-rights request and propagation backlogs. +/// +/// Data-rights timeliness is a deployment and governance policy, not a hard-coded product +/// constant. Callers provide evidence-backed limits for their named deployment profile. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct DataRightsBacklogPolicy { + /// Largest non-terminal participant request population still accepted. + pub max_active_request_count: u64, + /// Largest age in milliseconds of the oldest non-terminal request. + pub max_active_request_age_ms: u64, + /// Largest pending downstream propagation population still accepted. + pub max_pending_propagation_count: u64, + /// Largest age in milliseconds of the oldest pending downstream propagation. + pub max_pending_propagation_age_ms: u64, + /// Largest quarantined downstream propagation population still accepted. + pub max_quarantined_propagation_count: u64, +} + +/// Aggregate participant-rights backlog evidence without participant or tenant identifiers. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct PostgresDataRightsBacklogEvidence { + active_request_count: u64, + pending_propagation_count: u64, + quarantined_propagation_count: u64, + oldest_active_request_at_unix_ms: Option, + oldest_pending_propagation_event_at_unix_ms: Option, +} + +impl PostgresDataRightsBacklogEvidence { + /// Return the number of requested, identity-verified, or processing requests. + #[must_use] + pub const fn active_request_count(self) -> u64 { + self.active_request_count + } + + /// Return the number of downstream propagation records awaiting delivery. + #[must_use] + pub const fn pending_propagation_count(self) -> u64 { + self.pending_propagation_count + } + + /// Return the number of downstream propagation records quarantined for intervention. + #[must_use] + pub const fn quarantined_propagation_count(self) -> u64 { + self.quarantined_propagation_count + } + + /// Return the oldest original request time among non-terminal participant requests. + #[must_use] + pub const fn oldest_active_request_at_unix_ms(self) -> Option { + self.oldest_active_request_at_unix_ms + } + + /// Return the oldest event time among pending downstream propagations. + #[must_use] + pub const fn oldest_pending_propagation_event_at_unix_ms(self) -> Option { + self.oldest_pending_propagation_event_at_unix_ms + } +} + +/// Operator-supplied bounds for asynchronous scoring-job backlog evidence. +/// +/// Scoring timeliness is a deployment-profile policy. The product does not invent a +/// universal queue-depth or age SLO; callers supply evidence-backed limits. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct ScoringJobBacklogPolicy { + /// Largest queued, leased, or retry-scheduled job population still accepted. + pub max_active_job_count: u64, + /// Largest age in milliseconds of the oldest active scoring job. + pub max_active_job_age_ms: u64, + /// Largest quarantined scoring-job population still accepted. + pub max_quarantined_job_count: u64, +} + +/// Aggregate scoring-job backlog evidence without request, worker, or result identities. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct PostgresScoringJobBacklogEvidence { + active_job_count: u64, + quarantined_job_count: u64, + oldest_active_job_at_unix_ms: Option, +} + +impl PostgresScoringJobBacklogEvidence { + /// Return queued plus leased plus retry-scheduled scoring jobs. + #[must_use] + pub const fn active_job_count(self) -> u64 { + self.active_job_count + } + + /// Return the number of scoring jobs quarantined for operator attention. + #[must_use] + pub const fn quarantined_job_count(self) -> u64 { + self.quarantined_job_count + } + + /// Return the oldest `created_at` among active scoring jobs, if any. + #[must_use] + pub const fn oldest_active_job_at_unix_ms(self) -> Option { + self.oldest_active_job_at_unix_ms + } +} + +/// Fail-closed error while reading durable backlog evidence. +#[derive(Debug)] +#[non_exhaustive] +pub enum PostgresBacklogProbeError { + /// A stored event time violated the positive-millisecond persistence contract. + InvalidStoredValue, + /// `PostgreSQL` could not execute the aggregate backlog probe. + Database(postgres::Error), +} + +impl Display for PostgresBacklogProbeError { + fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result { + formatter.write_str(match self { + Self::InvalidStoredValue => "stored backlog evidence violates the persistence contract", + Self::Database(_) => "PostgreSQL backlog probe failed", + }) + } +} + +impl Error for PostgresBacklogProbeError { + fn source(&self) -> Option<&(dyn Error + 'static)> { + match self { + Self::Database(error) => Some(error), + Self::InvalidStoredValue => None, + } + } +} + +impl From for PostgresBacklogProbeError { + fn from(error: postgres::Error) -> Self { + Self::Database(error) + } +} + /// Classify server-version and transaction-read-only evidence without performing I/O. /// /// `PostgreSQL` 10 and later encode `server_version_num` as `major * 10000 + minor`, so @@ -128,6 +346,297 @@ pub fn probe_postgres_runtime( )) } +/// Apply partial indexes that keep operational-backlog readiness probes bounded. +/// +/// Call this after the integration, inbox-consumption, and data-rights migrations +/// so an existing installation receives the same readiness indexes as a newly +/// initialized database. The statements are idempotent (`CREATE INDEX IF NOT EXISTS`). +/// Directory-order recovery still applies `0020` as a file; this function is the +/// product-owned apply path that callers must use instead of a private `include_str!`. +/// +/// # Errors +/// +/// Returns the `PostgreSQL` driver error when a required relation is missing or +/// the index statements cannot be executed. +pub fn apply_backlog_health_index_migration( + client: &mut impl GenericClient, +) -> Result<(), postgres::Error> { + client.batch_execute(BACKLOG_HEALTH_INDEX_MIGRATION) +} + +/// Probe aggregate outbox and inbox-consumption backlog evidence in one database snapshot. +/// +/// This function intentionally does not classify the result against a universal threshold. +/// It reads only aggregate counts and the oldest server-authoritative event time from the +/// existing product-owned integration tables. Missing tables or query failures are errors, +/// and invalid non-positive stored event times fail closed instead of being normalized. +/// +/// # Errors +/// +/// Returns [`PostgresBacklogProbeError::InvalidStoredValue`] when a stored oldest-event +/// timestamp violates the positive-millisecond contract, or +/// [`PostgresBacklogProbeError::Database`] when the aggregate query cannot be executed. +pub fn probe_postgres_integration_backlog( + client: &mut impl GenericClient, +) -> Result { + let row = client.query_one( + "SELECT \ + (SELECT COUNT(*)::BIGINT FROM integration_outbox WHERE current_state = 'pending'), \ + (SELECT COUNT(*)::BIGINT FROM integration_outbox WHERE current_state = 'quarantined'), \ + (SELECT MIN(latest_event_at_unix_ms) FROM integration_outbox WHERE current_state = 'pending'), \ + (SELECT COUNT(*)::BIGINT FROM integration_consumption \ + WHERE consumption_state IN ('pending', 'processing')), \ + (SELECT COUNT(*)::BIGINT FROM integration_consumption \ + WHERE consumption_state = 'quarantined'), \ + (SELECT MIN(latest_event_at_unix_ms) FROM integration_consumption \ + WHERE consumption_state IN ('pending', 'processing'))", + &[], + )?; + + let pending_outbox_count: i64 = row.get(0); + let quarantined_outbox_count: i64 = row.get(1); + let oldest_pending_outbox_event_at_unix_ms: Option = row.get(2); + let active_consumption_count: i64 = row.get(3); + let quarantined_consumption_count: i64 = row.get(4); + let oldest_active_consumption_event_at_unix_ms: Option = row.get(5); + + Ok(PostgresIntegrationBacklogEvidence { + pending_outbox_count: pending_outbox_count.cast_unsigned(), + quarantined_outbox_count: quarantined_outbox_count.cast_unsigned(), + active_consumption_count: active_consumption_count.cast_unsigned(), + quarantined_consumption_count: quarantined_consumption_count.cast_unsigned(), + oldest_pending_outbox_event_at_unix_ms: positive_optional_millis( + oldest_pending_outbox_event_at_unix_ms, + )?, + oldest_active_consumption_event_at_unix_ms: positive_optional_millis( + oldest_active_consumption_event_at_unix_ms, + )?, + }) +} + +/// Classify aggregate integration backlog evidence against one explicit operator policy. +/// +/// A zero observation time or a stored event apparently occurring after the observation +/// is `Unknown`, not healthy. Otherwise any count or age beyond the supplied policy is +/// `Stalled`. No default SLO, alert threshold, or recovery promise is inferred here. +#[must_use] +pub fn classify_postgres_integration_backlog( + evidence: &PostgresIntegrationBacklogEvidence, + observed_at_unix_ms: u64, + policy: &IntegrationBacklogPolicy, +) -> BacklogHealth { + if observed_at_unix_ms == 0 + || evidence + .oldest_pending_outbox_event_at_unix_ms + .is_some_and(|timestamp| timestamp > observed_at_unix_ms) + || evidence + .oldest_active_consumption_event_at_unix_ms + .is_some_and(|timestamp| timestamp > observed_at_unix_ms) + { + return BacklogHealth::Unknown; + } + + if evidence.pending_outbox_count > policy.max_pending_outbox_count + || evidence.quarantined_outbox_count > policy.max_quarantined_outbox_count + || evidence.active_consumption_count > policy.max_active_consumption_count + || evidence.quarantined_consumption_count > policy.max_quarantined_consumption_count + { + return BacklogHealth::Stalled; + } + + if age_exceeds( + evidence.oldest_pending_outbox_event_at_unix_ms, + observed_at_unix_ms, + policy.max_pending_outbox_age_ms, + ) || age_exceeds( + evidence.oldest_active_consumption_event_at_unix_ms, + observed_at_unix_ms, + policy.max_active_consumption_age_ms, + ) { + return BacklogHealth::Stalled; + } + + BacklogHealth::WithinBounds +} + +/// Probe participant data-rights request and propagation backlog evidence. +/// +/// Active request age is measured from the original request time, so a recent lifecycle +/// transition cannot hide a participant request that has remained unresolved for too long. +/// Propagation age is measured from the latest durable propagation event. The query exposes +/// only aggregate counts and timestamps, never participant, tenant, scope, or payload data. +/// +/// # Errors +/// +/// Returns [`PostgresBacklogProbeError::InvalidStoredValue`] for non-positive oldest +/// timestamps, or [`PostgresBacklogProbeError::Database`] when the query cannot execute. +pub fn probe_postgres_data_rights_backlog( + client: &mut impl GenericClient, +) -> Result { + let row = client.query_one( + "SELECT \ + (SELECT COUNT(*)::BIGINT FROM data_rights_request_state \ + WHERE current_state IN ('requested', 'identity_verified', 'processing')), \ + (SELECT MIN(requested_at_unix_ms) FROM data_rights_request_state \ + WHERE current_state IN ('requested', 'identity_verified', 'processing')), \ + (SELECT COUNT(*)::BIGINT FROM data_rights_propagation_state \ + WHERE current_state = 'pending'), \ + (SELECT COUNT(*)::BIGINT FROM data_rights_propagation_state \ + WHERE current_state = 'quarantined'), \ + (SELECT MIN(latest_event_at_unix_ms) FROM data_rights_propagation_state \ + WHERE current_state = 'pending')", + &[], + )?; + + let active_request_count: i64 = row.get(0); + let oldest_active_request_at_unix_ms: Option = row.get(1); + let pending_propagation_count: i64 = row.get(2); + let quarantined_propagation_count: i64 = row.get(3); + let oldest_pending_propagation_event_at_unix_ms: Option = row.get(4); + + Ok(PostgresDataRightsBacklogEvidence { + active_request_count: active_request_count.cast_unsigned(), + pending_propagation_count: pending_propagation_count.cast_unsigned(), + quarantined_propagation_count: quarantined_propagation_count.cast_unsigned(), + oldest_active_request_at_unix_ms: positive_optional_millis( + oldest_active_request_at_unix_ms, + )?, + oldest_pending_propagation_event_at_unix_ms: positive_optional_millis( + oldest_pending_propagation_event_at_unix_ms, + )?, + }) +} + +/// Classify participant data-rights backlog evidence against an explicit operator policy. +/// +/// Missing observation time or future-dated durable evidence produces `Unknown`. Excess +/// request/propagation counts, quarantine population, or age produce `Stalled`. Otherwise +/// the measured data-rights backlog is within the caller's declared operating bounds. +#[must_use] +pub fn classify_postgres_data_rights_backlog( + evidence: &PostgresDataRightsBacklogEvidence, + observed_at_unix_ms: u64, + policy: &DataRightsBacklogPolicy, +) -> BacklogHealth { + if observed_at_unix_ms == 0 + || evidence + .oldest_active_request_at_unix_ms + .is_some_and(|timestamp| timestamp > observed_at_unix_ms) + || evidence + .oldest_pending_propagation_event_at_unix_ms + .is_some_and(|timestamp| timestamp > observed_at_unix_ms) + { + return BacklogHealth::Unknown; + } + + if evidence.active_request_count > policy.max_active_request_count + || evidence.pending_propagation_count > policy.max_pending_propagation_count + || evidence.quarantined_propagation_count > policy.max_quarantined_propagation_count + { + return BacklogHealth::Stalled; + } + + if age_exceeds( + evidence.oldest_active_request_at_unix_ms, + observed_at_unix_ms, + policy.max_active_request_age_ms, + ) || age_exceeds( + evidence.oldest_pending_propagation_event_at_unix_ms, + observed_at_unix_ms, + policy.max_pending_propagation_age_ms, + ) { + return BacklogHealth::Stalled; + } + + BacklogHealth::WithinBounds +} + +/// Probe aggregate scoring-job backlog evidence without identities or payloads. +/// +/// Active work is queued, leased, or retry-scheduled. Completed and cancelled jobs are +/// terminal and do not participate. Age is measured from `created_at` so a later lease +/// or retry transition cannot hide a job that has waited too long. The query returns +/// only counts and one oldest timestamp. +/// +/// # Errors +/// +/// Returns [`PostgresBacklogProbeError::InvalidStoredValue`] when the oldest created +/// time is not a positive unix-millisecond value, or +/// [`PostgresBacklogProbeError::Database`] when the query cannot execute. +pub fn probe_postgres_scoring_job_backlog( + client: &mut impl GenericClient, +) -> Result { + let row = client.query_one( + "SELECT \ + (SELECT COUNT(*)::BIGINT FROM scoring_job_state \ + WHERE scoring_state IN ('queued', 'leased', 'retry_scheduled')), \ + (SELECT COUNT(*)::BIGINT FROM scoring_job_state \ + WHERE scoring_state = 'quarantined'), \ + (SELECT (EXTRACT(EPOCH FROM MIN(created_at)) * 1000)::BIGINT \ + FROM scoring_job_state \ + WHERE scoring_state IN ('queued', 'leased', 'retry_scheduled'))", + &[], + )?; + + let active_job_count: i64 = row.get(0); + let quarantined_job_count: i64 = row.get(1); + let oldest_active_job_at_unix_ms: Option = row.get(2); + + Ok(PostgresScoringJobBacklogEvidence { + active_job_count: active_job_count.cast_unsigned(), + quarantined_job_count: quarantined_job_count.cast_unsigned(), + oldest_active_job_at_unix_ms: positive_optional_millis(oldest_active_job_at_unix_ms)?, + }) +} + +/// Classify scoring-job backlog evidence against an explicit operator policy. +/// +/// Missing observation time or future-dated created-at evidence produces `Unknown`. +/// Excess active or quarantined counts, or an active job older than the supplied +/// age bound, produce `Stalled`. +#[must_use] +pub fn classify_postgres_scoring_job_backlog( + evidence: &PostgresScoringJobBacklogEvidence, + observed_at_unix_ms: u64, + policy: &ScoringJobBacklogPolicy, +) -> BacklogHealth { + if observed_at_unix_ms == 0 + || evidence + .oldest_active_job_at_unix_ms + .is_some_and(|timestamp| timestamp > observed_at_unix_ms) + { + return BacklogHealth::Unknown; + } + + if evidence.active_job_count > policy.max_active_job_count + || evidence.quarantined_job_count > policy.max_quarantined_job_count + { + return BacklogHealth::Stalled; + } + + if age_exceeds( + evidence.oldest_active_job_at_unix_ms, + observed_at_unix_ms, + policy.max_active_job_age_ms, + ) { + return BacklogHealth::Stalled; + } + + BacklogHealth::WithinBounds +} + +fn positive_optional_millis(value: Option) -> Result, PostgresBacklogProbeError> { + match value { + Some(value) if value > 0 => Ok(Some(value.cast_unsigned())), + Some(_) => Err(PostgresBacklogProbeError::InvalidStoredValue), + None => Ok(None), + } +} + +fn age_exceeds(oldest_event_at: Option, observed_at: u64, maximum_age: u64) -> bool { + oldest_event_at.is_some_and(|timestamp| observed_at.saturating_sub(timestamp) > maximum_age) +} + /// Probe whether every caller-declared relation required by this application build exists. /// /// The application or packaged deployment remains responsible for declaring the exact @@ -158,3 +667,157 @@ pub fn probe_postgres_relation_integrity( } Ok(DataIntegrityHealth::Verified) } + +/// Combine independently measured operational-backlog families into one readiness signal. +/// +/// Scoring-job, integration, and data-rights evidence stay separately classified so a +/// healthy outbox cannot hide a stalled scoring queue. [`BacklogHealth::combine`] +/// keeps a known stall above an unknown observation, and unknown above a healthy one. +#[must_use] +pub fn classify_postgres_operational_backlog( + integration: &PostgresIntegrationBacklogEvidence, + data_rights: &PostgresDataRightsBacklogEvidence, + scoring_job: &PostgresScoringJobBacklogEvidence, + observed_at_unix_ms: u64, + integration_policy: &IntegrationBacklogPolicy, + data_rights_policy: &DataRightsBacklogPolicy, + scoring_job_policy: &ScoringJobBacklogPolicy, +) -> BacklogHealth { + classify_postgres_integration_backlog(integration, observed_at_unix_ms, integration_policy) + .combine(classify_postgres_data_rights_backlog( + data_rights, + observed_at_unix_ms, + data_rights_policy, + )) + .combine(classify_postgres_scoring_job_backlog( + scoring_job, + observed_at_unix_ms, + scoring_job_policy, + )) +} + +#[cfg(test)] +mod operational_backlog_composition_tests { + use super::{ + classify_postgres_operational_backlog, DataRightsBacklogPolicy, IntegrationBacklogPolicy, + PostgresDataRightsBacklogEvidence, PostgresIntegrationBacklogEvidence, + PostgresScoringJobBacklogEvidence, ScoringJobBacklogPolicy, + }; + use crate::health::BacklogHealth; + + fn integration_policy() -> IntegrationBacklogPolicy { + IntegrationBacklogPolicy { + max_pending_outbox_count: 4, + max_pending_outbox_age_ms: 5_000, + max_quarantined_outbox_count: 1, + max_active_consumption_count: 4, + max_active_consumption_age_ms: 5_000, + max_quarantined_consumption_count: 1, + } + } + + fn data_rights_policy() -> DataRightsBacklogPolicy { + DataRightsBacklogPolicy { + max_active_request_count: 4, + max_active_request_age_ms: 10_000, + max_pending_propagation_count: 4, + max_pending_propagation_age_ms: 5_000, + max_quarantined_propagation_count: 1, + } + } + + fn scoring_job_policy() -> ScoringJobBacklogPolicy { + ScoringJobBacklogPolicy { + max_active_job_count: 4, + max_active_job_age_ms: 5_000, + max_quarantined_job_count: 1, + } + } + + fn empty_integration() -> PostgresIntegrationBacklogEvidence { + PostgresIntegrationBacklogEvidence { + pending_outbox_count: 0, + quarantined_outbox_count: 0, + active_consumption_count: 0, + quarantined_consumption_count: 0, + oldest_pending_outbox_event_at_unix_ms: None, + oldest_active_consumption_event_at_unix_ms: None, + } + } + + fn empty_data_rights() -> PostgresDataRightsBacklogEvidence { + PostgresDataRightsBacklogEvidence { + active_request_count: 0, + pending_propagation_count: 0, + quarantined_propagation_count: 0, + oldest_active_request_at_unix_ms: None, + oldest_pending_propagation_event_at_unix_ms: None, + } + } + + fn empty_scoring_job() -> PostgresScoringJobBacklogEvidence { + PostgresScoringJobBacklogEvidence { + active_job_count: 0, + quarantined_job_count: 0, + oldest_active_job_at_unix_ms: None, + } + } + + #[test] + fn empty_families_are_within_bounds_without_invented_thresholds() { + assert_eq!( + classify_postgres_operational_backlog( + &empty_integration(), + &empty_data_rights(), + &empty_scoring_job(), + 10_000, + &integration_policy(), + &data_rights_policy(), + &scoring_job_policy(), + ), + BacklogHealth::WithinBounds + ); + } + + #[test] + fn stalled_scoring_job_cannot_be_masked_by_healthy_integration_or_data_rights() { + let stalled_scoring = PostgresScoringJobBacklogEvidence { + active_job_count: 1, + quarantined_job_count: 0, + oldest_active_job_at_unix_ms: Some(1_000), + }; + assert_eq!( + classify_postgres_operational_backlog( + &empty_integration(), + &empty_data_rights(), + &stalled_scoring, + 10_000, + &integration_policy(), + &data_rights_policy(), + &scoring_job_policy(), + ), + BacklogHealth::Stalled + ); + } + + #[test] + fn unknown_scoring_observation_cannot_be_masked_by_healthy_siblings() { + let future_scoring = PostgresScoringJobBacklogEvidence { + active_job_count: 1, + quarantined_job_count: 0, + oldest_active_job_at_unix_ms: Some(20_000), + }; + assert_eq!( + classify_postgres_operational_backlog( + &empty_integration(), + &empty_data_rights(), + &future_scoring, + 10_000, + &integration_policy(), + &data_rights_policy(), + &scoring_job_policy(), + ), + BacklogHealth::Unknown + ); + } +} diff --git a/src/postgres_scoring_job.rs b/src/postgres_scoring_job.rs index 08375bf5..16bcfc06 100644 --- a/src/postgres_scoring_job.rs +++ b/src/postgres_scoring_job.rs @@ -13,6 +13,8 @@ use std::error::Error; use std::fmt::{Display, Formatter}; const SCORING_JOB_MIGRATION: &str = include_str!("../migrations/0002_scoring_job_state.sql"); +const SCORING_JOB_HEALTH_INDEX_MIGRATION: &str = + include_str!("../migrations/0021_scoring_job_health_indexes.sql"); /// Outcome of persisting a scoring-job cancellation. #[derive(Clone, Copy, Debug, Eq, PartialEq)] @@ -173,13 +175,18 @@ impl From for ScoringJobPersistenceError { } } -/// Apply the idempotent scoring-job migration to a `PostgreSQL` connection. +/// Apply the idempotent scoring-job table and readiness-index migrations. +/// +/// The shipped base table remains immutable; forward-only health indexes are +/// applied afterwards so existing installations receive the same readiness +/// indexes as a newly initialized database. /// /// # Errors /// -/// Returns the `PostgreSQL` error if the migration cannot be applied. +/// Returns the `PostgreSQL` error if either migration cannot be applied. pub fn apply_scoring_job_migration(client: &mut impl GenericClient) -> Result<(), postgres::Error> { - client.batch_execute(SCORING_JOB_MIGRATION) + let migration_batch = format!("{SCORING_JOB_MIGRATION}\n{SCORING_JOB_HEALTH_INDEX_MIGRATION}"); + client.batch_execute(&migration_batch) } /// Persist the immutable identity of a fresh queued scoring job. diff --git a/tests/capability_health_contract.rs b/tests/capability_health_contract.rs index 8d7cf6ed..2dbf9365 100644 --- a/tests/capability_health_contract.rs +++ b/tests/capability_health_contract.rs @@ -111,6 +111,61 @@ fn integrity_and_backlog_failures_block_state_changing_readiness() { } } +#[test] +fn independent_backlog_signals_compose_without_masking_stalled_or_unknown_work() { + let cases = [ + ( + BacklogHealth::WithinBounds, + BacklogHealth::WithinBounds, + BacklogHealth::WithinBounds, + ), + ( + BacklogHealth::WithinBounds, + BacklogHealth::Unknown, + BacklogHealth::Unknown, + ), + ( + BacklogHealth::WithinBounds, + BacklogHealth::Stalled, + BacklogHealth::Stalled, + ), + ( + BacklogHealth::Unknown, + BacklogHealth::WithinBounds, + BacklogHealth::Unknown, + ), + ( + BacklogHealth::Unknown, + BacklogHealth::Unknown, + BacklogHealth::Unknown, + ), + ( + BacklogHealth::Unknown, + BacklogHealth::Stalled, + BacklogHealth::Stalled, + ), + ( + BacklogHealth::Stalled, + BacklogHealth::WithinBounds, + BacklogHealth::Stalled, + ), + ( + BacklogHealth::Stalled, + BacklogHealth::Unknown, + BacklogHealth::Stalled, + ), + ( + BacklogHealth::Stalled, + BacklogHealth::Stalled, + BacklogHealth::Stalled, + ), + ]; + + for (left, right, expected) in cases { + assert_eq!(left.combine(right), expected); + } +} + #[test] fn nonlive_process_and_unknown_required_capability_fail_closed() { let snapshot = RuntimeHealthSnapshot::new( diff --git a/tests/postgres_backlog_health_index_contract.rs b/tests/postgres_backlog_health_index_contract.rs new file mode 100644 index 00000000..6381825f --- /dev/null +++ b/tests/postgres_backlog_health_index_contract.rs @@ -0,0 +1,168 @@ +//! Real `PostgreSQL` contract for indexes that keep readiness backlog probes bounded. + +use postgres::{Client, NoTls}; +use psychometrics_commons_runtime::postgres_data_rights::apply_data_rights_migration; +use psychometrics_commons_runtime::postgres_health::apply_backlog_health_index_migration; +use psychometrics_commons_runtime::postgres_inbox_consumption::apply_inbox_consumption_migration; +use psychometrics_commons_runtime::postgres_integration::apply_integration_migration; +use std::sync::atomic::{AtomicU64, Ordering}; + +static SCHEMA_NONCE: AtomicU64 = AtomicU64::new(1); + +fn test_client() -> Client { + let connection = std::env::var("TEST_DATABASE_URL") + .expect("TEST_DATABASE_URL must identify the isolated CI PostgreSQL database"); + Client::connect(&connection, NoTls).expect("isolated CI PostgreSQL database must be reachable") +} + +fn isolated_client() -> (Client, String) { + let mut client = test_client(); + let nonce = SCHEMA_NONCE.fetch_add(1, Ordering::Relaxed); + let schema = format!("backlog_health_index_{}_{}", std::process::id(), nonce); + client + .batch_execute(&format!( + "CREATE SCHEMA {schema}; SET search_path TO {schema};" + )) + .expect("isolated backlog-health index schema should be created"); + apply_integration_migration(&mut client).expect("integration migration should apply"); + apply_inbox_consumption_migration(&mut client) + .expect("inbox-consumption migration should apply"); + apply_data_rights_migration(&mut client).expect("data-rights migration should apply"); + apply_backlog_health_index_migration(&mut client) + .expect("product backlog-health index migration should apply"); + (client, schema) +} + +fn cleanup(mut client: Client, schema: &str) { + client + .batch_execute(&format!( + "SET search_path TO public; DROP SCHEMA {schema} CASCADE;" + )) + .expect("isolated backlog-health index schema should be removed"); +} + +fn index_definition(client: &mut Client, index_name: &str) -> String { + client + .query_one( + "SELECT indexdef FROM pg_indexes \ + WHERE schemaname = current_schema() AND indexname = $1", + &[&index_name], + ) + .unwrap_or_else(|error| panic!("index {index_name} must exist: {error}")) + .get(0) +} + +fn assert_partial_health_index( + client: &mut Client, + index_name: &str, + table_name: &str, + timestamp_column: &str, + state_column: &str, + state_tokens: &[&str], +) { + let definition = index_definition(client, index_name).to_ascii_lowercase(); + assert!( + definition.contains(table_name), + "{index_name} must index {table_name}: {definition}" + ); + assert!( + definition.contains(timestamp_column), + "{index_name} must cover {timestamp_column}: {definition}" + ); + assert!( + definition.contains(" where "), + "{index_name} must be partial so terminal history does not dominate the readiness index: {definition}" + ); + assert!( + definition.contains(state_column), + "{index_name} predicate must constrain {state_column}: {definition}" + ); + for state in state_tokens { + assert!( + definition.contains(state), + "{index_name} predicate must include state {state}: {definition}" + ); + } +} + +#[test] +fn readiness_backlog_queries_have_state_selective_indexes() { + let (mut client, schema) = isolated_client(); + + assert_partial_health_index( + &mut client, + "integration_outbox_pending_health_idx", + "integration_outbox", + "latest_event_at_unix_ms", + "current_state", + &["pending"], + ); + assert_partial_health_index( + &mut client, + "integration_outbox_quarantined_health_idx", + "integration_outbox", + "latest_event_at_unix_ms", + "current_state", + &["quarantined"], + ); + assert_partial_health_index( + &mut client, + "integration_consumption_active_health_idx", + "integration_consumption", + "latest_event_at_unix_ms", + "consumption_state", + &["pending", "processing"], + ); + assert_partial_health_index( + &mut client, + "integration_consumption_quarantined_health_idx", + "integration_consumption", + "latest_event_at_unix_ms", + "consumption_state", + &["quarantined"], + ); + assert_partial_health_index( + &mut client, + "data_rights_request_active_health_idx", + "data_rights_request_state", + "requested_at_unix_ms", + "current_state", + &["requested", "identity_verified", "processing"], + ); + + let propagation_index = + index_definition(&mut client, "data_rights_propagation_state_idx").to_ascii_lowercase(); + assert!( + propagation_index.contains("current_state") + && propagation_index.contains("latest_event_at_unix_ms"), + "the existing propagation state/time index must continue covering propagation health probes: {propagation_index}" + ); + + apply_backlog_health_index_migration(&mut client) + .expect("readiness index apply must be idempotent for existing installations"); + + cleanup(client, &schema); +} + +#[test] +fn product_apply_path_fails_closed_when_required_backlog_relations_are_missing() { + let mut client = test_client(); + let nonce = SCHEMA_NONCE.fetch_add(1, Ordering::Relaxed); + let schema = format!( + "backlog_health_index_missing_{}_{}", + std::process::id(), + nonce + ); + client + .batch_execute(&format!( + "CREATE SCHEMA {schema}; SET search_path TO {schema};" + )) + .expect("isolated missing-relation schema should be created"); + + assert!( + apply_backlog_health_index_migration(&mut client).is_err(), + "readiness indexes must not be claimed when the owned backlog tables are absent" + ); + + cleanup(client, &schema); +} diff --git a/tests/postgres_backlog_probe_generic_client.rs b/tests/postgres_backlog_probe_generic_client.rs new file mode 100644 index 00000000..9da8f9f9 --- /dev/null +++ b/tests/postgres_backlog_probe_generic_client.rs @@ -0,0 +1,205 @@ +//! Real `PostgreSQL` coverage for backlog probes across supported generic client types. +//! +//! The production probes accept any `GenericClient`, so both a direct `Client` and a +//! caller-owned `Transaction` must exercise successful and failed query boundaries. + +use postgres::{Client, NoTls}; +use psychometrics_commons_runtime::postgres_data_rights::apply_data_rights_migration; +use psychometrics_commons_runtime::postgres_health::{ + probe_postgres_data_rights_backlog, probe_postgres_integration_backlog, + PostgresBacklogProbeError, +}; +use psychometrics_commons_runtime::postgres_inbox_consumption::apply_inbox_consumption_migration; +use psychometrics_commons_runtime::postgres_integration::apply_integration_migration; +use std::error::Error; +use std::sync::atomic::{AtomicU64, Ordering}; + +static SCHEMA_NONCE: AtomicU64 = AtomicU64::new(1); + +fn test_client() -> Client { + let connection = std::env::var("TEST_DATABASE_URL") + .expect("TEST_DATABASE_URL must identify the isolated CI PostgreSQL database"); + Client::connect(&connection, NoTls).expect("isolated CI PostgreSQL database must be reachable") +} + +fn create_isolated_schema(client: &mut Client, prefix: &str) -> String { + let nonce = SCHEMA_NONCE.fetch_add(1, Ordering::Relaxed); + let schema = format!("{prefix}_{}_{}", std::process::id(), nonce); + client + .batch_execute(&format!( + "CREATE SCHEMA {schema}; SET search_path TO {schema};" + )) + .expect("isolated backlog-probe schema should be created"); + schema +} + +fn cleanup_schema(mut client: Client, schema: &str) { + client + .batch_execute(&format!( + "SET search_path TO public; DROP SCHEMA {schema} CASCADE;" + )) + .expect("isolated backlog-probe schema should be removed"); +} + +#[test] +fn integration_probe_supports_transaction_success_and_direct_client_failure() { + let mut client = test_client(); + let schema = create_isolated_schema(&mut client, "integration_probe_generic_client"); + apply_integration_migration(&mut client).expect("integration migration should apply"); + apply_inbox_consumption_migration(&mut client) + .expect("inbox-consumption migration should apply"); + + let mut transaction = client.transaction().unwrap(); + let evidence = probe_postgres_integration_backlog(&mut transaction).unwrap(); + assert_eq!(evidence.pending_outbox_count(), 0); + assert_eq!(evidence.active_consumption_count(), 0); + transaction.rollback().unwrap(); + + client + .batch_execute("DROP TABLE integration_outbox CASCADE;") + .unwrap(); + assert!(matches!( + probe_postgres_integration_backlog(&mut client), + Err(PostgresBacklogProbeError::Database(_)) + )); + + cleanup_schema(client, &schema); +} + +#[test] +fn data_rights_probe_supports_transaction_success_and_direct_client_failure() { + let mut client = test_client(); + let schema = create_isolated_schema(&mut client, "data_rights_probe_generic_client"); + apply_integration_migration(&mut client).expect("integration migration should apply"); + apply_data_rights_migration(&mut client).expect("data-rights migration should apply"); + + let mut transaction = client.transaction().unwrap(); + let evidence = probe_postgres_data_rights_backlog(&mut transaction).unwrap(); + assert_eq!(evidence.active_request_count(), 0); + assert_eq!(evidence.pending_propagation_count(), 0); + transaction.rollback().unwrap(); + + client + .batch_execute("DROP TABLE data_rights_request_state CASCADE;") + .unwrap(); + assert!(matches!( + probe_postgres_data_rights_backlog(&mut client), + Err(PostgresBacklogProbeError::Database(_)) + )); + + cleanup_schema(client, &schema); +} + +#[test] +fn integration_probe_transaction_rejects_each_invalid_timestamp_independently() { + let mut client = test_client(); + let schema = create_isolated_schema(&mut client, "integration_probe_txn_ts"); + client + .batch_execute( + "CREATE TABLE integration_outbox (current_state TEXT, latest_event_at_unix_ms BIGINT);\ + CREATE TABLE integration_consumption (\ + consumption_state TEXT, latest_event_at_unix_ms BIGINT);\ + INSERT INTO integration_outbox VALUES ('pending', -1);\ + INSERT INTO integration_consumption VALUES ('pending', 4_000);", + ) + .unwrap(); + { + let mut transaction = client.transaction().unwrap(); + assert!(matches!( + probe_postgres_integration_backlog(&mut transaction), + Err(PostgresBacklogProbeError::InvalidStoredValue) + )); + transaction.rollback().unwrap(); + } + + client + .batch_execute( + "TRUNCATE integration_outbox, integration_consumption;\ + INSERT INTO integration_outbox VALUES ('pending', 4_000);\ + INSERT INTO integration_consumption VALUES ('pending', 0);", + ) + .unwrap(); + { + let mut transaction = client.transaction().unwrap(); + assert!(matches!( + probe_postgres_integration_backlog(&mut transaction), + Err(PostgresBacklogProbeError::InvalidStoredValue) + )); + transaction.rollback().unwrap(); + } + + cleanup_schema(client, &schema); +} + +#[test] +fn data_rights_probe_transaction_rejects_each_invalid_timestamp_independently() { + let mut client = test_client(); + let schema = create_isolated_schema(&mut client, "data_rights_probe_txn_ts"); + client + .batch_execute( + "CREATE TABLE data_rights_request_state (\ + current_state TEXT, requested_at_unix_ms BIGINT);\ + CREATE TABLE data_rights_propagation_state (\ + current_state TEXT, latest_event_at_unix_ms BIGINT);\ + INSERT INTO data_rights_request_state VALUES ('requested', -1);\ + INSERT INTO data_rights_propagation_state VALUES ('pending', 2_000);", + ) + .unwrap(); + { + let mut transaction = client.transaction().unwrap(); + assert!(matches!( + probe_postgres_data_rights_backlog(&mut transaction), + Err(PostgresBacklogProbeError::InvalidStoredValue) + )); + transaction.rollback().unwrap(); + } + + client + .batch_execute( + "TRUNCATE data_rights_request_state, data_rights_propagation_state;\ + INSERT INTO data_rights_request_state VALUES ('requested', 2_000);\ + INSERT INTO data_rights_propagation_state VALUES ('pending', 0);", + ) + .unwrap(); + { + let mut transaction = client.transaction().unwrap(); + assert!(matches!( + probe_postgres_data_rights_backlog(&mut transaction), + Err(PostgresBacklogProbeError::InvalidStoredValue) + )); + transaction.rollback().unwrap(); + } + + cleanup_schema(client, &schema); +} + +#[test] +fn backlog_probes_surface_transaction_query_failures() { + let mut client = test_client(); + let schema = create_isolated_schema(&mut client, "backlog_probe_txn_query"); + apply_integration_migration(&mut client).expect("integration migration should apply"); + apply_inbox_consumption_migration(&mut client) + .expect("inbox-consumption migration should apply"); + apply_data_rights_migration(&mut client).expect("data-rights migration should apply"); + + { + let mut transaction = client.transaction().unwrap(); + assert!(transaction.batch_execute("SELECT 1 / 0").is_err()); + let error = probe_postgres_integration_backlog(&mut transaction).unwrap_err(); + assert!(matches!(error, PostgresBacklogProbeError::Database(_))); + assert_eq!(error.to_string(), "PostgreSQL backlog probe failed"); + assert!(error.source().is_some()); + transaction.rollback().unwrap(); + } + { + let mut transaction = client.transaction().unwrap(); + assert!(transaction.batch_execute("SELECT 1 / 0").is_err()); + let error = probe_postgres_data_rights_backlog(&mut transaction).unwrap_err(); + assert!(matches!(error, PostgresBacklogProbeError::Database(_))); + assert_eq!(error.to_string(), "PostgreSQL backlog probe failed"); + assert!(error.source().is_some()); + transaction.rollback().unwrap(); + } + + cleanup_schema(client, &schema); +} diff --git a/tests/postgres_data_rights_backlog_health.rs b/tests/postgres_data_rights_backlog_health.rs new file mode 100644 index 00000000..3b9fdcf6 --- /dev/null +++ b/tests/postgres_data_rights_backlog_health.rs @@ -0,0 +1,326 @@ +//! Real `PostgreSQL` contract for participant data-rights backlog health evidence. + +use postgres::{Client, NoTls}; +use psychometrics_commons_runtime::health::BacklogHealth; +use psychometrics_commons_runtime::postgres_data_rights::apply_data_rights_migration; +use psychometrics_commons_runtime::postgres_health::{ + classify_postgres_data_rights_backlog, probe_postgres_data_rights_backlog, + DataRightsBacklogPolicy, PostgresBacklogProbeError, +}; +use psychometrics_commons_runtime::postgres_integration::apply_integration_migration; +use std::error::Error; +use std::sync::atomic::{AtomicU64, Ordering}; + +static SCHEMA_NONCE: AtomicU64 = AtomicU64::new(1); + +fn test_client() -> Client { + let connection = std::env::var("TEST_DATABASE_URL") + .expect("TEST_DATABASE_URL must identify the isolated CI PostgreSQL database"); + Client::connect(&connection, NoTls).expect("isolated CI PostgreSQL database must be reachable") +} + +fn isolated_client() -> (Client, String) { + let mut client = test_client(); + let nonce = SCHEMA_NONCE.fetch_add(1, Ordering::Relaxed); + let schema = format!("data_rights_backlog_{}_{}", std::process::id(), nonce); + client + .batch_execute(&format!( + "CREATE SCHEMA {schema}; SET search_path TO {schema};" + )) + .expect("isolated data-rights backlog schema should be created"); + apply_integration_migration(&mut client).expect("integration migration should apply"); + apply_data_rights_migration(&mut client).expect("data-rights migration should apply"); + (client, schema) +} + +fn cleanup(mut client: Client, schema: &str) { + client + .batch_execute(&format!( + "SET search_path TO public; DROP SCHEMA {schema} CASCADE;" + )) + .expect("isolated data-rights backlog schema should be removed"); +} + +fn policy() -> DataRightsBacklogPolicy { + DataRightsBacklogPolicy { + max_active_request_count: 4, + max_active_request_age_ms: 10_000, + max_pending_propagation_count: 4, + max_pending_propagation_age_ms: 5_000, + max_quarantined_propagation_count: 1, + } +} + +fn insert_request( + client: &mut Client, + suffix: &str, + state: &str, + requested_at: i64, + latest_event_at: i64, +) { + let request_ref = format!("request_{suffix}"); + client + .execute( + "INSERT INTO data_rights_request_state (\ + request_ref, tenant_ref, participant_ref, request_kind, scope_ref,\ + current_state, requested_at_unix_ms, latest_event_at_unix_ms\ + ) VALUES ($1,'tenant_rights_alpha','participant_rights_alpha','export',\ + 'scope_rights_alpha',$2,$3,$4)", + &[&request_ref, &state, &requested_at, &latest_event_at], + ) + .unwrap(); +} + +fn insert_propagation(client: &mut Client, suffix: &str, state: &str, event_at: i64) { + let request_ref = format!("request_propagation_{suffix}"); + let event_ref = format!("event_propagation_{suffix}"); + let dependent_system_ref = format!("dependent_system_{suffix}"); + // The propagation row has a request foreign key, so this fixture also adds one active + // processing request at time 1_000. That row intentionally contributes to request count + // and oldest-request age assertions below; it is not hidden probe behavior. + insert_request( + client, + &format!("propagation_{suffix}"), + "processing", + 1_000, + 1_500, + ); + client + .execute( + "INSERT INTO integration_outbox (\ + event_ref, event_type, schema_version, source_ref, tenant_ref, subject_ref,\ + occurred_at_unix_ms, correlation_ref, payload_digest, max_attempts,\ + current_state, latest_event_at_unix_ms\ + ) VALUES ($1,'participant.data_rights.changed','v1','psychometrics_commons',\ + 'tenant_rights_alpha','participant_rights_alpha',$2,\ + 'correlation_rights_alpha',\ + 'sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc',\ + 3,'pending',$2)", + &[&event_ref, &event_at], + ) + .unwrap(); + client + .execute( + "INSERT INTO data_rights_propagation_state (\ + request_ref, tenant_ref, dependent_system_ref, source_ref, event_ref,\ + current_state, latest_event_at_unix_ms\ + ) VALUES ($1,'tenant_rights_alpha',$2,'psychometrics_commons',$3,$4,$5)", + &[ + &request_ref, + &dependent_system_ref, + &event_ref, + &state, + &event_at, + ], + ) + .unwrap(); +} + +#[test] +fn empty_data_rights_backlog_is_within_caller_supplied_bounds() { + let (mut client, schema) = isolated_client(); + let evidence = probe_postgres_data_rights_backlog(&mut client).unwrap(); + + assert_eq!(evidence.active_request_count(), 0); + assert_eq!(evidence.pending_propagation_count(), 0); + assert_eq!(evidence.quarantined_propagation_count(), 0); + assert_eq!(evidence.oldest_active_request_at_unix_ms(), None); + assert_eq!(evidence.oldest_pending_propagation_event_at_unix_ms(), None); + assert_eq!( + classify_postgres_data_rights_backlog(&evidence, 20_000, &policy()), + BacklogHealth::WithinBounds + ); + + cleanup(client, &schema); +} + +#[test] +fn probe_keeps_request_age_separate_from_propagation_age() { + let (mut client, schema) = isolated_client(); + insert_request(&mut client, "requested_alpha", "requested", 2_000, 4_000); + insert_request( + &mut client, + "identity_verified_alpha", + "identity_verified", + 3_000, + 5_000, + ); + insert_request(&mut client, "completed_alpha", "completed", 500, 6_000); + insert_propagation(&mut client, "pending_alpha", "pending", 7_000); + insert_propagation(&mut client, "quarantined_alpha", "quarantined", 8_000); + + let evidence = probe_postgres_data_rights_backlog(&mut client).unwrap(); + assert_eq!(evidence.active_request_count(), 4); + assert_eq!(evidence.pending_propagation_count(), 1); + assert_eq!(evidence.quarantined_propagation_count(), 1); + assert_eq!(evidence.oldest_active_request_at_unix_ms(), Some(1_000)); + assert_eq!( + evidence.oldest_pending_propagation_event_at_unix_ms(), + Some(7_000) + ); + assert_eq!( + classify_postgres_data_rights_backlog(&evidence, 8_500, &policy()), + BacklogHealth::WithinBounds + ); + + cleanup(client, &schema); +} + +#[test] +fn participant_rights_backlog_fails_closed_by_count_age_or_quarantine_policy() { + let (mut client, schema) = isolated_client(); + insert_request(&mut client, "old_active_alpha", "processing", 2_000, 3_000); + insert_propagation(&mut client, "old_pending_alpha", "pending", 4_000); + insert_propagation( + &mut client, + "quarantined_policy_alpha", + "quarantined", + 4_500, + ); + let evidence = probe_postgres_data_rights_backlog(&mut client).unwrap(); + + let strict_request_count = DataRightsBacklogPolicy { + max_active_request_count: 0, + ..policy() + }; + assert_eq!( + classify_postgres_data_rights_backlog(&evidence, 5_000, &strict_request_count), + BacklogHealth::Stalled + ); + + let strict_propagation_count = DataRightsBacklogPolicy { + max_pending_propagation_count: 0, + ..policy() + }; + assert_eq!( + classify_postgres_data_rights_backlog(&evidence, 5_000, &strict_propagation_count), + BacklogHealth::Stalled + ); + + let strict_quarantine = DataRightsBacklogPolicy { + max_quarantined_propagation_count: 0, + ..policy() + }; + assert_eq!( + classify_postgres_data_rights_backlog(&evidence, 5_000, &strict_quarantine), + BacklogHealth::Stalled + ); + + assert_eq!( + classify_postgres_data_rights_backlog(&evidence, 12_001, &policy()), + BacklogHealth::Stalled + ); + + let request_age_relaxed = DataRightsBacklogPolicy { + max_active_request_age_ms: 20_000, + ..policy() + }; + assert_eq!( + classify_postgres_data_rights_backlog(&evidence, 9_001, &request_age_relaxed), + BacklogHealth::Stalled + ); + + cleanup(client, &schema); +} + +#[test] +fn future_data_rights_evidence_is_unknown_instead_of_healthy() { + let (mut client, schema) = isolated_client(); + insert_request( + &mut client, + "future_request_alpha", + "requested", + 10_000, + 10_000, + ); + let request_evidence = probe_postgres_data_rights_backlog(&mut client).unwrap(); + assert_eq!( + classify_postgres_data_rights_backlog(&request_evidence, 9_999, &policy()), + BacklogHealth::Unknown + ); + assert_eq!( + classify_postgres_data_rights_backlog(&request_evidence, 0, &policy()), + BacklogHealth::Unknown + ); + + cleanup(client, &schema); + + let (mut client, schema) = isolated_client(); + insert_propagation(&mut client, "future_propagation_alpha", "pending", 12_000); + let propagation_evidence = probe_postgres_data_rights_backlog(&mut client).unwrap(); + assert_eq!( + classify_postgres_data_rights_backlog(&propagation_evidence, 11_999, &policy()), + BacklogHealth::Unknown + ); + cleanup(client, &schema); +} + +#[test] +fn data_rights_probe_rejects_invalid_propagation_time_without_invalid_request_time() { + let mut client = test_client(); + let nonce = SCHEMA_NONCE.fetch_add(1, Ordering::Relaxed); + let schema = format!( + "data_rights_backlog_invalid_propagation_{}_{}", + std::process::id(), + nonce + ); + client + .batch_execute(&format!( + "CREATE SCHEMA {schema}; SET search_path TO {schema};\ + CREATE TABLE data_rights_request_state (\ + current_state TEXT, requested_at_unix_ms BIGINT);\ + CREATE TABLE data_rights_propagation_state (\ + current_state TEXT, latest_event_at_unix_ms BIGINT);\ + INSERT INTO data_rights_request_state VALUES ('requested', 2_000);\ + INSERT INTO data_rights_propagation_state VALUES ('pending', 0);" + )) + .unwrap(); + let error = probe_postgres_data_rights_backlog(&mut client).unwrap_err(); + assert!(matches!( + error, + PostgresBacklogProbeError::InvalidStoredValue + )); + client + .batch_execute(&format!( + "SET search_path TO public; DROP SCHEMA {schema} CASCADE;" + )) + .unwrap(); +} + +#[test] +fn data_rights_probe_rejects_invalid_stored_time_and_surfaces_database_failure() { + let mut client = test_client(); + let nonce = SCHEMA_NONCE.fetch_add(1, Ordering::Relaxed); + let schema = format!( + "data_rights_backlog_invalid_{}_{}", + std::process::id(), + nonce + ); + client + .batch_execute(&format!( + "CREATE SCHEMA {schema}; SET search_path TO {schema};\ + CREATE TABLE data_rights_request_state (\ + current_state TEXT, requested_at_unix_ms BIGINT);\ + CREATE TABLE data_rights_propagation_state (\ + current_state TEXT, latest_event_at_unix_ms BIGINT);\ + INSERT INTO data_rights_request_state VALUES ('requested', -1);" + )) + .unwrap(); + let error = probe_postgres_data_rights_backlog(&mut client).unwrap_err(); + assert!(matches!( + error, + PostgresBacklogProbeError::InvalidStoredValue + )); + assert!(error.source().is_none()); + + client + .batch_execute(&format!( + "SET search_path TO public; DROP SCHEMA {schema} CASCADE;" + )) + .unwrap(); + let mut transaction = client.transaction().unwrap(); + transaction.batch_execute("SELECT 1 / 0").unwrap_err(); + let error = probe_postgres_data_rights_backlog(&mut transaction).unwrap_err(); + assert!(matches!(error, PostgresBacklogProbeError::Database(_))); + assert!(error.source().is_some()); +} diff --git a/tests/postgres_integration_backlog_health.rs b/tests/postgres_integration_backlog_health.rs new file mode 100644 index 00000000..d58dfad4 --- /dev/null +++ b/tests/postgres_integration_backlog_health.rs @@ -0,0 +1,393 @@ +//! Real `PostgreSQL` contract for integration-backlog health evidence. + +use postgres::{Client, NoTls}; +use psychometrics_commons_runtime::health::BacklogHealth; +use psychometrics_commons_runtime::postgres_health::{ + classify_postgres_integration_backlog, probe_postgres_integration_backlog, + IntegrationBacklogPolicy, PostgresBacklogProbeError, +}; +use psychometrics_commons_runtime::postgres_inbox_consumption::apply_inbox_consumption_migration; +use psychometrics_commons_runtime::postgres_integration::apply_integration_migration; +use std::error::Error; +use std::sync::atomic::{AtomicU64, Ordering}; + +static SCHEMA_NONCE: AtomicU64 = AtomicU64::new(1); + +fn test_client() -> Client { + let connection = std::env::var("TEST_DATABASE_URL") + .expect("TEST_DATABASE_URL must identify the isolated CI PostgreSQL database"); + Client::connect(&connection, NoTls).expect("isolated CI PostgreSQL database must be reachable") +} + +fn isolated_client() -> (Client, String) { + let mut client = test_client(); + let nonce = SCHEMA_NONCE.fetch_add(1, Ordering::Relaxed); + let schema = format!( + "integration_backlog_health_{}_{}", + std::process::id(), + nonce + ); + client + .batch_execute(&format!( + "CREATE SCHEMA {schema}; SET search_path TO {schema};" + )) + .expect("isolated backlog-health schema should be created"); + apply_integration_migration(&mut client).expect("integration migration should apply"); + apply_inbox_consumption_migration(&mut client) + .expect("inbox consumption migration should apply"); + (client, schema) +} + +fn cleanup(mut client: Client, schema: &str) { + client + .batch_execute(&format!( + "SET search_path TO public; DROP SCHEMA {schema} CASCADE;" + )) + .expect("isolated backlog-health schema should be removed"); +} + +fn policy() -> IntegrationBacklogPolicy { + IntegrationBacklogPolicy { + max_pending_outbox_count: 4, + max_pending_outbox_age_ms: 5_000, + max_quarantined_outbox_count: 1, + max_active_consumption_count: 4, + max_active_consumption_age_ms: 5_000, + max_quarantined_consumption_count: 1, + } +} + +fn insert_outbox(client: &mut Client, event_ref: &str, state: &str, event_at: i64) { + client + .execute( + "INSERT INTO integration_outbox (\ + event_ref, event_type, schema_version, source_ref, tenant_ref, subject_ref,\ + occurred_at_unix_ms, correlation_ref, payload_digest, max_attempts,\ + current_state, latest_event_at_unix_ms\ + ) VALUES ($1,'assessment.result.changed','v1','psychometrics_commons',\ + 'tenant_backlog_alpha','subject_backlog_alpha',$2,\ + 'correlation_backlog_alpha',\ + 'sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa',\ + 3,$3,$2)", + &[&event_ref, &event_at, &state], + ) + .unwrap(); +} + +fn insert_consumption(client: &mut Client, suffix: &str, state: &str, event_at: i64) { + let source_event_ref = format!("source_event_{suffix}"); + let consumption_ref = format!("consumption_{suffix}"); + let side_effect_ref = format!("side_effect_{suffix}"); + client + .execute( + "INSERT INTO integration_inbox (\ + consumer_ref, source_ref, tenant_ref, source_event_ref, event_type,\ + schema_version, subject_ref, payload_digest, received_at_unix_ms\ + ) VALUES ('consumer_backlog_alpha','upstream_backlog_alpha',\ + 'tenant_backlog_alpha',$1,'assessment.result.changed','v1',\ + 'subject_backlog_alpha',\ + 'sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb',$2)", + &[&source_event_ref, &event_at], + ) + .unwrap(); + + match state { + "pending" => { + client + .execute( + "INSERT INTO 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\ + ) VALUES ('consumer_backlog_alpha','upstream_backlog_alpha',\ + 'tenant_backlog_alpha',$1,$2,$3,'pending',0,$4)", + &[ + &source_event_ref, + &consumption_ref, + &side_effect_ref, + &event_at, + ], + ) + .unwrap(); + } + "processing" => { + let claim_expires_at = event_at + 1_000; + client + .execute( + "INSERT INTO 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, claim_deadline_at\ + ) VALUES ('consumer_backlog_alpha','upstream_backlog_alpha',\ + 'tenant_backlog_alpha',$1,$2,$3,'processing',1,$4,$5,\ + clock_timestamp() + INTERVAL '1 second')", + &[ + &source_event_ref, + &consumption_ref, + &side_effect_ref, + &event_at, + &claim_expires_at, + ], + ) + .unwrap(); + } + "quarantined" => { + let cause_code = format!("cause_{suffix}"); + client + .execute( + "INSERT INTO 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, cause_code\ + ) VALUES ('consumer_backlog_alpha','upstream_backlog_alpha',\ + 'tenant_backlog_alpha',$1,$2,$3,'quarantined',0,$4,$5)", + &[ + &source_event_ref, + &consumption_ref, + &side_effect_ref, + &event_at, + &cause_code, + ], + ) + .unwrap(); + } + other => panic!("unsupported test state: {other}"), + } +} + +#[test] +fn empty_integration_backlog_is_observable_without_inventing_service_levels() { + let (mut client, schema) = isolated_client(); + let evidence = probe_postgres_integration_backlog(&mut client).unwrap(); + + assert_eq!(evidence.pending_outbox_count(), 0); + assert_eq!(evidence.quarantined_outbox_count(), 0); + assert_eq!(evidence.active_consumption_count(), 0); + assert_eq!(evidence.quarantined_consumption_count(), 0); + assert_eq!(evidence.oldest_pending_outbox_event_at_unix_ms(), None); + assert_eq!(evidence.oldest_active_consumption_event_at_unix_ms(), None); + assert_eq!( + classify_postgres_integration_backlog(&evidence, 10_000, &policy()), + BacklogHealth::WithinBounds + ); + + cleanup(client, &schema); +} + +#[test] +fn probe_reports_pending_processing_and_quarantined_work_without_payload_data() { + let (mut client, schema) = isolated_client(); + insert_outbox(&mut client, "event_pending_alpha", "pending", 2_000); + insert_outbox(&mut client, "event_quarantined_alpha", "quarantined", 2_500); + insert_consumption(&mut client, "pending_alpha", "pending", 3_000); + insert_consumption(&mut client, "processing_alpha", "processing", 3_500); + insert_consumption(&mut client, "quarantined_alpha", "quarantined", 4_000); + + let evidence = probe_postgres_integration_backlog(&mut client).unwrap(); + assert_eq!(evidence.pending_outbox_count(), 1); + assert_eq!(evidence.quarantined_outbox_count(), 1); + assert_eq!(evidence.active_consumption_count(), 2); + assert_eq!(evidence.quarantined_consumption_count(), 1); + assert_eq!( + evidence.oldest_pending_outbox_event_at_unix_ms(), + Some(2_000) + ); + assert_eq!( + evidence.oldest_active_consumption_event_at_unix_ms(), + Some(3_000) + ); + assert_eq!( + classify_postgres_integration_backlog(&evidence, 5_000, &policy()), + BacklogHealth::WithinBounds + ); + + cleanup(client, &schema); +} + +#[test] +fn caller_policy_can_fail_backlog_closed_by_count_or_age_without_hard_coded_slo() { + let (mut client, schema) = isolated_client(); + insert_outbox(&mut client, "event_pending_policy", "pending", 2_000); + insert_consumption(&mut client, "pending_policy", "pending", 3_000); + let evidence = probe_postgres_integration_backlog(&mut client).unwrap(); + + let strict_outbox_count = IntegrationBacklogPolicy { + max_pending_outbox_count: 0, + ..policy() + }; + assert_eq!( + classify_postgres_integration_backlog(&evidence, 4_000, &strict_outbox_count), + BacklogHealth::Stalled + ); + + let strict_consumption_count = IntegrationBacklogPolicy { + max_active_consumption_count: 0, + ..policy() + }; + assert_eq!( + classify_postgres_integration_backlog(&evidence, 4_000, &strict_consumption_count), + BacklogHealth::Stalled + ); + + assert_eq!( + classify_postgres_integration_backlog(&evidence, 8_001, &policy()), + BacklogHealth::Stalled + ); + + cleanup(client, &schema); +} + +#[test] +fn quarantine_limits_are_independent_operator_policy_inputs() { + let (mut client, schema) = isolated_client(); + insert_outbox( + &mut client, + "event_quarantined_policy", + "quarantined", + 2_000, + ); + insert_consumption(&mut client, "quarantined_policy", "quarantined", 2_500); + let evidence = probe_postgres_integration_backlog(&mut client).unwrap(); + + let strict_outbox_quarantine = IntegrationBacklogPolicy { + max_quarantined_outbox_count: 0, + ..policy() + }; + assert_eq!( + classify_postgres_integration_backlog(&evidence, 3_000, &strict_outbox_quarantine), + BacklogHealth::Stalled + ); + + let strict_consumption_quarantine = IntegrationBacklogPolicy { + max_quarantined_consumption_count: 0, + ..policy() + }; + assert_eq!( + classify_postgres_integration_backlog(&evidence, 3_000, &strict_consumption_quarantine), + BacklogHealth::Stalled + ); + + cleanup(client, &schema); +} + +#[test] +fn future_or_missing_observation_time_is_unknown_not_falsely_healthy() { + let (mut client, schema) = isolated_client(); + insert_outbox(&mut client, "event_future_alpha", "pending", 10_000); + let evidence = probe_postgres_integration_backlog(&mut client).unwrap(); + + assert_eq!( + classify_postgres_integration_backlog(&evidence, 0, &policy()), + BacklogHealth::Unknown + ); + assert_eq!( + classify_postgres_integration_backlog(&evidence, 9_999, &policy()), + BacklogHealth::Unknown + ); + + cleanup(client, &schema); +} + +#[test] +fn future_active_consumption_evidence_is_unknown_without_pending_outbox_work() { + let (mut client, schema) = isolated_client(); + insert_consumption(&mut client, "future_processing_alpha", "processing", 10_000); + let evidence = probe_postgres_integration_backlog(&mut client).unwrap(); + + assert_eq!(evidence.oldest_pending_outbox_event_at_unix_ms(), None); + assert_eq!( + classify_postgres_integration_backlog(&evidence, 9_999, &policy()), + BacklogHealth::Unknown + ); + + cleanup(client, &schema); +} + +#[test] +fn active_consumption_age_can_independently_stall_readiness() { + let (mut client, schema) = isolated_client(); + insert_consumption(&mut client, "old_processing_alpha", "processing", 2_000); + let evidence = probe_postgres_integration_backlog(&mut client).unwrap(); + + assert_eq!( + classify_postgres_integration_backlog(&evidence, 7_001, &policy()), + BacklogHealth::Stalled + ); + + cleanup(client, &schema); +} + +#[test] +fn probe_rejects_invalid_oldest_consumption_time_without_invalid_outbox_time() { + let mut client = test_client(); + let nonce = SCHEMA_NONCE.fetch_add(1, Ordering::Relaxed); + let schema = format!( + "integration_backlog_invalid_consumption_{}_{}", + std::process::id(), + nonce + ); + client + .batch_execute(&format!( + "CREATE SCHEMA {schema}; SET search_path TO {schema};\ + CREATE TABLE integration_outbox (current_state TEXT, latest_event_at_unix_ms BIGINT);\ + CREATE TABLE integration_consumption (consumption_state TEXT, latest_event_at_unix_ms BIGINT);\ + INSERT INTO integration_outbox VALUES ('pending', 4_000);\ + INSERT INTO integration_consumption VALUES ('pending', 0);" + )) + .unwrap(); + + let error = probe_postgres_integration_backlog(&mut client).unwrap_err(); + assert!(matches!( + error, + PostgresBacklogProbeError::InvalidStoredValue + )); + + client + .batch_execute(&format!( + "SET search_path TO public; DROP SCHEMA {schema} CASCADE;" + )) + .unwrap(); +} + +#[test] +fn probe_rejects_invalid_stored_timestamps_and_keeps_database_errors_typed() { + let mut client = test_client(); + let nonce = SCHEMA_NONCE.fetch_add(1, Ordering::Relaxed); + let schema = format!( + "integration_backlog_invalid_{}_{}", + std::process::id(), + nonce + ); + client + .batch_execute(&format!( + "CREATE SCHEMA {schema}; SET search_path TO {schema};\ + CREATE TABLE integration_outbox (current_state TEXT, latest_event_at_unix_ms BIGINT);\ + CREATE TABLE integration_consumption (consumption_state TEXT, latest_event_at_unix_ms BIGINT);\ + INSERT INTO integration_outbox VALUES ('pending', -1);" + )) + .unwrap(); + + let error = probe_postgres_integration_backlog(&mut client).unwrap_err(); + assert!(matches!( + error, + PostgresBacklogProbeError::InvalidStoredValue + )); + assert_eq!( + error.to_string(), + "stored backlog evidence violates the persistence contract" + ); + assert!(error.source().is_none()); + + client + .batch_execute(&format!( + "SET search_path TO public; DROP SCHEMA {schema} CASCADE;" + )) + .unwrap(); + + let mut transaction = client.transaction().unwrap(); + transaction.batch_execute("SELECT 1 / 0").unwrap_err(); + let error = probe_postgres_integration_backlog(&mut transaction).unwrap_err(); + assert!(matches!(error, PostgresBacklogProbeError::Database(_))); + assert_eq!(error.to_string(), "PostgreSQL backlog probe failed"); + assert!(error.source().is_some()); +} diff --git a/tests/postgres_recovery_invariants.rs b/tests/postgres_recovery_invariants.rs index e8af1d61..7ef9bb01 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, TIMESTAMPTZ '1970-01-01 00:00:13+00', 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,10 @@ 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), + "restored processing claim must keep its database-authoritative deadline" + ); let restored_snapshot = client .query_one( diff --git a/tests/postgres_runtime_readiness.rs b/tests/postgres_runtime_readiness.rs index eec8d1e9..0744ab47 100644 --- a/tests/postgres_runtime_readiness.rs +++ b/tests/postgres_runtime_readiness.rs @@ -146,3 +146,11 @@ fn relation_integrity_probe_surfaces_a_database_failure() { assert!(probe_postgres_relation_integrity(&mut transaction, &["pg_catalog.pg_class"]).is_err()); } + +#[test] +fn relation_integrity_probe_surfaces_a_closed_connection_failure() { + let mut client = test_client(); + let _ = client.batch_execute("SELECT pg_terminate_backend(pg_backend_pid())"); + + assert!(probe_postgres_relation_integrity(&mut client, &["pg_catalog.pg_class"]).is_err()); +} diff --git a/tests/postgres_scoring_job_backlog_health.rs b/tests/postgres_scoring_job_backlog_health.rs new file mode 100644 index 00000000..c37b049e --- /dev/null +++ b/tests/postgres_scoring_job_backlog_health.rs @@ -0,0 +1,362 @@ +//! Real `PostgreSQL` contract for scoring-job backlog health evidence. + +use postgres::{Client, NoTls}; +use psychometrics_commons_runtime::health::BacklogHealth; +use psychometrics_commons_runtime::postgres_health::{ + classify_postgres_scoring_job_backlog, probe_postgres_scoring_job_backlog, + PostgresBacklogProbeError, ScoringJobBacklogPolicy, +}; +use psychometrics_commons_runtime::postgres_scoring_job::apply_scoring_job_migration; +use std::error::Error; +use std::sync::atomic::{AtomicU64, Ordering}; + +static SCHEMA_NONCE: AtomicU64 = AtomicU64::new(1); + +fn test_client() -> Client { + let connection = std::env::var("TEST_DATABASE_URL") + .expect("TEST_DATABASE_URL must identify the isolated CI PostgreSQL database"); + Client::connect(&connection, NoTls).expect("isolated CI PostgreSQL database must be reachable") +} + +fn isolated_client() -> (Client, String) { + let mut client = test_client(); + let nonce = SCHEMA_NONCE.fetch_add(1, Ordering::Relaxed); + let schema = format!("scoring_job_backlog_{}_{}", std::process::id(), nonce); + client + .batch_execute(&format!( + "CREATE SCHEMA {schema}; SET search_path TO {schema};" + )) + .expect("isolated scoring-job backlog schema should be created"); + apply_scoring_job_migration(&mut client).expect("scoring-job migration should apply"); + (client, schema) +} + +fn cleanup(mut client: Client, schema: &str) { + client + .batch_execute(&format!( + "SET search_path TO public; DROP SCHEMA {schema} CASCADE;" + )) + .expect("isolated scoring-job backlog schema should be removed"); +} + +fn policy() -> ScoringJobBacklogPolicy { + ScoringJobBacklogPolicy { + max_active_job_count: 4, + max_active_job_age_ms: 5_000, + max_quarantined_job_count: 1, + } +} + +fn insert_job(client: &mut Client, job_ref: &str, state: &str, created_at_unix_ms: i64) { + let request_ref = format!("scoring_request_{job_ref}"); + let created_at = format!( + "TIMESTAMPTZ '1970-01-01 00:00:00+00' + INTERVAL '{created_at_unix_ms} milliseconds'" + ); + match state { + "queued" => { + client + .batch_execute(&format!( + "INSERT INTO scoring_job_state (\ + scoring_job_ref, scoring_request_ref, scoring_state, attempt_count,\ + max_attempts, created_at, updated_at\ + ) VALUES (\ + '{job_ref}', '{request_ref}', 'queued', 0, 3, {created_at}, {created_at}\ + )" + )) + .unwrap(); + } + "leased" => { + client + .batch_execute(&format!( + "INSERT INTO scoring_job_state (\ + scoring_job_ref, scoring_request_ref, scoring_state, attempt_count,\ + max_attempts, active_worker_ref, active_lease_ref, active_fencing_token,\ + active_lease_expires_at_unix_ms, created_at, updated_at\ + ) VALUES (\ + '{job_ref}', '{request_ref}', 'leased', 1, 3, 'worker_{job_ref}',\ + 'lease_{job_ref}', 1, {expiry}, {created_at}, {created_at}\ + )", + expiry = created_at_unix_ms + 1_000 + )) + .unwrap(); + } + "retry_scheduled" => { + client + .batch_execute(&format!( + "INSERT INTO scoring_job_state (\ + scoring_job_ref, scoring_request_ref, scoring_state, attempt_count,\ + max_attempts, next_attempt_at_unix_ms, last_failure_code, created_at,\ + updated_at\ + ) VALUES (\ + '{job_ref}', '{request_ref}', 'retry_scheduled', 1, 3, {next_attempt},\ + 'retryable_failure', {created_at}, {created_at}\ + )", + next_attempt = created_at_unix_ms + 500 + )) + .unwrap(); + } + "quarantined" => { + client + .batch_execute(&format!( + "INSERT INTO scoring_job_state (\ + scoring_job_ref, scoring_request_ref, scoring_state, attempt_count,\ + max_attempts, last_failure_code, created_at, updated_at\ + ) VALUES (\ + '{job_ref}', '{request_ref}', 'quarantined', 1, 3, 'poison_failure',\ + {created_at}, {created_at}\ + )" + )) + .unwrap(); + } + "completed" => { + client + .batch_execute(&format!( + "INSERT INTO scoring_job_state (\ + scoring_job_ref, scoring_request_ref, scoring_state, attempt_count,\ + max_attempts, result_ref, completed_fencing_token, created_at, updated_at\ + ) VALUES (\ + '{job_ref}', '{request_ref}', 'completed', 1, 3, 'result_{job_ref}', 1,\ + {created_at}, {created_at}\ + )" + )) + .unwrap(); + } + "cancelled" => { + client + .batch_execute(&format!( + "INSERT INTO scoring_job_state (\ + scoring_job_ref, scoring_request_ref, scoring_state, attempt_count,\ + max_attempts, created_at, updated_at\ + ) VALUES (\ + '{job_ref}', '{request_ref}', 'cancelled', 0, 3, {created_at}, {created_at}\ + )" + )) + .unwrap(); + } + other => panic!("unsupported scoring-job test state: {other}"), + } +} + +fn index_definition(client: &mut Client, index_name: &str) -> String { + client + .query_one( + "SELECT indexdef FROM pg_indexes \ + WHERE schemaname = current_schema() AND indexname = $1", + &[&index_name], + ) + .unwrap_or_else(|error| panic!("index {index_name} must exist: {error}")) + .get(0) +} + +#[test] +fn empty_scoring_job_backlog_is_observable_without_inventing_service_levels() { + let (mut client, schema) = isolated_client(); + let evidence = probe_postgres_scoring_job_backlog(&mut client).unwrap(); + + assert_eq!(evidence.active_job_count(), 0); + assert_eq!(evidence.quarantined_job_count(), 0); + assert_eq!(evidence.oldest_active_job_at_unix_ms(), None); + assert_eq!( + classify_postgres_scoring_job_backlog(&evidence, 10_000, &policy()), + BacklogHealth::WithinBounds + ); + + cleanup(client, &schema); +} + +#[test] +fn probe_counts_queued_leased_retry_and_quarantine_without_terminal_or_identity_data() { + let (mut client, schema) = isolated_client(); + insert_job(&mut client, "scoring_job_queued_alpha", "queued", 2_000); + insert_job(&mut client, "scoring_job_leased_alpha", "leased", 2_500); + insert_job( + &mut client, + "scoring_job_retry_alpha", + "retry_scheduled", + 3_000, + ); + insert_job( + &mut client, + "scoring_job_quarantined_alpha", + "quarantined", + 3_500, + ); + insert_job( + &mut client, + "scoring_job_completed_alpha", + "completed", + 1_000, + ); + insert_job( + &mut client, + "scoring_job_cancelled_alpha", + "cancelled", + 1_500, + ); + + let evidence = probe_postgres_scoring_job_backlog(&mut client).unwrap(); + assert_eq!(evidence.active_job_count(), 3); + assert_eq!(evidence.quarantined_job_count(), 1); + assert_eq!(evidence.oldest_active_job_at_unix_ms(), Some(2_000)); + assert_eq!( + classify_postgres_scoring_job_backlog(&evidence, 5_000, &policy()), + BacklogHealth::WithinBounds + ); + + cleanup(client, &schema); +} + +#[test] +fn caller_policy_can_fail_scoring_backlog_closed_by_count_or_age() { + let (mut client, schema) = isolated_client(); + insert_job(&mut client, "scoring_job_policy_queued", "queued", 2_000); + let evidence = probe_postgres_scoring_job_backlog(&mut client).unwrap(); + + let strict_count = ScoringJobBacklogPolicy { + max_active_job_count: 0, + ..policy() + }; + assert_eq!( + classify_postgres_scoring_job_backlog(&evidence, 4_000, &strict_count), + BacklogHealth::Stalled + ); + + assert_eq!( + classify_postgres_scoring_job_backlog(&evidence, 7_001, &policy()), + BacklogHealth::Stalled + ); + + cleanup(client, &schema); +} + +#[test] +fn quarantine_limit_is_an_independent_operator_policy_input() { + let (mut client, schema) = isolated_client(); + insert_job( + &mut client, + "scoring_job_policy_quarantine", + "quarantined", + 2_000, + ); + let evidence = probe_postgres_scoring_job_backlog(&mut client).unwrap(); + + let strict_quarantine = ScoringJobBacklogPolicy { + max_quarantined_job_count: 0, + ..policy() + }; + assert_eq!( + classify_postgres_scoring_job_backlog(&evidence, 3_000, &strict_quarantine), + BacklogHealth::Stalled + ); + + cleanup(client, &schema); +} + +#[test] +fn future_or_missing_observation_time_is_unknown_not_falsely_healthy() { + let (mut client, schema) = isolated_client(); + insert_job(&mut client, "scoring_job_future_alpha", "queued", 10_000); + let evidence = probe_postgres_scoring_job_backlog(&mut client).unwrap(); + + assert_eq!( + classify_postgres_scoring_job_backlog(&evidence, 0, &policy()), + BacklogHealth::Unknown + ); + assert_eq!( + classify_postgres_scoring_job_backlog(&evidence, 9_999, &policy()), + BacklogHealth::Unknown + ); + + cleanup(client, &schema); +} + +#[test] +fn non_positive_created_at_fails_closed_instead_of_normalizing() { + let (mut client, schema) = isolated_client(); + client + .batch_execute( + "INSERT INTO scoring_job_state (\ + scoring_job_ref, scoring_request_ref, scoring_state, attempt_count,\ + max_attempts, created_at, updated_at\ + ) VALUES (\ + 'scoring_job_invalid_created', 'scoring_request_invalid_created', 'queued', 0, 3,\ + TIMESTAMPTZ '1970-01-01 00:00:00+00', TIMESTAMPTZ '1970-01-01 00:00:00+00'\ + )", + ) + .unwrap(); + + assert!(matches!( + probe_postgres_scoring_job_backlog(&mut client), + Err(PostgresBacklogProbeError::InvalidStoredValue) + )); + + cleanup(client, &schema); +} + +#[test] +fn scoring_job_apply_path_creates_partial_readiness_indexes_idempotently() { + let (mut client, schema) = isolated_client(); + + for index_name in [ + "scoring_job_state_active_health_idx", + "scoring_job_state_quarantined_health_idx", + ] { + let definition = index_definition(&mut client, index_name).to_ascii_lowercase(); + assert!( + definition.contains("scoring_job_state"), + "{index_name} must index scoring_job_state: {definition}" + ); + assert!( + definition.contains("created_at"), + "{index_name} must cover created_at: {definition}" + ); + assert!( + definition.contains(" where "), + "{index_name} must be partial so terminal history does not dominate readiness: {definition}" + ); + assert!( + definition.contains("scoring_state"), + "{index_name} predicate must constrain scoring_state: {definition}" + ); + } + + let active = + index_definition(&mut client, "scoring_job_state_active_health_idx").to_ascii_lowercase(); + for state in ["queued", "leased", "retry_scheduled"] { + assert!( + active.contains(state), + "active readiness index must include {state}: {active}" + ); + } + let quarantined = index_definition(&mut client, "scoring_job_state_quarantined_health_idx") + .to_ascii_lowercase(); + assert!( + quarantined.contains("quarantined"), + "quarantine readiness index must include quarantined: {quarantined}" + ); + + apply_scoring_job_migration(&mut client) + .expect("scoring-job health index apply must be idempotent"); + + cleanup(client, &schema); +} + +#[test] +fn scoring_job_probe_supports_transaction_success_and_direct_client_failure() { + let (mut client, schema) = isolated_client(); + + let mut transaction = client.transaction().unwrap(); + let evidence = probe_postgres_scoring_job_backlog(&mut transaction).unwrap(); + assert_eq!(evidence.active_job_count(), 0); + transaction.rollback().unwrap(); + + client + .batch_execute("DROP TABLE scoring_job_state CASCADE;") + .unwrap(); + let error = probe_postgres_scoring_job_backlog(&mut client).unwrap_err(); + assert!(matches!(error, PostgresBacklogProbeError::Database(_))); + assert_eq!(error.to_string(), "PostgreSQL backlog probe failed"); + assert!(error.source().is_some()); + + cleanup(client, &schema); +}