Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

54 changes: 33 additions & 21 deletions crates/db-app/tests/cloudsync.rs
Original file line number Diff line number Diff line change
Expand Up @@ -181,26 +181,25 @@ fn sync_step_state(result: &CloudsyncNetworkResult) -> SyncStepState {
fn describe_sync_step(result: &CloudsyncNetworkResult) -> String {
let failure = |failure: &Option<serde_json::Value>| 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"
Comment thread
fastrepl-repo-automation[bot] marked this conversation as resolved.
},
send.local_version,
send.server_version,
send.chunks,
Expand Down Expand Up @@ -506,19 +505,19 @@ 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 send.last_failure.is_some() {
evidence.push("send failure present".to_string());
}
}
if let Some(receive) = result.receive {
if let Some(error) = receive.error {
if receive.error.is_some() {
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 receive.last_failure.is_some() {
clean_receive = false;
evidence.push(format!("receive failure: {last_failure}"));
evidence.push("receive failure present".to_string());
}
}
}
Expand Down Expand Up @@ -997,6 +996,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<CloudsyncNetworkResult>) -> Result<(), String> {
let steps = Arc::new(Mutex::new(VecDeque::from(steps)));
tokio::spawn(async move {
Expand Down
1 change: 1 addition & 0 deletions crates/db-core/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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"] }
Expand Down
4 changes: 2 additions & 2 deletions crates/db-core/src/cloudsync/runtime.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
};
Expand Down
50 changes: 50 additions & 0 deletions crates/db-core/src/cloudsync/runtime/background.rs
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,38 @@ struct SyncActivity {
error: Option<String>,
}

/// 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<bool>,
}

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,
Expand Down Expand Up @@ -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| {
Expand All @@ -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,
Expand Down
35 changes: 35 additions & 0 deletions crates/db-core/src/cloudsync/runtime/tests/result_state.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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());
}
1 change: 0 additions & 1 deletion plugins/db/src/runtime/recovery.rs
Original file line number Diff line number Diff line change
Expand Up @@ -806,7 +806,6 @@ impl PluginDbRuntime {
.unwrap()
.mark_failure(&generation, &message);
tracing::warn!(
%error,
phase = ?phase,
generation = %generation,
"CloudSync recovery remains pending"
Expand Down
Loading