diff --git a/CHANGELOG.md b/CHANGELOG.md index 0f83a9137..c1cc6e879 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ All notable changes to TEPP are documented here. The format follows Keep a Chang ### Added +- `persistence_postgres` backup/restore integrity: restored snapshots stay unusable until tenant, canonical `SHA-256`, knowledge-cutoff eligibility, temporal window order, and append-only triggers revalidate; SQL probes raise `restore integrity failed` (ADR 0013). - `persistence_postgres` concurrent document-write stress: atomic revise `DO` block that requires exactly one open `system_to` close, SQLSTATE mapping onto `ConcurrentWriteConflict` / `DuplicateDocumentRecord`, and live multi-session insert/revise/append-only proofs. No new migration number. - `tepp_api` naruon HTTP interchange: versioned `https` POST contracts for analysis-run create and modular export authorization that refuse table-access URLs, review/Copilot credential headers, reserved standard-header redefinition, principal-only export idempotency keys, and lexical inference claims (ADR 0011). - `persistence_postgres` audit-event SQL contracts: append-only insert that refuses empty, oversized, or hostile `action_code` values before SQL is rendered. diff --git a/crates/persistence_postgres/src/concurrent_write.rs b/crates/persistence_postgres/src/concurrent_write.rs index f1067f081..8febd9fd2 100644 --- a/crates/persistence_postgres/src/concurrent_write.rs +++ b/crates/persistence_postgres/src/concurrent_write.rs @@ -14,10 +14,13 @@ pub const DEADLOCK_DETECTED_SQLSTATE: &str = "40P01"; /// `PostgreSQL` `exclusion_violation` SQLSTATE. pub const EXCLUSION_VIOLATION_SQLSTATE: &str = "23P01"; +/// `PostgreSQL` `lock_not_available` SQLSTATE (`FOR UPDATE NOWAIT`). +pub const LOCK_NOT_AVAILABLE_SQLSTATE: &str = "55P03"; + /// Map a `PostgreSQL` SQLSTATE from a racing write onto a domain error. /// /// Unique identity collisions stay [`PersistenceError::DuplicateDocumentRecord`]. -/// Serialization, deadlock, and exclusion failures become +/// Serialization, deadlock, exclusion, and `NOWAIT` lock failures become /// [`PersistenceError::ConcurrentWriteConflict`]. Other codes stay unmapped so /// the transport can fail closed as a generic execution error. #[must_use] @@ -26,7 +29,8 @@ pub fn classify_write_conflict(sqlstate: &str) -> Option { UNIQUE_VIOLATION_SQLSTATE => Some(PersistenceError::DuplicateDocumentRecord), SERIALIZATION_FAILURE_SQLSTATE | DEADLOCK_DETECTED_SQLSTATE - | EXCLUSION_VIOLATION_SQLSTATE => Some(PersistenceError::ConcurrentWriteConflict), + | EXCLUSION_VIOLATION_SQLSTATE + | LOCK_NOT_AVAILABLE_SQLSTATE => Some(PersistenceError::ConcurrentWriteConflict), _ => None, } } @@ -34,8 +38,8 @@ pub fn classify_write_conflict(sqlstate: &str) -> Option { #[cfg(test)] mod tests { use super::{ - DEADLOCK_DETECTED_SQLSTATE, EXCLUSION_VIOLATION_SQLSTATE, SERIALIZATION_FAILURE_SQLSTATE, - UNIQUE_VIOLATION_SQLSTATE, classify_write_conflict, + DEADLOCK_DETECTED_SQLSTATE, EXCLUSION_VIOLATION_SQLSTATE, LOCK_NOT_AVAILABLE_SQLSTATE, + SERIALIZATION_FAILURE_SQLSTATE, UNIQUE_VIOLATION_SQLSTATE, classify_write_conflict, }; use crate::PersistenceError; @@ -57,6 +61,10 @@ mod tests { classify_write_conflict(EXCLUSION_VIOLATION_SQLSTATE), Some(PersistenceError::ConcurrentWriteConflict) ); + assert_eq!( + classify_write_conflict(LOCK_NOT_AVAILABLE_SQLSTATE), + Some(PersistenceError::ConcurrentWriteConflict) + ); assert_eq!(classify_write_conflict("00000"), None); assert_eq!(classify_write_conflict(""), None); assert_eq!(classify_write_conflict("P0001"), None); diff --git a/crates/persistence_postgres/src/document_sql.rs b/crates/persistence_postgres/src/document_sql.rs index 83fe8699c..ac7a6e68b 100644 --- a/crates/persistence_postgres/src/document_sql.rs +++ b/crates/persistence_postgres/src/document_sql.rs @@ -71,6 +71,9 @@ pub fn revise_document_atomic_sql(record: &DocumentRecord) -> Result "invalid source artifact", Self::InvalidAuditEvent => "invalid audit event", Self::ConcurrentWriteConflict => "concurrent write conflict", + Self::RestoreIntegrityFailed => "restore integrity failed", }; formatter.write_str(message) } @@ -190,6 +193,10 @@ mod tests { PersistenceError::InvalidAuditEvent.to_string(), "invalid audit event" ); + assert_eq!( + PersistenceError::RestoreIntegrityFailed.to_string(), + "restore integrity failed" + ); assert_eq!( MigrationContractError::SingleWordObjectName.to_string(), "single-word database object name" diff --git a/crates/persistence_postgres/src/lib.rs b/crates/persistence_postgres/src/lib.rs index 329f0bdb1..ceb979524 100644 --- a/crates/persistence_postgres/src/lib.rs +++ b/crates/persistence_postgres/src/lib.rs @@ -17,7 +17,9 @@ //! can belong to multiple entities and projects without atomistic collapse. //! Concurrent document revises use one transactional `DO` block that requires //! exactly one open row to close, and live `SQLx` maps racing SQLSTATEs onto -//! typed conflict errors. +//! typed conflict errors. Restore integrity probes refuse to mark analytical +//! state usable until tenant, digest, cutoff, temporal windows, and append-only +//! triggers revalidate. mod artifact_sql; mod concurrent_write; @@ -35,6 +37,7 @@ mod migration; mod model_run_sql; mod naming; mod relation_sql; +mod restore_integrity; mod sql_session; mod sqlx_gate; #[cfg(feature = "live-sqlx")] @@ -55,6 +58,8 @@ pub use artifact_sql::source_artifacts_are_idempotent_matches; pub use concurrent_write::DEADLOCK_DETECTED_SQLSTATE; /// `PostgreSQL` `exclusion_violation` SQLSTATE. pub use concurrent_write::EXCLUSION_VIOLATION_SQLSTATE; +/// `PostgreSQL` `lock_not_available` SQLSTATE (`FOR UPDATE NOWAIT`). +pub use concurrent_write::LOCK_NOT_AVAILABLE_SQLSTATE; /// `PostgreSQL` `serialization_failure` SQLSTATE. pub use concurrent_write::SERIALIZATION_FAILURE_SQLSTATE; /// `PostgreSQL` `unique_violation` SQLSTATE. @@ -149,6 +154,16 @@ pub use naming::is_multi_word_snake_case; pub use relation_sql::EventRelationRecord; /// Render insert SQL for a validated event relation. pub use relation_sql::insert_event_relation_sql; +/// Opaque usable-state token after restore integrity passes. +pub use restore_integrity::RestoreUsableState; +/// Restored snapshot values that must be revalidated before use. +pub use restore_integrity::RestoredAnalyticalSnapshot; +/// Physical tables a backup/restore pair must cover. +pub use restore_integrity::backup_scope_tables; +/// Mark restored analytical state usable only after integrity revalidation. +pub use restore_integrity::mark_restored_state_usable; +/// SQL probes that fail closed on unusable restored rows. +pub use restore_integrity::restore_integrity_probe_sqls; /// Recording SQL transport for offline contract tests. pub use sql_session::RecordingSqlSession; /// Live SQL transport contract. diff --git a/crates/persistence_postgres/src/live_repository.rs b/crates/persistence_postgres/src/live_repository.rs index 6541ffb16..358493b4f 100644 --- a/crates/persistence_postgres/src/live_repository.rs +++ b/crates/persistence_postgres/src/live_repository.rs @@ -28,6 +28,7 @@ use crate::model_run_sql::{ select_model_artifacts_by_run_sql, select_model_run_by_id_sql, }; use crate::relation_sql::{EventRelationRecord, insert_event_relation_sql}; +use crate::restore_integrity::restore_integrity_probe_sqls; use crate::sql_session::{SqlSession, apply_sql_batch}; use crate::{MigrationContractError, PersistenceError}; use temporal_core::{EventTime, SystemTime}; @@ -80,6 +81,18 @@ impl LiveDocumentRepository { apply_sql_batch(&mut self.session, catalog.up_sql()).map_err(LiveMigrationError::Transport) } + /// Revalidate restored physical rows before analytical state is usable. + /// + /// # Errors + /// + /// Returns transport failures, including a mapped restore-integrity raise. + pub fn assert_restore_integrity(&mut self) -> Result<(), PersistenceError> { + for sql in restore_integrity_probe_sqls() { + self.session.execute(&sql)?; + } + Ok(()) + } + /// Insert the first system-time version of a document identity. /// /// # Errors @@ -634,6 +647,14 @@ mod tests { let applied = repo.apply_migrations(&catalog).expect("migrate"); assert!(applied >= 1); assert!(!repo.session().executed().is_empty()); + repo.assert_restore_integrity() + .expect("restore integrity probes"); + assert!( + repo.session() + .executed() + .iter() + .any(|sql| sql.contains("restore integrity failed")) + ); repo.insert(&sample_record()).expect("insert"); let mut revised = sample_record(); diff --git a/crates/persistence_postgres/src/restore_integrity.rs b/crates/persistence_postgres/src/restore_integrity.rs new file mode 100644 index 000000000..c4cb960ea --- /dev/null +++ b/crates/persistence_postgres/src/restore_integrity.rs @@ -0,0 +1,245 @@ +//! Fail-closed restore integrity before analytical state is marked usable. + +use crate::PersistenceError; +use crate::cutoff::is_cutoff_eligible; +use temporal_core::{AvailableTime, EventTime, KnowledgeCutoff}; +use uuid::Uuid; + +/// Tables that a TEPP backup must include and a restore must revalidate. +pub const BACKUP_SCOPE_TABLES: &[&str] = &[ + "tenant_record", + "source_artifact", + "document_record", + "audit_event", + "reproducibility_manifest", + "corpus_split_manifest", + "model_run", + "model_artifact", +]; + +/// Restored row values that must be revalidated before use (ADR 0013). +/// +/// A backup copy is untrusted. Callers supply the reconstructed identities +/// and clocks; [`mark_restored_state_usable`] is the only constructor of a +/// usable analytical state. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct RestoredAnalyticalSnapshot { + /// Owning tenant; missing identity fails closed. + pub tenant_record_id: Option, + /// Canonical lowercase hex `SHA-256` of the bound source bytes. + pub content_sha256: String, + /// Availability time of the restored evidence. + pub available_time: AvailableTime, + /// Knowledge cutoff that the restored fit must honor. + pub knowledge_cutoff: KnowledgeCutoff, + /// Valid-time lower bound of the restored document version. + pub valid_from: EventTime, + /// Optional valid-time upper bound; must not precede `valid_from`. + pub valid_to: Option, + /// Whether append-only immutability triggers were present after restore. + pub append_only_triggers_present: bool, +} + +/// Opaque usable-state token produced only after restore integrity passes. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct RestoreUsableState { + usable: bool, +} + +impl RestoreUsableState { + /// Whether analytical reads may proceed against the restored snapshot. + #[must_use] + pub const fn is_usable(&self) -> bool { + self.usable + } +} + +/// Return the physical tables a backup/restore pair must cover. +#[must_use] +pub fn backup_scope_tables() -> &'static [&'static str] { + BACKUP_SCOPE_TABLES +} + +/// Identity tables that must keep an enabled append-only reject trigger. +const APPEND_ONLY_TRIGGER_TABLES: &[&str] = &[ + "source_artifact", + "audit_event", + "reproducibility_manifest", + "corpus_split_manifest", + "model_run", + "model_artifact", +]; + +/// SQL probes that fail closed when restored physical rows are unusable. +/// +/// Each statement is a `DO` block that raises `restore integrity failed` when +/// a digest, tenant, temporal-window, cutoff, or append-only trigger check +/// fails. Cutoff probes bind each `source_artifact` to same-tenant +/// `reproducibility_manifest` rows and fail when no applicable cutoff exists. +/// Append-only probes require an enabled `pg_trigger` on each identity table +/// linked to `reject_append_only_mutation`, not merely a same-named function. +#[must_use] +pub fn restore_integrity_probe_sqls() -> Vec { + vec![ + probe( + "digest", + "SELECT 1 FROM source_artifact \ + WHERE content_sha256 !~ '^[0-9a-f]{64}$' \ + OR tenant_record_id IS NULL", + ), + probe( + "document_window", + "SELECT 1 FROM document_record \ + WHERE tenant_record_id IS NULL \ + OR valid_to < valid_from \ + OR system_to < system_from", + ), + probe( + "missing_manifest", + "SELECT 1 FROM source_artifact sa \ + WHERE NOT EXISTS (\ + SELECT 1 FROM reproducibility_manifest rm \ + WHERE rm.tenant_record_id = sa.tenant_record_id)", + ), + probe( + "cutoff", + "SELECT 1 FROM source_artifact sa \ + WHERE available_time > (\ + SELECT MAX(rm.knowledge_cutoff) \ + FROM reproducibility_manifest rm \ + WHERE rm.tenant_record_id = sa.tenant_record_id)", + ), + probe("append_only", &append_only_trigger_predicate()), + ] +} + +fn append_only_trigger_predicate() -> String { + let tables = APPEND_ONLY_TRIGGER_TABLES + .iter() + .map(|name| format!("'{name}'")) + .collect::>() + .join(", "); + format!( + "SELECT 1 \ + FROM unnest(ARRAY[{tables}]) AS required(table_name) \ + WHERE NOT EXISTS (\ + SELECT 1 \ + FROM pg_trigger t \ + JOIN pg_class c ON c.oid = t.tgrelid \ + JOIN pg_namespace n ON n.oid = c.relnamespace \ + JOIN pg_proc p ON p.oid = t.tgfoid \ + WHERE n.nspname = current_schema() \ + AND c.relname = required.table_name \ + AND NOT t.tgisinternal \ + AND t.tgenabled <> 'D' \ + AND p.proname = 'reject_append_only_mutation')" + ) +} + +/// Revalidate a reconstructed snapshot and mark analytical state usable. +/// +/// # Errors +/// +/// Returns [`PersistenceError::RestoreIntegrityFailed`] when the tenant is +/// missing, the digest is not canonical `SHA-256`, availability exceeds the +/// cutoff, a valid window is inverted, or append-only triggers are absent. +pub fn mark_restored_state_usable( + snapshot: &RestoredAnalyticalSnapshot, +) -> Result { + if snapshot.tenant_record_id.is_none() { + return Err(PersistenceError::RestoreIntegrityFailed); + } + if !is_canonical_sha256(&snapshot.content_sha256) { + return Err(PersistenceError::RestoreIntegrityFailed); + } + if !is_cutoff_eligible(&snapshot.available_time, &snapshot.knowledge_cutoff) { + return Err(PersistenceError::RestoreIntegrityFailed); + } + if snapshot + .valid_to + .as_ref() + .is_some_and(|until| until.instant() < snapshot.valid_from.instant()) + { + return Err(PersistenceError::RestoreIntegrityFailed); + } + if !snapshot.append_only_triggers_present { + return Err(PersistenceError::RestoreIntegrityFailed); + } + Ok(RestoreUsableState { usable: true }) +} + +fn is_canonical_sha256(value: &str) -> bool { + value.len() == 64 + && value + .bytes() + .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f')) +} + +fn probe(tag: &str, predicate: &str) -> String { + format!( + "DO $tepp_restore_{tag}$\n\ + BEGIN\n\ + IF EXISTS ({predicate}) THEN\n\ + RAISE EXCEPTION 'restore integrity failed';\n\ + END IF;\n\ + END\n\ + $tepp_restore_{tag}$" + ) +} + +#[cfg(test)] +mod tests { + use super::{ + APPEND_ONLY_TRIGGER_TABLES, RestoredAnalyticalSnapshot, backup_scope_tables, + is_canonical_sha256, mark_restored_state_usable, restore_integrity_probe_sqls, + }; + use temporal_core::{AvailableTime, EventTime, KnowledgeCutoff}; + use uuid::Uuid; + + fn sample() -> RestoredAnalyticalSnapshot { + RestoredAnalyticalSnapshot { + tenant_record_id: Some(Uuid::nil()), + content_sha256: "09".repeat(32), + available_time: AvailableTime::parse_rfc3339("2026-01-01T00:00:00Z").expect("a"), + knowledge_cutoff: KnowledgeCutoff::parse_rfc3339("2026-01-01T00:00:00Z").expect("k"), + valid_from: EventTime::parse_rfc3339("2026-01-01T00:00:00Z").expect("vf"), + valid_to: Some(EventTime::parse_rfc3339("2026-01-02T00:00:00Z").expect("vt")), + append_only_triggers_present: true, + } + } + + #[test] + fn helpers_cover_digest_and_open_window_success() { + assert!(is_canonical_sha256(&"ab".repeat(32))); + assert!(!is_canonical_sha256("AB")); + assert!(mark_restored_state_usable(&sample()).is_ok()); + assert!( + restore_integrity_probe_sqls() + .iter() + .any(|sql| sql.contains("$tepp_restore_digest$")) + ); + assert!(backup_scope_tables().len() >= 8); + } + + #[test] + fn probe_sql_fails_closed_without_manifest_or_enabled_triggers() { + let joined = restore_integrity_probe_sqls().join("\n"); + assert!(joined.contains("$tepp_restore_missing_manifest$")); + assert!(joined.contains("rm.tenant_record_id = sa.tenant_record_id")); + assert!(joined.contains("MAX(rm.knowledge_cutoff)")); + assert!(joined.contains("pg_trigger")); + assert!(joined.contains("t.tgenabled")); + assert!(joined.contains("current_schema()")); + for table in APPEND_ONLY_TRIGGER_TABLES { + assert!( + joined.contains(&format!("'{table}'")), + "missing required trigger table {table}" + ); + } + assert!(!joined.contains( + "SELECT 1 WHERE NOT EXISTS (\ + SELECT 1 FROM pg_proc \ + WHERE proname = 'reject_append_only_mutation')" + )); + } +} diff --git a/crates/persistence_postgres/tests/concurrent_write_contract.rs b/crates/persistence_postgres/tests/concurrent_write_contract.rs index eb7341543..115544f07 100644 --- a/crates/persistence_postgres/tests/concurrent_write_contract.rs +++ b/crates/persistence_postgres/tests/concurrent_write_contract.rs @@ -1,9 +1,9 @@ //! Public concurrent-write classification and atomic revise contracts. use persistence_postgres::{ - DEADLOCK_DETECTED_SQLSTATE, DocumentRecord, EXCLUSION_VIOLATION_SQLSTATE, PersistenceError, - SERIALIZATION_FAILURE_SQLSTATE, UNIQUE_VIOLATION_SQLSTATE, classify_write_conflict, - revise_document_atomic_sql, + DEADLOCK_DETECTED_SQLSTATE, DocumentRecord, EXCLUSION_VIOLATION_SQLSTATE, + LOCK_NOT_AVAILABLE_SQLSTATE, PersistenceError, SERIALIZATION_FAILURE_SQLSTATE, + UNIQUE_VIOLATION_SQLSTATE, classify_write_conflict, revise_document_atomic_sql, }; use temporal_core::{AvailableTime, EventTime, SystemTime}; @@ -27,6 +27,7 @@ fn public_conflict_classifier_and_atomic_revise_sql_are_stable() { assert_eq!(SERIALIZATION_FAILURE_SQLSTATE, "40001"); assert_eq!(DEADLOCK_DETECTED_SQLSTATE, "40P01"); assert_eq!(EXCLUSION_VIOLATION_SQLSTATE, "23P01"); + assert_eq!(LOCK_NOT_AVAILABLE_SQLSTATE, "55P03"); assert_eq!( classify_write_conflict(UNIQUE_VIOLATION_SQLSTATE), Some(PersistenceError::DuplicateDocumentRecord) @@ -35,9 +36,14 @@ fn public_conflict_classifier_and_atomic_revise_sql_are_stable() { classify_write_conflict(SERIALIZATION_FAILURE_SQLSTATE), Some(PersistenceError::ConcurrentWriteConflict) ); + assert_eq!( + classify_write_conflict(LOCK_NOT_AVAILABLE_SQLSTATE), + Some(PersistenceError::ConcurrentWriteConflict) + ); let sql = revise_document_atomic_sql(&sample_record()).expect("atomic revise"); assert!(sql.contains("DO $tepp$")); + assert!(sql.contains("FOR UPDATE NOWAIT")); assert!(sql.contains("GET DIAGNOSTICS closed_count = ROW_COUNT")); assert!(sql.contains("closed_count <> 1")); assert!(sql.contains("ERRCODE = 'serialization_failure'")); diff --git a/crates/persistence_postgres/tests/live_postgres.rs b/crates/persistence_postgres/tests/live_postgres.rs index 868f33077..2e404ad07 100644 --- a/crates/persistence_postgres/tests/live_postgres.rs +++ b/crates/persistence_postgres/tests/live_postgres.rs @@ -13,12 +13,16 @@ use persistence_postgres::{ assume_app_runtime_role_sql, clear_session_tenant_sql, open_live_sqlx_pool, require_live_sqlx_config, reset_app_runtime_role_sql, set_session_tenant_sql, }; +use std::sync::mpsc; use std::sync::{Arc, Barrier}; use std::thread; +use std::time::Duration; use temporal_core::{AvailableTime, EventTime, SystemTime}; use uuid::Uuid; -const CONCURRENT_WRITERS: usize = 4; +const CONCURRENT_WRITERS: usize = 2; +/// Wall-clock budget for concurrent proofs; hang rather than block the live job forever. +const CONCURRENT_PROOF_TIMEOUT: Duration = Duration::from_secs(90); const LIVE_GATE_ENV: &str = "TEPP_LIVE_POSTGRES"; @@ -99,6 +103,7 @@ fn live_postgres_applies_migrations_and_document_sql() { repo.session_mut() .execute("SELECT 1") .expect("SELECT 1 through live transport"); + apply_sql_timeouts(&mut repo, "5s", "60s"); let catalog = MigrationCatalog::from_embedded().expect("embedded foundation catalog"); // Best-effort reset: empty service DBs lack tables/role; re-runs clean residual objects. @@ -110,6 +115,8 @@ fn live_postgres_applies_migrations_and_document_sql() { .apply_migrations(&catalog) .expect("foundation+RLS migrations must apply on live PostgreSQL"); assert!(applied >= 1); + repo.assert_restore_integrity() + .expect("empty restored catalog must pass integrity probes"); let tenant_record_id = Uuid::now_v7(); let document_record_id = Uuid::now_v7(); @@ -192,15 +199,32 @@ fn live_postgres_applies_migrations_and_document_sql() { exercise_typed_membership_assignments(&mut repo, tenant_record_id, available, system); prove_append_only_immutability(&mut repo, &manifest); prove_temporal_interval_ordering(&mut repo, tenant_record_id, source_artifact_id); + apply_sql_timeouts(&mut repo, "3s", "30s"); prove_concurrent_document_writes(&mut repo); prove_tenant_rls_isolation(&mut repo); } +fn apply_sql_timeouts( + repo: &mut LiveDocumentRepository, + lock_timeout: &str, + statement_timeout: &str, +) { + // Fail closed instead of hanging the live job on lock or statement stalls. + repo.session_mut() + .execute(&format!("SET lock_timeout = '{lock_timeout}'")) + .expect("lock_timeout"); + repo.session_mut() + .execute(&format!("SET statement_timeout = '{statement_timeout}'")) + .expect("statement_timeout"); +} + fn open_writer_repo() -> LiveDocumentRepository { let config = require_live_sqlx_config().expect("DATABASE_URL"); let options = LiveSqlxPoolOptions::new(1, 5_000).expect("writer pool"); let pool = open_live_sqlx_pool(&config, options).expect("writer pool open"); - LiveDocumentRepository::new(pool) + let mut repo = LiveDocumentRepository::new(pool); + apply_sql_timeouts(&mut repo, "3s", "15s"); + repo } fn is_closed_write_failure(error: PersistenceError) -> bool { @@ -261,18 +285,45 @@ fn sample_document( } } +fn join_with_timeout( + handle: thread::JoinHandle, + budget: Duration, + context: &'static str, +) -> T { + let (tx, rx) = mpsc::channel(); + thread::spawn(move || { + let _ = tx.send(handle.join()); + }); + match rx.recv_timeout(budget) { + Ok(Ok(value)) => value, + Ok(Err(panic_payload)) => { + panic!("{context}: worker thread panicked: {panic_payload:?}") + } + Err(mpsc::RecvTimeoutError::Timeout) => { + panic!("{context}: worker exceeded wall-clock budget") + } + Err(mpsc::RecvTimeoutError::Disconnected) => { + panic!("{context}: worker channel disconnected") + } + } +} + fn race_identical_writes( record: &DocumentRecord, revise: bool, context: &'static str, ) -> Vec> { + // Open pools before the barrier so a failed open cannot leave peers waiting forever. + let writers: Vec<_> = (0..CONCURRENT_WRITERS) + .map(|_| open_writer_repo()) + .collect(); let barrier = Arc::new(Barrier::new(CONCURRENT_WRITERS)); - (0..CONCURRENT_WRITERS) - .map(|_| { + let handles: Vec<_> = writers + .into_iter() + .map(|mut writer| { let barrier = Arc::clone(&barrier); let record = record.clone(); thread::spawn(move || { - let mut writer = open_writer_repo(); barrier.wait(); if revise { writer.revise(&record) @@ -281,7 +332,10 @@ fn race_identical_writes( } }) }) - .map(|handle| handle.join().expect(context)) + .collect(); + handles + .into_iter() + .map(|handle| join_with_timeout(handle, CONCURRENT_PROOF_TIMEOUT, context)) .collect() } @@ -352,23 +406,24 @@ fn prove_distinct_concurrent_inserts( for (document_record_id, digest) in &pairs { seed_source_artifact(repo, tenant_record_id, *document_record_id, digest); } + let writers: Vec<_> = (0..CONCURRENT_WRITERS) + .map(|_| open_writer_repo()) + .collect(); let barrier = Arc::new(Barrier::new(CONCURRENT_WRITERS)); - let handles: Vec<_> = pairs + let handles: Vec<_> = writers .into_iter() - .map(|(document_record_id, digest)| { + .zip(pairs) + .map(|(mut writer, (document_record_id, digest))| { let barrier = Arc::clone(&barrier); let record = sample_document(document_record_id, tenant_record_id, digest, 1, system); thread::spawn(move || { - let mut writer = open_writer_repo(); barrier.wait(); writer.insert(&record) }) }) .collect(); for handle in handles { - handle - .join() - .expect("distinct insert thread") + join_with_timeout(handle, CONCURRENT_PROOF_TIMEOUT, "distinct insert thread") .expect("independent document inserts must all succeed"); } } @@ -378,13 +433,16 @@ fn prove_concurrent_append_only_reject(source_artifact_id: Uuid) { "UPDATE source_artifact SET media_type_code = 'text/hostile' \ WHERE source_artifact_id = '{source_artifact_id}'::uuid" ); + let writers: Vec<_> = (0..CONCURRENT_WRITERS) + .map(|_| open_writer_repo()) + .collect(); let barrier = Arc::new(Barrier::new(CONCURRENT_WRITERS)); - let handles: Vec<_> = (0..CONCURRENT_WRITERS) - .map(|_| { + let handles: Vec<_> = writers + .into_iter() + .map(|mut writer| { let barrier = Arc::clone(&barrier); let sql = artifact_update.clone(); thread::spawn(move || { - let mut writer = open_writer_repo(); barrier.wait(); writer.session_mut().execute(&sql) }) @@ -392,7 +450,7 @@ fn prove_concurrent_append_only_reject(source_artifact_id: Uuid) { .collect(); for handle in handles { assert!( - handle.join().expect("mutation thread").is_err(), + join_with_timeout(handle, CONCURRENT_PROOF_TIMEOUT, "mutation thread").is_err(), "concurrent append-only UPDATE must fail" ); } diff --git a/crates/persistence_postgres/tests/restore_integrity_contract.rs b/crates/persistence_postgres/tests/restore_integrity_contract.rs new file mode 100644 index 000000000..f2a8d5f8d --- /dev/null +++ b/crates/persistence_postgres/tests/restore_integrity_contract.rs @@ -0,0 +1,87 @@ +//! Restored rows are untrusted until integrity revalidation (ADR 0013). + +use persistence_postgres::{ + PersistenceError, RestoredAnalyticalSnapshot, backup_scope_tables, mark_restored_state_usable, + restore_integrity_probe_sqls, +}; +use temporal_core::{AvailableTime, EventTime, KnowledgeCutoff}; +use uuid::Uuid; + +fn valid_snapshot() -> RestoredAnalyticalSnapshot { + RestoredAnalyticalSnapshot { + tenant_record_id: Some(Uuid::nil()), + content_sha256: "ab".repeat(32), + available_time: AvailableTime::parse_rfc3339("2026-01-01T00:00:00Z").expect("a"), + knowledge_cutoff: KnowledgeCutoff::parse_rfc3339("2026-06-01T00:00:00Z").expect("k"), + valid_from: EventTime::parse_rfc3339("2026-01-01T00:00:00Z").expect("vf"), + valid_to: None, + append_only_triggers_present: true, + } +} + +#[test] +fn valid_restored_snapshot_may_be_marked_usable() { + let usable = mark_restored_state_usable(&valid_snapshot()).expect("usable"); + assert!(usable.is_usable()); +} + +#[test] +fn restore_integrity_fails_closed_on_missing_or_hostile_fields() { + let mut missing_tenant = valid_snapshot(); + missing_tenant.tenant_record_id = None; + assert_eq!( + mark_restored_state_usable(&missing_tenant), + Err(PersistenceError::RestoreIntegrityFailed) + ); + + let mut bad_digest = valid_snapshot(); + bad_digest.content_sha256 = "NOPE".into(); + assert_eq!( + mark_restored_state_usable(&bad_digest), + Err(PersistenceError::RestoreIntegrityFailed) + ); + + let mut future_available = valid_snapshot(); + future_available.available_time = + AvailableTime::parse_rfc3339("2026-12-01T00:00:00Z").expect("later"); + assert_eq!( + mark_restored_state_usable(&future_available), + Err(PersistenceError::RestoreIntegrityFailed) + ); + + let mut inverted = valid_snapshot(); + inverted.valid_to = Some(EventTime::parse_rfc3339("2025-01-01T00:00:00Z").expect("earlier")); + assert_eq!( + mark_restored_state_usable(&inverted), + Err(PersistenceError::RestoreIntegrityFailed) + ); + + let mut no_triggers = valid_snapshot(); + no_triggers.append_only_triggers_present = false; + assert_eq!( + mark_restored_state_usable(&no_triggers), + Err(PersistenceError::RestoreIntegrityFailed) + ); +} + +#[test] +fn restore_probe_sql_covers_digest_cutoff_window_and_triggers() { + let probes = restore_integrity_probe_sqls(); + let joined = probes.join("\n"); + assert!(joined.contains("content_sha256")); + assert!(joined.contains("^[0-9a-f]{64}$")); + assert!(joined.contains("available_time")); + assert!(joined.contains("valid_from")); + assert!(joined.contains("valid_to")); + assert!(joined.contains("reject_append_only_mutation")); + assert!(joined.contains("restore integrity failed")); + assert!(joined.contains("missing_manifest")); + assert!(joined.contains("rm.tenant_record_id = sa.tenant_record_id")); + assert!(joined.contains("pg_trigger")); + assert!(joined.contains("t.tgenabled")); + assert!(joined.contains("'source_artifact'")); + assert!(joined.contains("'model_artifact'")); + assert!(backup_scope_tables().contains(&"source_artifact")); + assert!(backup_scope_tables().contains(&"reproducibility_manifest")); + assert!(backup_scope_tables().contains(&"document_record")); +} diff --git a/docs/OPERABILITY.md b/docs/OPERABILITY.md index 8f1118c32..336fc7e6e 100644 --- a/docs/OPERABILITY.md +++ b/docs/OPERABILITY.md @@ -52,7 +52,7 @@ LLM-backed semantic/interpreter functions use strict bounded requests and cached ## Database target recovery -Before PostgreSQL becomes production state, prove migrations and rollback, tenant isolation/RLS, temporal/lineage constraints, idempotency/concurrency, backup/restore, retention/deletion, and reconstruction from immutable artifacts. Concurrent document first-insert and revise stress is implemented on the active PR (atomic open-row close plus typed SQLSTATE mapping) and is not a protected-main claim until integration. Backup/restore and post-restore leakage/lineage revalidation remain accepted-target. A database recovery must re-run leakage/lineage validation before analytical state is marked usable. +Before PostgreSQL becomes production state, prove migrations and rollback, tenant isolation/RLS, temporal/lineage constraints, idempotency/concurrency, backup/restore, retention/deletion, and reconstruction from immutable artifacts. Concurrent document first-insert and revise stress is implemented-main. `persistence_postgres::mark_restored_state_usable` and `assert_restore_integrity` are the current fail-closed restore gate (active PR): they revalidate tenant identity, canonical digests, same-tenant knowledge-cutoff eligibility, temporal window order, and enabled append-only triggers. They do not yet revalidate relation-aware splits or full lineage graphs; those remain separate post-restore scientific steps. The gate does not replace operator `pg_dump`/`pg_restore` runbooks. ## Model release/cutover diff --git a/docs/TRACEABILITY.md b/docs/TRACEABILITY.md index afada87ae..c29d97433 100644 --- a/docs/TRACEABILITY.md +++ b/docs/TRACEABILITY.md @@ -17,7 +17,7 @@ The full APA 7th standards/literature register remains `docs/research/standards- | time-varying cross-classified multiple membership | PRD; ADR 0003 | `membership_core` network on protected main; multilevel estimators remaining | partial | | leakage-safe availability/cutoff snapshots | PRD; ADR 0002/0013 | `corpus_split` on protected main | implemented-main | | recovery metrics (RMSE, bias, coverage, graph, temporal order, Monte Carlo SE gates) | PRD; Test Strategy; ADR 0007/0014 | `validation_core` on protected main (PR #19); SE-aware Monte Carlo gates included | implemented-main | -| PostgreSQL bitemporal/lineage persistence | ADR 0013; Architecture/ERD | `persistence_postgres` migration contracts, in-memory adapters, live SQL session/document SQL port, tenant RLS (`0002` + session GUC/role helpers), `DATABASE_URL` SQLx gate, optional `live-sqlx` `PgPool` driver, exact-head live PostgreSQL CI with isolation proof, append-only immutability triggers (`0004`), temporal interval ordering CHECKs (`0005`), typed membership assignment (`0006` implemented-main), event-relation/mention/instance SQL (#37–#39 implemented-main), source-artifact SQL (#40 implemented-main), audit-event SQL (#41 implemented-main), concurrent document-write stress (active PR); remaining physical ERD/backup remaining | partial | +| PostgreSQL bitemporal/lineage persistence | ADR 0013; Architecture/ERD | `persistence_postgres` migration contracts, in-memory adapters, live SQL session/document SQL port, tenant RLS (`0002` + session GUC/role helpers), `DATABASE_URL` SQLx gate, optional `live-sqlx` `PgPool` driver, exact-head live PostgreSQL CI with isolation proof, append-only immutability triggers (`0004`), temporal interval ordering CHECKs (`0005`), typed membership assignment (`0006` implemented-main), event-relation/mention/instance SQL (#37–#39 implemented-main), source-artifact SQL (#40 implemented-main), audit-event SQL (#41 implemented-main), concurrent document-write stress (#43 implemented-main), backup/restore integrity revalidation (active PR); remaining physical ERD constraints | partial | | known-truth temporal/event simulation manifests | PRD; TRD; Test Strategy | `tepp_simulation` on protected main; recovery metrics in `validation_core` | implemented-main | | versioned service/API contracts and exports | PRD; API contract; ADR 0011/0013 | `tepp_api` analysis-run/export/JSON-LD/GraphML contracts on protected main (PR #21); HTTP service remaining accepted-target | partial | | immutable split/run/reproducibility manifests | ADR 0013; ERD | `tepp_api` reproducibility manifest contract on protected main; `persistence_postgres` append-only SQL insert/lookup for `reproducibility_manifest`, `corpus_split_manifest`, `model_run`, and `model_artifact` (migration `0003`); full physical ERD constraints remaining | partial | diff --git a/docs/adr/0013-bitemporal-persistence-reproducibility-and-split-authority.md b/docs/adr/0013-bitemporal-persistence-reproducibility-and-split-authority.md index 1cacd8119..a593ddb16 100644 --- a/docs/adr/0013-bitemporal-persistence-reproducibility-and-split-authority.md +++ b/docs/adr/0013-bitemporal-persistence-reproducibility-and-split-authority.md @@ -1,7 +1,7 @@ # ADR 0013 — Bitemporal persistence, reproducibility manifests, and split authority **Decision status:** Accepted -**Implementation maturity:** partial — migration contracts, cutoff eligibility, in-memory bitemporal adapters, live SQL session/migration port, document SQL contracts, `DATABASE_URL` SQLx gate, optional `live-sqlx` `PgPool` open/execute driver, exact-head live PostgreSQL CI, tenant RLS (`tepp_app_runtime` + session GUC), append-only reproducibility-manifest SQL insert/lookup, model-run / model-artifact / corpus-split-manifest chain (migration `0003`), append-only immutability triggers (migration `0004`), temporal interval ordering CHECK constraints (migration `0005`), typed membership-assignment storage (migration `0006`), event-relation/mention/instance SQL, source-artifact SQL, and audit-event action-code validation implemented-main; concurrent document-write stress (atomic revise + SQLSTATE mapping) on the active PR; backup/restore remain accepted-target +**Implementation maturity:** partial — migration contracts, cutoff eligibility, in-memory bitemporal adapters, live SQL session/migration port, document SQL contracts, `DATABASE_URL` SQLx gate, optional `live-sqlx` `PgPool` open/execute driver, exact-head live PostgreSQL CI, tenant RLS (`tepp_app_runtime` + session GUC), append-only reproducibility-manifest SQL insert/lookup, model-run / model-artifact / corpus-split-manifest chain (migration `0003`), append-only immutability triggers (migration `0004`), temporal interval ordering CHECK constraints (migration `0005`), typed membership-assignment storage (migration `0006`), event-relation/mention/instance SQL, source-artifact SQL, audit-event action-code validation, and concurrent document-write stress implemented-main; backup/restore integrity revalidation on the active PR **Date:** 2026-08-12 **Supersedes:** None; complements ADR 0002 (temporal semantics), ADR 0008 (evidence identity), and ADR 0011 (service ownership). diff --git a/docs/research/backup-restore-integrity.md b/docs/research/backup-restore-integrity.md new file mode 100644 index 000000000..c6fa2d34d --- /dev/null +++ b/docs/research/backup-restore-integrity.md @@ -0,0 +1,36 @@ +# Backup/restore integrity (doctoring) + +## Scope + +A restored PostgreSQL copy is untrusted. TEPP must not mark analytical state +usable until tenant identity, canonical content digests, knowledge-cutoff +eligibility, temporal window order, and append-only triggers are revalidated +(Jensen & Snodgrass, 1999; National Institute of Standards and Technology, +2010). This slice adds that fail-closed gate without a new migration number. + +It is not a substitute for operator backup tooling, RPO/RTO measurement, or +a claim of disaster-recovery certification. + +## Authority + +Jensen, C. S., & Snodgrass, R. T. (1999). Temporal data management. *IEEE +Transactions on Knowledge and Data Engineering, 11*(1), 36–44. +https://doi.org/10.1109/69.755613 + +National Institute of Standards and Technology. (2010). *Contingency planning +guide for federal information systems* (NIST SP 800-34 Rev. 1). U.S. +Department of Commerce. https://doi.org/10.6028/NIST.SP.800-34r1 + +Restored rows can silently invert valid/system windows or drop immutability +controls. Revalidation before use is the scientific recovery contract, not +an availability SLO. + +## Verification + +- valid reconstructed snapshots may be marked usable; +- missing tenant, non-canonical digest, future-available evidence, inverted + valid windows, and missing append-only triggers fail closed; +- probe SQL raises `restore integrity failed` for digest, window, missing + same-tenant manifest, cutoff, and enabled per-table append-only triggers; +- empty `reproducibility_manifest` or disabled/missing triggers fail closed; +- live PostgreSQL CI runs the probes after applying the embedded catalog. diff --git a/docs/research/task-8-bitemporal-persistence-foundations.md b/docs/research/task-8-bitemporal-persistence-foundations.md index 47a440d79..3a47c3220 100644 --- a/docs/research/task-8-bitemporal-persistence-foundations.md +++ b/docs/research/task-8-bitemporal-persistence-foundations.md @@ -9,7 +9,7 @@ Task 8 delivers storage-contract foundations for TEPP persistence under ADR 0013 3. knowledge-cutoff eligibility (`available_time <= knowledge_cutoff`); 4. in-memory bitemporal document versions with `as_known_at` / `as_valid_at` replay and append-only audit identity. -Live SQLx repositories and tenant RLS isolation are implemented; concurrent document-write stress is implemented on the active PR (not implemented-main). Backup/restore and full relation-aware split persistence remain accepted-target follow-ons behind the same contracts. +Live SQLx repositories, tenant RLS isolation, and concurrent document-write stress are implemented-main; backup/restore integrity revalidation is on the active PR; full relation-aware split persistence remains an accepted-target follow-on behind the same contracts. ## Authoritative sources diff --git a/docs/research/task-8-live-sql-transport.md b/docs/research/task-8-live-sql-transport.md index 8a93f9bc9..d7b413188 100644 --- a/docs/research/task-8-live-sql-transport.md +++ b/docs/research/task-8-live-sql-transport.md @@ -12,7 +12,7 @@ Extends Task 8 / ADR 0013 with: 6. optional `live-sqlx` feature compiling a real `SQLx`/`PgPool` open/execute driver behind validated URL and pool options; 7. exact-head live PostgreSQL CI (`live-postgres` job) that opens the pool, applies foundation+RLS migrations, exercises document insert/revise/as-of/audit SQL, and proves tenant isolation under `tepp_app_runtime` when `TEPP_LIVE_POSTGRES=1`. -Offline/`RecordingSqlSession` backends keep deterministic default CI free of a database process; `live-sqlx` fails closed without a reachable server. Tenant RLS migration `0002` (FORCE policies + `tepp_app_runtime` + session GUC) and live isolation proof are included. Concurrent document-write stress (atomic revise + SQLSTATE mapping) is on the active PR; remaining physical ERD constraints and backup/restore remain follow-ons. +Offline/`RecordingSqlSession` backends keep deterministic default CI free of a database process; `live-sqlx` fails closed without a reachable server. Tenant RLS migration `0002` (FORCE policies + `tepp_app_runtime` + session GUC), live isolation proof, and concurrent document-write stress are included; backup/restore integrity revalidation is on the active PR; remaining physical ERD constraints remain follow-ons. ## Authority diff --git a/docs/validation/temporal-event-foundation.md b/docs/validation/temporal-event-foundation.md index e367a798f..295fbae0f 100644 --- a/docs/validation/temporal-event-foundation.md +++ b/docs/validation/temporal-event-foundation.md @@ -18,7 +18,7 @@ This report tracks exact-head scientific and engineering evidence required befor | Event mention/instance | `event_core` | partial | — | unit + fail-closed promotion | Task 5 / PR #13 | | Multiple membership | `membership_core` | partial | — | unit + ESS weights | Task 7 / PR #12 + #25 | | Forward transition DAG | `relation_graph` | implemented-main | — | unit + cycle rejection | Task 6 / PR #14 | -| Bitemporal persistence + live SQL port | `persistence_postgres` | partial | concurrent document-write stress | migration contracts + recording transport + optional PgPool + live CI + tenant RLS + `0005`/`0006` + event relation/mention/instance + source-artifact + audit-event (#37–#41 implemented-main) + atomic revise / concurrent-write SQLSTATE mapping (active PR) | Task 8 / PR #16 + #23 + #26 + #27 + #29 + #30–#42 + concurrent-write stress | +| Bitemporal persistence + live SQL port | `persistence_postgres` | partial | backup/restore integrity | migration contracts + recording transport + optional PgPool + live CI + tenant RLS + `0005`/`0006` + event relation/mention/instance + source-artifact + audit-event + concurrent-write (#37–#43 implemented-main) + restore integrity probes (active PR) | Task 8 / PR #16 + #23 + #26 + #27 + #29 + #30–#43 + restore integrity | | Leakage-safe splits | `corpus_split` | implemented-main | — | cutoff + co-partition tests | Task 9 / PR #17 | | Truth corpora / manifests | `tepp_simulation` | implemented-main | — | deterministic generator tests | Task 10 / PR #18 | | Recovery metrics | `validation_core` | implemented-main | — | RMSE/bias/coverage/MC gates | Task 11 / PR #19 |