feat(scoring): persist response snapshot with dispatch atomically

src/lib.rs

@@ -25,6 +25,7 @@ pub mod postgres_instrument_release;
 pub mod postgres_integration;
 pub mod postgres_response_snapshot;
 pub mod postgres_scoring_job;
+pub mod postgres_scoring_orchestration;
 pub mod postgres_scoring_request;
 mod reference;
 pub mod research_release;

src/postgres_scoring_orchestration.rs

@@ -0,0 +1,138 @@
+//! Atomic `PostgreSQL` composition for response snapshots and scoring dispatch.
+//!
+//! This adapter closes the crash boundary between freezing accepted response evidence and
+//! durably scheduling its version-pinned scoring work. It composes existing immutable snapshot,
+//! scoring-request, scoring-job, and transactional-outbox adapters in one caller-owned local
+//! transaction. Psychometric arithmetic remains in `fast-mlsirm`.
+
+use crate::integration::IntegrationEvent;
+use crate::postgres_integration::PersistenceDisposition;
+use crate::postgres_response_snapshot::{
+    persist_response_snapshot, ResponseSnapshotPersistenceDisposition,
+    ResponseSnapshotPersistenceError,
+};
+use crate::postgres_scoring_job::ScoringJobPersistenceDisposition;
+use crate::postgres_scoring_request::{
+    persist_scoring_dispatch, ScoringDispatchPersistence, ScoringDispatchPersistenceError,
+    ScoringRequestPersistenceDisposition,
+};
+use crate::response::ResponseSnapshot;
+use crate::scoring::ScoringRequest;
+use crate::scoring_job::ScoringJob;
+use postgres::Transaction;
+use std::error::Error;
+use std::fmt::{Display, Formatter};
+
+/// Durable dispositions produced by one response-snapshot/scoring-dispatch transaction.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct SnapshotScoringPersistence {
+    response_snapshot: ResponseSnapshotPersistenceDisposition,
+    dispatch: ScoringDispatchPersistence,
+}
+
+impl SnapshotScoringPersistence {
+    /// Return whether immutable response-snapshot evidence was inserted or exactly replayed.
+    #[must_use]
+    pub const fn response_snapshot(self) -> ResponseSnapshotPersistenceDisposition {
+        self.response_snapshot
+    }
+
+    /// Return whether immutable scoring-request evidence was inserted or exactly replayed.
+    #[must_use]
+    pub const fn scoring_request(self) -> ScoringRequestPersistenceDisposition {
+        self.dispatch.scoring_request()
+    }
+
+    /// Return whether durable scoring-job evidence was inserted or exactly replayed.
+    #[must_use]
+    pub const fn scoring_job(self) -> ScoringJobPersistenceDisposition {
+        self.dispatch.scoring_job()
+    }
+
+    /// Return whether immutable transactional-outbox evidence was inserted or exactly replayed.
+    #[must_use]
+    pub const fn outbox(self) -> PersistenceDisposition {
+        self.dispatch.outbox()
+    }
+}
+
+/// Fail-closed error for response-snapshot/scoring-dispatch persistence.
+#[derive(Debug)]
+#[non_exhaustive]
+pub enum SnapshotScoringPersistenceError {
+    /// The scoring request does not name the exact supplied snapshot and session.
+    MismatchedSnapshotBinding,
+    /// Immutable response-snapshot persistence failed.
+    Snapshot(ResponseSnapshotPersistenceError),
+    /// Scoring request/job/outbox persistence failed.
+    Dispatch(ScoringDispatchPersistenceError),
+}
+
+impl Display for SnapshotScoringPersistenceError {
+    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
+        formatter.write_str(match self {
+            Self::MismatchedSnapshotBinding => {
+                "scoring request must bind the exact response snapshot and session"
+            }
+            Self::Snapshot(_) => "response snapshot persistence failed before scoring dispatch",
+            Self::Dispatch(_) => "scoring dispatch persistence failed after response snapshot",
+        })
+    }
+}
+
+impl Error for SnapshotScoringPersistenceError {
+    fn source(&self) -> Option<&(dyn Error + 'static)> {
+        match self {
+            Self::MismatchedSnapshotBinding => None,
+            Self::Snapshot(error) => Some(error),
+            Self::Dispatch(error) => Some(error),
+        }
+    }
+}
+
+/// Persist one frozen response snapshot and its scoring request/job/outbox as one transaction.
+///
+/// The request must name the supplied snapshot identity and the same session before any write is
+/// attempted. The caller owns the `READ COMMITTED` transaction and final commit/rollback decision.
+/// If snapshot persistence or any dispatch stage fails, the caller must roll back so a snapshot is
+/// never acknowledged without its local scoring work, and newly scheduled scoring work is never
+/// detached from its immutable response evidence. Exact replay remains idempotent at every
+/// composed adapter.
+///
+/// This function does not define the external scoring event schema and does not execute scoring;
+/// it only removes the local durability gap before asynchronous dispatch.
+///
+/// # Errors
+///
+/// Returns [`SnapshotScoringPersistenceError::MismatchedSnapshotBinding`] before writes for a
+/// snapshot/session mismatch, or preserves typed snapshot/dispatch persistence failures.
+pub fn persist_response_snapshot_and_scoring_dispatch(
+    transaction: &mut Transaction<'_>,
+    snapshot: &ResponseSnapshot,
+    request: &ScoringRequest,
+    job: &ScoringJob,
+    dispatch_event: &IntegrationEvent,
+    outbox_max_attempts: usize,
+) -> Result<SnapshotScoringPersistence, SnapshotScoringPersistenceError> {
+    if snapshot.snapshot_ref() != Some(request.response_snapshot_ref())
+        || snapshot.session_ref() != request.session_ref()
+    {
+        return Err(SnapshotScoringPersistenceError::MismatchedSnapshotBinding);
+    }
+
+    let response_snapshot = persist_response_snapshot(transaction, snapshot)
+        .map_err(SnapshotScoringPersistenceError::Snapshot)?;
+    let dispatch = persist_scoring_dispatch(
+        transaction,
+        request,
+        job,
+        dispatch_event,
+        outbox_max_attempts,
+    )
+    .map_err(SnapshotScoringPersistenceError::Dispatch)?;
+
+    Ok(SnapshotScoringPersistence {
+        response_snapshot,
+        dispatch,
+    })
+}

