v0.3.4-pre.005

crates/ksp-store-postgres-lib/src/lib.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-store-postgres-lib/src/lib.rs
-// version: 17
+// version: 18
 
 #![warn(missing_docs)]
 #![deny(unreachable_pub)]
@@ -27,7 +27,9 @@
 //! navigation index, external-schema compatibility and the bounded prerelease
 //! checksum transition from the provisional `pre.002` schema. `0.3.4-pre.004`
 //! adds backend-private RAW account state/observation read mapping and hostile-row
-//! guards while keeping all account writes, pagination and trait implementations closed.
+//! guards. `0.3.4-pre.005` adds atomic account state+observation acquisition writes
+//! with exact idempotence/conflict classification while pagination, additional-observation
+//! writes and all four account trait implementations remain closed.
 //!
 //! This crate depends on `ksp-store-api` and never on `ksp-store-lib`. The
 //! common facade consumes only this crate's narrow backend bridge and never
@@ -71,6 +73,8 @@ pub(crate) use self::migration::current_migration_version;
 pub(crate) use self::raw_account::get_raw_account_observation;
 /// Private RAW account state reader consumed by the physical backend runtime.
 pub(crate) use self::raw_account::get_raw_account_state;
+/// Private atomic RAW account acquisition writer consumed by the physical backend runtime.
+pub(crate) use self::raw_account::persist_raw_account_acquisition;
 /// Private RAW transaction cursor decoder consumed by the physical RAW module.
 pub(crate) use self::raw_transaction::cursor::decode_raw_transaction_cursor;
 /// Private RAW transaction cursor encoder consumed by the physical RAW module.

crates/ksp-store-postgres-lib/src/raw_account.rs

@@ -1,8 +1,12 @@
 // file: crates/ksp-store-postgres-lib/src/raw_account.rs
-// version: 2
+// version: 3
 
 const GET_ACCOUNT_OBSERVATION_SQL: &str = "SELECT observation_key, account_pubkey, account_slot::text AS account_slot_text, account_state_hash, provider, protocol, acquisition_method, origin, received_at_unix_millis, capture_session_id, commitment, endpoint_id, filter_id, observed_at_unix_millis, source_payload_hash, source_payload_size_bytes, is_startup, transaction_signature, write_version::text AS write_version_text FROM ksp_raw_account_observations WHERE observation_key = $1";
 const GET_ACCOUNT_STATE_SQL: &str = "SELECT pubkey, slot::text AS slot_text, state_hash, lamports::text AS lamports_text, owner, executable, rent_epoch::text AS rent_epoch_text, data FROM ksp_raw_account_states WHERE pubkey = $1 AND slot = $2::TEXT::NUMERIC AND state_hash = $3";
