v0.3.4-pre.006

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-store-postgres-lib/src/lib.rs
-// version: 18
+// version: 19
 
 #![warn(missing_docs)]
 #![deny(unreachable_pub)]
@@ -28,8 +28,9 @@
 //! 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. `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.
+//! with exact idempotence/conflict classification. `0.3.4-pre.006` adds additional
+//! account observations guarded by the existing state reference while pagination 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
@@ -75,6 +76,8 @@ pub(crate) use self::raw_account::get_raw_account_observation;
 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 additional RAW account observation writer consumed by the physical backend runtime.
+pub(crate) use self::raw_account::record_raw_account_observation;
 /// 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,11 +1,13 @@
 // file: crates/ksp-store-postgres-lib/src/raw_account.rs
-// version: 3
+// version: 4
 
 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_REFERENCE_SQL: &str =
+    "SELECT 1 FROM ksp_raw_account_states WHERE pubkey = $1 AND slot = $2::TEXT::NUMERIC AND state_hash = $3 FOR KEY SHARE";
 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 {
@@ -107,10 +109,7 @@ 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() {
@@ -189,6 +188,53 @@ pub(crate) async fn persist_raw_account_acquisition(
     return std::result::Result::Ok(ksp_store_api::RawAcquisitionWriteOutcome::new(entity_outcome, observation_outcome));
 }
 
+/// Persists one additional acquisition observation for an already durable RAW account state.
+pub(crate) async fn record_raw_account_observation(
+    pool: &deadpool_postgres::Pool,
+    network: &ksp_store_api::RawNetworkId,
+    observation: ksp_store_api::RawAccountObservation,
+) -> std::result::Result<ksp_store_api::RawObservationWriteOutcome, crate::PostgresBackendError> {
+    let input_result = ensure_observation_write_input(network, &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_observation_begin")),
+    };
+    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 reference_result = sql_transaction
+        .query_opt(LOCK_ACCOUNT_REFERENCE_SQL, &[&account_pubkey_bytes, &account_slot_text.as_str(), &account_state_hash_bytes])
+        .await;
+    match reference_result {
+        std::result::Result::Ok(std::option::Option::Some(_)) => {},
+        std::result::Result::Ok(std::option::Option::None) => {
+            return std::result::Result::Err(reference_not_found("raw_account_observation_reference"));
+        },
+        std::result::Result::Err(_) => return std::result::Result::Err(write_failed("raw_account_observation_lock_reference")),
+    }
+    let observation_result = persist_account_observation_row(&sql_transaction, network, &observation).await;
+    let 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_observation_commit"));
+    }
+    return std::result::Result::Ok(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,
@@ -508,16 +554,7 @@ async fn insert_account_state(
     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,
-            ],
+            &[&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 {
@@ -670,6 +707,13 @@ fn ensure_acquisition_inputs(
     return std::result::Result::Ok(());
 }
 
+fn ensure_observation_write_input(
+    network: &ksp_store_api::RawNetworkId,
+    observation: &ksp_store_api::RawAccountObservation,
+) -> std::result::Result<(), crate::PostgresBackendError> {
+    return ensure_network(network, observation.account(), "raw_account_observation_write_network");
+}
+
 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()
@@ -759,6 +803,10 @@ fn data_invalid(phase: &'static str) -> crate::PostgresBackendError {
     return crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::DataInvalid, phase);
 }
 
+fn reference_not_found(phase: &'static str) -> crate::PostgresBackendError {
+    return crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::ReferenceNotFound, phase);
+}
+
 fn write_failed(phase: &'static str) -> crate::PostgresBackendError {
     return crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::WriteFailed, phase);
 }

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-store-postgres-lib/src/runtime.rs
-// version: 12
+// version: 13
 
 const APPLICATION_NAME: &str = "ksp-store";
 const MAX_CONNECTION_URI_BYTES: usize = 4_096;
@@ -357,6 +357,14 @@ impl PostgresBackend {
         return crate::persist_raw_account_acquisition(&self.pool, &self.network, state, observation).await;
     }
 
+    /// Persists one additional RAW account observation for an already durable account state.
+    pub async fn record_raw_account_observation(
+        &self,
+        observation: ksp_store_api::RawAccountObservation,
+    ) -> std::result::Result<ksp_store_api::RawObservationWriteOutcome, crate::PostgresBackendError> {
+        return crate::record_raw_account_observation(&self.pool, &self.network, observation).await;
+    }
+
     /// Reads one canonical RAW transaction without exposing physical PostgreSQL row types.
     pub async fn get_raw_transaction(
         &self,

crates/ksp-store-postgres-lib/tests/dependency_boundary.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-store-postgres-lib/tests/dependency_boundary.rs
-// version: 19
+// version: 20
 
 #![warn(missing_docs)]
 #![deny(unreachable_pub)]
@@ -180,7 +180,7 @@ fn pre_005_raw_account_acquisition_is_atomic_idempotent_and_keeps_later_scope_cl
         assert!(raw.contains(required), "missing private RAW account acquisition contract: {required}");
     }
     assert!(runtime.contains("pub async fn persist_raw_account_acquisition"));
-    for forbidden in ["UPDATE ", "DELETE FROM", "ON CONFLICT DO UPDATE", " OFFSET ", "list_raw_account_states", "record_raw_account_observation"] {
+    for forbidden in ["UPDATE ", "DELETE FROM", "ON CONFLICT DO UPDATE", " OFFSET ", "list_raw_account_states"] {
         assert!(!raw.contains(forbidden), "pre.005 account module contains later/destructive scope: {forbidden}");
     }
     for forbidden in [
@@ -197,6 +197,36 @@ fn pre_005_raw_account_acquisition_is_atomic_idempotent_and_keeps_later_scope_cl
     return;
 }
 
+#[test]
+fn pre_006_raw_account_additional_observation_is_reference_guarded_and_cancellation_safe() {
+    let raw = include_str!("../src/raw_account.rs");
+    let runtime = include_str!("../src/runtime.rs");
+    for required in [
+        "LOCK_ACCOUNT_REFERENCE_SQL",
+        "FOR KEY SHARE",
+        "record_raw_account_observation",
+        "reference_not_found(\"raw_account_observation_reference\")",
+        "persist_account_observation_row",
+        "RawObservationWriteOutcome",
+        "raw_account_observation_commit",
+    ] {
+        assert!(raw.contains(required), "missing pre.006 additional-observation contract: {required}");
+    }
+    assert!(runtime.contains("pub async fn record_raw_account_observation"));
+    for forbidden in ["UPDATE ", "DELETE FROM", "ON CONFLICT DO UPDATE", " OFFSET ", "list_raw_account_states"] {
+        assert!(!raw.contains(forbidden), "pre.006 account module contains later/destructive scope: {forbidden}");
+    }
+    for forbidden in [
+        "impl ksp_store_api::RawAccountStateRead for PostgresBackend",
+        "impl ksp_store_api::RawAccountStateWrite for PostgresBackend",
+        "impl ksp_store_api::RawAccountObservationRead for PostgresBackend",
+        "impl ksp_store_api::RawAccountObservationWrite for PostgresBackend",
+    ] {
+        assert!(!runtime.contains(forbidden), "pre.006 opened RawAccount trait scope prematurely: {forbidden}");
+    }
+    return;
+}
+
 #[test]
 fn pre_004_raw_read_sql_and_mapping_remain_backend_private() {
     let crate_root = include_str!("../src/lib.rs");

crates/ksp-store-postgres-lib/tests/public_api.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-store-postgres-lib/tests/public_api.rs
-// version: 11
+// version: 12
 
 #![warn(missing_docs)]
 #![deny(unreachable_pub)]
@@ -88,6 +88,12 @@ fn pre_005_raw_account_acquisition_bridge_uses_only_backend_independent_models_a
     return;
 }
 
+#[test]
+fn pre_006_raw_account_observation_write_bridge_uses_only_backend_independent_models_and_outcomes() {
+    let _observation = ksp_store_postgres_lib::PostgresBackend::record_raw_account_observation;
+    return;
+}
+
 #[test]
 fn pre_004_raw_read_bridge_uses_only_backend_independent_models() {
     let _get = ksp_store_postgres_lib::PostgresBackend::get_raw_transaction;

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-store-postgres-lib/unit_tests/raw_account.rs
-// version: 2
+// version: 3
 
 fn network() -> ksp_store_api::RawNetworkId {
     return match ksp_store_api::RawNetworkId::new("devnet") {
@@ -322,3 +322,57 @@ fn pre_005_account_origin_encoding_is_exact_and_static() {
     }
     return;
 }
+
+#[test]
+fn pre_006_additional_observation_input_guard_rejects_wrong_network_before_io() {
+    let backend = network();
+    let foreign = match ksp_store_api::RawNetworkId::new("mainnet-beta") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => panic!("valid foreign network rejected: {error:?}"),
+    };
+    let observation = match super::decode_raw_account_observation_row(&foreign, observation_row()) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => panic!("valid foreign account observation rejected: {error:?}"),
+    };
+    let error = super::ensure_observation_write_input(&backend, &observation).err();
+    assert_eq!(error.map(|value| return value.kind()), std::option::Option::Some(crate::PostgresBackendErrorKind::WrongNetwork));
+    return;
+}
+
+#[test]
+fn pre_006_additional_observation_race_and_cancellation_contract_is_transaction_bounded() {
+    let source = include_str!("../src/raw_account.rs");
+    let start = match source.find("pub(crate) async fn record_raw_account_observation") {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => panic!("record_raw_account_observation function missing"),
+    };
+    let tail = &source[start..];
+    let end = match tail.find("fn raw_account_observation_db_row") {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => panic!("record_raw_account_observation boundary missing"),
+    };
+    let function = &tail[..end];
+    let begin = match function.find("client.transaction().await") {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => panic!("additional observation transaction begin missing"),
+    };
+    let key_share = match function.find("LOCK_ACCOUNT_REFERENCE_SQL") {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => panic!("additional observation reference lock missing"),
+    };
+    let persist = match function.find("persist_account_observation_row") {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => panic!("additional observation insert/compare missing"),
+    };
+    let commit = match function.find("sql_transaction.commit().await") {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => panic!("additional observation commit missing"),
+    };
+    assert!(begin < key_share);
+    assert!(key_share < persist);
+    assert!(persist < commit);
+    assert!(!function.contains("tokio::spawn"));
+    assert!(!function.contains("std::mem::forget"));
+    assert!(!function.contains("ON CONFLICT DO UPDATE"));
+    return;
+}