Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
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.

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
56 changes: 56 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,42 @@ struct SyncActivity {
error: Option<String>,
}

const LOGGED_FAILURE_FIELD_MAX_CHARS: usize = 64;

/// Fields of a server-reported `lastFailure` that are safe to log. The full
/// failure, and the error string built from it, can carry server messages.
#[derive(Debug, Default, Eq, PartialEq)]
pub(super) struct LoggedFailure {
pub(super) code: Option<String>,
pub(super) stage: Option<String>,
pub(super) retryable: Option<bool>,
pub(super) failed_at: Option<String>,
}

pub(super) fn logged_failure(failure: Option<&serde_json::Value>) -> LoggedFailure {
let Some(failure) = failure else {
return LoggedFailure::default();
};
LoggedFailure {
code: logged_failure_scalar(failure.get("code")),
stage: logged_failure_scalar(failure.get("stage")),
retryable: failure
.get("retryable")
.and_then(serde_json::Value::as_bool),
failed_at: logged_failure_scalar(failure.get("failedAt")),
}
}

fn logged_failure_scalar(value: Option<&serde_json::Value>) -> Option<String> {
match value? {
serde_json::Value::String(value) => {
Some(value.chars().take(LOGGED_FAILURE_FIELD_MAX_CHARS).collect())
}
serde_json::Value::Number(value) => Some(value.to_string()),
_ => None,
}
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(super) enum SyncLogLevel {
Info,
Expand Down Expand Up @@ -154,6 +190,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 +215,14 @@ 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,
send_failure_failed_at = ?send_failure.failed_at,
receive_failure_code = ?receive_failure.code,
receive_failure_stage = ?receive_failure.stage,
receive_failure_retryable = ?receive_failure.retryable,
receive_failure_failed_at = ?receive_failure.failed_at,
failures,
sent_bytes = activity.sent_bytes,
received_bytes = activity.received_bytes,
Expand Down
49 changes: 49 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,52 @@ async fn confirmed_prefix_retry_sends_the_tail_without_recording_a_sync_failure(
);
server.await.unwrap();
}

#[test]
fn logged_failure_keeps_only_safe_scalar_fields() {
let failure = serde_json::json!({
"jobId": "job-1",
"code": "apply_conflict",
"stage": "apply",
"message": "row with secret contents conflicts",
"retryable": true,
"failedAt": "2026-10-17T18:34:32Z",
});
assert_eq!(
logged_failure(Some(&failure)),
LoggedFailure {
code: Some("apply_conflict".to_string()),
stage: Some("apply".to_string()),
retryable: Some(true),
failed_at: Some("2026-10-17T18:34:32Z".to_string()),
}
);

let failure = serde_json::json!({
"code": 409,
"stage": {"message": "nested server text"},
"retryable": "yes",
"failedAt": 1_792_262_072,
});
assert_eq!(
logged_failure(Some(&failure)),
LoggedFailure {
code: Some("409".to_string()),
stage: None,
retryable: None,
failed_at: Some("1792262072".to_string()),
}
);

let failure = serde_json::json!({"code": "x".repeat(100)});
assert_eq!(
logged_failure(Some(&failure)).code.map(|code| code.len()),
Some(64)
);

assert_eq!(logged_failure(None), LoggedFailure::default());
assert_eq!(
logged_failure(Some(&serde_json::json!("previous apply failed"))),
LoggedFailure::default()
);
}
Loading