Sync polish round 1: data-loss races, merge correctness, responsiveness

src/connectSync.ts

@@ -1037,11 +1037,18 @@ export class ObsidianMirrorFileSystem implements MirrorFileSystem {
     return binaryInfo(bytes);
   }
 
-  async writeBinary(input: string, source: AsyncIterable<Uint8Array>): Promise<void> {
+  async writeBinary(input: string, source: AsyncIterable<Uint8Array>, expected?: MirrorBinaryInfo | null): Promise<void> {
     const path = assertVisibleBinaryPath(input);
+    // A stream can take seconds to consume. Remember its destination before
+    // reading any bytes, then recheck it after staging and folder creation.
+    const before = expected === undefined ? await this.inspectBinary(path) : expected;
     const bytes = await collectBinary(source);
     const slash = path.lastIndexOf("/");
     if (slash >= 0) await ensureFolder(this.vault, path.slice(0, slash));
+    const current = await this.inspectBinary(path);
+    if (current?.content_digest !== before?.content_digest || current?.size !== before?.size) {
+      throw new SyncError("sync_plan_stale", `${path} changed before it could be written. Review sync again.`);
+    }
     const existing = this.vault.getAbstractFileByPath(path);
     if (existing instanceof TFolder) throw new SyncError("mirror_path_collision", `A folder blocks the mirror file ${path}.`);
     this.assertActive();
@@ -1769,21 +1776,34 @@ export class ConnectSyncController {
     if (enrollment.collectionId !== profile.collectionId) {
       throw new SyncError("mirror_identity_conflict", "Connect approved a different collection. The existing mirror was not changed.");
     }
-    const nextProfile = profileFromEnrollment(enrollment, profile.selectiveSync);
-    if (oldState && enrollment.replicaId !== profile.replicaId) {
-      await this.stateStoreFor(nextProfile).write({
-        ...oldState,
-        replica_id: enrollment.replicaId,
-      });
-    }
-    await this.persistEnrollment(enrollment, profile.selectiveSync);
-    if (!copied && enrollment.replicaId !== profile.replicaId) {
-      this.clearCredentials(profile);
-      if ("clear" in oldStore && typeof oldStore.clear === "function") await oldStore.clear();
-    }
-    const status = await this.status();
-    if (!status) throw new SyncError("mirror_not_configured", "The reauthorized mirror profile could not be loaded.");
-    return status;
+    return this.withMirrorOperation(async () => {
+      // Browser approval can stay open while background sync advances or a
+      // disconnect retires this connection. Only migrate its latest checkpoint,
+      // under the same queue that protects all other mirror state operations.
+      const current = this.requireProfile();
+      if (current.collectionId !== profile.collectionId || current.replicaId !== profile.replicaId
+        || current.enrollmentId !== profile.enrollmentId) {
+        throw new SyncError("mirror_identity_conflict", "The connection changed during approval. The current mirror was not changed.");
+      }
+      const latestState = copied ? null : await oldStore.read();
+      if (latestState?.batch) {
+        throw new SyncError("mirror_recovery_required", "Resume the durable synchronization checkpoint before approving this vault again.");
+      }
+      const nextProfile = profileFromEnrollment(enrollment, current.selectiveSync);
+      if (latestState && enrollment.replicaId !== profile.replicaId) {
+        await this.stateStoreFor(nextProfile).write({
+          ...latestState,
+          replica_id: enrollment.replicaId,
+        });
+      }
+      await this.persistEnrollment(enrollment, current.selectiveSync);
+      if (!copied && enrollment.replicaId !== profile.replicaId) {
+        this.clearCredentials(profile);
+        if ("clear" in oldStore && typeof oldStore.clear === "function") await oldStore.clear();
+      }
+      const mirror = await this.createMirror();
+      return mirror.status();
+    });
   }
 
   async conflictComparison(
@@ -2017,32 +2037,55 @@ export class ConnectSyncController {
 
   private async writeConflictCopy(pathInput: string, document: string, label: string): Promise<string> {
     const path = safeMirrorPath(this.app.vault, pathInput);
-    const target = await this.conflictCopyPath(path, label);
-    await this.app.vault.create(target, document);
-    return target;
+    return this.createConflictCopy(path, new TextEncoder().encode(document).buffer, label);
   }
 
-  private async conflictCopyPath(path: string, label: string): Promise<string> {
+  private async createConflictCopy(path: string, document: ArrayBuffer, label: string): Promise<string> {
     const slash = path.lastIndexOf("/");
     const dot = path.lastIndexOf(".");
     const extension = dot > slash ? path.slice(dot) : "";
-    const base = extension ? path.slice(0, -extension.length) : path;
-    let target = `${base} (${label})${extension}`;
-    let suffix = 2;
-    while (this.app.vault.getAbstractFileByPath(target) || await this.app.vault.adapter.exists(target)) {
-      target = `${base} (${label} ${suffix})${extension}`;
-      suffix += 1;
+    const stem = extension ? path.slice(0, -extension.length) : path;
+    // A copy can itself conflict on another device. Keep the sibling names
+    // flat rather than producing copies of copies of copies.
+    const base = stem.replace(/(?: \((?:local|hosted) conflict copy(?: \d+)?\))+$/, "");
+    const bytes = new Uint8Array(document);
+    let suffix = 1;
+    while (true) {
+      const target = `${base} (${label}${suffix === 1 ? "" : ` ${suffix}`})${extension}`;
+      if (target === path) {
+        suffix += 1;
+        continue;
+      }
+      if (this.app.vault.getAbstractFileByPath(target) || await this.app.vault.adapter.exists(target)) {
+        // A stale decision may be retried after its copy was already saved.
+        // Reuse only byte-identical copies; never modify an older recovery file.
+        try {
+          const existing = new Uint8Array(await this.app.vault.adapter.readBinary(target));
+          if (existing.length === bytes.length && existing.every((byte, index) => byte === bytes[index])) return target;
+        } catch {
+          // Folders and unreadable copies still reserve their name.
+        }
+        suffix += 1;
+        continue;
+      }
+      try {
+        // Unlike adapter.copy, Vault.createBinary refuses an occupied target,
+        // including one that appears after the name check. Preserve exact bytes.
+        await this.app.vault.createBinary(target, document);
+        return target;
+      } catch (error) {
+        if (!await this.app.vault.adapter.exists(target)) throw error;
+        // A concurrent creator won this name. Inspect it before choosing again.
+      }
     }
-    return target;
   }
 
   async preserveConflictCopy(pathInput: string): Promise<string> {
     const path = safeMirrorPath(this.app.vault, pathInput);
     const existing = this.app.vault.getAbstractFileByPath(path);
     if (!(existing instanceof TFile)) throw new SyncError("mirror_conflict_copy_missing", `No local file exists at ${path}.`);
-    const target = await this.conflictCopyPath(path, "local conflict copy");
-    await this.app.vault.adapter.copy(path, target);
-    return target;
+    const document = await this.app.vault.adapter.readBinary(path);
+    return this.createConflictCopy(path, document, "local conflict copy");
   }
 
   async disconnect(removeSyncedFiles: boolean): Promise<DisconnectMirrorResult> {
@@ -2802,8 +2845,16 @@ export class ConnectSyncController {
       accessToken: this.readSecret("access", profile) ?? "",
       refreshCredential,
       accessTokenExpiresAt: profile.accessTokenExpiresAt,
-    });
-    await this.persistEnrollment(renewed, profile.selectiveSync);
+    }, { signal: this.lifetime.signal });
+    // Renewal may overlap a disconnect or a fresh browser enrollment. Never
+    // recreate the retired connection (or replace the new one's credentials).
+    const current = this.requireProfile();
+    if (current.collectionId !== profile.collectionId || current.replicaId !== profile.replicaId
+      || current.enrollmentId !== profile.enrollmentId) {
+      throw new SyncError("mirror_identity_conflict", "The connection changed while credentials were renewing. Retry with the current connection.");
+    }
+    // Settings can change independently while the request is in flight.
+    await this.persistEnrollment(renewed, current.selectiveSync);
     return renewed.accessToken;
   }
 

src/syncHistory.ts

@@ -95,7 +95,7 @@ export function historyFileFromReceipt(
  * A resumed batch already carries earlier receipts; they are the baseline.
  */
 export class ReceiptObservingStateStore implements MirrorStateStore {
-  private readonly seen = new Map<string, number>();
+  private readonly seen = new Map<string, { receipts: number; actions: Map<string, MirrorPlanAction> }>();
 
   constructor(
     private readonly inner: MirrorStateStore,
@@ -112,10 +112,17 @@ export class ReceiptObservingStateStore implements MirrorStateStore {
     if (!batch) return;
     const fingerprint = batch.plan.fingerprint;
     const previous = this.seen.get(fingerprint);
-    this.seen.set(fingerprint, batch.receipts.length);
-    if (previous === undefined) return;
-    for (const receipt of batch.receipts.slice(previous)) {
-      const action = batch.plan.actions.find((candidate) => candidate.action_id === receipt.action_id);
+    if (previous === undefined) {
+      this.seen.set(fingerprint, {
+        receipts: batch.receipts.length,
+        actions: new Map(batch.plan.actions.map((action) => [action.action_id, action])),
+      });
+      return;
+    }
+    const start = previous.receipts;
+    previous.receipts = batch.receipts.length;
+    for (const receipt of batch.receipts.slice(start)) {
+      const action = previous.actions.get(receipt.action_id);
       if (action) this.onReceipt(action, receipt);
     }
   }

src/syncMerge.ts

@@ -27,21 +27,23 @@ export function mergeDocuments(
   if (parts.every((part) => part === null)) {
     const merged = mergeLines(base, local, remote);
     return merged === null
-      ? { clean: false, reason: "Both versions changed the same lines." }
+      ? { clean: false, reason: "The line edits could not be merged safely." }
       : { clean: true, text: merged };
   }
   const [b, l, r] = parts;
   if (!b || !l || !r || b.open !== l.open || l.open !== r.open) {
     return { clean: false, reason: "Only one version has frontmatter." };
   }
+  const close = mergeValue(b.close, l.close, r.close);
+  if (close === null) return { clean: false, reason: "Both versions changed the frontmatter fence." };
   const frontmatter = mergeFrontmatter(b.frontmatter, l.frontmatter, r.frontmatter);
   if (frontmatter === null) return { clean: false, reason: "Both versions changed the same field." };
   const body = mergeLines(b.body, l.body, r.body);
-  if (body === null) return { clean: false, reason: "Both versions changed the same lines." };
-  if (frontmatter.length && !validFrontmatter(frontmatter.join("\n"))) {
+  if (body === null) return { clean: false, reason: "The line edits could not be merged safely." };
+  if (frontmatter.length && !validFrontmatter(frontmatter.join("\n") + "\n")) {
     return { clean: false, reason: "The combined frontmatter would not be valid YAML." };
   }
-  return { clean: true, text: [l.open, ...frontmatter, l.close].join("\n") + body };
+  return { clean: true, text: [l.open, ...frontmatter, close].join("\n") + body };
 }
 
 interface DocumentParts {
@@ -68,11 +70,22 @@ function splitDocument(document: string): DocumentParts | null {
 }
 
 function mergeLines(base: string, local: string, remote: string): string | null {
-  const regions = diff3Merge(local.split("\n"), base.split("\n"), remote.split("\n"), { excludeFalseConflicts: true });
+  if (local === remote || remote === base) return local;
+  if (local === base) return remote;
+  const b = base.split("\n");
+  const l = local.split("\n");
+  const r = remote.split("\n");
+  // node-diff3's LCS is quadratic in the worst case (especially repeated
+  // lines). Never freeze Obsidian's UI trying to merge a pathological note:
+  // the caller keeps both exact documents instead. Ordinary field/body edits
+  // above avoid diffing an unchanged body regardless of its size.
+  const work = b.length * (l.length + r.length);
+  if (work > 4_000_000) return null;
+  const regions = diff3Merge(l, b, r, { excludeFalseConflicts: true });
   const merged: string[] = [];
   for (const region of regions) {
     if (region.conflict) return null;
-    merged.push(...(region.ok ?? []));
+    for (const line of region.ok ?? []) merged.push(line);
   }
   return merged.join("\n");
 }
@@ -100,14 +113,21 @@ function fieldBlocks(lines: string[]): FieldBlocks | null {
       blocks.get(current)?.push(line);
     }
   }
-  return { order, blocks: new Map([...blocks].map(([key, value]) => [key, value.join("\n")])) };
+  return { order, blocks: new Map([...blocks].filter(([, value]) => value.length > 0).map(([key, value]) => [key, value.join("\n")])) };
 }
 
 function mergeFrontmatter(base: string[], local: string[], remote: string[]): string[] | null {
   const b = fieldBlocks(base);
   const l = fieldBlocks(local);
   const r = fieldBlocks(remote);
-  if (!b || !l || !r) {
+  // Field-wise merging cannot represent moves. A line merge either preserves
+  // the reordered text or conservatively keeps both versions.
+  const reordered = (ancestor: FieldBlocks, changed: FieldBlocks) => {
+    const common = new Set(ancestor.order.filter((key) => changed.blocks.has(key)));
+    return ancestor.order.filter((key) => common.has(key)).join("\n")
+      !== changed.order.filter((key) => common.has(key)).join("\n");
+  };
+  if (!b || !l || !r || reordered(b, l) || reordered(b, r)) {
     return mergeLines(base.join("\n"), local.join("\n"), remote.join("\n"))?.split("\n") ?? null;
   }
   const merged = new Map<string, string>();
@@ -123,11 +143,8 @@ function mergeFrontmatter(base: string[], local: string[], remote: string[]): st
     const before = r.order.slice(0, index).reverse().find((candidate) => order.includes(candidate));
     order.splice(before === undefined ? 0 : order.indexOf(before) + 1, 0, key);
   }
-  // An empty preamble is the absence of lines, not an empty one.
-  return order.flatMap((key) => {
-    const value = merged.get(key) ?? "";
-    return key === PREAMBLE && value === "" ? [] : value.split("\n");
-  });
+  // An absent preamble is different from one containing a single blank line.
+  return order.flatMap((key) => (merged.get(key) ?? "").split("\n"));
 }
 
 /** `undefined` is an absent field; `null` means both sides changed it differently. */

src/syncScheduler.ts

@@ -131,9 +131,12 @@ export class SyncScheduler {
         this.rerun = false;
         this.schedule(now);
       } else if (retryIn !== null) {
+        // A routine probe must not preempt the advertised failure backoff.
+        // Explicit edits, online and foreground events can still request sooner.
         this.schedule(now + retryIn);
+      } else {
+        this.schedule(now + (this.visible ? this.timing.probeVisibleMs : this.timing.probeHiddenMs));
       }
-      this.schedule(now + (this.visible ? this.timing.probeVisibleMs : this.timing.probeHiddenMs));
     }
   }
 

src/syncSession.ts

@@ -87,6 +87,8 @@ const EMPTY_STATE: SyncSessionState = {
 export class SyncSession {
   private current: SyncSessionState = { ...EMPTY_STATE };
   private readonly listeners = new Set<() => void>();
+  private safetyPreview: MdbaseSyncPreview | null = null;
+  private previewSafety: SyncPlanSafety | null = null;
 
   /** The last failure pinned to history, so automatic retries do not pin it again. */
   private pinnedFailure: string | null = null;
@@ -208,10 +210,16 @@ export class SyncSession {
     const result = await this.exclusive(async (): Promise<SyncNowResult> => {
       let applied = false;
       for (let round = 0; round < MAX_SYNC_ROUNDS; round += 1) {
+        if (this.current.paused) return "paused";
         const preview = await this.loadPreview();
+        // Inspection is not a transfer and cannot be cancelled by cancelSync().
+        // Honour Stop before starting any writes once that inspection returns.
+        if (this.current.paused) return "paused";
         if (!preview) return "failed";
         if (!preview.plan.actions.length) {
-          if (await this.settleConflicts()) continue;
+          const settled = await this.settleConflicts();
+          if (this.current.paused) return "paused";
+          if (settled) continue;
           if (!applied && !options.quiet) this.update({ message: "Already up to date." });
           return applied ? "applied" : "up_to_date";
         }
@@ -224,7 +232,10 @@ export class SyncSession {
         if (outcome === "cancelled") return "paused";
         if (outcome === "failed") return "failed";
         applied = true;
-        if (await this.settleConflicts()) continue;
+        if (this.current.paused) return "paused";
+        const settled = await this.settleConflicts();
+        if (this.current.paused) return "paused";
+        if (settled) continue;
         if (!this.current.preview) return "applied";
       }
       return applied ? "applied" : "up_to_date";
@@ -267,7 +278,14 @@ export class SyncSession {
   }
 
   safety(): SyncPlanSafety | null {
-    return this.current.preview ? syncPlanSafety(this.current.preview) : null;
+    const preview = this.current.preview;
+    // Engine previews are replaced, not mutated. Transfer progress can notify
+    // several surfaces per file; none needs to rescan the same consent policy.
+    if (preview !== this.safetyPreview) {
+      this.safetyPreview = preview;
+      this.previewSafety = preview ? syncPlanSafety(preview) : null;
+    }
+    return this.previewSafety;
   }
 
   /** Stop now and keep automatic sync off until the person resumes it. */