Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ All notable product and architecture changes are recorded here. Releases use imm
## Unreleased

### Added
- Atomic PostgreSQL scoring-job completion and integration-outbox composition: a fenced successful result and its bound outbox event commit together, exact replay stays idempotent, mixed exact dispositions reconcile legacy one-sided state, and a late outbox conflict rolls the completion back.
- Scoring-job cancel and lease-expiry fallback classification lock the current row until the caller transaction ends, so concurrent workers cannot rewrite terminal or unleased evidence.
- PostgreSQL operational-store readiness probe classifies the supported major version and write-readiness, and fails closed when a caller-declared required relation is missing.
- PostgreSQL scoring-job cancellation: queued, leased, or retry-scheduled work becomes cancelled without transferring a fence, exact replay is idempotent, and completed or quarantined evidence cannot be rewritten.
Expand Down
2 changes: 1 addition & 1 deletion docs/TRACEABILITY.md
Original file line number Diff line number Diff line change
Expand Up @@ -132,7 +132,7 @@ Still-Target logical modules/adapters include remaining product aggregate persis

### Active implementation work that is not protected-main truth

**Active PR** #76 data-rights processing-start persistence is not protected-main truth until an unchanged reviewed/check-clean head is integrated. Identity-verified requests persist an immutable operation identity and processing-start time under `FOR UPDATE` so later lifecycle composition cannot race the classified row. Dependent-system execution remains outside this slice.
**Active PR** #69 scoring-completion outbox persistence is not protected-main truth until an unchanged reviewed/check-clean head is integrated. Successful scoring completion commits the job outcome with a bound outbox event atomically, and mixed exact dispositions reconcile legacy one-sided completion or outbox state without rewriting immutable history. Live fast-mlsirm execution remains outside this slice.

## 5. ADR traceability by concern

Expand Down
1 change: 1 addition & 0 deletions src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ pub mod postgres_integration;
pub mod postgres_item_delivery;
pub mod postgres_response_snapshot;
pub mod postgres_result_snapshot;
pub mod postgres_scoring_completion;
pub mod postgres_scoring_job;
pub mod postgres_scoring_request;
mod reference;
Expand Down
135 changes: 135 additions & 0 deletions src/postgres_scoring_completion.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,135 @@
//! Atomic `PostgreSQL` composition for scoring completion and integration outbox evidence.
//!
//! This module does not calculate psychometric quantities or define an external event schema.
//! It composes the existing fenced scoring-job completion transition with the existing durable
//! transactional outbox inside one caller-owned transaction so a successful result transition
//! cannot commit while its integration evidence is lost.

use crate::integration::IntegrationEvent;
use crate::postgres_integration::{enqueue_outbox_event, PersistenceDisposition, PersistenceError};
use crate::postgres_scoring_job::{
record_successful_scoring_completion, ScoringJobCompletionDisposition,
ScoringJobPersistenceError,
};
use postgres::Transaction;
use std::error::Error;
use std::fmt::{Display, Formatter};

const SOURCE_REF: &str = "psychometrics_commons";

/// Durable dispositions produced by one atomic scoring-completion/outbox operation.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct ScoringCompletionOutboxPersistence {
completion: ScoringJobCompletionDisposition,
outbox: PersistenceDisposition,
}

impl ScoringCompletionOutboxPersistence {
/// Return whether scoring completion was newly committed or exactly replayed.
#[must_use]
pub const fn completion(self) -> ScoringJobCompletionDisposition {
self.completion
}

/// Return whether immutable outbox evidence was newly inserted or exactly replayed.
#[must_use]
pub const fn outbox(self) -> PersistenceDisposition {
self.outbox
}
}

/// Fail-closed error for atomic scoring completion and outbox persistence.
#[derive(Debug)]
#[non_exhaustive]
pub enum ScoringCompletionOutboxError {
/// The outbox envelope is not bound to this exact scoring-job completion boundary.
InvalidCompletionEnvelope,
/// The fenced scoring-job completion transition failed validation or persistence.
Completion(ScoringJobPersistenceError),
/// The immutable integration outbox event failed validation or persistence.
Outbox(PersistenceError),
}

impl Display for ScoringCompletionOutboxError {
fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
formatter.write_str(match self {
Self::InvalidCompletionEnvelope => {
"scoring completion outbox must bind the exact job and completion time"
}
Self::Completion(_) => "scoring completion persistence failed",
Self::Outbox(_) => "scoring completion outbox persistence failed",
})
}
}

impl Error for ScoringCompletionOutboxError {
fn source(&self) -> Option<&(dyn Error + 'static)> {
match self {
Self::InvalidCompletionEnvelope => None,
Self::Completion(error) => Some(error),
Self::Outbox(error) => Some(error),
}
}
}

