diff --git a/src/postgres_scoring_request.rs b/src/postgres_scoring_request.rs index e247d7a5..7561ed45 100644 --- a/src/postgres_scoring_request.rs +++ b/src/postgres_scoring_request.rs @@ -6,13 +6,20 @@ //! `fast-mlsirm` and does not store numeric scores. Replay requires //! `READ COMMITTED`. +use crate::integration::IntegrationEvent; +use crate::postgres_integration::{enqueue_outbox_event, PersistenceDisposition, PersistenceError}; +use crate::postgres_scoring_job::{ + persist_scoring_job, ScoringJobPersistenceDisposition, ScoringJobPersistenceError, +}; use crate::reference::normalized_reference; use crate::scoring::ScoringRequest; +use crate::scoring_job::ScoringJob; use postgres::Transaction; use std::error::Error; use std::fmt::{Display, Formatter}; const SCORING_REQUEST_MIGRATION: &str = include_str!("../migrations/0011_scoring_request.sql"); +const PRODUCT_EVENT_SOURCE: &str = "psychometrics_commons"; /// Outcome of persisting one immutable scoring request. #[derive(Clone, Copy, Debug, Eq, PartialEq)] @@ -24,6 +31,81 @@ pub enum ScoringRequestPersistenceDisposition { Duplicate, } +/// Durable dispositions produced by one atomic scoring-dispatch persistence call. +/// +/// Mixed dispositions are valid when this transaction safely reconciles pre-existing +/// exact evidence from an older write path. Every newly inserted row is still committed +/// or rolled back together by the caller-owned transaction. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct ScoringDispatchPersistence { + scoring_request: ScoringRequestPersistenceDisposition, + scoring_job: ScoringJobPersistenceDisposition, + outbox: PersistenceDisposition, +} + +impl ScoringDispatchPersistence { + /// Return whether immutable scoring-request evidence was inserted or replayed. + #[must_use] + pub const fn scoring_request(self) -> ScoringRequestPersistenceDisposition { + self.scoring_request + } + + /// Return whether immutable scoring-job evidence was inserted or replayed. + #[must_use] + pub const fn scoring_job(self) -> ScoringJobPersistenceDisposition { + self.scoring_job + } + + /// Return whether immutable outbox evidence was inserted or replayed. + #[must_use] + pub const fn outbox(self) -> PersistenceDisposition { + self.outbox + } +} + +/// Fail-closed error for one atomic scoring-dispatch persistence operation. +#[derive(Debug)] +#[non_exhaustive] +pub enum ScoringDispatchPersistenceError { + /// The scoring job names a different immutable scoring request. + MismatchedScoringRequest, + /// The dispatch outbox envelope is not causally bound to this local request/job pair. + InvalidDispatchEnvelope, + /// Immutable scoring-request persistence failed. + Request(ScoringRequestPersistenceError), + /// Durable scoring-job persistence failed. + Job(ScoringJobPersistenceError), + /// Transactional outbox persistence failed. + Outbox(PersistenceError), +} + +impl Display for ScoringDispatchPersistenceError { + fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result { + formatter.write_str(match self { + Self::MismatchedScoringRequest => { + "scoring job must reference the immutable scoring request in the same dispatch" + } + Self::InvalidDispatchEnvelope => { + "scoring dispatch outbox envelope must identify the local job and response snapshot" + } + Self::Request(_) => "scoring dispatch request persistence failed", + Self::Job(_) => "scoring dispatch job persistence failed", + Self::Outbox(_) => "scoring dispatch outbox persistence failed", + }) + } +} + +impl Error for ScoringDispatchPersistenceError { + fn source(&self) -> Option<&(dyn Error + 'static)> { + match self { + Self::MismatchedScoringRequest | Self::InvalidDispatchEnvelope => None, + Self::Request(error) => Some(error), + Self::Job(error) => Some(error), + Self::Outbox(error) => Some(error), + } + } +} + /// Fail-closed error for durable scoring-request persistence. #[derive(Debug)] #[non_exhaustive] @@ -86,6 +168,58 @@ pub fn apply_scoring_request_migration( client.batch_execute(SCORING_REQUEST_MIGRATION) } +/// Persist a scoring request, its fresh asynchronous job, and one outbox event atomically. +/// +/// The caller owns the transaction and therefore the final commit/rollback decision. This +/// function composes the existing immutable request, job, and transactional-outbox adapters +/// without introducing a second transaction boundary. The job must name exactly the supplied +/// scoring request. The outbox envelope must be emitted by this bounded context, use the scoring +/// job as its subject, and identify the immutable response snapshot as its causation reference. +/// Event type/schema, tenant, correlation, and payload semantics remain the caller's versioned +/// integration-contract responsibility after that causal binding is established. +/// +/// Exact replay is idempotent. Pre-existing exact evidence may produce mixed dispositions, +/// allowing a caller to reconcile a legacy partial state without rewriting immutable rows. +/// If any stage returns an error, callers must roll back the transaction rather than commit +/// earlier successful stages. +/// +/// # Errors +/// +/// Returns [`ScoringDispatchPersistenceError::MismatchedScoringRequest`] before writing when +/// the job is bound to another request, [`ScoringDispatchPersistenceError::InvalidDispatchEnvelope`] +/// when the outbox envelope is not causally bound to this local dispatch, or the typed request, +/// job, or outbox persistence error when one of those stages fails. +pub fn persist_scoring_dispatch( + transaction: &mut Transaction<'_>, + request: &ScoringRequest, + job: &ScoringJob, + dispatch_event: &IntegrationEvent, + outbox_max_attempts: usize, +) -> Result { + if job.scoring_request_ref() != request.scoring_request_ref() { + return Err(ScoringDispatchPersistenceError::MismatchedScoringRequest); + } + if dispatch_event.source() != PRODUCT_EVENT_SOURCE + || dispatch_event.subject_ref() != job.scoring_job_ref() + || dispatch_event.causation_ref() != Some(request.response_snapshot_ref()) + { + return Err(ScoringDispatchPersistenceError::InvalidDispatchEnvelope); + } + + let scoring_request = persist_scoring_request(transaction, request) + .map_err(ScoringDispatchPersistenceError::Request)?; + let scoring_job = + persist_scoring_job(transaction, job).map_err(ScoringDispatchPersistenceError::Job)?; + let outbox = enqueue_outbox_event(transaction, dispatch_event, outbox_max_attempts) + .map_err(ScoringDispatchPersistenceError::Outbox)?; + + Ok(ScoringDispatchPersistence { + scoring_request, + scoring_job, + outbox, + }) +} + /// Persist one immutable scoring-request identity. /// /// Exact replay of the same request identity and version bundle is idempotent. diff --git a/tests/postgres_scoring_dispatch_envelope.rs b/tests/postgres_scoring_dispatch_envelope.rs new file mode 100644 index 00000000..916ba10f --- /dev/null +++ b/tests/postgres_scoring_dispatch_envelope.rs @@ -0,0 +1,153 @@ +//! Scoring dispatch outbox evidence must be causally bound to the request and job being persisted. + +use postgres::{Client, NoTls}; +use psychometrics_commons_runtime::integration::IntegrationEvent; +use psychometrics_commons_runtime::postgres_integration::apply_integration_migration; +use psychometrics_commons_runtime::postgres_scoring_job::apply_scoring_job_migration; +use psychometrics_commons_runtime::postgres_scoring_request::{ + apply_scoring_request_migration, persist_scoring_dispatch, 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; + +const SCHEMA: &str = "scoring_dispatch_envelope_test"; +const DATABASE_TEST_LOCK_KEY: i64 = 0x5343_4453_5045_4E56; +const DIGEST: &str = "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; + +fn ready_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 + .query_one("SELECT pg_advisory_lock($1)", &[&DATABASE_TEST_LOCK_KEY]) + .expect("shared scoring-dispatch envelope test lock should be acquired"); + client + .batch_execute(&format!( + "DROP SCHEMA IF EXISTS {SCHEMA} CASCADE; + CREATE SCHEMA {SCHEMA}; + SET search_path TO {SCHEMA};" + )) + .unwrap(); + apply_integration_migration(&mut client).unwrap(); + apply_scoring_job_migration(&mut client).unwrap(); + apply_scoring_request_migration(&mut client).unwrap(); + client +} + +fn request() -> ScoringRequest { + let mut ledger = ResponseLedger::new("session_dispatch_envelope_alpha").unwrap(); + ledger + .record( + SessionState::Active, + ResponseWrite { + server_event_ref: "server_event_dispatch_envelope_alpha", + client_event_ref: "client_event_dispatch_envelope_alpha", + item_version_ref: "item_version_dispatch_envelope_alpha", + payload_digest: DIGEST, + }, + ) + .unwrap(); + let snapshot = ledger + .freeze_as( + SessionState::Completed, + "response_snapshot_dispatch_envelope_alpha", + ) + .unwrap(); + ScoringRequest::from_snapshot( + &snapshot, + ScoringRequestInput { + scoring_request_ref: "scoring_request_dispatch_envelope_alpha", + response_snapshot_ref: "response_snapshot_dispatch_envelope_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: Some("norm_version_big_five_ko_v1"), + requested_output_schema_version: 1, + }, + ) + .unwrap() +} + +fn event(source: &str, subject: &str, causation: Option<&str>) -> IntegrationEvent { + IntegrationEvent::new( + "event_scoring_dispatch_envelope_alpha", + "scoring.dispatch.requested", + "v1", + source, + "tenant_dispatch_envelope_alpha", + subject, + 10_000, + "correlation_dispatch_envelope_alpha", + causation, + DIGEST, + ) + .unwrap() +} + +#[test] +fn unrelated_source_subject_or_snapshot_causation_is_rejected_before_writes() { + let cases = [ + ( + "other_source", + "scoring_job_dispatch_envelope_alpha", + Some("response_snapshot_dispatch_envelope_alpha"), + ), + ( + "psychometrics_commons", + "scoring_job_dispatch_envelope_other", + Some("response_snapshot_dispatch_envelope_alpha"), + ), + ( + "psychometrics_commons", + "scoring_job_dispatch_envelope_alpha", + Some("response_snapshot_dispatch_envelope_other"), + ), + ( + "psychometrics_commons", + "scoring_job_dispatch_envelope_alpha", + None, + ), + ]; + + for (index, (source, subject, causation)) in cases.into_iter().enumerate() { + let mut client = ready_client(); + let request = request(); + let job = ScoringJob::new( + "scoring_job_dispatch_envelope_alpha", + request.scoring_request_ref(), + 3, + ) + .unwrap(); + let dispatch_event = event(source, subject, causation); + + let mut transaction = client.transaction().unwrap(); + assert!(matches!( + persist_scoring_dispatch(&mut transaction, &request, &job, &dispatch_event, 3), + Err(ScoringDispatchPersistenceError::InvalidDispatchEnvelope) + )); + transaction.rollback().unwrap(); + + for table in ["scoring_request", "scoring_job_state", "integration_outbox"] { + let count: i64 = client + .query_one(&format!("SELECT count(*) FROM {table}"), &[]) + .unwrap() + .get(0); + assert_eq!( + count, 0, + "invalid envelope case {index} must not write {table}" + ); + } + } + + let mut client = ready_client(); + client + .batch_execute(&format!( + "SET search_path TO public; + DROP SCHEMA IF EXISTS {SCHEMA} CASCADE;" + )) + .unwrap(); +} diff --git a/tests/postgres_scoring_dispatch_error_paths.rs b/tests/postgres_scoring_dispatch_error_paths.rs new file mode 100644 index 00000000..478913cf --- /dev/null +++ b/tests/postgres_scoring_dispatch_error_paths.rs @@ -0,0 +1,215 @@ +//! Failure-path coverage for atomic scoring-dispatch persistence. + +use postgres::{Client, IsolationLevel, NoTls}; +use psychometrics_commons_runtime::integration::IntegrationEvent; +use psychometrics_commons_runtime::postgres_integration::{ + apply_integration_migration, PersistenceError, +}; +use psychometrics_commons_runtime::postgres_scoring_job::{ + apply_scoring_job_migration, ScoringJobPersistenceError, +}; +use psychometrics_commons_runtime::postgres_scoring_request::{ + apply_scoring_request_migration, persist_scoring_dispatch, ScoringDispatchPersistenceError, + ScoringRequestPersistenceError, +}; +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 PAYLOAD_DIGEST: &str = + "sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc"; + +static ERROR_PATH_TEST_LOCK: Mutex<()> = Mutex::new(()); + +fn error_path_guard() -> MutexGuard<'static, ()> { + ERROR_PATH_TEST_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 scoring_dispatch_error_path_test;\ + SET search_path TO scoring_dispatch_error_path_test;", + ) + .unwrap(); + client +} + +fn reset_and_migrate(client: &mut Client) { + client + .batch_execute( + "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;", + ) + .unwrap(); + apply_integration_migration(client).unwrap(); + apply_scoring_job_migration(client).unwrap(); + apply_scoring_request_migration(client).unwrap(); +} + +fn request_named(scoring_request_ref: &str) -> ScoringRequest { + let mut ledger = ResponseLedger::new("session_dispatch_error_path").unwrap(); + ledger + .record( + SessionState::Active, + ResponseWrite { + server_event_ref: "server_event_dispatch_error_path", + client_event_ref: "client_event_dispatch_error_path", + item_version_ref: "item_version_dispatch_error_path", + payload_digest: PAYLOAD_DIGEST, + }, + ) + .unwrap(); + let snapshot = ledger + .freeze_as( + SessionState::Completed, + "response_snapshot_dispatch_error_path", + ) + .unwrap(); + ScoringRequest::from_snapshot( + &snapshot, + ScoringRequestInput { + scoring_request_ref, + response_snapshot_ref: "response_snapshot_dispatch_error_path", + 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() +} + +fn dispatch_event() -> IntegrationEvent { + IntegrationEvent::new( + "event_dispatch_error_path", + "scoring.dispatch.requested", + "v1", + "psychometrics_commons", + "tenant_dispatch_error_path", + "scoring_job_dispatch_error_path", + 20_000, + "correlation_dispatch_error_path", + Some("response_snapshot_dispatch_error_path"), + PAYLOAD_DIGEST, + ) + .unwrap() +} + +#[test] +fn request_isolation_failure_is_preserved_without_committed_partial_state() { + let _guard = error_path_guard(); + let mut client = test_client(); + reset_and_migrate(&mut client); + let request = request_named("scoring_request_dispatch_serializable"); + let job = ScoringJob::new( + "scoring_job_dispatch_error_path", + request.scoring_request_ref(), + 3, + ) + .unwrap(); + let event = dispatch_event(); + + let mut transaction = client + .build_transaction() + .isolation_level(IsolationLevel::Serializable) + .start() + .unwrap(); + assert!(matches!( + persist_scoring_dispatch(&mut transaction, &request, &job, &event, 3), + Err(ScoringDispatchPersistenceError::Request( + ScoringRequestPersistenceError::UnsupportedIsolationLevel + )) + )); + transaction.rollback().unwrap(); + + let request_count: i64 = client + .query_one("SELECT count(*) FROM scoring_request", &[]) + .unwrap() + .get(0); + assert_eq!(request_count, 0); +} + +#[test] +fn nonfresh_job_failure_rolls_back_request_insert() { + let _guard = error_path_guard(); + let mut client = test_client(); + reset_and_migrate(&mut client); + let request = request_named("scoring_request_dispatch_nonfresh"); + let mut job = ScoringJob::new( + "scoring_job_dispatch_error_path", + request.scoring_request_ref(), + 3, + ) + .unwrap(); + job.claim( + "worker_dispatch_error_path", + "lease_dispatch_error_path", + 20_000, + 30_000, + ) + .unwrap(); + let event = dispatch_event(); + + let mut transaction = client.transaction().unwrap(); + assert!(matches!( + persist_scoring_dispatch(&mut transaction, &request, &job, &event, 3), + Err(ScoringDispatchPersistenceError::Job( + ScoringJobPersistenceError::UnsupportedInitialState + )) + )); + transaction.rollback().unwrap(); + + let request_count: i64 = client + .query_one("SELECT count(*) FROM scoring_request", &[]) + .unwrap() + .get(0); + assert_eq!(request_count, 0); +} + +#[test] +fn dispatch_error_display_and_sources_are_typed() { + let cases = [ + ( + ScoringDispatchPersistenceError::MismatchedScoringRequest, + false, + ), + ( + ScoringDispatchPersistenceError::InvalidDispatchEnvelope, + false, + ), + ( + ScoringDispatchPersistenceError::Request( + ScoringRequestPersistenceError::InvalidReference, + ), + true, + ), + ( + ScoringDispatchPersistenceError::Job(ScoringJobPersistenceError::InvalidReference), + true, + ), + ( + ScoringDispatchPersistenceError::Outbox(PersistenceError::InvalidReference), + true, + ), + ]; + + for (error, has_source) in cases { + assert!(!error.to_string().is_empty()); + assert_eq!(error.source().is_some(), has_source); + } +} diff --git a/tests/postgres_scoring_dispatch_transaction.rs b/tests/postgres_scoring_dispatch_transaction.rs new file mode 100644 index 00000000..c1ef12b1 --- /dev/null +++ b/tests/postgres_scoring_dispatch_transaction.rs @@ -0,0 +1,291 @@ +//! Real `PostgreSQL` contract for atomic scoring-dispatch persistence. + +use postgres::{Client, NoTls}; +use psychometrics_commons_runtime::integration::IntegrationEvent; +use psychometrics_commons_runtime::postgres_integration::{ + apply_integration_migration, enqueue_outbox_event, PersistenceDisposition, +}; +use psychometrics_commons_runtime::postgres_scoring_job::{ + apply_scoring_job_migration, ScoringJobPersistenceDisposition, +}; +use psychometrics_commons_runtime::postgres_scoring_request::{ + apply_scoring_request_migration, persist_scoring_dispatch, ScoringDispatchPersistenceError, + ScoringRequestPersistenceDisposition, +}; +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::sync::{Mutex, MutexGuard}; + +const PAYLOAD_DIGEST_A: &str = + "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; +const PAYLOAD_DIGEST_B: &str = + "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"; + +static DISPATCH_TEST_LOCK: Mutex<()> = Mutex::new(()); + +fn dispatch_test_guard() -> MutexGuard<'static, ()> { + DISPATCH_TEST_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 scoring_dispatch_transaction_test;\ + SET search_path TO scoring_dispatch_transaction_test;", + ) + .unwrap(); + client +} + +fn reset_tables(client: &mut Client) { + client + .batch_execute( + "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;", + ) + .unwrap(); +} + +fn apply_migrations(client: &mut Client) { + apply_integration_migration(client).unwrap(); + apply_scoring_job_migration(client).unwrap(); + apply_scoring_request_migration(client).unwrap(); +} + +fn request_named( + session_ref: &str, + scoring_request_ref: &str, + snapshot_ref: &str, +) -> ScoringRequest { + let mut ledger = ResponseLedger::new(session_ref).unwrap(); + ledger + .record( + SessionState::Active, + ResponseWrite { + server_event_ref: "server_event_dispatch_one", + client_event_ref: "client_event_dispatch_one", + item_version_ref: "item_version_dispatch_one", + payload_digest: PAYLOAD_DIGEST_A, + }, + ) + .unwrap(); + let snapshot = ledger + .freeze_as(SessionState::Completed, snapshot_ref) + .unwrap(); + ScoringRequest::from_snapshot( + &snapshot, + ScoringRequestInput { + scoring_request_ref, + response_snapshot_ref: snapshot_ref, + 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: Some("norm_version_big_five_ko_v1"), + requested_output_schema_version: 1, + }, + ) + .unwrap() +} + +fn dispatch_event( + event_ref: &str, + digest: &str, + scoring_job_ref: &str, + response_snapshot_ref: &str, +) -> IntegrationEvent { + IntegrationEvent::new( + event_ref, + "scoring.dispatch.requested", + "v1", + "psychometrics_commons", + "tenant_dispatch_alpha", + scoring_job_ref, + 10_000, + "correlation_dispatch_alpha", + Some(response_snapshot_ref), + digest, + ) + .unwrap() +} + +#[test] +fn request_job_and_outbox_are_committed_and_replayed_as_one_dispatch() { + let _guard = dispatch_test_guard(); + let mut client = test_client(); + reset_tables(&mut client); + apply_migrations(&mut client); + + let request = request_named( + "session_dispatch_alpha", + "scoring_request_dispatch_alpha", + "response_snapshot_dispatch_alpha", + ); + let job = ScoringJob::new( + "scoring_job_dispatch_alpha", + request.scoring_request_ref(), + 3, + ) + .unwrap(); + let event = dispatch_event( + "event_scoring_dispatch_alpha", + PAYLOAD_DIGEST_A, + job.scoring_job_ref(), + request.response_snapshot_ref(), + ); + + let mut transaction = client.transaction().unwrap(); + let inserted = persist_scoring_dispatch(&mut transaction, &request, &job, &event, 3).unwrap(); + assert_eq!( + inserted.scoring_request(), + ScoringRequestPersistenceDisposition::Inserted + ); + assert_eq!( + inserted.scoring_job(), + ScoringJobPersistenceDisposition::Inserted + ); + assert_eq!(inserted.outbox(), PersistenceDisposition::Inserted); + transaction.commit().unwrap(); + + let mut transaction = client.transaction().unwrap(); + let duplicate = persist_scoring_dispatch(&mut transaction, &request, &job, &event, 3).unwrap(); + assert_eq!( + duplicate.scoring_request(), + ScoringRequestPersistenceDisposition::Duplicate + ); + assert_eq!( + duplicate.scoring_job(), + ScoringJobPersistenceDisposition::Duplicate + ); + assert_eq!(duplicate.outbox(), PersistenceDisposition::Duplicate); + transaction.commit().unwrap(); + + for table in ["scoring_request", "scoring_job_state", "integration_outbox"] { + let count: i64 = client + .query_one(&format!("SELECT count(*) FROM {table}"), &[]) + .unwrap() + .get(0); + assert_eq!(count, 1, "{table} must contain exactly one durable row"); + } +} + +#[test] +fn mismatched_job_request_fails_before_any_write() { + let _guard = dispatch_test_guard(); + let mut client = test_client(); + reset_tables(&mut client); + apply_migrations(&mut client); + + let request = request_named( + "session_dispatch_mismatch", + "scoring_request_dispatch_mismatch", + "response_snapshot_dispatch_mismatch", + ); + let job = ScoringJob::new("scoring_job_dispatch_alpha", "other_scoring_request", 3).unwrap(); + let event = dispatch_event( + "event_scoring_dispatch_mismatch", + PAYLOAD_DIGEST_A, + job.scoring_job_ref(), + request.response_snapshot_ref(), + ); + + let mut transaction = client.transaction().unwrap(); + assert!(matches!( + persist_scoring_dispatch(&mut transaction, &request, &job, &event, 3), + Err(ScoringDispatchPersistenceError::MismatchedScoringRequest) + )); + transaction.rollback().unwrap(); + + for table in ["scoring_request", "scoring_job_state", "integration_outbox"] { + let count: i64 = client + .query_one(&format!("SELECT count(*) FROM {table}"), &[]) + .unwrap() + .get(0); + assert_eq!( + count, 0, + "{table} must remain empty after rejected dispatch" + ); + } +} + +#[test] +fn outbox_conflict_rolls_back_request_and_job_insertions() { + let _guard = dispatch_test_guard(); + let mut client = test_client(); + reset_tables(&mut client); + apply_migrations(&mut client); + + let request = request_named( + "session_dispatch_conflict", + "scoring_request_dispatch_conflict", + "response_snapshot_dispatch_conflict", + ); + let job = ScoringJob::new( + "scoring_job_dispatch_alpha", + request.scoring_request_ref(), + 3, + ) + .unwrap(); + let existing_event = dispatch_event( + "event_scoring_dispatch_conflict", + PAYLOAD_DIGEST_A, + job.scoring_job_ref(), + request.response_snapshot_ref(), + ); + assert_eq!( + enqueue_outbox_event(&mut client, &existing_event, 3).unwrap(), + PersistenceDisposition::Inserted + ); + let conflicting_event = dispatch_event( + "event_scoring_dispatch_conflict", + PAYLOAD_DIGEST_B, + job.scoring_job_ref(), + request.response_snapshot_ref(), + ); + + let mut transaction = client.transaction().unwrap(); + assert!(matches!( + persist_scoring_dispatch(&mut transaction, &request, &job, &conflicting_event, 3), + Err(ScoringDispatchPersistenceError::Outbox(_)) + )); + transaction.rollback().unwrap(); + + let request_count: i64 = client + .query_one( + "SELECT count(*) FROM scoring_request WHERE scoring_request_ref = $1", + &[&request.scoring_request_ref()], + ) + .unwrap() + .get(0); + let job_count: i64 = client + .query_one( + "SELECT count(*) FROM scoring_job_state WHERE scoring_job_ref = $1", + &[&job.scoring_job_ref()], + ) + .unwrap() + .get(0); + let outbox_count: i64 = client + .query_one( + "SELECT count(*) FROM integration_outbox WHERE event_ref = $1", + &[&existing_event.event_ref()], + ) + .unwrap() + .get(0); + assert_eq!(request_count, 0); + assert_eq!(job_count, 0); + assert_eq!( + outbox_count, 1, + "pre-existing outbox evidence must remain intact" + ); +}