0.3.15-pre.012-fix.004

crates/ksp-store-postgres-lib/Cargo.toml

@@ -1,5 +1,5 @@
 # file: crates/ksp-store-postgres-lib/Cargo.toml
-# version: 3
+# version: 4
 
 [package]
 name = "ksp-store-postgres-lib"
@@ -14,6 +14,7 @@ ksp-store-api = { path = "../ksp-store-api" }
 rustls = { workspace = true, features = ["aws_lc_rs", "std", "tls12"] }
 rustls-native-certs.workspace = true
 sha2.workspace = true
+serde_json = { workspace = true }
 tokio = { workspace = true, features = ["rt", "time"] }
 tokio-postgres = { workspace = true, features = ["runtime"] }
 tokio-postgres-rustls = { workspace = true, features = ["aws-lc-rs"] }

crates/ksp-store-postgres-lib/src/raw_transaction.rs

@@ -1,10 +1,11 @@
 // file: crates/ksp-store-postgres-lib/src/raw_transaction.rs
-// version: 7
+// version: 8
 
 pub(crate) mod cursor;
 
 const DELETE_ARCHIVE_PAYLOAD_SQL: &str = "DELETE FROM ksp_raw_transaction_archive_payloads WHERE signature = $1";
 const GET_ARCHIVE_PAYLOAD_SQL: &str = "SELECT payload FROM ksp_raw_transaction_archive_payloads WHERE signature = $1";
+const GET_FIRST_TRANSACTION_OBSERVATION_SQL: &str = "SELECT observation_key, transaction_signature, provider, protocol, acquisition_method, origin, received_at_unix_millis, capture_session_id, commitment, endpoint_id, filter_id, observed_at_unix_millis, source_payload_hash, source_payload_size_bytes FROM ksp_raw_transaction_observations WHERE transaction_signature = $1 ORDER BY received_at_unix_millis ASC, observation_key ASC LIMIT 1";
 const GET_OBSERVATION_SQL: &str = "SELECT observation_key, transaction_signature, provider, protocol, acquisition_method, origin, received_at_unix_millis, capture_session_id, commitment, endpoint_id, filter_id, observed_at_unix_millis, source_payload_hash, source_payload_size_bytes FROM ksp_raw_transaction_observations WHERE observation_key = $1";
 const GET_RETENTION_SQL: &str = "SELECT retention_state FROM ksp_raw_transactions WHERE signature = $1";
 const GET_TOMBSTONE_SQL: &str = "SELECT signature, slot::text AS slot_text, block_time_unix_millis, format_id, format_version, content_hash, retention_state FROM ksp_raw_transactions WHERE signature = $1";
@@ -23,12 +24,56 @@ const LOCK_RETENTION_TRANSACTION_SQL: &str =
     "SELECT block_time_unix_millis, payload, retention_state FROM ksp_raw_transactions WHERE signature = $1 FOR UPDATE";
 const LOCK_TRANSACTION_SQL: &str = "SELECT signature, slot::text AS slot_text, block_time_unix_millis, format_id, format_version, content_hash, payload, retention_state, NULL::BYTEA AS archive_payload FROM ksp_raw_transactions WHERE signature = $1 FOR UPDATE";
 const LOCK_TRANSACTION_STATE_SQL: &str = "SELECT retention_state FROM ksp_raw_transactions WHERE signature = $1 FOR UPDATE";
+const RAW_TRANSACTION_CONTENT_CONFLICT_META_FIELDS: [&str; 15] = [
+    "err",
+    "status",
+    "fee",
+    "preBalances",
+    "postBalances",
+    "innerInstructions",
+    "logMessages",
+    "preTokenBalances",
+    "postTokenBalances",
+    "rewards",
+    "loadedAddresses",
+    "returnData",
+    "computeUnitsConsumed",
+    "costUnits",
+    "accounts",
+];
+const RAW_TRANSACTION_CONTENT_CONFLICT_PAYLOAD_FIELDS: [&str; 4] = ["transaction", "meta", "version", "transactionIndex"];
 const REHYDRATE_TRANSACTION_SQL: &str =
     "UPDATE ksp_raw_transactions SET block_time_unix_millis = $2, payload = $3, retention_state = 'full' WHERE signature = $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_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'";
 
