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

0.3.16-pre.008

Browse Source
This commit is contained in:
SinuS von SifriduS
2026-09-22 07:40:23 +02:00
parent de81084560
commit 14f89262ab
8 changed files with 533 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
crates/ksp-job-backfill-lib/unit_tests/persistence.rs
View File

@@ -1,5 +1,5 @@
// file: crates/ksp-job-backfill-lib/unit_tests/persistence.rs
// version: 2
// version: 3
#[derive(Clone, Copy)]
enum FakeResponse {
@@ -299,6 +299,31 @@ async fn v0_3_16_pre_007_durable_quarantined_conflict_preserves_observation_proj
return;
}
#[tokio::test]
async fn v0_3_16_pre_008_reobserved_quarantined_conflict_preserves_idempotent_observation() {
let (reference, transaction, observation) = match raw_acquisition_parts("devnet", 6, 18) {
std::option::Option::Some(parts) => parts,
std::option::Option::None => return,
};
let network = match raw_network("devnet") {
std::option::Option::Some(network) => network,
std::option::Option::None => return,
};
let response = store_variant_outcome(
ksp_store_lib::RawEntityWriteOutcome::AlreadyPresent,
ksp_store_lib::RawObservationWriteOutcome::AlreadyPresent,
ksp_store_lib::RawTransactionVariantWriteOutcome::QuarantinedConflict,
);
let port = FakePersistencePort::new(network, &[FakeResponse::Outcome(response)]);
let result = super::persist_available_with_port(&port, reference, transaction, observation).await;
assert!(result.is_ok());
if let std::result::Result::Ok(result) = result {
assert_eq!(result.entity(), crate::BackfillEntityPersistence::Conflict);
assert_eq!(result.observation(), crate::BackfillObservationPersistence::AlreadyPresent);
}
return;
}
#[tokio::test]
async fn pre_007_legacy_store_content_conflict_error_is_explicit_and_not_idempotent_success() {
let (reference, transaction, observation) = match raw_acquisition_parts("devnet", 6, 16) {

5
crates/ksp-store-postgres-lib/tests/hardening_completeness.rs
View File

@@ -1,5 +1,5 @@
// file: crates/ksp-store-postgres-lib/tests/hardening_completeness.rs
// version: 29
// version: 30
#![warn(missing_docs)]
#![deny(unreachable_pub)]
@@ -284,6 +284,9 @@ fn pre_009_live_raw_transaction_proof_is_opt_in_isolated_and_secret_safe() {
"prove_schema_update_policy",
"prove_concurrent_identical_insert",
"prove_concurrent_divergent_insert",
"prove_conflict_case_rollback",
"QuarantinedConflict",
"conflict_state_is_exact",
"prove_pagination",
"prove_inspection",
"prove_retention_and_rehydrate",

176
crates/ksp-store-postgres-lib/tests/postgres_raw_transaction_live.rs
View File

@@ -1,5 +1,5 @@
// file: crates/ksp-store-postgres-lib/tests/postgres_raw_transaction_live.rs
// version: 9
// version: 10
#![warn(missing_docs)]
#![deny(unreachable_pub)]
@@ -12,6 +12,68 @@
//! drops only the isolated schema it proved absent before the run.
const LIVE_CANCEL_WAIT: std::time::Duration = std::time::Duration::from_millis(300);
const LIVE_CONFLICT_STATE_SQL: &str = r#"SELECT
(SELECT COUNT(*) = 2 FROM ksp_raw_transaction_variants WHERE transaction_signature = $1)
AND (SELECT COUNT(*) = 1 FROM ksp_raw_transaction_canonical_selectors WHERE transaction_signature = $1)
AND (SELECT COUNT(*) = 2 FROM ksp_raw_transaction_observations WHERE transaction_signature = $1)
AND (SELECT COUNT(*) = 2 FROM ksp_raw_transaction_observation_variants WHERE transaction_signature = $1)
AND EXISTS (
SELECT 1
FROM ksp_raw_transaction_conflicts AS conflict
INNER JOIN ksp_raw_transaction_canonical_selectors AS selector
ON selector.transaction_signature = conflict.transaction_signature
WHERE conflict.transaction_signature = $1
AND conflict.status = 'open'
AND conflict.revision = 1
AND conflict.canonical_variant_id = selector.canonical_variant_id
AND conflict.incoming_variant_id <> conflict.canonical_variant_id
)
AND EXISTS (
SELECT 1
FROM ksp_raw_transactions AS canonical
INNER JOIN ksp_raw_transaction_canonical_selectors AS selector
ON selector.transaction_signature = canonical.signature
INNER JOIN ksp_raw_transaction_variants AS variant
ON variant.transaction_signature = selector.transaction_signature
AND variant.variant_id = selector.canonical_variant_id
WHERE canonical.signature = $1
AND variant.slot = canonical.slot
AND variant.block_time_unix_millis IS NOT DISTINCT FROM canonical.block_time_unix_millis
AND variant.format_id = canonical.format_id
AND variant.format_version = canonical.format_version
AND variant.content_hash = canonical.content_hash
AND variant.payload = canonical.payload
)
AND EXISTS (
SELECT 1 FROM ksp_raw_transaction_observation_variants
WHERE observation_key = $2 AND transaction_signature = $1
)
AND EXISTS (
SELECT 1 FROM ksp_raw_transaction_observation_variants
WHERE observation_key = $3 AND transaction_signature = $1
)"#;
const LIVE_CONFLICT_ROLLBACK_STATE_SQL: &str = r#"SELECT
(SELECT COUNT(*) = 1 FROM ksp_raw_transaction_variants WHERE transaction_signature = $1)
AND (SELECT COUNT(*) = 1 FROM ksp_raw_transaction_canonical_selectors WHERE transaction_signature = $1)
AND (SELECT COUNT(*) = 0 FROM ksp_raw_transaction_conflicts WHERE transaction_signature = $1)
AND (SELECT COUNT(*) = 1 FROM ksp_raw_transaction_observations WHERE transaction_signature = $1)
AND (SELECT COUNT(*) = 1 FROM ksp_raw_transaction_observation_variants WHERE transaction_signature = $1)
AND EXISTS (
SELECT 1
FROM ksp_raw_transactions AS canonical
INNER JOIN ksp_raw_transaction_canonical_selectors AS selector
ON selector.transaction_signature = canonical.signature
INNER JOIN ksp_raw_transaction_variants AS variant
ON variant.transaction_signature = selector.transaction_signature
AND variant.variant_id = selector.canonical_variant_id
WHERE canonical.signature = $1
AND variant.slot = canonical.slot
AND variant.block_time_unix_millis IS NOT DISTINCT FROM canonical.block_time_unix_millis
AND variant.format_id = canonical.format_id
AND variant.format_version = canonical.format_version
AND variant.content_hash = canonical.content_hash
AND variant.payload = canonical.payload
)"#;
const LIVE_INDEX_EXISTS_SQL: &str = r#"SELECT EXISTS (
SELECT 1 FROM pg_indexes
WHERE schemaname = current_schema()
@@ -214,6 +276,10 @@ async fn run_raw_transaction_scenario(admin: &mut tokio_postgres::Client, uri: &
if let std::result::Result::Err(error) = divergent_result {
return std::result::Result::Err(error);
}
let conflict_rollback_result = prove_conflict_case_rollback(admin, uri).await;
if let std::result::Result::Err(error) = conflict_rollback_result {
return std::result::Result::Err(error);
}
let race_result = prove_retention_races(uri).await;
if let std::result::Result::Err(error) = race_result {
return std::result::Result::Err(error);
@@ -414,23 +480,61 @@ async fn prove_concurrent_divergent_insert(admin: &tokio_postgres::Client, uri:
};
let pair = [first_result, second_result];
let inserted = pair.iter().filter(|value| return matches_inserted(value)).count();
let conflicts = pair
.iter()
.filter(|value| matches!(value, LivePersistResult::BackendError(ksp_store_postgres_lib::PostgresBackendErrorKind::Conflict)))
.count();
if inserted != 1 || conflicts != 1 {
let quarantined = pair.iter().filter(|value| return matches_quarantined_conflict(value)).count();
if inserted != 1 || quarantined != 1 {
return std::result::Result::Err(LiveFailure::new("concurrent_divergent_outcome"));
}
let reference = match raw_reference(30) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let first_key = ksp_store_api::RawObservationKey::new([31; 32]);
let second_key = ksp_store_api::RawObservationKey::new([32; 32]);
let first_exists = observation_exists(admin, &first_key).await;
let second_exists = observation_exists(admin, &second_key).await;
let count = match (first_exists, second_exists) {
(std::result::Result::Ok(first_value), std::result::Result::Ok(second_value)) => usize::from(first_value) + usize::from(second_value),
_ => return std::result::Result::Err(LiveFailure::new("concurrent_divergent_observation_probe")),
let state = conflict_state_is_exact(admin, &reference, &first_key, &second_key).await;
match state {
std::result::Result::Ok(true) => {},
std::result::Result::Ok(false) => return std::result::Result::Err(LiveFailure::new("concurrent_divergent_state")),
std::result::Result::Err(error) => return std::result::Result::Err(error),
}
return std::result::Result::Ok(());
}
async fn prove_conflict_case_rollback(admin: &tokio_postgres::Client, uri: &str) -> std::result::Result<(), LiveFailure> {
let seed = persist_once(uri.to_owned(), 82, 8_200, 82, 82).await;
match seed {
std::result::Result::Ok(LivePersistResult::Outcome(value)) if value.entity() == ksp_store_api::RawEntityWriteOutcome::Inserted => {},
_ => return std::result::Result::Err(LiveFailure::new("conflict_rollback_seed")),
}
let 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),
};
if count != 1 {
return std::result::Result::Err(LiveFailure::new("concurrent_divergent_rollback"));
let transaction = match raw_transaction(82, 8_200, 83) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let observation = match raw_observation(10, 82, 83) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let result = backend.persist_raw_transaction_acquisition(transaction, observation, ksp_store_api::RawTransactionAcquisitionMode::Normal).await;
match result {
std::result::Result::Err(error) if error.kind() == ksp_store_postgres_lib::PostgresBackendErrorKind::Conflict => {},
_ => return std::result::Result::Err(LiveFailure::new("conflict_rollback_expected_failure")),
}
let close_result = close_backend(backend).await;
if let std::result::Result::Err(error) = close_result {
return std::result::Result::Err(error);
}
let reference = match raw_reference(82) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let state = conflict_rollback_state_is_exact(admin, &reference).await;
match state {
std::result::Result::Ok(true) => {},
std::result::Result::Ok(false) => return std::result::Result::Err(LiveFailure::new("conflict_rollback_state")),
std::result::Result::Err(error) => return std::result::Result::Err(error),
}
return std::result::Result::Ok(());
}
@@ -971,6 +1075,16 @@ fn matches_inserted(value: &LivePersistResult) -> bool {
);
}
fn matches_quarantined_conflict(value: &LivePersistResult) -> bool {
return matches!(
value,
LivePersistResult::Outcome(outcome)
if outcome.entity() == ksp_store_api::RawEntityWriteOutcome::AlreadyPresent
&& outcome.observation() == ksp_store_api::RawObservationWriteOutcome::Inserted
&& outcome.transaction_variant() == std::option::Option::Some(ksp_store_api::RawTransactionVariantWriteOutcome::QuarantinedConflict)
);
}
fn matches_already_present(value: &LivePersistResult) -> bool {
return matches!(
value,
@@ -1277,6 +1391,42 @@ async fn observation_exists(client: &tokio_postgres::Client, key: &ksp_store_api
};
}
async fn conflict_state_is_exact(
client: &tokio_postgres::Client,
reference: &ksp_store_api::RawTransactionReference,
first_key: &ksp_store_api::RawObservationKey,
second_key: &ksp_store_api::RawObservationKey,
) -> std::result::Result<bool, LiveFailure> {
let signature = reference.signature();
let signature_bytes: &[u8] = signature.as_bytes();
let first_key_bytes: &[u8] = first_key.as_bytes();
let second_key_bytes: &[u8] = second_key.as_bytes();
let row = match client.query_one(LIVE_CONFLICT_STATE_SQL, &[&signature_bytes, &first_key_bytes, &second_key_bytes]).await {
std::result::Result::Ok(value) => value,
std::result::Result::Err(_) => return std::result::Result::Err(LiveFailure::new("conflict_state_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("conflict_state_decode")),
};
}
async fn conflict_rollback_state_is_exact(
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_CONFLICT_ROLLBACK_STATE_SQL, &[&signature_bytes]).await {
std::result::Result::Ok(value) => value,
std::result::Result::Err(_) => return std::result::Result::Err(LiveFailure::new("conflict_rollback_state_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("conflict_rollback_state_decode")),
};
}
async fn variant_projection_is_exact(
client: &tokio_postgres::Client,
reference: &ksp_store_api::RawTransactionReference,

49
crates/ksp-store-postgres-lib/tests/v003_variant_persistence.rs
View File

@@ -1,5 +1,5 @@
// file: crates/ksp-store-postgres-lib/tests/v003_variant_persistence.rs
// version: 4
// version: 5
#![warn(missing_docs)]
#![deny(unreachable_pub)]
@@ -145,3 +145,50 @@ fn v0_3_16_pre_006_promotion_is_inside_acquisition_transaction_and_preserves_pre
assert!(promotion.contains("next_transaction_canonical_revision"));
return;
}
#[test]
fn v0_3_16_pre_008_conflict_path_never_mutates_canonical_projection_or_selector() {
let source = include_str!("../src/raw_transaction.rs");
let start = source.find("ExistingTransactionMatch::ActiveConflict(variant_comparison) => {");
let end = source.find("ExistingTransactionMatch::Purged => {");
let (start, end) = match (start, end) {
(std::option::Option::Some(start), std::option::Option::Some(end)) if start < end => (start, end),
_ => panic!("pre.008 conflict branch markers are missing"),
};
let branch = &source[start..end];
assert!(branch.contains("persist_or_reuse_native_transaction_variant"));
assert!(branch.contains("open_or_update_raw_transaction_conflict"));
assert!(branch.contains("RawTransactionVariantWriteOutcome::QuarantinedConflict"));
assert!(!branch.contains("promote_more_complete_transaction_variant"));
assert!(!branch.contains("UPDATE_CANONICAL_TRANSACTION_PROJECTION_SQL"));
assert!(!branch.contains("UPDATE_TRANSACTION_VARIANT_SELECTOR_SQL"));
return;
}
#[test]
fn v0_3_16_pre_008_variant_transition_helpers_cannot_commit_independently() {
let source = include_str!("../src/raw_transaction.rs");
let promotion_start = source.find("async fn promote_more_complete_transaction_variant(");
let promotion_end = source.find("async fn preserve_previous_canonical_variant(");
let conflict_start = source.find("async fn open_or_update_raw_transaction_conflict(");
let conflict_end = source.find("async fn promote_more_complete_transaction_variant(");
let (promotion_start, promotion_end, conflict_start, conflict_end) = match (promotion_start, promotion_end, conflict_start, conflict_end) {
(Some(promotion_start), Some(promotion_end), Some(conflict_start), Some(conflict_end))
if promotion_start < promotion_end && conflict_start < conflict_end =>
{
(promotion_start, promotion_end, conflict_start, conflict_end)
},
_ => panic!("pre.008 transition helper markers are missing"),
};
let promotion = &source[promotion_start..promotion_end];
let conflict = &source[conflict_start..conflict_end];
for helper in [promotion, conflict] {
assert!(helper.contains("&deadpool_postgres::Transaction<'_>"));
assert!(!helper.contains(".commit().await"));
}
assert!(source.contains("const LOCK_TRANSACTION_SQL: &str = \"SELECT"));
assert!(source.contains("WHERE signature = $1 FOR UPDATE\";"));
assert!(source.contains("canonical_variant_id = $2::TEXT::NUMERIC AND canonical_revision = $3::TEXT::NUMERIC"));
assert!(source.contains("AND revision = $8::TEXT::NUMERIC"));
return;
}

46
crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime.rs
View File

@@ -1,5 +1,5 @@
// file: crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime.rs
// version: 14
// version: 15
struct ActiveTaskGuard {
active: std::sync::Arc<std::sync::atomic::AtomicUsize>,
@@ -526,6 +526,50 @@ async fn v0_3_16_pre_007_quarantined_conflict_keeps_worker_running_and_degraded(
return;
}
#[tokio::test(flavor = "current_thread")]
async fn v0_3_16_pre_008_quarantined_conflict_does_not_block_later_identity() {
let settings = match settings_with_persistence_concurrency(1) {
std::option::Option::Some(value) => value,
std::option::Option::None => return,
};
let network = settings.network().clone();
let port = std::sync::Arc::new(RuntimePersistencePort::new(network.clone(), RuntimePortResponse::QuarantinedConflict, true));
let runtime_port: super::PersistencePort = port.clone();
let handle = match super::start_foundation_with_port_and_source_spawner(
settings,
tokio::runtime::Handle::current(),
std::option::Option::Some(runtime_port),
move |children, _stop_receiver, admission_sender| {
let _abort_handle = children.spawn(async move {
for signature_byte in [20_u8, 21_u8] {
let ingress = match runtime_ingress(&network, signature_byte) {
std::option::Option::Some(value) => value,
std::option::Option::None => return std::result::Result::Err(crate::runtime_error("test.source_failed")),
};
if admission_sender.send(ingress).await.is_err() {
return std::result::Result::Err(crate::runtime_error("test.source_failed"));
}
}
return std::result::Result::Ok(());
});
},
) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(_) => return,
};
let source = handle.snapshot_source();
assert!(wait_for_completed(&port, 2).await);
let snapshot = source.current();
assert_eq!(snapshot.worker_snapshot().state(), ksp_worker_api::WorkerState::Running);
assert_eq!(snapshot.worker_snapshot().health(), ksp_worker_api::WorkerHealth::Degraded);
assert_eq!(snapshot.persisted_total(), 2);
assert_eq!(snapshot.content_conflict_total(), 2);
assert_eq!(snapshot.store_failure_total(), 0);
assert!(handle.request_stop());
assert!(matches!(handle.wait_terminal().await, std::result::Result::Ok(ksp_worker_api::WorkerState::Stopped)));
return;
}
#[tokio::test(flavor = "current_thread")]
async fn pre_007_legacy_store_conflict_error_becomes_terminal_after_private_drain() {
let settings = match settings_with_persistence_concurrency(1) {