fix(sync): protect native edits and connect Android realtime Pull

CHANGELOG.md

@@ -12,6 +12,13 @@
 
 <!-- ADD_NEW_HERE -->
 
+## 未发布 / Unreleased
+
+- **sync**：保护 Native 离线编辑、删除冲突和迟到 ACK；重试保留 mutationId，避免旧更新复活已删除笔记。Protect durable native edits, deletion conflicts and late ACKs without changing mutation identity.
+- **android**：复用 WebSocket 唤醒 Sync V2 Pull，提交后刷新界面，并通过前台周期 Pull 补偿丢失通知。Wake native Pull on notices/reconnect and refresh UI after committed Apply.
+- **sync**：修复同页创建后删除、工作区 Snapshot 清理和冲突后继续编辑的数据保护问题。Protect pending descendants, attachment bytes and newer local edits during recovery.
+- 验收范围与未完成平台见 `docs/sync-v2-reliability-validation.md`；这些改动尚未正式发布。See the validation record for evidence and remaining acceptance gaps; these changes are not released.
+
 ## v1.5.1 - 2026-10-08
 
 ### ✨ 新增

backend/src/routes/sync-v2.ts

@@ -804,6 +804,8 @@ app.post("/push", async (c) => {
           if (parsed.entityType === "note" && typeof serverPayload?.version === "number") {
             serverVersion=serverPayload.version;
             serverPayload = withEncryptedBlocksSupport(serverPayload);
+          } else if (parsed.entityType === "note" && !serverPayload) {
+            serverPayload = { id: parsed.entityId, __delete: true };
           }
         }
       }

backend/src/sync/apply.ts

@@ -248,6 +248,10 @@ function applyNote(db: Database.Database, input: ApplyMutationInput): number | n
     return nextVersion;
   }
 
