commit ba67c988ee14f2cfd45057b54f4d9c88175090c0
parent 678cc851ab64930404fab50d1ff72e58c20580d8
Author: triesap <tyson@radroots.org>
Date: Mon, 24 Aug 2026 09:32:53 +0000
refactor(rhi): fence reconciliation finalization
- bind sealed manifests to exact attempt and job identities
- validate live leases, dirty generations, policies, and commits
- require same-transaction revalidation before final writes
- qualify contracts, redaction, and the refreshed API surface
Diffstat:
13 files changed, 1055 insertions(+), 5 deletions(-)
diff --git a/AGENTS.md b/AGENTS.md
@@ -163,6 +163,10 @@
- Fence final work by dirty generation and policy digest. A stale worker cannot
overwrite newer evidence, and CAS loss leaves no partial report, outbox,
source completion, checkpoint, or job finalization.
+- Carry the exact committed attempt and job identities privately through the
+ sealed manifest, projection, and evaluation chain. A finalization preflight
+ is not commit authority: rerun its exact lease, generation, policy, and
+ attempt validator inside the Step 199 atomic transaction before any write.
## 6. Report, attestation, and publication invariants
diff --git a/README b/README
@@ -276,6 +276,25 @@ cancelled agreement claim is decisively invalid. Every evaluation retains its
projection and exposes exactly one closed stable primary reason code without
adding I/O or ambient authority.
+## Generation-fenced finalization preflight
+
+`RhiReconciliationAttemptRepository::prepare_finalization` consumes one sealed
+evaluation plus the exact live claimed lease. The manifest privately retains
+its Step 190 attempt and job identities without changing canonical manifest
+bytes or its digest, so an older attempt cannot be relabelled with a reclaimed
+worker's distinct attempt authority. One bounded read-only SQLite transaction
+checks the unexpired exact lease, current dirty generation, evidence-policy
+digest, and immutable committed-attempt row before returning a sealed
+`RhiReconciliationFinalizationFence`.
+
+The fence is a preflight capability, not durable commit proof: state may change
+after it is returned. Step 199 must rerun the same validator inside the final
+atomic transaction before its first write. This checkpoint writes no manifest,
+projection, report, signed event, outbox, checkpoint, or job state and performs
+no source, relay, network, filesystem, task, clock, or entropy operation. The
+exact machine contract is
+[`reconciliation_finalization.v1.json`](contracts/services_hardening/reconciliation_finalization.v1.json).
+
## Existing-state runtime foundation
`open_rhi_runtime_foundation` opens only an already initialized database from
diff --git a/contracts/api_baselines/rhi.txt b/contracts/api_baselines/rhi.txt
@@ -159,6 +159,17 @@ pub rhi::RhiReconciliationCommitErrorKind::LeaseLost
pub rhi::RhiReconciliationCommitErrorKind::Storage
impl rhi::RhiReconciliationCommitErrorKind
pub const fn rhi::RhiReconciliationCommitErrorKind::code(self) -> &'static str
+pub enum rhi::RhiReconciliationFinalizationErrorKind
+pub rhi::RhiReconciliationFinalizationErrorKind::AttemptUnavailable
+pub rhi::RhiReconciliationFinalizationErrorKind::CommitOutcomeUnknown
+pub rhi::RhiReconciliationFinalizationErrorKind::GenerationConflict
+pub rhi::RhiReconciliationFinalizationErrorKind::InvalidInput
+pub rhi::RhiReconciliationFinalizationErrorKind::InvalidMode
+pub rhi::RhiReconciliationFinalizationErrorKind::LeaseLost
+pub rhi::RhiReconciliationFinalizationErrorKind::ProjectionUnavailable
+pub rhi::RhiReconciliationFinalizationErrorKind::Storage
+impl rhi::RhiReconciliationFinalizationErrorKind
+pub const fn rhi::RhiReconciliationFinalizationErrorKind::code(self) -> &'static str
pub enum rhi::RhiReconciliationJobErrorKind
pub rhi::RhiReconciliationJobErrorKind::CommitOutcomeUnknown
pub rhi::RhiReconciliationJobErrorKind::DirtyGenerationConflict
@@ -656,6 +667,8 @@ pub async fn rhi::RhiReconciliationAttemptRepository<'_>::commit_source_replays<
impl rhi::RhiReconciliationAttemptRepository<'_>
pub const fn rhi::RhiReconciliationAttemptRepository<'_>::descriptor(&self) -> rhi::RhiStateRepositoryDescriptor
pub const fn rhi::RhiReconciliationAttemptRepository<'_>::kind(&self) -> rhi::RhiStateRepositoryKind
+impl rhi::RhiReconciliationAttemptRepository<'_>
+pub async fn rhi::RhiReconciliationAttemptRepository<'_>::prepare_finalization(&self, rhi::RhiReconciliationLease, rhi::RhiReconciliationEvaluation, rhi::RhiReconciliationUnixMilliseconds) -> core::result::Result<rhi::RhiReconciliationFinalizationFence, rhi::RhiReconciliationFinalizationError>
impl core::fmt::Debug for rhi::RhiReconciliationAttemptRepository<'_>
pub fn rhi::RhiReconciliationAttemptRepository<'_>::fmt(&self, &mut core::fmt::Formatter<'_>) -> core::fmt::Result
pub struct rhi::RhiReconciliationAttemptResults
@@ -684,6 +697,21 @@ pub const fn rhi::RhiReconciliationEvaluation::projection(&self) -> &rhi::RhiRec
pub const fn rhi::RhiReconciliationEvaluation::reason_codes(&self) -> &[rhi::RhiReconciliationReasonCode]
impl core::fmt::Debug for rhi::RhiReconciliationEvaluation
pub fn rhi::RhiReconciliationEvaluation::fmt(&self, &mut core::fmt::Formatter<'_>) -> core::fmt::Result
+pub struct rhi::RhiReconciliationFinalizationError
+impl rhi::RhiReconciliationFinalizationError
+pub const fn rhi::RhiReconciliationFinalizationError::code(self) -> &'static str
+pub const fn rhi::RhiReconciliationFinalizationError::kind(self) -> rhi::RhiReconciliationFinalizationErrorKind
+impl core::error::Error for rhi::RhiReconciliationFinalizationError
+impl core::fmt::Debug for rhi::RhiReconciliationFinalizationError
+pub fn rhi::RhiReconciliationFinalizationError::fmt(&self, &mut core::fmt::Formatter<'_>) -> core::fmt::Result
+impl core::fmt::Display for rhi::RhiReconciliationFinalizationError
+pub fn rhi::RhiReconciliationFinalizationError::fmt(&self, &mut core::fmt::Formatter<'_>) -> core::fmt::Result
+pub struct rhi::RhiReconciliationFinalizationFence
+impl rhi::RhiReconciliationFinalizationFence
+pub const fn rhi::RhiReconciliationFinalizationFence::contract_version(&self) -> u32
+pub const fn rhi::RhiReconciliationFinalizationFence::evaluation(&self) -> &rhi::RhiReconciliationEvaluation
+impl core::fmt::Debug for rhi::RhiReconciliationFinalizationFence
+pub fn rhi::RhiReconciliationFinalizationFence::fmt(&self, &mut core::fmt::Formatter<'_>) -> core::fmt::Result
pub struct rhi::RhiReconciliationJob
impl rhi::RhiReconciliationJob
pub const fn rhi::RhiReconciliationJob::attempt_count(self) -> u16
@@ -1272,6 +1300,7 @@ pub const rhi::RHI_PROVIDER_CONTRACT_VERSION: u32
pub const rhi::RHI_RECONCILIATION_ATTEMPT_CONTRACT_VERSION: u32
pub const rhi::RHI_RECONCILIATION_ATTEMPT_MAX_SOURCES: usize
pub const rhi::RHI_RECONCILIATION_COMMIT_CONTRACT_VERSION: u32
+pub const rhi::RHI_RECONCILIATION_FINALIZATION_CONTRACT_VERSION: u32
pub const rhi::RHI_RECONCILIATION_JOB_CONTRACT_VERSION: u32
pub const rhi::RHI_RECONCILIATION_JOB_MAX_ACTIVE: u32
pub const rhi::RHI_RECONCILIATION_MANIFEST_CONTRACT_VERSION: u32
diff --git a/contracts/services_hardening/reconciliation_finalization.v1.json b/contracts/services_hardening/reconciliation_finalization.v1.json
@@ -0,0 +1,86 @@
+{
+ "schema": "radroots.rhi.reconciliation-finalization",
+ "schema_version": 1,
+ "contract_version": 1,
+ "purpose": "read_only_preflight_for_the_later_generation_fenced_atomic_finalization",
+ "input": {
+ "evaluation_authority": "sealed_step_194_evaluation_only",
+ "lease_authority": "exact_live_claimed_reconciliation_lease",
+ "time_authority": "caller_injected_integer_utc_milliseconds",
+ "ownership": "evaluation_consumed_and_retained_inside_fence"
+ },
+ "private_identity_chain": {
+ "manifest_retains": [
+ "step_190_attempt_id",
+ "step_190_job_id"
+ ],
+ "canonical_manifest_wire_changed": false,
+ "required_equalities": [
+ "manifest_job_id_equals_lease_job_id",
+ "manifest_attempt_id_equals_domain_separated_current_lease_attempt_id",
+ "manifest_trade_equals_lease_job_trade",
+ "manifest_generation_equals_lease_job_input_generation",
+ "manifest_policy_digest_equals_lease_job_policy_digest"
+ ],
+ "projection_digest": "required_available_before_fence_creation",
+ "public_identity_access": false
+ },
+ "preflight_order": [
+ "writable_host",
+ "sealed_identity_chain",
+ "unexpired_input_lease",
+ "exact_current_lease",
+ "exact_current_dirty_generation_and_policy_digest",
+ "exact_committed_attempt_row"
+ ],
+ "durable_validation": {
+ "transaction": "one_bounded_read_only_service_sqlite_transaction",
+ "attempt_table": "evidence_reconciliations",
+ "attempt_match_cardinality": 1,
+ "mutation": false,
+ "step_199_requirement": "rerun_same_validator_inside_atomic_finalization_transaction_before_any_write",
+ "preflight_is_commit_authority": false,
+ "state_may_change_after_preflight": true
+ },
+ "result": {
+ "type": "RhiReconciliationFinalizationFence",
+ "sealed": true,
+ "clone": false,
+ "serialize": false,
+ "caller_forgeable": false,
+ "retains_evaluation": true,
+ "debug": "coverage_and_outcome_only",
+ "errors": "crate_owned_source_free_redacted"
+ },
+ "effects": {
+ "sqlite_read": true,
+ "sqlite_write": false,
+ "filesystem": false,
+ "source_or_relay": false,
+ "network": false,
+ "task_spawn": false,
+ "ambient_clock": false,
+ "ambient_entropy": false,
+ "report_or_attestation": false,
+ "publication": false,
+ "job_finalization": false
+ },
+ "forbidden": [
+ "old_attempt_relabelled_with_new_lease",
+ "stale_generation_fence",
+ "policy_digest_substitution",
+ "projection_digest_absence_as_finalizable",
+ "preflight_used_as_durable_commit_proof",
+ "report_or_outbox_write",
+ "checkpoint_advance",
+ "job_state_transition"
+ ],
+ "deferred": [
+ "canonical_report_statement_supersession_and_signed_attestation",
+ "manifest_projection_report_and_signed_event_schema",
+ "atomic_finalization_transaction",
+ "publication",
+ "job_finalization",
+ "integration_wave_qualification"
+ ]
+}
diff --git a/contracts/services_hardening/reconciliation_manifest.v1.json b/contracts/services_hardening/reconciliation_manifest.v1.json
@@ -86,6 +86,13 @@
"reducer_owner": "reconciliation_reducer.v1.json",
"outcome_owner": "reconciliation_outcome.v1.json"
},
+ "private_finalization_identity": {
+ "attempt_id": "exact_step_190_committed_attempt_id",
+ "job_id": "exact_step_190_claimed_job_id",
+ "canonical_manifest_wire_changed": false,
+ "public_access": false,
+ "owner": "reconciliation_finalization.v1.json"
+ },
"effects": {
"sqlite": false,
"filesystem": false,
diff --git a/src/lib.rs b/src/lib.rs
@@ -10,6 +10,7 @@ mod identity_credential;
mod identity_envelope;
mod reconciliation_attempt;
mod reconciliation_commit;
+mod reconciliation_finalization;
mod reconciliation_job;
mod reconciliation_manifest;
mod reconciliation_reducer;
@@ -83,6 +84,10 @@ pub use reconciliation_commit::{
RHI_RECONCILIATION_COMMIT_CONTRACT_VERSION, RhiReconciliationCommitError,
RhiReconciliationCommitErrorKind, RhiReconciliationSourceCommitOutcome,
};
+pub use reconciliation_finalization::{
+ RHI_RECONCILIATION_FINALIZATION_CONTRACT_VERSION, RhiReconciliationFinalizationError,
+ RhiReconciliationFinalizationErrorKind, RhiReconciliationFinalizationFence,
+};
pub use reconciliation_job::{
RHI_RECONCILIATION_JOB_CONTRACT_VERSION, RHI_RECONCILIATION_JOB_MAX_ACTIVE,
RhiReconciliationJob, RhiReconciliationJobError, RhiReconciliationJobErrorKind,
diff --git a/src/reconciliation_attempt.rs b/src/reconciliation_attempt.rs
@@ -631,7 +631,10 @@ fn absolute_deadline(
.map_err(|_| error(RhiReconciliationAttemptErrorKind::InvalidInput))
}
-fn attempt_id(job_id: RhiReconciliationJobId, attempt_count: u16) -> RhiReconciliationAttemptId {
+pub(crate) fn attempt_id(
+ job_id: RhiReconciliationJobId,
+ attempt_count: u16,
+) -> RhiReconciliationAttemptId {
let mut hasher = Sha256::new();
hasher.update(ATTEMPT_ID_DOMAIN);
hasher.update(job_id.as_bytes());
diff --git a/src/reconciliation_finalization.rs b/src/reconciliation_finalization.rs
@@ -0,0 +1,382 @@
+//! Generation-fenced reconciliation-finalization preflight.
+
+use core::fmt;
+use std::error::Error;
+
+use radroots_event::id::TradeId;
+use radroots_service_sqlite::{
+ ServiceSqliteTransaction, ServiceSqliteTransactionError, ServiceSqliteTransactionErrorKind,
+};
+
+use crate::{
+ RhiEvidencePolicyDigest, RhiReconciliationAttemptId, RhiReconciliationAttemptRepository,
+ RhiReconciliationEvaluation, RhiReconciliationJobId, RhiReconciliationJobState,
+ RhiReconciliationLease, RhiReconciliationUnixMilliseconds, RhiStateHostMode,
+ reconciliation_attempt::attempt_id,
+ reconciliation_job::{LeaseValidationError, validate_exact_lease},
+ source_ingest::{SourceOperationError, read_dirty},
+};
+
+/// Exact version of the reconciliation-finalization fence contract.
+pub const RHI_RECONCILIATION_FINALIZATION_CONTRACT_VERSION: u32 = 1;
+
+const MATCH_COMMITTED_ATTEMPT_SQL: &str = r#"SELECT COUNT(*)
+FROM evidence_reconciliations
+WHERE attempt_id = ? AND job_id = ? AND trade_id = ? AND input_generation = ?
+ AND evidence_policy_sha256 = ?"#;
+
+/// Stable source-free failure class for finalization preflight.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum RhiReconciliationFinalizationErrorKind {
+ InvalidMode,
+ InvalidInput,
+ LeaseLost,
+ GenerationConflict,
+ AttemptUnavailable,
+ ProjectionUnavailable,
+ Storage,
+ CommitOutcomeUnknown,
+}
+
+impl RhiReconciliationFinalizationErrorKind {
+ /// Returns the stable machine-readable failure code.
+ #[must_use]
+ pub const fn code(self) -> &'static str {
+ match self {
+ Self::InvalidMode => "reconciliation_finalization_mode_invalid",
+ Self::InvalidInput => "reconciliation_finalization_input_invalid",
+ Self::LeaseLost => "reconciliation_finalization_lease_lost",
+ Self::GenerationConflict => "reconciliation_finalization_generation_conflict",
+ Self::AttemptUnavailable => "reconciliation_finalization_attempt_unavailable",
+ Self::ProjectionUnavailable => "reconciliation_finalization_projection_unavailable",
+ Self::Storage => "reconciliation_finalization_storage_failed",
+ Self::CommitOutcomeUnknown => "reconciliation_finalization_outcome_unknown",
+ }
+ }
+}
+
+/// Redacted source-free finalization-preflight failure.
+#[derive(Clone, Copy, PartialEq, Eq)]
+pub struct RhiReconciliationFinalizationError {
+ kind: RhiReconciliationFinalizationErrorKind,
+}
+
+impl RhiReconciliationFinalizationError {
+ /// Returns the stable failure class.
+ #[must_use]
+ pub const fn kind(self) -> RhiReconciliationFinalizationErrorKind {
+ self.kind
+ }
+
+ /// Returns the stable machine-readable failure code.
+ #[must_use]
+ pub const fn code(self) -> &'static str {
+ self.kind.code()
+ }
+}
+
+impl fmt::Display for RhiReconciliationFinalizationError {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter.write_str(match self.kind {
+ RhiReconciliationFinalizationErrorKind::InvalidMode => {
+ "RHI reconciliation finalization requires writable state"
+ }
+ RhiReconciliationFinalizationErrorKind::InvalidInput => {
+ "RHI reconciliation finalization input is invalid"
+ }
+ RhiReconciliationFinalizationErrorKind::LeaseLost => {
+ "RHI reconciliation finalization lease is no longer authoritative"
+ }
+ RhiReconciliationFinalizationErrorKind::GenerationConflict => {
+ "RHI reconciliation finalization generation changed"
+ }
+ RhiReconciliationFinalizationErrorKind::AttemptUnavailable => {
+ "RHI reconciliation finalization attempt is unavailable"
+ }
+ RhiReconciliationFinalizationErrorKind::ProjectionUnavailable => {
+ "RHI reconciliation finalization projection is unavailable"
+ }
+ RhiReconciliationFinalizationErrorKind::Storage => {
+ "RHI reconciliation finalization preflight failed"
+ }
+ RhiReconciliationFinalizationErrorKind::CommitOutcomeUnknown => {
+ "RHI reconciliation finalization preflight outcome is unknown"
+ }
+ })
+ }
+}
+
+impl fmt::Debug for RhiReconciliationFinalizationError {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter
+ .debug_struct("RhiReconciliationFinalizationError")
+ .field("kind", &self.kind)
+ .finish()
+ }
+}
+
+impl Error for RhiReconciliationFinalizationError {}
+
+/// Sealed preflight capability for one exact evaluated reconciliation attempt.
+///
+/// This value proves only that its lease, generation, policy, and committed
+/// attempt matched during one bounded read-only preflight transaction. The
+/// eventual Step199 writer must rerun the same validator inside its atomic
+/// transaction before any mutation.
+///
+/// ```compile_fail
+/// use rhi::RhiReconciliationFinalizationFence;
+///
+/// let _forged = RhiReconciliationFinalizationFence { evaluation: todo!() };
+/// ```
+pub struct RhiReconciliationFinalizationFence {
+ lease: RhiReconciliationLease,
+ identity: FinalizationIdentity,
+ evaluation: RhiReconciliationEvaluation,
+}
+
+impl RhiReconciliationFinalizationFence {
+ /// Returns the exact finalization-fence contract version.
+ #[must_use]
+ pub const fn contract_version(&self) -> u32 {
+ RHI_RECONCILIATION_FINALIZATION_CONTRACT_VERSION
+ }
+
+ /// Returns the sealed Step194 evaluation retained by this preflight.
+ #[must_use]
+ pub const fn evaluation(&self) -> &RhiReconciliationEvaluation {
+ &self.evaluation
+ }
+}
+
+impl fmt::Debug for RhiReconciliationFinalizationFence {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter
+ .debug_struct("RhiReconciliationFinalizationFence")
+ .field("coverage", &self.evaluation.coverage())
+ .field("outcome", &self.evaluation.outcome())
+ .finish_non_exhaustive()
+ }
+}
+
+#[derive(Clone, Copy)]
+struct FinalizationIdentity {
+ attempt_id: RhiReconciliationAttemptId,
+ job_id: RhiReconciliationJobId,
+ trade_id: TradeId,
+ generation: u64,
+ policy_digest: RhiEvidencePolicyDigest,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum FinalizationOperationError {
+ LeaseLost,
+ GenerationConflict,
+ AttemptUnavailable,
+ Storage,
+}
+
+impl RhiReconciliationAttemptRepository<'_> {
+ /// Performs one bounded read-only preflight for later atomic finalization.
+ ///
+ /// The returned capability is not commit authority. Durable finalization
+ /// must rerun the same exact checks inside its Step199 write transaction.
+ pub async fn prepare_finalization(
+ &self,
+ lease: RhiReconciliationLease,
+ evaluation: RhiReconciliationEvaluation,
+ now: RhiReconciliationUnixMilliseconds,
+ ) -> Result<RhiReconciliationFinalizationFence, RhiReconciliationFinalizationError> {
+ if self.host().mode() != RhiStateHostMode::ReadWriteExisting {
+ return Err(failure(RhiReconciliationFinalizationErrorKind::InvalidMode));
+ }
+ let identity = finalization_identity(lease, &evaluation, now)?;
+ let fence = RhiReconciliationFinalizationFence {
+ lease,
+ identity,
+ evaluation,
+ };
+ self.host()
+ .sqlite_host()
+ .transaction(move |transaction| {
+ Box::pin(async move {
+ validate_finalization_fence(transaction, &fence, now).await?;
+ Ok(fence)
+ })
+ })
+ .await
+ .map_err(map_transaction_error)
+ }
+}
+
+pub(crate) async fn validate_finalization_fence(
+ transaction: &mut ServiceSqliteTransaction<'_>,
+ fence: &RhiReconciliationFinalizationFence,
+ now: RhiReconciliationUnixMilliseconds,
+) -> Result<(), FinalizationOperationError> {
+ validate_finalization_identity(transaction, fence.lease, fence.identity, now).await
+}
+
+fn finalization_identity(
+ lease: RhiReconciliationLease,
+ evaluation: &RhiReconciliationEvaluation,
+ now: RhiReconciliationUnixMilliseconds,
+) -> Result<FinalizationIdentity, RhiReconciliationFinalizationError> {
+ let job = lease.job();
+ let manifest = evaluation.projection().manifest();
+ if job.state() != RhiReconciliationJobState::Leased
+ || job.attempt_count() == 0
+ || manifest.job_id() != job.id()
+ || manifest.attempt_id() != attempt_id(job.id(), job.attempt_count())
+ || manifest.trade_id() != &job.trade_id()
+ || manifest.trade_generation() != job.input_generation()
+ || manifest.inner().evidence_policy_digest().as_bytes()
+ != job.evidence_policy_digest().as_bytes()
+ {
+ return Err(failure(
+ RhiReconciliationFinalizationErrorKind::InvalidInput,
+ ));
+ }
+ if now >= lease.lease_expires() {
+ return Err(failure(RhiReconciliationFinalizationErrorKind::LeaseLost));
+ }
+ if evaluation.projection().digest().is_none() {
+ return Err(failure(
+ RhiReconciliationFinalizationErrorKind::ProjectionUnavailable,
+ ));
+ }
+ Ok(FinalizationIdentity {
+ attempt_id: manifest.attempt_id(),
+ job_id: manifest.job_id(),
+ trade_id: job.trade_id(),
+ generation: job.input_generation(),
+ policy_digest: job.evidence_policy_digest(),
+ })
+}
+
+async fn validate_finalization_identity(
+ transaction: &mut ServiceSqliteTransaction<'_>,
+ lease: RhiReconciliationLease,
+ identity: FinalizationIdentity,
+ now: RhiReconciliationUnixMilliseconds,
+) -> Result<(), FinalizationOperationError> {
+ if now >= lease.lease_expires() {
+ return Err(FinalizationOperationError::LeaseLost);
+ }
+ validate_exact_lease(transaction, lease)
+ .await
+ .map_err(|error| match error {
+ LeaseValidationError::LeaseLost => FinalizationOperationError::LeaseLost,
+ LeaseValidationError::Storage => FinalizationOperationError::Storage,
+ })?;
+ read_dirty(transaction, identity.trade_id)
+ .await
+ .map_err(|error| match error {
+ SourceOperationError::GenerationConflict => {
+ FinalizationOperationError::GenerationConflict
+ }
+ SourceOperationError::Persistence(_) | SourceOperationError::Storage => {
+ FinalizationOperationError::Storage
+ }
+ })?
+ .filter(|dirty| {
+ dirty.generation.get() == identity.generation && dirty.policy == identity.policy_digest
+ })
+ .ok_or(FinalizationOperationError::GenerationConflict)?;
+ let generation =
+ i64::try_from(identity.generation).map_err(|_| FinalizationOperationError::Storage)?;
+ let count = sqlx::query_scalar::<_, i64>(MATCH_COMMITTED_ATTEMPT_SQL)
+ .bind(identity.attempt_id.as_bytes().as_slice())
+ .bind(identity.job_id.as_bytes().as_slice())
+ .bind(identity.trade_id.as_bytes().as_slice())
+ .bind(generation)
+ .bind(identity.policy_digest.as_bytes().as_slice())
+ .fetch_one(&mut *transaction)
+ .await
+ .map_err(|_| FinalizationOperationError::Storage)?;
+ if count == 1 {
+ Ok(())
+ } else {
+ Err(FinalizationOperationError::AttemptUnavailable)
+ }
+}
+
+fn map_transaction_error(
+ error: ServiceSqliteTransactionError<FinalizationOperationError>,
+) -> RhiReconciliationFinalizationError {
+ if error.kind() == ServiceSqliteTransactionErrorKind::CommitOutcomeUnknown {
+ return failure(RhiReconciliationFinalizationErrorKind::CommitOutcomeUnknown);
+ }
+ failure(match error.operation_error().copied() {
+ Some(FinalizationOperationError::LeaseLost) => {
+ RhiReconciliationFinalizationErrorKind::LeaseLost
+ }
+ Some(FinalizationOperationError::GenerationConflict) => {
+ RhiReconciliationFinalizationErrorKind::GenerationConflict
+ }
+ Some(FinalizationOperationError::AttemptUnavailable) => {
+ RhiReconciliationFinalizationErrorKind::AttemptUnavailable
+ }
+ Some(FinalizationOperationError::Storage) | None => {
+ RhiReconciliationFinalizationErrorKind::Storage
+ }
+ })
+}
+
+const fn failure(
+ kind: RhiReconciliationFinalizationErrorKind,
+) -> RhiReconciliationFinalizationError {
+ RhiReconciliationFinalizationError { kind }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn error_inventory_is_exact_source_free_and_redacted() {
+ let cases = [
+ (
+ RhiReconciliationFinalizationErrorKind::InvalidMode,
+ "reconciliation_finalization_mode_invalid",
+ ),
+ (
+ RhiReconciliationFinalizationErrorKind::InvalidInput,
+ "reconciliation_finalization_input_invalid",
+ ),
+ (
+ RhiReconciliationFinalizationErrorKind::LeaseLost,
+ "reconciliation_finalization_lease_lost",
+ ),
+ (
+ RhiReconciliationFinalizationErrorKind::GenerationConflict,
+ "reconciliation_finalization_generation_conflict",
+ ),
+ (
+ RhiReconciliationFinalizationErrorKind::AttemptUnavailable,
+ "reconciliation_finalization_attempt_unavailable",
+ ),
+ (
+ RhiReconciliationFinalizationErrorKind::ProjectionUnavailable,
+ "reconciliation_finalization_projection_unavailable",
+ ),
+ (
+ RhiReconciliationFinalizationErrorKind::Storage,
+ "reconciliation_finalization_storage_failed",
+ ),
+ (
+ RhiReconciliationFinalizationErrorKind::CommitOutcomeUnknown,
+ "reconciliation_finalization_outcome_unknown",
+ ),
+ ];
+ for (kind, code) in cases {
+ let error = failure(kind);
+ assert_eq!(error.kind(), kind);
+ assert_eq!(error.code(), code);
+ assert!(Error::source(&error).is_none());
+ let rendered = format!("{error} {error:?}");
+ assert!(!rendered.contains("11111111"));
+ assert!(!rendered.contains("trade-primary"));
+ }
+ }
+}
diff --git a/src/reconciliation_manifest.rs b/src/reconciliation_manifest.rs
@@ -16,7 +16,8 @@ use radroots_trade::evidence::{
use sha2::{Digest, Sha256};
use crate::{
- RhiReconciliationAttemptPlan, RhiReconciliationSourceCommitOutcome, RhiTradeSourceCompletion,
+ RhiReconciliationAttemptId, RhiReconciliationAttemptPlan, RhiReconciliationJobId,
+ RhiReconciliationSourceCommitOutcome, RhiTradeSourceCompletion,
reconciliation_commit::committed_inventory_digest,
reconciliation_replay::{
RhiReconciliationReplayCommitFact, RhiReconciliationReplayCommitParts,
@@ -113,6 +114,8 @@ impl Error for RhiReconciliationManifestError {}
/// ```
pub struct RhiReconciliationManifest {
inner: RadrootsTradeEvidenceManifestV1,
+ attempt_id: RhiReconciliationAttemptId,
+ job_id: RhiReconciliationJobId,
reducer_mutations: Box<[RhiReducerMutationMaterial]>,
}
@@ -181,6 +184,14 @@ impl RhiReconciliationManifest {
&self.inner
}
+ pub(crate) const fn attempt_id(&self) -> RhiReconciliationAttemptId {
+ self.attempt_id
+ }
+
+ pub(crate) const fn job_id(&self) -> RhiReconciliationJobId {
+ self.job_id
+ }
+
pub(crate) fn reducer_mutations(&self) -> &[RhiReducerMutationMaterial] {
&self.reducer_mutations
}
@@ -212,6 +223,8 @@ impl RhiReconciliationSourceCommitOutcome {
}
pub(crate) struct RhiCommittedManifestMaterial {
+ attempt_id: RhiReconciliationAttemptId,
+ job_id: RhiReconciliationJobId,
trade_id: TradeId,
generation: NonZeroU64,
policy_digest: RadrootsTradeEvidencePolicyDigestV1,
@@ -298,6 +311,8 @@ pub(crate) fn committed_manifest_material(
}
Ok(RhiCommittedManifestMaterial {
+ attempt_id: plan.id(),
+ job_id: plan.job_id(),
trade_id,
generation,
policy_digest,
@@ -341,6 +356,8 @@ fn freeze_manifest(
.map_err(|_| error(RhiReconciliationManifestErrorKind::InvalidCommittedInventory))?;
Ok(RhiReconciliationManifest {
inner,
+ attempt_id: material.attempt_id,
+ job_id: material.job_id,
reducer_mutations: material.reducer_mutations,
})
}
diff --git a/tests/package_boundary.rs b/tests/package_boundary.rs
@@ -13,6 +13,7 @@ const RUNTIME_ADAPTER_CONTRACT: &str =
const RUNTIME_FOUNDATION: &str = include_str!("../src/runtime_foundation.rs");
const RECONCILIATION_ATTEMPTS: &str = include_str!("../src/reconciliation_attempt.rs");
const RECONCILIATION_COMMIT: &str = include_str!("../src/reconciliation_commit.rs");
+const RECONCILIATION_FINALIZATION: &str = include_str!("../src/reconciliation_finalization.rs");
const RECONCILIATION_MANIFEST: &str = include_str!("../src/reconciliation_manifest.rs");
const RECONCILIATION_REDUCER: &str = include_str!("../src/reconciliation_reducer.rs");
const RECONCILIATION_JOBS: &str = include_str!("../src/reconciliation_job.rs");
@@ -23,6 +24,8 @@ const RECONCILIATION_REPLAY_CONTRACT: &str =
include_str!("../contracts/services_hardening/reconciliation_replay.v1.json");
const RECONCILIATION_COMMIT_CONTRACT: &str =
include_str!("../contracts/services_hardening/reconciliation_commit.v1.json");
+const RECONCILIATION_FINALIZATION_CONTRACT: &str =
+ include_str!("../contracts/services_hardening/reconciliation_finalization.v1.json");
const RECONCILIATION_MANIFEST_CONTRACT: &str =
include_str!("../contracts/services_hardening/reconciliation_manifest.v1.json");
const RECONCILIATION_REDUCER_CONTRACT: &str =
@@ -47,6 +50,7 @@ const SOURCES: &[&str] = &[
include_str!("../src/identity_envelope.rs"),
include_str!("../src/reconciliation_attempt.rs"),
include_str!("../src/reconciliation_commit.rs"),
+ include_str!("../src/reconciliation_finalization.rs"),
include_str!("../src/reconciliation_job.rs"),
include_str!("../src/reconciliation_manifest.rs"),
include_str!("../src/reconciliation_reducer.rs"),
@@ -119,6 +123,7 @@ fn state_catalog_module_is_private_and_root_api_is_curated() {
"identity_envelope",
"reconciliation_attempt",
"reconciliation_commit",
+ "reconciliation_finalization",
"reconciliation_job",
"reconciliation_manifest",
"reconciliation_reducer",
@@ -164,6 +169,9 @@ fn state_catalog_module_is_private_and_root_api_is_curated() {
"RhiReconciliationAttemptResults",
"RhiReconciliationSourceCommitOutcome",
"RhiReconciliationCommitErrorKind",
+ "RhiReconciliationFinalizationFence",
+ "RhiReconciliationFinalizationErrorKind",
+ "RHI_RECONCILIATION_FINALIZATION_CONTRACT_VERSION",
"RhiReconciliationManifest",
"RhiReconciliationManifestErrorKind",
"RhiReconciliationScopePrerequisites",
@@ -261,6 +269,50 @@ fn reconciliation_commit_is_atomic_bounded_and_sealed() {
}
#[test]
+fn reconciliation_finalization_is_attempt_bound_nonmutating_and_revalidated() {
+ let contract: serde_json::Value = serde_json::from_str(RECONCILIATION_FINALIZATION_CONTRACT)
+ .expect("reconciliation-finalization contract");
+ assert_eq!(
+ contract["schema"],
+ "radroots.rhi.reconciliation-finalization"
+ );
+ assert_eq!(contract["contract_version"], 1);
+ assert_eq!(contract["effects"]["sqlite_read"], true);
+ assert_eq!(contract["effects"]["sqlite_write"], false);
+ assert_eq!(
+ contract["durable_validation"]["preflight_is_commit_authority"],
+ false
+ );
+ for required in [
+ "pub async fn prepare_finalization(",
+ "validate_finalization_fence(",
+ "validate_exact_lease(transaction, lease)",
+ "read_dirty(transaction, identity.trade_id)",
+ "MATCH_COMMITTED_ATTEMPT_SQL",
+ "attempt_id(job.id(), job.attempt_count())",
+ ] {
+ assert!(
+ RECONCILIATION_FINALIZATION.contains(required),
+ "reconciliation finalization is missing {required}"
+ );
+ }
+ for forbidden in [
+ "INSERT ",
+ "UPDATE ",
+ "DELETE ",
+ "tokio::spawn",
+ "SystemTime",
+ ] {
+ assert!(
+ !RECONCILIATION_FINALIZATION.contains(forbidden),
+ "reconciliation finalization gained forbidden authority {forbidden}"
+ );
+ }
+ assert!(!ROOT.contains("pub mod reconciliation_finalization"));
+ assert!(!PUBLIC_API.contains("rhi::reconciliation_finalization::"));
+}
+
+#[test]
fn reconciliation_manifest_is_canonical_sealed_and_effect_free() {
let contract: serde_json::Value = serde_json::from_str(RECONCILIATION_MANIFEST_CONTRACT)
.expect("reconciliation-manifest contract");
@@ -402,7 +454,7 @@ fn public_errors_are_crate_owned_redacted_and_source_free() {
.lines()
.filter(|line| line.starts_with("pub struct rhi::") && line.ends_with("Error"))
.count();
- assert_eq!(public_error_count, 22);
+ assert_eq!(public_error_count, 23);
}
#[test]
@@ -826,6 +878,9 @@ fn readme_freezes_the_root_only_boundary_and_exact_baseline() {
"[`reconciliation_reducer.v1.json`](contracts/services_hardening/reconciliation_reducer.v1.json)",
"binds the promoted shared `radroots.trade.reducer.v1`",
"[`reconciliation_outcome.v1.json`](contracts/services_hardening/reconciliation_outcome.v1.json)",
+ "## Generation-fenced finalization preflight",
+ "[`reconciliation_finalization.v1.json`](contracts/services_hardening/reconciliation_finalization.v1.json)",
+ "Step 199 must rerun the same validator inside the final",
"Coverage is exactly `Missing`, `Partial`, `ScopeSatisfied`, or `Unsupported`",
"Missing, partial, unsupported,",
"The Step 192 integration-wave qualification proves that concurrent exact",
diff --git a/tests/services_hardening_reconciliation_finalization_contract.rs b/tests/services_hardening_reconciliation_finalization_contract.rs
@@ -0,0 +1,130 @@
+#![forbid(unsafe_code)]
+
+use rhi::RHI_RECONCILIATION_FINALIZATION_CONTRACT_VERSION;
+use serde_json::json;
+
+const CONTRACT: &str =
+ include_str!("../contracts/services_hardening/reconciliation_finalization.v1.json");
+const ROOT: &str = include_str!("../src/lib.rs");
+const SOURCE: &str = include_str!("../src/reconciliation_finalization.rs");
+const MANIFEST: &str = include_str!("../src/reconciliation_manifest.rs");
+const README: &str = include_str!("../README");
+
+#[test]
+fn machine_contract_freezes_the_complete_step_195_boundary() {
+ let contract: serde_json::Value = serde_json::from_str(CONTRACT).expect("contract");
+ assert_eq!(
+ contract["schema"],
+ "radroots.rhi.reconciliation-finalization"
+ );
+ assert_eq!(contract["schema_version"], 1);
+ assert_eq!(
+ contract["contract_version"],
+ RHI_RECONCILIATION_FINALIZATION_CONTRACT_VERSION
+ );
+ assert_eq!(
+ contract["private_identity_chain"]["manifest_retains"],
+ json!(["step_190_attempt_id", "step_190_job_id"])
+ );
+ assert_eq!(
+ contract["private_identity_chain"]["canonical_manifest_wire_changed"],
+ false
+ );
+ assert_eq!(
+ contract["preflight_order"],
+ json!([
+ "writable_host",
+ "sealed_identity_chain",
+ "unexpired_input_lease",
+ "exact_current_lease",
+ "exact_current_dirty_generation_and_policy_digest",
+ "exact_committed_attempt_row"
+ ])
+ );
+ assert_eq!(contract["durable_validation"]["mutation"], false);
+ assert_eq!(
+ contract["durable_validation"]["step_199_requirement"],
+ "rerun_same_validator_inside_atomic_finalization_transaction_before_any_write"
+ );
+ assert_eq!(
+ contract["durable_validation"]["preflight_is_commit_authority"],
+ false
+ );
+ assert_eq!(contract["effects"]["sqlite_read"], true);
+ assert_eq!(contract["effects"]["sqlite_write"], false);
+ for effect in [
+ "filesystem",
+ "source_or_relay",
+ "network",
+ "task_spawn",
+ "ambient_clock",
+ "ambient_entropy",
+ "report_or_attestation",
+ "publication",
+ "job_finalization",
+ ] {
+ assert_eq!(contract["effects"][effect], false, "effect {effect}");
+ }
+}
+
+#[test]
+fn finalization_boundary_is_sealed_identity_bound_and_nonmutating() {
+ assert!(ROOT.contains("mod reconciliation_finalization;"));
+ assert!(!ROOT.contains("pub mod reconciliation_finalization;"));
+ for required in [
+ "RhiReconciliationFinalizationFence",
+ "RhiReconciliationFinalizationError",
+ "RhiReconciliationFinalizationErrorKind",
+ "RHI_RECONCILIATION_FINALIZATION_CONTRACT_VERSION",
+ ] {
+ assert!(ROOT.contains(required), "root API is missing {required}");
+ }
+ for required in [
+ "pub async fn prepare_finalization(",
+ "validate_finalization_fence(",
+ "validate_exact_lease(transaction, lease)",
+ "read_dirty(transaction, identity.trade_id)",
+ "MATCH_COMMITTED_ATTEMPT_SQL",
+ "manifest.attempt_id() != attempt_id(job.id(), job.attempt_count())",
+ "now >= lease.lease_expires()",
+ "evaluation.projection().digest().is_none()",
+ ] {
+ assert!(
+ SOURCE.contains(required),
+ "finalization boundary is missing {required}"
+ );
+ }
+ for required in [
+ "attempt_id: RhiReconciliationAttemptId",
+ "job_id: RhiReconciliationJobId",
+ "attempt_id: plan.id()",
+ "job_id: plan.job_id()",
+ ] {
+ assert!(
+ MANIFEST.contains(required),
+ "manifest is missing {required}"
+ );
+ }
+ for forbidden in [
+ "INSERT ",
+ "UPDATE ",
+ "DELETE ",
+ "std::fs",
+ "std::net",
+ "tokio::spawn",
+ "SystemTime",
+ "thread_rng",
+ "OsRng",
+ "pub fn new(",
+ "pub const fn new(",
+ ] {
+ assert!(
+ !SOURCE.contains(forbidden),
+ "finalization boundary gained forbidden authority {forbidden}"
+ );
+ }
+ assert!(README.contains("## Generation-fenced finalization preflight"));
+ assert!(README.contains(
+ "[`reconciliation_finalization.v1.json`](contracts/services_hardening/reconciliation_finalization.v1.json)"
+ ));
+}
diff --git a/tests/services_hardening_reconciliation_jobs.rs b/tests/services_hardening_reconciliation_jobs.rs
@@ -9,8 +9,9 @@ use rhi::{
RadrootsHostEnvironment, RadrootsPathResolver, RadrootsPlatform,
RhiReconciliationAttemptErrorKind, RhiReconciliationAttemptPlan,
RhiReconciliationAttemptResults, RhiReconciliationCommitErrorKind,
- RhiReconciliationJobErrorKind, RhiReconciliationJobPolicy, RhiReconciliationJobState,
- RhiReconciliationLease, RhiReconciliationLeaseOwner, RhiReconciliationRetryDelayMilliseconds,
+ RhiReconciliationFinalizationErrorKind, RhiReconciliationJobErrorKind,
+ RhiReconciliationJobPolicy, RhiReconciliationJobState, RhiReconciliationLease,
+ RhiReconciliationLeaseOwner, RhiReconciliationRetryDelayMilliseconds,
RhiReconciliationScopePrerequisites, RhiReconciliationSourceReplayPlan,
RhiReconciliationSourceResult, RhiReconciliationUnixMilliseconds, RhiRuntimeContext,
RhiStateMetadata, RhiTradeMutationAdmissionLimits, RhiTradeMutationAuthoredTimePolicy,
@@ -1412,6 +1413,176 @@ async fn incomplete_results_never_advance_and_stale_leases_fail_closed() {
drop((configuration, metadata, runtime, root));
}
+#[tokio::test]
+async fn finalization_preflight_is_exact_nonmutating_and_redacted() {
+ let (root, runtime, metadata, configuration, host, lease, evaluation) =
+ finalization_fixture("finalization-exact").await;
+ let before = finalization_snapshot(&runtime).await;
+ let fence = host
+ .repositories()
+ .reconciliation_attempts()
+ .prepare_finalization(lease, evaluation, now(1_784_347_206_000))
+ .await
+ .expect("finalization preflight");
+ assert_eq!(fence.contract_version(), 1);
+ assert_eq!(
+ fence
+ .evaluation()
+ .projection()
+ .manifest()
+ .trade_generation(),
+ 2
+ );
+ assert!(fence.evaluation().projection().digest().is_some());
+ let rendered = format!("{fence:?}");
+ assert!(!rendered.contains("11111111"));
+ assert!(!rendered.contains("reconciliation_attempt"));
+ assert_eq!(finalization_snapshot(&runtime).await, before);
+ host.close().await.expect("close");
+ drop((fence, configuration, metadata, runtime, root));
+}
+
+#[tokio::test]
+async fn finalization_rejects_stale_generation_policy_and_expired_or_lost_leases() {
+ let (root, runtime, metadata, configuration, host, lease, evaluation) =
+ finalization_fixture("finalization-generation").await;
+ write_dirty(
+ &runtime,
+ lease.job().trade_id(),
+ 3,
+ *lease.job().evidence_policy_digest().as_bytes(),
+ 1_784_347_207,
+ )
+ .await;
+ let error = host
+ .repositories()
+ .reconciliation_attempts()
+ .prepare_finalization(lease, evaluation, now(1_784_347_207_000))
+ .await
+ .expect_err("stale generation");
+ assert_eq!(
+ error.kind(),
+ RhiReconciliationFinalizationErrorKind::GenerationConflict
+ );
+ assert!(Error::source(&error).is_none());
+ assert!(!format!("{error} {error:?}").contains("11111111"));
+ host.close().await.expect("close");
+ drop((configuration, metadata, runtime, root));
+
+ let (root, runtime, metadata, configuration, host, lease, evaluation) =
+ finalization_fixture("finalization-policy").await;
+ write_dirty(
+ &runtime,
+ lease.job().trade_id(),
+ 3,
+ [0xa5; 32],
+ 1_784_347_207,
+ )
+ .await;
+ let error = host
+ .repositories()
+ .reconciliation_attempts()
+ .prepare_finalization(lease, evaluation, now(1_784_347_207_000))
+ .await
+ .expect_err("stale policy");
+ assert_eq!(
+ error.kind(),
+ RhiReconciliationFinalizationErrorKind::GenerationConflict
+ );
+ host.close().await.expect("close");
+ drop((configuration, metadata, runtime, root));
+
+ let (root, runtime, metadata, configuration, host, lease, evaluation) =
+ finalization_fixture("finalization-expired").await;
+ let error = host
+ .repositories()
+ .reconciliation_attempts()
+ .prepare_finalization(lease, evaluation, lease.lease_expires())
+ .await
+ .expect_err("expired lease");
+ assert_eq!(
+ error.kind(),
+ RhiReconciliationFinalizationErrorKind::LeaseLost
+ );
+ host.close().await.expect("close");
+ drop((configuration, metadata, runtime, root));
+
+ let (root, runtime, metadata, configuration, host, lease, evaluation) =
+ finalization_fixture("finalization-lost").await;
+ host.repositories()
+ .reconciliation_jobs()
+ .record_failure(
+ lease,
+ now(1_784_347_206_000),
+ RhiReconciliationRetryDelayMilliseconds::new(1).expect("delay"),
+ )
+ .await
+ .expect("release lease");
+ let error = host
+ .repositories()
+ .reconciliation_attempts()
+ .prepare_finalization(lease, evaluation, now(1_784_347_206_001))
+ .await
+ .expect_err("lost lease");
+ assert_eq!(
+ error.kind(),
+ RhiReconciliationFinalizationErrorKind::LeaseLost
+ );
+ host.close().await.expect("close");
+ drop((configuration, metadata, runtime, root));
+}
+
+#[tokio::test]
+async fn finalization_rejects_cross_attempt_relabelling_and_read_only_hosts() {
+ let (root, runtime, metadata, configuration, host, lease, evaluation) =
+ finalization_fixture("finalization-attempt").await;
+ let jobs = host.repositories().reconciliation_jobs();
+ jobs.record_failure(
+ lease,
+ now(1_784_347_206_000),
+ RhiReconciliationRetryDelayMilliseconds::new(1).expect("delay"),
+ )
+ .await
+ .expect("retry schedule");
+ let next_lease = jobs
+ .claim_next(owner(0x93), now(1_784_347_206_001))
+ .await
+ .expect("next claim")
+ .expect("reclaimed job");
+ assert_eq!(next_lease.job().attempt_count(), 2);
+ let error = host
+ .repositories()
+ .reconciliation_attempts()
+ .prepare_finalization(next_lease, evaluation, now(1_784_347_206_002))
+ .await
+ .expect_err("old evaluation cannot be relabelled");
+ assert_eq!(
+ error.kind(),
+ RhiReconciliationFinalizationErrorKind::InvalidInput
+ );
+ host.close().await.expect("close");
+ drop((configuration, metadata, runtime, root));
+
+ let (root, runtime, metadata, configuration, host, lease, evaluation) =
+ finalization_fixture("finalization-inspection").await;
+ host.close().await.expect("close writer");
+ let inspection = open_rhi_state_inspection(&runtime, &metadata)
+ .await
+ .expect("inspection");
+ let error = inspection
+ .repositories()
+ .reconciliation_attempts()
+ .prepare_finalization(lease, evaluation, now(1_784_347_206_000))
+ .await
+ .expect_err("inspection cannot prepare finalization");
+ assert_eq!(
+ error.kind(),
+ RhiReconciliationFinalizationErrorKind::InvalidMode
+ );
+ inspection.close().await.expect("inspection close");
+ drop((configuration, metadata, runtime, root));
+}
+
async fn fixture_connection(runtime: &RhiRuntimeContext) -> SqliteConnection {
let options = SqliteConnectOptions::new()
.filename(runtime.artifacts().state_database())
@@ -1422,6 +1593,134 @@ async fn fixture_connection(runtime: &RhiRuntimeContext) -> SqliteConnection {
.expect("offline fixture connection")
}
+async fn finalization_snapshot(
+ runtime: &RhiRuntimeContext,
+) -> (i64, i64, i64, i64, i64, i64, i64, i64) {
+ let mut connection = fixture_connection(runtime).await;
+ let row: (i64, i64, i64, i64, i64, i64, i64, i64) = sqlx::query_as(
+ r#"SELECT
+ (SELECT COUNT(*) FROM reconciliation_jobs),
+ (SELECT COUNT(*) FROM reconciliation_jobs WHERE state = 'leased'),
+ (SELECT SUM(revision) FROM reconciliation_jobs),
+ (SELECT COUNT(*) FROM evidence_reconciliations),
+ (SELECT COUNT(*) FROM evidence_reconciliation_sources),
+ (SELECT COUNT(*) FROM relay_checkpoints),
+ (SELECT generation FROM trade_dirty_generations LIMIT 1),
+ (SELECT COUNT(*) FROM sqlite_schema WHERE name IN (
+ 'evidence_manifests', 'trade_projections', 'attestation_reports',
+ 'signed_attestation_events', 'publication_outbox'
+ ))"#,
+ )
+ .fetch_one(&mut connection)
+ .await
+ .expect("finalization snapshot");
+ connection.close().await.expect("snapshot close");
+ row
+}
+
+async fn finalization_fixture(
+ instance: &str,
+) -> (
+ tempfile::TempDir,
+ RhiRuntimeContext,
+ RhiStateMetadata,
+ rhi::RhiConfigDocumentV1,
+ rhi::RhiStateHost,
+ RhiReconciliationLease,
+ rhi::RhiReconciliationEvaluation,
+) {
+ let started_ms = 1_784_347_200_000;
+ let (root, runtime, metadata, configuration, host, first_lease, first_plan) =
+ replay_fixture(instance, EXAMPLE, started_ms).await;
+ let wire = replay_wire();
+ let first_request = &first_plan.requests()[0];
+ let first_replay = RhiReconciliationSourceReplayPlan::from_request(
+ &first_plan,
+ first_request,
+ &configuration,
+ None,
+ )
+ .expect("first replay plan")
+ .finish(
+ first_request,
+ RhiTradeSourceCompletion::Complete,
+ now(started_ms),
+ now(started_ms + 2_000),
+ [admitted_replay(&configuration, &wire, 1_784_347_200)],
+ )
+ .expect("first replay");
+ let first = host
+ .repositories()
+ .reconciliation_attempts()
+ .commit_source_replays(first_lease, first_plan, [first_replay])
+ .await
+ .expect("first commit");
+ assert!(first.dirty_generation_advanced());
+ let cursor = first.committed_cursors()[0].clone();
+ drop(first);
+
+ let jobs = host.repositories().reconciliation_jobs();
+ let scheduled = jobs
+ .schedule_trade(
+ first_lease.job().trade_id(),
+ configured_policy(&configuration),
+ now(started_ms + 3_000),
+ )
+ .await
+ .expect("schedule final generation");
+ assert_eq!(scheduled.job().input_generation(), 2);
+ let lease = jobs
+ .claim_next(owner(0x92), now(started_ms + 3_000))
+ .await
+ .expect("claim final generation")
+ .expect("final job");
+ let plan =
+ RhiReconciliationAttemptPlan::from_claim(lease, &configuration, now(started_ms + 3_000))
+ .expect("final plan");
+ let request = &plan.requests()[0];
+ let replay = RhiReconciliationSourceReplayPlan::from_request(
+ &plan,
+ request,
+ &configuration,
+ Some(cursor),
+ )
+ .expect("final replay plan")
+ .finish(
+ request,
+ RhiTradeSourceCompletion::Complete,
+ now(started_ms + 3_000),
+ now(started_ms + 5_000),
+ [admitted_replay(&configuration, &wire, 1_784_347_203)],
+ )
+ .expect("final replay");
+ let committed = host
+ .repositories()
+ .reconciliation_attempts()
+ .commit_source_replays(lease, plan, [replay])
+ .await
+ .expect("final commit");
+ assert!(!committed.dirty_generation_advanced());
+ let manifest = committed
+ .into_evidence_manifest(
+ UnixTimeSeconds::new(1_784_347_206),
+ RhiReconciliationScopePrerequisites::Satisfied,
+ )
+ .expect("final manifest");
+ assert_eq!(manifest.trade_generation(), 2);
+ let projection = reduce_rhi_reconciliation_manifest(manifest).expect("final projection");
+ let claim = *projection.root_mutation_id().expect("root claim");
+ let evaluation = rhi::evaluate_rhi_reconciliation_claim(projection, claim);
+ (
+ root,
+ runtime,
+ metadata,
+ configuration,
+ host,
+ lease,
+ evaluation,
+ )
+}
+
async fn attempt_fixture(
instance: &str,
) -> (
diff --git a/tests/services_hardening_reconciliation_manifest_contract.rs b/tests/services_hardening_reconciliation_manifest_contract.rs
@@ -82,6 +82,18 @@ fn machine_contract_freezes_the_complete_step_191_boundary() {
contract["private_reducer_material"]["outcome_owner"],
"reconciliation_outcome.v1.json"
);
+ assert_eq!(
+ contract["private_finalization_identity"]["attempt_id"],
+ "exact_step_190_committed_attempt_id"
+ );
+ assert_eq!(
+ contract["private_finalization_identity"]["job_id"],
+ "exact_step_190_claimed_job_id"
+ );
+ assert_eq!(
+ contract["private_finalization_identity"]["canonical_manifest_wire_changed"],
+ false
+ );
assert!(
!contract["deferred"]
.as_array()
@@ -120,6 +132,8 @@ fn manifest_boundary_is_sealed_canonical_redacted_and_effect_free() {
"i64::try_from(observed_at.get())",
"pub const fn shared_manifest_contract_id(&self)",
"pub const fn shared_manifest_contract_version(&self)",
+ "attempt_id: plan.id()",
+ "job_id: plan.job_id()",
] {
assert!(
SOURCE.contains(required),
