From 4dda6400ce3eebb9893a2b0ec5facd731d549cd2 Mon Sep 17 00:00:00 2001 From: Kang Date: Tue, 11 Aug 2026 12:03:39 +0800 Subject: [PATCH] feat(cloudSync): enhance conflict resolution and snapshot management (#397, #405) - Introduced a new action for resolving cloud sync conflicts by recovering the current remote snapshot. - Updated the SyncBackupHistoryPanel and SyncBackupTab components to handle remote inconsistency scenarios, providing appropriate user feedback and actions. - Added a utility function to determine if a remote conflict is inconsistent, improving conflict detection logic. - Enhanced the CloudSyncManager to manage remote snapshot resolutions, including handling legacy snapshots and ensuring consistency during sync operations. - Implemented new error handling for cloud sync operations, improving robustness and user experience. --- src-tauri/src/config/cloud_sync.rs | 12 + src-tauri/src/core/cloud_sync/gc.rs | 220 ++++++++++ src-tauri/src/core/cloud_sync/manager.rs | 402 ++++++++++++------ src-tauri/src/core/cloud_sync/migration.rs | 76 ++++ src-tauri/src/core/cloud_sync/mod.rs | 3 + src-tauri/src/core/cloud_sync/operator.rs | 117 +++-- src-tauri/src/core/cloud_sync/protocol.rs | 391 +++++++++++++++++ src-tauri/src/core/cloud_sync/remote.rs | 9 + src-tauri/src/core/portable_snapshot/hash.rs | 2 +- src-tauri/src/error.rs | 33 ++ .../panel/SyncBackupHistoryPanel.tsx | 48 ++- src/components/settings/SyncBackupTab.tsx | 78 +++- src/lib/cloudSync.ts | 4 + src/types/global.d.ts | 4 + 14 files changed, 1222 insertions(+), 177 deletions(-) create mode 100644 src-tauri/src/core/cloud_sync/gc.rs create mode 100644 src-tauri/src/core/cloud_sync/migration.rs create mode 100644 src-tauri/src/core/cloud_sync/protocol.rs diff --git a/src-tauri/src/config/cloud_sync.rs b/src-tauri/src/config/cloud_sync.rs index a94e00af..66d6381a 100644 --- a/src-tauri/src/config/cloud_sync.rs +++ b/src-tauri/src/config/cloud_sync.rs @@ -231,12 +231,20 @@ pub struct CloudSyncState { pub struct CloudConflictPreview { pub detected_at_ms: u64, pub provider: String, + #[serde(default = "default_conflict_kind")] + pub kind: String, pub local_payload_hash: String, pub remote_payload_hash: String, pub remote_revision: String, pub remote_created_at_ms: u64, #[serde(default)] pub remote_device_id: String, + #[serde(default)] + pub recovery_revision: Option, + #[serde(default)] + pub recovery_payload_hash: Option, + #[serde(default)] + pub recovery_created_at_ms: Option, pub message: String, } @@ -322,6 +330,10 @@ fn default_status_state() -> String { "idle".to_string() } +fn default_conflict_kind() -> String { + "content_conflict".to_string() +} + pub fn load_cloud_sync_settings(app: &AppHandle) -> AppResult { let _ = app; storage::load_settings_doc(SettingsDocKey::CloudSyncSettings) diff --git a/src-tauri/src/core/cloud_sync/gc.rs b/src-tauri/src/core/cloud_sync/gc.rs new file mode 100644 index 00000000..433fdc57 --- /dev/null +++ b/src-tauri/src/core/cloud_sync/gc.rs @@ -0,0 +1,220 @@ +use std::collections::HashSet; +use std::time::Duration; + +use crate::error::AppResult; + +use super::operator::CloudRemote; +use super::protocol::sync_snapshot_file; +use super::remote::{ + RemoteSyncPointer, SYNC_SNAPSHOTS_DIR, current_time_ms, is_legacy_sync_snapshot_path, + remote_path, +}; + +pub(super) const SYNC_SNAPSHOT_KEEP_RECENT: usize = 5; +pub(super) const SYNC_SNAPSHOT_GC_GRACE_PERIOD: Duration = Duration::from_secs(24 * 60 * 60); + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(super) struct SnapshotGcEntry { + pub path: String, + pub revision_id: String, + pub created_at_ms: u64, + pub deletable: bool, +} + +pub(super) async fn cleanup_sync_snapshots( + remote: &CloudRemote, + remote_root: &str, + latest: Option<&RemoteSyncPointer>, +) { + let result = collect_snapshots(remote, remote_root) + .await + .map(|snapshots| { + plan_snapshot_gc( + snapshots, + latest.map(|pointer| pointer.revision_id.as_str()), + current_time_ms(), + SYNC_SNAPSHOT_KEEP_RECENT, + SYNC_SNAPSHOT_GC_GRACE_PERIOD, + ) + }); + + let paths = match result { + Ok(paths) => paths, + Err(error) => { + tracing::warn!("Failed to plan cloud sync snapshot cleanup: {}", error); + return; + } + }; + + for path in paths { + if let Err(error) = remote.delete(&path).await { + tracing::warn!( + path = %path, + error = %error, + "Failed to delete old cloud sync snapshot" + ); + } + } +} + +async fn collect_snapshots( + remote: &CloudRemote, + remote_root: &str, +) -> AppResult> { + let prefix = remote_path(remote_root, SYNC_SNAPSHOTS_DIR); + let paths = remote.list_files(&prefix).await?; + let mut snapshots = Vec::new(); + for path in paths + .into_iter() + .filter(|path| is_legacy_sync_snapshot_path(path, remote_root)) + { + let Some(revision_id) = snapshot_revision_from_path(&path) else { + snapshots.push(SnapshotGcEntry { + path, + revision_id: String::new(), + created_at_ms: 0, + deletable: false, + }); + continue; + }; + let pointer = RemoteSyncPointer { + schema_version: 2, + revision_id: revision_id.clone(), + created_at_ms: 0, + payload_hash: String::new(), + device_id: String::new(), + app_version: String::new(), + }; + match read_snapshot_for_gc(remote, remote_root, &pointer).await { + Ok((created_at_ms, payload_hash)) => snapshots.push(SnapshotGcEntry { + path, + revision_id, + created_at_ms, + deletable: !payload_hash.is_empty(), + }), + Err(error) => { + tracing::warn!( + path = %path, + error = %error, + "Cloud sync snapshot is not readable; keeping it during cleanup" + ); + snapshots.push(SnapshotGcEntry { + path, + revision_id, + created_at_ms: 0, + deletable: false, + }); + } + } + } + Ok(snapshots) +} + +async fn read_snapshot_for_gc( + remote: &CloudRemote, + remote_root: &str, + pointer: &RemoteSyncPointer, +) -> AppResult<(u64, String)> { + let Some(raw) = remote + .read_if_exists(&remote_path( + remote_root, + &sync_snapshot_file(&pointer.revision_id), + )) + .await? + else { + return Ok((0, String::new())); + }; + let decrypted = super::crypto::decrypt_snapshot_bytes(&raw)?; + let snapshot = crate::core::portable_snapshot::decode_portable_snapshot(&decrypted)?; + if snapshot.revision_id != pointer.revision_id { + return Err(crate::error::CloudSyncError::RevisionMismatch { + pointer_revision: pointer.revision_id.clone(), + snapshot_revision: snapshot.revision_id, + } + .into()); + } + Ok((snapshot.created_at_ms, snapshot.payload_hash)) +} + +pub(super) fn plan_snapshot_gc( + mut snapshots: Vec, + latest_revision: Option<&str>, + now_ms: u64, + keep_recent: usize, + grace_period: Duration, +) -> Vec { + let mut protected: HashSet = HashSet::new(); + if let Some(latest_revision) = latest_revision { + protected.insert(latest_revision.to_string()); + } + + snapshots.sort_by_key(|snapshot| snapshot.created_at_ms); + for snapshot in snapshots.iter().rev().take(keep_recent) { + protected.insert(snapshot.revision_id.clone()); + } + + let grace_ms = u64::try_from(grace_period.as_millis()).unwrap_or(u64::MAX); + snapshots + .into_iter() + .filter(|snapshot| snapshot.deletable) + .filter(|snapshot| !protected.contains(&snapshot.revision_id)) + .filter(|snapshot| now_ms.saturating_sub(snapshot.created_at_ms) > grace_ms) + .map(|snapshot| snapshot.path) + .collect() +} + +fn snapshot_revision_from_path(path: &str) -> Option { + let filename = path.rsplit('/').next()?; + filename + .strip_suffix(".redb.enc") + .filter(|revision| !revision.is_empty()) + .map(ToString::to_string) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn entry(revision: &str, created_at_ms: u64) -> SnapshotGcEntry { + SnapshotGcEntry { + path: format!("nyaterm/sync/snapshots/{revision}.redb.enc"), + revision_id: revision.to_string(), + created_at_ms, + deletable: true, + } + } + + #[test] + fn gc_keeps_latest_even_when_it_is_old() { + let delete = plan_snapshot_gc( + vec![ + entry("r1", 1), + entry("r2", 2), + entry("r3", 3), + entry("r4", 4), + entry("r5", 5), + entry("r6", 6), + entry("r7", 7), + ], + Some("r1"), + 100_000_000, + 5, + Duration::from_secs(0), + ); + + assert!(!delete.iter().any(|path| path.contains("r1.redb.enc"))); + } + + #[test] + fn gc_protects_recent_orphans() { + let delete = plan_snapshot_gc( + vec![entry("old", 1), entry("fresh", 99_000)], + None, + 100_000, + 0, + Duration::from_secs(2), + ); + + assert_eq!(delete, vec!["nyaterm/sync/snapshots/old.redb.enc"]); + } +} diff --git a/src-tauri/src/core/cloud_sync/manager.rs b/src-tauri/src/core/cloud_sync/manager.rs index 17702084..52d86072 100644 --- a/src-tauri/src/core/cloud_sync/manager.rs +++ b/src-tauri/src/core/cloud_sync/manager.rs @@ -13,18 +13,24 @@ use crate::config::{ }; use crate::error::{AppError, AppResult}; -use super::crypto::{decrypt_snapshot_bytes, encrypt_snapshot_bytes, require_master_password}; +use super::crypto::require_master_password; +use super::gc::{SYNC_SNAPSHOT_GC_GRACE_PERIOD, cleanup_sync_snapshots}; use super::history_log::{log_history_entry, read_cloud_sync_history_from_logs}; +use super::migration::{ + RemoteSnapshotResolution, recover_current_remote_snapshot, resolve_remote_snapshot, +}; use super::operator::{build_remote, ensure_remote_layout}; +use super::protocol::{ + commit_sync_pointer, ensure_remote_head_unchanged, pointer_from_snapshot, upload_sync_snapshot, + verify_uploaded_sync_snapshot, write_current_sync_snapshot_compat, +}; use super::remote::{ - RemoteSyncPointer, SYNC_CURRENT_FILE, SYNC_SNAPSHOTS_DIR, current_time_ms, elapsed_ms, - is_legacy_sync_snapshot_path, legacy_sync_snapshot_file, load_sync_pointer, remote_path, - write_sync_pointer, + RemoteSyncPointer, SYNC_SNAPSHOTS_DIR, current_time_ms, elapsed_ms, load_sync_pointer, + remote_path, }; use crate::core::portable_snapshot::{ PortableSnapshot, PortableSnapshotKind, apply_portable_snapshot, build_portable_snapshot, - decode_portable_snapshot, encode_portable_snapshot, }; const CLOUD_SYNC_STARTUP_CHECK_TIMEOUT: Duration = Duration::from_secs(30); @@ -326,6 +332,7 @@ impl CloudSyncManager { match action { "upload_local" => self.push_snapshot("resolve_upload", true).await, "download_remote" => self.pull_snapshot("resolve_download", true).await, + "recover_current_remote" => self.recover_current_remote(action).await, _ => Err(AppError::Config(format!( "Unsupported conflict resolution action '{}'", action @@ -426,7 +433,7 @@ impl CloudSyncManager { config::save_cloud_sync_state(&self.app()?, &state)?; } - let Some(remote) = latest else { + let Some(remote_pointer) = latest else { self.set_status( "idle", "No remote sync snapshot found".to_string(), @@ -437,13 +444,47 @@ impl CloudSyncManager { return Ok(RemoteCheckOutcome::NoRemote); }; + match trace_cloud_sync_step(trigger, "resolve_remote_snapshot", async { + resolve_remote_snapshot(&remote, &settings.remote_root, &remote_pointer).await + }) + .await? + { + RemoteSnapshotResolution::Current(_) | RemoteSnapshotResolution::LegacyMigrated(_) => {} + RemoteSnapshotResolution::Inconsistent { + pointer, + recovery_candidate, + } => { + let conflict = remote_inconsistent_preview( + &settings, + &local_hash, + &pointer, + &recovery_candidate, + ); + self.append_history(CloudSyncHistoryEntry { + id: uuid::Uuid::new_v4().to_string(), + timestamp_ms: current_time_ms(), + kind: "sync".to_string(), + status: "conflict".to_string(), + trigger: trigger.to_string(), + provider: Some(settings.provider.clone()), + revision: Some(pointer.revision_id.clone()), + duration_ms: None, + message: conflict.message.clone(), + }) + .await; + self.set_status("conflict", conflict.message.clone(), None, Some(conflict)) + .await; + return Ok(RemoteCheckOutcome::Conflict); + } + } + let state = self.state.lock().await.clone(); - match decide_remote_check(&state, &local_hash, &remote, allow_auto_pull) { + match decide_remote_check(&state, &local_hash, &remote_pointer, allow_auto_pull) { RemoteCheckDecision::UpToDate => { { let mut state = self.state.lock().await; state.last_synced_payload_hash = Some(local_hash); - state.last_applied_remote_revision = Some(remote.revision_id.clone()); + state.last_applied_remote_revision = Some(remote_pointer.revision_id.clone()); state.last_checked_at_ms = Some(current_time_ms()); config::save_cloud_sync_state(&self.app()?, &state)?; } @@ -452,7 +493,7 @@ impl CloudSyncManager { Ok(RemoteCheckOutcome::UpToDate) } RemoteCheckDecision::Conflict => { - let conflict = cloud_conflict_preview(&settings, &local_hash, &remote); + let conflict = cloud_conflict_preview(&settings, &local_hash, &remote_pointer); self.append_history(CloudSyncHistoryEntry { id: uuid::Uuid::new_v4().to_string(), timestamp_ms: current_time_ms(), @@ -460,7 +501,7 @@ impl CloudSyncManager { status: "conflict".to_string(), trigger: trigger.to_string(), provider: Some(settings.provider.clone()), - revision: Some(remote.revision_id.clone()), + revision: Some(remote_pointer.revision_id.clone()), duration_ms: None, message: conflict.message.clone(), }) @@ -725,12 +766,36 @@ impl CloudSyncManager { }) .await?; - if let Some(remote) = &latest { - if remote.payload_hash == local_hash { + if let Some(remote_pointer) = &latest { + if remote_pointer.payload_hash == local_hash { + match trace_cloud_sync_step(trigger, "resolve_remote_snapshot", async { + resolve_remote_snapshot(&remote, &settings.remote_root, remote_pointer).await + }) + .await? + { + RemoteSnapshotResolution::Current(_) + | RemoteSnapshotResolution::LegacyMigrated(_) => {} + RemoteSnapshotResolution::Inconsistent { + pointer, + recovery_candidate, + } => { + let conflict = remote_inconsistent_preview( + &settings, + &local_hash, + &pointer, + &recovery_candidate, + ); + self.set_status("conflict", conflict.message.clone(), None, Some(conflict)) + .await; + return Err(AppError::Config( + "Cloud sync remote metadata is inconsistent".to_string(), + )); + } + } { let mut state = self.state.lock().await; state.last_synced_payload_hash = Some(local_hash); - state.last_applied_remote_revision = Some(remote.revision_id.clone()); + state.last_applied_remote_revision = Some(remote_pointer.revision_id.clone()); state.last_checked_at_ms = Some(current_time_ms()); config::save_cloud_sync_state(&self.app()?, &state)?; } @@ -762,11 +827,15 @@ impl CloudSyncManager { let conflict = CloudConflictPreview { detected_at_ms: current_time_ms(), provider: settings.provider.clone(), + kind: "content_conflict".to_string(), local_payload_hash: local_hash.clone(), remote_payload_hash: remote.payload_hash.clone(), remote_revision: remote.revision_id.clone(), remote_created_at_ms: remote.created_at_ms, remote_device_id: remote.device_id.clone(), + recovery_revision: None, + recovery_payload_hash: None, + recovery_created_at_ms: None, message: "Both local and cloud state changed since last sync".to_string(), }; self.append_history(CloudSyncHistoryEntry { @@ -797,18 +866,24 @@ impl CloudSyncManager { None, ) .await; - trace_cloud_sync_step(trigger, "write_current_sync_snapshot", async { - write_current_sync_snapshot(&remote, &settings.remote_root, &envelope).await + trace_cloud_sync_step(trigger, "upload_sync_snapshot", async { + upload_sync_snapshot(&remote, &settings.remote_root, &envelope).await }) .await?; - let pointer = RemoteSyncPointer { - revision_id: envelope.revision_id.clone(), - created_at_ms: envelope.created_at_ms, - payload_hash: envelope.payload_hash.clone(), - device_id: envelope.device_id.clone(), - app_version: envelope.app_version.clone(), - }; + let pointer = pointer_from_snapshot(&envelope); + trace_cloud_sync_step(trigger, "verify_uploaded_sync_snapshot", async { + verify_uploaded_sync_snapshot(&remote, &settings.remote_root, &pointer).await + }) + .await?; + + if !force { + trace_cloud_sync_step(trigger, "recheck_sync_pointer", async { + ensure_remote_head_unchanged(&remote, &settings.remote_root, latest.as_ref()).await + }) + .await?; + } + self.set_status( "running", "Updating cloud sync pointer".to_string(), @@ -816,11 +891,23 @@ impl CloudSyncManager { None, ) .await; - trace_cloud_sync_step(trigger, "write_sync_pointer", async { - write_sync_pointer(&remote, &settings.remote_root, &pointer).await + trace_cloud_sync_step(trigger, "commit_sync_pointer", async { + commit_sync_pointer(&remote, &settings.remote_root, &pointer).await }) .await?; - schedule_cleanup_legacy_sync_snapshots(remote.clone(), settings.remote_root.clone()); + if let Err(error) = + trace_cloud_sync_step(trigger, "write_current_sync_snapshot_compat", async { + write_current_sync_snapshot_compat(&remote, &settings.remote_root, &envelope).await + }) + .await + { + tracing::warn!( + error = %error, + revision = %envelope.revision_id, + "Compatible current cloud sync snapshot write failed after commit" + ); + } + schedule_sync_snapshot_gc(remote.clone(), settings.remote_root.clone(), Some(pointer)); { let mut state = self.state.lock().await; @@ -858,6 +945,74 @@ impl CloudSyncManager { self.pull_snapshot_locked(trigger, force).await } + async fn recover_current_remote(self: &Arc, trigger: &str) -> AppResult<()> { + let _guard = self.operation_lock.lock().await; + let _ = require_master_password()?; + let settings = self.settings.lock().await.clone(); + if !settings.enabled { + return Err(AppError::Config( + "Cloud sync is disabled in settings".to_string(), + )); + } + + let started = Instant::now(); + self.set_status( + "running", + "Recovering incomplete cloud sync metadata".to_string(), + Some("sync_recover".to_string()), + None, + ) + .await; + let remote = trace_cloud_sync_step(trigger, "build_remote", async { + self.build_remote_with_recovery(settings.clone()).await + }) + .await?; + trace_cloud_sync_step(trigger, "ensure_remote_layout", async { + ensure_remote_layout(&remote, &settings.remote_root).await + }) + .await?; + let envelope = trace_cloud_sync_step(trigger, "recover_current_remote_snapshot", async { + recover_current_remote_snapshot(&remote, &settings.remote_root).await + }) + .await?; + trace_cloud_sync_step(trigger, "apply_portable_snapshot", async { + apply_portable_snapshot(&self.app()?, &envelope).await + }) + .await?; + let pointer = pointer_from_snapshot(&envelope); + schedule_sync_snapshot_gc(remote.clone(), settings.remote_root.clone(), Some(pointer)); + + { + let mut state = self.state.lock().await; + state.last_synced_payload_hash = Some(envelope.payload_hash.clone()); + state.last_applied_remote_revision = Some(envelope.revision_id.clone()); + state.last_synced_at_ms = Some(current_time_ms()); + state.last_checked_at_ms = Some(current_time_ms()); + config::save_cloud_sync_state(&self.app()?, &state)?; + } + + self.append_history(CloudSyncHistoryEntry { + id: uuid::Uuid::new_v4().to_string(), + timestamp_ms: current_time_ms(), + kind: "sync".to_string(), + status: "success".to_string(), + trigger: trigger.to_string(), + provider: Some(settings.provider.clone()), + revision: Some(envelope.revision_id.clone()), + duration_ms: Some(elapsed_ms(started.elapsed())), + message: "Cloud sync metadata recovered from current snapshot".to_string(), + }) + .await; + self.set_status( + "idle", + "Cloud sync metadata recovered".to_string(), + None, + None, + ) + .await; + Ok(()) + } + async fn pull_snapshot_locked(self: &Arc, trigger: &str, force: bool) -> AppResult<()> { let _ = require_master_password()?; let settings = self.settings.lock().await.clone(); @@ -922,6 +1077,44 @@ impl CloudSyncManager { .as_deref() .map_or(true, |revision| revision != latest.revision_id); + let remote_envelope = + match trace_cloud_sync_step(trigger, "resolve_remote_snapshot", async { + resolve_remote_snapshot(&remote, &settings.remote_root, &latest).await + }) + .await? + { + RemoteSnapshotResolution::Current(snapshot) + | RemoteSnapshotResolution::LegacyMigrated(snapshot) => snapshot, + RemoteSnapshotResolution::Inconsistent { + pointer, + recovery_candidate, + } => { + let conflict = remote_inconsistent_preview( + &settings, + &local_envelope.payload_hash, + &pointer, + &recovery_candidate, + ); + self.append_history(CloudSyncHistoryEntry { + id: uuid::Uuid::new_v4().to_string(), + timestamp_ms: current_time_ms(), + kind: "sync".to_string(), + status: "conflict".to_string(), + trigger: trigger.to_string(), + provider: Some(settings.provider.clone()), + revision: Some(pointer.revision_id.clone()), + duration_ms: Some(elapsed_ms(started.elapsed())), + message: conflict.message.clone(), + }) + .await; + self.set_status("conflict", conflict.message.clone(), None, Some(conflict)) + .await; + return Err(AppError::Config( + "Cloud sync remote metadata is inconsistent".to_string(), + )); + } + }; + if latest.payload_hash == local_envelope.payload_hash { { let mut state = self.state.lock().await; @@ -944,11 +1137,15 @@ impl CloudSyncManager { let conflict = CloudConflictPreview { detected_at_ms: current_time_ms(), provider: settings.provider.clone(), + kind: "content_conflict".to_string(), local_payload_hash: local_envelope.payload_hash.clone(), remote_payload_hash: latest.payload_hash.clone(), remote_revision: latest.revision_id.clone(), remote_created_at_ms: latest.created_at_ms, remote_device_id: latest.device_id.clone(), + recovery_revision: None, + recovery_payload_hash: None, + recovery_created_at_ms: None, message: "Both local and cloud state changed since last sync".to_string(), }; self.append_history(CloudSyncHistoryEntry { @@ -981,10 +1178,7 @@ impl CloudSyncManager { None, ) .await; - let envelope = trace_cloud_sync_step(trigger, "read_sync_snapshot", async { - read_sync_snapshot(&remote, &settings.remote_root, &latest).await - }) - .await?; + let envelope = remote_envelope; self.set_status( "running", "Applying cloud sync snapshot".to_string(), @@ -1003,11 +1197,23 @@ impl CloudSyncManager { None, ) .await; - trace_cloud_sync_step(trigger, "write_current_sync_snapshot", async { - write_current_sync_snapshot(&remote, &settings.remote_root, &envelope).await - }) - .await?; - schedule_cleanup_legacy_sync_snapshots(remote.clone(), settings.remote_root.clone()); + if let Err(error) = + trace_cloud_sync_step(trigger, "write_current_sync_snapshot_compat", async { + write_current_sync_snapshot_compat(&remote, &settings.remote_root, &envelope).await + }) + .await + { + tracing::warn!( + error = %error, + revision = %envelope.revision_id, + "Compatible current cloud sync snapshot refresh failed after pull" + ); + } + schedule_sync_snapshot_gc( + remote.clone(), + settings.remote_root.clone(), + Some(latest.clone()), + ); { let mut state = self.state.lock().await; @@ -1292,52 +1498,6 @@ where result } -async fn write_current_sync_snapshot( - remote: &super::operator::CloudRemote, - remote_root: &str, - envelope: &PortableSnapshot, -) -> AppResult<()> { - let encoded = encode_portable_snapshot(envelope)?; - let encrypted = encrypt_snapshot_bytes(&encoded)?; - remote - .write(&remote_path(remote_root, SYNC_CURRENT_FILE), encrypted) - .await -} - -async fn read_sync_snapshot( - remote: &super::operator::CloudRemote, - remote_root: &str, - pointer: &RemoteSyncPointer, -) -> AppResult { - if let Some(raw) = remote - .read_if_exists(&remote_path(remote_root, SYNC_CURRENT_FILE)) - .await? - { - let envelope = decode_remote_sync_snapshot(&raw)?; - if envelope.revision_id == pointer.revision_id { - return Ok(envelope); - } - - tracing::warn!( - current_revision = %envelope.revision_id, - pointer_revision = %pointer.revision_id, - "Cloud sync current snapshot revision differs from latest pointer; trying legacy revision path" - ); - } - - let legacy_path = remote_path( - remote_root, - &legacy_sync_snapshot_file(&pointer.revision_id), - ); - let raw = remote.read(&legacy_path).await?; - decode_remote_sync_snapshot(&raw) -} - -fn decode_remote_sync_snapshot(raw: &[u8]) -> AppResult { - let decrypted = decrypt_snapshot_bytes(raw)?; - decode_portable_snapshot(&decrypted) -} - fn decide_remote_check( state: &CloudSyncState, local_hash: &str, @@ -1374,25 +1534,52 @@ fn cloud_conflict_preview( CloudConflictPreview { detected_at_ms: current_time_ms(), provider: settings.provider.clone(), + kind: "content_conflict".to_string(), local_payload_hash: local_hash.to_string(), remote_payload_hash: remote.payload_hash.clone(), remote_revision: remote.revision_id.clone(), remote_created_at_ms: remote.created_at_ms, remote_device_id: remote.device_id.clone(), + recovery_revision: None, + recovery_payload_hash: None, + recovery_created_at_ms: None, message: "Both local and cloud state changed since last sync".to_string(), } } -fn schedule_cleanup_legacy_sync_snapshots( +fn remote_inconsistent_preview( + settings: &CloudSyncSettings, + local_hash: &str, + pointer: &RemoteSyncPointer, + recovery_candidate: &PortableSnapshot, +) -> CloudConflictPreview { + CloudConflictPreview { + detected_at_ms: current_time_ms(), + provider: settings.provider.clone(), + kind: "remote_inconsistent".to_string(), + local_payload_hash: local_hash.to_string(), + remote_payload_hash: pointer.payload_hash.clone(), + remote_revision: pointer.revision_id.clone(), + remote_created_at_ms: pointer.created_at_ms, + remote_device_id: pointer.device_id.clone(), + recovery_revision: Some(recovery_candidate.revision_id.clone()), + recovery_payload_hash: Some(recovery_candidate.payload_hash.clone()), + recovery_created_at_ms: Some(recovery_candidate.created_at_ms), + message: "Remote cloud sync metadata is incomplete. The latest pointer references a missing snapshot, but current.redb.enc contains a recoverable snapshot.".to_string(), + } +} + +fn schedule_sync_snapshot_gc( remote: super::operator::CloudRemote, remote_root: String, + latest: Option, ) { async_runtime::spawn(async move { let result = with_operation_timeout( - "cleanup_legacy_sync_snapshots", + "cleanup_sync_snapshots", CLOUD_SYNC_CLEANUP_TIMEOUT, async { - cleanup_legacy_sync_snapshots(&remote, &remote_root).await; + cleanup_sync_snapshots(&remote, &remote_root, latest.as_ref()).await; Ok(()) }, ) @@ -1401,48 +1588,13 @@ fn schedule_cleanup_legacy_sync_snapshots( if let Err(error) = result { tracing::warn!( error = %error, - "Legacy cloud sync snapshot cleanup did not complete" + grace_hours = SYNC_SNAPSHOT_GC_GRACE_PERIOD.as_secs() / 3600, + "Cloud sync snapshot cleanup did not complete" ); } }); } -async fn cleanup_legacy_sync_snapshots(remote: &super::operator::CloudRemote, remote_root: &str) { - let prefix = remote_path(remote_root, SYNC_SNAPSHOTS_DIR); - let paths = match trace_cloud_sync_step( - "cleanup_legacy_sync_snapshots", - "list_legacy_snapshots", - remote.list_files(&prefix), - ) - .await - { - Ok(paths) => paths, - Err(error) => { - tracing::warn!("Failed to list legacy cloud sync snapshots: {}", error); - return; - } - }; - - for path in paths - .into_iter() - .filter(|path| is_legacy_sync_snapshot_path(path, remote_root)) - { - if let Err(error) = trace_cloud_sync_step( - "cleanup_legacy_sync_snapshots", - "delete_legacy_snapshot", - remote.delete(&path), - ) - .await - { - tracing::warn!( - path = %path, - error = %error, - "Failed to delete legacy cloud sync snapshot" - ); - } - } -} - fn is_automatic_trigger(trigger: &str) -> bool { matches!( trigger, @@ -1451,7 +1603,10 @@ fn is_automatic_trigger(trigger: &str) -> bool { } fn is_non_retryable_automatic_error(error: &AppError) -> bool { - matches!(error, AppError::Auth(_) | AppError::Config(_)) + matches!( + error, + AppError::Auth(_) | AppError::Config(_) | AppError::Crypto(_) | AppError::CloudSync(_) + ) } fn should_record_startup_check_failure(error: &AppError) -> bool { @@ -1518,6 +1673,7 @@ mod tests { fn remote_pointer(revision_id: &str, payload_hash: &str) -> RemoteSyncPointer { RemoteSyncPointer { + schema_version: 2, revision_id: revision_id.to_string(), created_at_ms: 2, payload_hash: payload_hash.to_string(), @@ -1658,11 +1814,15 @@ mod tests { conflict: Some(CloudConflictPreview { detected_at_ms: 1, provider: "webdav".to_string(), + kind: "content_conflict".to_string(), local_payload_hash: "local".to_string(), remote_payload_hash: "remote".to_string(), remote_revision: "revision".to_string(), remote_created_at_ms: 2, remote_device_id: "device".to_string(), + recovery_revision: None, + recovery_payload_hash: None, + recovery_created_at_ms: None, message: "conflict".to_string(), }), ..CloudSyncStatus::default() diff --git a/src-tauri/src/core/cloud_sync/migration.rs b/src-tauri/src/core/cloud_sync/migration.rs new file mode 100644 index 00000000..e63957f7 --- /dev/null +++ b/src-tauri/src/core/cloud_sync/migration.rs @@ -0,0 +1,76 @@ +use crate::core::portable_snapshot::PortableSnapshot; +use crate::error::{AppError, AppResult, CloudSyncError}; + +use super::operator::CloudRemote; +use super::protocol::{ + commit_sync_pointer, pointer_from_snapshot, read_current_sync_snapshot_compat, + read_snapshot_for_pointer, upload_sync_snapshot, validate_snapshot_against_pointer, + verify_uploaded_sync_snapshot, +}; +use super::remote::RemoteSyncPointer; + +#[derive(Debug)] +pub(super) enum RemoteSnapshotResolution { + Current(PortableSnapshot), + LegacyMigrated(PortableSnapshot), + Inconsistent { + pointer: RemoteSyncPointer, + recovery_candidate: PortableSnapshot, + }, +} + +pub(super) async fn resolve_remote_snapshot( + remote: &CloudRemote, + remote_root: &str, + pointer: &RemoteSyncPointer, +) -> AppResult { + match read_snapshot_for_pointer(remote, remote_root, pointer).await { + Ok(snapshot) => return Ok(RemoteSnapshotResolution::Current(snapshot)), + Err(AppError::CloudSync(CloudSyncError::SnapshotMissing { .. })) => {} + Err(error) => return Err(error), + } + + let Some(current) = read_current_sync_snapshot_compat(remote, remote_root).await? else { + return Err(CloudSyncError::SnapshotMissing { + revision: pointer.revision_id.clone(), + } + .into()); + }; + + if validate_snapshot_against_pointer(pointer, ¤t).is_ok() { + migrate_legacy_snapshot(remote, remote_root, pointer, ¤t).await?; + return Ok(RemoteSnapshotResolution::LegacyMigrated(current)); + } + + Ok(RemoteSnapshotResolution::Inconsistent { + pointer: pointer.clone(), + recovery_candidate: current, + }) +} + +pub(super) async fn migrate_legacy_snapshot( + remote: &CloudRemote, + remote_root: &str, + pointer: &RemoteSyncPointer, + snapshot: &PortableSnapshot, +) -> AppResult<()> { + upload_sync_snapshot(remote, remote_root, snapshot).await?; + verify_uploaded_sync_snapshot(remote, remote_root, pointer).await?; + Ok(()) +} + +pub(super) async fn recover_current_remote_snapshot( + remote: &CloudRemote, + remote_root: &str, +) -> AppResult { + let Some(snapshot) = read_current_sync_snapshot_compat(remote, remote_root).await? else { + return Err(AppError::Config( + "No current cloud sync snapshot is available for recovery".to_string(), + )); + }; + let pointer = pointer_from_snapshot(&snapshot); + upload_sync_snapshot(remote, remote_root, &snapshot).await?; + verify_uploaded_sync_snapshot(remote, remote_root, &pointer).await?; + commit_sync_pointer(remote, remote_root, &pointer).await?; + Ok(snapshot) +} diff --git a/src-tauri/src/core/cloud_sync/mod.rs b/src-tauri/src/core/cloud_sync/mod.rs index 2fd5e477..2c12eed4 100644 --- a/src-tauri/src/core/cloud_sync/mod.rs +++ b/src-tauri/src/core/cloud_sync/mod.rs @@ -1,8 +1,11 @@ pub mod crypto; +mod gc; mod github_gist_auth; mod history_log; mod manager; +mod migration; mod operator; +mod protocol; mod remote; pub use github_gist_auth::{ diff --git a/src-tauri/src/core/cloud_sync/operator.rs b/src-tauri/src/core/cloud_sync/operator.rs index d835d3d8..574a864d 100644 --- a/src-tauri/src/core/cloud_sync/operator.rs +++ b/src-tauri/src/core/cloud_sync/operator.rs @@ -3,6 +3,9 @@ use std::io; use std::sync::Arc; use std::time::Duration; +#[cfg(test)] +use std::sync::Mutex as StdMutex; + use base64::Engine; use base64::engine::general_purpose::{STANDARD as BASE64_STANDARD, URL_SAFE_NO_PAD}; use http::header::{AUTHORIZATION, WWW_AUTHENTICATE}; @@ -36,6 +39,8 @@ pub(super) enum CloudRemote { OpenDal(Operator), GiteeSnippet(GiteeSnippetRemote), GithubGist(GithubGistRemote), + #[cfg(test)] + Memory(MemoryRemote), } impl CloudRemote { @@ -44,6 +49,8 @@ impl CloudRemote { Self::OpenDal(operator) => operator.create_dir(path).await.map_err(map_storage_error), Self::GiteeSnippet(_) => Ok(()), Self::GithubGist(_) => Ok(()), + #[cfg(test)] + Self::Memory(remote) => remote.create_dir(path), } } @@ -52,18 +59,8 @@ impl CloudRemote { Self::OpenDal(operator) => operator.exists(path).await.map_err(map_storage_error), Self::GiteeSnippet(remote) => remote.exists(path).await, Self::GithubGist(remote) => remote.exists(path).await, - } - } - - pub(super) async fn read(&self, path: &str) -> AppResult> { - match self { - Self::OpenDal(operator) => Ok(operator - .read(path) - .await - .map_err(map_storage_error)? - .to_vec()), - Self::GiteeSnippet(remote) => remote.read(path).await, - Self::GithubGist(remote) => remote.read(path).await, + #[cfg(test)] + Self::Memory(remote) => remote.exists(path), } } @@ -83,6 +80,8 @@ impl CloudRemote { } Self::GiteeSnippet(remote) => remote.read_if_exists(path).await, Self::GithubGist(remote) => remote.read_if_exists(path).await, + #[cfg(test)] + Self::Memory(remote) => remote.read_if_exists(path), } } @@ -97,6 +96,8 @@ impl CloudRemote { } Self::GiteeSnippet(remote) => remote.write(path, &content).await, Self::GithubGist(remote) => remote.write(path, &content).await, + #[cfg(test)] + Self::Memory(remote) => remote.write(path, content), } } @@ -105,6 +106,8 @@ impl CloudRemote { Self::OpenDal(operator) => operator.delete(path).await.map_err(map_storage_error), Self::GiteeSnippet(remote) => remote.delete(path).await, Self::GithubGist(remote) => remote.delete(path).await, + #[cfg(test)] + Self::Memory(remote) => remote.delete(path), } } @@ -125,10 +128,88 @@ impl CloudRemote { } Self::GiteeSnippet(remote) => remote.list_files(path).await, Self::GithubGist(remote) => remote.list_files(path).await, + #[cfg(test)] + Self::Memory(remote) => remote.list_files(path), } } } +#[cfg(test)] +#[derive(Clone, Default)] +pub(super) struct MemoryRemote { + files: Arc>>>, + fail_writes: Arc>>, +} + +#[cfg(test)] +impl MemoryRemote { + pub(super) fn with_files(files: HashMap>) -> Self { + Self { + files: Arc::new(StdMutex::new(files)), + fail_writes: Arc::new(StdMutex::new(Vec::new())), + } + } + + pub(super) fn fail_next_write_containing(&self, needle: &str) { + self.fail_writes + .lock() + .expect("lock fail writes") + .push(needle.to_string()); + } + + pub(super) fn file(&self, path: &str) -> Option> { + self.files.lock().expect("lock files").get(path).cloned() + } + + fn create_dir(&self, _path: &str) -> AppResult<()> { + Ok(()) + } + + fn exists(&self, path: &str) -> AppResult { + Ok(self.files.lock().expect("lock files").contains_key(path)) + } + + fn read_if_exists(&self, path: &str) -> AppResult>> { + Ok(self.files.lock().expect("lock files").get(path).cloned()) + } + + fn write(&self, path: &str, content: Vec) -> AppResult<()> { + let mut fail_writes = self.fail_writes.lock().expect("lock fail writes"); + if let Some(index) = fail_writes + .iter() + .position(|needle| path.contains(needle.as_str())) + { + fail_writes.remove(index); + return Err(AppError::Io(io::Error::new( + io::ErrorKind::Other, + format!("injected memory write failure for {path}"), + ))); + } + drop(fail_writes); + self.files + .lock() + .expect("lock files") + .insert(path.to_string(), content); + Ok(()) + } + + fn delete(&self, path: &str) -> AppResult<()> { + self.files.lock().expect("lock files").remove(path); + Ok(()) + } + + fn list_files(&self, path: &str) -> AppResult> { + Ok(self + .files + .lock() + .expect("lock files") + .keys() + .filter(|key| key.starts_with(path)) + .cloned() + .collect()) + } +} + pub(super) fn build_remote(settings: &CloudSyncSettings) -> AppResult { opendal::install_default(); match settings.provider.as_str() { @@ -478,12 +559,6 @@ impl GiteeSnippetRemote { Ok(snippet.files.contains_key(&gitee_remote_filename(path))) } - async fn read(&self, path: &str) -> AppResult> { - self.read_if_exists(path) - .await? - .ok_or_else(|| AppError::Config(format!("Gitee snippet file '{}' not found", path))) - } - async fn read_if_exists(&self, path: &str) -> AppResult>> { let filename = gitee_remote_filename(path); if let Ok(content) = self.fetch_raw_filename(&filename).await { @@ -717,12 +792,6 @@ impl GithubGistRemote { Ok(gist.files.contains_key(&github_gist_remote_filename(path))) } - async fn read(&self, path: &str) -> AppResult> { - self.read_if_exists(path) - .await? - .ok_or_else(|| AppError::Config(format!("GitHub Gist file '{}' not found", path))) - } - async fn read_if_exists(&self, path: &str) -> AppResult>> { let filename = github_gist_remote_filename(path); let gist = self.fetch_gist().await?; diff --git a/src-tauri/src/core/cloud_sync/protocol.rs b/src-tauri/src/core/cloud_sync/protocol.rs new file mode 100644 index 00000000..0c5edab1 --- /dev/null +++ b/src-tauri/src/core/cloud_sync/protocol.rs @@ -0,0 +1,391 @@ +use crate::core::portable_snapshot::{ + PortableSnapshot, decode_portable_snapshot, encode_portable_snapshot, +}; +use crate::error::{AppError, AppResult, CloudSyncError}; + +use super::crypto::{decrypt_snapshot_bytes, encrypt_snapshot_bytes}; +use super::operator::CloudRemote; +use super::remote::{ + REMOTE_SYNC_POINTER_SCHEMA_VERSION, RemoteSyncPointer, SYNC_CURRENT_FILE, + legacy_sync_snapshot_file, load_sync_pointer, remote_path, write_sync_pointer, +}; + +pub(super) fn sync_snapshot_file(revision: &str) -> String { + legacy_sync_snapshot_file(revision) +} + +pub(super) fn sync_snapshot_path(remote_root: &str, revision: &str) -> String { + remote_path(remote_root, &sync_snapshot_file(revision)) +} + +pub(super) fn pointer_from_snapshot(snapshot: &PortableSnapshot) -> RemoteSyncPointer { + RemoteSyncPointer { + schema_version: REMOTE_SYNC_POINTER_SCHEMA_VERSION, + revision_id: snapshot.revision_id.clone(), + created_at_ms: snapshot.created_at_ms, + payload_hash: snapshot.payload_hash.clone(), + device_id: snapshot.device_id.clone(), + app_version: snapshot.app_version.clone(), + } +} + +pub(super) async fn upload_sync_snapshot( + remote: &CloudRemote, + remote_root: &str, + snapshot: &PortableSnapshot, +) -> AppResult<()> { + let encoded = encode_portable_snapshot(snapshot)?; + let encrypted = encrypt_snapshot_bytes(&encoded)?; + remote + .write( + &sync_snapshot_path(remote_root, &snapshot.revision_id), + encrypted, + ) + .await +} + +pub(super) async fn verify_uploaded_sync_snapshot( + remote: &CloudRemote, + remote_root: &str, + pointer: &RemoteSyncPointer, +) -> AppResult { + read_snapshot_for_pointer(remote, remote_root, pointer).await +} + +pub(super) async fn read_snapshot_for_pointer( + remote: &CloudRemote, + remote_root: &str, + pointer: &RemoteSyncPointer, +) -> AppResult { + let path = sync_snapshot_path(remote_root, &pointer.revision_id); + let Some(raw) = remote.read_if_exists(&path).await? else { + return Err(CloudSyncError::SnapshotMissing { + revision: pointer.revision_id.clone(), + } + .into()); + }; + let snapshot = decode_remote_sync_snapshot(&raw, &pointer.revision_id)?; + validate_snapshot_against_pointer(pointer, &snapshot)?; + Ok(snapshot) +} + +pub(super) async fn write_current_sync_snapshot_compat( + remote: &CloudRemote, + remote_root: &str, + snapshot: &PortableSnapshot, +) -> AppResult<()> { + let encoded = encode_portable_snapshot(snapshot)?; + let encrypted = encrypt_snapshot_bytes(&encoded)?; + remote + .write(&remote_path(remote_root, SYNC_CURRENT_FILE), encrypted) + .await +} + +pub(super) async fn read_current_sync_snapshot_compat( + remote: &CloudRemote, + remote_root: &str, +) -> AppResult> { + let Some(raw) = remote + .read_if_exists(&remote_path(remote_root, SYNC_CURRENT_FILE)) + .await? + else { + return Ok(None); + }; + decode_remote_sync_snapshot(&raw, "current").map(Some) +} + +pub(super) async fn commit_sync_pointer( + remote: &CloudRemote, + remote_root: &str, + pointer: &RemoteSyncPointer, +) -> AppResult<()> { + write_sync_pointer(remote, remote_root, pointer).await +} + +pub(super) async fn ensure_remote_head_unchanged( + remote: &CloudRemote, + remote_root: &str, + expected: Option<&RemoteSyncPointer>, +) -> AppResult<()> { + let actual = load_sync_pointer(remote, remote_root).await?; + let expected_revision = expected.map(|pointer| pointer.revision_id.clone()); + let actual_revision = actual.as_ref().map(|pointer| pointer.revision_id.clone()); + if expected_revision != actual_revision { + return Err(CloudSyncError::ConcurrentUpdate { + expected_revision, + actual_revision, + } + .into()); + } + Ok(()) +} + +pub(super) fn validate_snapshot_against_pointer( + pointer: &RemoteSyncPointer, + snapshot: &PortableSnapshot, +) -> AppResult<()> { + if snapshot.revision_id != pointer.revision_id { + return Err(CloudSyncError::RevisionMismatch { + pointer_revision: pointer.revision_id.clone(), + snapshot_revision: snapshot.revision_id.clone(), + } + .into()); + } + + if snapshot.payload_hash != pointer.payload_hash { + return Err(CloudSyncError::HashMismatch { + expected: pointer.payload_hash.clone(), + actual: snapshot.payload_hash.clone(), + } + .into()); + } + + Ok(()) +} + +fn decode_remote_sync_snapshot(raw: &[u8], revision: &str) -> AppResult { + let decrypted = decrypt_snapshot_bytes(raw).map_err(|error| match error { + AppError::CloudSync(_) => error, + _ => CloudSyncError::CorruptedSnapshot { + revision: revision.to_string(), + } + .into(), + })?; + decode_portable_snapshot(&decrypted).map_err(|error| match error { + AppError::CloudSync(_) => error, + _ => CloudSyncError::CorruptedSnapshot { + revision: revision.to_string(), + } + .into(), + }) +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + use std::sync::Mutex; + + use crate::config::AppSettings; + use crate::core::portable_snapshot::{ + PortableAppSettings, PortableSnapshotKind, calculate_payload_hash, + }; + use crate::utils::crypto::set_master_password; + + use super::super::migration::{RemoteSnapshotResolution, resolve_remote_snapshot}; + use super::super::operator::MemoryRemote; + use super::super::remote::{load_sync_pointer, remote_path}; + use super::*; + + static MASTER_PASSWORD_TEST_LOCK: Mutex<()> = Mutex::new(()); + + fn memory_remote() -> (MemoryRemote, CloudRemote) { + let memory = MemoryRemote::with_files(HashMap::new()); + let remote = CloudRemote::Memory(memory.clone()); + (memory, remote) + } + + fn sample_snapshot(revision_id: &str, created_at_ms: u64) -> PortableSnapshot { + let settings = PortableAppSettings::from_app_settings( + &AppSettings::default(), + &PortableSnapshotKind::Sync, + ); + let mut snapshot = PortableSnapshot { + schema_version: 3, + snapshot_kind: PortableSnapshotKind::Sync, + revision_id: revision_id.to_string(), + device_id: "device".to_string(), + created_at_ms, + payload_hash: String::new(), + app_version: "test".to_string(), + settings, + sessions: Default::default(), + keys: Default::default(), + passwords: Default::default(), + credentials: Default::default(), + otp: Default::default(), + proxies: Default::default(), + proxy_groups: Default::default(), + tunnels: Default::default(), + tunnel_groups: Default::default(), + quick_commands: Default::default(), + history: Default::default(), + master_key_token: None, + known_hosts: String::new(), + notes: Default::default(), + }; + snapshot.payload_hash = calculate_payload_hash(&snapshot).expect("hash snapshot"); + snapshot + } + + async fn write_committed_snapshot( + remote: &CloudRemote, + revision_id: &str, + ) -> RemoteSyncPointer { + let snapshot = sample_snapshot(revision_id, 1); + let pointer = pointer_from_snapshot(&snapshot); + upload_sync_snapshot(remote, "nyaterm", &snapshot) + .await + .expect("upload snapshot"); + commit_sync_pointer(remote, "nyaterm", &pointer) + .await + .expect("commit pointer"); + pointer + } + + #[tokio::test(flavor = "current_thread")] + async fn snapshot_upload_failure_leaves_latest_unchanged() { + let _guard = MASTER_PASSWORD_TEST_LOCK.lock().expect("lock password"); + set_master_password(Some("secret".to_string())); + let (memory, remote) = memory_remote(); + let old_pointer = write_committed_snapshot(&remote, "r1").await; + memory.fail_next_write_containing("snapshots/r2"); + + let new_snapshot = sample_snapshot("r2", 2); + let result = upload_sync_snapshot(&remote, "nyaterm", &new_snapshot).await; + + assert!(result.is_err()); + let latest = load_sync_pointer(&remote, "nyaterm") + .await + .expect("load latest") + .expect("latest"); + assert_eq!(latest.revision_id, old_pointer.revision_id); + set_master_password(None); + } + + #[tokio::test(flavor = "current_thread")] + async fn pointer_write_failure_keeps_old_revision_readable() { + let _guard = MASTER_PASSWORD_TEST_LOCK.lock().expect("lock password"); + set_master_password(Some("secret".to_string())); + let (memory, remote) = memory_remote(); + let old_pointer = write_committed_snapshot(&remote, "r1").await; + let new_snapshot = sample_snapshot("r2", 2); + let new_pointer = pointer_from_snapshot(&new_snapshot); + + upload_sync_snapshot(&remote, "nyaterm", &new_snapshot) + .await + .expect("upload new snapshot"); + memory.fail_next_write_containing("latest.redb"); + let result = commit_sync_pointer(&remote, "nyaterm", &new_pointer).await; + + assert!(result.is_err()); + let old_snapshot = read_snapshot_for_pointer(&remote, "nyaterm", &old_pointer) + .await + .expect("old snapshot readable"); + assert_eq!(old_snapshot.revision_id, "r1"); + set_master_password(None); + } + + #[tokio::test(flavor = "current_thread")] + async fn current_pointer_mismatch_returns_inconsistent_resolution() { + let _guard = MASTER_PASSWORD_TEST_LOCK.lock().expect("lock password"); + set_master_password(Some("secret".to_string())); + let (_memory, remote) = memory_remote(); + let pointer = pointer_from_snapshot(&sample_snapshot("r1", 1)); + commit_sync_pointer(&remote, "nyaterm", &pointer) + .await + .expect("commit pointer"); + write_current_sync_snapshot_compat(&remote, "nyaterm", &sample_snapshot("r2", 2)) + .await + .expect("write current"); + + let resolution = resolve_remote_snapshot(&remote, "nyaterm", &pointer) + .await + .expect("resolve remote"); + + match resolution { + RemoteSnapshotResolution::Inconsistent { + pointer, + recovery_candidate, + } => { + assert_eq!(pointer.revision_id, "r1"); + assert_eq!(recovery_candidate.revision_id, "r2"); + } + _ => panic!("expected inconsistent remote"), + } + set_master_password(None); + } + + #[tokio::test(flavor = "current_thread")] + async fn new_protocol_reads_latest_snapshot() { + let _guard = MASTER_PASSWORD_TEST_LOCK.lock().expect("lock password"); + set_master_password(Some("secret".to_string())); + let (_memory, remote) = memory_remote(); + let pointer = write_committed_snapshot(&remote, "r2").await; + + let snapshot = read_snapshot_for_pointer(&remote, "nyaterm", &pointer) + .await + .expect("read snapshot"); + + assert_eq!(snapshot.revision_id, "r2"); + set_master_password(None); + } + + #[tokio::test(flavor = "current_thread")] + async fn legacy_current_is_migrated_to_snapshot_path() { + let _guard = MASTER_PASSWORD_TEST_LOCK.lock().expect("lock password"); + set_master_password(Some("secret".to_string())); + let (memory, remote) = memory_remote(); + let snapshot = sample_snapshot("r1", 1); + let pointer = pointer_from_snapshot(&snapshot); + commit_sync_pointer(&remote, "nyaterm", &pointer) + .await + .expect("commit pointer"); + write_current_sync_snapshot_compat(&remote, "nyaterm", &snapshot) + .await + .expect("write current"); + + let resolution = resolve_remote_snapshot(&remote, "nyaterm", &pointer) + .await + .expect("resolve legacy"); + + assert!(matches!( + resolution, + RemoteSnapshotResolution::LegacyMigrated(_) + )); + assert!( + memory + .file(&remote_path("nyaterm", &sync_snapshot_file("r1"))) + .is_some() + ); + set_master_password(None); + } + + #[tokio::test(flavor = "current_thread")] + async fn concurrent_update_is_detected_before_pointer_commit() { + let _guard = MASTER_PASSWORD_TEST_LOCK.lock().expect("lock password"); + set_master_password(Some("secret".to_string())); + let (_memory, remote) = memory_remote(); + let base_pointer = write_committed_snapshot(&remote, "r1").await; + let next_pointer = pointer_from_snapshot(&sample_snapshot("r2", 2)); + commit_sync_pointer(&remote, "nyaterm", &next_pointer) + .await + .expect("commit competing pointer"); + + let result = ensure_remote_head_unchanged(&remote, "nyaterm", Some(&base_pointer)).await; + + assert!(matches!( + result, + Err(AppError::CloudSync(CloudSyncError::ConcurrentUpdate { .. })) + )); + let latest = load_sync_pointer(&remote, "nyaterm") + .await + .expect("load latest") + .expect("latest"); + assert_eq!(latest.revision_id, "r2"); + set_master_password(None); + } + + #[test] + fn pointer_snapshot_hash_mismatch_is_rejected() { + let snapshot = sample_snapshot("r1", 1); + let mut pointer = pointer_from_snapshot(&snapshot); + pointer.payload_hash = "wrong".to_string(); + + let result = validate_snapshot_against_pointer(&pointer, &snapshot); + + assert!(matches!( + result, + Err(AppError::CloudSync(CloudSyncError::HashMismatch { .. })) + )); + } +} diff --git a/src-tauri/src/core/cloud_sync/remote.rs b/src-tauri/src/core/cloud_sync/remote.rs index 472e9e68..46d5aec4 100644 --- a/src-tauri/src/core/cloud_sync/remote.rs +++ b/src-tauri/src/core/cloud_sync/remote.rs @@ -13,6 +13,7 @@ use super::operator::CloudRemote; pub(super) const SYNC_CURRENT_FILE: &str = "sync/current.redb.enc"; pub(super) const SYNC_LATEST_FILE: &str = "sync/latest.redb"; pub(super) const SYNC_SNAPSHOTS_DIR: &str = "sync/snapshots/"; +pub(super) const REMOTE_SYNC_POINTER_SCHEMA_VERSION: u32 = 2; const REMOTE_SYNC_POINTER_TABLE: TableDefinition<&str, &str> = TableDefinition::new("sync_pointer"); @@ -20,6 +21,8 @@ const REMOTE_SYNC_POINTER_KEY: &str = "latest"; #[derive(Debug, Clone, Serialize, Deserialize)] pub(super) struct RemoteSyncPointer { + #[serde(default = "default_remote_sync_pointer_schema_version")] + pub schema_version: u32, pub revision_id: String, pub created_at_ms: u64, pub payload_hash: String, @@ -27,6 +30,10 @@ pub(super) struct RemoteSyncPointer { pub app_version: String, } +fn default_remote_sync_pointer_schema_version() -> u32 { + 1 +} + pub(super) fn remote_path(base_root: &str, child: &str) -> String { let root = base_root.trim().trim_matches('/'); let child = child.trim().trim_start_matches('/'); @@ -211,6 +218,7 @@ mod tests { #[test] fn remote_redb_metadata_roundtrips() { let pointer = RemoteSyncPointer { + schema_version: REMOTE_SYNC_POINTER_SCHEMA_VERSION, revision_id: "rev".to_string(), created_at_ms: 1, payload_hash: "hash".to_string(), @@ -226,6 +234,7 @@ mod tests { assert_eq!(decoded.revision_id, pointer.revision_id); assert_eq!(decoded.payload_hash, pointer.payload_hash); + assert_eq!(decoded.schema_version, REMOTE_SYNC_POINTER_SCHEMA_VERSION); } #[test] diff --git a/src-tauri/src/core/portable_snapshot/hash.rs b/src-tauri/src/core/portable_snapshot/hash.rs index 2537f421..481f932b 100644 --- a/src-tauri/src/core/portable_snapshot/hash.rs +++ b/src-tauri/src/core/portable_snapshot/hash.rs @@ -14,7 +14,7 @@ fn validate_portable_snapshot(snapshot: &PortableSnapshot) -> AppResult<()> { Ok(()) } -fn calculate_payload_hash(snapshot: &PortableSnapshot) -> AppResult { +pub(crate) fn calculate_payload_hash(snapshot: &PortableSnapshot) -> AppResult { let payload_bytes = serde_json::to_vec(&SnapshotHashInput { settings: &snapshot.settings, sessions: &snapshot.sessions, diff --git a/src-tauri/src/error.rs b/src-tauri/src/error.rs index bbdb0711..08fdb603 100644 --- a/src-tauri/src/error.rs +++ b/src-tauri/src/error.rs @@ -46,10 +46,43 @@ pub enum AppError { #[error("Crypto error: {0}")] Crypto(String), + #[error("{0}")] + CloudSync(#[from] CloudSyncError), + #[error("Translation error: {0}")] Translation(String), } +#[derive(Debug, thiserror::Error, Clone, PartialEq, Eq)] +pub enum CloudSyncError { + #[error( + "Remote sync metadata is inconsistent: latest points to {revision} but the referenced snapshot is missing." + )] + SnapshotMissing { revision: String }, + + #[error( + "Remote sync snapshot revision mismatch: latest points to {pointer_revision} but snapshot contains {snapshot_revision}." + )] + RevisionMismatch { + pointer_revision: String, + snapshot_revision: String, + }, + + #[error("Remote sync snapshot hash mismatch: expected {expected} but got {actual}.")] + HashMismatch { expected: String, actual: String }, + + #[error( + "Remote sync was updated by another device: expected {expected_revision:?} but found {actual_revision:?}." + )] + ConcurrentUpdate { + expected_revision: Option, + actual_revision: Option, + }, + + #[error("Remote sync snapshot {revision} is corrupted.")] + CorruptedSnapshot { revision: String }, +} + impl Serialize for AppError { fn serialize(&self, serializer: S) -> Result where diff --git a/src/components/panel/SyncBackupHistoryPanel.tsx b/src/components/panel/SyncBackupHistoryPanel.tsx index b5ab05ee..821ae30e 100644 --- a/src/components/panel/SyncBackupHistoryPanel.tsx +++ b/src/components/panel/SyncBackupHistoryPanel.tsx @@ -21,6 +21,7 @@ import { formatCloudProvider, formatDuration, formatTimestamp, + isRemoteInconsistentConflict, shortValue, } from "@/lib/cloudSync"; import { getErrorMessage } from "@/lib/errors"; @@ -216,7 +217,7 @@ function SyncBackupHistoryPanel() { }, []); const handleResolveConflict = useCallback( - async (action: "download_remote" | "upload_local") => { + async (action: "download_remote" | "upload_local" | "recover_current_remote") => { setRunningAction(action); try { await invoke("resolve_cloud_sync_conflict", { action }); @@ -224,7 +225,9 @@ function SyncBackupHistoryPanel() { toast.success( action === "download_remote" ? t("settings.syncResolveDownloadSuccess") - : t("settings.syncResolveUploadSuccess"), + : action === "recover_current_remote" + ? t("settings.syncRecoverCurrentSuccess") + : t("settings.syncResolveUploadSuccess"), ); } catch (error) { toast.error(getErrorMessage(error)); @@ -258,6 +261,7 @@ function SyncBackupHistoryPanel() { const syncActionDisabled = loading || runningAction !== null || status.state === "running"; const canRunSyncAction = status.enabled && !syncActionDisabled; + const isRemoteInconsistent = isRemoteInconsistentConflict(status.conflict); const kindLabels = useMemo( () => ({ @@ -363,7 +367,9 @@ function SyncBackupHistoryPanel() {
- {t("settings.syncConflictTitle")} + {isRemoteInconsistent + ? t("settings.syncRemoteIncompleteTitle") + : t("settings.syncConflictTitle")}
@@ -384,18 +390,36 @@ function SyncBackupHistoryPanel() { label={t("settings.payloadHashLabel")} value={shortValue(status.conflict.remote_payload_hash, 10)} /> + {isRemoteInconsistent ? ( + + ) : null}
- + {isRemoteInconsistent ? ( + + ) : ( + + )}
+ {isRemoteInconsistent ? ( +
+
+ {t("settings.currentRemoteSnapshot")} +
+
+ {shortValue(status.conflict.recovery_revision, 10)} +
+
+ {formatTimestamp(status.conflict.recovery_created_at_ms)} +
+
+ ) : null}
- + {isRemoteInconsistent ? ( + + ) : ( + + )}