v0.3.6-pre.007

crates/ksp-job-backfill-lib/src/error.rs

@@ -1,10 +1,12 @@
 // file: crates/ksp-job-backfill-lib/src/error.rs
-// version: 2
+// version: 3
 
 /// Error code used when a signature page violates a bounded discovery invariant.
 pub const ERROR_CODE_BACKFILL_DISCOVERY_INVALID: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("job_backfill", "discovery_invalid");
 /// Error code used when paginated discovery cannot advance its exclusive RPC cursor safely.
 pub const ERROR_CODE_BACKFILL_DISCOVERY_STALLED: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("job_backfill", "discovery_stalled");
+/// Error code used when Store persistence returns an impossible Backfill state or targets a different network.
+pub const ERROR_CODE_BACKFILL_PERSISTENCE_INVALID: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("job_backfill", "persistence_invalid");
 /// Error code used when deterministic Transport-to-RAW conversion violates the v1 contract.
 pub const ERROR_CODE_BACKFILL_RAW_CONVERSION_INVALID: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("job_backfill", "raw_conversion_invalid");
 /// Error code used when one Backfill request violates its bounded admission contract.

crates/ksp-job-backfill-lib/src/lib.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-job-backfill-lib/src/lib.rs
-// version: 2
+// version: 3
 
 #![warn(missing_docs)]
 #![deny(unreachable_pub)]
@@ -10,13 +10,14 @@
 //! This tranche owns explicit admission, network-scoped candidate identity, deterministic
 //! `getSignaturesForAddress` pagination and canonical RAW v1 conversion through observed
 //! `getTransaction`. Transport retains provider/endpoint selection and retry; Store retains
-//! durable idempotence and persistence. Persistence, concurrency, checkpointing, cancellation
+//! durable idempotence through the atomic Store facade. Concurrency, checkpointing, cancellation
 //! and concrete latest-value snapshots are added by later v0.3.6 tranches.
 
 mod constants;
 mod conversion;
 mod discovery;
 mod error;
+mod persistence;
 mod request;
 
 /// Result of hydrating one deterministic candidate through observed `getTransaction`.
@@ -43,12 +44,22 @@ pub use self::discovery::discover_backfill_candidates;
 pub use self::error::ERROR_CODE_BACKFILL_DISCOVERY_INVALID;
 /// Error code used when paginated discovery cannot advance its exclusive RPC cursor safely.
 pub use self::error::ERROR_CODE_BACKFILL_DISCOVERY_STALLED;
+/// Error code used when Store persistence returns an impossible Backfill state or targets a different network.
+pub use self::error::ERROR_CODE_BACKFILL_PERSISTENCE_INVALID;
 /// Error code used when deterministic Transport-to-RAW conversion violates the v1 contract.
 pub use self::error::ERROR_CODE_BACKFILL_RAW_CONVERSION_INVALID;
 /// Error code used when one Backfill request violates its bounded admission contract.
 pub use self::error::ERROR_CODE_BACKFILL_REQUEST_INVALID;
 /// Error code used when one transaction signature text violates the bounded Base58-shape contract.
 pub use self::error::ERROR_CODE_BACKFILL_SIGNATURE_INVALID;
+/// Canonical entity disposition produced by one Backfill Store persistence attempt.
+pub use self::persistence::BackfillEntityPersistence;
+/// Observation disposition produced by one Backfill Store persistence attempt.
+pub use self::persistence::BackfillObservationPersistence;
+/// Stable Backfill projection of one hydration persistence result.
+pub use self::persistence::BackfillPersistenceOutcome;
+/// Persists one hydrated result through the backend-neutral atomic Store contract.
+pub use self::persistence::persist_backfill_hydration;
 /// Commitment levels intentionally admitted by the historical Backfill vertical.
 pub use self::request::BackfillCommitment;
 /// Fully explicit bounded request for one historical transaction Backfill Job.

crates/ksp-job-backfill-lib/src/persistence.rs

