// file: crates/ksp-job-backfill-lib/src/persistence.rs // version: 3 /// 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: crate::BackfillEntityPersistence, observation: crate::BackfillObservationPersistence, } impl crate::BackfillPersistenceOutcome { /// Creates one internally classified persistence result. pub(crate) fn new( reference: ksp_store_lib::RawTransactionReference, entity: crate::BackfillEntityPersistence, observation: crate::BackfillObservationPersistence, ) -> Self { return Self { reference, entity, observation }; } /// 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) -> crate::BackfillEntityPersistence { return self.entity; } /// Returns the acquisition-observation persistence disposition. #[must_use] pub const fn observation(&self) -> crate::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 { 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>; } 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> { return ksp_store_lib::RawTransactionWrite::persist_raw_transaction_acquisition(self, transaction, observation, mode); } } async fn persist_hydration_with_port

(port: &P, hydration: crate::BackfillHydrationOutcome) -> ksp_core_lib::Result 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(crate::BackfillPersistenceOutcome::new( reference, crate::BackfillEntityPersistence::Missing, crate::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

( port: &P, reference: ksp_store_lib::RawTransactionReference, transaction: ksp_store_lib::RawTransaction, observation: ksp_store_lib::RawTransactionObservation, ) -> ksp_core_lib::Result 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(crate::BackfillPersistenceOutcome::new( reference, crate::BackfillEntityPersistence::Conflict, crate::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 { 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(crate::BackfillPersistenceOutcome::new( reference, crate::BackfillEntityPersistence::Inserted, crate::BackfillObservationPersistence::Inserted, )); } if entity == ksp_store_lib::RawEntityWriteOutcome::AlreadyPresent && observation == ksp_store_lib::RawObservationWriteOutcome::Inserted { return std::result::Result::Ok(crate::BackfillPersistenceOutcome::new( reference, crate::BackfillEntityPersistence::AlreadyPresent, crate::BackfillObservationPersistence::Inserted, )); } if entity == ksp_store_lib::RawEntityWriteOutcome::AlreadyPresent && observation == ksp_store_lib::RawObservationWriteOutcome::AlreadyPresent { return std::result::Result::Ok(crate::BackfillPersistenceOutcome::new( reference, crate::BackfillEntityPersistence::AlreadyPresent, crate::BackfillObservationPersistence::AlreadyPresent, )); } if entity == ksp_store_lib::RawEntityWriteOutcome::SkippedPurged && observation == ksp_store_lib::RawObservationWriteOutcome::NotRecorded { return std::result::Result::Ok(crate::BackfillPersistenceOutcome::new( reference, crate::BackfillEntityPersistence::SkippedPurged, crate::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;