+struct RawPayloadContentConflictDiagnostic {
+    available: bool,
+    meta_mismatch: bool,
+    meta_mismatch_fields: std::string::String,
+    other_mismatch: bool,
+    transaction_index_mismatch: bool,
+    transaction_mismatch: bool,
+    version_mismatch: bool,
+}
+
+struct RawTransactionContentConflictDiagnostic {
+    block_time_mismatch: bool,
+    content_hash_mismatch: bool,
+    format_id_mismatch: bool,
+    format_version_mismatch: bool,
+    meta_mismatch_fields: std::string::String,
+    payload_bytes_mismatch: bool,
+    payload_diagnostic_available: bool,
+    payload_meta_mismatch: bool,
+    payload_other_mismatch: bool,
+    payload_transaction_index_mismatch: bool,
+    payload_transaction_mismatch: bool,
+    payload_version_mismatch: bool,
+    slot_mismatch: bool,
+}
+
 struct RawListDbRow {
     signature: std::vec::Vec<u8>,
     slot_text: std::string::String,
@@ -619,7 +664,12 @@ pub(crate) async fn persist_raw_transaction_acquisition(
         let comparison_result = compare_existing_transaction(network, locked, &raw_transaction);
         let comparison = match comparison_result {
             std::result::Result::Ok(value) => value,
-            std::result::Result::Err(error) => return std::result::Result::Err(error),
+            std::result::Result::Err(error) => {
+                if error.kind() == crate::PostgresBackendErrorKind::Conflict && error.phase() == "raw_acquisition_content_conflict" {
+                    log_raw_transaction_content_conflict_provenance(&sql_transaction, network, &observation).await;
+                }
+                return std::result::Result::Err(error);
+            },
         };
         match comparison {
             ExistingTransactionMatch::Active => ksp_store_api::RawEntityWriteOutcome::AlreadyPresent,
@@ -1074,9 +1124,178 @@ fn compare_existing_transaction(
     if raw_transactions_equal(&stored, incoming) {
         return std::result::Result::Ok(ExistingTransactionMatch::Active);
     }
+    log_raw_transaction_content_conflict(network, &stored, incoming);
     return std::result::Result::Err(conflict("raw_acquisition_content_conflict"));
 }
 
+fn raw_transaction_content_conflict_diagnostic(
+    stored: &ksp_store_api::RawTransaction,
+    incoming: &ksp_store_api::RawTransaction,
+) -> RawTransactionContentConflictDiagnostic {
+    let payload = raw_payload_content_conflict_diagnostic(stored.payload().bytes(), incoming.payload().bytes());
+    return RawTransactionContentConflictDiagnostic {
+        block_time_mismatch: stored.block_time() != incoming.block_time(),
+        content_hash_mismatch: stored.payload().content_hash() != incoming.payload().content_hash(),
+        format_id_mismatch: stored.payload().format_id() != incoming.payload().format_id(),
+        format_version_mismatch: stored.payload().format_version() != incoming.payload().format_version(),
+        meta_mismatch_fields: payload.meta_mismatch_fields,
+        payload_bytes_mismatch: stored.payload().bytes() != incoming.payload().bytes(),
+        payload_diagnostic_available: payload.available,
+        payload_meta_mismatch: payload.meta_mismatch,
+        payload_other_mismatch: payload.other_mismatch,
+        payload_transaction_index_mismatch: payload.transaction_index_mismatch,
+        payload_transaction_mismatch: payload.transaction_mismatch,
+        payload_version_mismatch: payload.version_mismatch,
+        slot_mismatch: stored.slot() != incoming.slot(),
+    };
+}
+
+fn raw_payload_content_conflict_diagnostic(stored: &[u8], incoming: &[u8]) -> RawPayloadContentConflictDiagnostic {
+    let stored = serde_json::from_slice::<serde_json::Value>(stored);
+    let incoming = serde_json::from_slice::<serde_json::Value>(incoming);
+    let (stored, incoming) = match (stored, incoming) {
+        (std::result::Result::Ok(serde_json::Value::Object(stored)), std::result::Result::Ok(serde_json::Value::Object(incoming))) => (stored, incoming),
+        _ => {
+            return RawPayloadContentConflictDiagnostic {
+                available: false,
+                meta_mismatch: false,
+                meta_mismatch_fields: "unavailable".to_owned(),
+                other_mismatch: false,
+                transaction_index_mismatch: false,
+                transaction_mismatch: false,
+                version_mismatch: false,
+            };
+        },
+    };
+    let transaction_mismatch = stored.get("transaction") != incoming.get("transaction");
+    let meta_mismatch = stored.get("meta") != incoming.get("meta");
+    let version_mismatch = stored.get("version") != incoming.get("version");
+    let transaction_index_mismatch = stored.get("transactionIndex") != incoming.get("transactionIndex");
+    let other_mismatch = json_object_other_fields_mismatch(&stored, &incoming, RAW_TRANSACTION_CONTENT_CONFLICT_PAYLOAD_FIELDS.as_slice());
+    let meta_mismatch_fields = raw_meta_content_conflict_fields(stored.get("meta"), incoming.get("meta"));
+    return RawPayloadContentConflictDiagnostic {
+        available: true,
+        meta_mismatch,
+        meta_mismatch_fields,
+        other_mismatch,
+        transaction_index_mismatch,
+        transaction_mismatch,
+        version_mismatch,
+    };
+}
+
+fn raw_meta_content_conflict_fields(stored: std::option::Option<&serde_json::Value>, incoming: std::option::Option<&serde_json::Value>) -> std::string::String {
+    if stored == incoming {
+        return "none".to_owned();
+    }
+    let (stored, incoming) = match (stored, incoming) {
+        (std::option::Option::Some(serde_json::Value::Object(stored)), std::option::Option::Some(serde_json::Value::Object(incoming))) => (stored, incoming),
+        _ => return "shape".to_owned(),
+    };
+    let mut fields = std::vec::Vec::new();
+    for field in RAW_TRANSACTION_CONTENT_CONFLICT_META_FIELDS {
+        if stored.get(field) != incoming.get(field) {
+            fields.push(field);
+        }
+    }
+    if json_object_other_fields_mismatch(stored, incoming, RAW_TRANSACTION_CONTENT_CONFLICT_META_FIELDS.as_slice()) {
+        fields.push("other");
+    }
+    if fields.is_empty() {
+        return "shape".to_owned();
+    }
+    return fields.join(",");
+}
+
+fn json_object_other_fields_mismatch(
+    stored: &serde_json::Map<std::string::String, serde_json::Value>,
+    incoming: &serde_json::Map<std::string::String, serde_json::Value>,
+    known_fields: &[&str],
+) -> bool {
+    let stored_other = stored.iter().filter(|(key, _)| return !known_fields.contains(&key.as_str())).collect::<std::collections::BTreeMap<_, _>>();
+    let incoming_other = incoming.iter().filter(|(key, _)| return !known_fields.contains(&key.as_str())).collect::<std::collections::BTreeMap<_, _>>();
+    return stored_other != incoming_other;
+}
+
+fn log_raw_transaction_content_conflict(
+    network: &ksp_store_api::RawNetworkId,
+    stored: &ksp_store_api::RawTransaction,
+    incoming: &ksp_store_api::RawTransaction,
+) {
+    let diagnostic = raw_transaction_content_conflict_diagnostic(stored, incoming);
+    ksp_logging_lib::warn!(
+        target: crate::TRACING_TARGET,
+        domain = "store.raw_transaction.content_conflict",
+        network = network.as_str(),
+        slot_mismatch = diagnostic.slot_mismatch,
+        block_time_mismatch = diagnostic.block_time_mismatch,
+        format_id_mismatch = diagnostic.format_id_mismatch,
+        format_version_mismatch = diagnostic.format_version_mismatch,
+        content_hash_mismatch = diagnostic.content_hash_mismatch,
+        payload_bytes_mismatch = diagnostic.payload_bytes_mismatch,
+        payload_diagnostic_available = diagnostic.payload_diagnostic_available,
+        transaction_mismatch = diagnostic.payload_transaction_mismatch,
+        meta_mismatch = diagnostic.payload_meta_mismatch,
+        version_mismatch = diagnostic.payload_version_mismatch,
+        transaction_index_mismatch = diagnostic.payload_transaction_index_mismatch,
+        payload_other_mismatch = diagnostic.payload_other_mismatch,
+        meta_mismatch_fields = diagnostic.meta_mismatch_fields.as_str(),
+        "PostgreSQL Store rejected divergent canonical RAW transaction"
+    );
+    return;
+}
+
+async fn log_raw_transaction_content_conflict_provenance(
+    sql_transaction: &deadpool_postgres::Transaction<'_>,
+    network: &ksp_store_api::RawNetworkId,
+    incoming_observation: &ksp_store_api::RawTransactionObservation,
+) {
+    let signature = incoming_observation.transaction().signature();
+    let signature_bytes: &[u8] = signature.as_bytes();
+    let stored_observation = match sql_transaction.query_opt(GET_FIRST_TRANSACTION_OBSERVATION_SQL, &[&signature_bytes]).await {
+        std::result::Result::Ok(std::option::Option::Some(row)) => match raw_observation_db_row(&row) {
+            std::result::Result::Ok(row) => decode_raw_observation_row(network, row).ok(),
+            std::result::Result::Err(_) => std::option::Option::None,
+        },
+        std::result::Result::Ok(std::option::Option::None) | std::result::Result::Err(_) => std::option::Option::None,
+    };
+    let incoming = incoming_observation.provenance();
+    let incoming_endpoint = incoming.endpoint_id().map(|value| return value.as_str()).unwrap_or("none");
+    let incoming_commitment = incoming.commitment().map(|value| return value.as_str()).unwrap_or("none");
+    let stored_available = stored_observation.is_some();
+    let (stored_provider, stored_protocol, stored_method, stored_endpoint, stored_commitment) = match stored_observation.as_ref() {
+        std::option::Option::Some(observation) => {
+            let provenance = observation.provenance();
+            (
+                provenance.provider().as_str(),
+                provenance.protocol().as_str(),
+                provenance.acquisition_method().as_str(),
+                provenance.endpoint_id().map(|value| return value.as_str()).unwrap_or("none"),
+                provenance.commitment().map(|value| return value.as_str()).unwrap_or("none"),
+            )
+        },
+        std::option::Option::None => ("unknown", "unknown", "unknown", "unknown", "unknown"),
+    };
+    ksp_logging_lib::warn!(
+        target: crate::TRACING_TARGET,
+        domain = "store.raw_transaction.content_conflict",
+        network = network.as_str(),
+        stored_observation_available = stored_available,
+        stored_provider = stored_provider,
+        stored_protocol = stored_protocol,
+        stored_acquisition_method = stored_method,
+        stored_endpoint_id = stored_endpoint,
+        stored_commitment = stored_commitment,
+        incoming_provider = incoming.provider().as_str(),
+        incoming_protocol = incoming.protocol().as_str(),
+        incoming_acquisition_method = incoming.acquisition_method().as_str(),
+        incoming_endpoint_id = incoming_endpoint,
+        incoming_commitment = incoming_commitment,
+        "PostgreSQL Store classified canonical RAW transaction conflict provenance"
+    );
+    return;
+}
+
 fn raw_transactions_equal(left: &ksp_store_api::RawTransaction, right: &ksp_store_api::RawTransaction) -> bool {
     return left.reference() == right.reference()
         && left.slot() == right.slot()

crates/ksp-store-postgres-lib/tests/dependency_boundary.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-store-postgres-lib/tests/dependency_boundary.rs
-// version: 29
+// version: 30
 
 #![warn(missing_docs)]
 #![deny(unreachable_pub)]
@@ -10,9 +10,17 @@
 #[test]
 fn pre_005_backend_owns_exact_physical_runtime_dependencies_without_reverse_facade_edge() {
     let manifest = include_str!("../Cargo.toml");
-    for required in
-        ["deadpool-postgres", "ksp-logging-lib", "ksp-store-api", "rustls", "rustls-native-certs", "sha2", "tokio-postgres", "tokio-postgres-rustls"]
-    {
+    for required in [
+        "deadpool-postgres",
+        "ksp-logging-lib",
+        "ksp-store-api",
+        "rustls",
+        "rustls-native-certs",
+        "serde_json",
+        "sha2",
+        "tokio-postgres",
+        "tokio-postgres-rustls",
+    ] {
         assert!(manifest.contains(required), "missing PostgreSQL backend dependency: {required}");
     }
     for forbidden in ["ksp-store-lib", "ksp-config-lib", "ksp-materializer", "ksp-program", "ksp-onchain-transport-lib", "ksp-offchain-transport-lib", "sqlx"] {

crates/ksp-store-postgres-lib/tests/hardening_completeness.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-store-postgres-lib/tests/hardening_completeness.rs
-// version: 23
+// version: 24
 
 #![warn(missing_docs)]
 #![deny(unreachable_pub)]
@@ -136,6 +136,7 @@ fn pre_009_backend_modules_exports_and_manifest_dependencies_are_exact() {
         "ksp-store-api",
         "rustls",
         "rustls-native-certs",
+        "serde_json",
         "sha2",
         "tokio",
         "tokio-postgres",
@@ -561,3 +562,47 @@ fn v0_3_8_pre_005_raw_account_keyset_sql_remains_offset_free_while_inspection_is
     }
     return;
 }
+
+#[test]
+fn v0_3_15_pre_012_fix_004_content_conflict_diagnostic_is_field_only_bounded_and_provenance_safe() {
+    let source = include_str!("../src/raw_transaction.rs");
+    for required in [
+        "slot_mismatch",
+        "block_time_mismatch",
+        "transaction_mismatch",
+        "meta_mismatch",
+        "version_mismatch",
+        "transaction_index_mismatch",
+        "meta_mismatch_fields",
+        "stored_provider",
+        "stored_protocol",
+        "stored_acquisition_method",
+        "stored_endpoint_id",
+        "incoming_provider",
+        "incoming_protocol",
+        "incoming_acquisition_method",
+        "incoming_endpoint_id",
+        "RAW_TRANSACTION_CONTENT_CONFLICT_META_FIELDS",
+        "\"rewards\"",
+        "\"logMessages\"",
+        "\"innerInstructions\"",
+    ] {
+        assert!(source.contains(required), "missing bounded RAW conflict diagnostic field: {required}");
+    }
+    let start = source.find("fn log_raw_transaction_content_conflict(");
+    assert!(start.is_some(), "missing RAW conflict structural logger");
+    let Some(start) = start else {
+        return;
+    };
+    let end = source[start..].find("fn raw_transactions_equal(");
+    assert!(end.is_some(), "missing RAW conflict logger end marker");
+    let Some(end) = end else {
+        return;
+    };
+    let diagnostic_surface = &source[start..start + end];
+    for forbidden in ["\n        signature =", "payload =", "payload_bytes =", "content_hash =", "source_payload_hash =", "capture_session_id", "filter_id"] {
+        assert!(!diagnostic_surface.contains(forbidden), "RAW conflict logger exposes forbidden material: {forbidden}");
+    }
+    assert!(source.contains("fields.push(\"other\")"));
+    return;
+}

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: 6
+// version: 7
 
 fn network() -> ksp_store_api::RawNetworkId {
     return match ksp_store_api::RawNetworkId::new("devnet") {
@@ -392,6 +392,43 @@ fn pre_005_purged_tombstone_matches_only_retained_identity_metadata() {
     return;
 }
 
+#[test]
+fn v0_3_15_pre_012_fix_004_content_conflict_diagnostic_classifies_raw_v1_components_without_payload_values() {
+    let stored = raw_transaction(
+        9,
+        br#"{"transaction":["AQID","base64"],"meta":{"err":null,"fee":5000,"logMessages":["left"],"rewards":[]},"version":0,"transactionIndex":7}"#,
+        7,
+    );
+    let incoming = raw_transaction(9, br#"{"transaction":["AQID","base64"],"meta":{"err":null,"fee":5000,"logMessages":["right"],"rewards":[{"lamports":1}]},"version":0,"transactionIndex":8}"#, 8);
+    let diagnostic = super::raw_transaction_content_conflict_diagnostic(&stored, &incoming);
+    assert!(!diagnostic.slot_mismatch);
+    assert!(!diagnostic.block_time_mismatch);
+    assert!(!diagnostic.format_id_mismatch);
+    assert!(!diagnostic.format_version_mismatch);
+    assert!(diagnostic.content_hash_mismatch);
+    assert!(diagnostic.payload_bytes_mismatch);
+    assert!(diagnostic.payload_diagnostic_available);
+    assert!(!diagnostic.payload_transaction_mismatch);
+    assert!(diagnostic.payload_meta_mismatch);
+    assert!(!diagnostic.payload_version_mismatch);
+    assert!(diagnostic.payload_transaction_index_mismatch);
+    assert!(!diagnostic.payload_other_mismatch);
+    assert_eq!(diagnostic.meta_mismatch_fields, "logMessages,rewards");
+    return;
+}
+
+#[test]
+fn v0_3_15_pre_012_fix_004_content_conflict_diagnostic_collapses_unknown_meta_keys_to_other() {
+    let stored = raw_transaction(9, br#"{"transaction":["AQID","base64"],"meta":{"err":null,"futureField":1},"version":0}"#, 7);
+    let incoming = raw_transaction(9, br#"{"transaction":["BAUG","base64"],"meta":{"err":null,"futureField":2},"version":1}"#, 8);
+    let diagnostic = super::raw_transaction_content_conflict_diagnostic(&stored, &incoming);
+    assert!(diagnostic.payload_transaction_mismatch);
+    assert!(diagnostic.payload_meta_mismatch);
+    assert!(diagnostic.payload_version_mismatch);
+    assert_eq!(diagnostic.meta_mismatch_fields, "other");
+    return;
+}
+
 #[test]
 fn pre_005_observation_idempotence_compares_reference_and_complete_provenance() {
     let incoming_transaction = raw_transaction(8, &[1, 2, 3], 4);