v0.3.4-pre.007

2026-08-30 22:47:33 +02:00
parent fccb7d876c
commit 902b5fba99
12 changed files with 904 additions and 16 deletions
--- a/crates/ksp-store-postgres-lib/src/raw_account.rs
+++ b/crates/ksp-store-postgres-lib/src/raw_account.rs
@@ -1,15 +1,27 @@
 // file: crates/ksp-store-postgres-lib/src/raw_account.rs
-// version: 4
+// version: 5
+
+pub(crate) mod cursor;

 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 LIST_ACCOUNT_STATES_ASC_SQL: &str = "SELECT pubkey, slot::text AS slot_text, state_hash FROM ksp_raw_account_states WHERE ($1::TEXT IS NULL OR slot >= $1::TEXT::NUMERIC) AND ($2::TEXT IS NULL OR slot <= $2::TEXT::NUMERIC) AND ($3::TEXT IS NULL OR (slot, pubkey, state_hash) > ($3::TEXT::NUMERIC, $4::BYTEA, $5::BYTEA)) ORDER BY slot ASC, pubkey ASC, state_hash ASC LIMIT $6";
+const LIST_ACCOUNT_STATES_BY_PUBKEY_ASC_SQL: &str = "SELECT pubkey, slot::text AS slot_text, state_hash FROM ksp_raw_account_states WHERE pubkey = $1 AND ($2::TEXT IS NULL OR slot >= $2::TEXT::NUMERIC) AND ($3::TEXT IS NULL OR slot <= $3::TEXT::NUMERIC) AND ($4::TEXT IS NULL OR (slot, pubkey, state_hash) > ($4::TEXT::NUMERIC, $5::BYTEA, $6::BYTEA)) ORDER BY slot ASC, pubkey ASC, state_hash ASC LIMIT $7";
+const LIST_ACCOUNT_STATES_BY_PUBKEY_DESC_SQL: &str = "SELECT pubkey, slot::text AS slot_text, state_hash FROM ksp_raw_account_states WHERE pubkey = $1 AND ($2::TEXT IS NULL OR slot >= $2::TEXT::NUMERIC) AND ($3::TEXT IS NULL OR slot <= $3::TEXT::NUMERIC) AND ($4::TEXT IS NULL OR (slot, pubkey, state_hash) < ($4::TEXT::NUMERIC, $5::BYTEA, $6::BYTEA)) ORDER BY slot DESC, pubkey DESC, state_hash DESC LIMIT $7";
+const LIST_ACCOUNT_STATES_DESC_SQL: &str = "SELECT pubkey, slot::text AS slot_text, state_hash FROM ksp_raw_account_states WHERE ($1::TEXT IS NULL OR slot >= $1::TEXT::NUMERIC) AND ($2::TEXT IS NULL OR slot <= $2::TEXT::NUMERIC) AND ($3::TEXT IS NULL OR (slot, pubkey, state_hash) < ($3::TEXT::NUMERIC, $4::BYTEA, $5::BYTEA)) ORDER BY slot DESC, pubkey DESC, state_hash DESC LIMIT $6";
 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 RawAccountListDbRow {
+    pubkey: std::vec::Vec<u8>,
+    slot_text: std::string::String,
+    state_hash: std::vec::Vec<u8>,
+}
+
 struct RawAccountObservationDbRow {
    account_pubkey: std::vec::Vec<u8>,
    account_slot_text: std::string::String,
@@ -93,6 +105,104 @@ pub(crate) async fn get_raw_account_state(
    return std::result::Result::Ok(std::option::Option::Some(decoded));
 }

+/// Lists deterministic canonical RAW account-state references using PostgreSQL keyset pagination.
+pub(crate) async fn list_raw_account_states(
+    pool: &deadpool_postgres::Pool,
+    network: &ksp_store_api::RawNetworkId,
+    query: &ksp_store_api::RawAccountStateQuery,
+) -> std::result::Result<ksp_store_api::RawPage<ksp_store_api::RawAccountStateReference>, crate::PostgresBackendError> {
+    if query.network() != network {
+        return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::WrongNetwork, "raw_account_list_network"));
+    }
+    let (requested_usize, sql_limit) = match crate::raw_account_physical_page_limit(query.page().limit().get()) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let decoded_cursor = match query.page().cursor() {
+        std::option::Option::Some(value) => match crate::decode_raw_account_cursor(query, value) {
+            std::result::Result::Ok(decoded) => std::option::Option::Some(decoded),
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        },
+        std::option::Option::None => std::option::Option::None,
+    };
+    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 slots = query.slots();
+    let start_text = slots.start_inclusive().map(|value| return value.to_string());
+    let end_text = slots.end_inclusive().map(|value| return value.to_string());
+    let cursor_slot_text = decoded_cursor.as_ref().map(|value| return value.last_slot.to_string());
+    let cursor_pubkey = decoded_cursor.as_ref().map(|value| return value.last_pubkey.to_vec());
+    let cursor_state_hash = decoded_cursor.as_ref().map(|value| return value.last_state_hash.to_vec());
+    let rows_result = match query.pubkey() {
+        std::option::Option::Some(pubkey) => {
+            let pubkey_bytes: &[u8] = pubkey.as_ref();
+            let sql = match query.direction() {
+                ksp_store_api::RawSortDirection::Ascending => LIST_ACCOUNT_STATES_BY_PUBKEY_ASC_SQL,
+                ksp_store_api::RawSortDirection::Descending => LIST_ACCOUNT_STATES_BY_PUBKEY_DESC_SQL,
+                _ => {
+                    return std::result::Result::Err(crate::PostgresBackendError::new(
+                        crate::PostgresBackendErrorKind::QueryInvalid,
+                        "raw_account_list_direction",
+                    ));
+                },
+            };
+            client.query(sql, &[&pubkey_bytes, &start_text, &end_text, &cursor_slot_text, &cursor_pubkey, &cursor_state_hash, &sql_limit]).await
+        },
+        std::option::Option::None => {
+            let sql = match query.direction() {
+                ksp_store_api::RawSortDirection::Ascending => LIST_ACCOUNT_STATES_ASC_SQL,
+                ksp_store_api::RawSortDirection::Descending => LIST_ACCOUNT_STATES_DESC_SQL,
+                _ => {
+                    return std::result::Result::Err(crate::PostgresBackendError::new(
+                        crate::PostgresBackendErrorKind::QueryInvalid,
+                        "raw_account_list_direction",
+                    ));
+                },
+            };
+            client.query(sql, &[&start_text, &end_text, &cursor_slot_text, &cursor_pubkey, &cursor_state_hash, &sql_limit]).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_list_query"));
+        },
+    };
+    let mut decoded = std::vec::Vec::with_capacity(rows.len());
+    for row in rows {
+        let physical = match raw_account_list_db_row(&row) {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        let item = match decode_raw_account_list_row(network, physical) {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        decoded.push(item);
+    }
+    let has_more = decoded.len() > requested_usize;
+    if has_more {
+        decoded.truncate(requested_usize);
+    }
+    let next_cursor = if has_more {
+        let last = match decoded.last() {
+            std::option::Option::Some(value) => value,
+            std::option::Option::None => return std::result::Result::Err(data_invalid("raw_account_list_page")),
+        };
+        match crate::encode_raw_account_cursor(query, last.0, last.1.pubkey(), &last.1.state_hash()) {
+            std::result::Result::Ok(value) => std::option::Option::Some(value),
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        }
+    } else {
+        std::option::Option::None
+    };
+    let items = decoded.into_iter().map(|value| return value.1).collect();
+    return std::result::Result::Ok(ksp_store_api::RawPage::new(items, next_cursor));
+}
+
 /// Reads one persisted RAW account observation by producer-owned idempotence key.
 pub(crate) async fn get_raw_account_observation(
    pool: &deadpool_postgres::Pool,
@@ -235,6 +345,42 @@ pub(crate) async fn record_raw_account_observation(
    return std::result::Result::Ok(outcome);
 }

+fn raw_account_list_db_row(row: &tokio_postgres::Row) -> std::result::Result<RawAccountListDbRow, crate::PostgresBackendError> {
+    let pubkey = match row.try_get::<_, std::vec::Vec<u8>>("pubkey") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_account_list_decode")),
+    };
+    let slot_text = match row.try_get::<_, std::string::String>("slot_text") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_account_list_decode")),
+    };
+    let state_hash = match row.try_get::<_, std::vec::Vec<u8>>("state_hash") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_account_list_decode")),
+    };
+    return std::result::Result::Ok(RawAccountListDbRow { pubkey, slot_text, state_hash });
+}
+
+fn decode_raw_account_list_row(
+    network: &ksp_store_api::RawNetworkId,
+    row: RawAccountListDbRow,
+) -> std::result::Result<(u64, ksp_store_api::RawAccountStateReference), crate::PostgresBackendError> {
+    let pubkey = match fixed_bytes::<32>(row.pubkey, "raw_account_list_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_list_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_list_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 reference = ksp_store_api::RawAccountStateReference::new(network.clone(), pubkey, slot, state_hash);
+    return std::result::Result::Ok((slot, reference));
+}
+
 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,