fix(sync): 恢复过期会话并持久化上传尝试结果

frontend/src-tauri/src/sync_auth.rs

@@ -158,6 +158,27 @@ pub async fn client(
         saved.allow_test_http,
     )
 }
+/// The coordinator must serialize calls. Only use for read-only or durably idempotent work:
+/// a 401 repeats the operation once with the same device after rotating its session.
+pub async fn authenticated<T, F, Fut>(
+    credentials: &Credentials,
+    endpoint: &str,
+    account: &str,
+    mut action: F,
+) -> Result<T>
+where
+    F: FnMut(SyncClient) -> Fut,
+    Fut: Future<Output = Result<T>>,
+{
+    let first = client(credentials, endpoint, account, false).await?;
+    match guarded(credentials, action(first)).await {
+        Err(error) if error.status == 401 => {
+            let refreshed = client(credentials, endpoint, account, true).await?;
+            guarded(credentials, action(refreshed)).await
+        }
+        result => result,
+    }
+}
 pub async fn logout(credentials: &Credentials, endpoint: &str, account: &str) -> Result<()> {
     let client = client(credentials, endpoint, account, false).await?;
     // A failed server revocation is reported; the encrypted record remains available for retry.

frontend/src-tauri/src/sync_client.rs

@@ -30,6 +30,20 @@ impl SyncError {
     }
 }
 type Result<T> = std::result::Result<T, SyncError>;
