// file: crates/ksp-worker-raw-transaction-ingest-lib/src/persistence.rs // version: 1 /// Canonical entity disposition produced by one Worker Store persistence attempt. #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub(crate) enum RawTransactionIngestEntityPersistence { /// The canonical RAW transaction was inserted for the first time. Inserted, /// Identical canonical RAW transaction content was already durable. AlreadyPresent, /// Normal persistence respected an existing purge tombstone. SkippedPurged, } /// Observation disposition produced by one Worker Store persistence attempt. #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub(crate) enum RawTransactionIngestObservationPersistence { /// The deterministic acquisition observation was inserted for the first time. Inserted, /// The same deterministic acquisition observation was already durable. AlreadyPresent, /// No observation was recorded because the canonical entity was intentionally skipped. NotRecorded, } /// Classified successful outcome of one atomic Worker Store persistence attempt. #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub(crate) struct RawTransactionIngestPersistenceOutcome { entity: crate::RawTransactionIngestEntityPersistence, observation: crate::RawTransactionIngestObservationPersistence, } impl crate::RawTransactionIngestPersistenceOutcome { /// Returns the canonical entity persistence disposition. #[must_use] pub(crate) const fn entity(self) -> crate::RawTransactionIngestEntityPersistence { return self.entity; } /// Returns the acquisition-observation persistence disposition. #[must_use] pub(crate) const fn observation(self) -> crate::RawTransactionIngestObservationPersistence { return self.observation; } } /// Private backend-neutral Store persistence port used by the Worker and deterministic tests. pub(crate) trait RawTransactionIngestPersistencePort: std::marker::Send + std::marker::Sync { /// Returns whether the persistence target belongs to the requested logical network. fn network_matches(&self, network: &ksp_store_lib::RawNetworkId) -> bool; /// Executes one atomic canonical RAW transaction plus observation Store write. 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 crate::RawTransactionIngestPersistencePort for ksp_store_lib::Store { fn network_matches(&self, network: &ksp_store_lib::RawNetworkId) -> bool { return self.runtime_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); } } /// Persists one already-canonical Worker acquisition through the private Store port in `Normal` mode. pub(crate) async fn persist_raw_transaction_ingest_acquisition

( port: &P, acquisition: ksp_raw_transaction_lib::RawTransactionAcquisition, ) -> ksp_core_lib::Result where P: crate::RawTransactionIngestPersistencePort + ?Sized, { let reference = acquisition.transaction().reference().clone(); if !port.network_matches(reference.network()) { return std::result::Result::Err(crate::runtime_error("persistence.store_network_mismatch")); } if acquisition.observation().transaction() != &reference { return std::result::Result::Err(crate::runtime_error("persistence.acquisition_reference_mismatch")); } let (transaction, observation) = acquisition.into_parts(); let result = port.persist_acquisition(transaction, observation, ksp_store_lib::RawTransactionAcquisitionMode::Normal).await; return match result { std::result::Result::Ok(outcome) => map_store_outcome(outcome), std::result::Result::Err(error) => map_store_error(error), }; } fn map_store_error(error: ksp_core_lib::Error) -> ksp_core_lib::Result { if error.code() == ksp_store_lib::ERROR_CODE_RAW_CONFLICT { return std::result::Result::Err(crate::content_conflict_error()); } return std::result::Result::Err(crate::store_error(error.code())); } fn map_store_outcome(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::RawTransactionIngestPersistenceOutcome { entity: crate::RawTransactionIngestEntityPersistence::Inserted, observation: crate::RawTransactionIngestObservationPersistence::Inserted, }); } if entity == ksp_store_lib::RawEntityWriteOutcome::AlreadyPresent && observation == ksp_store_lib::RawObservationWriteOutcome::Inserted { return std::result::Result::Ok(crate::RawTransactionIngestPersistenceOutcome { entity: crate::RawTransactionIngestEntityPersistence::AlreadyPresent, observation: crate::RawTransactionIngestObservationPersistence::Inserted, }); } if entity == ksp_store_lib::RawEntityWriteOutcome::AlreadyPresent && observation == ksp_store_lib::RawObservationWriteOutcome::AlreadyPresent { return std::result::Result::Ok(crate::RawTransactionIngestPersistenceOutcome { entity: crate::RawTransactionIngestEntityPersistence::AlreadyPresent, observation: crate::RawTransactionIngestObservationPersistence::AlreadyPresent, }); } if entity == ksp_store_lib::RawEntityWriteOutcome::SkippedPurged && observation == ksp_store_lib::RawObservationWriteOutcome::NotRecorded { return std::result::Result::Ok(crate::RawTransactionIngestPersistenceOutcome { entity: crate::RawTransactionIngestEntityPersistence::SkippedPurged, observation: crate::RawTransactionIngestObservationPersistence::NotRecorded, }); } return std::result::Result::Err(crate::runtime_error("persistence.store_outcome_invalid")); } #[cfg(test)] #[path = "../unit_tests/persistence.rs"] mod tests;