Sasedev/khadhroony-bot3
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v0.5.3-pre.005-fix010

Browse Source
This commit is contained in:
SinuS von SifriduS
2026-08-14 14:40:36 +02:00
parent b6583bd9fb
commit 734a27f2ca
59 changed files with 4337 additions and 747 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
ks-store/src/postgres/migrations.rs
View File

@@ -1,13 +1,11 @@
// file: ks-store/src/postgres/migrations.rs
// version: 11
// version: 12
//! PostgreSQL schema resources and migration conventions for the storage backend.
/// Canonical Solana table prefix.
#[cfg(test)]
const SOLANA_TABLE_PREFIX: &str = "k_sol_";
/// Historical table prefix rejected by the `0.5.3` baseline initializer.
pub(crate) const LEGACY_SOLANA_TABLE_PREFIX: &str = "kb_sol_";
/// Advisory lock id used while applying idempotent store schema statements.
pub(crate) const STORE_SCHEMA_ADVISORY_LOCK_ID: i64 = 2_024_000_503;
/// PostgreSQL table `k_sol_raw_transactions`.
@@ -386,6 +384,28 @@ fn create_pg_table_if_not_exists_k_sol_raw_transactions() -> &'static str {
);
}
/// Embedded `CREATE TABLE` resources defining the current managed PostgreSQL column contract.
pub(crate) fn store_table_create_statements() -> std::vec::Vec<&'static str> {
return vec![
create_pg_table_if_not_exists_k_sol_core_account_keys(),
create_pg_table_if_not_exists_k_sol_core_account_states(),
create_pg_table_if_not_exists_k_sol_core_balance_changes(),
create_pg_table_if_not_exists_k_sol_core_inner_instructions(),
create_pg_table_if_not_exists_k_sol_core_instructions(),
create_pg_table_if_not_exists_k_sol_core_logs(),
create_pg_table_if_not_exists_k_sol_core_return_data(),
create_pg_table_if_not_exists_k_sol_core_transactions(),
create_pg_table_if_not_exists_k_sol_decode_coverage_declarations(),
create_pg_table_if_not_exists_k_sol_decode_coverage_observations(),
create_pg_table_if_not_exists_k_sol_decode_events(),
create_pg_table_if_not_exists_k_sol_mat_outputs(),
create_pg_table_if_not_exists_k_sol_obs_account_observations(),
create_pg_table_if_not_exists_k_sol_obs_transaction_observations(),
create_pg_table_if_not_exists_k_sol_ops_processing_ledger(),
create_pg_table_if_not_exists_k_sol_raw_transactions(),
];
}
fn add_pg_constraint_if_not_exists_ck_k_sol_core_account_keys_index_non_negative() -> &'static str {
return include_str!(
"../../migrations/postgres/schema/constraints/add_constraint_if_not_exists_ck_k_sol_core_account_keys_index_non_negative.sql"
@@ -2245,6 +2265,7 @@ mod tests {
+ super::CORE_STORE_TABLE_NAMES.len()
+ super::DECODE_STORE_TABLE_NAMES.len();
assert_eq!(table_count, 16);
assert_eq!(crate::store_table_create_statements().len(), table_count);
assert_eq!(crate::expected_postgres_index_names().len(), 79);
}

5
ks-store/src/postgres/query.rs
View File

@@ -1,5 +1,5 @@
// file: ks-store/src/postgres/query.rs
// version: 6
// version: 9
//! PostgreSQL query modules.
@@ -35,6 +35,7 @@ pub(crate) use self::core_queries::update_core_instruction_lifecycle;
pub(crate) use self::decode_pipeline_queries::is_decode_current;
pub(crate) use self::decode_pipeline_queries::list_decode_coverage_summary;
pub(crate) use self::decode_pipeline_queries::list_decode_inputs;
pub(crate) use self::decode_pipeline_queries::list_materialized_output_page;
pub(crate) use self::decode_pipeline_queries::list_materialized_outputs;
pub(crate) use self::decode_pipeline_queries::mark_decode_failed;
pub(crate) use self::decode_pipeline_queries::persist_decode_coverage_declarations;
@@ -56,7 +57,7 @@ pub(crate) use self::raw_queries::update_raw_payload_lifecycle;
pub(crate) use self::replay_candidate_queries::list_replay_entity_summaries;
pub(crate) use self::replay_candidate_queries::list_replay_program_summaries;
pub(crate) use self::replay_candidate_queries::list_replay_transaction_candidates;
pub(crate) use self::schema_queries::apply_store_schema;
pub(crate) use self::schema_queries::apply_missing_store_schema;
pub(crate) use self::schema_queries::load_expected_index_counts;
pub(crate) use self::table_diagnostics_queries::load_table_statistics;
pub(crate) use self::table_diagnostics_queries::table_exists;

4
ks-store/src/postgres/query/core_extraction_queries.rs
View File

@@ -1,5 +1,5 @@
// file: ks-store/src/postgres/query/core_extraction_queries.rs
// version: 9
// version: 10
//! PostgreSQL queries for atomic canonical transaction to core extraction.
@@ -871,7 +871,7 @@ mod tests {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => panic!("unexpected connect error: {error}"),
};
let schema_result = store.initialize_store_schema().await;
let schema_result = store.initialize_missing_store_schema().await;
if let std::result::Result::Err(error) = schema_result {
panic!("unexpected schema error: {error}");
}

4
ks-store/src/postgres/query/core_queries.rs
View File

@@ -1,5 +1,5 @@
// file: ks-store/src/postgres/query/core_queries.rs
// version: 11
// version: 12
//! PostgreSQL queries for normalized Solana core storage.
@@ -1181,7 +1181,7 @@ mod tests {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => panic!("unexpected connect error: {error}"),
};
let schema_result = store.initialize_store_schema().await;
let schema_result = store.initialize_missing_store_schema().await;
if let std::result::Result::Err(error) = schema_result {
panic!("unexpected schema error: {error}");
}

298
ks-store/src/postgres/query/decode_pipeline_queries.rs
View File

@@ -1,5 +1,5 @@
// file: ks-store/src/postgres/query/decode_pipeline_queries.rs
// version: 12
// version: 14
//! PostgreSQL queries for contextual decode, coverage and materialization persistence.
@@ -22,6 +22,30 @@ struct MaterializedOutputDatabaseRow {
updated_at: std::string::String,
}
#[derive(sqlx::FromRow)]
struct MaterializedOutputPageDatabaseRow {
id: i64,
processor_name: std::string::String,
processor_version: std::string::String,
input_key: std::string::String,
output_key: std::string::String,
source_event_key: std::string::String,
source_decoder_name: std::string::String,
source_decoder_version: std::string::String,
signature: std::string::String,
slot: i64,
materialized_family: std::string::String,
payload_json: serde_json::Value,
created_at: std::string::String,
updated_at: std::string::String,
}
#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
struct MaterializedOutputPageCursor {
slot: i64,
id: i64,
}
pub(crate) async fn list_decode_inputs(
pool: &sqlx::PgPool,
filter: &crate::DecodeSelectionFilter,
@@ -89,6 +113,229 @@ pub(crate) async fn list_materialized_outputs(
return std::result::Result::Ok(output);
}
pub(crate) async fn list_materialized_output_page(
pool: &sqlx::PgPool,
filter: &crate::MaterializedOutputPageFilter,
page_request: &crate::PageRequest,
) -> ks_core::Result<crate::CountedPageSlice<crate::MaterializedOutputQueryRow>> {
if page_request.limit == 0 || page_request.limit > crate::MAX_PAGE_SIZE {
return std::result::Result::Err(ks_core::Error::db(
"materialized output page size is outside repository bounds",
));
}
let cursor_result = decode_materialized_output_page_cursor(page_request.cursor.as_deref());
let cursor = match cursor_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let payload_match_groups_result = materialized_output_payload_match_groups(filter);
let payload_match_groups = match payload_match_groups_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let count_started_at = std::time::Instant::now();
let count_result = sqlx::query_scalar::<sqlx::Postgres, i64>(
"SELECT COUNT(*)::BIGINT FROM k_sol_mat_outputs WHERE ($1::TEXT IS NULL OR processor_name = $1) AND ($2::TEXT IS NULL OR materialized_family = $2) AND ($3::TEXT IS NULL OR source_decoder_name = $3) AND ($4::TEXT IS NULL OR POSITION(LOWER($4) IN LOWER(signature)) > 0) AND NOT EXISTS (SELECT 1 FROM jsonb_array_elements($5::JSONB) AS match_group(value) WHERE NOT EXISTS (SELECT 1 FROM jsonb_array_elements(match_group.value) AS candidate(value) WHERE payload_jsonb @> candidate.value))",
)
.bind(filter.processor_name.as_deref())
.bind(filter.materialized_family.as_deref())
.bind(filter.source_decoder_name.as_deref())
.bind(filter.signature_contains.as_deref())
.bind(&payload_match_groups)
.fetch_one(pool)
.await;
let count = match count_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => {
trace_query_failure("count_materialized_output_page", count_started_at, &error);
return std::result::Result::Err(ks_core::Error::db(format!(
"postgres materialized output page count failed: {error}"
)));
},
};
let total_rows_result = u64::try_from(count);
let total_rows = match total_rows_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => {
return std::result::Result::Err(ks_core::Error::db(format!(
"postgres materialized output page count conversion failed: {error}"
)));
},
};
trace_query_success("count_materialized_output_page", count_started_at, 1);
let query_limit = i64::from(page_request.limit) + 1;
let started_at = std::time::Instant::now();
let query_result = sqlx::query_as::<sqlx::Postgres, MaterializedOutputPageDatabaseRow>(
"SELECT id, processor_name, processor_version, input_key, output_key, source_event_key, source_decoder_name, source_decoder_version, signature, slot, materialized_family, payload_jsonb AS payload_json, created_at::TEXT AS created_at, updated_at::TEXT AS updated_at FROM k_sol_mat_outputs WHERE ($1::TEXT IS NULL OR processor_name = $1) AND ($2::TEXT IS NULL OR materialized_family = $2) AND ($3::TEXT IS NULL OR source_decoder_name = $3) AND ($4::TEXT IS NULL OR POSITION(LOWER($4) IN LOWER(signature)) > 0) AND NOT EXISTS (SELECT 1 FROM jsonb_array_elements($5::JSONB) AS match_group(value) WHERE NOT EXISTS (SELECT 1 FROM jsonb_array_elements(match_group.value) AS candidate(value) WHERE payload_jsonb @> candidate.value)) AND ($6::BIGINT IS NULL OR slot < $6 OR (slot = $6 AND id < $7)) ORDER BY slot DESC, id DESC LIMIT $8",
)
.bind(filter.processor_name.as_deref())
.bind(filter.materialized_family.as_deref())
.bind(filter.source_decoder_name.as_deref())
.bind(filter.signature_contains.as_deref())
.bind(&payload_match_groups)
.bind(cursor.as_ref().map(|value| return value.slot))
.bind(cursor.as_ref().map(|value| return value.id))
.bind(query_limit)
.fetch_all(pool)
.await;
let rows = match query_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => {
trace_query_failure("list_materialized_output_page", started_at, &error);
return std::result::Result::Err(ks_core::Error::db(format!(
"postgres materialized output page query failed: {error}"
)));
},
};
let mut output = std::vec::Vec::with_capacity(rows.len().min(usize::from(page_request.limit)));
for row in rows.iter().take(usize::from(page_request.limit)) {
let mapped_result = materialized_output_page_row(row);
let mapped = match mapped_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
output.push(mapped);
}
let next_cursor = if rows.len() > usize::from(page_request.limit) {
let last_index = usize::from(page_request.limit) - 1;
match rows.get(last_index) {
std::option::Option::Some(value) => {
let encoded_result = encode_materialized_output_page_cursor(value.slot, value.id);
match encoded_result {
std::result::Result::Ok(value) => std::option::Option::Some(value),
std::result::Result::Err(error) => return std::result::Result::Err(error),
}
},
std::option::Option::None => std::option::Option::None,
}
} else {
std::option::Option::None
};
trace_query_success("list_materialized_output_page", started_at, output.len());
return std::result::Result::Ok(crate::CountedPageSlice::new(output, next_cursor, total_rows));
}
fn materialized_output_payload_match_groups(
filter: &crate::MaterializedOutputPageFilter,
) -> ks_core::Result<serde_json::Value> {
if filter.payload_match_groups.len() > 8 {
return std::result::Result::Err(ks_core::Error::db(
"materialized output payload filter has too many match groups",
));
}
for group in &filter.payload_match_groups {
if group.is_empty() || group.len() > 4 {
return std::result::Result::Err(ks_core::Error::db(
"materialized output payload filter group size is outside repository bounds",
));
}
if group.iter().any(|candidate| return !candidate.is_object()) {
return std::result::Result::Err(ks_core::Error::db(
"materialized output payload filter alternatives must be JSON objects",
));
}
}
let value = serde_json::Value::Array(
filter
.payload_match_groups
.iter()
.map(|group| return serde_json::Value::Array(group.clone()))
.collect(),
);
let encoded_result = serde_json::to_vec(&value);
let encoded = match encoded_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => {
return std::result::Result::Err(ks_core::Error::db(format!(
"materialized output payload filter serialization failed: {error}"
)));
},
};
if encoded.len() > 8_192 {
return std::result::Result::Err(ks_core::Error::db(
"materialized output payload filter exceeds repository bounds",
));
}
return std::result::Result::Ok(value);
}
fn materialized_output_page_row(
row: &MaterializedOutputPageDatabaseRow,
) -> ks_core::Result<crate::MaterializedOutputQueryRow> {
let slot_result = u64::try_from(row.slot);
let slot = match slot_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => {
return std::result::Result::Err(ks_core::Error::db(format!(
"postgres materialized output page slot conversion failed: {error}"
)));
},
};
return std::result::Result::Ok(crate::MaterializedOutputQueryRow {
processor_name: row.processor_name.clone(),
processor_version: row.processor_version.clone(),
input_key: row.input_key.clone(),
output_key: row.output_key.clone(),
source_event_key: row.source_event_key.clone(),
source_decoder_name: row.source_decoder_name.clone(),
source_decoder_version: row.source_decoder_version.clone(),
signature: row.signature.clone(),
slot,
materialized_family: row.materialized_family.clone(),
payload_json: row.payload_json.clone(),
created_at: row.created_at.clone(),
updated_at: row.updated_at.clone(),
});
}
fn decode_materialized_output_page_cursor(
cursor: std::option::Option<&str>,
) -> ks_core::Result<std::option::Option<MaterializedOutputPageCursor>> {
let value = match cursor {
std::option::Option::Some(raw) => raw,
std::option::Option::None => return std::result::Result::Ok(std::option::Option::None),
};
if value.len() > crate::MAX_PAGE_CURSOR_LENGTH {
return std::result::Result::Err(ks_core::Error::db(
"materialized output page cursor exceeds maximum encoded length",
));
}
let decode_result = serde_json::from_str::<MaterializedOutputPageCursor>(value);
let decoded = match decode_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => {
return std::result::Result::Err(ks_core::Error::db(format!(
"materialized output page cursor is invalid: {error}"
)));
},
};
if decoded.slot < 0 || decoded.id <= 0 {
return std::result::Result::Err(ks_core::Error::db(
"materialized output page cursor values are invalid",
));
}
return std::result::Result::Ok(std::option::Option::Some(decoded));
}
fn encode_materialized_output_page_cursor(
slot: i64,
id: i64,
) -> ks_core::Result<std::string::String> {
let cursor = MaterializedOutputPageCursor { slot, id };
let encode_result = serde_json::to_string(&cursor);
return match encode_result {
std::result::Result::Ok(value) if value.len() <= crate::MAX_PAGE_CURSOR_LENGTH => {
std::result::Result::Ok(value)
},
std::result::Result::Ok(_value) => std::result::Result::Err(ks_core::Error::db(
"materialized output page cursor exceeds maximum encoded length",
)),
std::result::Result::Err(error) => std::result::Result::Err(ks_core::Error::db(format!(
"materialized output page cursor serialization failed: {error}"
))),
};
}
pub(crate) async fn is_decode_current(
pool: &sqlx::PgPool,
identity: &crate::ProcessingLedgerIdentity,
@@ -1004,10 +1251,55 @@ mod tests {
};
let pool_result = sqlx::PgPool::connect(url.as_str()).await;
let pool = result_or_panic(pool_result);
result_or_panic(crate::apply_store_schema(&pool).await);
result_or_panic(crate::apply_missing_store_schema(&pool).await);
return std::option::Option::Some(pool);
}
#[test]
fn materialized_output_page_uses_bounded_structural_filters_without_offset() {
let filter = crate::MaterializedOutputPageFilter::new(
std::option::Option::None,
std::option::Option::None,
std::option::Option::None,
std::option::Option::None,
std::vec![std::vec![serde_json::json!({"operation": "transfer"})]],
);
let encoded_result = super::materialized_output_payload_match_groups(&filter);
let encoded = match encoded_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => panic!("unexpected payload filter error: {error}"),
};
assert_eq!(encoded, serde_json::json!([[{"operation": "transfer"}]]));
let source = include_str!("decode_pipeline_queries.rs");
let tests_marker = source.find("#[cfg(test)]");
let production_source = match tests_marker {
std::option::Option::Some(index) => &source[..index],
std::option::Option::None => source,
};
assert_eq!(production_source.matches("jsonb_array_elements($5::JSONB)").count(), 2);
assert!(!production_source.contains(" OFFSET "));
}
#[test]
fn materialized_output_payload_filter_rejects_invalid_group_shapes() {
let empty_group = crate::MaterializedOutputPageFilter::new(
std::option::Option::None,
std::option::Option::None,
std::option::Option::None,
std::option::Option::None,
std::vec![std::vec::Vec::new()],
);
assert!(super::materialized_output_payload_match_groups(&empty_group).is_err());
let scalar = crate::MaterializedOutputPageFilter::new(
std::option::Option::None,
std::option::Option::None,
std::option::Option::None,
std::option::Option::None,
std::vec![std::vec![serde_json::json!("invalid")]],
);
assert!(super::materialized_output_payload_match_groups(&scalar).is_err());
}
#[tokio::test]
async fn optional_postgres_coverage_declarations_report_insert_skip_and_update_from_env() {
let _postgres_guard = crate::postgres_test_guard().await;
@@ -1152,7 +1444,7 @@ mod tests {
let _postgres_guard = crate::postgres_test_guard().await;
let pool_result = sqlx::PgPool::connect(url.as_str()).await;
let pool = result_or_panic(pool_result);
result_or_panic(crate::apply_store_schema(&pool).await);
result_or_panic(crate::apply_missing_store_schema(&pool).await);
execute_sql(
&pool,
"DELETE FROM k_sol_decode_events WHERE processor_name = 'decode_atomic_rollback_test'",

4
ks-store/src/postgres/query/raw_queries.rs
View File

@@ -1,5 +1,5 @@
// file: ks-store/src/postgres/query/raw_queries.rs
// version: 9
// version: 10
//! PostgreSQL canonical transaction and acquisition observation SQL queries.
@@ -463,7 +463,7 @@ mod tests {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => panic!("unexpected connect error: {error}"),
};
let schema_result = store.initialize_store_schema().await;
let schema_result = store.initialize_missing_store_schema().await;
if let std::result::Result::Err(error) = schema_result {
panic!("unexpected schema error: {error}");
}

695
ks-store/src/postgres/query/replay_candidate_queries.rs
View File

@@ -1,5 +1,5 @@
// file: ks-store/src/postgres/query/replay_candidate_queries.rs
// version: 8
// version: 10
//! Read-only PostgreSQL queries for replay candidate discovery.
@@ -42,18 +42,41 @@ struct ReplayEntitySummaryRow {
max_slot: i64,
}
#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
struct ReplayTransactionPageCursor {
slot: i64,
signature: std::string::String,
newest_first: bool,
}
#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
struct ReplayProgramPageCursor {
transaction_count: i64,
program_id: std::string::String,
}
#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
struct ReplayEntityPageCursor {
entity_kind: std::string::String,
transaction_count: i64,
entity_value: std::string::String,
}
pub(crate) async fn list_replay_transaction_candidates(
pool: &sqlx::PgPool,
filter: &crate::ReplayTransactionFilter,
) -> ks_core::Result<std::vec::Vec<crate::ReplayTransactionCandidate>> {
let min_slot_result =
crate::postgres::query::replay_candidate_queries::optional_sql_bigint(filter.min_slot);
page_request: &crate::PageRequest,
) -> ks_core::Result<crate::CountedPageSlice<crate::ReplayTransactionCandidate>> {
let page_result = validate_page_request(page_request);
if let std::result::Result::Err(error) = page_result {
return std::result::Result::Err(error);
}
let min_slot_result = optional_sql_bigint(filter.min_slot);
let min_slot = match min_slot_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let max_slot_result =
crate::postgres::query::replay_candidate_queries::optional_sql_bigint(filter.max_slot);
let max_slot_result = optional_sql_bigint(filter.max_slot);
let max_slot = match max_slot_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
@@ -62,103 +85,127 @@ pub(crate) async fn list_replay_transaction_candidates(
std::option::Option::Some(value) => std::option::Option::Some(value.as_sql()),
std::option::Option::None => std::option::Option::None,
};
let sql = if filter.newest_first {
crate::postgres::query::replay_candidate_queries::transaction_candidate_sql_desc()
} else {
crate::postgres::query::replay_candidate_queries::transaction_candidate_sql_asc()
let cursor_result =
decode_transaction_cursor(page_request.cursor.as_deref(), filter.newest_first);
let cursor = match cursor_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let total_rows_result =
count_replay_transaction_candidates(pool, filter, min_slot, max_slot, entity_kind).await;
let total_rows = match total_rows_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let sql = if filter.newest_first {
transaction_candidate_sql_desc()
} else {
transaction_candidate_sql_asc()
};
let query_limit = i64::from(page_request.limit) + 1;
let started_at = std::time::Instant::now();
let query_result = sqlx::query_as::<
sqlx::Postgres,
crate::postgres::query::replay_candidate_queries::ReplayTransactionCandidateRow,
>(sql)
.bind(filter.signature_contains.as_deref())
.bind(min_slot)
.bind(max_slot)
.bind(filter.raw_processing_state.as_deref())
.bind(filter.ledger_status.as_deref())
.bind(filter.program_id.as_deref())
.bind(filter.program_scope.as_sql())
.bind(entity_kind)
.bind(filter.entity_value.as_deref())
.bind(i64::from(filter.limit))
.fetch_all(pool)
.await;
let query_result = sqlx::query_as::<sqlx::Postgres, ReplayTransactionCandidateRow>(sql)
.bind(filter.signature_contains.as_deref())
.bind(min_slot)
.bind(max_slot)
.bind(filter.raw_processing_state.as_deref())
.bind(filter.ledger_status.as_deref())
.bind(filter.program_id.as_deref())
.bind(filter.program_scope.as_sql())
.bind(entity_kind)
.bind(filter.entity_value.as_deref())
.bind(cursor.as_ref().map(|value| return value.slot))
.bind(cursor.as_ref().map(|value| return value.signature.as_str()))
.bind(query_limit)
.fetch_all(pool)
.await;
let rows = match query_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => {
trace_query_failure("list_replay_transaction_candidates", started_at, &error);
trace_query_failure("list_replay_transaction_candidate_page", started_at, &error);
return std::result::Result::Err(ks_core::Error::db(format!(
"postgres replay transaction candidate query failed: {error}"
)));
},
};
let mut output = std::vec::Vec::with_capacity(rows.len());
for row in rows {
let mut output = std::vec::Vec::with_capacity(rows.len().min(usize::from(page_request.limit)));
for row in rows.iter().take(usize::from(page_request.limit)) {
output.push(crate::ReplayTransactionCandidate {
signature: row.signature,
signature: row.signature.clone(),
slot: row.slot,
raw_processing_state: row.raw_processing_state,
retention_state: row.retention_state,
raw_processing_state: row.raw_processing_state.clone(),
retention_state: row.retention_state.clone(),
has_core_transaction: row.has_core_transaction,
transaction_failed: row.transaction_failed,
ledger_status: row.ledger_status,
processor_version: row.processor_version,
ledger_status: row.ledger_status.clone(),
processor_version: row.processor_version.clone(),
attempt_count: row.attempt_count,
top_level_instruction_count: row.top_level_instruction_count,
inner_instruction_count: row.inner_instruction_count,
top_level_program_count: row.top_level_program_count,
inner_program_count: row.inner_program_count,
updated_at: row.updated_at,
updated_at: row.updated_at.clone(),
});
}
trace_query_success("list_replay_transaction_candidates", started_at, output.len());
return std::result::Result::Ok(output);
let next_cursor = if rows.len() > usize::from(page_request.limit) {
match output.last() {
std::option::Option::Some(value) => {
let encoded_result = encode_transaction_cursor(value, filter.newest_first);
match encoded_result {
std::result::Result::Ok(value) => std::option::Option::Some(value),
std::result::Result::Err(error) => return std::result::Result::Err(error),
}
},
std::option::Option::None => std::option::Option::None,
}
} else {
std::option::Option::None
};
trace_query_success("list_replay_transaction_candidate_page", started_at, output.len());
return std::result::Result::Ok(crate::CountedPageSlice::new(output, next_cursor, total_rows));
}
pub(crate) async fn list_replay_program_summaries(
pool: &sqlx::PgPool,
filter: &crate::ReplayProgramFilter,
) -> ks_core::Result<std::vec::Vec<crate::ReplayProgramSummary>> {
page_request: &crate::PageRequest,
) -> ks_core::Result<crate::CountedPageSlice<crate::ReplayProgramSummary>> {
let page_result = validate_page_request(page_request);
if let std::result::Result::Err(error) = page_result {
return std::result::Result::Err(error);
}
let cursor_result = decode_program_cursor(page_request.cursor.as_deref());
let cursor = match cursor_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let total_rows_result = count_replay_program_summaries(pool, filter).await;
let total_rows = match total_rows_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let query_limit = i64::from(page_request.limit) + 1;
let started_at = std::time::Instant::now();
let query_result = sqlx::query_as::<sqlx::Postgres, crate::postgres::query::replay_candidate_queries::ReplayProgramSummaryRow>(
r#"WITH occurrences AS (
SELECT program_id, signature, slot, 'top_level'::TEXT AS scope FROM k_sol_core_instructions
UNION ALL
SELECT program_id, signature, slot, 'inner'::TEXT AS scope FROM k_sol_core_inner_instructions
UNION ALL
SELECT program_id, signature, slot, 'logs'::TEXT AS scope FROM k_sol_core_logs WHERE program_id IS NOT NULL
)
SELECT program_id,
COUNT(DISTINCT signature)::BIGINT AS transaction_count,
COUNT(*) FILTER (WHERE scope = 'top_level')::BIGINT AS top_level_instruction_count,
COUNT(*) FILTER (WHERE scope = 'inner')::BIGINT AS inner_instruction_count,
COUNT(*) FILTER (WHERE scope = 'logs')::BIGINT AS log_count,
MIN(slot)::BIGINT AS min_slot,
MAX(slot)::BIGINT AS max_slot
FROM occurrences
WHERE ($1::TEXT IS NULL OR program_id ILIKE '%' || $1 || '%')
GROUP BY program_id
ORDER BY transaction_count DESC, program_id ASC
LIMIT $2"#,
)
.bind(filter.program_id_contains.as_deref())
.bind(i64::from(filter.limit))
.fetch_all(pool)
.await;
let query_result = sqlx::query_as::<sqlx::Postgres, ReplayProgramSummaryRow>(PROGRAM_PAGE_SQL)
.bind(filter.program_id_contains.as_deref())
.bind(cursor.as_ref().map(|value| return value.transaction_count))
.bind(cursor.as_ref().map(|value| return value.program_id.as_str()))
.bind(query_limit)
.fetch_all(pool)
.await;
let rows = match query_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => {
trace_query_failure("list_replay_program_summaries", started_at, &error);
trace_query_failure("list_replay_program_summary_page", started_at, &error);
return std::result::Result::Err(ks_core::Error::db(format!(
"postgres replay program summary query failed: {error}"
)));
},
};
let mut output = std::vec::Vec::with_capacity(rows.len());
for row in rows {
let mut output = std::vec::Vec::with_capacity(rows.len().min(usize::from(page_request.limit)));
for row in rows.iter().take(usize::from(page_request.limit)) {
output.push(crate::ReplayProgramSummary {
program_id: row.program_id,
program_id: row.program_id.clone(),
transaction_count: row.transaction_count,
top_level_instruction_count: row.top_level_instruction_count,
inner_instruction_count: row.inner_instruction_count,
@@ -167,74 +214,429 @@ pub(crate) async fn list_replay_program_summaries(
max_slot: row.max_slot,
});
}
trace_query_success("list_replay_program_summaries", started_at, output.len());
return std::result::Result::Ok(output);
let next_cursor = if rows.len() > usize::from(page_request.limit) {
match output.last() {
std::option::Option::Some(value) => {
let encoded_result = encode_program_cursor(value);
match encoded_result {
std::result::Result::Ok(value) => std::option::Option::Some(value),
std::result::Result::Err(error) => return std::result::Result::Err(error),
}
},
std::option::Option::None => std::option::Option::None,
}
} else {
std::option::Option::None
};
trace_query_success("list_replay_program_summary_page", started_at, output.len());
return std::result::Result::Ok(crate::CountedPageSlice::new(output, next_cursor, total_rows));
}
pub(crate) async fn list_replay_entity_summaries(
pool: &sqlx::PgPool,
filter: &crate::ReplayEntityFilter,
) -> ks_core::Result<std::vec::Vec<crate::ReplayEntitySummary>> {
page_request: &crate::PageRequest,
) -> ks_core::Result<crate::CountedPageSlice<crate::ReplayEntitySummary>> {
let page_result = validate_page_request(page_request);
if let std::result::Result::Err(error) = page_result {
return std::result::Result::Err(error);
}
let entity_kind = filter.entity_kind.as_sql();
let cursor_result = decode_entity_cursor(page_request.cursor.as_deref(), entity_kind);
let cursor = match cursor_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let total_rows_result = count_replay_entity_summaries(pool, filter).await;
let total_rows = match total_rows_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let query_limit = i64::from(page_request.limit) + 1;
let started_at = std::time::Instant::now();
let query_result = sqlx::query_as::<
sqlx::Postgres,
crate::postgres::query::replay_candidate_queries::ReplayEntitySummaryRow,
>(
r#"WITH entities AS (
SELECT 'mint'::TEXT AS entity_kind, mint AS entity_value, signature, slot
FROM k_sol_core_balance_changes
WHERE mint IS NOT NULL
UNION ALL
SELECT 'owner'::TEXT AS entity_kind, owner AS entity_value, signature, slot
FROM k_sol_core_balance_changes
WHERE owner IS NOT NULL
UNION ALL
SELECT 'account_key'::TEXT AS entity_kind, account_key AS entity_value, signature, slot
FROM k_sol_core_account_keys
)
SELECT entity_kind,
entity_value,
COUNT(DISTINCT signature)::BIGINT AS transaction_count,
COUNT(*)::BIGINT AS occurrence_count,
MIN(slot)::BIGINT AS min_slot,
MAX(slot)::BIGINT AS max_slot
FROM entities
WHERE entity_kind = $1
AND ($2::TEXT IS NULL OR entity_value ILIKE '%' || $2 || '%')
GROUP BY entity_kind, entity_value
ORDER BY transaction_count DESC, entity_value ASC
LIMIT $3"#,
)
.bind(entity_kind)
.bind(filter.entity_value_contains.as_deref())
.bind(i64::from(filter.limit))
.fetch_all(pool)
.await;
let query_result = sqlx::query_as::<sqlx::Postgres, ReplayEntitySummaryRow>(ENTITY_PAGE_SQL)
.bind(entity_kind)
.bind(filter.entity_value_contains.as_deref())
.bind(cursor.as_ref().map(|value| return value.transaction_count))
.bind(cursor.as_ref().map(|value| return value.entity_value.as_str()))
.bind(query_limit)
.fetch_all(pool)
.await;
let rows = match query_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => {
trace_query_failure("list_replay_entity_summaries", started_at, &error);
trace_query_failure("list_replay_entity_summary_page", started_at, &error);
return std::result::Result::Err(ks_core::Error::db(format!(
"postgres replay entity summary query failed: {error}"
)));
},
};
let mut output = std::vec::Vec::with_capacity(rows.len());
for row in rows {
let mut output = std::vec::Vec::with_capacity(rows.len().min(usize::from(page_request.limit)));
for row in rows.iter().take(usize::from(page_request.limit)) {
output.push(crate::ReplayEntitySummary {
entity_kind: row.entity_kind,
entity_value: row.entity_value,
entity_kind: row.entity_kind.clone(),
entity_value: row.entity_value.clone(),
transaction_count: row.transaction_count,
occurrence_count: row.occurrence_count,
min_slot: row.min_slot,
max_slot: row.max_slot,
});
}
trace_query_success("list_replay_entity_summaries", started_at, output.len());
return std::result::Result::Ok(output);
let next_cursor = if rows.len() > usize::from(page_request.limit) {
match output.last() {
std::option::Option::Some(value) => {
let encoded_result = encode_entity_cursor(value);
match encoded_result {
std::result::Result::Ok(value) => std::option::Option::Some(value),
std::result::Result::Err(error) => return std::result::Result::Err(error),
}
},
std::option::Option::None => std::option::Option::None,
}
} else {
std::option::Option::None
};
trace_query_success("list_replay_entity_summary_page", started_at, output.len());
return std::result::Result::Ok(crate::CountedPageSlice::new(output, next_cursor, total_rows));
}
async fn count_replay_transaction_candidates(
pool: &sqlx::PgPool,
filter: &crate::ReplayTransactionFilter,
min_slot: std::option::Option<i64>,
max_slot: std::option::Option<i64>,
entity_kind: std::option::Option<&str>,
) -> ks_core::Result<u64> {
let started_at = std::time::Instant::now();
let query_result = sqlx::query_scalar::<sqlx::Postgres, i64>(TRANSACTION_COUNT_SQL)
.bind(filter.signature_contains.as_deref())
.bind(min_slot)
.bind(max_slot)
.bind(filter.raw_processing_state.as_deref())
.bind(filter.ledger_status.as_deref())
.bind(filter.program_id.as_deref())
.bind(filter.program_scope.as_sql())
.bind(entity_kind)
.bind(filter.entity_value.as_deref())
.fetch_one(pool)
.await;
let count = match query_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => {
trace_query_failure("count_replay_transaction_candidates", started_at, &error);
return std::result::Result::Err(ks_core::Error::db(format!(
"postgres replay transaction candidate count failed: {error}"
)));
},
};
trace_query_success("count_replay_transaction_candidates", started_at, 1);
return unsigned_count(count, "replay transaction candidate count is negative");
}
async fn count_replay_program_summaries(
pool: &sqlx::PgPool,
filter: &crate::ReplayProgramFilter,
) -> ks_core::Result<u64> {
let started_at = std::time::Instant::now();
let query_result = sqlx::query_scalar::<sqlx::Postgres, i64>(PROGRAM_COUNT_SQL)
.bind(filter.program_id_contains.as_deref())
.fetch_one(pool)
.await;
let count = match query_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => {
trace_query_failure("count_replay_program_summaries", started_at, &error);
return std::result::Result::Err(ks_core::Error::db(format!(
"postgres replay program summary count failed: {error}"
)));
},
};
trace_query_success("count_replay_program_summaries", started_at, 1);
return unsigned_count(count, "replay program summary count is negative");
}
async fn count_replay_entity_summaries(
pool: &sqlx::PgPool,
filter: &crate::ReplayEntityFilter,
) -> ks_core::Result<u64> {
let started_at = std::time::Instant::now();
let query_result = sqlx::query_scalar::<sqlx::Postgres, i64>(ENTITY_COUNT_SQL)
.bind(filter.entity_kind.as_sql())
.bind(filter.entity_value_contains.as_deref())
.fetch_one(pool)
.await;
let count = match query_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => {
trace_query_failure("count_replay_entity_summaries", started_at, &error);
return std::result::Result::Err(ks_core::Error::db(format!(
"postgres replay entity summary count failed: {error}"
)));
},
};
trace_query_success("count_replay_entity_summaries", started_at, 1);
return unsigned_count(count, "replay entity summary count is negative");
}
fn validate_page_request(page_request: &crate::PageRequest) -> ks_core::Result<()> {
if page_request.limit == 0 || page_request.limit > crate::MAX_PAGE_SIZE {
return std::result::Result::Err(ks_core::Error::db(
"replay candidate page size is outside repository bounds",
));
}
return std::result::Result::Ok(());
}
fn unsigned_count(value: i64, message: &str) -> ks_core::Result<u64> {
let conversion_result = u64::try_from(value);
return match conversion_result {
std::result::Result::Ok(converted) => std::result::Result::Ok(converted),
std::result::Result::Err(_error) => std::result::Result::Err(ks_core::Error::db(message)),
};
}
fn decode_transaction_cursor(
cursor: std::option::Option<&str>,
newest_first: bool,
) -> ks_core::Result<std::option::Option<ReplayTransactionPageCursor>> {
let decoded_result = decode_cursor::<ReplayTransactionPageCursor>(cursor, "transaction");
let decoded = match decoded_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
if decoded.as_ref().is_some_and(|value| {
return value.slot < 0
|| value.signature.trim().is_empty()
|| value.newest_first != newest_first;
}) {
return std::result::Result::Err(ks_core::Error::db(
"replay transaction page cursor values are invalid",
));
}
return std::result::Result::Ok(decoded);
}
fn decode_program_cursor(
cursor: std::option::Option<&str>,
) -> ks_core::Result<std::option::Option<ReplayProgramPageCursor>> {
let decoded_result = decode_cursor::<ReplayProgramPageCursor>(cursor, "program");
let decoded = match decoded_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
if decoded.as_ref().is_some_and(|value| {
return value.transaction_count < 0 || value.program_id.trim().is_empty();
}) {
return std::result::Result::Err(ks_core::Error::db(
"replay program page cursor values are invalid",
));
}
return std::result::Result::Ok(decoded);
}
fn decode_entity_cursor(
cursor: std::option::Option<&str>,
entity_kind: &str,
) -> ks_core::Result<std::option::Option<ReplayEntityPageCursor>> {
let decoded_result = decode_cursor::<ReplayEntityPageCursor>(cursor, "entity");
let decoded = match decoded_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
if decoded.as_ref().is_some_and(|value| {
return value.transaction_count < 0
|| value.entity_value.trim().is_empty()
|| value.entity_kind != entity_kind;
}) {
return std::result::Result::Err(ks_core::Error::db(
"replay entity page cursor values are invalid",
));
}
return std::result::Result::Ok(decoded);
}
fn decode_cursor<T>(
cursor: std::option::Option<&str>,
label: &str,
) -> ks_core::Result<std::option::Option<T>>
where
T: serde::de::DeserializeOwned,
{
let raw = match cursor {
std::option::Option::Some(value) => value,
std::option::Option::None => return std::result::Result::Ok(std::option::Option::None),
};
let decode_result = serde_json::from_str::<T>(raw);
return match decode_result {
std::result::Result::Ok(value) => std::result::Result::Ok(std::option::Option::Some(value)),
std::result::Result::Err(_error) => std::result::Result::Err(ks_core::Error::db(format!(
"replay {label} page cursor is invalid"
))),
};
}
fn encode_transaction_cursor(
row: &crate::ReplayTransactionCandidate,
newest_first: bool,
) -> ks_core::Result<std::string::String> {
return encode_cursor(
&ReplayTransactionPageCursor {
slot: row.slot,
signature: row.signature.clone(),
newest_first,
},
"transaction",
);
}
fn encode_program_cursor(
row: &crate::ReplayProgramSummary,
) -> ks_core::Result<std::string::String> {
return encode_cursor(
&ReplayProgramPageCursor {
transaction_count: row.transaction_count,
program_id: row.program_id.clone(),
},
"program",
);
}
fn encode_entity_cursor(row: &crate::ReplayEntitySummary) -> ks_core::Result<std::string::String> {
return encode_cursor(
&ReplayEntityPageCursor {
entity_kind: row.entity_kind.clone(),
transaction_count: row.transaction_count,
entity_value: row.entity_value.clone(),
},
"entity",
);
}
fn encode_cursor<T>(value: &T, label: &str) -> ks_core::Result<std::string::String>
where
T: serde::Serialize,
{
let encode_result = serde_json::to_string(value);
return match encode_result {
std::result::Result::Ok(encoded) if encoded.len() <= crate::MAX_PAGE_CURSOR_LENGTH => {
std::result::Result::Ok(encoded)
},
std::result::Result::Ok(_encoded) => std::result::Result::Err(ks_core::Error::db(format!(
"replay {label} page cursor exceeds maximum encoded length"
))),
std::result::Result::Err(error) => std::result::Result::Err(ks_core::Error::db(format!(
"replay {label} page cursor serialization failed: {error}"
))),
};
}
const TRANSACTION_COUNT_SQL: &str = r#"SELECT COUNT(*)::BIGINT
FROM k_sol_raw_transactions raw
LEFT JOIN LATERAL (
SELECT status
FROM k_sol_ops_processing_ledger
WHERE stage = 'core_extraction'
AND processor_name = 'canonical_to_core'
AND input_key = raw.signature
ORDER BY updated_at DESC, id DESC
LIMIT 1
) ledger ON TRUE
WHERE ($1::TEXT IS NULL OR raw.signature ILIKE '%' || $1 || '%')
AND ($2::BIGINT IS NULL OR raw.slot >= $2)
AND ($3::BIGINT IS NULL OR raw.slot <= $3)
AND ($4::TEXT IS NULL OR raw.processing_state = $4)
AND ($5::TEXT IS NULL OR ($5 = 'not_started' AND ledger.status IS NULL) OR ledger.status = $5)
AND ($6::TEXT IS NULL OR
($7 = 'any' AND (
EXISTS (SELECT 1 FROM k_sol_core_instructions candidate_top_level WHERE candidate_top_level.signature = raw.signature AND candidate_top_level.program_id = $6) OR
EXISTS (SELECT 1 FROM k_sol_core_inner_instructions candidate_inner WHERE candidate_inner.signature = raw.signature AND candidate_inner.program_id = $6) OR
EXISTS (SELECT 1 FROM k_sol_core_logs candidate_log WHERE candidate_log.signature = raw.signature AND candidate_log.program_id = $6)
)) OR
($7 = 'top_level' AND EXISTS (SELECT 1 FROM k_sol_core_instructions candidate_top_level WHERE candidate_top_level.signature = raw.signature AND candidate_top_level.program_id = $6)) OR
($7 = 'inner' AND EXISTS (SELECT 1 FROM k_sol_core_inner_instructions candidate_inner WHERE candidate_inner.signature = raw.signature AND candidate_inner.program_id = $6)) OR
($7 = 'logs' AND EXISTS (SELECT 1 FROM k_sol_core_logs candidate_log WHERE candidate_log.signature = raw.signature AND candidate_log.program_id = $6)))
AND ($8::TEXT IS NULL OR
($8 = 'mint' AND EXISTS (SELECT 1 FROM k_sol_core_balance_changes candidate_balance WHERE candidate_balance.signature = raw.signature AND candidate_balance.mint = $9)) OR
($8 = 'owner' AND EXISTS (SELECT 1 FROM k_sol_core_balance_changes candidate_balance WHERE candidate_balance.signature = raw.signature AND candidate_balance.owner = $9)) OR
($8 = 'account_key' AND EXISTS (SELECT 1 FROM k_sol_core_account_keys candidate_account WHERE candidate_account.signature = raw.signature AND candidate_account.account_key = $9)))"#;
const PROGRAM_PAGE_SQL: &str = r#"WITH occurrences AS (
SELECT program_id, signature, slot, 'top_level'::TEXT AS scope FROM k_sol_core_instructions
UNION ALL
SELECT program_id, signature, slot, 'inner'::TEXT AS scope FROM k_sol_core_inner_instructions
UNION ALL
SELECT program_id, signature, slot, 'logs'::TEXT AS scope FROM k_sol_core_logs WHERE program_id IS NOT NULL
), summaries AS (
SELECT program_id,
COUNT(DISTINCT signature)::BIGINT AS transaction_count,
COUNT(*) FILTER (WHERE scope = 'top_level')::BIGINT AS top_level_instruction_count,
COUNT(*) FILTER (WHERE scope = 'inner')::BIGINT AS inner_instruction_count,
COUNT(*) FILTER (WHERE scope = 'logs')::BIGINT AS log_count,
MIN(slot)::BIGINT AS min_slot,
MAX(slot)::BIGINT AS max_slot
FROM occurrences
WHERE ($1::TEXT IS NULL OR program_id ILIKE '%' || $1 || '%')
GROUP BY program_id
)
SELECT program_id, transaction_count, top_level_instruction_count, inner_instruction_count, log_count, min_slot, max_slot
FROM summaries
WHERE ($2::BIGINT IS NULL OR transaction_count < $2 OR (transaction_count = $2 AND program_id > $3))
ORDER BY transaction_count DESC, program_id ASC
LIMIT $4"#;
const PROGRAM_COUNT_SQL: &str = r#"WITH occurrences AS (
SELECT program_id FROM k_sol_core_instructions
UNION ALL
SELECT program_id FROM k_sol_core_inner_instructions
UNION ALL
SELECT program_id FROM k_sol_core_logs WHERE program_id IS NOT NULL
), summaries AS (
SELECT program_id
FROM occurrences
WHERE ($1::TEXT IS NULL OR program_id ILIKE '%' || $1 || '%')
GROUP BY program_id
)
SELECT COUNT(*)::BIGINT FROM summaries"#;
const ENTITY_PAGE_SQL: &str = r#"WITH entities AS (
SELECT 'mint'::TEXT AS entity_kind, mint AS entity_value, signature, slot FROM k_sol_core_balance_changes WHERE mint IS NOT NULL
UNION ALL
SELECT 'owner'::TEXT AS entity_kind, owner AS entity_value, signature, slot FROM k_sol_core_balance_changes WHERE owner IS NOT NULL
UNION ALL
SELECT 'account_key'::TEXT AS entity_kind, account_key AS entity_value, signature, slot FROM k_sol_core_account_keys
), summaries AS (
SELECT entity_kind,
entity_value,
COUNT(DISTINCT signature)::BIGINT AS transaction_count,
COUNT(*)::BIGINT AS occurrence_count,
MIN(slot)::BIGINT AS min_slot,
MAX(slot)::BIGINT AS max_slot
FROM entities
WHERE entity_kind = $1
AND ($2::TEXT IS NULL OR entity_value ILIKE '%' || $2 || '%')
GROUP BY entity_kind, entity_value
)
SELECT entity_kind, entity_value, transaction_count, occurrence_count, min_slot, max_slot
FROM summaries
WHERE ($3::BIGINT IS NULL OR transaction_count < $3 OR (transaction_count = $3 AND entity_value > $4))
ORDER BY transaction_count DESC, entity_value ASC
LIMIT $5"#;
const ENTITY_COUNT_SQL: &str = r#"WITH entities AS (
SELECT 'mint'::TEXT AS entity_kind, mint AS entity_value FROM k_sol_core_balance_changes WHERE mint IS NOT NULL
UNION ALL
SELECT 'owner'::TEXT AS entity_kind, owner AS entity_value FROM k_sol_core_balance_changes WHERE owner IS NOT NULL
UNION ALL
SELECT 'account_key'::TEXT AS entity_kind, account_key AS entity_value FROM k_sol_core_account_keys
), summaries AS (
SELECT entity_kind, entity_value
FROM entities
WHERE entity_kind = $1
AND ($2::TEXT IS NULL OR entity_value ILIKE '%' || $2 || '%')
GROUP BY entity_kind, entity_value
)
SELECT COUNT(*)::BIGINT FROM summaries"#;
fn transaction_candidate_sql_desc() -> &'static str {
return crate::postgres::query::replay_candidate_queries::transaction_candidate_sql("DESC");
}
@@ -310,8 +712,9 @@ fn transaction_candidate_sql(order: &str) -> &'static str {
WHERE candidate_account.signature = raw.signature
AND candidate_account.account_key = $9
)))
AND ($10::BIGINT IS NULL OR raw.slot < $10 OR (raw.slot = $10 AND raw.signature < $11))
ORDER BY raw.slot DESC, raw.signature DESC
LIMIT $10"#;
LIMIT $12"#;
}
return r#"SELECT raw.signature,
raw.slot,
@@ -378,8 +781,9 @@ fn transaction_candidate_sql(order: &str) -> &'static str {
WHERE candidate_account.signature = raw.signature
AND candidate_account.account_key = $9
)))
AND ($10::BIGINT IS NULL OR raw.slot > $10 OR (raw.slot = $10 AND raw.signature > $11))
ORDER BY raw.slot ASC, raw.signature ASC
LIMIT $10"#;
LIMIT $12"#;
}
fn optional_sql_bigint(
@@ -413,6 +817,49 @@ fn trace_query_failure(action: &str, started_at: std::time::Instant, error: &sql
#[cfg(test)]
mod tests {
#[test]
fn replay_candidate_queries_are_cursor_based_and_counted() {
let source = include_str!("replay_candidate_queries.rs");
let tests_marker = source.find("#[cfg(test)]");
let production_source = match tests_marker {
std::option::Option::Some(index) => &source[..index],
std::option::Option::None => source,
};
assert!(!production_source.contains(" OFFSET "));
assert!(production_source.contains("TRANSACTION_COUNT_SQL"));
assert!(production_source.contains("PROGRAM_COUNT_SQL"));
assert!(production_source.contains("ENTITY_COUNT_SQL"));
assert!(production_source.contains("CountedPageSlice::new"));
}
#[test]
fn transaction_cursor_rejects_order_direction_changes() {
let row = crate::ReplayTransactionCandidate {
signature: "signature".to_string(),
slot: 42,
raw_processing_state: "received".to_string(),
retention_state: "retained".to_string(),
has_core_transaction: false,
transaction_failed: std::option::Option::None,
ledger_status: "not_started".to_string(),
processor_version: std::option::Option::None,
attempt_count: 0,
top_level_instruction_count: 0,
inner_instruction_count: 0,
top_level_program_count: 0,
inner_program_count: 0,
updated_at: "now".to_string(),
};
let encoded_result = super::encode_transaction_cursor(&row, true);
let encoded = match encoded_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => panic!("unexpected encode error: {error}"),
};
let decoded_result =
super::decode_transaction_cursor(std::option::Option::Some(encoded.as_str()), false);
assert!(decoded_result.is_err());
}
#[tokio::test]
async fn optional_postgres_replay_candidate_queries_from_env() {
let database_url = match std::env::var("KS_SECRET_POSTGRES_TEST_URL") {
@@ -430,7 +877,7 @@ mod tests {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => panic!("unexpected connect error: {error}"),
};
let schema_result = store.initialize_store_schema().await;
let schema_result = store.initialize_missing_store_schema().await;
if let std::result::Result::Err(error) = schema_result {
panic!("unexpected store schema error: {error}");
}
@@ -444,7 +891,6 @@ mod tests {
crate::ReplayProgramScope::Any,
std::option::Option::None,
std::option::Option::None,
10,
true,
);
let transaction_filter = match transaction_filter_result {
@@ -453,16 +899,24 @@ mod tests {
panic!("unexpected transaction filter error: {error}")
},
};
let transaction_result = store.replay_transaction_candidates(&transaction_filter).await;
let transaction_page = crate::PageRequest::new(10, std::option::Option::None);
let transaction_page = match transaction_page {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => panic!("unexpected transaction page error: {error}"),
};
let transaction_result = store
.replay_transaction_candidates(&transaction_filter, &transaction_page)
.await;
if let std::result::Result::Err(error) = transaction_result {
panic!("unexpected transaction candidate query error: {error}");
}
let program_filter_result = crate::ReplayProgramFilter::new(std::option::Option::None, 10);
let program_filter_result = crate::ReplayProgramFilter::new(std::option::Option::None);
let program_filter = match program_filter_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => panic!("unexpected program filter error: {error}"),
};
let program_result = store.replay_program_summaries(&program_filter).await;
let program_result =
store.replay_program_summaries(&program_filter, &transaction_page).await;
if let std::result::Result::Err(error) = program_result {
panic!("unexpected program summary query error: {error}");
}
@@ -472,14 +926,15 @@ mod tests {
crate::ReplayEntityKind::AccountKey,
] {
let entity_filter_result =
crate::ReplayEntityFilter::new(entity_kind, std::option::Option::None, 10);
crate::ReplayEntityFilter::new(entity_kind, std::option::Option::None);
let entity_filter = match entity_filter_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => {
panic!("unexpected entity filter error: {error}")
},
};
let entity_result = store.replay_entity_summaries(&entity_filter).await;
let entity_result =
store.replay_entity_summaries(&entity_filter, &transaction_page).await;
if let std::result::Result::Err(error) = entity_result {
panic!("unexpected entity summary query error: {error}");
}

79
ks-store/src/postgres/query/schema_queries.rs
View File

@@ -1,23 +1,20 @@
// file: ks-store/src/postgres/query/schema_queries.rs
// version: 1
// version: 5
//! PostgreSQL orchestration for the complete store baseline schema.
//! PostgreSQL additive-only orchestration for the managed store schema.
pub(crate) async fn apply_store_schema(pool: &sqlx::PgPool) -> ks_core::Result<()> {
pub(crate) async fn apply_missing_store_schema(pool: &sqlx::PgPool) -> ks_core::Result<()> {
let resource_validation_result = crate::validate_postgres_sql_resources();
if let std::result::Result::Err(error) = resource_validation_result {
return std::result::Result::Err(error);
}
let legacy_result = legacy_store_schema_detected(pool).await;
let legacy_detected = match legacy_result {
let statements_result = crate::load_postgres_schema_addition_statements(pool).await;
let statements = match statements_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
if legacy_detected {
return std::result::Result::Err(crate::storage_contract_error(
"store_legacy_schema_detected",
"historical store schema detected; rebuild the database before initialization",
));
if statements.is_empty() {
return std::result::Result::Ok(());
}
let transaction_result = pool.begin().await;
let mut transaction = match transaction_result {
@@ -37,20 +34,8 @@ pub(crate) async fn apply_store_schema(pool: &sqlx::PgPool) -> ks_core::Result<(
"postgres store schema advisory lock failed: {error}"
)));
}
let raw_result =
execute_schema_group(&mut transaction, "raw", crate::raw_store_schema_statements()).await;
if let std::result::Result::Err(error) = raw_result {
return std::result::Result::Err(error);
}
let core_result =
execute_schema_group(&mut transaction, "core", crate::core_store_schema_statements()).await;
if let std::result::Result::Err(error) = core_result {
return std::result::Result::Err(error);
}
let decode_result =
execute_schema_group(&mut transaction, "decode", crate::decode_store_schema_statements())
.await;
if let std::result::Result::Err(error) = decode_result {
let execution_result = execute_additive_schema_group(&mut transaction, statements).await;
if let std::result::Result::Err(error) = execution_result {
return std::result::Result::Err(error);
}
let commit_result = transaction.commit().await;
@@ -62,57 +47,39 @@ pub(crate) async fn apply_store_schema(pool: &sqlx::PgPool) -> ks_core::Result<(
return std::result::Result::Ok(());
}
async fn execute_schema_group(
async fn execute_additive_schema_group(
transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
model: &str,
statements: std::vec::Vec<&'static str>,
statements: std::vec::Vec<std::string::String>,
) -> ks_core::Result<()> {
for (statement_index, statement) in statements.into_iter().enumerate() {
tracing::trace!(
target: crate::TRACING_TARGET,
backend = "postgres",
domain = "ks-store.pg",
action = "schema_statement",
model,
action = "schema_additive_statement",
statement_index,
sql = statement,
"execute PostgreSQL schema statement"
"execute additive PostgreSQL schema statement"
);
let execution_result = sqlx::query(statement).execute(&mut **transaction).await;
let execution_result =
sqlx::query(sqlx::AssertSqlSafe(statement)).execute(&mut **transaction).await;
if let std::result::Result::Err(error) = execution_result {
tracing::error!(
target: crate::TRACING_TARGET,
backend = "postgres",
domain = "ks-store.pg",
action = "schema_statement",
model,
action = "schema_additive_statement",
statement_index,
"PostgreSQL schema statement failed"
"additive PostgreSQL schema statement failed"
);
return std::result::Result::Err(ks_core::Error::db(format!(
"postgres store schema statement failed for model {model} at index {statement_index}: {error}"
"postgres additive store schema statement failed at index {statement_index}: {error}"
)));
}
}
return std::result::Result::Ok(());
}
async fn legacy_store_schema_detected(pool: &sqlx::PgPool) -> ks_core::Result<bool> {
let pattern = format!("{}%", crate::LEGACY_SOLANA_TABLE_PREFIX);
let query_result = sqlx::query_scalar::<sqlx::Postgres, bool>(
"SELECT EXISTS(SELECT 1 FROM pg_class object JOIN pg_namespace namespace ON namespace.oid = object.relnamespace WHERE namespace.nspname = current_schema() AND object.relkind IN ('r', 'p', 'S', 'i') AND object.relname LIKE $1)",
)
.bind(pattern)
.fetch_one(pool)
.await;
return match query_result {
std::result::Result::Ok(value) => std::result::Result::Ok(value),
std::result::Result::Err(error) => std::result::Result::Err(ks_core::Error::db(format!(
"postgres historical schema detection failed: {error}"
))),
};
}
pub(crate) async fn load_expected_index_counts(pool: &sqlx::PgPool) -> ks_core::Result<(u32, u32)> {
let names = crate::expected_postgres_index_names()
.iter()
@@ -150,13 +117,3 @@ pub(crate) async fn load_expected_index_counts(pool: &sqlx::PgPool) -> ks_core::
};
return std::result::Result::Ok((expected_count, available_count));
}
#[cfg(test)]
mod tests {
#[test]
fn legacy_prefix_is_distinct_from_the_candidate_baseline_prefix() {
assert_eq!(crate::LEGACY_SOLANA_TABLE_PREFIX, "kb_sol_");
assert!(crate::RAW_TRANSACTIONS_TABLE_NAME.starts_with("k_sol_"));
assert!(!crate::RAW_TRANSACTIONS_TABLE_NAME.starts_with(crate::LEGACY_SOLANA_TABLE_PREFIX));
}
}

1645
ks-store/src/postgres/schema_contract.rs Normal file
View File

File diff suppressed because it is too large Load Diff

86
ks-store/src/postgres/store.rs
View File

@@ -1,5 +1,5 @@
// file: ks-store/src/postgres/store.rs
// version: 13
// version: 17
//! PostgreSQL store implementation kept behind the backend-agnostic `Store` facade.
@@ -214,12 +214,29 @@ impl crate::PostgresStore {
return crate::load_expected_index_counts(&self.pool).await;
}
/// Applies each idempotent store schema once per invocation in dependency order.
pub(crate) async fn initialize_store_schema(&self) -> ks_core::Result<()> {
tracing::debug!(target: crate::TRACING_TARGET, backend = "postgres", domain = "ks-store.pg", action = "schema_initialize", "initialize PostgreSQL store schema");
let result = crate::apply_store_schema(&self.pool).await;
/// Verifies existing managed PostgreSQL tables, columns, keys, and indexes.
pub(crate) async fn verify_managed_schema_compatibility(
&self,
) -> ks_core::Result<crate::PostgresSchemaCompatibilitySummary> {
let result = crate::load_postgres_schema_compatibility(&self.pool).await;
return match result {
std::result::Result::Ok(summary) => {
tracing::debug!(target: crate::TRACING_TARGET, backend = "postgres", domain = "ks-store.pg", action = "schema_compatibility_verify", compatible = true, complete = summary.is_complete(), expected_tables = summary.expected_table_count, compatible_tables = summary.compatible_table_count, expected_columns = summary.expected_column_count, compatible_columns = summary.compatible_column_count, expected_keys = summary.expected_key_count, compatible_keys = summary.compatible_key_count, expected_indexes = summary.expected_index_count, compatible_indexes = summary.compatible_index_count, missing_tables = summary.missing_table_names.len(), "PostgreSQL managed schema compatibility verified");
std::result::Result::Ok(summary)
},
std::result::Result::Err(error) => {
tracing::error!(target: crate::TRACING_TARGET, backend = "postgres", domain = "ks-store.pg", action = "schema_compatibility_verify", compatible = false, "PostgreSQL managed schema compatibility verification failed");
std::result::Result::Err(error)
},
};
}
/// Applies additive-only initialization for absent managed tables, columns, keys, and indexes.
pub(crate) async fn initialize_missing_store_schema(&self) -> ks_core::Result<()> {
tracing::debug!(target: crate::TRACING_TARGET, backend = "postgres", domain = "ks-store.pg", action = "schema_initialize_missing", "initialize missing PostgreSQL store schema resources additively");
let result = crate::apply_missing_store_schema(&self.pool).await;
if result.is_ok() {
tracing::debug!(target: crate::TRACING_TARGET, backend = "postgres", domain = "ks-store.pg", action = "schema_initialize", initialized = true, "PostgreSQL store schema initialized");
tracing::debug!(target: crate::TRACING_TARGET, backend = "postgres", domain = "ks-store.pg", action = "schema_initialize_missing", initialized = true, "missing PostgreSQL store schema resources initialized additively");
}
return result;
}
@@ -282,6 +299,30 @@ impl crate::PostgresStore {
));
}
if available_count == expected_count {
let compatibility_result = self.verify_managed_schema_compatibility().await;
let compatibility = match compatibility_result {
std::result::Result::Ok(value) => value,
std::result::Result::Err(_error) => {
return std::result::Result::Ok(crate::StoreMigrationSnapshot::new(
crate::StoreMigrationStatus::Drift,
std::option::Option::None,
std::vec::Vec::new(),
std::option::Option::Some(std::string::String::from(
"managed store schema is structurally incompatible",
)),
));
},
};
if !compatibility.is_complete() {
return std::result::Result::Ok(crate::StoreMigrationSnapshot::new(
crate::StoreMigrationStatus::Drift,
std::option::Option::None,
std::vec::Vec::new(),
std::option::Option::Some(std::string::String::from(
"managed store schema is incomplete",
)),
));
}
return std::result::Result::Ok(crate::StoreMigrationSnapshot::new(
crate::StoreMigrationStatus::Current,
std::option::Option::Some(crate::STORE_SCHEMA_CONTRACT_VERSION.to_string()),
@@ -329,31 +370,40 @@ impl crate::PostgresStore {
});
}
/// Lists bounded raw transaction candidates enriched with Core and ledger diagnostics.
/// Lists one counted raw-transaction candidate block.
pub(crate) async fn replay_transaction_candidates(
&self,
filter: &crate::ReplayTransactionFilter,
) -> ks_core::Result<std::vec::Vec<crate::ReplayTransactionCandidate>> {
tracing::debug!(target: crate::TRACING_TARGET, backend = "postgres", domain = "ks-store.pg", action = "replay_transaction_candidates", "query PostgreSQL replay transaction candidates");
return crate::list_replay_transaction_candidates(&self.pool, filter).await;
page_request: &crate::PageRequest,
) -> ks_core::Result<crate::CountedPageSlice<crate::ReplayTransactionCandidate>> {
return crate::list_replay_transaction_candidates(&self.pool, filter, page_request).await;
}
/// Lists bounded program summaries across top-level, inner and reliably linked logs.
/// Lists one counted program-summary block.
pub(crate) async fn replay_program_summaries(
&self,
filter: &crate::ReplayProgramFilter,
) -> ks_core::Result<std::vec::Vec<crate::ReplayProgramSummary>> {
tracing::debug!(target: crate::TRACING_TARGET, backend = "postgres", domain = "ks-store.pg", action = "replay_program_summaries", "query PostgreSQL replay program summaries");
return crate::list_replay_program_summaries(&self.pool, filter).await;
page_request: &crate::PageRequest,
) -> ks_core::Result<crate::CountedPageSlice<crate::ReplayProgramSummary>> {
return crate::list_replay_program_summaries(&self.pool, filter, page_request).await;
}
/// Lists bounded mint, owner or account-key summaries from Core facts.
/// Lists one counted mint, owner or account-key summary block.
pub(crate) async fn replay_entity_summaries(
&self,
filter: &crate::ReplayEntityFilter,
) -> ks_core::Result<std::vec::Vec<crate::ReplayEntitySummary>> {
tracing::debug!(target: crate::TRACING_TARGET, backend = "postgres", domain = "ks-store.pg", action = "replay_entity_summaries", "query PostgreSQL replay entity summaries");
return crate::list_replay_entity_summaries(&self.pool, filter).await;
page_request: &crate::PageRequest,
) -> ks_core::Result<crate::CountedPageSlice<crate::ReplayEntitySummary>> {
return crate::list_replay_entity_summaries(&self.pool, filter, page_request).await;
}
/// Lists one counted materialized-output block using stable cursor pagination.
pub(crate) async fn materialized_output_page(
&self,
filter: &crate::MaterializedOutputPageFilter,
page_request: &crate::PageRequest,
) -> ks_core::Result<crate::CountedPageSlice<crate::MaterializedOutputQueryRow>> {
return crate::list_materialized_output_page(&self.pool, filter, page_request).await;
}
/// Reads diagnostics for raw store resources without changing the schema.