@@ -0,0 +1,201 @@
+// file: crates/ksp-job-backfill-lib/src/persistence.rs
+// version: 1
+
+/// Canonical entity disposition produced by one Backfill Store persistence attempt.
+#[non_exhaustive]
+#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
+pub enum BackfillEntityPersistence {
+    /// The canonical RAW transaction was inserted for the first time.
+    Inserted,
+    /// Identical canonical RAW transaction content was already durable.
+    AlreadyPresent,
+    /// A durable purge tombstone prevented normal Backfill rehydration.
+    SkippedPurged,
+    /// `getTransaction` returned JSON `null`, so no canonical Store write was attempted.
+    Missing,
+    /// Store reported divergent canonical content for the same network-scoped transaction identity.
+    Conflict,
+}
+
+/// Observation disposition produced by one Backfill Store persistence attempt.
+#[non_exhaustive]
+#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
+pub enum BackfillObservationPersistence {
+    /// The deterministic acquisition observation was inserted for the first time.
+    Inserted,
+    /// The same deterministic acquisition observation was already durable.
+    AlreadyPresent,
+    /// Store intentionally recorded no observation because normal persistence skipped a purged entity or detected a conflict.
+    NotRecorded,
+    /// No observation existed because hydration returned `Missing` before persistence.
+    NotApplicable,
+}
+
+/// Stable Backfill projection of one hydration persistence result.
+///
+/// The canonical transaction identity remains `(network, signature)`. Provider, endpoint and
+/// transport details can affect the persisted observation but never the transaction identity.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct BackfillPersistenceOutcome {
+    reference: ksp_store_lib::RawTransactionReference,
+    entity: BackfillEntityPersistence,
+    observation: BackfillObservationPersistence,
+}
+
+impl BackfillPersistenceOutcome {
+    /// Returns the network-scoped canonical transaction identity classified by this result.
+    #[must_use]
+    pub fn reference(&self) -> &ksp_store_lib::RawTransactionReference {
+        return &self.reference;
+    }
+
+    /// Returns the canonical transaction persistence disposition.
+    #[must_use]
+    pub const fn entity(&self) -> BackfillEntityPersistence {
+        return self.entity;
+    }
+
+    /// Returns the acquisition-observation persistence disposition.
+    #[must_use]
+    pub const fn observation(&self) -> BackfillObservationPersistence {
+        return self.observation;
+    }
+}
+
+/// Persists one hydrated Backfill result through the backend-neutral Store facade.
+///
+/// Available acquisitions use the existing atomic transaction-plus-observation Store contract in
+/// `Normal` mode. A durable purge is therefore respected and never rehydrated implicitly.
+/// `Missing` performs no Store write. Stable Store content conflicts are projected explicitly as
+/// [`BackfillEntityPersistence::Conflict`] rather than being silently treated as idempotent skips.
+/// Other Store failures remain errors.
+pub async fn persist_backfill_hydration(
+    store: &ksp_store_lib::Store,
+    hydration: crate::BackfillHydrationOutcome,
+) -> ksp_core_lib::Result<BackfillPersistenceOutcome> {
+    return persist_hydration_with_port(store, hydration).await;
+}
+
+trait RawTransactionPersistencePort: std::marker::Send + std::marker::Sync {
+    fn network_matches(&self, network: &ksp_store_lib::RawNetworkId) -> bool;
+
+    fn persist_acquisition<'a>(
+        &'a self,
+        transaction: ksp_store_lib::RawTransaction,
+        observation: ksp_store_lib::RawTransactionObservation,
+        mode: ksp_store_lib::RawTransactionAcquisitionMode,
+    ) -> ksp_store_lib::StoreApiFuture<'a, ksp_store_lib::Result<ksp_store_lib::RawAcquisitionWriteOutcome>>;
+}
+
+impl RawTransactionPersistencePort for ksp_store_lib::Store {
+    fn network_matches(&self, network: &ksp_store_lib::RawNetworkId) -> bool {
+        let snapshot = self.runtime_snapshot();
+        return snapshot.network() == network;
+    }
+
+    fn persist_acquisition<'a>(
+        &'a self,
+        transaction: ksp_store_lib::RawTransaction,
+        observation: ksp_store_lib::RawTransactionObservation,
+        mode: ksp_store_lib::RawTransactionAcquisitionMode,
+    ) -> ksp_store_lib::StoreApiFuture<'a, ksp_store_lib::Result<ksp_store_lib::RawAcquisitionWriteOutcome>> {
+        return ksp_store_lib::RawTransactionWrite::persist_raw_transaction_acquisition(self, transaction, observation, mode);
+    }
+}
+
+async fn persist_hydration_with_port<P>(port: &P, hydration: crate::BackfillHydrationOutcome) -> ksp_core_lib::Result<BackfillPersistenceOutcome>
+where
+    P: RawTransactionPersistencePort,
+{
+    let reference = hydration.reference().clone();
+    if !port.network_matches(reference.network()) {
+        return std::result::Result::Err(persistence_error("store.network"));
+    }
+    return match hydration {
+        crate::BackfillHydrationOutcome::Missing(_) => std::result::Result::Ok(BackfillPersistenceOutcome {
+            reference,
+            entity: BackfillEntityPersistence::Missing,
+            observation: BackfillObservationPersistence::NotApplicable,
+        }),
+        crate::BackfillHydrationOutcome::Available(acquisition) => {
+            let (transaction, observation) = acquisition.into_parts();
+            persist_available_with_port(port, reference, transaction, observation).await
+        },
+    };
+}
+
+async fn persist_available_with_port<P>(
+    port: &P,
+    reference: ksp_store_lib::RawTransactionReference,
+    transaction: ksp_store_lib::RawTransaction,
+    observation: ksp_store_lib::RawTransactionObservation,
+) -> ksp_core_lib::Result<BackfillPersistenceOutcome>
+where
+    P: RawTransactionPersistencePort,
+{
+    if !port.network_matches(reference.network()) {
+        return std::result::Result::Err(persistence_error("store.network"));
+    }
+    if transaction.reference() != &reference || observation.transaction() != &reference {
+        return std::result::Result::Err(persistence_error("acquisition.reference"));
+    }
+    let result = port.persist_acquisition(transaction, observation, ksp_store_lib::RawTransactionAcquisitionMode::Normal).await;
+    return match result {
+        std::result::Result::Ok(outcome) => map_store_outcome(reference, outcome),
+        std::result::Result::Err(error) => {
+            if error.code() == ksp_store_lib::ERROR_CODE_RAW_CONFLICT {
+                return std::result::Result::Ok(BackfillPersistenceOutcome {
+                    reference,
+                    entity: BackfillEntityPersistence::Conflict,
+                    observation: BackfillObservationPersistence::NotRecorded,
+                });
+            }
+            std::result::Result::Err(error)
+        },
+    };
+}
+
+fn map_store_outcome(
+    reference: ksp_store_lib::RawTransactionReference,
+    outcome: ksp_store_lib::RawAcquisitionWriteOutcome,
+) -> ksp_core_lib::Result<BackfillPersistenceOutcome> {
+    let entity = outcome.entity();
+    let observation = outcome.observation();
+    if entity == ksp_store_lib::RawEntityWriteOutcome::Inserted && observation == ksp_store_lib::RawObservationWriteOutcome::Inserted {
+        return std::result::Result::Ok(BackfillPersistenceOutcome {
+            reference,
+            entity: BackfillEntityPersistence::Inserted,
+            observation: BackfillObservationPersistence::Inserted,
+        });
+    }
+    if entity == ksp_store_lib::RawEntityWriteOutcome::AlreadyPresent && observation == ksp_store_lib::RawObservationWriteOutcome::Inserted {
+        return std::result::Result::Ok(BackfillPersistenceOutcome {
+            reference,
+            entity: BackfillEntityPersistence::AlreadyPresent,
+            observation: BackfillObservationPersistence::Inserted,
+        });
+    }
+    if entity == ksp_store_lib::RawEntityWriteOutcome::AlreadyPresent && observation == ksp_store_lib::RawObservationWriteOutcome::AlreadyPresent {
+        return std::result::Result::Ok(BackfillPersistenceOutcome {
+            reference,
+            entity: BackfillEntityPersistence::AlreadyPresent,
+            observation: BackfillObservationPersistence::AlreadyPresent,
+        });
+    }
+    if entity == ksp_store_lib::RawEntityWriteOutcome::SkippedPurged && observation == ksp_store_lib::RawObservationWriteOutcome::NotRecorded {
+        return std::result::Result::Ok(BackfillPersistenceOutcome {
+            reference,
+            entity: BackfillEntityPersistence::SkippedPurged,
+            observation: BackfillObservationPersistence::NotRecorded,
+        });
+    }
+    return std::result::Result::Err(persistence_error("store.outcome"));
+}
+
+fn persistence_error(field: &'static str) -> ksp_core_lib::Error {
+    return ksp_core_lib::Error::new(crate::ERROR_CODE_BACKFILL_PERSISTENCE_INVALID, "invalid Backfill Store persistence state").with_context("field", field);
+}
+
+#[cfg(test)]
+#[path = "../unit_tests/persistence.rs"]
+mod tests;

