diff --git a/Cargo.lock b/Cargo.lock index 94f2676bdc7..7c3bd6f6d03 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4424,6 +4424,7 @@ dependencies = [ "db-change", "libsqlite3-sys", "serde", + "serde_json", "sqlx", "tempfile", "thiserror 2.0.18", diff --git a/crates/db-app/tests/cloudsync.rs b/crates/db-app/tests/cloudsync.rs index 0e8c3068cc6..3ff6df3cea6 100644 --- a/crates/db-app/tests/cloudsync.rs +++ b/crates/db-app/tests/cloudsync.rs @@ -133,11 +133,7 @@ async fn sync_ok(db: &Db, label: &str) { }) .await; let status = db.cloudsync_status().await.unwrap(); - assert!( - status.last_error.is_none(), - "{label} recorded a sync error: {:?}", - status.last_error - ); + assert!(status.last_error.is_none(), "{label} recorded a sync error"); } #[derive(Debug, PartialEq)] @@ -181,26 +177,25 @@ fn sync_step_state(result: &CloudsyncNetworkResult) -> SyncStepState { fn describe_sync_step(result: &CloudsyncNetworkResult) -> String { let failure = |failure: &Option| match failure { None => "none".to_string(), - Some(failure) => { - let field = |key: &str| { - failure - .get(key) - .map_or_else(|| "?".to_string(), |value| value.to_string()) - }; - format!( - "code={} stage={} retryable={}", - field("code"), - field("stage"), - field("retryable") - ) - } + Some(failure) => format!( + "present retryable={:?}", + failure + .get("retryable") + .and_then(serde_json::Value::as_bool) + ), }; let send = result.send.as_ref().map_or_else( || "none".to_string(), |send| { format!( "status={} local_version={} server_version={} chunks={} bytes={} last_failure={}", - send.status, + if send.status.eq_ignore_ascii_case("synced") { + "synced" + } else if send.status.eq_ignore_ascii_case("syncing") { + "syncing" + } else { + "unknown" + }, send.local_version, send.server_version, send.chunks, @@ -256,8 +251,8 @@ where ); } } - Err(error) => { - history.push(format!("transport error: {error}")); + Err(_error) => { + history.push("transport error present".to_string()); transport_failures += 1; if transport_failures == SYNC_ATTEMPTS { panic!( @@ -498,6 +493,22 @@ async fn expect_policy_sync_completed_or_denied(db: &Db, operation: &str) -> Res } tokio::time::sleep(SYNC_RETRY_INTERVAL).await; }; + let status = db.cloudsync_status().await.unwrap(); + policy_sync_outcome( + operation, + steps, + outcome.map_err(|error| error.to_string()), + status.last_error.as_deref(), + ) +} + +fn policy_sync_outcome( + operation: &str, + steps: usize, + outcome: Result, + runtime_error: Option<&str>, +) -> Result<(), String> { + let mut denied = runtime_error.is_some_and(is_rls_policy_denial); let mut evidence = vec![format!("bounded sync steps: {steps}")]; let mut clean_send = false; let mut clean_receive = true; @@ -506,33 +517,36 @@ async fn expect_policy_sync_completed_or_denied(db: &Db, operation: &str) -> Res if let Some(send) = result.send { clean_send = send.status.eq_ignore_ascii_case("synced") && send.last_failure.is_none(); - evidence.push(format!("send status: {}", send.status)); - if let Some(last_failure) = send.last_failure { - evidence.push(format!("send failure: {last_failure}")); + evidence.push(format!("send settled: {clean_send}")); + if let Some(failure) = &send.last_failure { + denied |= is_rls_policy_denial(&failure.to_string()); + evidence.push("send failure present".to_string()); } } if let Some(receive) = result.receive { - if let Some(error) = receive.error { + if let Some(error) = &receive.error { + denied |= is_rls_policy_denial(error); clean_receive = false; - evidence.push(format!("receive error: {error}")); + evidence.push("receive error present".to_string()); } - if let Some(last_failure) = receive.last_failure { + if let Some(failure) = &receive.last_failure { + denied |= is_rls_policy_denial(&failure.to_string()); clean_receive = false; - evidence.push(format!("receive failure: {last_failure}")); + evidence.push("receive failure present".to_string()); } } } - Err(error) => evidence.push(error.to_string()), + Err(error) => { + denied |= is_rls_policy_denial(&error); + evidence.push("transport error present".to_string()); + } } - let status = db.cloudsync_status().await.unwrap(); - let runtime_clean = status.last_error.is_none(); - if let Some(error) = &status.last_error { - evidence.push(format!("runtime error: {error}")); + let runtime_clean = runtime_error.is_none(); + if !runtime_clean { + evidence.push("runtime error present".to_string()); } - if evidence.iter().any(|entry| is_rls_policy_denial(entry)) - || (clean_send && clean_receive && runtime_clean) - { + if denied || (clean_send && clean_receive && runtime_clean) { Ok(()) } else { Err(format!( @@ -997,6 +1011,19 @@ fn upload_step( } } +#[test] +fn sync_diagnostics_do_not_expose_server_text() { + let secret = "private-server-content"; + let failure = serde_json::json!({"code": secret, "stage": secret, "retryable": secret}); + let mut result = upload_step(secret, 0, Some(failure.clone())); + let receive = result.receive.as_mut().unwrap(); + receive.error = Some(secret.to_string()); + receive.last_failure = Some(failure); + let diagnostic = describe_sync_step(&result); + assert!(!diagnostic.contains(secret)); + assert!(diagnostic.contains("last_failure=present")); +} + async fn scripted_sync_outcome(steps: Vec) -> Result<(), String> { let steps = Arc::new(Mutex::new(VecDeque::from(steps))); tokio::spawn(async move { @@ -1032,7 +1059,7 @@ async fn bounded_out_of_sync_upload_continues_until_synced_or_fails_with_diagnos .unwrap_err(); assert!(message.contains("did not settle after"), "{message}"); assert!( - message.contains("status=out-of-sync local_version=2 server_version=0"), + message.contains("status=unknown local_version=2 server_version=0"), "{message}" ); @@ -1047,8 +1074,26 @@ async fn bounded_out_of_sync_upload_continues_until_synced_or_fails_with_diagnos .unwrap_err(); assert!(message.contains("reported a sync failure"), "{message}"); assert!( - message.contains(r#"code="internal_error" stage="apply_payload" retryable=true"#), + message.contains("last_failure=present retryable=Some(true)"), "{message}" ); assert!(!message.contains("server detail"), "{message}"); } + +#[test] +fn policy_diagnostics_redact_runtime_errors_without_losing_denial_classification() { + let secret = "private server contents"; + let message = + policy_sync_outcome("policy", 1, Err(secret.to_string()), Some(secret)).unwrap_err(); + assert!(!message.contains(secret)); + assert!(message.contains("runtime error present")); + assert!( + policy_sync_outcome( + "policy", + 1, + Err(secret.to_string()), + Some("RLS policy denied: private server contents") + ) + .is_ok() + ); +} diff --git a/crates/db-core/Cargo.toml b/crates/db-core/Cargo.toml index cebaeab1f81..702500a7c56 100644 --- a/crates/db-core/Cargo.toml +++ b/crates/db-core/Cargo.toml @@ -13,6 +13,7 @@ anlg-db-change = { workspace = true } backon = { workspace = true } libsqlite3-sys = "0.35" serde = { workspace = true, features = ["derive"] } +serde_json = { workspace = true } sqlx = { workspace = true, features = ["runtime-tokio", "sqlite"] } thiserror = { workspace = true } tokio = { workspace = true, features = ["macros", "rt", "sync", "time"] } diff --git a/crates/db-core/src/cloudsync/runtime.rs b/crates/db-core/src/cloudsync/runtime.rs index fc66f837ca1..7531c6bfdca 100644 --- a/crates/db-core/src/cloudsync/runtime.rs +++ b/crates/db-core/src/cloudsync/runtime.rs @@ -13,9 +13,9 @@ mod background; pub(super) use background::cloudsync_activity_paused; #[cfg(test)] use background::{ - CLOUDSYNC_PROGRESS_INTERVAL, CloudsyncWake, SyncLogLevel, cloudsync_busy_delay, + CLOUDSYNC_PROGRESS_INTERVAL, CloudsyncWake, LoggedFailure, SyncLogLevel, cloudsync_busy_delay, cloudsync_next_delay, cloudsync_request_pending, cloudsync_wake_deadline, - drain_pending_changes, merge_bounded_sync_results, next_synced_change, + drain_pending_changes, logged_failure, merge_bounded_sync_results, next_synced_change, pending_cloudsync_payload_exists, sync_result_log_level, sync_result_needs_receive_progress, wait_for_retry_request_or_shutdown, }; diff --git a/crates/db-core/src/cloudsync/runtime/background.rs b/crates/db-core/src/cloudsync/runtime/background.rs index 902d178d5a2..ffc0eb02522 100644 --- a/crates/db-core/src/cloudsync/runtime/background.rs +++ b/crates/db-core/src/cloudsync/runtime/background.rs @@ -20,6 +20,38 @@ struct SyncActivity { error: Option, } +/// Server strings are untrusted even when short; only known classifications +/// and a typed retry flag may leave the process through logging. +#[derive(Debug, Default, Eq, PartialEq)] +pub(super) struct LoggedFailure { + pub(super) code: Option<&'static str>, + pub(super) stage: Option<&'static str>, + pub(super) retryable: Option, +} + +pub(super) fn logged_failure(failure: Option<&serde_json::Value>) -> LoggedFailure { + let Some(failure) = failure else { + return LoggedFailure::default(); + }; + LoggedFailure { + code: match failure.get("code").and_then(serde_json::Value::as_str) { + Some("apply_conflict") => Some("apply_conflict"), + Some("check_failed") => Some("check_failed"), + Some(_) => Some("unknown"), + None => None, + }, + stage: match failure.get("stage").and_then(serde_json::Value::as_str) { + Some("apply") => Some("apply"), + Some("check") => Some("check"), + Some(_) => Some("unknown"), + None => None, + }, + retryable: failure + .get("retryable") + .and_then(serde_json::Value::as_bool), + } +} + #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub(super) enum SyncLogLevel { Info, @@ -154,6 +186,18 @@ fn log_sync_result( failures: u32, ) { if activity.status == CloudsyncActivityStatus::Failed { + let send_failure = logged_failure( + result + .send + .as_ref() + .and_then(|send| send.last_failure.as_ref()), + ); + let receive_failure = logged_failure( + result + .receive + .as_ref() + .and_then(|receive| receive.last_failure.as_ref()), + ); tracing::warn!( trigger = ?activity.trigger, send_failed = result.send.as_ref().is_some_and(|send| { @@ -167,6 +211,12 @@ fn log_sync_result( error_kind = ?activity.error.as_deref().map(|error| { anlg_cloudsync::Error::Io(std::io::Error::other(error)).kind() }), + send_failure_code = ?send_failure.code, + send_failure_stage = ?send_failure.stage, + send_failure_retryable = ?send_failure.retryable, + receive_failure_code = ?receive_failure.code, + receive_failure_stage = ?receive_failure.stage, + receive_failure_retryable = ?receive_failure.retryable, failures, sent_bytes = activity.sent_bytes, received_bytes = activity.received_bytes, diff --git a/crates/db-core/src/cloudsync/runtime/tests/result_state.rs b/crates/db-core/src/cloudsync/runtime/tests/result_state.rs index cf7f23f4a46..bc2e4360391 100644 --- a/crates/db-core/src/cloudsync/runtime/tests/result_state.rs +++ b/crates/db-core/src/cloudsync/runtime/tests/result_state.rs @@ -435,3 +435,38 @@ async fn confirmed_prefix_retry_sends_the_tail_without_recording_a_sync_failure( ); server.await.unwrap(); } + +#[test] +fn logged_failure_does_not_expose_server_text() { + let failure = serde_json::json!({ + "code": "apply_conflict", "stage": "apply", "retryable": true, + "failedAt": "secret timestamp text", "message": "secret row contents" + }); + assert_eq!( + logged_failure(Some(&failure)), + LoggedFailure { + code: Some("apply_conflict"), + stage: Some("apply"), + retryable: Some(true), + } + ); + let failure = serde_json::json!({ + "code": "secret", "stage": "secret", "retryable": "secret", + "failedAt": "secret", "message": "secret" + }); + assert_eq!( + logged_failure(Some(&failure)), + LoggedFailure { + code: Some("unknown"), + stage: Some("unknown"), + retryable: None, + } + ); + assert_eq!( + logged_failure(Some(&serde_json::json!({ + "code": 409, "stage": {"message": "secret"}, "retryable": [] + }))), + LoggedFailure::default() + ); + assert_eq!(logged_failure(None), LoggedFailure::default()); +} diff --git a/plugins/db/src/runtime/recovery.rs b/plugins/db/src/runtime/recovery.rs index c9e13503f1c..8f899452663 100644 --- a/plugins/db/src/runtime/recovery.rs +++ b/plugins/db/src/runtime/recovery.rs @@ -806,7 +806,6 @@ impl PluginDbRuntime { .unwrap() .mark_failure(&generation, &message); tracing::warn!( - %error, phase = ?phase, generation = %generation, "CloudSync recovery remains pending"