+  // An update based on an existing revision is not a new entity. An old device
+  // must not resurrect a permanently deleted note, even after feed retention.
+  if (input.baseVersion !== undefined) throw new SyncError("VERSION_CONFLICT", "远端笔记已删除，请保留冲突或另存为新笔记");
+
   // 新建：客户端生成 UUID，离线也能创建，不依赖服务端分配 ID。
   const version = Math.max(1, num(p.version, 1));
   db.prepare(`

backend/src/sync/engine.ts

@@ -718,7 +718,9 @@ export class SyncEngine {
     const deletions: EngineRemotePayload[] = [];
     const wanted = new Set<string>();
 
-    for (const item of items) {
+    // Only the last operation per identity describes its state in Snapshot.
+    const latest = new Map(items.map((item) => [`${item.entityType}\u0000${item.entityId}`, item]));
+    for (const item of latest.values()) {
       if (item.operation === "delete") {
         deletions.push(item.entityType === "knowledge_tree_node"
           ? {
@@ -779,8 +781,7 @@ export class SyncEngine {
       );
     }
 
-    // 先应用 upsert 再应用 delete：
-    // 同一轮里若既有创建又有删除，删除应当是最终状态。
+    // Apply only the final operation for each entity in this feed page.
     return [...upserts, ...deletions];
   }
 
@@ -990,27 +991,34 @@ export class SyncEngine {
         ${ids.length ? `AND id NOT IN (${placeholders})` : ""}
         AND NOT EXISTS (SELECT 1 FROM sync_outbox o WHERE o.profileId=? AND o.scopeKey=?
           AND o.entityType=? AND o.entityId=${table}.id AND o.status IN ('pending','inflight','failed'))
-        ${extra}`).run(scope.workspaceId,...ids,this.profileId,scope.scopeKey,entityType);
+        AND NOT EXISTS (SELECT 1 FROM sync_conflicts c WHERE c.profileId=? AND c.scopeKey=?
+          AND c.entityType=? AND c.entityId=${table}.id AND c.status='unresolved')
+        ${extra}`).run(scope.workspaceId,...ids,this.profileId,scope.scopeKey,entityType,this.profileId,scope.scopeKey,entityType);
     };
     const removeComposite=(table:string,entityType:string,idSql:string,scopeSql:string)=>{
       const ids=[...(seen.get(entityType) || [])];
       const placeholders=ids.map(()=>"?").join(",");
       this.db.prepare(`DELETE FROM ${table} WHERE ${scopeSql}
         ${ids.length ? `AND (${idSql}) NOT IN (${placeholders})` : ""}
         AND NOT EXISTS (SELECT 1 FROM sync_outbox o WHERE o.profileId=? AND o.scopeKey=?
-          AND o.entityType=? AND o.entityId=(${idSql}) AND o.status IN ('pending','inflight','failed'))`)
-        .run(scope.workspaceId,...ids,this.profileId,scope.scopeKey,entityType);
+          AND o.entityType=? AND o.entityId=(${idSql}) AND o.status IN ('pending','inflight','failed'))
+        AND NOT EXISTS (SELECT 1 FROM sync_conflicts c WHERE c.profileId=? AND c.scopeKey=?
+          AND c.entityType=? AND c.entityId=(${idSql}) AND c.status='unresolved')`)
+        .run(scope.workspaceId,...ids,this.profileId,scope.scopeKey,entityType,this.profileId,scope.scopeKey,entityType);
     };
     runWithOutboxSuppressed(()=>runChangeFeedSuppressed(this.db,()=>this.db.transaction(()=>{
       removeMissing("attachments","attachment");
       removeComposite("favorites","favorite","favorites.userId || ':' || favorites.noteId","favorites.workspaceId=?");
       removeComposite("note_tags","note_tag","note_tags.noteId || ':' || note_tags.tagId","note_tags.noteId IN (SELECT id FROM notes WHERE workspaceId=?)");
       removeComposite("task_reminders","task_reminder","task_reminders.id","task_reminders.taskId IN (SELECT id FROM tasks WHERE workspaceId=?)");
-      removeMissing("tasks","task");
+      removeMissing("tasks","task","AND NOT EXISTS (SELECT 1 FROM task_reminders r WHERE r.taskId=tasks.id)");
       removeMissing("diaries","diary");
       removeMissing("mindmaps","mindmap");
-      removeMissing("notes","note");
-      removeMissing("notebooks","notebook","AND NOT EXISTS (SELECT 1 FROM notes n WHERE n.notebookId=notebooks.id)");
+      removeMissing("notes","note",`AND NOT EXISTS (SELECT 1 FROM attachments a WHERE a.noteId=notes.id)
+        AND NOT EXISTS (SELECT 1 FROM note_tags nt WHERE nt.noteId=notes.id)
+        AND NOT EXISTS (SELECT 1 FROM favorites f WHERE f.noteId=notes.id)`);
+      removeMissing("notebooks","notebook",`AND NOT EXISTS (SELECT 1 FROM notes n WHERE n.notebookId=notebooks.id)
+        AND NOT EXISTS (SELECT 1 FROM notebooks child WHERE child.parentId=notebooks.id)`);
       removeMissing("tags","tag","AND NOT EXISTS (SELECT 1 FROM note_tags nt WHERE nt.tagId=tags.id)");
     })()));
   }

backend/src/sync/push.ts

@@ -72,9 +72,8 @@ export function coalesceMutations(rows: SyncOutboxRow[]): CoalescedMutation[] {
     existing.operation = row.operation;
     existing.payload = parsePayload(row.payload);
     // baseVersion 保持最早那条：它才代表"这串修改的共同祖先"。
-    if (existing.baseVersion === undefined && row.baseVersion !== null) {
-      existing.baseVersion = row.baseVersion;
-    }
+    // An absent first base means creation. Later local revisions must not turn
+    // that creation into an update of a server entity that does not exist yet.
   }
 
   return order.map((key) => byEntity.get(key) as CoalescedMutation);

backend/tests/sync-v2-android-server.ts

