v0.3.4-pre.004

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-store-postgres-lib/src/lib.rs
-// version: 16
+// version: 17
 
 #![warn(missing_docs)]
 #![deny(unreachable_pub)]
@@ -25,7 +25,9 @@
 //! tables with canonical state/observation PKs and the observation-state FK.
 //! `0.3.4-pre.003` completes V002 with exact physical bounds, one unfiltered
 //! navigation index, external-schema compatibility and the bounded prerelease
-//! checksum transition from the provisional `pre.002` schema.
+//! 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.
 //!
 //! 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
@@ -35,6 +37,7 @@ mod constants;
 mod error;
 mod health;
 mod migration;
+mod raw_account;
 mod raw_transaction;
 mod runtime;
 mod schema;
@@ -64,6 +67,10 @@ pub(crate) use self::health::probe_health;
 pub(crate) use self::migration::bootstrap;
 /// Current embedded migration version consumed by the private health probe.
 pub(crate) use self::migration::current_migration_version;
+/// Private RAW account observation reader consumed by the physical backend runtime.
+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 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

@@ -0,0 +1,403 @@
+// file: crates/ksp-store-postgres-lib/src/raw_account.rs
+// version: 1
+
+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";
+
+struct RawAccountObservationDbRow {
+    account_pubkey: std::vec::Vec<u8>,
+    account_slot_text: std::string::String,
+    account_state_hash: std::vec::Vec<u8>,
+    acquisition_method: std::string::String,
+    capture_session_id: std::option::Option<std::string::String>,
+    commitment: std::option::Option<std::string::String>,
+    endpoint_id: std::option::Option<std::string::String>,
+    filter_id: std::option::Option<std::string::String>,
+    is_startup: std::option::Option<bool>,
+    observation_key: std::vec::Vec<u8>,
+    observed_at_unix_millis: std::option::Option<i64>,
+    origin: std::string::String,
+    protocol: std::string::String,
+    provider: std::string::String,
+    received_at_unix_millis: i64,
+    source_payload_hash: std::option::Option<std::vec::Vec<u8>>,
+    source_payload_size_bytes: std::option::Option<i64>,
+    transaction_signature: std::option::Option<std::vec::Vec<u8>>,
+    write_version_text: std::option::Option<std::string::String>,
+}
+
+struct RawAccountStateDbRow {
+    data: std::vec::Vec<u8>,
+    executable: bool,
+    lamports_text: std::string::String,
+    owner: std::vec::Vec<u8>,
+    pubkey: std::vec::Vec<u8>,
+    rent_epoch_text: std::string::String,
+    slot_text: std::string::String,
+    state_hash: std::vec::Vec<u8>,
+}
+
+/// Reads one complete canonical RAW account state from the physical PostgreSQL backend.
+pub(crate) async fn get_raw_account_state(
+    pool: &deadpool_postgres::Pool,
+    network: &ksp_store_api::RawNetworkId,
+    reference: &ksp_store_api::RawAccountStateReference,
+) -> std::result::Result<std::option::Option<ksp_store_api::RawAccountState>, crate::PostgresBackendError> {
+    let network_result = ensure_network(network, reference, "raw_account_state_network");
+    if let std::result::Result::Err(error) = network_result {
+        return std::result::Result::Err(error);
+    }
+    let client_result = pool.get().await;
+    let 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 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 rows_result = client.query(GET_ACCOUNT_STATE_SQL, &[&pubkey_bytes, &slot_text, &state_hash_bytes]).await;
+    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_state_query"));
+        },
+    };
+    if rows.is_empty() {
+        return std::result::Result::Ok(std::option::Option::None);
+    }
+    if rows.len() != 1 {
+        return std::result::Result::Err(data_invalid("raw_account_state_cardinality"));
+    }
+    let row = match rows.first() {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return std::result::Result::Err(data_invalid("raw_account_state_cardinality")),
+    };
+    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),
+    };
+    let decoded = match decode_raw_account_state_row(network, physical) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    if decoded.reference() != reference {
+        return std::result::Result::Err(data_invalid("raw_account_state_reference"));
+    }
+    return std::result::Result::Ok(std::option::Option::Some(decoded));
+}
+
+/// Reads one persisted RAW account observation by producer-owned idempotence key.
+pub(crate) async fn get_raw_account_observation(
+    pool: &deadpool_postgres::Pool,
+    network: &ksp_store_api::RawNetworkId,
+    observation_key: &ksp_store_api::RawObservationKey,
+) -> std::result::Result<std::option::Option<ksp_store_api::RawAccountObservation>, crate::PostgresBackendError> {
+    let client_result = pool.get().await;
+    let 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 observation_key_bytes: &[u8] = observation_key.as_bytes();
+    let rows_result = client.query(GET_ACCOUNT_OBSERVATION_SQL, &[&observation_key_bytes]).await;
+    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"));
+        },
+    };
+    if rows.is_empty() {
+        return std::result::Result::Ok(std::option::Option::None);
+    }
+    if rows.len() != 1 {
+        return std::result::Result::Err(data_invalid("raw_account_observation_cardinality"));
+    }
+    let row = match rows.first() {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return std::result::Result::Err(data_invalid("raw_account_observation_cardinality")),
+    };
+    let physical = match raw_account_observation_db_row(row) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let decoded = 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 decoded.observation_key() != *observation_key {
+        return std::result::Result::Err(data_invalid("raw_account_observation_key"));
+    }
+    return std::result::Result::Ok(std::option::Option::Some(decoded));
+}
+
+fn raw_account_observation_db_row(row: &tokio_postgres::Row) -> std::result::Result<RawAccountObservationDbRow, crate::PostgresBackendError> {
+    return std::result::Result::Ok(RawAccountObservationDbRow {
+        account_pubkey: row.try_get("account_pubkey").map_err(|_| data_invalid("raw_account_observation_decode"))?,
+        account_slot_text: row.try_get("account_slot_text").map_err(|_| data_invalid("raw_account_observation_decode"))?,
+        account_state_hash: row.try_get("account_state_hash").map_err(|_| data_invalid("raw_account_observation_decode"))?,
+        acquisition_method: row.try_get("acquisition_method").map_err(|_| data_invalid("raw_account_observation_decode"))?,
+        capture_session_id: row.try_get("capture_session_id").map_err(|_| data_invalid("raw_account_observation_decode"))?,
+        commitment: row.try_get("commitment").map_err(|_| data_invalid("raw_account_observation_decode"))?,
+        endpoint_id: row.try_get("endpoint_id").map_err(|_| data_invalid("raw_account_observation_decode"))?,
+        filter_id: row.try_get("filter_id").map_err(|_| data_invalid("raw_account_observation_decode"))?,
+        is_startup: row.try_get("is_startup").map_err(|_| data_invalid("raw_account_observation_decode"))?,
+        observation_key: row.try_get("observation_key").map_err(|_| data_invalid("raw_account_observation_decode"))?,
+        observed_at_unix_millis: row.try_get("observed_at_unix_millis").map_err(|_| data_invalid("raw_account_observation_decode"))?,
+        origin: row.try_get("origin").map_err(|_| data_invalid("raw_account_observation_decode"))?,
+        protocol: row.try_get("protocol").map_err(|_| data_invalid("raw_account_observation_decode"))?,
+        provider: row.try_get("provider").map_err(|_| data_invalid("raw_account_observation_decode"))?,
+        received_at_unix_millis: row.try_get("received_at_unix_millis").map_err(|_| data_invalid("raw_account_observation_decode"))?,
+        source_payload_hash: row.try_get("source_payload_hash").map_err(|_| data_invalid("raw_account_observation_decode"))?,
+        source_payload_size_bytes: row.try_get("source_payload_size_bytes").map_err(|_| data_invalid("raw_account_observation_decode"))?,
+        transaction_signature: row.try_get("transaction_signature").map_err(|_| data_invalid("raw_account_observation_decode"))?,
+        write_version_text: row.try_get("write_version_text").map_err(|_| data_invalid("raw_account_observation_decode"))?,
+    });
+}
+
+fn raw_account_state_db_row(row: &tokio_postgres::Row) -> std::result::Result<RawAccountStateDbRow, crate::PostgresBackendError> {
+    return std::result::Result::Ok(RawAccountStateDbRow {
+        data: row.try_get("data").map_err(|_| data_invalid("raw_account_state_decode"))?,
+        executable: row.try_get("executable").map_err(|_| data_invalid("raw_account_state_decode"))?,
+        lamports_text: row.try_get("lamports_text").map_err(|_| data_invalid("raw_account_state_decode"))?,
+        owner: row.try_get("owner").map_err(|_| data_invalid("raw_account_state_decode"))?,
+        pubkey: row.try_get("pubkey").map_err(|_| data_invalid("raw_account_state_decode"))?,
+        rent_epoch_text: row.try_get("rent_epoch_text").map_err(|_| data_invalid("raw_account_state_decode"))?,
+        slot_text: row.try_get("slot_text").map_err(|_| data_invalid("raw_account_state_decode"))?,
+        state_hash: row.try_get("state_hash").map_err(|_| data_invalid("raw_account_state_decode"))?,
+    });
+}
+
+fn decode_raw_account_observation_row(
+    network: &ksp_store_api::RawNetworkId,
+    row: RawAccountObservationDbRow,
+) -> std::result::Result<ksp_store_api::RawAccountObservation, crate::PostgresBackendError> {
+    let observation_key = match fixed_bytes::<32>(row.observation_key, "raw_account_observation_key") {
+        std::result::Result::Ok(value) => ksp_store_api::RawObservationKey::new(value),
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let pubkey = match fixed_bytes::<32>(row.account_pubkey, "raw_account_observation_pubkey") {
+        std::result::Result::Ok(value) => ksp_store_api::Pubkey::new_from_array(value),
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let slot = match decode_u64_decimal(row.account_slot_text.as_str(), "raw_account_observation_slot") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let state_hash = match fixed_bytes::<32>(row.account_state_hash, "raw_account_observation_state_hash") {
+        std::result::Result::Ok(value) => ksp_store_api::RawContentHash::new(value),
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let provider = match decode_provenance_code(row.provider) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let protocol = match decode_provenance_code(row.protocol) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let acquisition_method = match decode_provenance_code(row.acquisition_method) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let origin = match decode_origin(row.origin.as_str()) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let received_at = match decode_timestamp_i64(row.received_at_unix_millis, "raw_account_observation_received_at") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let mut provenance = ksp_store_api::RawAcquisitionProvenance::new(provider, protocol, acquisition_method, origin, received_at);
+    provenance = match row.capture_session_id {
+        std::option::Option::Some(value) => match decode_provenance_code(value) {
+            std::result::Result::Ok(code) => provenance.with_capture_session_id(code),
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        },
+        std::option::Option::None => provenance,
+    };
+    provenance = match row.commitment {
+        std::option::Option::Some(value) => match decode_provenance_code(value) {
+            std::result::Result::Ok(code) => provenance.with_commitment(code),
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        },
+        std::option::Option::None => provenance,
+    };
+    provenance = match row.endpoint_id {
+        std::option::Option::Some(value) => match decode_provenance_code(value) {
+            std::result::Result::Ok(code) => provenance.with_endpoint_id(code),
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        },
+        std::option::Option::None => provenance,
+    };
+    provenance = match row.filter_id {
+        std::option::Option::Some(value) => match decode_provenance_code(value) {
+            std::result::Result::Ok(code) => provenance.with_filter_id(code),
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        },
+        std::option::Option::None => provenance,
+    };
+    provenance = match row.observed_at_unix_millis {
+        std::option::Option::Some(value) => {
+            let timestamp = match decode_timestamp_i64(value, "raw_account_observation_observed_at") {
+                std::result::Result::Ok(decoded) => decoded,
+                std::result::Result::Err(error) => return std::result::Result::Err(error),
+            };
+            match provenance.try_with_observed_at(timestamp) {
+                std::result::Result::Ok(updated) => updated,
+                std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_account_observation_time_order")),
+            }
+        },
+        std::option::Option::None => provenance,
+    };
+    provenance = match row.source_payload_hash {
+        std::option::Option::Some(value) => {
+            let hash = match fixed_bytes::<32>(value, "raw_account_observation_source_hash") {
+                std::result::Result::Ok(decoded) => ksp_store_api::RawContentHash::new(decoded),
+                std::result::Result::Err(error) => return std::result::Result::Err(error),
+            };
+            provenance.with_source_payload_hash(hash)
+        },
+        std::option::Option::None => provenance,
+    };
+    provenance = match row.source_payload_size_bytes {
+        std::option::Option::Some(value) => {
+            let size = match decode_u64_i64(value, "raw_account_observation_source_size") {
+                std::result::Result::Ok(decoded) => decoded,
+                std::result::Result::Err(error) => return std::result::Result::Err(error),
+            };
+            match provenance.try_with_source_payload_size_bytes(size) {
+                std::result::Result::Ok(updated) => updated,
+                std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_account_observation_source_size")),
+            }
+        },
+        std::option::Option::None => provenance,
+    };
+    let reference = ksp_store_api::RawAccountStateReference::new(network.clone(), pubkey, slot, state_hash);
+    let mut observation = ksp_store_api::RawAccountObservation::new(observation_key, reference, provenance);
+    observation = match row.is_startup {
+        std::option::Option::Some(value) => observation.with_is_startup(value),
+        std::option::Option::None => observation,
+    };
+    observation = match row.transaction_signature {
+        std::option::Option::Some(value) => {
+            let signature = match fixed_bytes::<64>(value, "raw_account_observation_transaction_signature") {
+                std::result::Result::Ok(decoded) => ksp_store_api::RawTransactionSignature::new(decoded),
+                std::result::Result::Err(error) => return std::result::Result::Err(error),
+            };
+            observation.with_transaction_signature(signature)
+        },
+        std::option::Option::None => observation,
+    };
+    observation = match row.write_version_text {
+        std::option::Option::Some(value) => match decode_u64_decimal(value.as_str(), "raw_account_observation_write_version") {
+            std::result::Result::Ok(decoded) => observation.with_write_version(decoded),
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        },
+        std::option::Option::None => observation,
+    };
+    return std::result::Result::Ok(observation);
+}
+
+fn decode_raw_account_state_row(
+    network: &ksp_store_api::RawNetworkId,
+    row: RawAccountStateDbRow,
+) -> std::result::Result<ksp_store_api::RawAccountState, crate::PostgresBackendError> {
+    let pubkey = match fixed_bytes::<32>(row.pubkey, "raw_account_state_pubkey") {
+        std::result::Result::Ok(value) => ksp_store_api::Pubkey::new_from_array(value),
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let slot = match decode_u64_decimal(row.slot_text.as_str(), "raw_account_state_slot") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let state_hash = match fixed_bytes::<32>(row.state_hash, "raw_account_state_hash") {
+        std::result::Result::Ok(value) => ksp_store_api::RawContentHash::new(value),
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let lamports = match decode_u64_decimal(row.lamports_text.as_str(), "raw_account_state_lamports") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let owner = match fixed_bytes::<32>(row.owner, "raw_account_state_owner") {
+        std::result::Result::Ok(value) => ksp_store_api::Pubkey::new_from_array(value),
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let rent_epoch = match decode_u64_decimal(row.rent_epoch_text.as_str(), "raw_account_state_rent_epoch") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let reference = ksp_store_api::RawAccountStateReference::new(network.clone(), pubkey, slot, state_hash);
+    return match ksp_store_api::RawAccountState::try_new(reference, lamports, owner, row.executable, rent_epoch, row.data.into_boxed_slice()) {
+        std::result::Result::Ok(value) => std::result::Result::Ok(value),
+        std::result::Result::Err(_) => std::result::Result::Err(data_invalid("raw_account_state_model")),
+    };
+}
+
+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),
+        "import" => std::result::Result::Ok(ksp_store_api::RawAcquisitionOrigin::Import),
+        "live" => std::result::Result::Ok(ksp_store_api::RawAcquisitionOrigin::Live),
+        "repair" => std::result::Result::Ok(ksp_store_api::RawAcquisitionOrigin::Repair),
+        "replay" => std::result::Result::Ok(ksp_store_api::RawAcquisitionOrigin::Replay),
+        _ => std::result::Result::Err(data_invalid("raw_account_observation_origin")),
+    };
+}
+
+fn decode_provenance_code(value: std::string::String) -> std::result::Result<ksp_store_api::RawProvenanceCode, crate::PostgresBackendError> {
+    return match ksp_store_api::RawProvenanceCode::new(value) {
+        std::result::Result::Ok(decoded) => std::result::Result::Ok(decoded),
+        std::result::Result::Err(_) => std::result::Result::Err(data_invalid("raw_account_provenance_code")),
+    };
+}
+
+fn decode_timestamp_i64(value: i64, phase: &'static str) -> std::result::Result<ksp_store_api::RawTimestamp, crate::PostgresBackendError> {
+    let unsigned = match u64::try_from(value) {
+        std::result::Result::Ok(decoded) => decoded,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid(phase)),
+    };
+    return match ksp_store_api::RawTimestamp::from_unix_millis(unsigned) {
+        std::result::Result::Ok(decoded) => std::result::Result::Ok(decoded),
+        std::result::Result::Err(_) => std::result::Result::Err(data_invalid(phase)),
+    };
+}
+
+fn decode_u64_decimal(value: &str, phase: &'static str) -> std::result::Result<u64, crate::PostgresBackendError> {
+    return match value.parse::<u64>() {
+        std::result::Result::Ok(decoded) => std::result::Result::Ok(decoded),
+        std::result::Result::Err(_) => std::result::Result::Err(data_invalid(phase)),
+    };
+}
+
+fn decode_u64_i64(value: i64, phase: &'static str) -> std::result::Result<u64, crate::PostgresBackendError> {
+    return match u64::try_from(value) {
+        std::result::Result::Ok(decoded) => std::result::Result::Ok(decoded),
+        std::result::Result::Err(_) => std::result::Result::Err(data_invalid(phase)),
+    };
+}
+
+fn fixed_bytes<const N: usize>(value: std::vec::Vec<u8>, phase: &'static str) -> std::result::Result<[u8; N], crate::PostgresBackendError> {
+    return match <[u8; N]>::try_from(value.as_slice()) {
+        std::result::Result::Ok(decoded) => std::result::Result::Ok(decoded),
+        std::result::Result::Err(_) => std::result::Result::Err(data_invalid(phase)),
+    };
+}
+
+fn ensure_network(
+    network: &ksp_store_api::RawNetworkId,
+    reference: &ksp_store_api::RawAccountStateReference,
+    phase: &'static str,
+) -> std::result::Result<(), crate::PostgresBackendError> {
+    if reference.network() != network {
+        return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::WrongNetwork, phase));
+    }
+    return std::result::Result::Ok(());
+}
+
+fn data_invalid(phase: &'static str) -> crate::PostgresBackendError {
+    return crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::DataInvalid, 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: 10
+// version: 11
 
 const APPLICATION_NAME: &str = "ksp-store";
 const MAX_CONNECTION_URI_BYTES: usize = 4_096;
@@ -332,6 +332,22 @@ impl PostgresBackend {
         return crate::probe_health(&self.pool).await;
     }
 
+    /// Reads one complete canonical RAW account state without exposing physical PostgreSQL row types.
+    pub async fn get_raw_account_state(
+        &self,
+        reference: &ksp_store_api::RawAccountStateReference,
+    ) -> std::result::Result<std::option::Option<ksp_store_api::RawAccountState>, crate::PostgresBackendError> {
+        return crate::get_raw_account_state(&self.pool, &self.network, reference).await;
+    }
+
+    /// Reads one persisted RAW account observation by producer-owned idempotence key.
+    pub async fn get_raw_account_observation(
+        &self,
+        observation_key: &ksp_store_api::RawObservationKey,
+    ) -> std::result::Result<std::option::Option<ksp_store_api::RawAccountObservation>, crate::PostgresBackendError> {
+        return crate::get_raw_account_observation(&self.pool, &self.network, observation_key).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: 16
+// version: 17
 
 #![warn(missing_docs)]
 #![deny(unreachable_pub)]
@@ -126,7 +126,7 @@ fn pre_003_fix_001_migration_engine_uses_split_schema_contract_and_binds_network
 }
 
 #[test]
-fn pre_003_v002_schema_is_complete_but_still_does_not_open_account_repository_or_runtime_dispatch() {
+fn pre_003_v002_schema_is_complete_without_account_trait_or_write_dispatch() {
     let crate_root = include_str!("../src/lib.rs");
     let migration = include_str!("../src/migration.rs");
     let schema = include_str!("../src/schema.rs");
@@ -148,9 +148,45 @@ fn pre_003_v002_schema_is_complete_but_still_does_not_open_account_repository_or
     assert!(slot_check.contains("slot >= 0 AND slot <= 18446744073709551615"));
     assert!(index.contains("ON ksp_raw_account_states (slot, pubkey, state_hash)"));
     assert!(!index.contains("WHERE"));
-    assert!(!crate_root.contains("mod raw_account;"));
     assert!(!runtime.contains("impl ksp_store_api::RawAccount"));
-    assert!(!std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("src/raw_account.rs").exists());
+    return;
+}
+
+#[test]
+fn pre_004_raw_account_read_sql_and_mapping_remain_backend_private_and_read_only() {
+    let crate_root = include_str!("../src/lib.rs");
+    let raw = include_str!("../src/raw_account.rs");
+    let runtime = include_str!("../src/runtime.rs");
+    assert!(crate_root.contains("mod raw_account;"));
+    assert!(!crate_root.contains("pub mod raw_account"));
+    for required in [
+        "GET_ACCOUNT_STATE_SQL",
+        "GET_ACCOUNT_OBSERVATION_SQL",
+        "slot::text AS slot_text",
+        "lamports::text AS lamports_text",
+        "rent_epoch::text AS rent_epoch_text",
+        "write_version::text AS write_version_text",
+        "RawAccountState::try_new",
+        "RawAccountObservation::new",
+        "PostgresBackendErrorKind::DataInvalid",
+        "PostgresBackendErrorKind::WrongNetwork",
+    ] {
+        assert!(raw.contains(required), "missing private RAW account read mapping contract: {required}");
+    }
+    for forbidden in ["INSERT INTO", "UPDATE ", "DELETE FROM", "ON CONFLICT", "FOR UPDATE", " OFFSET ", "list_raw_account_states"] {
+        assert!(!raw.contains(forbidden), "pre.004 account module contains write/pagination material: {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.004 opened RawAccount trait scope prematurely: {forbidden}");
+    }
+    for forbidden in ["std::env", "dotenv", "ksp_store_lib", "ksp_config_lib"] {
+        assert!(!raw.contains(forbidden), "RAW account module contains forbidden ownership material: {forbidden}");
+    }
     return;
 }
 

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-store-postgres-lib/tests/hardening_completeness.rs
-// version: 12
+// version: 13
 
 #![warn(missing_docs)]
 #![deny(unreachable_pub)]
@@ -109,7 +109,8 @@ fn assert_pre_io_rejection(connection_uri: &str, tls_mode: ksp_store_postgres_li
 #[test]
 fn pre_009_backend_modules_exports_and_manifest_dependencies_are_exact() {
     let crate_root = include_str!("../src/lib.rs");
-    for required in ["mod constants;", "mod error;", "mod health;", "mod migration;", "mod raw_transaction;", "mod runtime;", "mod schema;"] {
+    for required in ["mod constants;", "mod error;", "mod health;", "mod migration;", "mod raw_account;", "mod raw_transaction;", "mod runtime;", "mod schema;"]
+    {
         assert!(crate_root.contains(required), "missing PostgreSQL backend module: {required}");
     }
     assert!(!crate_root.contains("pub mod "));
@@ -172,6 +173,7 @@ fn pre_009_backend_error_bridge_cannot_retain_external_error_or_secret_text() {
         runtime,
         include_str!("../src/migration.rs"),
         include_str!("../src/health.rs"),
+        include_str!("../src/raw_account.rs"),
         include_str!("../src/raw_transaction.rs"),
         include_str!("../src/raw_transaction/cursor.rs"),
     ] {
@@ -192,11 +194,13 @@ fn pre_009_backend_has_no_env_bypass_reverse_facade_edge_or_raw_account_trait_im
 {}
 {}
 {}
+{}
 {}",
         include_str!("../src/error.rs"),
         include_str!("../src/health.rs"),
         include_str!("../src/lib.rs"),
         include_str!("../src/migration.rs"),
+        include_str!("../src/raw_account.rs"),
         include_str!("../src/raw_transaction.rs"),
         include_str!("../src/raw_transaction/cursor.rs"),
         include_str!("../src/runtime.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: 9
+// version: 10
 
 #![warn(missing_docs)]
 #![deny(unreachable_pub)]
@@ -75,6 +75,13 @@ fn pre_003_retention_compaction_error_code_matches_store_contract_value() {
     return;
 }
 
+#[test]
+fn pre_004_raw_account_read_bridge_uses_only_backend_independent_models() {
+    let _state = ksp_store_postgres_lib::PostgresBackend::get_raw_account_state;
+    let _observation = ksp_store_postgres_lib::PostgresBackend::get_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

@@ -0,0 +1,221 @@
+// file: crates/ksp-store-postgres-lib/unit_tests/raw_account.rs
+// version: 1
+
+fn network() -> ksp_store_api::RawNetworkId {
+    return match ksp_store_api::RawNetworkId::new("devnet") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => panic!("valid test network rejected: {error:?}"),
+    };
+}
+
+fn state_row() -> super::RawAccountStateDbRow {
+    return super::RawAccountStateDbRow {
+        data: vec![1, 2, 3, 4],
+        executable: true,
+        lamports_text: u64::MAX.to_string(),
+        owner: vec![2; 32],
+        pubkey: vec![1; 32],
+        rent_epoch_text: u64::MAX.to_string(),
+        slot_text: u64::MAX.to_string(),
+        state_hash: vec![3; 32],
+    };
+}
+
+fn observation_row() -> super::RawAccountObservationDbRow {
+    return super::RawAccountObservationDbRow {
+        account_pubkey: vec![1; 32],
+        account_slot_text: u64::MAX.to_string(),
+        account_state_hash: vec![3; 32],
+        acquisition_method: "account_subscribe".to_owned(),
+        capture_session_id: std::option::Option::Some("session_1".to_owned()),
+        commitment: std::option::Option::Some("confirmed".to_owned()),
+        endpoint_id: std::option::Option::Some("endpoint_1".to_owned()),
+        filter_id: std::option::Option::Some("filter_1".to_owned()),
+        is_startup: std::option::Option::Some(true),
+        observation_key: vec![4; 32],
+        observed_at_unix_millis: std::option::Option::Some(1_700_000_000_000),
+        origin: "live".to_owned(),
+        protocol: "yellowstone_grpc".to_owned(),
+        provider: "publicnode".to_owned(),
+        received_at_unix_millis: 1_700_000_000_001,
+        source_payload_hash: std::option::Option::Some(vec![5; 32]),
+        source_payload_size_bytes: std::option::Option::Some(67_108_864),
+        transaction_signature: std::option::Option::Some(vec![6; 64]),
+        write_version_text: std::option::Option::Some(u64::MAX.to_string()),
+    };
+}
+
+#[test]
+fn pre_004_account_state_row_round_trips_complete_u64_domain_and_bytes() {
+    let network = network();
+    let decoded = super::decode_raw_account_state_row(&network, state_row());
+    let state = match decoded {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => panic!("valid account state row rejected: {error:?}"),
+    };
+    assert_eq!(state.reference().network().as_str(), "devnet");
+    assert_eq!(state.reference().pubkey(), &ksp_store_api::Pubkey::new_from_array([1_u8; 32]));
+    assert_eq!(state.reference().slot(), u64::MAX);
+    assert_eq!(state.reference().state_hash(), ksp_store_api::RawContentHash::new([3_u8; 32]));
+    assert_eq!(state.lamports(), u64::MAX);
+    assert_eq!(state.owner(), &ksp_store_api::Pubkey::new_from_array([2_u8; 32]));
+    assert!(state.executable());
+    assert_eq!(state.rent_epoch(), u64::MAX);
+    assert_eq!(state.data(), &[1, 2, 3, 4]);
+    return;
+}
+
+#[test]
+fn pre_004_account_state_accepts_empty_data_and_rejects_oversized_data() {
+    let network = network();
+    let mut empty = state_row();
+    empty.data = std::vec::Vec::new();
+    let empty = super::decode_raw_account_state_row(&network, empty);
+    assert!(matches!(empty, std::result::Result::Ok(_)));
+    let mut oversized = state_row();
+    oversized.data = vec![0_u8; ksp_store_api::MAX_RAW_ACCOUNT_DATA_BYTES + 1];
+    let oversized = super::decode_raw_account_state_row(&network, oversized);
+    assert_eq!(oversized.err().map(|value| return value.kind()), std::option::Option::Some(crate::PostgresBackendErrorKind::DataInvalid));
+    return;
+}
+
+#[test]
+fn pre_004_account_state_rejects_hostile_width_and_decimal_rows_without_echo() {
+    let network = network();
+    let mut bad_pubkey = state_row();
+    bad_pubkey.pubkey = vec![0xA5; 31];
+    let error = super::decode_raw_account_state_row(&network, bad_pubkey).err();
+    assert_eq!(error.map(|value| return value.kind()), std::option::Option::Some(crate::PostgresBackendErrorKind::DataInvalid));
+    let mut bad_slot = state_row();
+    bad_slot.slot_text = "18446744073709551616-HOSTILE".to_owned();
+    let error = match super::decode_raw_account_state_row(&network, bad_slot) {
+        std::result::Result::Err(value) => value,
+        std::result::Result::Ok(_) => panic!("hostile slot unexpectedly decoded"),
+    };
+    assert_eq!(error.kind(), crate::PostgresBackendErrorKind::DataInvalid);
+    assert!(!std::format!("{error:?}").contains("HOSTILE"));
+    let mut bad_lamports = state_row();
+    bad_lamports.lamports_text = "-1".to_owned();
+    assert_eq!(
+        super::decode_raw_account_state_row(&network, bad_lamports).err().map(|value| return value.kind()),
+        std::option::Option::Some(crate::PostgresBackendErrorKind::DataInvalid)
+    );
+    return;
+}
+
+#[test]
+fn pre_004_account_observation_round_trips_complete_provenance_and_yellowstone_metadata() {
+    let network = network();
+    let decoded = super::decode_raw_account_observation_row(&network, observation_row());
+    let observation = match decoded {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => panic!("valid account observation row rejected: {error:?}"),
+    };
+    assert_eq!(observation.observation_key(), ksp_store_api::RawObservationKey::new([4_u8; 32]));
+    assert_eq!(observation.account().network().as_str(), "devnet");
+    assert_eq!(observation.account().pubkey(), &ksp_store_api::Pubkey::new_from_array([1_u8; 32]));
+    assert_eq!(observation.account().slot(), u64::MAX);
+    assert_eq!(observation.account().state_hash(), ksp_store_api::RawContentHash::new([3_u8; 32]));
+    assert_eq!(observation.provenance().provider().as_str(), "publicnode");
+    assert_eq!(observation.provenance().protocol().as_str(), "yellowstone_grpc");
+    assert_eq!(observation.provenance().acquisition_method().as_str(), "account_subscribe");
+    assert_eq!(observation.provenance().capture_session_id().map(|value| return value.as_str()), std::option::Option::Some("session_1"));
+    assert_eq!(observation.provenance().commitment().map(|value| return value.as_str()), std::option::Option::Some("confirmed"));
+    assert_eq!(observation.provenance().endpoint_id().map(|value| return value.as_str()), std::option::Option::Some("endpoint_1"));
+    assert_eq!(observation.provenance().filter_id().map(|value| return value.as_str()), std::option::Option::Some("filter_1"));
+    assert_eq!(observation.provenance().received_at().unix_millis(), 1_700_000_000_001);
+    assert_eq!(observation.provenance().observed_at().map(|value| return value.unix_millis()), std::option::Option::Some(1_700_000_000_000));
+    assert_eq!(observation.provenance().source_payload_hash(), std::option::Option::Some(ksp_store_api::RawContentHash::new([5_u8; 32])));
+    assert_eq!(observation.provenance().source_payload_size_bytes(), std::option::Option::Some(ksp_store_api::MAX_RAW_SOURCE_PAYLOAD_BYTES));
+    assert_eq!(observation.is_startup(), std::option::Option::Some(true));
+    assert_eq!(observation.transaction_signature(), std::option::Option::Some(ksp_store_api::RawTransactionSignature::new([6_u8; 64])));
+    assert_eq!(observation.write_version(), std::option::Option::Some(u64::MAX));
+    return;
+}
+
+#[test]
+fn pre_004_account_observation_keeps_optional_metadata_absent() {
+    let network = network();
+    let mut row = observation_row();
+    row.capture_session_id = std::option::Option::None;
+    row.commitment = std::option::Option::None;
+    row.endpoint_id = std::option::Option::None;
+    row.filter_id = std::option::Option::None;
+    row.observed_at_unix_millis = std::option::Option::None;
+    row.source_payload_hash = std::option::Option::None;
+    row.source_payload_size_bytes = std::option::Option::None;
+    row.is_startup = std::option::Option::None;
+    row.transaction_signature = std::option::Option::None;
+    row.write_version_text = std::option::Option::None;
+    let decoded = super::decode_raw_account_observation_row(&network, row);
+    let observation = match decoded {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => panic!("valid sparse account observation rejected: {error:?}"),
+    };
+    assert!(observation.provenance().capture_session_id().is_none());
+    assert!(observation.provenance().commitment().is_none());
+    assert!(observation.provenance().endpoint_id().is_none());
+    assert!(observation.provenance().filter_id().is_none());
+    assert!(observation.provenance().observed_at().is_none());
+    assert!(observation.provenance().source_payload_hash().is_none());
+    assert!(observation.provenance().source_payload_size_bytes().is_none());
+    assert!(observation.is_startup().is_none());
+    assert!(observation.transaction_signature().is_none());
+    assert!(observation.write_version().is_none());
+    return;
+}
+
+#[test]
+fn pre_004_account_observation_rejects_hostile_rows_without_echoing_values() {
+    let network = network();
+    let mut bad_signature = observation_row();
+    bad_signature.transaction_signature = std::option::Option::Some(vec![8; 63]);
+    assert_eq!(
+        super::decode_raw_account_observation_row(&network, bad_signature).err().map(|value| return value.kind()),
+        std::option::Option::Some(crate::PostgresBackendErrorKind::DataInvalid)
+    );
+    let mut bad_write_version = observation_row();
+    bad_write_version.write_version_text = std::option::Option::Some("18446744073709551616-HOSTILE".to_owned());
+    let error = match super::decode_raw_account_observation_row(&network, bad_write_version) {
+        std::result::Result::Err(value) => value,
+        std::result::Result::Ok(_) => panic!("hostile write version unexpectedly decoded"),
+    };
+    assert_eq!(error.kind(), crate::PostgresBackendErrorKind::DataInvalid);
+    assert!(!std::format!("{error:?}").contains("HOSTILE"));
+    let mut bad_origin = observation_row();
+    bad_origin.origin = "hostile-origin".to_owned();
+    assert_eq!(
+        super::decode_raw_account_observation_row(&network, bad_origin).err().map(|value| return value.kind()),
+        std::option::Option::Some(crate::PostgresBackendErrorKind::DataInvalid)
+    );
+    let mut bad_time = observation_row();
+    bad_time.observed_at_unix_millis = std::option::Option::Some(bad_time.received_at_unix_millis + 1);
+    assert_eq!(
+        super::decode_raw_account_observation_row(&network, bad_time).err().map(|value| return value.kind()),
+        std::option::Option::Some(crate::PostgresBackendErrorKind::DataInvalid)
+    );
+    return;
+}
+
+#[test]
+fn pre_004_account_wrong_network_guard_is_pre_io_and_static() {
+    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 reference = ksp_store_api::RawAccountStateReference::new(
+        foreign,
+        ksp_store_api::Pubkey::new_from_array([1_u8; 32]),
+        1,
+        ksp_store_api::RawContentHash::new([2_u8; 32]),
+    );
+    let error = super::ensure_network(&backend, &reference, "raw_account_state_network").err();
+    let error = match error {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => panic!("wrong-network account reference unexpectedly accepted"),
+    };
+    assert_eq!(error.kind(), crate::PostgresBackendErrorKind::WrongNetwork);
+    assert_eq!(error.phase(), "raw_account_state_network");
+    return;
+}