feat(persistence): concurrent document-write stress without 0007

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.

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<PersistenceError> {
+    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);
+    }
+}

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<String, PersistenceError> {
+    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(

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)
     }
@@ -159,6 +162,10 @@ mod tests {
             PersistenceError::InvalidMembershipAssignment.to_string(),
             "invalid membership assignment"
         );
+        assert_eq!(
+            PersistenceError::ConcurrentWriteConflict.to_string(),
+            "concurrent write conflict"
+        );
         assert_eq!(
             PersistenceError::InvalidEventRelation.to_string(),
             "invalid event relation"

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.

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<S: SqlSession> LiveDocumentRepository<S> {
         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"),

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)
+}

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)
+    );
+}