starred/nyaterm
1
0
Fork 0
mirror of https://github.com/nyakang/nyaterm.git synced 2026-09-22 08:01:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(cloudSync): enhance conflict resolution and snapshot management (#397, #405)

Browse Source 
- 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.
This commit is contained in:
Kang
2026-08-11 12:05:00 +08:00
parent 21d0f04ed5
commit 4dda6400ce
14 changed files with 1222 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/config/cloud_sync.rs
+12
View File
@@ -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<String>,
#[serde(default)]
pub recovery_payload_hash: Option<String>,
#[serde(default)]
pub recovery_created_at_ms: Option<u64>,
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<CloudSyncSettings> {
let _ = app;
storage::load_settings_doc(SettingsDocKey::CloudSyncSettings)
src-tauri/src/core/cloud_sync/gc.rs
+220
View File
@@ -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<Vec<SnapshotGcEntry>> {
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<SnapshotGcEntry>,
latest_revision: Option<&str>,
now_ms: u64,
keep_recent: usize,
grace_period: Duration,
) -> Vec<String> {
let mut protected: HashSet<String> = 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<String> {
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"]);
}
}
src-tauri/src/core/cloud_sync/manager.rs
+281 -121
View File
@@ -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<Self>, 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<Self>, 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<PortableSnapshot> {
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<PortableSnapshot> {
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<RemoteSyncPointer>,
) {
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()
src-tauri/src/core/cloud_sync/migration.rs
+76
View File
@@ -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<RemoteSnapshotResolution> {
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, &current).is_ok() {
migrate_legacy_snapshot(remote, remote_root, pointer, &current).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<PortableSnapshot> {
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)
}
src-tauri/src/core/cloud_sync/mod.rs
+3
View File
@@ -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::{
src-tauri/src/core/cloud_sync/operator.rs
+93 -24
View File
@@ -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<Vec<u8>> {
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<StdMutex<HashMap<String, Vec<u8>>>>,
fail_writes: Arc<StdMutex<Vec<String>>>,
}
#[cfg(test)]
impl MemoryRemote {
pub(super) fn with_files(files: HashMap<String, Vec<u8>>) -> 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<Vec<u8>> {
self.files.lock().expect("lock files").get(path).cloned()
}
fn create_dir(&self, _path: &str) -> AppResult<()> {
Ok(())
}
fn exists(&self, path: &str) -> AppResult<bool> {
Ok(self.files.lock().expect("lock files").contains_key(path))
}
fn read_if_exists(&self, path: &str) -> AppResult<Option<Vec<u8>>> {
Ok(self.files.lock().expect("lock files").get(path).cloned())
}
fn write(&self, path: &str, content: Vec<u8>) -> 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<Vec<String>> {
Ok(self
.files
.lock()
.expect("lock files")
.keys()
.filter(|key| key.starts_with(path))
.cloned()
.collect())
}
}
pub(super) fn build_remote(settings: &CloudSyncSettings) -> AppResult<CloudRemote> {
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<Vec<u8>> {
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<Option<Vec<u8>>> {
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<Vec<u8>> {
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<Option<Vec<u8>>> {
let filename = github_gist_remote_filename(path);
let gist = self.fetch_gist().await?;
src-tauri/src/core/cloud_sync/protocol.rs
+391
View File
@@ -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<PortableSnapshot> {
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<PortableSnapshot> {
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<Option<PortableSnapshot>> {
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<PortableSnapshot> {
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 { .. }))
));
}
}
src-tauri/src/core/cloud_sync/remote.rs
+9
View File
@@ -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]
src-tauri/src/core/portable_snapshot/hash.rs
+1 -1
View File
@@ -14,7 +14,7 @@ fn validate_portable_snapshot(snapshot: &PortableSnapshot) -> AppResult<()> {
Ok(())
}
fn calculate_payload_hash(snapshot: &PortableSnapshot) -> AppResult<String> {
pub(crate) fn calculate_payload_hash(snapshot: &PortableSnapshot) -> AppResult<String> {
let payload_bytes = serde_json::to_vec(&SnapshotHashInput {
settings: &snapshot.settings,
sessions: &snapshot.sessions,
src-tauri/src/error.rs
+33
View File
@@ -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<String>,
actual_revision: Option<String>,
},
#[error("Remote sync snapshot {revision} is corrupted.")]
CorruptedSnapshot { revision: String },
}
impl Serialize for AppError {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
src/components/panel/SyncBackupHistoryPanel.tsx
+36 -12
View File
@@ -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() {
<div className="flex items-center gap-2 border-b border-amber-500/20 px-3 py-2.5">
<MdWarning className="shrink-0 text-base text-amber-500" />
<span className="flex-1 text-sm font-medium text-amber-500">
{t("settings.syncConflictTitle")}
{isRemoteInconsistent
? t("settings.syncRemoteIncompleteTitle")
: t("settings.syncConflictTitle")}
</span>
</div>
@@ -384,18 +390,36 @@ function SyncBackupHistoryPanel() {
label={t("settings.payloadHashLabel")}
value={shortValue(status.conflict.remote_payload_hash, 10)}
/>
{isRemoteInconsistent ? (
<StatRow
label={t("settings.currentRemoteSnapshot")}
value={shortValue(status.conflict.recovery_revision, 10)}
/>
) : null}
</div>
<div className="flex gap-2 px-3 pb-3">
<Button
variant="outline"
size="sm"
className="flex-1 text-xs"
onClick={() => void handleResolveConflict("download_remote")}
disabled={runningAction !== null}
>
{t("settings.downloadRemoteVersion")}
</Button>
{isRemoteInconsistent ? (
<Button
variant="outline"
size="sm"
className="flex-1 text-xs"
onClick={() => void handleResolveConflict("recover_current_remote")}
disabled={runningAction !== null}
>
{t("settings.useCurrentRemoteSnapshot")}
</Button>
) : (
<Button
variant="outline"
size="sm"
className="flex-1 text-xs"
onClick={() => void handleResolveConflict("download_remote")}
disabled={runningAction !== null}
>
{t("settings.downloadRemoteVersion")}
</Button>
)}
<Button
size="sm"
className="flex-1 text-xs"
src/components/settings/SyncBackupTab.tsx
+59 -19
View File
@@ -15,6 +15,7 @@ import {
formatCloudProvider,
formatTimestamp,
getCloudSyncValidationErrors,
isRemoteInconsistentConflict,
secretInputValue,
secretPlaceholder,
shortValue,
@@ -120,6 +121,7 @@ export function SyncBackupTab({ onNavigateSecurity }: SyncBackupTabProps) {
const canRunConfigDependentActions = canUseCommittedProvider && !isDirty && !isSaving;
const canRunEnabledActions = canRunConfigDependentActions && committedCloudSync.enabled;
const isBusy = loading || isSaving || runningAction !== null;
const isRemoteInconsistent = isRemoteInconsistentConflict(status.conflict);
const updateCloudSync = useCallback(
(patch: Partial<CloudSyncSettings>) => {
@@ -1077,7 +1079,11 @@ export function SyncBackupTab({ onNavigateSecurity }: SyncBackupTabProps) {
>
{status.conflict ? (
<div className="rounded-lg border border-amber-500/30 bg-amber-500/10 px-4 py-4">
<div className="text-sm font-semibold">{t("settings.syncConflictTitle")}</div>
<div className="text-sm font-semibold">
{isRemoteInconsistent
? t("settings.syncRemoteIncompleteTitle")
: t("settings.syncConflictTitle")}
</div>
<p className="mt-1 text-xs leading-5 text-muted-foreground">
<span className="break-words [overflow-wrap:anywhere]">
{status.conflict.message}
@@ -1103,26 +1109,60 @@ export function SyncBackupTab({ onNavigateSecurity }: SyncBackupTabProps) {
{formatTimestamp(status.conflict.remote_created_at_ms)}
</div>
</div>
{isRemoteInconsistent ? (
<div className="rounded-md border border-border/70 bg-background/70 px-3 py-3 md:col-span-2">
<div className="text-[0.6875rem] uppercase tracking-[0.14em] text-muted-foreground">
{t("settings.currentRemoteSnapshot")}
</div>
<div className="mt-2 text-xs font-medium">
{shortValue(status.conflict.recovery_revision, 10)}
</div>
<div className="mt-1 text-[0.6875rem] text-muted-foreground">
{formatTimestamp(status.conflict.recovery_created_at_ms)}
</div>
</div>
) : null}
</div>
<div className="mt-3 flex flex-wrap gap-2">
<Button
variant="outline"
size="sm"
onClick={() =>
void runAction(
"resolve-download",
t("settings.syncResolveDownloadSuccess"),
() =>
invoke("resolve_cloud_sync_conflict", {
action: "download_remote",
}),
{ allowWhenDisabled: true },
)
}
disabled={isBusy || !canRunConfigDependentActions}
>
{t("settings.downloadRemoteVersion")}
</Button>
{isRemoteInconsistent ? (
<Button
variant="outline"
size="sm"
onClick={() =>
void runAction(
"resolve-recover-current",
t("settings.syncRecoverCurrentSuccess"),
() =>
invoke("resolve_cloud_sync_conflict", {
action: "recover_current_remote",
}),
{ allowWhenDisabled: true },
)
}
disabled={isBusy || !canRunConfigDependentActions}
>
{t("settings.useCurrentRemoteSnapshot")}
</Button>
) : (
<Button
variant="outline"
size="sm"
onClick={() =>
void runAction(
"resolve-download",
t("settings.syncResolveDownloadSuccess"),
() =>
invoke("resolve_cloud_sync_conflict", {
action: "download_remote",
}),
{ allowWhenDisabled: true },
)
}
disabled={isBusy || !canRunConfigDependentActions}
>
{t("settings.downloadRemoteVersion")}
</Button>
)}
<Button
size="sm"
onClick={() =>
src/lib/cloudSync.ts
+4
View File
@@ -233,3 +233,7 @@ export function shortValue(value?: string | null, size = 8) {
export function hasConflict(conflict?: CloudConflictPreview | null) {
return Boolean(conflict?.remote_revision);
}
export function isRemoteInconsistentConflict(conflict?: CloudConflictPreview | null) {
return conflict?.kind === "remote_inconsistent";
}
src/types/global.d.ts
Vendored
+4
View File
@@ -1627,11 +1627,15 @@ export interface GithubGistDeviceFlowPoll {
export interface CloudConflictPreview {
detected_at_ms: number;
provider: string;
kind?: "content_conflict" | "remote_inconsistent";
local_payload_hash: string;
remote_payload_hash: string;
remote_revision: string;
remote_created_at_ms: number;
remote_device_id: string;
recovery_revision?: string | null;
recovery_payload_hash?: string | null;
recovery_created_at_ms?: number | null;
message: string;
}