0.3.15-pre.014-fix.001

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-store-postgres-lib/src/raw_transaction.rs
-// version: 9
+// version: 10
 
 pub(crate) mod cursor;
 
@@ -691,7 +691,7 @@ pub(crate) async fn persist_raw_transaction_acquisition(
             },
         };
         match comparison {
-            ExistingTransactionMatch::Active => ksp_store_api::RawEntityWriteOutcome::AlreadyPresent,
+            ExistingTransactionMatch::Active | ExistingTransactionMatch::ActiveIncomingTruncatedLogs => ksp_store_api::RawEntityWriteOutcome::AlreadyPresent,
             ExistingTransactionMatch::Purged => {
                 if mode == ksp_store_api::RawTransactionAcquisitionMode::ForceRehydrate {
                     let rehydrate_result = rehydrate_transaction(&sql_transaction, &raw_transaction).await;
@@ -853,6 +853,7 @@ pub(crate) async fn transition_raw_transaction_retention(
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 enum ExistingTransactionMatch {
     Active,
+    ActiveIncomingTruncatedLogs,
     Purged,
 }
 
@@ -1143,10 +1144,86 @@ fn compare_existing_transaction(
     if raw_transactions_equal(&stored, incoming) {
         return std::result::Result::Ok(ExistingTransactionMatch::Active);
     }
+    if raw_transaction_incoming_truncated_log_messages_compatible(&stored, incoming) {
+        log_raw_transaction_compatible_truncated_log_messages(network, &stored, incoming);
+        return std::result::Result::Ok(ExistingTransactionMatch::ActiveIncomingTruncatedLogs);
+    }
     log_raw_transaction_content_conflict(network, &stored, incoming);
     return std::result::Result::Err(conflict("raw_acquisition_content_conflict"));
 }
 
+fn raw_transaction_incoming_truncated_log_messages_compatible(stored: &ksp_store_api::RawTransaction, incoming: &ksp_store_api::RawTransaction) -> bool {
+    if stored.reference() != incoming.reference()
+        || stored.slot() != incoming.slot()
+        || stored.block_time() != incoming.block_time()
+        || stored.payload().format_id() != incoming.payload().format_id()
+        || stored.payload().format_version() != incoming.payload().format_version()
+    {
+        return false;
+    }
+    let stored_payload = serde_json::from_slice::<serde_json::Value>(stored.payload().bytes());
+    let incoming_payload = serde_json::from_slice::<serde_json::Value>(incoming.payload().bytes());
+    let (stored_payload, incoming_payload) = match (stored_payload, incoming_payload) {
+        (std::result::Result::Ok(serde_json::Value::Object(stored)), std::result::Result::Ok(serde_json::Value::Object(incoming))) => (stored, incoming),
+        _ => return false,
+    };
+    if stored_payload.get("transaction") != incoming_payload.get("transaction")
+        || stored_payload.get("version") != incoming_payload.get("version")
+        || stored_payload.get("transactionIndex") != incoming_payload.get("transactionIndex")
+        || json_object_other_fields_mismatch(&stored_payload, &incoming_payload, RAW_TRANSACTION_CONTENT_CONFLICT_PAYLOAD_FIELDS.as_slice())
+    {
+        return false;
+    }
+    let (stored_meta, incoming_meta) = match (stored_payload.get("meta"), incoming_payload.get("meta")) {
+        (std::option::Option::Some(serde_json::Value::Object(stored)), std::option::Option::Some(serde_json::Value::Object(incoming))) => (stored, incoming),
+        _ => return false,
+    };
+    if raw_meta_other_than_log_messages_mismatch(stored_meta, incoming_meta) {
+        return false;
+    }
+    let (stored_logs, incoming_logs) = match (stored_meta.get("logMessages"), incoming_meta.get("logMessages")) {
+        (std::option::Option::Some(serde_json::Value::Array(stored)), std::option::Option::Some(serde_json::Value::Array(incoming))) => {
+            (stored.as_slice(), incoming.as_slice())
+        },
+        _ => return false,
+    };
+    return raw_log_messages_incoming_truncated_compatible(stored_logs, incoming_logs);
+}
+
+fn raw_meta_other_than_log_messages_mismatch(
+    stored: &serde_json::Map<std::string::String, serde_json::Value>,
+    incoming: &serde_json::Map<std::string::String, serde_json::Value>,
+) -> bool {
+    let stored_without_logs = stored.iter().filter(|(key, _)| return key.as_str() != "logMessages").collect::<std::collections::BTreeMap<_, _>>();
+    let incoming_without_logs = incoming.iter().filter(|(key, _)| return key.as_str() != "logMessages").collect::<std::collections::BTreeMap<_, _>>();
+    return stored_without_logs != incoming_without_logs;
+}
+
+fn raw_log_messages_incoming_truncated_compatible(stored: &[serde_json::Value], incoming: &[serde_json::Value]) -> bool {
+    if raw_log_messages_exact_truncation_marker_count(stored) != 0 {
+        return false;
+    }
+    let marker_index = match incoming.iter().position(raw_log_message_is_exact_truncation_marker) {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return false,
+    };
+    if raw_log_messages_exact_truncation_marker_count(incoming) != 1 || marker_index >= stored.len() {
+        return false;
+    }
+    if incoming[..marker_index] != stored[..marker_index] {
+        return false;
+    }
+    return true;
+}
+
+fn raw_log_messages_exact_truncation_marker_count(values: &[serde_json::Value]) -> usize {
+    return values.iter().filter(|value| return raw_log_message_is_exact_truncation_marker(value)).count();
+}
+
+fn raw_log_message_is_exact_truncation_marker(value: &serde_json::Value) -> bool {
+    return matches!(value, serde_json::Value::String(line) if line == "Log truncated");
+}
+
 fn raw_transaction_content_conflict_diagnostic(
     stored: &ksp_store_api::RawTransaction,
     incoming: &ksp_store_api::RawTransaction,
@@ -1374,6 +1451,25 @@ fn json_object_other_fields_mismatch(
     return stored_other != incoming_other;
 }
 
+fn log_raw_transaction_compatible_truncated_log_messages(
+    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::debug!(
+        target: crate::TRACING_TARGET,
+        domain = "store.raw_transaction.compatible_partial",
+        network = network.as_str(),
+        slot = stored.slot(),
+        stored_log_messages_count = diagnostic.log_messages.stored_count,
+        incoming_log_messages_count = diagnostic.log_messages.incoming_count,
+        incoming_log_messages_has_truncation_marker = diagnostic.log_messages.incoming_has_truncation_marker,
+        "PostgreSQL Store accepted compatible truncated RAW transaction logs without replacing canonical content"
+    );
+    return;
+}
+
 fn log_raw_transaction_content_conflict(
     network: &ksp_store_api::RawNetworkId,
     stored: &ksp_store_api::RawTransaction,