/// Persist one fenced successful scoring completion and one immutable outbox event atomically.
///
/// Before any write, the outbox event must be emitted by `psychometrics_commons`, identify the
/// exact scoring job as its subject, and carry the same server-authoritative completion time.
/// Event type, tenant, schema version, correlation, causation, payload digest, event identity,
/// and `scoring_result_ref` stay on the caller's versioned integration contract; this composition
/// does not add a schema that binds those fields. A new `event_ref` after an already-accepted
/// completion therefore inserts a second outbox row rather than rewriting historical completion
/// evidence.
///
/// The caller supplies and owns a `READ COMMITTED` transaction. After envelope validation, the
/// scoring-job transition runs first and the outbox insert runs in the same transaction. If either
/// stage fails, callers must roll the transaction back so no newly written completion state can
/// survive without its outbox evidence. Exact replay remains idempotent at both existing adapters,
/// and mixed exact dispositions can safely reconcile legacy partial state without mutating
/// historical evidence.
///
/// # Errors
///
/// Returns [`ScoringCompletionOutboxError::InvalidCompletionEnvelope`] before writes when the
/// integration event is bound to another source, job, or completion time,
/// [`ScoringCompletionOutboxError::Completion`] when fenced scoring completion fails, or
/// [`ScoringCompletionOutboxError::Outbox`] when the integration event cannot be persisted.
#[allow(clippy::too_many_arguments)]
pub fn record_successful_scoring_completion_with_outbox(
transaction: &mut Transaction<'_>,
scoring_job_ref: &str,
fencing_token: u64,
scoring_result_ref: &str,
completed_at_unix_ms: u64,
completion_event: &IntegrationEvent,
outbox_max_attempts: usize,
) -> Result<ScoringCompletionOutboxPersistence, ScoringCompletionOutboxError> {
validate_completion_envelope(scoring_job_ref, completed_at_unix_ms, completion_event)?;
let completion = record_successful_scoring_completion(
transaction,
scoring_job_ref,
fencing_token,
scoring_result_ref,
completed_at_unix_ms,
)
.map_err(ScoringCompletionOutboxError::Completion)?;
let outbox = enqueue_outbox_event(transaction, completion_event, outbox_max_attempts)
.map_err(ScoringCompletionOutboxError::Outbox)?;

Ok(ScoringCompletionOutboxPersistence { completion, outbox })
}

fn validate_completion_envelope(
scoring_job_ref: &str,
completed_at_unix_ms: u64,
completion_event: &IntegrationEvent,
) -> Result<(), ScoringCompletionOutboxError> {
if completion_event.source() != SOURCE_REF
|| completion_event.subject_ref() != scoring_job_ref
|| completion_event.occurred_at_unix_ms() != completed_at_unix_ms

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The bind set is source + job + occurred_at / completed_at. Leave ResultSnapshot and scoring_result_ref to the worker/schema PR. Keep event_ref stable across retries: a new identity after Duplicate completion inserts a second outbox row rather than reconciling.

{
return Err(ScoringCompletionOutboxError::InvalidCompletionEnvelope);
}
Ok(())
}
11 changes: 8 additions & 3 deletions tests/postgres_recovery_invariants.rs
Original file line number Diff line number Diff line change
Expand Up @@ -84,11 +84,11 @@ fn seed_recovery_critical_state(client: &mut Client) {
INSERT INTO {SOURCE_SCHEMA}.integration_consumption (
consumer_ref, source_ref, tenant_ref, source_event_ref, consumption_ref,
side_effect_ref, consumption_state, fencing_token, latest_event_at_unix_ms,
claim_expires_at_unix_ms, completion_evidence_ref, cause_code
claim_expires_at_unix_ms, claim_deadline_at, completion_evidence_ref, cause_code
) VALUES (
'consumer_recovery_alpha', 'dependency_recovery_alpha', 'tenant_recovery_alpha',
'event_dependency_alpha', 'consumption_recovery_alpha', 'effect_recovery_alpha',
'processing', 7, 12000, 13000, NULL, NULL
'processing', 7, 12000, 13000, TIMESTAMPTZ '1970-01-01 00:00:13+00', NULL, NULL
);
INSERT INTO {SOURCE_SCHEMA}.response_snapshot (
snapshot_ref, session_ref, event_count, last_sequence
Expand Down Expand Up @@ -150,7 +150,8 @@ fn assert_restored_evidence(client: &mut Client) {
let restored_consumption = client
.query_one(
&format!(
"SELECT consumption_state, fencing_token, claim_expires_at_unix_ms
"SELECT consumption_state, fencing_token, claim_expires_at_unix_ms,
claim_deadline_at IS NOT NULL
FROM {RESTORED_SCHEMA}.integration_consumption
WHERE consumption_ref = 'consumption_recovery_alpha'"
),
Expand All @@ -160,6 +161,10 @@ fn assert_restored_evidence(client: &mut Client) {
assert_eq!(restored_consumption.get::<_, String>(0), "processing");
assert_eq!(restored_consumption.get::<_, i64>(1), 7);
assert_eq!(restored_consumption.get::<_, Option<i64>>(2), Some(13000));
assert!(
restored_consumption.get::<_, bool>(3),
"restored processing claim must keep its database-authoritative deadline"
);

let restored_snapshot = client
.query_one(
Expand Down
Loading
Loading