diff --git a/Cargo.toml b/Cargo.toml index a81d91d..3c98ddd 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,12 +1,12 @@ # file: Cargo.toml -# version: 644 +# version: 645 [workspace] resolver = "3" members = ["crates/ksp-app-backfill-desk", "crates/ksp-app-config-desk", "crates/ksp-app-raw-transaction-ingest-desk", "crates/ksp-app-solprices-desk", "crates/ksp-app-store-desk", "crates/ksp-app-wallet-desk", "crates/ksp-config-lib", "crates/ksp-core-lib", "crates/ksp-interface-lib", "crates/ksp-job-api", "crates/ksp-job-backfill-lib", "crates/ksp-logging-lib", "crates/ksp-offchain-transport-lib", "crates/ksp-onchain-transport-lib", "crates/ksp-program-api", "crates/ksp-raw-transaction-lib", "crates/ksp-store-api", "crates/ksp-store-lib", "crates/ksp-store-postgres-lib", "crates/ksp-wallet-lib", "crates/ksp-worker-api", "crates/ksp-worker-raw-transaction-ingest-lib"] [workspace.package] -version = "0.3.16-pre.5.fix.1" +version = "0.3.16-pre.6" edition = "2024" license = "MIT" repository = "https://git.sasedev.com/Sasedev/khadhroony-solana-project" diff --git a/crates/ksp-store-postgres-lib/src/raw_transaction.rs b/crates/ksp-store-postgres-lib/src/raw_transaction.rs index eb8f7c8..30e5bfc 100644 --- a/crates/ksp-store-postgres-lib/src/raw_transaction.rs +++ b/crates/ksp-store-postgres-lib/src/raw_transaction.rs @@ -1,5 +1,5 @@ // file: crates/ksp-store-postgres-lib/src/raw_transaction.rs -// version: 12 +// version: 13 pub(crate) mod cursor; @@ -48,6 +48,7 @@ const REHYDRATE_TRANSACTION_SQL: &str = "UPDATE ksp_raw_transactions SET block_time_unix_millis = $2, payload = $3, retention_state = 'full' WHERE signature = $1"; const SELECT_CANONICAL_VARIANT_ID_SQL: &str = "SELECT canonical_variant_id::TEXT AS canonical_variant_id_text FROM ksp_raw_transaction_canonical_selectors WHERE transaction_signature = $1"; +const SELECT_CANONICAL_VARIANT_SELECTOR_SQL: &str = "SELECT canonical_variant_id::TEXT AS canonical_variant_id_text, canonical_revision::TEXT AS canonical_revision_text FROM ksp_raw_transaction_canonical_selectors WHERE transaction_signature = $1"; const SELECT_EXACT_NATIVE_VARIANT_SQL: &str = "SELECT variant_id::TEXT AS variant_id_text FROM ksp_raw_transaction_variants WHERE transaction_signature = $1 AND origin_kind = 'native' AND slot = $2::TEXT::NUMERIC AND block_time_unix_millis IS NOT DISTINCT FROM $3 AND format_id = $4 AND format_version = $5 AND content_hash = $6 AND payload = $7 AND retention_state = 'full' ORDER BY variant_id ASC LIMIT 1"; const SELECT_MAX_VARIANT_ID_SQL: &str = "SELECT COALESCE(MAX(variant_id), 0)::TEXT AS variant_id_text FROM ksp_raw_transaction_variants WHERE transaction_signature = $1"; @@ -55,8 +56,11 @@ const SELECT_TRANSACTION_VARIANT_LINK_SQL: &str = "SELECT transaction_signature, variant_id::TEXT AS variant_id_text FROM ksp_raw_transaction_observation_variants WHERE observation_key = $1"; const UPDATE_ARCHIVED_TRANSACTION_SQL: &str = "UPDATE ksp_raw_transactions SET payload = NULL, retention_state = 'archived' WHERE signature = $1 AND retention_state = 'full'"; +const UPDATE_CANONICAL_TRANSACTION_PROJECTION_SQL: &str = "UPDATE ksp_raw_transactions SET slot = $2::TEXT::NUMERIC, block_time_unix_millis = $3, format_id = $4, format_version = $5, content_hash = $6, payload = $7, retention_state = 'full' WHERE signature = $1"; const UPDATE_PURGED_TRANSACTION_SQL: &str = "UPDATE ksp_raw_transactions SET block_time_unix_millis = NULL, payload = NULL, retention_state = 'purged' WHERE signature = $1 AND retention_state = 'archived'"; const UPDATE_REHYDRATED_TRANSACTION_VARIANT_SQL: &str = "UPDATE ksp_raw_transaction_variants SET block_time_unix_millis = $3, payload = $4, retention_state = 'full' WHERE transaction_signature = $1 AND variant_id = $2::TEXT::NUMERIC"; +const UPDATE_TRANSACTION_VARIANT_SELECTOR_SQL: &str = "UPDATE ksp_raw_transaction_canonical_selectors SET canonical_variant_id = $4::TEXT::NUMERIC, canonical_revision = $5::TEXT::NUMERIC, updated_at_unix_millis = $6 WHERE transaction_signature = $1 AND canonical_variant_id = $2::TEXT::NUMERIC AND canonical_revision = $3::TEXT::NUMERIC"; +const UPDATE_TRANSACTION_VARIANT_TO_FULL_SQL: &str = "UPDATE ksp_raw_transaction_variants SET payload = $8, retention_state = 'full' WHERE transaction_signature = $1 AND variant_id = $2::TEXT::NUMERIC AND slot = $3::TEXT::NUMERIC AND block_time_unix_millis IS NOT DISTINCT FROM $4 AND format_id = $5 AND format_version = $6 AND content_hash = $7 AND ((retention_state = 'archived' AND payload IS NULL) OR (retention_state = 'full' AND payload = $8))"; struct RawLogMessagesContentConflictDiagnostic { available: bool, @@ -697,10 +701,10 @@ pub(crate) async fn persist_raw_transaction_acquisition( std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; - let (entity_outcome, observed_variant_id) = if inserted { - (ksp_store_api::RawEntityWriteOutcome::Inserted, canonical_variant_id) + let (entity_outcome, observed_variant_id, canonical_promotion) = if inserted { + (ksp_store_api::RawEntityWriteOutcome::Inserted, canonical_variant_id, std::option::Option::None) } else { - let comparison_result = compare_existing_transaction(network, locked, &raw_transaction); + let comparison_result = compare_existing_transaction(network, locked.clone(), &raw_transaction); let comparison = match comparison_result { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => { @@ -711,14 +715,28 @@ pub(crate) async fn persist_raw_transaction_acquisition( }, }; match comparison { - ExistingTransactionMatch::Active => (ksp_store_api::RawEntityWriteOutcome::AlreadyPresent, canonical_variant_id), + ExistingTransactionMatch::Active => (ksp_store_api::RawEntityWriteOutcome::AlreadyPresent, canonical_variant_id, std::option::Option::None), ExistingTransactionMatch::ActiveCompatibleLessComplete => { let variant_result = persist_or_reuse_native_transaction_variant(&sql_transaction, &raw_transaction, linked_at).await; let variant_id = match variant_result { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; - (ksp_store_api::RawEntityWriteOutcome::AlreadyPresent, variant_id) + (ksp_store_api::RawEntityWriteOutcome::AlreadyPresent, variant_id, std::option::Option::None) + }, + ExistingTransactionMatch::ActiveCompatibleMoreComplete => { + let variant_result = persist_or_reuse_native_transaction_variant(&sql_transaction, &raw_transaction, linked_at).await; + let variant_id = match variant_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let promotion_result = + promote_more_complete_transaction_variant(&sql_transaction, &locked, &raw_transaction, canonical_variant_id, variant_id, linked_at).await; + let promotion = match promotion_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + (ksp_store_api::RawEntityWriteOutcome::AlreadyPresent, variant_id, std::option::Option::Some(promotion)) }, ExistingTransactionMatch::Purged => { if mode == ksp_store_api::RawTransactionAcquisitionMode::ForceRehydrate { @@ -730,7 +748,7 @@ pub(crate) async fn persist_raw_transaction_acquisition( if let std::result::Result::Err(error) = variant_result { return std::result::Result::Err(error); } - (ksp_store_api::RawEntityWriteOutcome::Rehydrated, canonical_variant_id) + (ksp_store_api::RawEntityWriteOutcome::Rehydrated, canonical_variant_id, std::option::Option::None) } else { let commit_result = sql_transaction.commit().await; if commit_result.is_err() { @@ -757,6 +775,9 @@ pub(crate) async fn persist_raw_transaction_acquisition( if commit_result.is_err() { return std::result::Result::Err(write_failed("raw_acquisition_commit")); } + if let std::option::Option::Some(transition) = canonical_promotion { + log_raw_transaction_canonical_promotion_committed(network, raw_transaction.slot(), transition); + } return std::result::Result::Ok(ksp_store_api::RawAcquisitionWriteOutcome::new(entity_outcome, observation_outcome)); } @@ -893,10 +914,19 @@ pub(crate) async fn transition_raw_transaction_retention( return commit_retention_outcome(sql_transaction, ksp_store_api::RawRetentionWriteOutcome::Applied).await; } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +struct RawCanonicalPromotionTransition { + from_revision: u64, + from_variant_id: u64, + to_revision: u64, + to_variant_id: u64, +} + #[derive(Clone, Copy, Debug, Eq, PartialEq)] enum ExistingTransactionMatch { Active, ActiveCompatibleLessComplete, + ActiveCompatibleMoreComplete, Purged, } @@ -1229,6 +1259,182 @@ async fn persist_or_reuse_native_transaction_variant( }; } +async fn promote_more_complete_transaction_variant( + sql_transaction: &deadpool_postgres::Transaction<'_>, + locked: &RawTransactionDbRow, + incoming: &ksp_store_api::RawTransaction, + expected_canonical_variant_id: u64, + promoted_variant_id: u64, + updated_at_unix_millis: i64, +) -> std::result::Result { + if promoted_variant_id == expected_canonical_variant_id { + return std::result::Result::Err(data_invalid("raw_variant_promotion_identity")); + } + let signature = incoming.reference().signature(); + let signature_bytes: &[u8] = signature.as_bytes(); + if locked.signature.as_slice() != signature_bytes { + return std::result::Result::Err(data_invalid("raw_variant_promotion_reference")); + } + let selector_result = sql_transaction.query_opt(SELECT_CANONICAL_VARIANT_SELECTOR_SQL, &[&signature_bytes]).await; + let selector_row = match selector_result { + std::result::Result::Ok(std::option::Option::Some(value)) => value, + std::result::Result::Ok(std::option::Option::None) => return std::result::Result::Err(data_invalid("raw_variant_promotion_selector_missing")), + std::result::Result::Err(_) => return std::result::Result::Err(write_failed("raw_variant_promotion_selector_query")), + }; + let canonical_variant_id_text = match selector_row.try_get::<_, std::string::String>("canonical_variant_id_text") { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_variant_promotion_selector_id")), + }; + let current_canonical_variant_id = match decode_u64_decimal(canonical_variant_id_text.as_str(), "raw_variant_promotion_selector_id") { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + if current_canonical_variant_id != expected_canonical_variant_id { + return std::result::Result::Err(data_invalid("raw_variant_promotion_selector_changed")); + } + let canonical_revision_text = match selector_row.try_get::<_, std::string::String>("canonical_revision_text") { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_variant_promotion_revision")), + }; + let current_revision = match decode_u64_decimal(canonical_revision_text.as_str(), "raw_variant_promotion_revision") { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let next_revision = match next_transaction_canonical_revision(current_revision) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let preserve_result = preserve_previous_canonical_variant(sql_transaction, locked, expected_canonical_variant_id).await; + if let std::result::Result::Err(error) = preserve_result { + return std::result::Result::Err(error); + } + let archive_cleanup_result = sql_transaction.execute(DELETE_ARCHIVE_PAYLOAD_SQL, &[&signature_bytes]).await; + match archive_cleanup_result { + std::result::Result::Ok(count) if count <= 1 => {}, + std::result::Result::Ok(_) => return std::result::Result::Err(data_invalid("raw_variant_promotion_archive_cleanup_count")), + std::result::Result::Err(_) => return std::result::Result::Err(write_failed("raw_variant_promotion_archive_cleanup")), + } + let slot_text = incoming.slot().to_string(); + let block_time = match incoming.block_time() { + std::option::Option::Some(value) => match i64::try_from(value.unix_millis()) { + std::result::Result::Ok(decoded) => std::option::Option::Some(decoded), + std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_variant_promotion_block_time")), + }, + std::option::Option::None => std::option::Option::None, + }; + let format_version = i64::from(incoming.payload().format_version()); + let content_hash = incoming.payload().content_hash(); + let content_hash_bytes: &[u8] = content_hash.as_bytes(); + let payload_bytes = incoming.payload().bytes(); + let projection_result = sql_transaction + .execute( + UPDATE_CANONICAL_TRANSACTION_PROJECTION_SQL, + &[ + &signature_bytes, + &slot_text.as_str(), + &block_time, + &incoming.payload().format_id().as_str(), + &format_version, + &content_hash_bytes, + &payload_bytes, + ], + ) + .await; + match projection_result { + std::result::Result::Ok(1) => {}, + std::result::Result::Ok(_) => return std::result::Result::Err(data_invalid("raw_variant_promotion_projection_count")), + std::result::Result::Err(_) => return std::result::Result::Err(write_failed("raw_variant_promotion_projection")), + } + let expected_variant_id_text = expected_canonical_variant_id.to_string(); + let expected_revision_text = current_revision.to_string(); + let promoted_variant_id_text = promoted_variant_id.to_string(); + let next_revision_text = next_revision.to_string(); + let selector_update_result = sql_transaction + .execute( + UPDATE_TRANSACTION_VARIANT_SELECTOR_SQL, + &[ + &signature_bytes, + &expected_variant_id_text.as_str(), + &expected_revision_text.as_str(), + &promoted_variant_id_text.as_str(), + &next_revision_text.as_str(), + &updated_at_unix_millis, + ], + ) + .await; + return match selector_update_result { + std::result::Result::Ok(1) => std::result::Result::Ok(RawCanonicalPromotionTransition { + from_revision: current_revision, + from_variant_id: expected_canonical_variant_id, + to_revision: next_revision, + to_variant_id: promoted_variant_id, + }), + std::result::Result::Ok(0) => std::result::Result::Err(data_invalid("raw_variant_promotion_stale_selector")), + std::result::Result::Ok(_) => std::result::Result::Err(data_invalid("raw_variant_promotion_selector_count")), + std::result::Result::Err(_) => std::result::Result::Err(write_failed("raw_variant_promotion_selector_update")), + }; +} + +async fn preserve_previous_canonical_variant( + sql_transaction: &deadpool_postgres::Transaction<'_>, + locked: &RawTransactionDbRow, + canonical_variant_id: u64, +) -> std::result::Result<(), crate::PostgresBackendError> { + let state = match decode_retention_state(locked.retention_state.as_str()) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let preserved_payload = if state == ksp_store_api::RawRetentionState::Full { + if locked.archive_payload.is_some() { + return std::result::Result::Err(data_invalid("raw_variant_promotion_full_shape")); + } + match locked.payload.as_ref() { + std::option::Option::Some(value) => value.as_slice(), + std::option::Option::None => return std::result::Result::Err(data_invalid("raw_variant_promotion_full_payload")), + } + } else if state == ksp_store_api::RawRetentionState::Archived { + if locked.payload.is_some() { + return std::result::Result::Err(data_invalid("raw_variant_promotion_archived_shape")); + } + match locked.archive_payload.as_ref() { + std::option::Option::Some(value) => value.as_slice(), + std::option::Option::None => return std::result::Result::Err(data_invalid("raw_variant_promotion_archived_payload")), + } + } else { + return std::result::Result::Err(data_invalid("raw_variant_promotion_retention_state")); + }; + let signature_bytes: &[u8] = locked.signature.as_slice(); + let canonical_variant_id_text = canonical_variant_id.to_string(); + let content_hash_bytes: &[u8] = locked.content_hash.as_slice(); + let update_result = sql_transaction + .execute( + UPDATE_TRANSACTION_VARIANT_TO_FULL_SQL, + &[ + &signature_bytes, + &canonical_variant_id_text.as_str(), + &locked.slot_text.as_str(), + &locked.block_time_unix_millis, + &locked.format_id.as_str(), + &locked.format_version, + &content_hash_bytes, + &preserved_payload, + ], + ) + .await; + return match update_result { + std::result::Result::Ok(1) => std::result::Result::Ok(()), + std::result::Result::Ok(_) => std::result::Result::Err(data_invalid("raw_variant_promotion_preserve_count")), + std::result::Result::Err(_) => std::result::Result::Err(write_failed("raw_variant_promotion_preserve")), + }; +} + +fn next_transaction_canonical_revision(current: u64) -> std::result::Result { + return match current.checked_add(1) { + std::option::Option::Some(value) if value != 0 => std::result::Result::Ok(value), + _ => std::result::Result::Err(data_invalid("raw_variant_canonical_revision_exhausted")), + }; +} + async fn persist_transaction_variant_link( sql_transaction: &deadpool_postgres::Transaction<'_>, observation: &ksp_store_api::RawTransactionObservation, @@ -1449,9 +1655,8 @@ fn compare_existing_transaction( log_raw_transaction_compatible_truncated_log_messages(network, &stored, incoming); std::result::Result::Ok(ExistingTransactionMatch::ActiveCompatibleLessComplete) }, - ksp_store_api::RawTransactionVariantRelation::CompatibleMoreComplete - | ksp_store_api::RawTransactionVariantRelation::Conflict - | ksp_store_api::RawTransactionVariantRelation::Incomparable => { + ksp_store_api::RawTransactionVariantRelation::CompatibleMoreComplete => std::result::Result::Ok(ExistingTransactionMatch::ActiveCompatibleMoreComplete), + ksp_store_api::RawTransactionVariantRelation::Conflict | ksp_store_api::RawTransactionVariantRelation::Incomparable => { log_raw_transaction_content_conflict(network, &stored, incoming); std::result::Result::Err(conflict("raw_acquisition_content_conflict")) }, @@ -1705,6 +1910,21 @@ fn log_raw_transaction_compatible_truncated_log_messages( return; } +fn log_raw_transaction_canonical_promotion_committed(network: &ksp_store_api::RawNetworkId, slot: u64, transition: RawCanonicalPromotionTransition) { + ksp_logging_lib::info!( + target: crate::TRACING_TARGET, + domain = "store.raw_transaction.canonical_promotion", + network = network.as_str(), + slot = slot, + from_variant_id = transition.from_variant_id, + to_variant_id = transition.to_variant_id, + from_revision = transition.from_revision, + to_revision = transition.to_revision, + "PostgreSQL Store committed a proved RAW transaction canonical promotion" + ); + return; +} + fn log_raw_transaction_content_conflict( network: &ksp_store_api::RawNetworkId, stored: &ksp_store_api::RawTransaction, diff --git a/crates/ksp-store-postgres-lib/tests/v003_variant_persistence.rs b/crates/ksp-store-postgres-lib/tests/v003_variant_persistence.rs index 7d0b4c0..f49af43 100644 --- a/crates/ksp-store-postgres-lib/tests/v003_variant_persistence.rs +++ b/crates/ksp-store-postgres-lib/tests/v003_variant_persistence.rs @@ -1,5 +1,5 @@ // file: crates/ksp-store-postgres-lib/tests/v003_variant_persistence.rs -// version: 2 +// version: 3 #![warn(missing_docs)] #![deny(unreachable_pub)] @@ -59,12 +59,52 @@ fn v0_3_16_pre_004_legacy_observation_without_mapping_is_not_fabricated() { } #[test] -fn v0_3_16_pre_005_shared_comparator_is_active_but_durable_conflict_and_promotion_stay_deferred() { +fn v0_3_16_pre_006_shared_comparator_promotes_only_more_complete_and_keeps_conflict_deferred() { let source = include_str!("../src/raw_transaction.rs"); assert!(source.contains("ksp_store_api::compare_raw_transaction_variants")); - assert!(source.contains("ksp_store_api::RawTransactionVariantRelation::CompatibleLessComplete")); - assert!(source.contains("ksp_store_api::RawTransactionVariantRelation::CompatibleMoreComplete")); + assert!(source.contains("ExistingTransactionMatch::ActiveCompatibleLessComplete")); + assert!(source.contains("ExistingTransactionMatch::ActiveCompatibleMoreComplete")); + assert!(source.contains("promote_more_complete_transaction_variant")); + assert!(source.contains("UPDATE ksp_raw_transaction_canonical_selectors SET canonical_variant_id")); + assert!(source.contains("canonical_revision = $5::TEXT::NUMERIC")); + assert!(source.contains("UPDATE ksp_raw_transactions SET slot = $2::TEXT::NUMERIC")); + assert!(source.contains("PostgreSQL Store committed a proved RAW transaction canonical promotion")); + assert!(!source.contains("log_raw_transaction_compatible_more_complete_log_messages")); assert!(!source.contains("INSERT INTO ksp_raw_transaction_conflicts")); - assert!(!source.contains("UPDATE ksp_raw_transaction_canonical_selectors SET canonical_variant_id")); + return; +} + +#[test] +fn v0_3_16_pre_006_promotion_is_inside_acquisition_transaction_and_preserves_previous_canonical() { + let source = include_str!("../src/raw_transaction.rs"); + let start = source.find("pub(crate) async fn persist_raw_transaction_acquisition("); + let end = source.find("pub(crate) async fn record_raw_transaction_observation("); + let (start, end) = match (start, end) { + (std::option::Option::Some(start), std::option::Option::Some(end)) if start < end => (start, end), + _ => panic!("pre.006 acquisition function markers are missing"), + }; + let body = &source[start..end]; + let persist_variant = body.find("persist_or_reuse_native_transaction_variant"); + let promote = body.find("promote_more_complete_transaction_variant"); + let observation = body.find("persist_observation_row"); + let commit = body.rfind("sql_transaction.commit().await"); + let transition_log = body.find("log_raw_transaction_canonical_promotion_committed"); + assert!(matches!( + (persist_variant, promote, observation, commit, transition_log), + (Some(persist_variant), Some(promote), Some(observation), Some(commit), Some(transition_log)) + if persist_variant < promote && promote < observation && observation < commit && commit < transition_log + )); + let promotion_start = source.find("async fn promote_more_complete_transaction_variant("); + let promotion_end = source.find("async fn persist_transaction_variant_link("); + let (promotion_start, promotion_end) = match (promotion_start, promotion_end) { + (std::option::Option::Some(start), std::option::Option::Some(end)) if start < end => (start, end), + _ => panic!("pre.006 promotion helper markers are missing"), + }; + let promotion = &source[promotion_start..promotion_end]; + assert!(promotion.contains("preserve_previous_canonical_variant")); + assert!(promotion.contains("DELETE_ARCHIVE_PAYLOAD_SQL")); + assert!(promotion.contains("UPDATE_CANONICAL_TRANSACTION_PROJECTION_SQL")); + assert!(promotion.contains("UPDATE_TRANSACTION_VARIANT_SELECTOR_SQL")); + assert!(promotion.contains("next_transaction_canonical_revision")); return; } diff --git a/crates/ksp-store-postgres-lib/unit_tests/raw_transaction.rs b/crates/ksp-store-postgres-lib/unit_tests/raw_transaction.rs index 4ca07b0..bc0145c 100644 --- a/crates/ksp-store-postgres-lib/unit_tests/raw_transaction.rs +++ b/crates/ksp-store-postgres-lib/unit_tests/raw_transaction.rs @@ -1,5 +1,5 @@ // file: crates/ksp-store-postgres-lib/unit_tests/raw_transaction.rs -// version: 12 +// version: 13 fn network() -> ksp_store_api::RawNetworkId { return match ksp_store_api::RawNetworkId::new("devnet") { @@ -870,3 +870,33 @@ fn v0_3_16_pre_004_variant_link_contract_keeps_legacy_observations_unfabricated( assert!(!super::INSERT_TRANSACTION_VARIANT_LINK_SQL.contains("ON CONFLICT")); return; } + +#[test] +fn v0_3_16_pre_006_canonical_revision_is_monotonic_and_bounded() { + assert_eq!(super::next_transaction_canonical_revision(1), std::result::Result::Ok(2)); + let exhausted = super::next_transaction_canonical_revision(u64::MAX).err(); + assert_eq!(exhausted.as_ref().map(|value| return value.kind()), std::option::Option::Some(crate::PostgresBackendErrorKind::DataInvalid)); + assert_eq!(exhausted.as_ref().map(|value| return value.phase()), std::option::Option::Some("raw_variant_canonical_revision_exhausted")); + return; +} + +#[test] +fn v0_3_16_pre_006_promotion_sql_is_compare_guarded_and_projection_complete() { + assert!(super::SELECT_CANONICAL_VARIANT_SELECTOR_SQL.contains("canonical_revision::TEXT")); + assert!(super::UPDATE_TRANSACTION_VARIANT_SELECTOR_SQL.contains("canonical_variant_id = $2::TEXT::NUMERIC")); + assert!(super::UPDATE_TRANSACTION_VARIANT_SELECTOR_SQL.contains("canonical_revision = $3::TEXT::NUMERIC")); + assert!(super::UPDATE_TRANSACTION_VARIANT_SELECTOR_SQL.contains("canonical_variant_id = $4::TEXT::NUMERIC")); + assert!(super::UPDATE_CANONICAL_TRANSACTION_PROJECTION_SQL.contains("slot = $2::TEXT::NUMERIC")); + assert!(super::UPDATE_CANONICAL_TRANSACTION_PROJECTION_SQL.contains("content_hash = $6")); + assert!(super::UPDATE_CANONICAL_TRANSACTION_PROJECTION_SQL.contains("payload = $7")); + assert!(super::UPDATE_CANONICAL_TRANSACTION_PROJECTION_SQL.contains("retention_state = 'full'")); + assert!(super::UPDATE_TRANSACTION_VARIANT_TO_FULL_SQL.contains("slot = $3::TEXT::NUMERIC")); + assert!(super::UPDATE_TRANSACTION_VARIANT_TO_FULL_SQL.contains("block_time_unix_millis IS NOT DISTINCT FROM $4")); + assert!(super::UPDATE_TRANSACTION_VARIANT_TO_FULL_SQL.contains("format_id = $5")); + assert!(super::UPDATE_TRANSACTION_VARIANT_TO_FULL_SQL.contains("format_version = $6")); + assert!(super::UPDATE_TRANSACTION_VARIANT_TO_FULL_SQL.contains("content_hash = $7")); + assert!(super::UPDATE_TRANSACTION_VARIANT_TO_FULL_SQL.contains("retention_state = 'archived'")); + assert!(super::UPDATE_TRANSACTION_VARIANT_TO_FULL_SQL.contains("retention_state = 'full' AND payload = $8")); + assert!(!super::UPDATE_TRANSACTION_VARIANT_TO_FULL_SQL.contains("DELETE")); + return; +} diff --git a/deltas/0.3.16/pre.006.md b/deltas/0.3.16/pre.006.md new file mode 100644 index 0000000..1f39bd7 --- /dev/null +++ b/deltas/0.3.16/pre.006.md @@ -0,0 +1,217 @@ + + + +# Delta `0.3.16-pre.006` — promotion atomique du canonique RAW + +## Base requise + +```text +0.3.16-pre.005-fix.001 appliqué +workspace.package.version = 0.3.16-pre.5.fix.1 +``` + +Le gate opérateur fourni pour la base est propre : audits Rust/export/KSP/Markdown, `cargo check --workspace`, Clippy `-D warnings`, toutes les suites `ksp-store-api` et `ksp-store-postgres-lib` ciblées passent. Les preuves PostgreSQL live restent opt-in. + +L'audit de l'archive source fournie a relevé un unique artefact ignoré, `scripts/__pycache__/audit_rust_general_rules.cpython-313.pyc`, alors que `.gitignore` exclut `__pycache__/`. Cet artefact n'appartient pas au source ni au delta et est explicitement exclu de l'archive `pre.006`. Aucun autre candidat correspondant aux familles ignorées de `.gitignore` n'a été trouvé. + +## Version + +Cette tranche modifie le runtime PostgreSQL : + +```text +workspace.package.version = 0.3.16-pre.6 +``` + +## Objectif + +Rendre effectif le sens déjà reconnu par le comparateur partagé : + +```text +canonique logMessages tronqué ++ +incoming logMessages complet strictement prouvé +=> CompatibleMoreComplete +=> variante entrante durable +=> promotion canonique atomique +=> ancien canonique conservé +``` + +Aucune règle de dominance nouvelle n'est ajoutée. `logMessages` reste la seule preuve automatique de complétude et les autres différences continuent à suivre la politique fail-closed de `pre.005`. + +## Promotion atomique + +Le chemin PostgreSQL `persist_raw_transaction_acquisition` traite désormais `CompatibleMoreComplete` dans la même transaction que l'acquisition : + +```text +lock identité V001 +ensure bootstrap V003 +compare canonique / incoming +persist ou reuse variante incoming exacte +charger selector + revision +conserver l'ancien canonique +mettre à jour la projection V001 +mettre à jour selector + canonical_revision +persist observation +lier observation -> variante incoming +COMMIT +``` + +Le selector est modifié avec une garde exacte sur : + +```text +transaction_signature +canonical_variant_id attendu +canonical_revision attendue +``` + +La nouvelle révision est calculée par incrément borné `u64`; l'épuisement de la révision est fail-closed avant mutation canonique. + +Toute cardinalité inattendue ou changement du selector observé pendant la transaction provoque une erreur et laisse PostgreSQL rollbacker l'ensemble de la promotion. + +## Projection V001 cohérente + +Lors d'une promotion, la projection de compatibilité `ksp_raw_transactions` est remplacée par la représentation entrante complète : + +```text +slot +block_time_unix_millis +format_id +format_version +content_hash +payload +retention_state = full +``` + +Le selector V003 n'est déplacé qu'au sein de la même transaction. Il ne peut donc pas être durablement avancé si la projection V001 échoue, ni inversement. + +## Conservation de l'ancien canonique + +L'ancienne variante canonique n'est jamais supprimée ni écrasée par la nouvelle variante. + +Cas `Full` : l'ancienne variante V003 est confirmée byte-exactement sur slot, block time, format, version, hash et payload avant tout remplacement de projection. + +Cas projection V001 `Archived` : les bytes encore disponibles dans `ksp_raw_transaction_archive_payloads` sont d'abord copiés ou confirmés byte-exacts dans l'ancienne variante V003, avec la même vérification de métadonnées, puis l'archive V001 devenue obsolète est supprimée avant de remettre la projection V001 à `Full` sur le nouveau canonique. + +Cette copie de sauvegarde est une nécessité de promotion/rollback local, pas l'ouverture de la politique générale de rétention des variantes prévue en `0.3.17`. + +Aucune reconstruction réseau n'est utilisée. + +## Révision et historique minimal + +`canonical_revision` devient effectivement monotone lors d'une promotion automatique et constitue la garde durable du selector courant. + +Un journal opérationnel minimal est émis uniquement après le `COMMIT` réussi de l'acquisition. Il porte seulement le réseau, le slot, l'ancien/nouveau `variant_id` et l'ancienne/nouvelle `canonical_revision`; aucun payload ni signature n'est journalisé. Une promotion rollbackée n'est donc jamais annoncée comme commitée. + +Cette tranche n'ajoute pas de table append-only supplémentaire : le journal durable complet des actions/résolutions et les participants de conflit restent explicitement différés par le plan 038 à `0.3.17`. `pre.006` conserve donc le scope V003 déjà figé et n'altère aucun checksum de migration. + +## Conflits toujours différés + +Cette tranche ne modifie pas le comportement durable de : + +```text +Conflict +Incomparable +``` + +Ils restent projetés vers `raw_acquisition_content_conflict` jusqu'à `0.3.16-pre.007`. + +En particulier, `pre.006` n'écrit toujours pas dans : + +```text +ksp_raw_transaction_conflicts +``` + +et ne rend encore aucun conflit non terminal côté Worker. + +## Tests et canaris + +Les tests unitaires PostgreSQL ajoutés verrouillent : + +- l'incrément borné de `canonical_revision` ; +- la garde selector par ancien `canonical_variant_id` et ancienne revision ; +- la projection V001 complète vers le nouveau canonique `Full` ; +- la conservation locale d'une ancienne variante `Archived` ou déjà `Full`. + +Le canari `v003_variant_persistence.rs` est avancé de `pre.005` vers `pre.006` et exige désormais : + +- le chemin `ActiveCompatibleMoreComplete` ; +- la persistance de la variante avant promotion ; +- la promotion avant l'observation et avant le commit final ; +- l'update du selector et de `canonical_revision` ; +- l'update de la projection V001 ; +- la conservation de l'ancien canonique ; +- le journal opérationnel de promotion placé après le commit réussi ; +- l'absence persistante d'écriture dans la table de conflits avant `pre.007`. + +## Fichiers modifiés + +```text +Cargo.toml +crates/ksp-store-postgres-lib/src/raw_transaction.rs +crates/ksp-store-postgres-lib/tests/v003_variant_persistence.rs +crates/ksp-store-postgres-lib/unit_tests/raw_transaction.rs +``` + +## Fichier ajouté + +```text +deltas/0.3.16/pre.006.md +``` + +## Fichiers supprimés + +```text +aucun +``` + +## Validation exécutée lors de la génération + +Sur le checkout complet final : + +```bash +python3 scripts/audit_rust_workspace_rules.py +python3 scripts/audit_markdown_tables.py README.md RULES.md ROADMAP.md CHANGELOG.md docs prompts crates deltas +``` + +Résultat : audits Rust/export/KSP et Markdown propres. + +Le ZIP minimal final est ensuite appliqué sur une copie fraîche de `0.3.16-pre.005-fix.001`. Les mêmes audits y restent propres, les cinq fichiers livrés sont byte-identiques au checkout de génération et les 98 ressources de migration PostgreSQL restent byte-identiques à la base. + +Des canaris statiques supplémentaires ont été rejoués sur le source final et sur le replay pour confirmer l'ordre : variante -> promotion -> observation -> mapping -> commit -> journal opérationnel, l'existence des updates selector/projection, la conservation byte-exacte de l'ancien canonique et l'absence d'écriture de conflit. + +Le toolchain Rust n'est pas installé dans l'environnement de génération ; `cargo fmt`, `cargo check`, Clippy et les tests Cargo ne sont donc pas déclarés exécutés ici. Le gate Cargo propre de la base provient du log opérateur fourni avec `pre.005-fix.001`. + +## Validation opérateur demandée + +```bash +cargo fmt --all +cargo fmt --all -- --check + +python3 scripts/audit_rust_workspace_rules.py +python3 scripts/audit_markdown_tables.py README.md RULES.md ROADMAP.md CHANGELOG.md docs prompts crates deltas + +cargo check --workspace +cargo clippy --workspace --all-targets --all-features -- -D warnings + +cargo test -p ksp-store-api --all-targets --all-features +cargo test -p ksp-store-postgres-lib --all-targets --all-features +``` + +Les preuves PostgreSQL live restent opt-in pour cette tranche ; elles pourront être rejouées séparément sur une base dédiée. + +## Décisions prises + +- une promotion automatique n'est admise que pour `CompatibleMoreComplete` déjà prouvé par le comparateur partagé ; +- la projection V001 et le selector V003 restent une seule mutation transactionnelle ; +- l'ancien canonique est conservé localement avant tout remplacement de projection ; +- `canonical_revision` est incrémentée avec garde compare-and-set ; +- aucune migration existante n'est modifiée ; +- le journal append-only complet reste reporté selon le plan 038. + +## Questions ouvertes + +Aucune question bloquante pour cette tranche. Les conflict cases durables et l'outcome Worker non terminal appartiennent à `pre.007`. + +## Suite + +Après gate propre : `0.3.16-pre.007` — conflit durable minimal + intégration Worker : variante divergente conservée, case `Open`, outcome Store non terminal, arbitrage run-local limité à l'autorité du Store et health `Degraded` lorsque requis.