@@ -0,0 +1,49 @@
+// Explicit test helper, not a *.test.ts suite or production route. Loopback only.
+import fs from "node:fs";
+import path from "node:path";
+import crypto from "node:crypto";
+import { Hono } from "hono";
+import { cors } from "hono/cors";
+import { serve } from "@hono/node-server";
+
+async function main() {
+  const directory = process.env.SYNC_ACCEPTANCE_DIRECTORY;
+  if (!directory) throw new Error("SYNC_ACCEPTANCE_DIRECTORY required");
+  fs.mkdirSync(directory, { recursive: true });
+  process.env.DB_PATH = path.join(directory, "server.db");
+  process.env.ELECTRON_USER_DATA = directory;
+  process.env.JWT_SECRET = "isolated-sync-acceptance-secret-2026";
+  const { getDb } = await import("../src/db/schema");
+  const { signLoginToken, verifyLoginToken } = await import("../src/lib/auth-security");
+  const { attachRealtimeServer, broadcastToUser } = await import("../src/services/realtime");
+  const { setSyncBroadcaster } = await import("../src/sync/notify");
+  const { default: routes } = await import("../src/routes/sync-v2");
+  const userId = "isolated-android-sync-user", db = getDb();
+  db.prepare("INSERT OR IGNORE INTO users (id,username,passwordHash,createdAt,updatedAt) VALUES (?,?,'test-only','now','now')").run(userId, userId);
+  const fixtureToken = () => signLoginToken({ userId, username: userId, tokenVersion: 0 });
+  const app = new Hono();
+  app.use("*", cors({ origin: "https://localhost" }));
+  app.get("/acceptance/config", (c) => c.json({ userId, token: fixtureToken(), serverUrl: "http://127.0.0.1:47831" }));
+  app.use("/api/*", async (c, next) => {
+    const identity = verifyLoginToken(c.req.header("Authorization")?.replace(/^Bearer /, "") || "");
+    if (!identity) return c.json({ error: "UNAUTHORIZED" }, 401);
+    c.req.raw.headers.set("X-User-Id", identity.userId);
+    await next();
+  });
+  app.route("/api/sync/v2", routes);
+  app.get("/acceptance/note/:id", (c) => c.json(db.prepare("SELECT * FROM notes WHERE id=? AND userId=?").get(c.req.param("id"), userId) || null));
+  app.post("/acceptance/edit/:id", async (c) => {
+    const note = db.prepare("SELECT * FROM notes WHERE id=? AND userId=?").get(c.req.param("id"), userId) as Record<string, unknown> | undefined;
+    if (!note) return c.json({ error: "NOT_FOUND" }, 404);
+    const { text } = await c.req.json<{text:string}>();
+    const content = note.contentFormat === "markdown" ? text : JSON.stringify({ type: "doc", content: [{ type: "paragraph", content: [{ type: "text", text }] }] });
+    return app.request("http://localhost/api/sync/v2/push", { method: "POST", headers: { Authorization: `Bearer ${fixtureToken()}`, "Content-Type": "application/json" },
+      body: JSON.stringify({ scopeKey: "personal", deviceId: "http-client", mutations: [{ mutationId: crypto.randomUUID(), entityType: "note", entityId: note.id, operation: "upsert", baseVersion: note.version,
+        payload: { ...note, content, contentText: text } }] }) });
+  });
+  const server = serve({ fetch: app.fetch, hostname: "127.0.0.1", port: 47831 });
+  attachRealtimeServer(server);
+  setSyncBroadcaster((recipient, notice) => broadcastToUser(recipient, notice as never));
+  fs.writeFileSync(path.join(directory, "ready"), "ready");
+}
+void main();

backend/tests/sync-v2-engine.test.ts

@@ -706,6 +706,24 @@ test("无变更时也推进游标并 ACK，避免重复扫描", async () => {
   assert.deepEqual(remote.ackCalls, [42]);
 });
 
+test("同一批创建后删除只应用最终删除，不等待已不存在的 Snapshot", async () => {
+  resetSyncTables();
+  const db = getDb();
+  const { engine, remote, profileId } = createEngine();
+  remote.serverSequence = 43;
+  remote.changesQueue.push({
+    serverSequence: 43, nextSequence: 43, hasMore: false, resetRequired: false,
+    items: [
+      { sequence: 42, entityType: "note", entityId: "created-then-deleted", operation: "upsert" },
+      { sequence: 43, entityType: "note", entityId: "created-then-deleted", operation: "delete" },
+    ],
+  });
+  const status = await engine.syncOnce();
+  assert.notEqual(status.state, "error");
+  assert.equal(sync.getSyncState(db, profileId)?.lastSequence, 43);
+  assert.deepEqual(remote.ackCalls, [43]);
+});
+
 test("Change Feed upsert 缺少 Snapshot payload 时不推进游标或 ACK", async () => {
   resetSyncTables();
   const db = getDb();