tests/postgres_snapshot_scoring_error_paths.rs

@@ -0,0 +1,196 @@
+//! Failure-boundary coverage for atomic snapshot and scoring dispatch persistence.
+
+use postgres::{Client, IsolationLevel, NoTls};
+use psychometrics_commons_runtime::integration::IntegrationEvent;
+use psychometrics_commons_runtime::postgres_integration::apply_integration_migration;
+use psychometrics_commons_runtime::postgres_response_snapshot::{
+    apply_response_snapshot_migration, ResponseSnapshotPersistenceError,
+};
+use psychometrics_commons_runtime::postgres_scoring_job::apply_scoring_job_migration;
+use psychometrics_commons_runtime::postgres_scoring_orchestration::{
+    persist_response_snapshot_and_scoring_dispatch, SnapshotScoringPersistenceError,
+};
+use psychometrics_commons_runtime::postgres_scoring_request::{
+    apply_scoring_request_migration, ScoringDispatchPersistenceError,
+};
+use psychometrics_commons_runtime::response::{ResponseLedger, ResponseWrite};
+use psychometrics_commons_runtime::scoring::{ScoringRequest, ScoringRequestInput};
+use psychometrics_commons_runtime::scoring_job::ScoringJob;
+use psychometrics_commons_runtime::session::SessionState;
+use std::error::Error;
+use std::sync::{Mutex, MutexGuard};
+
+const DIGEST: &str = "sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc";
+static ERROR_PATH_LOCK: Mutex<()> = Mutex::new(());
+
+fn test_guard() -> MutexGuard<'static, ()> {
+    ERROR_PATH_LOCK
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+}
+
+fn test_client() -> Client {
+    let connection = std::env::var("TEST_DATABASE_URL")
+        .expect("TEST_DATABASE_URL must identify the isolated CI PostgreSQL database");
+    let mut client = Client::connect(&connection, NoTls)
+        .expect("isolated CI PostgreSQL database must be reachable");
+    client
+        .batch_execute(
+            "CREATE SCHEMA IF NOT EXISTS snapshot_scoring_error_test;\
+             SET search_path TO snapshot_scoring_error_test;\
+             DROP TABLE IF EXISTS integration_delivery_attempt;\
+             DROP TABLE IF EXISTS integration_inbox;\
+             DROP TABLE IF EXISTS integration_outbox;\
+             DROP TABLE IF EXISTS scoring_job_state;\
+             DROP TABLE IF EXISTS scoring_request;\
+             DROP TABLE IF EXISTS response_snapshot_entry;\
+             DROP TABLE IF EXISTS response_snapshot;",
+        )
+        .unwrap();
+    apply_integration_migration(&mut client).unwrap();
+    apply_scoring_job_migration(&mut client).unwrap();
+    apply_response_snapshot_migration(&mut client).unwrap();
+    apply_scoring_request_migration(&mut client).unwrap();
+    client
+}
+
+fn evidence() -> (
+    psychometrics_commons_runtime::response::ResponseSnapshot,
+    ScoringRequest,
+    IntegrationEvent,
+) {
+    let mut ledger = ResponseLedger::new("session_snapshot_error_alpha").unwrap();
+    ledger
+        .record(
+            SessionState::Active,
+            ResponseWrite {
+                server_event_ref: "server_event_snapshot_error_alpha",
+                client_event_ref: "client_event_snapshot_error_alpha",
+                item_version_ref: "item_version_snapshot_error_alpha",
+                payload_digest: DIGEST,
+            },
+        )
+        .unwrap();
+    let snapshot = ledger
+        .freeze_as(
+            SessionState::Completed,
+            "response_snapshot_error_alpha",
+        )
+        .unwrap();
+    let request = ScoringRequest::from_snapshot(
+        &snapshot,
+        ScoringRequestInput {
+            scoring_request_ref: "scoring_request_snapshot_error_alpha",
+            response_snapshot_ref: "response_snapshot_error_alpha",
+            assessment_spec_ref: "assessment_spec_big_five_v1",
+            instrument_version_ref: "instrument_version_big_five_ko_v1",
+            scoring_version_ref: "scoring_version_big_five_v1",
+            calibration_reference: "calibration_big_five_ko_v1",
+            norm_version_ref: None,
+            requested_output_schema_version: 1,
+        },
+    )
+    .unwrap();
+    let event = IntegrationEvent::new(
+        "event_snapshot_error_alpha",
+        "scoring.dispatch.requested",
+        "v1",
+        "psychometrics_commons",
+        "tenant_snapshot_error_alpha",
+        "scoring_job_snapshot_error_alpha",
+        10_000,
+        "correlation_snapshot_error_alpha",
+        Some("response_snapshot_error_alpha"),
+        DIGEST,
+    )
+    .unwrap();
+    (snapshot, request, event)
+}
+
+#[test]
+fn snapshot_isolation_failure_is_preserved_before_any_write() {
+    let _guard = test_guard();
+    let mut client = test_client();
+    let (snapshot, request, event) = evidence();
+    let job = ScoringJob::new(
+        "scoring_job_snapshot_error_alpha",
+        request.scoring_request_ref(),
+        3,
+    )
+    .unwrap();
+    let mut transaction = client
+        .build_transaction()
+        .isolation_level(IsolationLevel::Serializable)
+        .start()
+        .unwrap();
+    let error = persist_response_snapshot_and_scoring_dispatch(
+        &mut transaction,
+        &snapshot,
+        &request,
+        &job,
+        &event,
+        3,
+    )
+    .unwrap_err();
+    assert!(matches!(
+        error,
+        SnapshotScoringPersistenceError::Snapshot(
+            ResponseSnapshotPersistenceError::UnsupportedIsolationLevel
+        )
+    ));
+    assert!(error.source().is_some());
+    transaction.rollback().unwrap();
+}
+
+#[test]
+fn dispatch_binding_failure_rolls_back_the_new_snapshot() {
+    let _guard = test_guard();
+    let mut client = test_client();
+    let (snapshot, request, event) = evidence();
+    let mismatched_job = ScoringJob::new(
+        "scoring_job_snapshot_error_alpha",
+        "scoring_request_other",
+        3,
+    )
+    .unwrap();
+    let mut transaction = client.transaction().unwrap();
+    let error = persist_response_snapshot_and_scoring_dispatch(
+        &mut transaction,
+        &snapshot,
+        &request,
+        &mismatched_job,
+        &event,
+        3,
+    )
+    .unwrap_err();
+    assert!(matches!(
+        error,
+        SnapshotScoringPersistenceError::Dispatch(
+            ScoringDispatchPersistenceError::MismatchedScoringRequest
+        )
+    ));
+    assert!(error.source().is_some());
+    transaction.rollback().unwrap();
+
+    let snapshot_count: i64 = client
+        .query_one("SELECT count(*) FROM response_snapshot", &[])
+        .unwrap()
+        .get(0);
+    assert_eq!(snapshot_count, 0);
+}
+
+#[test]
+fn every_error_family_has_stable_operator_facing_display() {
+    let errors = [
+        SnapshotScoringPersistenceError::MismatchedSnapshotBinding,
+        SnapshotScoringPersistenceError::Snapshot(
+            ResponseSnapshotPersistenceError::InvalidReference,
+        ),
+        SnapshotScoringPersistenceError::Dispatch(
+            ScoringDispatchPersistenceError::MismatchedScoringRequest,
+        ),
+    ];
+    for error in errors {
+        assert!(!error.to_string().is_empty());
+    }
+}