Kronecker/NotesAgentic
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 5 Wiki Activity

fix(sync): 跨工作区重启保留重试与暂停状态

Browse Source
This commit is contained in:
admin
2026-09-08 19:58:39 +08:00
parent 5c995e66be
commit 9e295b9a5d
7 changed files with 227 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frontend/src-tauri/src/lib.rs
+2
View File
@@ -22,3 +22,5 @@ pub mod sync_resolution;
pub mod sync_state;
pub mod workspace;
pub mod workspace_broker;
pub mod sync_retry;
frontend/src-tauri/src/sync_commands.rs
+71 -44
View File
@@ -13,7 +13,7 @@ use std::{
atomic::{AtomicU64, Ordering},
Mutex,
},
time::{Duration, Instant},
time::{Duration, SystemTime, UNIX_EPOCH},
};
use tauri::State;
use zeroize::Zeroizing;
@@ -26,10 +26,6 @@ pub struct Runtime {
#[derive(Default)]
struct Progress {
running: bool,
error: Option<String>,
failures: u32,
retry: Option<Instant>,
halted: bool,
}
impl Runtime {
pub fn cancel(&self) {
@@ -63,6 +59,21 @@ pub async fn sync_login(host: State<'_, Host>, request: Login) -> Result<Connect
)
.await
.map_err(|e| e.code)?;
with_workspace(&host, |ws| {
if let Some(binding) = ws.sync_binding()? {
if binding.endpoint == endpoint && binding.account == request.account {
ws.sync_retry_clear(&binding.id)?;
}
}
Ok(())
})
.or_else(|error| {
if error == "VAULT_NOT_OPEN" {
Ok(())
} else {
Err(error)
}
})?;
host.sync.status.lock().map_err(|_| "HOST_BUSY")?.clear();
Ok(Connection {
endpoint,
@@ -223,7 +234,13 @@ pub fn sync_unbind(host: State<'_, Host>, binding_id: String) -> Result<(), Stri
#[tauri::command]
pub fn sync_pause(host: State<'_, Host>, binding_id: String, paused: bool) -> Result<(), String> {
host.sync.cancel();
with_workspace(&host, |ws| ws.sync_pause(&binding_id, paused))?;
with_workspace(&host, |ws| {
ws.sync_pause(&binding_id, paused)?;
if !paused {
ws.sync_retry_clear(&binding_id)?;
}
Ok(())
})?;
host.sync
.status
.lock()
@@ -247,13 +264,18 @@ pub fn sync_status(host: State<'_, Host>) -> Result<Value, String> {
.unwrap_or_default();
Ok((
ws.vault_id.clone(),
binding,
binding.clone(),
paused,
conflicts,
ws.pending_count()?,
binding
.as_ref()
.map(|b| ws.sync_retry(&b.id))
.transpose()?
.unwrap_or_default(),
))
})?;
let (vault_id, binding, paused, conflicts, pending) = snapshot;
let (vault_id, binding, paused, conflicts, pending, retry) = snapshot;
let credential_state = if let Some(b) = &binding {
sync_auth::available(&host.credentials, &b.endpoint, &b.account)
.map(|exists| {
@@ -271,7 +293,7 @@ pub fn sync_status(host: State<'_, Host>) -> Result<Value, String> {
let statuses = host.sync.status.lock().map_err(|_| "HOST_BUSY")?;
let status = binding.as_ref().and_then(|b| statuses.get(&b.id));
Ok(
json!({"vault_id":vault_id,"binding":binding,"paused":paused,"pending":pending,"conflicts":conflicts,"credential_state":credential_state,"running":status.is_some_and(|s|s.running),"error":status.and_then(|s|s.error.as_ref()),"retry_in":status.and_then(|s|s.retry).map(|time|time.saturating_duration_since(Instant::now()).as_secs())}),
json!({"vault_id":vault_id,"binding":binding,"paused":paused,"pending":pending,"conflicts":conflicts,"credential_state":credential_state,"running":status.is_some_and(|s|s.running),"error":retry.error,"retry_in":retry.retry_at.map(|_| retry.remaining(now())),"failures":retry.failures,"halted":retry.halted}),
)
}
#[tauri::command]
@@ -315,49 +337,54 @@ pub async fn run(host: &Host, manual: bool) -> Result<(), String> {
if with_workspace(host, |ws| ws.sync_paused(&binding.id))? {
return Err("SYNC_PAUSED".into());
}
{
let mut statuses = host.sync.status.lock().map_err(|_| "HOST_BUSY")?;
let state = statuses.entry(binding.id.clone()).or_default();
if !manual && (state.halted || state.retry.is_some_and(|v| v > Instant::now())) {
return Ok(());
}
state.running = true;
state.error = None;
let retry = with_workspace(host, |ws| ws.sync_retry(&binding.id))?;
if !manual && (retry.halted || retry.remaining(now()) > 0) {
return Ok(());
}
host.sync
.status
.lock()
.map_err(|_| "HOST_BUSY")?
.entry(binding.id.clone())
.or_default()
.running = true;
let result = cycle(host, &binding).await;
let mut statuses = host.sync.status.lock().map_err(|_| "HOST_BUSY")?;
let state = statuses.entry(binding.id.clone()).or_default();
state.running = false;
host.sync
.status
.lock()
.map_err(|_| "HOST_BUSY")?
.entry(binding.id.clone())
.or_default()
.running = false;
match result {
Ok(()) => {
*state = Progress::default();
Ok(())
}
Ok(()) => with_workspace(host, |ws| ws.sync_retry_clear(&binding.id)),
Err(error) => {
state.failures = if error.code == "CREDENTIALS_LOCKED" {
0
} else {
state.failures.saturating_add(1)
};
state.error = Some(error.code.clone());
state.halted = matches!(error.status, 401 | 403 | 413 | 426 | 507)
|| matches!(
error.code.as_str(),
"PROTOCOL_INCOMPATIBLE" | "SYNC_LOGIN_REQUIRED"
);
state.retry = Some(
Instant::now()
+ Duration::from_secs(
error
.retry_after
.unwrap_or(2u64.saturating_pow(state.failures.min(8)))
.clamp(1, 3600),
),
);
if !matches!(
error.code.as_str(),
"SYNC_CANCELLED" | "SYNC_BINDING_CHANGED" | "VAULT_CHANGED"
) {
with_workspace(host, |ws| {
ws.sync_retry_fail(
&binding.id,
&error.code,
error.status,
error.retry_after,
now(),
)
})?;
}
Err(error.code)
}
}
}
fn now() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_secs()
.min(i64::MAX as u64) as i64
}
async fn cycle(host: &Host, binding: &Binding) -> Result<(), SyncError> {
let epoch = host.sync.epoch.load(Ordering::SeqCst);
let work = async {
frontend/src-tauri/src/sync_retry.rs
+130
View File
@@ -0,0 +1,130 @@
//! Binding-scoped retry decisions survive restart; wall time is bounded after clock changes.
use crate::workspace::{Result, Workspace};
use rusqlite::{params, OptionalExtension};
use serde::Serialize;
#[derive(Default, Debug, Serialize)]
pub struct Retry {
pub error: Option<String>,
pub failures: u32,
pub retry_at: Option<i64>,
pub halted: bool,
}
impl Retry {
pub fn remaining(&self, now: i64) -> u64 {
self.retry_at
.map(|at| at.saturating_sub(now).clamp(0, 3600) as u64)
.unwrap_or(0)
}
}
impl Workspace {
pub fn sync_retry(&self, binding: &str) -> Result<Retry> {
self.check_binding(binding)?;
Ok(self
.db
.query_row(
"SELECT error,failures,retry_at,halted FROM sync_retry WHERE binding=?1",
[binding],
|row| {
Ok(Retry {
error: row.get(0)?,
failures: row.get(1)?,
retry_at: row.get(2)?,
halted: row.get(3)?,
})
},
)
.optional()?
.unwrap_or_default())
}
pub fn sync_retry_clear(&mut self, binding: &str) -> Result<()> {
self.check_binding(binding)?;
self.db
.execute("DELETE FROM sync_retry WHERE binding=?1", [binding])?;
Ok(())
}
pub fn sync_retry_fail(
&mut self,
binding: &str,
code: &str,
status: u16,
retry_after: Option<u64>,
now: i64,
) -> Result<()> {
let previous = self.sync_retry(binding)?;
if code == "SYNC_CANCELLED" {
return Ok(());
}
// Only retain a bounded machine code, never an arbitrary remote response string.
let code = if !code.is_empty()
&& code.len() <= 80
&& code
.bytes()
.all(|b| b.is_ascii_uppercase() || b.is_ascii_digit() || b == b'_')
{
code
} else {
"SYNC_REMOTE_ERROR"
};
let failures = if code == "CREDENTIALS_LOCKED" {
0
} else {
previous.failures.saturating_add(1)
};
let halted = matches!(status, 401 | 403 | 413 | 426 | 507)
|| matches!(code, "PROTOCOL_INCOMPATIBLE" | "SYNC_LOGIN_REQUIRED");
let retry_at = now.saturating_add(
retry_after
.unwrap_or(2u64.saturating_pow(failures.min(8)))
.clamp(1, 3600) as i64,
);
self.db.execute("INSERT INTO sync_retry VALUES (?1,?2,?3,?4,?5) ON CONFLICT(binding) DO UPDATE SET error=excluded.error,failures=excluded.failures,retry_at=excluded.retry_at,halted=excluded.halted", params![binding,code,failures,retry_at,halted])?;
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn retry_history_survives_twenty_reopens_and_isolates_bindings() {
let dir = tempfile::tempdir().unwrap();
let mut ws = Workspace::open(dir.path()).unwrap();
let b = ws
.sync_bind_empty("https://sync.example", "remote", "account")
.unwrap();
for attempt in 1..=20 {
ws.sync_retry_fail(&b.id, "NETWORK_ERROR", 0, Some(120), 1000)
.unwrap();
drop(ws);
ws = Workspace::open(dir.path()).unwrap();
let state = ws.sync_retry(&b.id).unwrap();
assert_eq!(state.failures, attempt);
assert_eq!(state.remaining(1010), 110);
assert_eq!(state.remaining(-10000), 3600);
assert_eq!(state.remaining(1200), 0);
}
ws.sync_retry_fail(&b.id, "SYNC_CANCELLED", 0, None, 1100)
.unwrap();
assert_eq!(ws.sync_retry(&b.id).unwrap().failures, 20);
ws.sync_retry_fail(&b.id, "UNAUTHORIZED", 401, None, 1100)
.unwrap();
drop(ws);
ws = Workspace::open(dir.path()).unwrap();
assert!(ws.sync_retry(&b.id).unwrap().halted);
ws.sync_retry_clear(&b.id).unwrap();
assert!(ws.sync_retry(&b.id).unwrap().error.is_none());
ws.sync_retry_fail(&b.id, "secret response body", 500, None, 1100)
.unwrap();
assert_eq!(
ws.sync_retry(&b.id).unwrap().error.as_deref(),
Some("SYNC_REMOTE_ERROR")
);
ws.sync_unbind(&b.id).unwrap();
let next = ws
.sync_bind_empty("https://sync.example", "other", "account")
.unwrap();
assert!(ws.sync_retry(&next.id).unwrap().error.is_none());
assert!(ws.sync_retry_clear(&b.id).is_err());
}
}
frontend/src-tauri/src/workspace.rs
+4 -3
View File
@@ -135,10 +135,10 @@ impl Workspace {
let db = Connection::open(db_path)?;
db.execute_batch("PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL;")?;
let version: i64 = db.query_row("PRAGMA user_version", [], |r| r.get(0))?;
if version > 8 {
if version > 9 {
return Err(HostError::new("SCHEMA_INCOMPATIBLE"));
}
if (1..8).contains(&version) {
if (1..9).contains(&version) {
// Independent, complete SQLite backup before the schema ownership change.
let backup = managed.join(format!("host-schema{version}-{}.sqlite3", Uuid::new_v4()));
db.execute("VACUUM INTO ?1", [backup.to_string_lossy().as_ref()])?;
@@ -162,6 +162,7 @@ impl Workspace {
CREATE TABLE IF NOT EXISTS sync_initial_items (binding TEXT NOT NULL,sequence INTEGER NOT NULL,revision TEXT NOT NULL,PRIMARY KEY(binding,sequence));
CREATE TABLE IF NOT EXISTS payloads (operation_id TEXT PRIMARY KEY,hash TEXT NOT NULL,size INTEGER NOT NULL);
CREATE TABLE IF NOT EXISTS sync_observed (file_id TEXT PRIMARY KEY,path TEXT NOT NULL,hash TEXT NOT NULL,deleted INTEGER NOT NULL);
CREATE TABLE IF NOT EXISTS sync_retry (binding TEXT PRIMARY KEY,error TEXT,failures INTEGER NOT NULL,retry_at INTEGER,halted INTEGER NOT NULL);
CREATE TABLE IF NOT EXISTS sync_preferences (binding TEXT PRIMARY KEY,paused INTEGER NOT NULL DEFAULT 0);
CREATE TABLE IF NOT EXISTS sync_resolutions (binding TEXT NOT NULL,sequence INTEGER NOT NULL,choice TEXT NOT NULL,destination TEXT NOT NULL,expected TEXT NOT NULL,operation_id TEXT NOT NULL,rename_id TEXT NOT NULL,copy_id TEXT NOT NULL,state TEXT NOT NULL,PRIMARY KEY(binding,sequence));")?;
let has_origin: bool = db.query_row(
@@ -178,7 +179,7 @@ impl Workspace {
if version < 7 {
db.execute_batch("INSERT OR IGNORE INTO sync_observed SELECT f.id,COALESCE((SELECT o.path FROM outbox o WHERE o.file_id=f.id AND o.state IN ('pending','queued') ORDER BY rowid DESC LIMIT 1),(SELECT h.path FROM sync_heads h JOIN sync_bindings b ON h.binding=b.id WHERE h.file_id=f.id AND b.state='active'),f.path),COALESCE((SELECT o.hash FROM outbox o WHERE o.file_id=f.id AND o.state IN ('pending','queued') ORDER BY rowid DESC LIMIT 1),(SELECT h.hash FROM sync_heads h JOIN sync_bindings b ON h.binding=b.id WHERE h.file_id=f.id AND b.state='active'),f.hash),f.deleted FROM files f;")?;
}
db.execute_batch("PRAGMA user_version=8; COMMIT;")?;
db.execute_batch("PRAGMA user_version=9; COMMIT;")?;
let vault_id: String = db
.query_row("SELECT id FROM identity", [], |r| r.get(0))
.optional()?