crates/ksp-job-backfill-lib/tests/dependency_boundary.rs

@@ -1,10 +1,10 @@
 // file: crates/ksp-job-backfill-lib/tests/dependency_boundary.rs
-// version: 2
+// version: 3
 
-//! Dependency firewall canaries for Backfill discovery and RAW v1 conversion.
+//! Dependency firewall canaries through Backfill Store persistence.
 
 #[test]
-fn pre_006_manifest_uses_only_planned_ksp_edges_and_backend_neutral_store() {
+fn pre_007_manifest_uses_only_planned_ksp_edges_and_backend_neutral_store() {
     let manifest = include_str!("../Cargo.toml");
     for required in [
         "ksp-core-lib = { path = \"../ksp-core-lib\" }",
@@ -36,7 +36,7 @@ fn pre_006_manifest_uses_only_planned_ksp_edges_and_backend_neutral_store() {
 }
 
 #[test]
-fn pre_006_production_sources_keep_transport_and_store_in_their_owned_layers() {
+fn pre_007_production_sources_keep_transport_and_store_in_their_owned_layers() {
     let non_conversion_sources = [
         include_str!("../src/constants.rs"),
         include_str!("../src/discovery.rs"),
@@ -62,6 +62,15 @@ fn pre_006_production_sources_keep_transport_and_store_in_their_owned_layers() {
     for forbidden in ["ksp_config_lib::", "ksp_interface_lib::", "ksp_store_api::", "ksp_store_postgres_lib::", "reqwest::", "std::env", "tonic::"] {
         assert!(!conversion.contains(forbidden), "forbidden RAW conversion path detected: {forbidden}");
     }
+    let persistence = include_str!("../src/persistence.rs");
+    assert!(persistence.contains("persist_raw_transaction_acquisition"));
+    assert!(persistence.contains("RawTransactionAcquisitionMode::Normal"));
+    assert!(persistence.contains("ERROR_CODE_RAW_CONFLICT"));
+    assert!(!persistence.contains("record_raw_transaction_observation"));
+    assert!(!persistence.contains("RawTransactionAcquisitionMode::ForceRehydrate"));
+    for forbidden in ["ksp_config_lib::", "ksp_interface_lib::", "ksp_store_api::", "ksp_store_postgres_lib::", "reqwest::", "std::env", "tonic::"] {
+        assert!(!persistence.contains(forbidden), "forbidden Store persistence path detected: {forbidden}");
+    }
     let discovery = include_str!("../src/discovery.rs");
     assert!(discovery.contains("get_signatures_for_address"));
     assert!(!discovery.contains("execute_standard_rpc"));

crates/ksp-job-backfill-lib/tests/public_api.rs

@@ -1,7 +1,7 @@
 // file: crates/ksp-job-backfill-lib/tests/public_api.rs
-// version: 2
+// version: 3
 
-//! Public API canaries for bounded Backfill discovery and RAW v1 conversion.
+//! Public API canaries for bounded Backfill discovery, RAW v1 conversion and Store persistence.
 
 #[test]
 fn pre_005_request_scope_and_discovery_contracts_are_available_from_crate_root() {
@@ -86,3 +86,13 @@ fn pre_006_raw_conversion_contract_is_available_from_crate_root() {
     assert_eq!(ksp_job_backfill_lib::ERROR_CODE_BACKFILL_RAW_CONVERSION_INVALID, ksp_core_lib::ErrorCode::new("job_backfill", "raw_conversion_invalid"));
     return;
 }
+
+#[test]
+fn pre_007_store_persistence_contract_is_available_from_crate_root() {
+    let _persist = ksp_job_backfill_lib::persist_backfill_hydration;
+    let _outcome: std::option::Option<ksp_job_backfill_lib::BackfillPersistenceOutcome> = std::option::Option::None;
+    let _entity = ksp_job_backfill_lib::BackfillEntityPersistence::AlreadyPresent;
+    let _observation = ksp_job_backfill_lib::BackfillObservationPersistence::AlreadyPresent;
+    assert_eq!(ksp_job_backfill_lib::ERROR_CODE_BACKFILL_PERSISTENCE_INVALID, ksp_core_lib::ErrorCode::new("job_backfill", "persistence_invalid"));
+    return;
+}

crates/ksp-job-backfill-lib/tests/release_completeness.rs

@@ -1,10 +1,10 @@
 // file: crates/ksp-job-backfill-lib/tests/release_completeness.rs
-// version: 2
+// version: 3
 
-//! Completeness canaries for the `pre.006` Backfill RAW conversion tranche.
+//! Completeness canaries through the `pre.007` Backfill Store persistence tranche.
 
 #[test]
-fn pre_006_production_module_inventory_is_exact() -> std::io::Result<()> {
+fn pre_007_production_module_inventory_is_exact() -> std::io::Result<()> {
     let source_root = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("src");
     let entries = match std::fs::read_dir(source_root) {
         std::result::Result::Ok(value) => value,
@@ -32,12 +32,12 @@ fn pre_006_production_module_inventory_is_exact() -> std::io::Result<()> {
         }
     }
     names.sort_unstable();
-    assert_eq!(names, std::vec!["constants.rs", "conversion.rs", "discovery.rs", "error.rs", "lib.rs", "request.rs"]);
+    assert_eq!(names, std::vec!["constants.rs", "conversion.rs", "discovery.rs", "error.rs", "lib.rs", "persistence.rs", "request.rs"]);
     return std::result::Result::Ok(());
 }
 
 #[test]
-fn pre_006_surface_adds_raw_conversion_without_persistence_or_checkpoint_runtime() {
+fn pre_007_surface_adds_store_persistence_without_checkpoint_runtime() {
     let root = include_str!("../src/lib.rs");
     for required in [
         "BackfillCandidate",
@@ -56,11 +56,16 @@ fn pre_006_surface_adds_raw_conversion_without_persistence_or_checkpoint_runtime
         "RAW_TRANSACTION_FORMAT_ID",
         "RAW_TRANSACTION_FORMAT_VERSION",
         "ERROR_CODE_BACKFILL_RAW_CONVERSION_INVALID",
+        "BackfillPersistenceOutcome",
+        "BackfillEntityPersistence",
+        "BackfillObservationPersistence",
+        "persist_backfill_hydration",
+        "ERROR_CODE_BACKFILL_PERSISTENCE_INVALID",
     ] {
         assert!(root.contains(required), "required pre.006 public contract missing: {required}");
     }
-    for forbidden in ["persist_raw_transaction_acquisition", "BackfillCheckpoint", "BackfillJobHandle", "JobSnapshotSource"] {
-        assert!(!root.contains(forbidden), "later Backfill tranche leaked into pre.006: {forbidden}");
+    for forbidden in ["BackfillCheckpoint", "BackfillJobHandle", "JobSnapshotSource"] {
+        assert!(!root.contains(forbidden), "later Backfill tranche leaked into pre.007: {forbidden}");
     }
     assert!(!root.contains("pub mod "));
     return;

crates/ksp-job-backfill-lib/unit_tests/persistence.rs

@@ -0,0 +1,348 @@
+// file: crates/ksp-job-backfill-lib/unit_tests/persistence.rs
+// version: 1
+
+#[derive(Clone, Copy)]
+enum FakeResponse {
+    Outcome(ksp_store_lib::RawAcquisitionWriteOutcome),
+    Conflict,
+    Failure,
+}
+
+struct FakePersistencePort {
+    network: ksp_store_lib::RawNetworkId,
+    responses: std::sync::Mutex<std::collections::VecDeque<FakeResponse>>,
+    calls: std::sync::atomic::AtomicUsize,
+    normal_mode_only: std::sync::atomic::AtomicBool,
+}
+
+impl FakePersistencePort {
+    fn new(network: ksp_store_lib::RawNetworkId, responses: &[FakeResponse]) -> Self {
+        return Self {
+            network,
+            responses: std::sync::Mutex::new(responses.iter().copied().collect()),
+            calls: std::sync::atomic::AtomicUsize::new(0),
+            normal_mode_only: std::sync::atomic::AtomicBool::new(true),
+        };
+    }
+
+    fn calls(&self) -> usize {
+        return self.calls.load(std::sync::atomic::Ordering::SeqCst);
+    }
+
+    fn used_only_normal_mode(&self) -> bool {
+        return self.normal_mode_only.load(std::sync::atomic::Ordering::SeqCst);
+    }
+}
+
+impl super::RawTransactionPersistencePort for FakePersistencePort {
+    fn network_matches(&self, network: &ksp_store_lib::RawNetworkId) -> bool {
+        return &self.network == network;
+    }
+
+    fn persist_acquisition<'a>(
+        &'a self,
+        transaction: ksp_store_lib::RawTransaction,
+        observation: ksp_store_lib::RawTransactionObservation,
+        mode: ksp_store_lib::RawTransactionAcquisitionMode,
+    ) -> ksp_store_lib::StoreApiFuture<'a, ksp_store_lib::Result<ksp_store_lib::RawAcquisitionWriteOutcome>> {
+        self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+        if mode != ksp_store_lib::RawTransactionAcquisitionMode::Normal {
+            self.normal_mode_only.store(false, std::sync::atomic::Ordering::SeqCst);
+        }
+        let result = if transaction.reference() != observation.transaction() {
+            std::result::Result::Err(ksp_core_lib::Error::new(
+                ksp_core_lib::ErrorCode::new("test", "reference_mismatch"),
+                "fake persistence reference mismatch",
+            ))
+        } else {
+            let response = match self.responses.lock() {
+                std::result::Result::Ok(mut responses) => responses.pop_front(),
+                std::result::Result::Err(_) => std::option::Option::None,
+            };
+            match response {
+                std::option::Option::Some(FakeResponse::Outcome(outcome)) => std::result::Result::Ok(outcome),
+                std::option::Option::Some(FakeResponse::Conflict) => {
+                    std::result::Result::Err(ksp_core_lib::Error::new(ksp_store_lib::ERROR_CODE_RAW_CONFLICT, "fake canonical content conflict"))
+                },
+                std::option::Option::Some(FakeResponse::Failure) | std::option::Option::None => {
+                    std::result::Result::Err(ksp_core_lib::Error::new(ksp_core_lib::ErrorCode::new("test", "store_failure"), "fake Store failure"))
+                },
+            }
+        };
+        return std::boxed::Box::pin(async move {
+            return result;
+        });
+    }
+}
+
+fn raw_network(value: &str) -> std::option::Option<ksp_store_lib::RawNetworkId> {
+    return match ksp_store_lib::RawNetworkId::new(value) {
+        std::result::Result::Ok(network) => std::option::Option::Some(network),
+        std::result::Result::Err(_) => std::option::Option::None,
+    };
+}
+
+fn raw_reference(network: &str, signature_byte: u8) -> std::option::Option<ksp_store_lib::RawTransactionReference> {
+    let network = match raw_network(network) {
+        std::option::Option::Some(network) => network,
+        std::option::Option::None => return std::option::Option::None,
+    };
+    return std::option::Option::Some(ksp_store_lib::RawTransactionReference::new(network, ksp_store_lib::RawTransactionSignature::new([signature_byte; 64])));
+}
+
+fn raw_acquisition_parts(
+    network: &str,
+    signature_byte: u8,
+    observation_byte: u8,
+) -> std::option::Option<(ksp_store_lib::RawTransactionReference, ksp_store_lib::RawTransaction, ksp_store_lib::RawTransactionObservation)> {
+    let reference = match raw_reference(network, signature_byte) {
+        std::option::Option::Some(reference) => reference,
+        std::option::Option::None => return std::option::Option::None,
+    };
+    let format_id = match ksp_store_lib::RawFormatId::new(crate::RAW_TRANSACTION_FORMAT_ID) {
+        std::result::Result::Ok(format_id) => format_id,
+        std::result::Result::Err(_) => return std::option::Option::None,
+    };
+    let payload = ksp_store_lib::RawPayload::try_new(
+        format_id,
+        crate::RAW_TRANSACTION_FORMAT_VERSION,
+        std::vec![signature_byte].into_boxed_slice(),
+        ksp_store_lib::RawContentHash::new([signature_byte; 32]),
+    );
+    let payload = match payload {
+        std::result::Result::Ok(payload) => payload,
+        std::result::Result::Err(_) => return std::option::Option::None,
+    };
+    let received_at = match ksp_store_lib::RawTimestamp::from_unix_millis(1_700_000_000_000) {
+        std::result::Result::Ok(received_at) => received_at,
+        std::result::Result::Err(_) => return std::option::Option::None,
+    };
+    let provider = match ksp_store_lib::RawProvenanceCode::new("provider") {
+        std::result::Result::Ok(provider) => provider,
+        std::result::Result::Err(_) => return std::option::Option::None,
+    };
+    let protocol = match ksp_store_lib::RawProvenanceCode::new("solana.http.json_rpc") {
+        std::result::Result::Ok(protocol) => protocol,
+        std::result::Result::Err(_) => return std::option::Option::None,
+    };
+    let method = match ksp_store_lib::RawProvenanceCode::new("getTransaction") {
+        std::result::Result::Ok(method) => method,
+        std::result::Result::Err(_) => return std::option::Option::None,
+    };
+    let provenance = ksp_store_lib::RawAcquisitionProvenance::new(provider, protocol, method, ksp_store_lib::RawAcquisitionOrigin::Backfill, received_at);
+    let transaction = ksp_store_lib::RawTransaction::new(reference.clone(), 42, std::option::Option::None, payload);
+    let observation =
+        ksp_store_lib::RawTransactionObservation::new(ksp_store_lib::RawObservationKey::new([observation_byte; 32]), reference.clone(), provenance);
+    return std::option::Option::Some((reference, transaction, observation));
+}
+
+fn store_outcome(
+    entity: ksp_store_lib::RawEntityWriteOutcome,
+    observation: ksp_store_lib::RawObservationWriteOutcome,
+) -> ksp_store_lib::RawAcquisitionWriteOutcome {
+    return ksp_store_lib::RawAcquisitionWriteOutcome::new(entity, observation);
+}
+
+#[tokio::test]
+async fn pre_007_missing_skips_store_and_preserves_network_scoped_identity() {
+    let reference = match raw_reference("devnet", 1) {
+        std::option::Option::Some(reference) => reference,
+        std::option::Option::None => return,
+    };
+    let network = match raw_network("devnet") {
+        std::option::Option::Some(network) => network,
+        std::option::Option::None => return,
+    };
+    let port = FakePersistencePort::new(network, &[]);
+    let result = super::persist_hydration_with_port(&port, crate::BackfillHydrationOutcome::Missing(reference.clone())).await;
+    assert!(result.is_ok());
+    if let std::result::Result::Ok(result) = result {
+        assert_eq!(result.reference(), &reference);
+        assert_eq!(result.entity(), crate::BackfillEntityPersistence::Missing);
+        assert_eq!(result.observation(), crate::BackfillObservationPersistence::NotApplicable);
+    }
+    assert_eq!(port.calls(), 0);
+    return;
+}
+
+#[tokio::test]
+async fn pre_007_store_network_mismatch_is_rejected_before_any_write() {
+    let reference = match raw_reference("devnet", 2) {
+        std::option::Option::Some(reference) => reference,
+        std::option::Option::None => return,
+    };
+    let network = match raw_network("mainnet") {
+        std::option::Option::Some(network) => network,
+        std::option::Option::None => return,
+    };
+    let port = FakePersistencePort::new(network, &[]);
+    let result = super::persist_hydration_with_port(&port, crate::BackfillHydrationOutcome::Missing(reference)).await;
+    assert!(result.is_err());
+    if let std::result::Result::Err(error) = result {
+        assert_eq!(error.code(), crate::ERROR_CODE_BACKFILL_PERSISTENCE_INVALID);
+    }
+    assert_eq!(port.calls(), 0);
+    return;
+}
+
+#[tokio::test]
+async fn pre_007_atomic_insert_maps_entity_and_observation_without_second_write() {
+    let (reference, transaction, observation) = match raw_acquisition_parts("devnet", 3, 13) {
+        std::option::Option::Some(parts) => parts,
+        std::option::Option::None => return,
+    };
+    let network = match raw_network("devnet") {
+        std::option::Option::Some(network) => network,
+        std::option::Option::None => return,
+    };
+    let response = store_outcome(ksp_store_lib::RawEntityWriteOutcome::Inserted, ksp_store_lib::RawObservationWriteOutcome::Inserted);
+    let port = FakePersistencePort::new(network, &[FakeResponse::Outcome(response)]);
+    let result = super::persist_available_with_port(&port, reference.clone(), transaction, observation).await;
+    assert!(result.is_ok());
+    if let std::result::Result::Ok(result) = result {
+        assert_eq!(result.reference(), &reference);
+        assert_eq!(result.entity(), crate::BackfillEntityPersistence::Inserted);
+        assert_eq!(result.observation(), crate::BackfillObservationPersistence::Inserted);
+    }
+    assert_eq!(port.calls(), 1);
+    assert!(port.used_only_normal_mode());
+    return;
+}
+
+#[tokio::test]
+async fn pre_007_existing_entity_distinguishes_new_from_idempotent_observation() {
+    let network = match raw_network("devnet") {
+        std::option::Option::Some(network) => network,
+        std::option::Option::None => return,
+    };
+    let first = store_outcome(ksp_store_lib::RawEntityWriteOutcome::AlreadyPresent, ksp_store_lib::RawObservationWriteOutcome::Inserted);
+    let second = store_outcome(ksp_store_lib::RawEntityWriteOutcome::AlreadyPresent, ksp_store_lib::RawObservationWriteOutcome::AlreadyPresent);
+    let port = FakePersistencePort::new(network, &[FakeResponse::Outcome(first), FakeResponse::Outcome(second)]);
+    let (reference, transaction, observation) = match raw_acquisition_parts("devnet", 4, 14) {
+        std::option::Option::Some(parts) => parts,
+        std::option::Option::None => return,
+    };
+    let result = super::persist_available_with_port(&port, reference, transaction, observation).await;
+    assert!(result.is_ok());
+    if let std::result::Result::Ok(result) = result {
+        assert_eq!(result.entity(), crate::BackfillEntityPersistence::AlreadyPresent);
+        assert_eq!(result.observation(), crate::BackfillObservationPersistence::Inserted);
+    }
+    let (reference, transaction, observation) = match raw_acquisition_parts("devnet", 4, 14) {
+        std::option::Option::Some(parts) => parts,
+        std::option::Option::None => return,
+    };
+    let rerun = super::persist_available_with_port(&port, reference, transaction, observation).await;
+    assert!(rerun.is_ok());
+    if let std::result::Result::Ok(rerun) = rerun {
+        assert_eq!(rerun.entity(), crate::BackfillEntityPersistence::AlreadyPresent);
+        assert_eq!(rerun.observation(), crate::BackfillObservationPersistence::AlreadyPresent);
+    }
+    assert_eq!(port.calls(), 2);
+    assert!(port.used_only_normal_mode());
+    return;
+}
+
+#[tokio::test]
+async fn pre_007_normal_backfill_respects_purged_tombstone_without_observation() {
+    let (reference, transaction, observation) = match raw_acquisition_parts("devnet", 5, 15) {
+        std::option::Option::Some(parts) => parts,
+        std::option::Option::None => return,
+    };
+    let network = match raw_network("devnet") {
+        std::option::Option::Some(network) => network,
+        std::option::Option::None => return,
+    };
+    let response = store_outcome(ksp_store_lib::RawEntityWriteOutcome::SkippedPurged, ksp_store_lib::RawObservationWriteOutcome::NotRecorded);
+    let port = FakePersistencePort::new(network, &[FakeResponse::Outcome(response)]);
+    let result = super::persist_available_with_port(&port, reference, transaction, observation).await;
+    assert!(result.is_ok());
+    if let std::result::Result::Ok(result) = result {
+        assert_eq!(result.entity(), crate::BackfillEntityPersistence::SkippedPurged);
+        assert_eq!(result.observation(), crate::BackfillObservationPersistence::NotRecorded);
+    }
+    assert!(port.used_only_normal_mode());
+    return;
+}
+
+#[tokio::test]
+async fn pre_007_store_content_conflict_is_explicit_and_not_idempotent_success() {
+    let (reference, transaction, observation) = match raw_acquisition_parts("devnet", 6, 16) {
+        std::option::Option::Some(parts) => parts,
+        std::option::Option::None => return,
+    };
+    let network = match raw_network("devnet") {
+        std::option::Option::Some(network) => network,
+        std::option::Option::None => return,
+    };
+    let port = FakePersistencePort::new(network, &[FakeResponse::Conflict]);
+    let result = super::persist_available_with_port(&port, reference, transaction, observation).await;
+    assert!(result.is_ok());
+    if let std::result::Result::Ok(result) = result {
+        assert_eq!(result.entity(), crate::BackfillEntityPersistence::Conflict);
+        assert_eq!(result.observation(), crate::BackfillObservationPersistence::NotRecorded);
+        assert_ne!(result.entity(), crate::BackfillEntityPersistence::AlreadyPresent);
+    }
+    return;
+}
+
+#[tokio::test]
+async fn pre_007_non_conflict_store_failure_propagates_unchanged() {
+    let (reference, transaction, observation) = match raw_acquisition_parts("devnet", 7, 17) {
+        std::option::Option::Some(parts) => parts,
+        std::option::Option::None => return,
+    };
+    let network = match raw_network("devnet") {
+        std::option::Option::Some(network) => network,
+        std::option::Option::None => return,
+    };
+    let port = FakePersistencePort::new(network, &[FakeResponse::Failure]);
+    let result = super::persist_available_with_port(&port, reference, transaction, observation).await;
+    assert!(result.is_err());
+    if let std::result::Result::Err(error) = result {
+        assert_eq!(error.code(), ksp_core_lib::ErrorCode::new("test", "store_failure"));
+    }
+    return;
+}
+
+#[test]
+fn pre_007_normal_mode_rejects_impossible_store_outcome_combinations() {
+    let reference = match raw_reference("devnet", 8) {
+        std::option::Option::Some(reference) => reference,
+        std::option::Option::None => return,
+    };
+    let impossible = [
+        store_outcome(ksp_store_lib::RawEntityWriteOutcome::Rehydrated, ksp_store_lib::RawObservationWriteOutcome::Inserted),
+        store_outcome(ksp_store_lib::RawEntityWriteOutcome::Inserted, ksp_store_lib::RawObservationWriteOutcome::AlreadyPresent),
+        store_outcome(ksp_store_lib::RawEntityWriteOutcome::SkippedPurged, ksp_store_lib::RawObservationWriteOutcome::Inserted),
+    ];
+    for outcome in impossible {
+        let result = super::map_store_outcome(reference.clone(), outcome);
+        assert!(result.is_err());
+        if let std::result::Result::Err(error) = result {
+            assert_eq!(error.code(), crate::ERROR_CODE_BACKFILL_PERSISTENCE_INVALID);
+        }
+    }
+    return;
+}
+
+#[tokio::test]
+async fn pre_007_mismatched_transaction_observation_reference_is_rejected_before_store() {
+    let (reference, transaction, _) = match raw_acquisition_parts("devnet", 9, 19) {
+        std::option::Option::Some(parts) => parts,
+        std::option::Option::None => return,
+    };
+    let (_, _, observation) = match raw_acquisition_parts("devnet", 10, 20) {
+        std::option::Option::Some(parts) => parts,
+        std::option::Option::None => return,
+    };
+    let network = match raw_network("devnet") {
+        std::option::Option::Some(network) => network,
+        std::option::Option::None => return,
+    };
+    let port = FakePersistencePort::new(network, &[]);
+    let result = super::persist_available_with_port(&port, reference, transaction, observation).await;
+    assert!(result.is_err());
+    assert_eq!(port.calls(), 0);
+    return;
+}