okooooooo/nyaterm
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(sync): implement gist capacity handling and recovery actions

Browse Source 
- Added logic to detect gist capacity sync failures and introduced a recovery button in the SyncBackupTab for users to clean remote snapshots and retry the push.
- Updated internationalization files to include new strings for the recovery action and hints regarding gist capacity limits.
- Enhanced cloud sync management to prune snapshots before uploads to avoid exceeding gist file limits, improving overall sync reliability.
- Implemented tests to ensure proper functionality of the new gist handling features.
This commit is contained in:
jeeinn committed 2026-09-29 23:53:41 +08:00
1 parent 1ef75d1bbb
commit a096009812
11 files changed
+429 -36
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src-tauri/src/core/cloud_sync/gc.rs
+105 -1
View File
@@ -12,6 +12,8 @@ use super::remote::{
pub(super) const SYNC_SNAPSHOT_KEEP_RECENT: usize = 5;
pub(super) const SYNC_SNAPSHOT_GC_GRACE_PERIOD: Duration = Duration::from_secs(24 * 60 * 60);
/// Gitee gist hard limit is 10 files; keep headroom for the next upload.
pub(super) const GIST_REMOTE_MAX_FILES: usize = 10;
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) struct SnapshotGcEntry {
@@ -21,10 +23,15 @@ pub(super) struct SnapshotGcEntry {
pub deletable: bool,
}
pub(super) fn is_gist_provider(provider: &str) -> bool {
matches!(provider, "gitee_snippet" | "github_gist")
}
pub(super) async fn cleanup_sync_snapshots(
remote: &CloudRemote,
remote_root: &str,
latest: Option<&RemoteSyncPointer>,
grace_period: Duration,
) {
let result = collect_snapshots(remote, remote_root)
.await
@@ -34,7 +41,7 @@ pub(super) async fn cleanup_sync_snapshots(
latest.map(|pointer| pointer.revision_id.as_str()),
current_time_ms(),
SYNC_SNAPSHOT_KEEP_RECENT,
SYNC_SNAPSHOT_GC_GRACE_PERIOD,
grace_period,
)
});
@@ -46,6 +53,76 @@ pub(super) async fn cleanup_sync_snapshots(
}
};
delete_snapshot_paths(remote, paths).await;
}
/// Free gist capacity before uploading a new snapshot file.
///
/// Keeps at most `keep_recent - 1` snapshots so the upcoming upload stays within
/// `keep_recent`, and deletes additional oldest snapshots when total gist files
/// would exceed `GIST_REMOTE_MAX_FILES - 1`.
pub(super) async fn prune_gist_snapshots_before_upload(
remote: &CloudRemote,
remote_root: &str,
latest: Option<&RemoteSyncPointer>,
) -> AppResult<()> {
let total_files = remote.list_files("").await?.len();
let snapshots = collect_snapshots(remote, remote_root).await?;
let keep_before_upload = SYNC_SNAPSHOT_KEEP_RECENT.saturating_sub(1);
let mut paths = plan_snapshot_gc(
snapshots.clone(),
latest.map(|pointer| pointer.revision_id.as_str()),
current_time_ms(),
keep_before_upload,
Duration::from_secs(0),
);
let projected_total = total_files.saturating_sub(paths.len()).saturating_add(1);
if projected_total > GIST_REMOTE_MAX_FILES {
let extra_needed = projected_total - GIST_REMOTE_MAX_FILES;
let mut protected: HashSet<String> = HashSet::new();
if let Some(latest) = latest {
protected.insert(latest.revision_id.clone());
}
let mut candidates = snapshots;
candidates.sort_by_key(|snapshot| snapshot.created_at_ms);
let mut extras = Vec::new();
for snapshot in candidates {
if extras.len() >= extra_needed {
break;
}
if !snapshot.deletable {
// Capacity prune: drop unreadable orphans too when over the hard limit.
if snapshot.revision_id.is_empty() || !protected.contains(&snapshot.revision_id) {
if !paths.contains(&snapshot.path) {
extras.push(snapshot.path);
}
}
continue;
}
if protected.contains(&snapshot.revision_id) {
continue;
}
if paths.contains(&snapshot.path) {
continue;
}
extras.push(snapshot.path);
}
paths.extend(extras);
}
if !paths.is_empty() {
tracing::info!(
delete_count = paths.len(),
total_files,
"Pruning gist snapshots before upload"
);
}
delete_snapshot_paths(remote, paths).await;
Ok(())
}
async fn delete_snapshot_paths(remote: &CloudRemote, paths: Vec<String>) {
for path in paths {
if let Err(error) = remote.delete(&path).await {
tracing::warn!(
@@ -217,4 +294,31 @@ mod tests {
assert_eq!(delete, vec!["nyaterm/sync/snapshots/old.redb.enc"]);
}
#[test]
fn gist_pre_upload_keep_leaves_one_slot() {
let delete = plan_snapshot_gc(
vec![
entry("r1", 1),
entry("r2", 2),
entry("r3", 3),
entry("r4", 4),
entry("r5", 5),
],
Some("r5"),
100_000,
SYNC_SNAPSHOT_KEEP_RECENT.saturating_sub(1),
Duration::from_secs(0),
);
assert_eq!(delete, vec!["nyaterm/sync/snapshots/r1.redb.enc"]);
}
#[test]
fn is_gist_provider_detects_snippet_backends() {
assert!(is_gist_provider("gitee_snippet"));
assert!(is_gist_provider("github_gist"));
assert!(!is_gist_provider("webdav"));
assert!(!is_gist_provider("s3"));
}
}
src-tauri/src/core/cloud_sync/manager.rs
+129 -21
View File
@@ -11,10 +11,13 @@ use crate::config::{
self, CloudConflictPreview, CloudSyncHistoryEntry, CloudSyncSettings, CloudSyncState,
CloudSyncStatus,
};
use crate::error::{AppError, AppResult};
use crate::error::{AppError, AppResult, CloudSyncError};
use super::crypto::require_master_password;
use super::gc::{SYNC_SNAPSHOT_GC_GRACE_PERIOD, cleanup_sync_snapshots};
use super::gc::{
SYNC_SNAPSHOT_GC_GRACE_PERIOD, cleanup_sync_snapshots, is_gist_provider,
prune_gist_snapshots_before_upload,
};
use super::history_log::{log_history_entry, read_cloud_sync_history_from_logs};
use super::migration::{
RemoteSnapshotResolution, recover_current_remote_snapshot, resolve_remote_snapshot,
@@ -957,16 +960,80 @@ impl CloudSyncManager {
None,
)
.await;
trace_cloud_sync_step(trigger, "upload_sync_snapshot", async {
upload_sync_snapshot(&remote, &settings.remote_root, &envelope).await
})
.await?;
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?;
let gist_backend = is_gist_provider(&settings.provider);
if gist_backend {
trace_cloud_sync_step(trigger, "prune_gist_snapshots_before_upload", async {
prune_gist_snapshots_before_upload(&remote, &settings.remote_root, latest.as_ref())
.await
})
.await?;
}
let mut upload_attempts = 0;
loop {
upload_attempts += 1;
let upload_result = trace_cloud_sync_step(trigger, "upload_sync_snapshot", async {
upload_sync_snapshot(&remote, &settings.remote_root, &envelope).await
})
.await;
match upload_result {
Ok(()) => {}
Err(error)
if gist_backend
&& upload_attempts < 2
&& matches!(
&error,
AppError::CloudSync(CloudSyncError::RemoteFileRejected { .. })
) =>
{
tracing::warn!(
error = %error,
attempt = upload_attempts,
"Gist upload rejected; pruning and retrying once"
);
prune_gist_snapshots_before_upload(
&remote,
&settings.remote_root,
latest.as_ref(),
)
.await?;
continue;
}
Err(error) => return Err(error),
}
match trace_cloud_sync_step(trigger, "verify_uploaded_sync_snapshot", async {
verify_uploaded_sync_snapshot(&remote, &settings.remote_root, &pointer).await
})
.await
{
Ok(_) => break,
Err(error)
if gist_backend
&& upload_attempts < 2
&& matches!(
&error,
AppError::CloudSync(CloudSyncError::SnapshotNotAccepted { .. })
| AppError::CloudSync(CloudSyncError::RemoteFileRejected { .. })
) =>
{
tracing::warn!(
error = %error,
attempt = upload_attempts,
"Gist upload verify failed; pruning and retrying once"
);
prune_gist_snapshots_before_upload(
&remote,
&settings.remote_root,
latest.as_ref(),
)
.await?;
}
Err(error) => return Err(error),
}
}
if !force {
trace_cloud_sync_step(trigger, "recheck_sync_pointer", async {
@@ -998,8 +1065,13 @@ impl CloudSyncManager {
"Compatible current cloud sync snapshot write failed after commit"
);
}
self.schedule_sync_snapshot_gc(remote.clone(), settings.remote_root.clone(), Some(pointer))
.await?;
self.schedule_sync_snapshot_gc(
remote.clone(),
settings.remote_root.clone(),
Some(pointer),
&settings.provider,
)
.await?;
{
let mut state = self.state.lock().await;
@@ -1074,8 +1146,13 @@ impl CloudSyncManager {
})
.await?;
let pointer = pointer_from_snapshot(&envelope);
self.schedule_sync_snapshot_gc(remote.clone(), settings.remote_root.clone(), Some(pointer))
.await?;
self.schedule_sync_snapshot_gc(
remote.clone(),
settings.remote_root.clone(),
Some(pointer),
&settings.provider,
)
.await?;
{
let mut state = self.state.lock().await;
@@ -1323,6 +1400,7 @@ impl CloudSyncManager {
remote.clone(),
settings.remote_root.clone(),
Some(pointer.clone()),
&settings.provider,
)
.await?;
@@ -1364,11 +1442,20 @@ impl CloudSyncManager {
remote: super::operator::CloudRemote,
remote_root: String,
latest: Option<RemoteSyncPointer>,
provider: &str,
) -> AppResult<()> {
let gist_backend = is_gist_provider(provider);
let grace_period = if gist_backend {
Duration::from_secs(0)
} else {
SYNC_SNAPSHOT_GC_GRACE_PERIOD
};
let now = current_time_ms();
{
let mut state = self.state.lock().await;
if !maintenance_due(state.last_gc_attempt_at_ms, now, CLOUD_SYNC_GC_INTERVAL_MS) {
if !gist_backend
&& !maintenance_due(state.last_gc_attempt_at_ms, now, CLOUD_SYNC_GC_INTERVAL_MS)
{
tracing::info!("Cloud sync snapshot cleanup skipped by daily throttle");
return Ok(());
}
@@ -1381,7 +1468,8 @@ impl CloudSyncManager {
"cleanup_sync_snapshots",
CLOUD_SYNC_CLEANUP_TIMEOUT,
async {
cleanup_sync_snapshots(&remote, &remote_root, latest.as_ref()).await;
cleanup_sync_snapshots(&remote, &remote_root, latest.as_ref(), grace_period)
.await;
Ok(())
},
)
@@ -1390,7 +1478,7 @@ impl CloudSyncManager {
if let Err(error) = result {
tracing::warn!(
error = %error,
grace_hours = SYNC_SNAPSHOT_GC_GRACE_PERIOD.as_secs() / 3600,
grace_hours = grace_period.as_secs() / 3600,
"Cloud sync snapshot cleanup did not complete"
);
}
@@ -1802,10 +1890,15 @@ fn is_automatic_trigger(trigger: &str) -> bool {
}
fn is_non_retryable_automatic_error(error: &AppError) -> bool {
matches!(
error,
AppError::Auth(_) | AppError::Config(_) | AppError::Crypto(_) | AppError::CloudSync(_)
)
match error {
AppError::CloudSync(
CloudSyncError::SnapshotNotAccepted { .. } | CloudSyncError::RemoteFileRejected { .. },
) => false,
AppError::Auth(_) | AppError::Config(_) | AppError::Crypto(_) | AppError::CloudSync(_) => {
true
}
_ => false,
}
}
fn should_record_startup_check_failure(error: &AppError) -> bool {
@@ -2092,6 +2185,21 @@ mod tests {
assert!(!is_non_retryable_automatic_error(&AppError::Io(
std::io::Error::new(std::io::ErrorKind::TimedOut, "timeout")
)));
assert!(!is_non_retryable_automatic_error(&AppError::CloudSync(
CloudSyncError::SnapshotNotAccepted {
revision: "r1".to_string(),
}
)));
assert!(!is_non_retryable_automatic_error(&AppError::CloudSync(
CloudSyncError::RemoteFileRejected {
filename: "nyaterm-x.blob".to_string(),
}
)));
assert!(is_non_retryable_automatic_error(&AppError::CloudSync(
CloudSyncError::SnapshotMissing {
revision: "r1".to_string(),
}
)));
}
#[test]
src-tauri/src/core/cloud_sync/operator.rs
+100 -11
View File
@@ -1,4 +1,4 @@
use std::collections::HashMap;
use std::collections::{HashMap, HashSet};
use std::io;
use std::sync::Arc;
use std::time::Duration;
@@ -136,6 +136,7 @@ fn map_optional_read(result: Result<Buffer, Error>) -> AppResult<Option<Vec<u8>>
pub(super) struct MemoryRemote {
files: Arc<StdMutex<HashMap<String, Vec<u8>>>>,
fail_writes: Arc<StdMutex<Vec<String>>>,
drop_writes: Arc<StdMutex<Vec<String>>>,
}
#[cfg(test)]
@@ -144,6 +145,7 @@ impl MemoryRemote {
Self {
files: Arc::new(StdMutex::new(files)),
fail_writes: Arc::new(StdMutex::new(Vec::new())),
drop_writes: Arc::new(StdMutex::new(Vec::new())),
}
}
@@ -154,6 +156,13 @@ impl MemoryRemote {
.push(needle.to_string());
}
pub(super) fn drop_next_write_containing(&self, needle: &str) {
self.drop_writes
.lock()
.expect("lock drop writes")
.push(needle.to_string());
}
pub(super) fn file(&self, path: &str) -> Option<Vec<u8>> {
self.files.lock().expect("lock files").get(path).cloned()
}
@@ -183,6 +192,17 @@ impl MemoryRemote {
)));
}
drop(fail_writes);
let mut drop_writes = self.drop_writes.lock().expect("lock drop writes");
if let Some(index) = drop_writes
.iter()
.position(|needle| path.contains(needle.as_str()))
{
drop_writes.remove(index);
return Ok(());
}
drop(drop_writes);
self.files
.lock()
.expect("lock files")
@@ -530,6 +550,8 @@ struct GiteeSnippetFile {
content: Option<String>,
#[serde(default)]
raw_url: Option<String>,
#[serde(default)]
truncated: bool,
}
impl GiteeSnippetRemote {
@@ -582,13 +604,17 @@ impl GiteeSnippetRemote {
let Some(file) = snippet.files.get(&filename) else {
return Ok(None);
};
let content = match file
.content
.as_deref()
.filter(|value| !value.trim().is_empty())
{
Some(content) => content.to_string(),
None => self.fetch_raw_file(&filename, file).await?,
let content = if file.truncated {
self.fetch_raw_file(&filename, file).await?
} else {
match file
.content
.as_deref()
.filter(|value| !value.trim().is_empty())
{
Some(content) => content.to_string(),
None => self.fetch_raw_file(&filename, file).await?,
}
};
decode_gitee_file_content(&content).map(Some)
}
@@ -675,7 +701,7 @@ impl GiteeSnippetRemote {
&self,
files: serde_json::Map<String, serde_json::Value>,
) -> AppResult<()> {
let body = gitee_patch_body(self.access_token.as_str(), files);
let body = gitee_patch_body(self.access_token.as_str(), files.clone());
let url = format!("{}/gists/{}", self.api_endpoint, self.gist_id);
let response = self
.client
@@ -684,7 +710,9 @@ impl GiteeSnippetRemote {
.send()
.await
.map_err(map_gitee_client_error)?;
let _: serde_json::Value = decode_gitee_response(response).await?;
let snippet: GiteeSnippet = decode_gitee_response(response).await?;
let response_files: HashSet<String> = snippet.files.keys().cloned().collect();
ensure_gist_patch_accepted(&files, &response_files)?;
Ok(())
}
}
@@ -716,6 +744,28 @@ fn gitee_patch_body(
})
}
fn ensure_gist_patch_accepted(
requested: &serde_json::Map<String, serde_json::Value>,
response_filenames: &HashSet<String>,
) -> AppResult<()> {
for (filename, value) in requested {
let should_exist = !value.is_null();
let exists = response_filenames.contains(filename);
if should_exist && !exists {
return Err(crate::error::CloudSyncError::RemoteFileRejected {
filename: filename.clone(),
}
.into());
}
if !should_exist && exists {
return Err(AppError::Config(format!(
"Remote gist still contains deleted file '{filename}'"
)));
}
}
Ok(())
}
fn decode_gitee_file_content(content: &str) -> AppResult<Vec<u8>> {
BASE64_STANDARD
.decode(content.trim())
@@ -920,7 +970,9 @@ impl GithubGistRemote {
.send()
.await
.map_err(map_github_gist_client_error)?;
let _: serde_json::Value = decode_github_gist_response(response).await?;
let gist: GithubGist = decode_github_gist_response(response).await?;
let response_files: HashSet<String> = gist.files.keys().cloned().collect();
ensure_gist_patch_accepted(files, &response_files)?;
Ok(())
}
}
@@ -1266,6 +1318,33 @@ mod tests {
);
}
#[test]
fn gist_patch_requires_written_files_in_response() {
let mut requested = serde_json::Map::new();
requested.insert(
"nyaterm-new.blob".to_string(),
serde_json::json!({ "content": "abc" }),
);
let response = HashSet::from(["nyaterm-old.blob".to_string()]);
let err = ensure_gist_patch_accepted(&requested, &response).expect_err("missing file");
assert!(matches!(
err,
AppError::CloudSync(crate::error::CloudSyncError::RemoteFileRejected { filename })
if filename == "nyaterm-new.blob"
));
}
#[test]
fn gist_patch_accepts_when_response_contains_file() {
let mut requested = serde_json::Map::new();
requested.insert(
"nyaterm-new.blob".to_string(),
serde_json::json!({ "content": "abc" }),
);
let response = HashSet::from(["nyaterm-new.blob".to_string()]);
ensure_gist_patch_accepted(&requested, &response).expect("accepted");
}
#[test]
fn webdav_remote_layout_paths_support_empty_and_nested_roots() {
let webdav = webdav_remote();
@@ -1413,6 +1492,16 @@ mod tests {
assert!(file.truncated);
}
#[test]
fn gitee_snippet_file_deserializes_truncated_flag() {
let file: GiteeSnippetFile = serde_json::from_str(
r#"{"content":"partial","raw_url":"https://gitee.com/raw","truncated":true}"#,
)
.expect("deserialize gitee file");
assert!(file.truncated);
}
#[test]
fn github_gist_update_conflict_is_retryable() {
let error = AppError::Config(
src-tauri/src/core/cloud_sync/protocol.rs
+41 -3
View File
@@ -1,7 +1,7 @@
use crate::core::portable_snapshot::{
DecodedPortableSnapshot, PortableSnapshot, encode_portable_snapshot,
};
use crate::error::{AppResult, CloudSyncError};
use crate::error::{AppError, AppResult, CloudSyncError};
use super::crypto::encrypt_snapshot_bytes;
use super::operator::CloudRemote;
@@ -54,7 +54,17 @@ pub(super) async fn verify_uploaded_sync_snapshot(
remote_root: &str,
pointer: &RemoteSyncPointer,
) -> AppResult<PortableSnapshot> {
read_snapshot_for_pointer(remote, remote_root, pointer).await
match read_snapshot_for_pointer(remote, remote_root, pointer).await {
Ok(snapshot) => Ok(snapshot),
Err(AppError::CloudSync(CloudSyncError::SnapshotMissing { revision })) => {
tracing::warn!(
revision = %revision,
"Uploaded sync snapshot is missing on remote; latest pointer was not changed"
);
Err(CloudSyncError::SnapshotNotAccepted { revision }.into())
}
Err(error) => Err(error),
}
}
pub(super) async fn read_snapshot_for_pointer(
@@ -251,7 +261,7 @@ mod tests {
PortableAppSettings, PortableSnapshotKind, calculate_payload_hash,
calculate_v3_raw_payload_hash, encode_v3_raw_snapshot_redb_for_test,
};
use crate::error::AppError;
use crate::error::{AppError, CloudSyncError};
use crate::utils::crypto::set_master_password;
use super::super::migration::{RemoteSnapshotResolution, resolve_remote_snapshot};
@@ -483,6 +493,34 @@ mod tests {
set_master_password(None);
}
#[tokio::test(flavor = "current_thread")]
async fn verify_uploaded_missing_snapshot_maps_to_not_accepted() {
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);
memory.drop_next_write_containing("snapshots/r2");
upload_sync_snapshot(&remote, "nyaterm", &new_snapshot)
.await
.expect("upload reports success even when silently dropped");
let result = verify_uploaded_sync_snapshot(&remote, "nyaterm", &new_pointer).await;
assert!(matches!(
result,
Err(AppError::CloudSync(CloudSyncError::SnapshotNotAccepted { revision }))
if revision == "r2"
));
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");
src-tauri/src/error.rs
+8
View File
@@ -64,6 +64,14 @@ pub enum CloudSyncError {
)]
SnapshotMissing { revision: String },
#[error(
"New sync snapshot was not accepted by remote storage (candidate revision {revision}; latest was not changed)."
)]
SnapshotNotAccepted { revision: String },
#[error("Remote sync storage rejected file '{filename}' (gist may be at file capacity).")]
RemoteFileRejected { filename: String },
#[error(
"Remote sync snapshot revision mismatch: latest points to {pointer_revision} but snapshot contains {snapshot_revision}."
)]