+const INSERT_ACCOUNT_OBSERVATION_SQL: &str = "INSERT INTO ksp_raw_account_observations (observation_key, account_pubkey, account_slot, account_state_hash, provider, protocol, acquisition_method, origin, received_at_unix_millis, capture_session_id, commitment, endpoint_id, filter_id, observed_at_unix_millis, source_payload_hash, source_payload_size_bytes, is_startup, transaction_signature, write_version) VALUES ($1, $2, $3::TEXT::NUMERIC, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19::TEXT::NUMERIC) ON CONFLICT (observation_key) DO NOTHING RETURNING observation_key";
+const INSERT_ACCOUNT_STATE_SQL: &str = "INSERT INTO ksp_raw_account_states (pubkey, slot, state_hash, lamports, owner, executable, rent_epoch, data) VALUES ($1, $2::TEXT::NUMERIC, $3, $4::TEXT::NUMERIC, $5, $6, $7::TEXT::NUMERIC, $8) ON CONFLICT (pubkey, slot, state_hash) DO NOTHING RETURNING pubkey";
+const LOCK_ACCOUNT_OBSERVATION_SQL: &str = "SELECT observation_key, account_pubkey, account_slot::text AS account_slot_text, account_state_hash, provider, protocol, acquisition_method, origin, received_at_unix_millis, capture_session_id, commitment, endpoint_id, filter_id, observed_at_unix_millis, source_payload_hash, source_payload_size_bytes, is_startup, transaction_signature, write_version::text AS write_version_text FROM ksp_raw_account_observations WHERE observation_key = $1 FOR UPDATE";
+const LOCK_ACCOUNT_STATE_SQL: &str = "SELECT pubkey, slot::text AS slot_text, state_hash, lamports::text AS lamports_text, owner, executable, rent_epoch::text AS rent_epoch_text, data FROM ksp_raw_account_states WHERE pubkey = $1 AND slot = $2::TEXT::NUMERIC AND state_hash = $3 FOR UPDATE";
 
 struct RawAccountObservationDbRow {
     account_pubkey: std::vec::Vec<u8>,
@@ -103,7 +107,10 @@ pub(crate) async fn get_raw_account_observation(
     let rows = match rows_result {
         std::result::Result::Ok(value) => value,
         std::result::Result::Err(_) => {
-            return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::ReadFailed, "raw_account_observation_query"));
+            return std::result::Result::Err(crate::PostgresBackendError::new(
+                crate::PostgresBackendErrorKind::ReadFailed,
+                "raw_account_observation_query",
+            ));
         },
     };
     if rows.is_empty() {
@@ -130,6 +137,58 @@ pub(crate) async fn get_raw_account_observation(
     return std::result::Result::Ok(std::option::Option::Some(decoded));
 }
 
+/// Persists one canonical RAW account state and its acquisition observation atomically.
+pub(crate) async fn persist_raw_account_acquisition(
+    pool: &deadpool_postgres::Pool,
+    network: &ksp_store_api::RawNetworkId,
+    state: ksp_store_api::RawAccountState,
+    observation: ksp_store_api::RawAccountObservation,
+) -> std::result::Result<ksp_store_api::RawAcquisitionWriteOutcome, crate::PostgresBackendError> {
+    let input_result = ensure_acquisition_inputs(network, &state, &observation);
+    if let std::result::Result::Err(error) = input_result {
+        return std::result::Result::Err(error);
+    }
+    let client_result = pool.get().await;
+    let mut client = match client_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(crate::map_pool_error(error)),
+    };
+    let sql_transaction_result = client.transaction().await;
+    let sql_transaction = match sql_transaction_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(write_failed("raw_account_acquisition_begin")),
+    };
+    let insert_result = insert_account_state(&sql_transaction, &state).await;
+    let inserted = match insert_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let entity_outcome = if inserted {
+        ksp_store_api::RawEntityWriteOutcome::Inserted
+    } else {
+        let locked_result = load_locked_account_state(&sql_transaction, network, state.reference()).await;
+        let locked = match locked_result {
+            std::result::Result::Ok(std::option::Option::Some(value)) => value,
+            std::result::Result::Ok(std::option::Option::None) => return std::result::Result::Err(data_invalid("raw_account_acquisition_conflict_missing")),
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        if !raw_account_states_equal(&locked, &state) {
+            return std::result::Result::Err(conflict("raw_account_acquisition_content_conflict"));
+        }
+        ksp_store_api::RawEntityWriteOutcome::AlreadyPresent
+    };
+    let observation_result = persist_account_observation_row(&sql_transaction, network, &observation).await;
+    let observation_outcome = match observation_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let commit_result = sql_transaction.commit().await;
+    if commit_result.is_err() {
+        return std::result::Result::Err(write_failed("raw_account_acquisition_commit"));
+    }
+    return std::result::Result::Ok(ksp_store_api::RawAcquisitionWriteOutcome::new(entity_outcome, observation_outcome));
+}
+
 fn raw_account_observation_db_row(row: &tokio_postgres::Row) -> std::result::Result<RawAccountObservationDbRow, crate::PostgresBackendError> {
     let account_pubkey: std::vec::Vec<u8> = match row.try_get("account_pubkey") {
         std::result::Result::Ok(value) => value,
@@ -432,6 +491,209 @@ fn decode_raw_account_state_row(
     };
 }
 
+async fn insert_account_state(
+    sql_transaction: &deadpool_postgres::Transaction<'_>,
+    state: &ksp_store_api::RawAccountState,
+) -> std::result::Result<bool, crate::PostgresBackendError> {
+    let reference = state.reference();
+    let pubkey_bytes: &[u8] = reference.pubkey().as_ref();
+    let slot_text = reference.slot().to_string();
+    let state_hash = reference.state_hash();
+    let state_hash_bytes: &[u8] = state_hash.as_bytes();
+    let lamports_text = state.lamports().to_string();
+    let owner_bytes: &[u8] = state.owner().as_ref();
+    let executable = state.executable();
+    let rent_epoch_text = state.rent_epoch().to_string();
+    let data = state.data();
+    let row_result = sql_transaction
+        .query_opt(
+            INSERT_ACCOUNT_STATE_SQL,
+            &[
+                &pubkey_bytes,
+                &slot_text.as_str(),
+                &state_hash_bytes,
+                &lamports_text.as_str(),
+                &owner_bytes,
+                &executable,
+                &rent_epoch_text.as_str(),
+                &data,
+            ],
+        )
+        .await;
+    return match row_result {
+        std::result::Result::Ok(std::option::Option::Some(_)) => std::result::Result::Ok(true),
+        std::result::Result::Ok(std::option::Option::None) => std::result::Result::Ok(false),
+        std::result::Result::Err(_) => std::result::Result::Err(write_failed("raw_account_acquisition_insert_state")),
+    };
+}
+
+async fn load_locked_account_state(
+    sql_transaction: &deadpool_postgres::Transaction<'_>,
+    network: &ksp_store_api::RawNetworkId,
+    reference: &ksp_store_api::RawAccountStateReference,
+) -> std::result::Result<std::option::Option<ksp_store_api::RawAccountState>, crate::PostgresBackendError> {
+    let pubkey_bytes: &[u8] = reference.pubkey().as_ref();
+    let slot_text = reference.slot().to_string();
+    let state_hash = reference.state_hash();
+    let state_hash_bytes: &[u8] = state_hash.as_bytes();
+    let row_result = sql_transaction.query_opt(LOCK_ACCOUNT_STATE_SQL, &[&pubkey_bytes, &slot_text.as_str(), &state_hash_bytes]).await;
+    let row = match row_result {
+        std::result::Result::Ok(std::option::Option::Some(value)) => value,
+        std::result::Result::Ok(std::option::Option::None) => return std::result::Result::Ok(std::option::Option::None),
+        std::result::Result::Err(_) => return std::result::Result::Err(write_failed("raw_account_acquisition_lock_state")),
+    };
+    let physical = match raw_account_state_db_row(&row) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    return match decode_raw_account_state_row(network, physical) {
+        std::result::Result::Ok(value) => std::result::Result::Ok(std::option::Option::Some(value)),
+        std::result::Result::Err(error) => std::result::Result::Err(error),
+    };
+}
+
+async fn persist_account_observation_row(
+    sql_transaction: &deadpool_postgres::Transaction<'_>,
+    network: &ksp_store_api::RawNetworkId,
+    observation: &ksp_store_api::RawAccountObservation,
+) -> std::result::Result<ksp_store_api::RawObservationWriteOutcome, crate::PostgresBackendError> {
+    let provenance = observation.provenance();
+    let observation_key = observation.observation_key();
+    let observation_key_bytes: &[u8] = observation_key.as_bytes();
+    let account = observation.account();
+    let account_pubkey_bytes: &[u8] = account.pubkey().as_ref();
+    let account_slot_text = account.slot().to_string();
+    let account_state_hash = account.state_hash();
+    let account_state_hash_bytes: &[u8] = account_state_hash.as_bytes();
+    let origin = match encode_origin(provenance.origin()) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let received_at = match i64::try_from(provenance.received_at().unix_millis()) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_account_observation_received_at_encode")),
+    };
+    let observed_at = match provenance.observed_at() {
+        std::option::Option::Some(value) => match i64::try_from(value.unix_millis()) {
+            std::result::Result::Ok(decoded) => std::option::Option::Some(decoded),
+            std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_account_observation_observed_at_encode")),
+        },
+        std::option::Option::None => std::option::Option::None,
+    };
+    let source_payload_size = match provenance.source_payload_size_bytes() {
+        std::option::Option::Some(value) => match i64::try_from(value) {
+            std::result::Result::Ok(decoded) => std::option::Option::Some(decoded),
+            std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_account_observation_source_size_encode")),
+        },
+        std::option::Option::None => std::option::Option::None,
+    };
+    let capture_session_id = provenance.capture_session_id().map(|value| return value.as_str());
+    let commitment = provenance.commitment().map(|value| return value.as_str());
+    let endpoint_id = provenance.endpoint_id().map(|value| return value.as_str());
+    let filter_id = provenance.filter_id().map(|value| return value.as_str());
+    let source_payload_hash = provenance.source_payload_hash();
+    let source_payload_hash_bytes: std::option::Option<&[u8]> = source_payload_hash.as_ref().map(|value| return &value.as_bytes()[..]);
+    let is_startup = observation.is_startup();
+    let transaction_signature = observation.transaction_signature();
+    let transaction_signature_bytes: std::option::Option<&[u8]> = transaction_signature.as_ref().map(|value| return &value.as_bytes()[..]);
+    let write_version_text = observation.write_version().map(|value| return value.to_string());
+    let row_result = sql_transaction
+        .query_opt(
+            INSERT_ACCOUNT_OBSERVATION_SQL,
+            &[
+                &observation_key_bytes,
+                &account_pubkey_bytes,
+                &account_slot_text.as_str(),
+                &account_state_hash_bytes,
+                &provenance.provider().as_str(),
+                &provenance.protocol().as_str(),
+                &provenance.acquisition_method().as_str(),
+                &origin,
+                &received_at,
+                &capture_session_id,
+                &commitment,
+                &endpoint_id,
+                &filter_id,
+                &observed_at,
+                &source_payload_hash_bytes,
+                &source_payload_size,
+                &is_startup,
+                &transaction_signature_bytes,
+                &write_version_text,
+            ],
+        )
+        .await;
+    let inserted = match row_result {
+        std::result::Result::Ok(std::option::Option::Some(_)) => true,
+        std::result::Result::Ok(std::option::Option::None) => false,
+        std::result::Result::Err(_) => return std::result::Result::Err(write_failed("raw_account_observation_insert")),
+    };
+    if inserted {
+        return std::result::Result::Ok(ksp_store_api::RawObservationWriteOutcome::Inserted);
+    }
+    let existing_result = sql_transaction.query_opt(LOCK_ACCOUNT_OBSERVATION_SQL, &[&observation_key_bytes]).await;
+    let existing_row = match existing_result {
+        std::result::Result::Ok(std::option::Option::Some(value)) => value,
+        std::result::Result::Ok(std::option::Option::None) => return std::result::Result::Err(data_invalid("raw_account_observation_conflict_missing")),
+        std::result::Result::Err(_) => return std::result::Result::Err(write_failed("raw_account_observation_conflict_query")),
+    };
+    let physical = match raw_account_observation_db_row(&existing_row) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let stored = match decode_raw_account_observation_row(network, physical) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    if stored.eq(observation) {
+        return std::result::Result::Ok(ksp_store_api::RawObservationWriteOutcome::AlreadyPresent);
+    }
+    return std::result::Result::Err(conflict("raw_account_observation_content_conflict"));
+}
+
+fn ensure_acquisition_inputs(
+    network: &ksp_store_api::RawNetworkId,
+    state: &ksp_store_api::RawAccountState,
+    observation: &ksp_store_api::RawAccountObservation,
+) -> std::result::Result<(), crate::PostgresBackendError> {
+    let state_network_result = ensure_network(network, state.reference(), "raw_account_acquisition_state_network");
+    if let std::result::Result::Err(error) = state_network_result {
+        return std::result::Result::Err(error);
+    }
+    let observation_network_result = ensure_network(network, observation.account(), "raw_account_acquisition_observation_network");
+    if let std::result::Result::Err(error) = observation_network_result {
+        return std::result::Result::Err(error);
+    }
+    if observation.account() != state.reference() {
+        return std::result::Result::Err(conflict("raw_account_acquisition_reference_mismatch"));
+    }
+    return std::result::Result::Ok(());
+}
+
+fn raw_account_states_equal(left: &ksp_store_api::RawAccountState, right: &ksp_store_api::RawAccountState) -> bool {
+    return left.reference() == right.reference()
+        && left.lamports() == right.lamports()
+        && left.owner() == right.owner()
+        && left.executable() == right.executable()
+        && left.rent_epoch() == right.rent_epoch()
+        && left.data() == right.data();
+}
+
+fn encode_origin(origin: ksp_store_api::RawAcquisitionOrigin) -> std::result::Result<&'static str, crate::PostgresBackendError> {
+    return match origin {
+        ksp_store_api::RawAcquisitionOrigin::Backfill => std::result::Result::Ok("backfill"),
+        ksp_store_api::RawAcquisitionOrigin::Import => std::result::Result::Ok("import"),
+        ksp_store_api::RawAcquisitionOrigin::Live => std::result::Result::Ok("live"),
+        ksp_store_api::RawAcquisitionOrigin::Repair => std::result::Result::Ok("repair"),
+        ksp_store_api::RawAcquisitionOrigin::Replay => std::result::Result::Ok("replay"),
+        _ => std::result::Result::Err(data_invalid("raw_account_observation_origin_encode")),
+    };
+}
+
+fn conflict(phase: &'static str) -> crate::PostgresBackendError {
+    return crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::Conflict, phase);
+}
+
 fn decode_origin(value: &str) -> std::result::Result<ksp_store_api::RawAcquisitionOrigin, crate::PostgresBackendError> {
     return match value {
         "backfill" => std::result::Result::Ok(ksp_store_api::RawAcquisitionOrigin::Backfill),
@@ -497,6 +759,10 @@ fn data_invalid(phase: &'static str) -> crate::PostgresBackendError {
     return crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::DataInvalid, phase);
 }
 
+fn write_failed(phase: &'static str) -> crate::PostgresBackendError {
+    return crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::WriteFailed, phase);
+}
+
 #[cfg(test)]
 #[path = "../unit_tests/raw_account.rs"]
 mod tests;

crates/ksp-store-postgres-lib/src/runtime.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-store-postgres-lib/src/runtime.rs
-// version: 11
+// version: 12
 
 const APPLICATION_NAME: &str = "ksp-store";
 const MAX_CONNECTION_URI_BYTES: usize = 4_096;
@@ -348,6 +348,15 @@ impl PostgresBackend {
         return crate::get_raw_account_observation(&self.pool, &self.network, observation_key).await;
     }
 
+    /// Persists one complete RAW account state and its acquisition observation atomically.
+    pub async fn persist_raw_account_acquisition(
+        &self,
+        state: ksp_store_api::RawAccountState,
+        observation: ksp_store_api::RawAccountObservation,
+    ) -> std::result::Result<ksp_store_api::RawAcquisitionWriteOutcome, crate::PostgresBackendError> {
+        return crate::persist_raw_account_acquisition(&self.pool, &self.network, state, observation).await;
+    }
+
     /// Reads one canonical RAW transaction without exposing physical PostgreSQL row types.
     pub async fn get_raw_transaction(
         &self,