+struct Attempt<'a, W: WorkspaceAccess> {
+    workspace: &'a W,
+    job: &'a Job,
+    finished: bool,
+}
+impl<W: WorkspaceAccess> Drop for Attempt<'_, W> {
+    fn drop(&mut self) {
+        if !self.finished {
+            let _ = self
+                .workspace
+                .access(|ws| ws.sync_attempt_interrupt(self.job));
+        }
+    }
+}
 pub trait WorkspaceAccess: Send + Sync {
     fn access<T>(
         &self,
@@ -309,19 +323,36 @@ impl SyncClient {
         if job.state == "conflict" {
             return Err(SyncError::new("REVISION_CONFLICT"));
         }
-        if job.operation == "put" && job.base_revision.is_none() {
-            self.upload(workspace, binding, &job).await?;
+        workspace.access(|ws| ws.sync_attempt_start(&job))?;
+        let mut attempt = Attempt {
+            workspace,
+            job: &job,
+            finished: false,
+        };
+        let result = async {
+            if job.operation == "put" && job.base_revision.is_none() {
+                self.upload(workspace, binding, &job).await?;
+            }
+            let payload = workspace.access(|ws| ws.sync_commit_payload(&job))?;
+            let revision = self
+                .json(
+                    Method::POST,
+                    &format!("sync/v1/vaults/{}/revisions", binding.remote_vault),
+                    Some(payload),
+                )
+                .await?;
+            workspace.access(|ws| ws.sync_ack(&job, &revision))?;
+            Ok(true)
         }
-        let payload = workspace.access(|ws| ws.sync_commit_payload(&job))?;
-        let revision = self
-            .json(
-                Method::POST,
-                &format!("sync/v1/vaults/{}/revisions", binding.remote_vault),
-                Some(payload),
+        .await;
+        workspace.access(|ws| {
+            ws.sync_attempt_finish(
+                &job,
+                result.as_ref().err().map(|e: &SyncError| e.code.as_str()),
             )
-            .await?;
-        workspace.access(|ws| ws.sync_ack(&job, &revision))?;
-        Ok(true)
+        })?;
+        attempt.finished = true;
+        result
     }
     pub async fn pull_page(
         &self,
@@ -577,3 +608,40 @@ fn identifier(value: &str) -> Result<()> {
     }
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[tokio::test]
+    async fn dropping_a_pending_attempt_marks_interruption_without_restart() {
+        let dir = tempfile::tempdir().unwrap();
+        let mut ws = Workspace::open(dir.path()).unwrap();
+        ws.write("cancel.md", "", b"retained", "local").unwrap();
+        let binding = ws
+            .sync_bind_empty("https://sync.example", "remote", "account")
+            .unwrap();
+        let job = ws.sync_next(&binding.id).unwrap().unwrap();
+        let workspace = Arc::new(Mutex::new(ws));
+        let future = async {
+            workspace.access(|ws| ws.sync_attempt_start(&job)).unwrap();
+            let _attempt = Attempt {
+                workspace: &workspace,
+                job: &job,
+                finished: false,
+            };
+            std::future::pending::<()>().await;
+        };
+        assert!(tokio::time::timeout(Duration::from_millis(10), future)
+            .await
+            .is_err());
+        let rows = workspace
+            .access(|ws| ws.sync_attempts(&binding.id))
+            .unwrap();
+        assert_eq!(rows[0]["outcome"], "interrupted");
+        assert_eq!(rows[0]["attempts"], 1);
+        assert_eq!(
+            workspace.access(|ws| ws.read("cancel.md")).unwrap().content,
+            "retained"
+        );
+    }
+}

frontend/src-tauri/src/sync_commands.rs

@@ -268,6 +268,11 @@ pub fn sync_status(host: State<'_, Host>) -> Result<Value, String> {
             paused,
             conflicts,
             ws.pending_count()?,
+            binding
+                .as_ref()
+                .map(|b| ws.sync_attempts(&b.id))
+                .transpose()?
+                .unwrap_or_default(),
             binding
                 .as_ref()
                 .map(|b| ws.sync_retry(&b.id))
@@ -275,7 +280,7 @@ pub fn sync_status(host: State<'_, Host>) -> Result<Value, String> {
                 .unwrap_or_default(),
         ))
     })?;
-    let (vault_id, binding, paused, conflicts, pending, retry) = snapshot;
+    let (vault_id, binding, paused, conflicts, pending, attempts, retry) = snapshot;
     let credential_state = if let Some(b) = &binding {
         sync_auth::available(&host.credentials, &b.endpoint, &b.account)
             .map(|exists| {
@@ -293,7 +298,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":retry.error,"retry_in":retry.retry_at.map(|_| retry.remaining(now())),"failures":retry.failures,"halted":retry.halted}),
+        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,"attempts":attempts}),
     )
 }
 #[tauri::command]
@@ -387,15 +392,11 @@ fn now() -> i64 {
 
 async fn cycle(host: &Host, binding: &Binding) -> Result<(), SyncError> {
     let epoch = host.sync.epoch.load(Ordering::SeqCst);
-    let work = async {
-        let client = sync_auth::client(
-            &host.credentials,
-            &binding.endpoint,
-            &binding.account,
-            false,
-        )
-        .await?;
-        let work = async {
+    let work = sync_auth::authenticated(
+        &host.credentials,
+        &binding.endpoint,
+        &binding.account,
+        |client| async move {
             client.handshake().await?;
             host.workspace.access(|ws| ws.sync_discover(&binding.id))?;
             for _ in 0..10 {
@@ -418,9 +419,8 @@ async fn cycle(host: &Host, binding: &Binding) -> Result<(), SyncError> {
             }
             client.pull_page(&host.workspace, binding).await?;
             Ok(())
-        };
-        sync_auth::guarded(&host.credentials, work).await
-    };
+        },
+    );
     tokio::pin!(work);
     let mut tick = tokio::time::interval(Duration::from_millis(50));
     loop {

frontend/src-tauri/src/sync_retry.rs

@@ -17,7 +17,57 @@ impl Retry {
             .unwrap_or(0)
     }
 }
+fn machine_code(code: &str) -> &str {
+    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"
+    }
+}
 impl Workspace {
+    pub fn sync_attempt_start(&self, job: &crate::sync_state::Job) -> Result<()> {
+        self.check_job(job)?;
+        self.db.execute("INSERT INTO sync_attempts VALUES (?1,?2,1,'running',NULL) ON CONFLICT(binding,operation_id) DO UPDATE SET attempts=MIN(attempts+1,2147483647),outcome='running',error=NULL", params![job.binding, job.operation_id])?;
+        Ok(())
+    }
+    pub fn sync_attempt_interrupt(&self, job: &crate::sync_state::Job) -> Result<()> {
+        self.check_job(job)?;
+        self.db.execute("UPDATE sync_attempts SET outcome=CASE WHEN EXISTS(SELECT 1 FROM sync_jobs j WHERE j.binding=?1 AND j.operation_id=?2 AND j.state='acked') THEN 'succeeded' ELSE 'interrupted' END WHERE binding=?1 AND operation_id=?2 AND outcome='running'", params![job.binding,job.operation_id])?;
+        Ok(())
+    }
+    pub fn sync_attempt_finish(
+        &self,
+        job: &crate::sync_state::Job,
+        error: Option<&str>,
+    ) -> Result<()> {
+        self.check_job(job)?;
+        self.db.execute(
+            "UPDATE sync_attempts SET outcome=?3,error=?4 WHERE binding=?1 AND operation_id=?2",
+            params![
+                job.binding,
+                job.operation_id,
+                if error.is_some() {
+                    "failed"
+                } else {
+                    "succeeded"
+                },
+                error.map(machine_code)
+            ],
+        )?;
+        Ok(())
+    }
+    pub fn sync_attempts(&self, binding: &str) -> Result<Vec<serde_json::Value>> {
+        self.check_binding(binding)?;
+        let mut statement = self.db.prepare("SELECT a.operation_id,j.path,a.attempts,a.outcome,a.error,j.state FROM sync_attempts a JOIN sync_jobs j ON j.binding=a.binding AND j.operation_id=a.operation_id WHERE a.binding=?1 AND j.state NOT IN ('acked','archived') ORDER BY a.rowid LIMIT 20")?;
+        let rows = statement.query_map([binding], |row| Ok(serde_json::json!({"operation_id":row.get::<_,String>(0)?,"path":row.get::<_,String>(1)?,"attempts":row.get::<_,i64>(2)?,"outcome":row.get::<_,String>(3)?,"error":row.get::<_,Option<String>>(4)?,"state":row.get::<_,String>(5)?})))?;
+        Ok(rows.collect::<std::result::Result<Vec<_>, _>>()?)
+    }
+
     pub fn sync_retry(&self, binding: &str) -> Result<Retry> {
         self.check_binding(binding)?;
         Ok(self
@@ -56,16 +106,7 @@ impl Workspace {
             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 code = machine_code(code);
         let failures = if code == "CREDENTIALS_LOCKED" {
             0
         } else {
@@ -87,6 +128,50 @@ impl Workspace {
 mod tests {
     use super::*;
     #[test]
+    fn interrupted_job_attempts_survive_twenty_restarts_without_changing_commit_base() {
+        let dir = tempfile::tempdir().unwrap();
+        let mut ws = Workspace::open(dir.path()).unwrap();
+        ws.write("attempt.md", "", b"body", "local").unwrap();
+        let binding = ws
+            .sync_bind_empty("https://sync.example", "remote", "account")
+            .unwrap();
+        let job = ws.sync_next(&binding.id).unwrap().unwrap();
+        let payload = ws.sync_commit_payload(&job).unwrap();
+        for count in 1..=20 {
+            ws.sync_attempt_start(&job).unwrap();
+            drop(ws);
+            ws = Workspace::open(dir.path()).unwrap();
+            let rows = ws.sync_attempts(&binding.id).unwrap();
+            assert_eq!(rows[0]["attempts"], count);
+            assert_eq!(rows[0]["outcome"], "interrupted");
+            assert_eq!(ws.sync_commit_payload(&job).unwrap(), payload);
+        }
+        ws.sync_attempt_finish(&job, Some("response with secret body"))
+            .unwrap();
+        assert_eq!(
+            ws.sync_attempts(&binding.id).unwrap()[0]["error"],
+            "SYNC_REMOTE_ERROR"
+        );
+        ws.sync_attempt_start(&job).unwrap();
+        let mut revision = payload.clone();
+        revision["hash"] = payload["content_hash"].clone();
+        revision["vault_id"] = serde_json::json!("remote");
+        revision["sequence"] = serde_json::json!(1);
+        ws.sync_ack(&job, &revision).unwrap();
+        drop(ws);
+        ws = Workspace::open(dir.path()).unwrap();
+        assert!(ws.sync_attempts(&binding.id).unwrap().is_empty());
+        assert_eq!(
+            ws.db
+                .query_row("SELECT outcome FROM sync_attempts", [], |r| r
+                    .get::<_, String>(0))
+                .unwrap(),
+            "succeeded"
+        );
+        ws.sync_unbind(&binding.id).unwrap();
+        assert!(ws.sync_attempt_finish(&job, None).is_err());
+    }
+    #[test]
     fn retry_history_survives_twenty_reopens_and_isolates_bindings() {
         let dir = tempfile::tempdir().unwrap();
         let mut ws = Workspace::open(dir.path()).unwrap();

frontend/src-tauri/src/sync_state.rs

@@ -197,7 +197,7 @@ impl Workspace {
         }
         Ok(job)
     }
-    fn check_job(&self, job: &Job) -> Result<()> {
+    pub(crate) fn check_job(&self, job: &Job) -> Result<()> {
         self.check_binding(&job.binding)?;
         let state: String = self.db.query_row(
             "SELECT state FROM sync_jobs WHERE binding=?1 AND operation_id=?2",

frontend/src-tauri/src/workspace.rs

@@ -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 > 9 {
+        if version > 10 {
             return Err(HostError::new("SCHEMA_INCOMPATIBLE"));
         }
-        if (1..9).contains(&version) {
+        if (1..10).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_attempts (binding TEXT NOT NULL,operation_id TEXT NOT NULL,attempts INTEGER NOT NULL,outcome TEXT NOT NULL,error TEXT,PRIMARY KEY(binding,operation_id));
             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));")?;
@@ -179,7 +180,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=9; COMMIT;")?;
+        db.execute_batch("UPDATE sync_attempts SET outcome=CASE WHEN EXISTS(SELECT 1 FROM sync_jobs j WHERE j.binding=sync_attempts.binding AND j.operation_id=sync_attempts.operation_id AND j.state='acked') THEN 'succeeded' ELSE 'interrupted' END WHERE outcome='running'; PRAGMA user_version=10; COMMIT;")?;
         let vault_id: String = db
             .query_row("SELECT id FROM identity", [], |r| r.get(0))
             .optional()?

frontend/src-tauri/tests/sync_push.rs

@@ -618,6 +618,14 @@ async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicate
     }
     std::fs::remove_file(root.path().join("interrupt-upload")).unwrap();
     let large_workspace = Arc::new(Mutex::new(Workspace::open(large_root.path()).unwrap()));
+    let interrupted = large_workspace
+        .lock()
+        .unwrap()
+        .sync_attempts(&large_binding.id)
+        .unwrap();
+    assert_eq!(interrupted.len(), 1);
+    assert_eq!(interrupted[0]["attempts"], 10);
+    assert_eq!(interrupted[0]["outcome"], "interrupted");
     assert!(client
         .push_one(&large_workspace, &large_binding)
         .await
@@ -707,6 +715,59 @@ async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicate
         .unwrap()
         .unlock(Zeroizing::new(b"fixture-stronghold-password".to_vec()))
         .unwrap();
+    // Expire the access token early in this isolated fixture; the refresh token stays valid.
+    let database = rusqlite::Connection::open(root.path().join("sync.sqlite3")).unwrap();
+    database
+        .execute("UPDATE sessions SET expires=0", [])
+        .unwrap();
+    let attempts = std::sync::atomic::AtomicUsize::new(0);
+    let recovered = sync_auth::authenticated(&credentials, &canonical, "rust-fixture", |client| {
+        attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+        async move {
+            client
+                .json(reqwest::Method::GET, "sync/v1/vaults", None)
+                .await
+        }
+    })
+    .await
+    .unwrap();
+    assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 2);
+    assert!(recovered["items"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .any(|v| v["id"] == remote));
+    // Even a repeated 401 must stop after one rotation, rather than refresh indefinitely.
+    attempts.store(0, std::sync::atomic::Ordering::SeqCst);
+    let denied = sync_auth::authenticated(&credentials, &canonical, "rust-fixture", |_client| {
+        attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+        async {
+            Err::<(), _>(notesagent_host::sync_client::SyncError {
+                code: "UNAUTHORIZED".into(),
+                status: 401,
+                retry_after: None,
+            })
+        }
+    })
+    .await
+    .unwrap_err();
+    assert_eq!(denied.status, 401);
+    assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 2);
+    attempts.store(0, std::sync::atomic::Ordering::SeqCst);
+    let denied = sync_auth::authenticated(&credentials, &canonical, "rust-fixture", |_client| {
+        attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+        async {
+            Err::<(), _>(notesagent_host::sync_client::SyncError {
+                code: "FORBIDDEN".into(),
+                status: 403,
+                retry_after: None,
+            })
+        }
+    })
+    .await
+    .unwrap_err();
+    assert_eq!(denied.status, 403);
+    assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 1);
     let restored = sync_auth::client(&credentials, &canonical, "rust-fixture", false)
         .await
         .unwrap();
@@ -723,6 +784,30 @@ async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicate
             .status,
         401
     );
+    sync_auth::login(
+        &credentials,
+        &endpoint,
+        "rust-fixture",
+        Zeroizing::new("controlled-fixture-password".into()),
+        "Revoked Host",
+        true,
+    )
+    .await
+    .unwrap();
+    database.execute("DELETE FROM sessions", []).unwrap();
+    attempts.store(0, std::sync::atomic::Ordering::SeqCst);
+    let revoked = sync_auth::authenticated(&credentials, &canonical, "rust-fixture", |client| {
+        attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+        async move {
+            client
+                .json(reqwest::Method::GET, "sync/v1/vaults", None)
+                .await
+        }
+    })
+    .await
+    .unwrap_err();
+    assert_eq!(revoked.status, 401);
+    assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 1);
 }
 
 #[tokio::test]