From 1f132d4f607eea43f2b82fbe5cf45346f6477855 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 13 Aug 2026 16:26:36 +0900 Subject: [PATCH 1/9] feat(persistence): concurrent document-write stress without 0007 Atomic revise requires exactly one open system_to close in one DO block. Racing SQLSTATEs map to typed conflict errors. Live CI races identical inserts/revises and keeps append-only mutations fail-closed. --- CHANGELOG.md | 1 + .../src/concurrent_write.rs | 64 +++++ .../persistence_postgres/src/document_sql.rs | 48 +++- crates/persistence_postgres/src/error.rs | 5 + crates/persistence_postgres/src/lib.rs | 16 ++ .../src/live_repository.rs | 18 +- crates/persistence_postgres/src/sqlx_live.rs | 12 +- .../tests/concurrent_write_contract.rs | 54 +++++ .../tests/live_postgres.rs | 222 +++++++++++++++++- docs/ERD.md | 2 +- docs/OPERABILITY.md | 2 +- docs/TRACEABILITY.md | 2 +- ...nce-reproducibility-and-split-authority.md | 2 +- .../concurrent-document-write-stress.md | 29 +++ ...sk-8-bitemporal-persistence-foundations.md | 2 +- docs/research/task-8-live-sql-transport.md | 2 +- docs/validation/temporal-event-foundation.md | 2 +- 17 files changed, 467 insertions(+), 16 deletions(-) create mode 100644 crates/persistence_postgres/src/concurrent_write.rs create mode 100644 crates/persistence_postgres/tests/concurrent_write_contract.rs create mode 100644 docs/research/concurrent-document-write-stress.md diff --git a/CHANGELOG.md b/CHANGELOG.md index 4477ef811..0f83a9137 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` 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. - `persistence_postgres` event-instance SQL contracts: bitemporal insert and as-known-at lookup that refuse inverted valid/system windows and hostile type/lifecycle labels before SQL is rendered. diff --git a/crates/persistence_postgres/src/concurrent_write.rs b/crates/persistence_postgres/src/concurrent_write.rs new file mode 100644 index 000000000..f1067f081 --- /dev/null +++ b/crates/persistence_postgres/src/concurrent_write.rs @@ -0,0 +1,64 @@ +//! Classify concurrent-write SQLSTATE codes without a live server. + +use crate::PersistenceError; + +/// `PostgreSQL` `unique_violation` SQLSTATE. +pub const UNIQUE_VIOLATION_SQLSTATE: &str = "23505"; + +/// `PostgreSQL` `serialization_failure` SQLSTATE. +pub const SERIALIZATION_FAILURE_SQLSTATE: &str = "40001"; + +/// `PostgreSQL` `deadlock_detected` SQLSTATE. +pub const DEADLOCK_DETECTED_SQLSTATE: &str = "40P01"; + +/// `PostgreSQL` `exclusion_violation` SQLSTATE. +pub const EXCLUSION_VIOLATION_SQLSTATE: &str = "23P01"; + +/// Map a `PostgreSQL` SQLSTATE from a racing write onto a domain error. +/// +/// Unique identity collisions stay [`PersistenceError::DuplicateDocumentRecord`]. +/// Serialization, deadlock, and exclusion failures become +/// [`PersistenceError::ConcurrentWriteConflict`]. Other codes stay unmapped so +/// the transport can fail closed as a generic execution error. +#[must_use] +pub fn classify_write_conflict(sqlstate: &str) -> Option { + match sqlstate { + UNIQUE_VIOLATION_SQLSTATE => Some(PersistenceError::DuplicateDocumentRecord), + SERIALIZATION_FAILURE_SQLSTATE + | DEADLOCK_DETECTED_SQLSTATE + | EXCLUSION_VIOLATION_SQLSTATE => Some(PersistenceError::ConcurrentWriteConflict), + _ => None, + } +} + +#[cfg(test)] +mod tests { + use super::{ + DEADLOCK_DETECTED_SQLSTATE, EXCLUSION_VIOLATION_SQLSTATE, SERIALIZATION_FAILURE_SQLSTATE, + UNIQUE_VIOLATION_SQLSTATE, classify_write_conflict, + }; + use crate::PersistenceError; + + #[test] + fn known_sqlstates_map_and_unknown_codes_stay_unmapped() { + assert_eq!( + classify_write_conflict(UNIQUE_VIOLATION_SQLSTATE), + Some(PersistenceError::DuplicateDocumentRecord) + ); + assert_eq!( + classify_write_conflict(SERIALIZATION_FAILURE_SQLSTATE), + Some(PersistenceError::ConcurrentWriteConflict) + ); + assert_eq!( + classify_write_conflict(DEADLOCK_DETECTED_SQLSTATE), + Some(PersistenceError::ConcurrentWriteConflict) + ); + assert_eq!( + classify_write_conflict(EXCLUSION_VIOLATION_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 809b80d8b..2ba3d21c5 100644 --- a/crates/persistence_postgres/src/document_sql.rs +++ b/crates/persistence_postgres/src/document_sql.rs @@ -53,6 +53,38 @@ pub fn revise_document_sqls(record: &DocumentRecord) -> Result<[String; 2], Pers Ok([close, insert]) } +/// Render one transactional revise that fails closed unless exactly one open row closes. +/// +/// The `DO` block updates the current `system_to IS NULL` version, requires that +/// close to affect exactly one row, then inserts the successor. Concurrent +/// revisers serialize on the open-row lock; the loser raises +/// `serialization_failure` instead of leaving two open versions or a silent +/// no-op. Digest validation matches [`insert_document_sql`]. +/// +/// # Errors +/// +/// Returns [`PersistenceError::InvalidContentDigest`] when the digest is not a +/// 64-character hexadecimal `SHA-256` string. +pub fn revise_document_atomic_sql(record: &DocumentRecord) -> Result { + let insert = insert_document_sql(record)?; + Ok(format!( + "DO $tepp$ \ + DECLARE closed_count integer; \ + BEGIN \ + UPDATE document_record SET system_to = '{system_from}'::timestamptz \ + WHERE document_record_id = '{document_id}'::uuid AND system_to IS NULL; \ + GET DIAGNOSTICS closed_count = ROW_COUNT; \ + IF closed_count <> 1 THEN \ + RAISE EXCEPTION 'concurrent document revision conflict' \ + USING ERRCODE = 'serialization_failure'; \ + END IF; \ + {insert}; \ + END $tepp$", + system_from = record.system_from.to_rfc3339(), + document_id = record.document_record_id, + )) +} + /// Render as-known-at selection for one document identity. #[must_use] pub fn as_known_at_sql(document_record_id: uuid::Uuid, known_at_rfc3339: &str) -> String { @@ -149,7 +181,8 @@ fn validate_digest(digest: &str) -> Result<(), PersistenceError> { mod tests { use super::{ append_audit_sql, as_known_at_sql, as_valid_at_sql, escape_literal, insert_document_sql, - optional_timestamptz, revise_document_sqls, validate_audit_action, validate_digest, + optional_timestamptz, revise_document_atomic_sql, revise_document_sqls, validate_audit_action, + validate_digest, }; use crate::PersistenceError; use crate::document_store::{AuditEvent, DocumentRecord}; @@ -186,6 +219,19 @@ mod tests { assert!(close.contains("system_to IS NULL")); assert!(reopen.contains("INSERT INTO document_record")); + let atomic = revise_document_atomic_sql(&record).expect("atomic"); + assert!(atomic.contains("DO $tepp$")); + assert!(atomic.contains("GET DIAGNOSTICS closed_count = ROW_COUNT")); + assert!(atomic.contains("serialization_failure")); + assert!(atomic.contains("INSERT INTO document_record")); + assert_eq!( + revise_document_atomic_sql(&DocumentRecord { + content_digest: "nope".into(), + ..sample_record() + }), + Err(PersistenceError::InvalidContentDigest) + ); + let known = as_known_at_sql(uuid::Uuid::nil(), "2026-03-01T00:00:00Z"); assert!(known.contains("system_from <=")); let valid = as_valid_at_sql( diff --git a/crates/persistence_postgres/src/error.rs b/crates/persistence_postgres/src/error.rs index 5d32a4416..e1a312f72 100644 --- a/crates/persistence_postgres/src/error.rs +++ b/crates/persistence_postgres/src/error.rs @@ -38,6 +38,8 @@ pub enum PersistenceError { InvalidSourceArtifact, /// An audit action code was empty, oversized, or hostile. InvalidAuditEvent, + /// A concurrent writer won the open-row lock or serialization contest. + ConcurrentWriteConflict, } impl fmt::Display for PersistenceError { @@ -59,6 +61,7 @@ impl fmt::Display for PersistenceError { Self::ConflictingSourceArtifact => "conflicting source artifact", Self::InvalidSourceArtifact => "invalid source artifact", Self::InvalidAuditEvent => "invalid audit event", + Self::ConcurrentWriteConflict => "concurrent write conflict", }; formatter.write_str(message) } @@ -158,6 +161,8 @@ mod tests { assert_eq!( PersistenceError::InvalidMembershipAssignment.to_string(), "invalid membership assignment" + PersistenceError::ConcurrentWriteConflict.to_string(), + "concurrent write conflict" ); assert_eq!( PersistenceError::InvalidEventRelation.to_string(), diff --git a/crates/persistence_postgres/src/lib.rs b/crates/persistence_postgres/src/lib.rs index 28e5a16a1..329f0bdb1 100644 --- a/crates/persistence_postgres/src/lib.rs +++ b/crates/persistence_postgres/src/lib.rs @@ -15,8 +15,12 @@ //! contracts chain immutable run identities to those manifests. Typed //! membership-assignment SQL (migration `0006`) replaces the polymorphic 0001 stub so documents //! 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. mod artifact_sql; +mod concurrent_write; mod cutoff; mod document_sql; mod document_store; @@ -47,6 +51,16 @@ pub use artifact_sql::insert_source_artifact_sql; pub use artifact_sql::select_source_artifact_by_id_sql; /// Compare two source artifacts for idempotent-retry equality. pub use artifact_sql::source_artifacts_are_idempotent_matches; +/// `PostgreSQL` `deadlock_detected` SQLSTATE. +pub use concurrent_write::DEADLOCK_DETECTED_SQLSTATE; +/// `PostgreSQL` `exclusion_violation` SQLSTATE. +pub use concurrent_write::EXCLUSION_VIOLATION_SQLSTATE; +/// `PostgreSQL` `serialization_failure` SQLSTATE. +pub use concurrent_write::SERIALIZATION_FAILURE_SQLSTATE; +/// `PostgreSQL` `unique_violation` SQLSTATE. +pub use concurrent_write::UNIQUE_VIOLATION_SQLSTATE; +/// Map a racing-write SQLSTATE onto a domain persistence error. +pub use concurrent_write::classify_write_conflict; /// Knowledge-cutoff eligibility for historical analytical reads. pub use cutoff::is_cutoff_eligible; /// Render append-only audit insert SQL. @@ -57,6 +71,8 @@ pub use document_sql::as_known_at_sql; pub use document_sql::as_valid_at_sql; /// Render open-document insert SQL. pub use document_sql::insert_document_sql; +/// Render one transactional revise that fails closed unless one open row closes. +pub use document_sql::revise_document_atomic_sql; /// Render revise close+insert SQL pair. pub use document_sql::revise_document_sqls; /// Append-only audit event. diff --git a/crates/persistence_postgres/src/live_repository.rs b/crates/persistence_postgres/src/live_repository.rs index dc8a26093..6541ffb16 100644 --- a/crates/persistence_postgres/src/live_repository.rs +++ b/crates/persistence_postgres/src/live_repository.rs @@ -5,7 +5,8 @@ use crate::artifact_sql::{ select_source_artifact_by_id_sql, }; use crate::document_sql::{ - append_audit_sql, as_known_at_sql, as_valid_at_sql, insert_document_sql, revise_document_sqls, + append_audit_sql, as_known_at_sql, as_valid_at_sql, insert_document_sql, + revise_document_atomic_sql, }; use crate::document_store::{AuditEvent, DocumentRecord}; use crate::instance_sql::{ @@ -89,15 +90,14 @@ impl LiveDocumentRepository { self.session.execute(&sql) } - /// Close the open system-time row and insert a revision. + /// Close the open system-time row and insert a revision atomically. /// /// # Errors /// - /// Returns digest or transport failures. + /// Returns digest, concurrent-write, or transport failures. pub fn revise(&mut self, record: &DocumentRecord) -> Result<(), PersistenceError> { - let [close, insert] = revise_document_sqls(record)?; - self.session.execute(&close)?; - self.session.execute(&insert) + let sql = revise_document_atomic_sql(record)?; + self.session.execute(&sql) } /// Issue as-known-at SQL for a document identity. @@ -641,6 +641,12 @@ mod tests { revised.system_from = SystemTime::parse_rfc3339("2026-02-01T00:00:00Z").expect("later system"); repo.revise(&revised).expect("revise"); + assert!( + repo.session() + .executed() + .iter() + .any(|sql| sql.contains("DO $tepp$") && sql.contains("GET DIAGNOSTICS")) + ); repo.submit_as_known_at( uuid::Uuid::nil(), &SystemTime::parse_rfc3339("2026-02-01T00:00:00Z").expect("k"), diff --git a/crates/persistence_postgres/src/sqlx_live.rs b/crates/persistence_postgres/src/sqlx_live.rs index b05f42014..f9e1170c0 100644 --- a/crates/persistence_postgres/src/sqlx_live.rs +++ b/crates/persistence_postgres/src/sqlx_live.rs @@ -5,6 +5,7 @@ //! for the success path. Unreachable-host failure is still unit-tested. use crate::PersistenceError; +use crate::classify_write_conflict; use crate::live_pool::{LiveSqlxPool, LiveSqlxPoolOptions}; use crate::sqlx_gate::LiveSqlxConfig; use std::sync::Arc; @@ -61,6 +62,15 @@ impl SqlxTransport { self.runtime .block_on(async { sqlx::query(sql).execute(&self.pool).await }) .map(|_| ()) - .map_err(|_| PersistenceError::SqlExecutionFailed) + .map_err(|error| map_sqlx_error(&error)) } } + +fn map_sqlx_error(error: &sqlx::Error) -> PersistenceError { + error + .as_database_error() + .and_then(sqlx::error::DatabaseError::code) + .as_deref() + .and_then(classify_write_conflict) + .unwrap_or(PersistenceError::SqlExecutionFailed) +} diff --git a/crates/persistence_postgres/tests/concurrent_write_contract.rs b/crates/persistence_postgres/tests/concurrent_write_contract.rs new file mode 100644 index 000000000..eb7341543 --- /dev/null +++ b/crates/persistence_postgres/tests/concurrent_write_contract.rs @@ -0,0 +1,54 @@ +//! 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, +}; +use temporal_core::{AvailableTime, EventTime, SystemTime}; + +fn sample_record() -> DocumentRecord { + DocumentRecord { + document_record_id: uuid::Uuid::nil(), + tenant_record_id: uuid::Uuid::nil(), + content_digest: "ab".repeat(32), + available_time: AvailableTime::parse_rfc3339("2026-01-01T00:00:00Z").expect("a"), + valid_from: EventTime::parse_rfc3339("2026-01-01T00:00:00Z").expect("v"), + valid_to: None, + system_from: SystemTime::parse_rfc3339("2026-02-01T00:00:00Z").expect("s"), + system_to: None, + revision_number: 2, + } +} + +#[test] +fn public_conflict_classifier_and_atomic_revise_sql_are_stable() { + assert_eq!(UNIQUE_VIOLATION_SQLSTATE, "23505"); + assert_eq!(SERIALIZATION_FAILURE_SQLSTATE, "40001"); + assert_eq!(DEADLOCK_DETECTED_SQLSTATE, "40P01"); + assert_eq!(EXCLUSION_VIOLATION_SQLSTATE, "23P01"); + assert_eq!( + classify_write_conflict(UNIQUE_VIOLATION_SQLSTATE), + Some(PersistenceError::DuplicateDocumentRecord) + ); + assert_eq!( + classify_write_conflict(SERIALIZATION_FAILURE_SQLSTATE), + Some(PersistenceError::ConcurrentWriteConflict) + ); + + let sql = revise_document_atomic_sql(&sample_record()).expect("atomic revise"); + assert!(sql.contains("DO $tepp$")); + assert!(sql.contains("GET DIAGNOSTICS closed_count = ROW_COUNT")); + assert!(sql.contains("closed_count <> 1")); + assert!(sql.contains("ERRCODE = 'serialization_failure'")); + assert!(sql.contains("UPDATE document_record")); + assert!(sql.contains("INSERT INTO document_record")); + assert!(sql.contains("system_to IS NULL")); + + let mut invalid = sample_record(); + invalid.content_digest = "short".into(); + assert_eq!( + revise_document_atomic_sql(&invalid), + Err(PersistenceError::InvalidContentDigest) + ); +} diff --git a/crates/persistence_postgres/tests/live_postgres.rs b/crates/persistence_postgres/tests/live_postgres.rs index 67cbb1db2..2ce9b30f6 100644 --- a/crates/persistence_postgres/tests/live_postgres.rs +++ b/crates/persistence_postgres/tests/live_postgres.rs @@ -12,10 +12,18 @@ use persistence_postgres::{ ModelRunRecord, ReproducibilityManifestRecord, SqlSession, apply_sql_batch, 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, + LiveSqlxPoolOptions, MigrationCatalog, ModelArtifactRecord, ModelRunRecord, PersistenceError, + ReproducibilityManifestRecord, SqlSession, apply_sql_batch, 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::{Arc, Barrier}; +use std::thread; use temporal_core::{AvailableTime, EventTime, SystemTime}; use uuid::Uuid; +const CONCURRENT_WRITERS: usize = 4; + const LIVE_GATE_ENV: &str = "TEPP_LIVE_POSTGRES"; fn live_postgres_requested() -> bool { @@ -39,7 +47,7 @@ fn seed_tenant_and_artifact( source_artifact_id: Uuid, content_digest: &str, ) { - let (available, _valid, system) = sample_times(); + let (_available, _valid, system) = sample_times(); let tenant_sql = format!( "INSERT INTO tenant_record (tenant_record_id, tenant_status_code, system_time) \ VALUES ('{tenant_record_id}'::uuid, 'active', '{system}'::timestamptz)", @@ -48,7 +56,16 @@ fn seed_tenant_and_artifact( repo.session_mut() .execute(&tenant_sql) .expect("insert tenant_record"); + seed_source_artifact(repo, tenant_record_id, source_artifact_id, content_digest); +} +fn seed_source_artifact( + repo: &mut LiveDocumentRepository, + tenant_record_id: Uuid, + source_artifact_id: Uuid, + content_digest: &str, +) { + let (available, _valid, system) = sample_times(); let artifact_sql = format!( "INSERT INTO source_artifact (\ source_artifact_id, tenant_record_id, content_sha256, source_size_bytes, \ @@ -179,9 +196,212 @@ 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); + prove_concurrent_document_writes(&mut repo); prove_tenant_rls_isolation(&mut repo); } +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) +} + +fn is_closed_write_failure(error: PersistenceError) -> bool { + matches!( + error, + PersistenceError::DuplicateDocumentRecord + | PersistenceError::ConcurrentWriteConflict + | PersistenceError::SqlExecutionFailed + ) +} + +fn assert_single_winner(results: Vec>, context: &str) { + let successes = results.iter().filter(|result| result.is_ok()).count(); + assert_eq!(successes, 1, "{context}: expected exactly one winner"); + assert!( + results + .into_iter() + .filter_map(Result::err) + .all(is_closed_write_failure), + "{context}: losers must fail closed" + ); +} + +fn document_row_guard(document_record_id: Uuid, expected_rows: u64, expected_open: u64) -> String { + format!( + "DO $tepp$ BEGIN \ + IF (SELECT COUNT(*) FROM document_record \ + WHERE document_record_id = '{document_record_id}'::uuid) <> {expected_rows} THEN \ + RAISE EXCEPTION 'unexpected document_record row count'; \ + END IF; \ + IF (SELECT COUNT(*) FROM document_record \ + WHERE document_record_id = '{document_record_id}'::uuid \ + AND system_to IS NULL) <> {expected_open} THEN \ + RAISE EXCEPTION 'unexpected open document_record count'; \ + END IF; \ + END $tepp$" + ) +} + +fn sample_document( + document_record_id: Uuid, + tenant_record_id: Uuid, + content_digest: String, + revision_number: u64, + system: SystemTime, +) -> DocumentRecord { + let (available, valid, _) = sample_times(); + DocumentRecord { + document_record_id, + tenant_record_id, + content_digest, + available_time: available, + valid_from: valid, + valid_to: None, + system_from: system, + system_to: None, + revision_number, + } +} + +fn race_identical_writes( + record: &DocumentRecord, + revise: bool, + context: &'static str, +) -> Vec> { + let barrier = Arc::new(Barrier::new(CONCURRENT_WRITERS)); + (0..CONCURRENT_WRITERS) + .map(|_| { + 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) + } else { + writer.insert(&record) + } + }) + }) + .map(|handle| handle.join().expect(context)) + .collect() +} + +/// Concurrent first inserts and revises must leave exactly one open version. +fn prove_concurrent_document_writes( + repo: &mut LiveDocumentRepository, +) { + let tenant_record_id = Uuid::now_v7(); + let first_document_id = Uuid::now_v7(); + let first_digest = "a1".repeat(32); + seed_tenant_and_artifact(repo, tenant_record_id, first_document_id, &first_digest); + let (_, _, system) = sample_times(); + let first = sample_document(first_document_id, tenant_record_id, first_digest, 1, system); + assert_single_winner( + race_identical_writes(&first, false, "insert thread"), + "concurrent first insert", + ); + repo.session_mut() + .execute(&document_row_guard(first_document_id, 1, 1)) + .expect("exactly one first insert row"); + + let revised = sample_document( + first_document_id, + tenant_record_id, + "b2".repeat(32), + 2, + SystemTime::parse_rfc3339("2026-02-01T00:00:00Z").expect("later"), + ); + assert_single_winner( + race_identical_writes(&revised, true, "revise thread"), + "concurrent revise", + ); + repo.session_mut() + .execute(&document_row_guard(first_document_id, 2, 1)) + .expect("closed first version plus one open revision"); + + prove_missing_open_row_fails(repo, tenant_record_id); + prove_distinct_concurrent_inserts(repo, tenant_record_id, system); + prove_concurrent_append_only_reject(first_document_id); +} + +fn prove_missing_open_row_fails( + repo: &mut LiveDocumentRepository, + tenant_record_id: Uuid, +) { + let missing = sample_document( + Uuid::now_v7(), + tenant_record_id, + "c3".repeat(32), + 2, + SystemTime::parse_rfc3339("2026-03-01T00:00:00Z").expect("missing"), + ); + let missing_error = repo.revise(&missing).expect_err("missing open row"); + assert!( + is_closed_write_failure(missing_error), + "revise without an open row must fail closed" + ); +} + +fn prove_distinct_concurrent_inserts( + repo: &mut LiveDocumentRepository, + tenant_record_id: Uuid, + system: SystemTime, +) { + let pairs: Vec<(Uuid, String)> = (0..CONCURRENT_WRITERS) + .map(|index| (Uuid::now_v7(), format!("{index:02x}") + &"e5".repeat(31))) + .collect(); + for (document_record_id, digest) in &pairs { + seed_source_artifact(repo, tenant_record_id, *document_record_id, digest); + } + let barrier = Arc::new(Barrier::new(CONCURRENT_WRITERS)); + let handles: Vec<_> = pairs + .into_iter() + .map(|(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") + .expect("independent document inserts must all succeed"); + } +} + +fn prove_concurrent_append_only_reject(source_artifact_id: Uuid) { + let artifact_update = format!( + "UPDATE source_artifact SET media_type_code = 'text/hostile' \ + WHERE source_artifact_id = '{source_artifact_id}'::uuid" + ); + let barrier = Arc::new(Barrier::new(CONCURRENT_WRITERS)); + let handles: Vec<_> = (0..CONCURRENT_WRITERS) + .map(|_| { + 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) + }) + }) + .collect(); + for handle in handles { + assert!( + handle.join().expect("mutation thread").is_err(), + "concurrent append-only UPDATE must fail" + ); + } +} + /// Append-only triggers must reject UPDATE/DELETE on identity tables. fn prove_append_only_immutability( repo: &mut LiveDocumentRepository, diff --git a/docs/ERD.md b/docs/ERD.md index 651055f2b..7fc04c89f 100644 --- a/docs/ERD.md +++ b/docs/ERD.md @@ -3,7 +3,7 @@ **Status:** Accepted logical target model with current implementation maturity explicitly marked. **Last reviewed:** 2026-08-13 -Protected main implements storage-independent domain objects plus `persistence_postgres` foundation tables (`0001`), tenant row-level security (`0002`), the model-run/artifact chain (`0003`), append-only immutability triggers (`0004`), temporal interval ordering CHECKs (`0005`), typed membership assignment (`0006`), event-relation vocabulary SQL, and event-mention SQL as executable migration/application contracts with live CI. Event-instance insert/as-known-at SQL is on the active PR and is not implemented-main until exact-head checks, review, and protected-main integration complete. Broader planned ERD entities, concurrent-write acceptance, and backup/recovery gates remain accepted-target. +Protected main implements storage-independent domain objects plus `persistence_postgres` foundation tables (`0001`), tenant row-level security (`0002`), the model-run/artifact chain (`0003`), append-only immutability triggers (`0004`), temporal interval ordering CHECKs (`0005`), typed membership assignment (`0006`), event-relation/mention/instance SQL, source-artifact SQL, audit-event SQL, and naruon HTTP interchange contracts as executable migration/application contracts with live CI. Concurrent document-write stress (atomic revise + live multi-session proof) is on the active PR and is not implemented-main until exact-head checks, review, and protected-main integration complete. Broader planned ERD entities and backup/recovery gates remain accepted-target. ## Current domain foundation diff --git a/docs/OPERABILITY.md b/docs/OPERABILITY.md index 1f1bf262a..8f1118c32 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. 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 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. ## Model release/cutover diff --git a/docs/TRACEABILITY.md b/docs/TRACEABILITY.md index fcecaef2a..afada87ae 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 (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 (active PR); remaining physical ERD/backup remaining | 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 ecc2890b5..1cacd8119 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, and source-artifact SQL implemented-main; audit-event action-code validation on the active PR; concurrent write stress and 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, 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 **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/concurrent-document-write-stress.md b/docs/research/concurrent-document-write-stress.md new file mode 100644 index 000000000..f7ed096c3 --- /dev/null +++ b/docs/research/concurrent-document-write-stress.md @@ -0,0 +1,29 @@ +# Concurrent document-write stress (doctoring) + +## Scope + +This increment proves the ADR 0013 concurrent-write verification item for `document_record` without allocating migration `0007` while `0006` remains in flight. + +`LiveDocumentRepository::revise` now submits one `DO` block that: + +1. updates the unique open version (`system_to IS NULL`); +2. requires `GET DIAGNOSTICS ROW_COUNT = 1`; +3. inserts the successor revision in the same implicit transaction; and +4. raises `serialization_failure` when another session already closed the open row. + +Live `SQLx` maps PostgreSQL `unique_violation` to `DuplicateDocumentRecord` and `serialization_failure` / `deadlock_detected` / `exclusion_violation` to `ConcurrentWriteConflict`. Distinct-identity inserts remain compatible. Append-only `source_artifact` mutations still fail closed under concurrent sessions via migration `0004`. + +The increment does not add a partial unique index on the open row, persist event-level membership, or implement backup/restore. + +## Authority + +Berenson, H., Bernstein, P., Gray, J., Melton, J., O'Neil, E., & O'Neil, P. (1995). A critique of ANSI SQL isolation levels. *ACM SIGMOD Record, 24*(2), 1–10. https://doi.org/10.1145/568271.223785 + +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 + +PostgreSQL Global Development Group. (2024). *Transaction isolation*. In *PostgreSQL 16 documentation*. https://www.postgresql.org/docs/16/transaction-iso.html + +## Verification + +- unit/contract tests cover atomic revise SQL text, digest refusal, and every classified SQLSTATE including the unmapped default; +- live PostgreSQL CI (`TEPP_LIVE_POSTGRES=1`) races four sessions on the same first insert and the same revision, asserts exactly one winner, preserves the closed first version, accepts independent inserts, and rejects concurrent append-only `source_artifact` updates. diff --git a/docs/research/task-8-bitemporal-persistence-foundations.md b/docs/research/task-8-bitemporal-persistence-foundations.md index ae5c88e16..47a440d79 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 write stress, backup/restore, and full relation-aware split persistence remain accepted-target follow-ons behind the same contracts. +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. ## Authoritative sources diff --git a/docs/research/task-8-live-sql-transport.md b/docs/research/task-8-live-sql-transport.md index 05a03b45a..8a93f9bc9 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; remaining physical ERD constraints, concurrent write stress, 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) 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. ## Authority diff --git a/docs/validation/temporal-event-foundation.md b/docs/validation/temporal-event-foundation.md index 247bb5f68..e367a798f 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 | audit-event SQL | migration contracts + recording transport + optional PgPool + live CI + tenant RLS + `0005`/`0006` + event relation/mention/instance + source-artifact (#37–#40 implemented-main) + audit action-code validation (active PR) | Task 8 / PR #16 + #23 + #26 + #27 + #29 + #30–#40 + audit-event SQL | +| 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 | | 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 | From 155e6f94093b5107885b91fd6d112981680e7bee Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 13 Aug 2026 17:34:44 +0900 Subject: [PATCH 2/9] fix(persistence): rebase concurrent write stress onto main Resolve CHANGELOG/docs/error conflicts after membership 0006, restore stable error-message assertions, and keep ConcurrentWriteConflict export. --- crates/persistence_postgres/src/error.rs | 2 ++ crates/persistence_postgres/tests/live_postgres.rs | 6 +----- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/crates/persistence_postgres/src/error.rs b/crates/persistence_postgres/src/error.rs index e1a312f72..958995c82 100644 --- a/crates/persistence_postgres/src/error.rs +++ b/crates/persistence_postgres/src/error.rs @@ -161,6 +161,8 @@ mod tests { assert_eq!( PersistenceError::InvalidMembershipAssignment.to_string(), "invalid membership assignment" + ); + assert_eq!( PersistenceError::ConcurrentWriteConflict.to_string(), "concurrent write conflict" ); diff --git a/crates/persistence_postgres/tests/live_postgres.rs b/crates/persistence_postgres/tests/live_postgres.rs index 2ce9b30f6..868f33077 100644 --- a/crates/persistence_postgres/tests/live_postgres.rs +++ b/crates/persistence_postgres/tests/live_postgres.rs @@ -9,13 +9,9 @@ use persistence_postgres::{ AuditEvent, CorpusSplitManifestRecord, DocumentRecord, LiveDocumentRepository, LiveSqlxPoolOptions, MembershipAssignmentRecord, MigrationCatalog, ModelArtifactRecord, - ModelRunRecord, ReproducibilityManifestRecord, SqlSession, apply_sql_batch, + ModelRunRecord, PersistenceError, ReproducibilityManifestRecord, SqlSession, apply_sql_batch, 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, - LiveSqlxPoolOptions, MigrationCatalog, ModelArtifactRecord, ModelRunRecord, PersistenceError, - ReproducibilityManifestRecord, SqlSession, apply_sql_batch, 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::{Arc, Barrier}; use std::thread; From 4122a90a4e0a795799a79a0e274ccf15fedbd6c4 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 13 Aug 2026 18:02:40 +0900 Subject: [PATCH 3/9] ci: re-trigger exact-head checks after cancelled jobs From 7703ebb619611884331d194ef9ec22407e6972ec Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Fri, 14 Aug 2026 06:58:15 +0900 Subject: [PATCH 4/9] ci: re-trigger exact-head checks after cancelled jobs From 67eed37bcf2c237d11890602fe8e878be8176ed7 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Fri, 14 Aug 2026 07:00:19 +0900 Subject: [PATCH 5/9] ci: re-trigger exact-head checks for ready PR #43 From 7423f1688d5cee6ab0d406604983dc9840f169b2 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Fri, 14 Aug 2026 07:02:17 +0900 Subject: [PATCH 6/9] ci: single re-trigger for exact-head PR #43 From 8853e9b8b1ad299dd1e80c897369b51b81986cdd Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Fri, 14 Aug 2026 07:17:13 +0900 Subject: [PATCH 7/9] ci: re-trigger exact-head checks for PR #43 From 57f5e5e0ffcff9748b5b6cf17f469773091391d3 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Fri, 14 Aug 2026 07:18:35 +0900 Subject: [PATCH 8/9] ci: re-trigger exact-head checks for PR #43 after thrash cancel From eeab74fd3a8e60f7d2e44abca20c4e6724f01d37 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Fri, 14 Aug 2026 07:24:35 +0900 Subject: [PATCH 9/9] style(persistence): rustfmt atomic revise import grouping for CI cargo fmt --check failed on document_sql test imports after concurrent-write atomic revise. --- crates/persistence_postgres/src/document_sql.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/persistence_postgres/src/document_sql.rs b/crates/persistence_postgres/src/document_sql.rs index 2ba3d21c5..83fe8699c 100644 --- a/crates/persistence_postgres/src/document_sql.rs +++ b/crates/persistence_postgres/src/document_sql.rs @@ -181,8 +181,8 @@ fn validate_digest(digest: &str) -> Result<(), PersistenceError> { mod tests { use super::{ append_audit_sql, as_known_at_sql, as_valid_at_sql, escape_literal, insert_document_sql, - optional_timestamptz, revise_document_atomic_sql, revise_document_sqls, validate_audit_action, - validate_digest, + optional_timestamptz, revise_document_atomic_sql, revise_document_sqls, + validate_audit_action, validate_digest, }; use crate::PersistenceError; use crate::document_store::{AuditEvent, DocumentRecord};