1217 lines
62 KiB
Rust
1217 lines
62 KiB
Rust
// file: crates/ksp-store-postgres-lib/tests/postgres_raw_account_live.rs
|
|
// version: 1
|
|
|
|
#![warn(missing_docs)]
|
|
#![deny(unreachable_pub)]
|
|
#![forbid(unsafe_code)]
|
|
|
|
//! Opt-in real PostgreSQL proof for the complete RawAccountState vertical slice.
|
|
//!
|
|
//! The test reads one dedicated PostgreSQL URI from stdin, refuses to start
|
|
//! when any managed KSP Store table already exists, never prints the URI, and
|
|
//! drops only the isolated schema it proved absent before the run.
|
|
|
|
const LIVE_ACCOUNT_INDEX_EXISTS_SQL: &str = r#"SELECT EXISTS (
|
|
SELECT 1 FROM pg_indexes
|
|
WHERE schemaname = current_schema()
|
|
AND indexname = 'ix_ksp_raw_account_states_slot_pubkey_state_hash'
|
|
)"#;
|
|
const LIVE_ACCOUNT_STATE_EXISTS_SQL: &str =
|
|
"SELECT EXISTS (SELECT 1 FROM ksp_raw_account_states WHERE pubkey = $1 AND slot = $2::TEXT::NUMERIC AND state_hash = $3)";
|
|
const LIVE_CANCEL_WAIT: std::time::Duration = std::time::Duration::from_millis(300);
|
|
const LIVE_LOCK_ACCOUNT_OBSERVATION_SQL: &str = "SELECT observation_key FROM ksp_raw_account_observations WHERE observation_key = $1 FOR UPDATE";
|
|
const LIVE_MANAGED_SCHEMA_DROP_SQL: &str = r#"DROP TABLE IF EXISTS ksp_raw_account_observations;
|
|
DROP TABLE IF EXISTS ksp_raw_account_states;
|
|
DROP TABLE IF EXISTS ksp_raw_transaction_observations;
|
|
DROP TABLE IF EXISTS ksp_raw_transaction_archive_payloads;
|
|
DROP TABLE IF EXISTS ksp_raw_transactions;
|
|
DROP TABLE IF EXISTS ksp_store_identity;
|
|
DROP TABLE IF EXISTS ksp_store_schema_migrations;"#;
|
|
const LIVE_MANAGED_SCHEMA_EXISTS_SQL: &str = r#"SELECT EXISTS (
|
|
SELECT 1 FROM information_schema.tables
|
|
WHERE table_schema = current_schema()
|
|
AND table_name IN (
|
|
'ksp_store_schema_migrations',
|
|
'ksp_store_identity',
|
|
'ksp_raw_transactions',
|
|
'ksp_raw_transaction_observations',
|
|
'ksp_raw_transaction_archive_payloads',
|
|
'ksp_raw_account_states',
|
|
'ksp_raw_account_observations'
|
|
)
|
|
AND table_type = 'BASE TABLE'
|
|
)"#;
|
|
const LIVE_MAX_URI_BYTES: usize = 4_096;
|
|
const LIVE_TRANSACTION_EXISTS_SQL: &str = "SELECT EXISTS (SELECT 1 FROM ksp_raw_transactions WHERE signature = $1)";
|
|
|
|
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
|
struct LiveFailure {
|
|
phase: &'static str,
|
|
}
|
|
|
|
impl LiveFailure {
|
|
const fn new(phase: &'static str) -> Self {
|
|
return Self { phase };
|
|
}
|
|
|
|
const fn phase(&self) -> &'static str {
|
|
return self.phase;
|
|
}
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
enum LiveAccountPersistResult {
|
|
BackendError(ksp_store_postgres_lib::PostgresBackendErrorKind),
|
|
Outcome(ksp_store_api::RawAcquisitionWriteOutcome),
|
|
}
|
|
|
|
#[test]
|
|
#[ignore = "opt-in real PostgreSQL RawAccountState proof; reads one dedicated URI from stdin"]
|
|
fn pre_009_real_postgres_raw_account_vertical_slice_is_atomic_concurrent_and_recoverable() {
|
|
eprintln!("KSP Store RawAccountState live proof: reading one dedicated URI from stdin without echoing it from the test.");
|
|
let uri_result = read_uri_from_stdin();
|
|
let uri = match uri_result {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => panic!("PostgreSQL RawAccountState live input rejected at phase {}", error.phase()),
|
|
};
|
|
let runtime_result = tokio::runtime::Builder::new_current_thread().enable_all().build();
|
|
let runtime = match runtime_result {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => panic!("PostgreSQL RawAccountState live runtime could not be constructed"),
|
|
};
|
|
let outcome = runtime.block_on(run_live_test(uri.as_str()));
|
|
if let std::result::Result::Err(error) = outcome {
|
|
panic!("PostgreSQL RawAccountState live proof failed at safe phase {}", error.phase());
|
|
}
|
|
return;
|
|
}
|
|
|
|
fn read_uri_from_stdin() -> std::result::Result<std::string::String, LiveFailure> {
|
|
let mut input = std::string::String::new();
|
|
let read_result = std::io::stdin().read_line(&mut input);
|
|
match read_result {
|
|
std::result::Result::Ok(0) | std::result::Result::Err(_) => return std::result::Result::Err(LiveFailure::new("stdin_read")),
|
|
std::result::Result::Ok(_) => {},
|
|
}
|
|
let uri = input.trim().to_owned();
|
|
if uri.is_empty() || uri.len() > LIVE_MAX_URI_BYTES {
|
|
return std::result::Result::Err(LiveFailure::new("stdin_uri"));
|
|
}
|
|
return std::result::Result::Ok(uri);
|
|
}
|
|
|
|
async fn run_live_test(uri: &str) -> std::result::Result<(), LiveFailure> {
|
|
let admin_result = connect_admin(uri).await;
|
|
let mut admin = match admin_result {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let preexisting_result = managed_schema_exists(&admin).await;
|
|
let preexisting = match preexisting_result {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
if preexisting {
|
|
return std::result::Result::Err(LiveFailure::new("managed_schema_preexisting_refusal"));
|
|
}
|
|
let major_result = postgres_major(&admin).await;
|
|
let major = match major_result {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
if major < 15 {
|
|
return std::result::Result::Err(LiveFailure::new("postgres_major_unsupported"));
|
|
}
|
|
eprintln!("KSP Store RawAccountState live proof: server major {major}");
|
|
let mut owns_schema = false;
|
|
let scenario = run_raw_account_scenario(&mut admin, uri, &mut owns_schema).await;
|
|
let cleanup = if owns_schema { drop_managed_schema(&admin).await } else { std::result::Result::Ok(()) };
|
|
if let std::result::Result::Err(error) = cleanup {
|
|
return std::result::Result::Err(error);
|
|
}
|
|
if let std::result::Result::Err(error) = scenario {
|
|
return std::result::Result::Err(error);
|
|
}
|
|
let remains_result = managed_schema_exists(&admin).await;
|
|
match remains_result {
|
|
std::result::Result::Ok(false) => {},
|
|
std::result::Result::Ok(true) => return std::result::Result::Err(LiveFailure::new("cleanup_verification")),
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
}
|
|
return std::result::Result::Ok(());
|
|
}
|
|
|
|
async fn run_raw_account_scenario(admin: &mut tokio_postgres::Client, uri: &str, owns_schema: &mut bool) -> std::result::Result<(), LiveFailure> {
|
|
let initial_result = open_backend(uri, "devnet", true, true).await;
|
|
let initial = match initial_result {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
*owns_schema = true;
|
|
let initial_health = initial.health().await;
|
|
if !initial_health.is_ready() || initial_health.migration_version() != std::option::Option::Some(2) || initial_health.pending_migration_count() != 0 {
|
|
return std::result::Result::Err(LiveFailure::new("initial_health"));
|
|
}
|
|
let wrong_network_result = open_backend_result(uri, "testnet", true, true).await;
|
|
match wrong_network_result {
|
|
std::result::Result::Ok(std::result::Result::Err(error)) if error.kind() == ksp_store_postgres_lib::PostgresBackendErrorKind::MigrationMismatch => {},
|
|
std::result::Result::Ok(std::result::Result::Ok(backend)) => {
|
|
let _ = close_backend(backend).await;
|
|
return std::result::Result::Err(LiveFailure::new("wrong_network_accepted"));
|
|
},
|
|
_ => return std::result::Result::Err(LiveFailure::new("wrong_network_classification")),
|
|
}
|
|
let initial_close = close_backend(initial).await;
|
|
if let std::result::Result::Err(error) = initial_close {
|
|
return std::result::Result::Err(error);
|
|
}
|
|
let schema_result = prove_schema_update_policy(admin, uri).await;
|
|
if let std::result::Result::Err(error) = schema_result {
|
|
return std::result::Result::Err(error);
|
|
}
|
|
let backend_result = open_backend(uri, "devnet", true, true).await;
|
|
let backend = match backend_result {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
for proof in [
|
|
prove_state_extremes(&backend).await,
|
|
prove_atomic_idempotence_and_conflicts(&backend).await,
|
|
prove_additional_observations(&backend).await,
|
|
prove_pagination_and_cursors(&backend).await,
|
|
prove_cross_family_coexistence(&backend, admin).await,
|
|
] {
|
|
if let std::result::Result::Err(error) = proof {
|
|
return std::result::Result::Err(error);
|
|
}
|
|
}
|
|
let backend_close = close_backend(backend).await;
|
|
if let std::result::Result::Err(error) = backend_close {
|
|
return std::result::Result::Err(error);
|
|
}
|
|
let identical_result = prove_concurrent_identical_insert(uri).await;
|
|
if let std::result::Result::Err(error) = identical_result {
|
|
return std::result::Result::Err(error);
|
|
}
|
|
let divergent_result = prove_concurrent_divergent_insert(uri).await;
|
|
if let std::result::Result::Err(error) = divergent_result {
|
|
return std::result::Result::Err(error);
|
|
}
|
|
let cancellation_result = prove_cancellation_rollback(admin, uri).await;
|
|
if let std::result::Result::Err(error) = cancellation_result {
|
|
return std::result::Result::Err(error);
|
|
}
|
|
let reopened_result = open_backend(uri, "devnet", true, true).await;
|
|
let reopened = match reopened_result {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let reopened_health = reopened.health().await;
|
|
if !reopened_health.is_ready() || reopened_health.migration_version() != std::option::Option::Some(2) {
|
|
return std::result::Result::Err(LiveFailure::new("reopen_health"));
|
|
}
|
|
let reference = match account_reference(10, 100, 10) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
match reopened.get_raw_account_state(&reference).await {
|
|
std::result::Result::Ok(std::option::Option::Some(value)) if account_state_matches(&value, 10, 100, 10, 1_000, 11, false, 12, 4, 10) => {},
|
|
_ => return std::result::Result::Err(LiveFailure::new("reopen_account_read")),
|
|
}
|
|
let transaction_reference = match transaction_reference(90) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
match reopened.get_raw_transaction(&transaction_reference).await {
|
|
std::result::Result::Ok(std::option::Option::Some(_)) => {},
|
|
_ => return std::result::Result::Err(LiveFailure::new("reopen_transaction_read")),
|
|
}
|
|
return close_backend(reopened).await;
|
|
}
|
|
|
|
async fn prove_schema_update_policy(admin: &tokio_postgres::Client, uri: &str) -> std::result::Result<(), LiveFailure> {
|
|
let drop_result = admin.execute("DROP INDEX ix_ksp_raw_account_states_slot_pubkey_state_hash", &[]).await;
|
|
match drop_result {
|
|
std::result::Result::Ok(_) => {},
|
|
std::result::Result::Err(_) => return std::result::Result::Err(LiveFailure::new("schema_drop_account_index")),
|
|
}
|
|
let blocked_result = open_backend_result(uri, "devnet", true, false).await;
|
|
match blocked_result {
|
|
std::result::Result::Ok(std::result::Result::Err(error)) if error.kind() == ksp_store_postgres_lib::PostgresBackendErrorKind::MigrationMismatch => {},
|
|
std::result::Result::Ok(std::result::Result::Ok(backend)) => {
|
|
let _ = close_backend(backend).await;
|
|
return std::result::Result::Err(LiveFailure::new("schema_autoupdate_disabled_accepted"));
|
|
},
|
|
_ => return std::result::Result::Err(LiveFailure::new("schema_autoupdate_disabled_classification")),
|
|
}
|
|
let repaired_result = open_backend(uri, "devnet", true, true).await;
|
|
let repaired = match repaired_result {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let index_result = account_index_exists(admin).await;
|
|
match index_result {
|
|
std::result::Result::Ok(true) => {},
|
|
std::result::Result::Ok(false) => return std::result::Result::Err(LiveFailure::new("schema_account_index_not_repaired")),
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
}
|
|
return close_backend(repaired).await;
|
|
}
|
|
|
|
async fn prove_state_extremes(backend: &ksp_store_postgres_lib::PostgresBackend) -> std::result::Result<(), LiveFailure> {
|
|
let max_state = match account_state(1, u64::MAX, 1, u64::MAX, 2, true, u64::MAX, 0, 1) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let max_observation = match account_observation(1, 1, u64::MAX, 1, 1, true) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let max_persist = backend.persist_raw_account_acquisition(max_state, max_observation).await;
|
|
match max_persist {
|
|
std::result::Result::Ok(value)
|
|
if value.entity() == ksp_store_api::RawEntityWriteOutcome::Inserted
|
|
&& value.observation() == ksp_store_api::RawObservationWriteOutcome::Inserted => {},
|
|
_ => return std::result::Result::Err(LiveFailure::new("u64_max_insert")),
|
|
}
|
|
let max_reference = match account_reference(1, u64::MAX, 1) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
match backend.get_raw_account_state(&max_reference).await {
|
|
std::result::Result::Ok(std::option::Option::Some(value)) if account_state_matches(&value, 1, u64::MAX, 1, u64::MAX, 2, true, u64::MAX, 0, 1) => {},
|
|
_ => return std::result::Result::Err(LiveFailure::new("u64_max_read")),
|
|
}
|
|
let max_observation_key = ksp_store_api::RawObservationKey::new([1; 32]);
|
|
match backend.get_raw_account_observation(&max_observation_key).await {
|
|
std::result::Result::Ok(std::option::Option::Some(value))
|
|
if value.is_startup() == std::option::Option::Some(true)
|
|
&& value.write_version() == std::option::Option::Some(u64::MAX)
|
|
&& value.transaction_signature() == std::option::Option::Some(ksp_store_api::RawTransactionSignature::new([21; 64])) => {},
|
|
_ => return std::result::Result::Err(LiveFailure::new("u64_max_observation_read")),
|
|
}
|
|
let full_state = match account_state(2, 2, 2, 2, 3, false, 2, ksp_store_api::MAX_RAW_ACCOUNT_DATA_BYTES, 2) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let full_observation = match account_observation(2, 2, 2, 2, 2, false) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let full_persist = backend.persist_raw_account_acquisition(full_state, full_observation).await;
|
|
match full_persist {
|
|
std::result::Result::Ok(value) if value.entity() == ksp_store_api::RawEntityWriteOutcome::Inserted => {},
|
|
_ => return std::result::Result::Err(LiveFailure::new("max_data_insert")),
|
|
}
|
|
let full_reference = match account_reference(2, 2, 2) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
match backend.get_raw_account_state(&full_reference).await {
|
|
std::result::Result::Ok(std::option::Option::Some(value))
|
|
if account_state_matches(&value, 2, 2, 2, 2, 3, false, 2, ksp_store_api::MAX_RAW_ACCOUNT_DATA_BYTES, 2) => {},
|
|
_ => return std::result::Result::Err(LiveFailure::new("max_data_read")),
|
|
}
|
|
return std::result::Result::Ok(());
|
|
}
|
|
|
|
async fn prove_atomic_idempotence_and_conflicts(backend: &ksp_store_postgres_lib::PostgresBackend) -> std::result::Result<(), LiveFailure> {
|
|
let state = match account_state(10, 100, 10, 1_000, 11, false, 12, 4, 10) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let observation = match account_observation(10, 10, 100, 10, 10, true) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let first = backend.persist_raw_account_acquisition(state, observation).await;
|
|
match first {
|
|
std::result::Result::Ok(value)
|
|
if value.entity() == ksp_store_api::RawEntityWriteOutcome::Inserted
|
|
&& value.observation() == ksp_store_api::RawObservationWriteOutcome::Inserted => {},
|
|
_ => return std::result::Result::Err(LiveFailure::new("atomic_insert")),
|
|
}
|
|
let identical_state = match account_state(10, 100, 10, 1_000, 11, false, 12, 4, 10) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let identical_observation = match account_observation(10, 10, 100, 10, 10, true) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let second = backend.persist_raw_account_acquisition(identical_state, identical_observation).await;
|
|
match second {
|
|
std::result::Result::Ok(value)
|
|
if value.entity() == ksp_store_api::RawEntityWriteOutcome::AlreadyPresent
|
|
&& value.observation() == ksp_store_api::RawObservationWriteOutcome::AlreadyPresent => {},
|
|
_ => return std::result::Result::Err(LiveFailure::new("atomic_idempotent")),
|
|
}
|
|
let divergent_state = match account_state(10, 100, 10, 1_001, 11, false, 12, 4, 99) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let divergent_observation = match account_observation(11, 10, 100, 10, 11, false) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
match backend.persist_raw_account_acquisition(divergent_state, divergent_observation).await {
|
|
std::result::Result::Err(error) if error.kind() == ksp_store_postgres_lib::PostgresBackendErrorKind::Conflict => {},
|
|
_ => return std::result::Result::Err(LiveFailure::new("state_conflict")),
|
|
}
|
|
let same_slot_first = match account_state(20, 120, 20, 2_000, 21, false, 22, 3, 20) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let same_slot_first_observation = match account_observation(20, 20, 120, 20, 20, false) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let same_slot_second = match account_state(20, 120, 21, 2_001, 21, false, 22, 3, 21) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let same_slot_second_observation = match account_observation(21, 20, 120, 21, 21, false) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
for result in [
|
|
backend.persist_raw_account_acquisition(same_slot_first, same_slot_first_observation).await,
|
|
backend.persist_raw_account_acquisition(same_slot_second, same_slot_second_observation).await,
|
|
] {
|
|
match result {
|
|
std::result::Result::Ok(value) if value.entity() == ksp_store_api::RawEntityWriteOutcome::Inserted => {},
|
|
_ => return std::result::Result::Err(LiveFailure::new("same_slot_hash_insert")),
|
|
}
|
|
}
|
|
let collision_state = match account_state(30, 130, 30, 3_000, 31, false, 32, 3, 30) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let collision_observation = match account_observation(10, 30, 130, 30, 30, false) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
match backend.persist_raw_account_acquisition(collision_state, collision_observation).await {
|
|
std::result::Result::Err(error) if error.kind() == ksp_store_postgres_lib::PostgresBackendErrorKind::Conflict => {},
|
|
_ => return std::result::Result::Err(LiveFailure::new("observation_collision_conflict")),
|
|
}
|
|
let collision_reference = match account_reference(30, 130, 30) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
match backend.get_raw_account_state(&collision_reference).await {
|
|
std::result::Result::Ok(std::option::Option::None) => {},
|
|
_ => return std::result::Result::Err(LiveFailure::new("observation_collision_rollback")),
|
|
}
|
|
return std::result::Result::Ok(());
|
|
}
|
|
|
|
async fn prove_additional_observations(backend: &ksp_store_postgres_lib::PostgresBackend) -> std::result::Result<(), LiveFailure> {
|
|
let additional = match account_observation(41, 10, 100, 10, 41, true) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let expected = additional.clone();
|
|
let first = backend.record_raw_account_observation(additional).await;
|
|
if first != std::result::Result::Ok(ksp_store_api::RawObservationWriteOutcome::Inserted) {
|
|
return std::result::Result::Err(LiveFailure::new("additional_observation_insert"));
|
|
}
|
|
let identical = match account_observation(41, 10, 100, 10, 41, true) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let second = backend.record_raw_account_observation(identical).await;
|
|
if second != std::result::Result::Ok(ksp_store_api::RawObservationWriteOutcome::AlreadyPresent) {
|
|
return std::result::Result::Err(LiveFailure::new("additional_observation_idempotent"));
|
|
}
|
|
let key = ksp_store_api::RawObservationKey::new([41; 32]);
|
|
match backend.get_raw_account_observation(&key).await {
|
|
std::result::Result::Ok(std::option::Option::Some(value)) if value == expected => {},
|
|
_ => return std::result::Result::Err(LiveFailure::new("additional_observation_round_trip")),
|
|
}
|
|
let divergent = match account_observation(41, 10, 100, 10, 42, false) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
match backend.record_raw_account_observation(divergent).await {
|
|
std::result::Result::Err(error) if error.kind() == ksp_store_postgres_lib::PostgresBackendErrorKind::Conflict => {},
|
|
_ => return std::result::Result::Err(LiveFailure::new("additional_observation_conflict")),
|
|
}
|
|
let missing = match account_observation(42, 99, 999, 99, 42, false) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
match backend.record_raw_account_observation(missing).await {
|
|
std::result::Result::Err(error) if error.kind() == ksp_store_postgres_lib::PostgresBackendErrorKind::ReferenceNotFound => {},
|
|
_ => return std::result::Result::Err(LiveFailure::new("missing_reference")),
|
|
}
|
|
return std::result::Result::Ok(());
|
|
}
|
|
|
|
async fn prove_pagination_and_cursors(backend: &ksp_store_postgres_lib::PostgresBackend) -> std::result::Result<(), LiveFailure> {
|
|
let rows = [(50_u8, 200_u64, 50_u8), (50_u8, 200_u64, 51_u8), (50_u8, 201_u64, 52_u8), (51_u8, 201_u64, 53_u8), (52_u8, 202_u64, 54_u8)];
|
|
for (index, (pubkey_seed, slot, hash_seed)) in rows.into_iter().enumerate() {
|
|
let state = match account_state(pubkey_seed, slot, hash_seed, 5_000_u64 + slot, 60, false, 61, 2, hash_seed) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let observation_seed = 50_u8.wrapping_add(index as u8);
|
|
let observation = match account_observation(observation_seed, pubkey_seed, slot, hash_seed, observation_seed, false) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
match backend.persist_raw_account_acquisition(state, observation).await {
|
|
std::result::Result::Ok(value) if value.entity() == ksp_store_api::RawEntityWriteOutcome::Inserted => {},
|
|
_ => return std::result::Result::Err(LiveFailure::new("pagination_seed")),
|
|
}
|
|
}
|
|
let ascending = match collect_account_pages(backend, std::option::Option::None, 200, 202, ksp_store_api::RawSortDirection::Ascending).await {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let expected_ascending = [(200, 50, 50), (200, 50, 51), (201, 50, 52), (201, 51, 53), (202, 52, 54)];
|
|
if ascending.as_slice() != expected_ascending.as_slice() {
|
|
return std::result::Result::Err(LiveFailure::new("pagination_ascending"));
|
|
}
|
|
let descending = match collect_account_pages(backend, std::option::Option::None, 200, 202, ksp_store_api::RawSortDirection::Descending).await {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let expected_descending = [(202, 52, 54), (201, 51, 53), (201, 50, 52), (200, 50, 51), (200, 50, 50)];
|
|
if descending.as_slice() != expected_descending.as_slice() {
|
|
return std::result::Result::Err(LiveFailure::new("pagination_descending"));
|
|
}
|
|
let filtered = match collect_account_pages(
|
|
backend,
|
|
std::option::Option::Some(ksp_store_api::Pubkey::new_from_array([50; 32])),
|
|
200,
|
|
201,
|
|
ksp_store_api::RawSortDirection::Ascending,
|
|
)
|
|
.await
|
|
{
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let expected_filtered = [(200, 50, 50), (200, 50, 51), (201, 50, 52)];
|
|
if filtered.as_slice() != expected_filtered.as_slice() {
|
|
return std::result::Result::Err(LiveFailure::new("pagination_pubkey_filter"));
|
|
}
|
|
let first_query = match account_query(std::option::Option::None, 200, 202, ksp_store_api::RawSortDirection::Ascending, std::option::Option::None) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let first_page = match backend.list_raw_account_states(&first_query).await {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::result::Result::Err(LiveFailure::new("cursor_first_page")),
|
|
};
|
|
let cursor = match first_page.next_cursor() {
|
|
std::option::Option::Some(value) => value.clone(),
|
|
std::option::Option::None => return std::result::Result::Err(LiveFailure::new("cursor_missing")),
|
|
};
|
|
let other_direction =
|
|
match account_query(std::option::Option::None, 200, 202, ksp_store_api::RawSortDirection::Descending, std::option::Option::Some(cursor.clone())) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
match backend.list_raw_account_states(&other_direction).await {
|
|
std::result::Result::Err(error) if error.kind() == ksp_store_postgres_lib::PostgresBackendErrorKind::QueryInvalid => {},
|
|
_ => return std::result::Result::Err(LiveFailure::new("cursor_cross_direction")),
|
|
}
|
|
let mut hostile_bytes = cursor.as_bytes().to_vec();
|
|
let last_index = match hostile_bytes.len().checked_sub(1) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return std::result::Result::Err(LiveFailure::new("cursor_hostile_index")),
|
|
};
|
|
hostile_bytes[last_index] ^= 0xff;
|
|
let hostile_cursor = match ksp_store_api::RawPageCursor::try_new(hostile_bytes.into_boxed_slice()) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::result::Result::Err(LiveFailure::new("cursor_hostile_model")),
|
|
};
|
|
let hostile_query =
|
|
match account_query(std::option::Option::None, 200, 202, ksp_store_api::RawSortDirection::Ascending, std::option::Option::Some(hostile_cursor)) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
match backend.list_raw_account_states(&hostile_query).await {
|
|
std::result::Result::Err(error) if error.kind() == ksp_store_postgres_lib::PostgresBackendErrorKind::QueryInvalid => {},
|
|
_ => return std::result::Result::Err(LiveFailure::new("cursor_hostile_digest")),
|
|
}
|
|
let mut transaction_family_bytes = vec![0_u8; 109];
|
|
transaction_family_bytes[0..4].copy_from_slice(b"KSPT");
|
|
transaction_family_bytes[4] = 1;
|
|
let transaction_family_cursor = match ksp_store_api::RawPageCursor::try_new(transaction_family_bytes.into_boxed_slice()) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::result::Result::Err(LiveFailure::new("cursor_cross_family_model")),
|
|
};
|
|
let transaction_family_query = match account_query(
|
|
std::option::Option::None,
|
|
200,
|
|
202,
|
|
ksp_store_api::RawSortDirection::Ascending,
|
|
std::option::Option::Some(transaction_family_cursor),
|
|
) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
match backend.list_raw_account_states(&transaction_family_query).await {
|
|
std::result::Result::Err(error) if error.kind() == ksp_store_postgres_lib::PostgresBackendErrorKind::QueryInvalid => {},
|
|
_ => return std::result::Result::Err(LiveFailure::new("cursor_cross_family")),
|
|
}
|
|
return std::result::Result::Ok(());
|
|
}
|
|
|
|
async fn prove_cross_family_coexistence(
|
|
backend: &ksp_store_postgres_lib::PostgresBackend,
|
|
admin: &tokio_postgres::Client,
|
|
) -> std::result::Result<(), LiveFailure> {
|
|
let account_state = match account_state(90, 900, 90, 9_000, 91, false, 92, 3, 90) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let account_observation = match account_observation(90, 90, 900, 90, 90, true) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
match backend.persist_raw_account_acquisition(account_state, account_observation).await {
|
|
std::result::Result::Ok(value) if value.entity() == ksp_store_api::RawEntityWriteOutcome::Inserted => {},
|
|
_ => return std::result::Result::Err(LiveFailure::new("coexistence_account_insert")),
|
|
}
|
|
let transaction = match raw_transaction(90, 900, 90) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let transaction_observation = match raw_transaction_observation(90, 90, 90) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
match backend
|
|
.persist_raw_transaction_acquisition(transaction, transaction_observation, ksp_store_api::RawTransactionAcquisitionMode::Normal)
|
|
.await
|
|
{
|
|
std::result::Result::Ok(value) if value.entity() == ksp_store_api::RawEntityWriteOutcome::Inserted => {},
|
|
_ => return std::result::Result::Err(LiveFailure::new("coexistence_transaction_insert")),
|
|
}
|
|
let transaction_reference = match transaction_reference(90) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
match backend.get_raw_transaction(&transaction_reference).await {
|
|
std::result::Result::Ok(std::option::Option::Some(_)) => {},
|
|
_ => return std::result::Result::Err(LiveFailure::new("coexistence_transaction_read")),
|
|
}
|
|
let account_key = ksp_store_api::RawObservationKey::new([90; 32]);
|
|
match backend.get_raw_account_observation(&account_key).await {
|
|
std::result::Result::Ok(std::option::Option::Some(_)) => {},
|
|
_ => return std::result::Result::Err(LiveFailure::new("coexistence_account_observation")),
|
|
}
|
|
let transaction_exists_result = transaction_exists(admin, &transaction_reference).await;
|
|
match transaction_exists_result {
|
|
std::result::Result::Ok(true) => {},
|
|
_ => return std::result::Result::Err(LiveFailure::new("coexistence_transaction_physical")),
|
|
}
|
|
return std::result::Result::Ok(());
|
|
}
|
|
|
|
async fn prove_concurrent_identical_insert(uri: &str) -> std::result::Result<(), LiveFailure> {
|
|
let first = tokio::spawn(persist_account_once(uri.to_owned(), 70, 700, 70, 7_000, 70, 70));
|
|
let second = tokio::spawn(persist_account_once(uri.to_owned(), 70, 700, 70, 7_000, 70, 70));
|
|
let first_result = match joined_account_persist(first.await, "concurrent_identical_first") {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let second_result = match joined_account_persist(second.await, "concurrent_identical_second") {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
if !one_inserted_one_already([first_result, second_result]) {
|
|
return std::result::Result::Err(LiveFailure::new("concurrent_identical_outcome"));
|
|
}
|
|
return std::result::Result::Ok(());
|
|
}
|
|
|
|
async fn prove_concurrent_divergent_insert(uri: &str) -> std::result::Result<(), LiveFailure> {
|
|
let first = tokio::spawn(persist_account_once(uri.to_owned(), 71, 710, 71, 7_100, 71, 71));
|
|
let second = tokio::spawn(persist_account_once(uri.to_owned(), 71, 710, 71, 7_101, 72, 72));
|
|
let first_result = match joined_account_persist(first.await, "concurrent_divergent_first") {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let second_result = match joined_account_persist(second.await, "concurrent_divergent_second") {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let inserted = [&first_result, &second_result]
|
|
.into_iter()
|
|
.filter(|value| matches!(value, LiveAccountPersistResult::Outcome(outcome) if outcome.entity() == ksp_store_api::RawEntityWriteOutcome::Inserted))
|
|
.count();
|
|
let conflicts = [&first_result, &second_result]
|
|
.into_iter()
|
|
.filter(|value| matches!(value, LiveAccountPersistResult::BackendError(ksp_store_postgres_lib::PostgresBackendErrorKind::Conflict)))
|
|
.count();
|
|
if inserted != 1 || conflicts != 1 {
|
|
return std::result::Result::Err(LiveFailure::new("concurrent_divergent_outcome"));
|
|
}
|
|
return std::result::Result::Ok(());
|
|
}
|
|
|
|
async fn prove_cancellation_rollback(admin: &mut tokio_postgres::Client, uri: &str) -> std::result::Result<(), LiveFailure> {
|
|
let seed_state = match account_state(80, 800, 80, 8_000, 81, false, 82, 3, 80) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let seed_observation = match account_observation(80, 80, 800, 80, 80, false) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let seed_backend = match open_backend(uri, "devnet", true, true).await {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let seeded = seed_backend.persist_raw_account_acquisition(seed_state, seed_observation).await;
|
|
match seeded {
|
|
std::result::Result::Ok(value) if value.entity() == ksp_store_api::RawEntityWriteOutcome::Inserted => {},
|
|
_ => return std::result::Result::Err(LiveFailure::new("cancellation_seed")),
|
|
}
|
|
let close_seed = close_backend(seed_backend).await;
|
|
if let std::result::Result::Err(error) = close_seed {
|
|
return std::result::Result::Err(error);
|
|
}
|
|
let lock_transaction = match admin.transaction().await {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::result::Result::Err(LiveFailure::new("cancellation_lock_begin")),
|
|
};
|
|
let key = ksp_store_api::RawObservationKey::new([80; 32]);
|
|
let key_bytes: &[u8] = key.as_bytes();
|
|
let lock_result = lock_transaction.query_one(LIVE_LOCK_ACCOUNT_OBSERVATION_SQL, &[&key_bytes]).await;
|
|
if lock_result.is_err() {
|
|
return std::result::Result::Err(LiveFailure::new("cancellation_lock_observation"));
|
|
}
|
|
let cancel_backend = match open_backend(uri, "devnet", true, true).await {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let state = match account_state(81, 810, 81, 8_100, 82, false, 83, 3, 81) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let observation = match account_observation(80, 81, 810, 81, 81, false) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let mut task = tokio::spawn(async move {
|
|
return cancel_backend.persist_raw_account_acquisition(state, observation).await;
|
|
});
|
|
let premature = tokio::time::timeout(LIVE_CANCEL_WAIT, &mut task).await;
|
|
if premature.is_ok() {
|
|
let _ = lock_transaction.rollback().await;
|
|
return std::result::Result::Err(LiveFailure::new("cancellation_operation_not_blocked"));
|
|
}
|
|
task.abort();
|
|
let cancelled = task.await;
|
|
match cancelled {
|
|
std::result::Result::Err(error) if error.is_cancelled() => {},
|
|
_ => {
|
|
let _ = lock_transaction.rollback().await;
|
|
return std::result::Result::Err(LiveFailure::new("cancellation_join"));
|
|
},
|
|
}
|
|
let unlock_result = lock_transaction.rollback().await;
|
|
if unlock_result.is_err() {
|
|
return std::result::Result::Err(LiveFailure::new("cancellation_unlock"));
|
|
}
|
|
let reference = match account_reference(81, 810, 81) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let exists_result = account_state_exists(admin, &reference).await;
|
|
match exists_result {
|
|
std::result::Result::Ok(false) => {},
|
|
std::result::Result::Ok(true) => return std::result::Result::Err(LiveFailure::new("cancellation_left_state")),
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
}
|
|
return std::result::Result::Ok(());
|
|
}
|
|
|
|
async fn collect_account_pages(
|
|
backend: &ksp_store_postgres_lib::PostgresBackend,
|
|
pubkey: std::option::Option<ksp_store_api::Pubkey>,
|
|
start: u64,
|
|
end: u64,
|
|
direction: ksp_store_api::RawSortDirection,
|
|
) -> std::result::Result<std::vec::Vec<(u64, u8, u8)>, LiveFailure> {
|
|
let mut output = std::vec::Vec::new();
|
|
let mut cursor = std::option::Option::None;
|
|
loop {
|
|
let query = match account_query(pubkey, start, end, direction, cursor) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let page = match backend.list_raw_account_states(&query).await {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::result::Result::Err(LiveFailure::new("pagination_query")),
|
|
};
|
|
for reference in page.items() {
|
|
let pubkey_bytes: &[u8] = reference.pubkey().as_ref();
|
|
let pubkey_seed = match pubkey_bytes.first() {
|
|
std::option::Option::Some(value) => *value,
|
|
std::option::Option::None => return std::result::Result::Err(LiveFailure::new("pagination_pubkey_decode")),
|
|
};
|
|
let hash_seed = reference.state_hash().as_bytes()[0];
|
|
output.push((reference.slot(), pubkey_seed, hash_seed));
|
|
}
|
|
cursor = page.next_cursor().cloned();
|
|
if cursor.is_none() {
|
|
break;
|
|
}
|
|
}
|
|
return std::result::Result::Ok(output);
|
|
}
|
|
|
|
async fn persist_account_once(
|
|
uri: std::string::String,
|
|
pubkey_seed: u8,
|
|
slot: u64,
|
|
hash_seed: u8,
|
|
lamports: u64,
|
|
data_seed: u8,
|
|
observation_seed: u8,
|
|
) -> std::result::Result<LiveAccountPersistResult, LiveFailure> {
|
|
let backend = match open_backend(uri.as_str(), "devnet", true, true).await {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let state = match account_state(pubkey_seed, slot, hash_seed, lamports, 100, false, 101, 4, data_seed) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let observation = match account_observation(observation_seed, pubkey_seed, slot, hash_seed, observation_seed, false) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let persisted = backend.persist_raw_account_acquisition(state, observation).await;
|
|
let result = match persisted {
|
|
std::result::Result::Ok(value) => LiveAccountPersistResult::Outcome(value),
|
|
std::result::Result::Err(error) => LiveAccountPersistResult::BackendError(error.kind()),
|
|
};
|
|
let close_result = close_backend(backend).await;
|
|
if let std::result::Result::Err(error) = close_result {
|
|
return std::result::Result::Err(error);
|
|
}
|
|
return std::result::Result::Ok(result);
|
|
}
|
|
|
|
fn joined_account_persist(
|
|
joined: std::result::Result<std::result::Result<LiveAccountPersistResult, LiveFailure>, tokio::task::JoinError>,
|
|
phase: &'static str,
|
|
) -> std::result::Result<LiveAccountPersistResult, LiveFailure> {
|
|
return match joined {
|
|
std::result::Result::Ok(std::result::Result::Ok(value)) => std::result::Result::Ok(value),
|
|
_ => std::result::Result::Err(LiveFailure::new(phase)),
|
|
};
|
|
}
|
|
|
|
fn one_inserted_one_already(values: [LiveAccountPersistResult; 2]) -> bool {
|
|
let inserted = values
|
|
.iter()
|
|
.filter(|value| matches!(value, LiveAccountPersistResult::Outcome(outcome) if outcome.entity() == ksp_store_api::RawEntityWriteOutcome::Inserted))
|
|
.count();
|
|
let already = values
|
|
.iter()
|
|
.filter(|value| matches!(value, LiveAccountPersistResult::Outcome(outcome) if outcome.entity() == ksp_store_api::RawEntityWriteOutcome::AlreadyPresent))
|
|
.count();
|
|
let errors = values.iter().filter(|value| matches!(value, LiveAccountPersistResult::BackendError(_))).count();
|
|
return inserted == 1 && already == 1 && errors == 0;
|
|
}
|
|
|
|
fn account_query(
|
|
pubkey: std::option::Option<ksp_store_api::Pubkey>,
|
|
start: u64,
|
|
end: u64,
|
|
direction: ksp_store_api::RawSortDirection,
|
|
cursor: std::option::Option<ksp_store_api::RawPageCursor>,
|
|
) -> std::result::Result<ksp_store_api::RawAccountStateQuery, LiveFailure> {
|
|
let network = match network("devnet") {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let slots = match ksp_store_api::RawSlotRange::new(std::option::Option::Some(start), std::option::Option::Some(end)) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::result::Result::Err(LiveFailure::new("model_slot_range")),
|
|
};
|
|
let limit = match ksp_store_api::RawPageLimit::new(2) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::result::Result::Err(LiveFailure::new("model_page_limit")),
|
|
};
|
|
let page = match cursor {
|
|
std::option::Option::Some(value) => ksp_store_api::RawPageRequest::after(limit, value),
|
|
std::option::Option::None => ksp_store_api::RawPageRequest::first(limit),
|
|
};
|
|
return std::result::Result::Ok(ksp_store_api::RawAccountStateQuery::new(network, pubkey, slots, direction, page));
|
|
}
|
|
|
|
fn account_reference(pubkey_seed: u8, slot: u64, hash_seed: u8) -> std::result::Result<ksp_store_api::RawAccountStateReference, LiveFailure> {
|
|
let network = match network("devnet") {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
return std::result::Result::Ok(ksp_store_api::RawAccountStateReference::new(
|
|
network,
|
|
ksp_store_api::Pubkey::new_from_array([pubkey_seed; 32]),
|
|
slot,
|
|
ksp_store_api::RawContentHash::new([hash_seed; 32]),
|
|
));
|
|
}
|
|
|
|
fn account_state(
|
|
pubkey_seed: u8,
|
|
slot: u64,
|
|
hash_seed: u8,
|
|
lamports: u64,
|
|
owner_seed: u8,
|
|
executable: bool,
|
|
rent_epoch: u64,
|
|
data_len: usize,
|
|
data_seed: u8,
|
|
) -> std::result::Result<ksp_store_api::RawAccountState, LiveFailure> {
|
|
let reference = match account_reference(pubkey_seed, slot, hash_seed) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let data = vec![data_seed; data_len].into_boxed_slice();
|
|
return match ksp_store_api::RawAccountState::try_new(
|
|
reference,
|
|
lamports,
|
|
ksp_store_api::Pubkey::new_from_array([owner_seed; 32]),
|
|
executable,
|
|
rent_epoch,
|
|
data,
|
|
) {
|
|
std::result::Result::Ok(value) => std::result::Result::Ok(value),
|
|
std::result::Result::Err(_) => std::result::Result::Err(LiveFailure::new("model_account_state")),
|
|
};
|
|
}
|
|
|
|
fn account_observation(
|
|
observation_seed: u8,
|
|
pubkey_seed: u8,
|
|
slot: u64,
|
|
hash_seed: u8,
|
|
provenance_seed: u8,
|
|
yellowstone_metadata: bool,
|
|
) -> std::result::Result<ksp_store_api::RawAccountObservation, LiveFailure> {
|
|
let reference = match account_reference(pubkey_seed, slot, hash_seed) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let provenance = match acquisition_provenance(provenance_seed, "account-subscribe") {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let observation = ksp_store_api::RawAccountObservation::new(ksp_store_api::RawObservationKey::new([observation_seed; 32]), reference, provenance);
|
|
if !yellowstone_metadata {
|
|
return std::result::Result::Ok(observation);
|
|
}
|
|
return std::result::Result::Ok(
|
|
observation
|
|
.with_is_startup(true)
|
|
.with_transaction_signature(ksp_store_api::RawTransactionSignature::new([provenance_seed.wrapping_add(20); 64]))
|
|
.with_write_version(if slot == u64::MAX { u64::MAX } else { 10_000_u64 + u64::from(provenance_seed) }),
|
|
);
|
|
}
|
|
|
|
fn acquisition_provenance(provenance_seed: u8, method_name: &str) -> std::result::Result<ksp_store_api::RawAcquisitionProvenance, LiveFailure> {
|
|
let provider = match provenance_code("live-provider") {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let protocol = match provenance_code("yellowstone-grpc") {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let method = match provenance_code(method_name) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let received = match ksp_store_api::RawTimestamp::from_unix_millis(1_800_000_000_000_u64 + u64::from(provenance_seed)) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::result::Result::Err(LiveFailure::new("model_received_at")),
|
|
};
|
|
let observed = match ksp_store_api::RawTimestamp::from_unix_millis(1_799_999_999_000_u64 + u64::from(provenance_seed)) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::result::Result::Err(LiveFailure::new("model_observed_at")),
|
|
};
|
|
let capture_session = match provenance_code("session-live") {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let commitment = match provenance_code("confirmed") {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let endpoint = match provenance_code("endpoint-live") {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let filter = match provenance_code("filter-live") {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let provenance = ksp_store_api::RawAcquisitionProvenance::new(provider, protocol, method, ksp_store_api::RawAcquisitionOrigin::Live, received)
|
|
.with_capture_session_id(capture_session)
|
|
.with_commitment(commitment)
|
|
.with_endpoint_id(endpoint)
|
|
.with_filter_id(filter)
|
|
.with_source_payload_hash(ksp_store_api::RawContentHash::new([provenance_seed.wrapping_add(50); 32]));
|
|
let provenance = match provenance.try_with_observed_at(observed) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::result::Result::Err(LiveFailure::new("model_observed_order")),
|
|
};
|
|
return match provenance.try_with_source_payload_size_bytes(1_024_u64 + u64::from(provenance_seed)) {
|
|
std::result::Result::Ok(value) => std::result::Result::Ok(value),
|
|
std::result::Result::Err(_) => std::result::Result::Err(LiveFailure::new("model_source_size")),
|
|
};
|
|
}
|
|
|
|
fn account_state_matches(
|
|
state: &ksp_store_api::RawAccountState,
|
|
pubkey_seed: u8,
|
|
slot: u64,
|
|
hash_seed: u8,
|
|
lamports: u64,
|
|
owner_seed: u8,
|
|
executable: bool,
|
|
rent_epoch: u64,
|
|
data_len: usize,
|
|
data_seed: u8,
|
|
) -> bool {
|
|
return state.reference().network().as_str() == "devnet"
|
|
&& state.reference().pubkey() == &ksp_store_api::Pubkey::new_from_array([pubkey_seed; 32])
|
|
&& state.reference().slot() == slot
|
|
&& state.reference().state_hash() == ksp_store_api::RawContentHash::new([hash_seed; 32])
|
|
&& state.lamports() == lamports
|
|
&& state.owner() == &ksp_store_api::Pubkey::new_from_array([owner_seed; 32])
|
|
&& state.executable() == executable
|
|
&& state.rent_epoch() == rent_epoch
|
|
&& state.data_len() == data_len
|
|
&& state.data().iter().all(|value| return *value == data_seed);
|
|
}
|
|
|
|
fn raw_transaction(signature_seed: u8, slot: u64, payload_seed: u8) -> std::result::Result<ksp_store_api::RawTransaction, LiveFailure> {
|
|
let reference = match transaction_reference(signature_seed) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let format_id = match ksp_store_api::RawFormatId::new("ksp-live-raw-v1") {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::result::Result::Err(LiveFailure::new("model_format")),
|
|
};
|
|
let content_hash = ksp_store_api::RawContentHash::new([payload_seed.wrapping_add(100); 32]);
|
|
let payload_bytes = [payload_seed, payload_seed.wrapping_add(1), payload_seed.wrapping_add(2), payload_seed.wrapping_add(3)].to_vec().into_boxed_slice();
|
|
let payload = match ksp_store_api::RawPayload::try_new(format_id, 1, payload_bytes, content_hash) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::result::Result::Err(LiveFailure::new("model_payload")),
|
|
};
|
|
let timestamp = match ksp_store_api::RawTimestamp::from_unix_millis(1_700_000_000_000_u64 + slot) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::result::Result::Err(LiveFailure::new("model_block_time")),
|
|
};
|
|
return std::result::Result::Ok(ksp_store_api::RawTransaction::new(reference, slot, std::option::Option::Some(timestamp), payload));
|
|
}
|
|
|
|
fn raw_transaction_observation(
|
|
observation_seed: u8,
|
|
signature_seed: u8,
|
|
provenance_seed: u8,
|
|
) -> std::result::Result<ksp_store_api::RawTransactionObservation, LiveFailure> {
|
|
let transaction = match transaction_reference(signature_seed) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let provenance = match acquisition_provenance(provenance_seed, "transaction-subscribe") {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
return std::result::Result::Ok(ksp_store_api::RawTransactionObservation::new(
|
|
ksp_store_api::RawObservationKey::new([observation_seed; 32]),
|
|
transaction,
|
|
provenance,
|
|
));
|
|
}
|
|
|
|
fn transaction_reference(signature_seed: u8) -> std::result::Result<ksp_store_api::RawTransactionReference, LiveFailure> {
|
|
let network = match network("devnet") {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
return std::result::Result::Ok(ksp_store_api::RawTransactionReference::new(network, ksp_store_api::RawTransactionSignature::new([signature_seed; 64])));
|
|
}
|
|
|
|
fn provenance_code(value: &str) -> std::result::Result<ksp_store_api::RawProvenanceCode, LiveFailure> {
|
|
return match ksp_store_api::RawProvenanceCode::new(value) {
|
|
std::result::Result::Ok(code) => std::result::Result::Ok(code),
|
|
std::result::Result::Err(_) => std::result::Result::Err(LiveFailure::new("model_provenance_code")),
|
|
};
|
|
}
|
|
|
|
fn network(value: &str) -> std::result::Result<ksp_store_api::RawNetworkId, LiveFailure> {
|
|
return match ksp_store_api::RawNetworkId::new(value) {
|
|
std::result::Result::Ok(network) => std::result::Result::Ok(network),
|
|
std::result::Result::Err(_) => std::result::Result::Err(LiveFailure::new("network")),
|
|
};
|
|
}
|
|
|
|
fn settings(
|
|
uri: &str,
|
|
network_name: &str,
|
|
schema_autocreate: bool,
|
|
schema_autoupdate: bool,
|
|
) -> std::result::Result<ksp_store_postgres_lib::PostgresBackendSettings, LiveFailure> {
|
|
let network = match network(network_name) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
return std::result::Result::Ok(ksp_store_postgres_lib::PostgresBackendSettings::with_schema_policy(
|
|
network,
|
|
uri,
|
|
8,
|
|
std::time::Duration::from_secs(10),
|
|
std::time::Duration::from_secs(5),
|
|
std::time::Duration::from_secs(10),
|
|
std::time::Duration::from_secs(5),
|
|
ksp_store_postgres_lib::PostgresBackendTlsMode::Disabled,
|
|
schema_autocreate,
|
|
schema_autoupdate,
|
|
std::time::Duration::from_secs(30),
|
|
std::time::Duration::from_secs(10),
|
|
));
|
|
}
|
|
|
|
async fn open_backend(
|
|
uri: &str,
|
|
network_name: &str,
|
|
schema_autocreate: bool,
|
|
schema_autoupdate: bool,
|
|
) -> std::result::Result<ksp_store_postgres_lib::PostgresBackend, LiveFailure> {
|
|
let opened = open_backend_result(uri, network_name, schema_autocreate, schema_autoupdate).await;
|
|
return match opened {
|
|
std::result::Result::Ok(std::result::Result::Ok(value)) => std::result::Result::Ok(value),
|
|
std::result::Result::Ok(std::result::Result::Err(error)) => std::result::Result::Err(LiveFailure::new(error.phase())),
|
|
std::result::Result::Err(error) => std::result::Result::Err(error),
|
|
};
|
|
}
|
|
|
|
async fn open_backend_result(
|
|
uri: &str,
|
|
network_name: &str,
|
|
schema_autocreate: bool,
|
|
schema_autoupdate: bool,
|
|
) -> std::result::Result<std::result::Result<ksp_store_postgres_lib::PostgresBackend, ksp_store_postgres_lib::PostgresBackendError>, LiveFailure> {
|
|
let backend_settings = match settings(uri, network_name, schema_autocreate, schema_autoupdate) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
return std::result::Result::Ok(ksp_store_postgres_lib::PostgresBackend::open(backend_settings).await);
|
|
}
|
|
|
|
async fn close_backend(backend: ksp_store_postgres_lib::PostgresBackend) -> std::result::Result<(), LiveFailure> {
|
|
return match backend.close(std::time::Duration::from_secs(5)).await {
|
|
std::result::Result::Ok(()) => std::result::Result::Ok(()),
|
|
std::result::Result::Err(_) => std::result::Result::Err(LiveFailure::new("backend_close")),
|
|
};
|
|
}
|
|
|
|
async fn connect_admin(uri: &str) -> std::result::Result<tokio_postgres::Client, LiveFailure> {
|
|
let parsed = uri.parse::<tokio_postgres::Config>();
|
|
let mut config = match parsed {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::result::Result::Err(LiveFailure::new("admin_config")),
|
|
};
|
|
config.ssl_mode(tokio_postgres::config::SslMode::Disable);
|
|
config.ssl_negotiation(tokio_postgres::config::SslNegotiation::Postgres);
|
|
let connected = config.connect(tokio_postgres::NoTls).await;
|
|
let (client, connection) = match connected {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::result::Result::Err(LiveFailure::new("admin_connect")),
|
|
};
|
|
let _connection_task = tokio::spawn(async move {
|
|
let _result = connection.await;
|
|
return;
|
|
});
|
|
return std::result::Result::Ok(client);
|
|
}
|
|
|
|
async fn postgres_major(client: &tokio_postgres::Client) -> std::result::Result<u32, LiveFailure> {
|
|
let row_result = client.query_one("SHOW server_version_num", &[]).await;
|
|
let row = match row_result {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::result::Result::Err(LiveFailure::new("server_version")),
|
|
};
|
|
let value = match row.try_get::<usize, std::string::String>(0) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::result::Result::Err(LiveFailure::new("server_version_decode")),
|
|
};
|
|
let version_num = match value.parse::<u32>() {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::result::Result::Err(LiveFailure::new("server_version_parse")),
|
|
};
|
|
return std::result::Result::Ok(version_num / 10_000);
|
|
}
|
|
|
|
async fn managed_schema_exists(client: &tokio_postgres::Client) -> std::result::Result<bool, LiveFailure> {
|
|
let row = match client.query_one(LIVE_MANAGED_SCHEMA_EXISTS_SQL, &[]).await {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::result::Result::Err(LiveFailure::new("managed_schema_probe")),
|
|
};
|
|
return match row.try_get::<usize, bool>(0) {
|
|
std::result::Result::Ok(value) => std::result::Result::Ok(value),
|
|
std::result::Result::Err(_) => std::result::Result::Err(LiveFailure::new("managed_schema_probe_decode")),
|
|
};
|
|
}
|
|
|
|
async fn drop_managed_schema(client: &tokio_postgres::Client) -> std::result::Result<(), LiveFailure> {
|
|
return match client.batch_execute(LIVE_MANAGED_SCHEMA_DROP_SQL).await {
|
|
std::result::Result::Ok(()) => std::result::Result::Ok(()),
|
|
std::result::Result::Err(_) => std::result::Result::Err(LiveFailure::new("managed_schema_cleanup")),
|
|
};
|
|
}
|
|
|
|
async fn account_index_exists(client: &tokio_postgres::Client) -> std::result::Result<bool, LiveFailure> {
|
|
let row = match client.query_one(LIVE_ACCOUNT_INDEX_EXISTS_SQL, &[]).await {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::result::Result::Err(LiveFailure::new("account_index_probe")),
|
|
};
|
|
return match row.try_get::<usize, bool>(0) {
|
|
std::result::Result::Ok(value) => std::result::Result::Ok(value),
|
|
std::result::Result::Err(_) => std::result::Result::Err(LiveFailure::new("account_index_probe_decode")),
|
|
};
|
|
}
|
|
|
|
async fn account_state_exists(client: &tokio_postgres::Client, reference: &ksp_store_api::RawAccountStateReference) -> std::result::Result<bool, LiveFailure> {
|
|
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 = match client.query_one(LIVE_ACCOUNT_STATE_EXISTS_SQL, &[&pubkey_bytes, &slot_text, &state_hash_bytes]).await {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::result::Result::Err(LiveFailure::new("account_state_exists_probe")),
|
|
};
|
|
return match row.try_get::<usize, bool>(0) {
|
|
std::result::Result::Ok(value) => std::result::Result::Ok(value),
|
|
std::result::Result::Err(_) => std::result::Result::Err(LiveFailure::new("account_state_exists_decode")),
|
|
};
|
|
}
|
|
|
|
async fn transaction_exists(client: &tokio_postgres::Client, reference: &ksp_store_api::RawTransactionReference) -> std::result::Result<bool, LiveFailure> {
|
|
let signature = reference.signature();
|
|
let signature_bytes: &[u8] = signature.as_bytes();
|
|
let row = match client.query_one(LIVE_TRANSACTION_EXISTS_SQL, &[&signature_bytes]).await {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::result::Result::Err(LiveFailure::new("transaction_exists_probe")),
|
|
};
|
|
return match row.try_get::<usize, bool>(0) {
|
|
std::result::Result::Ok(value) => std::result::Result::Ok(value),
|
|
std::result::Result::Err(_) => std::result::Result::Err(LiveFailure::new("transaction_exists_decode")),
|
|
};
|
|
}
|