Harden local-only queue wipe sync

Author: arul28

apps/desktop/src/main/services/state/kvDb.sync.test.ts

@@ -22,6 +22,26 @@ function makeDbPath(prefix: string): string {
   return path.join(root, ".ade", "kv.sqlite");
 }
 
+function packedTextPrimaryKey(text: string): { type: "bytes"; base64: string } {
+  const textBytes = Buffer.from(text, "utf8");
+  return {
+    type: "bytes",
+    base64: Buffer.concat([Buffer.from([0x01, 0x0b, textBytes.length]), textBytes]).toString("base64"),
+  };
+}
+
+function syncPrimaryKeyMatchesText(value: unknown, text: string): boolean {
+  if (value === text) return true;
+  return Boolean(
+    value
+    && typeof value === "object"
+    && "type" in value
+    && (value as { type?: unknown }).type === "bytes"
+    && "base64" in value
+    && (value as { base64?: unknown }).base64 === packedTextPrimaryKey(text).base64,
+  );
+}
+
 describe.skipIf(!isCrsqliteAvailable())("kvDb sync foundation", () => {
   it("persists a stable local site id and marks CRR tables", async () => {
     const dbPath = makeDbPath("ade-kvdb-sync-site-");
@@ -181,6 +201,7 @@ describe.skipIf(!isCrsqliteAvailable())("kvDb sync foundation", () => {
   it("does not replicate queue_landing_state overhaul wipe deletes to synced peers", async () => {
     const dbPathA = makeDbPath("ade-kvdb-sync-queue-wipe-a-");
     const dbA = await openKvDb(dbPathA, createLogger() as any);
+    const wipeMarker = "queue_landing_state.wiped_for_stacked_overhaul.v1";
     const projectId = "project-queue-wipe";
     const groupId = "group-queue-wipe";
     const queueId = "queue-wipe-1";
@@ -211,7 +232,7 @@ describe.skipIf(!isCrsqliteAvailable())("kvDb sync foundation", () => {
     ).toBe(queueId);
 
     const versionBeforeWipe = dbA.sync.getDbVersion();
-    dbA.run("delete from kv where key = ?", ["queue_landing_state.wiped_for_stacked_overhaul.v1"]);
+    dbA.run("delete from kv where key = ?", [wipeMarker]);
     dbA.close();
 
     const dbAReopened = await openKvDb(dbPathA, createLogger() as any);
@@ -221,12 +242,36 @@ describe.skipIf(!isCrsqliteAvailable())("kvDb sync foundation", () => {
 
     const wipeChanges = dbAReopened.sync.exportChangesSince(versionBeforeWipe);
     expect(wipeChanges.some((change) => change.table === "queue_landing_state")).toBe(false);
+    expect(wipeChanges.some((change) => change.table === "kv" && syncPrimaryKeyMatchesText(change.pk, wipeMarker))).toBe(false);
 
     dbB.sync.applyChanges(wipeChanges);
     expect(
       dbB.get<{ id: string }>("select id from queue_landing_state where id = ?", [queueId])?.id,
     ).toBe(queueId);
 
+    const markerValueBeforeApply = dbB.get<{ value: string }>(
+      "select value from kv where key = ?",
+      [wipeMarker],
+    )?.value ?? null;
+    const versionBeforeMarkerApply = dbB.sync.getDbVersion();
+    const markerApplyResult = dbB.sync.applyChanges([{
+      table: "kv",
+      pk: packedTextPrimaryKey(wipeMarker),
+      cid: "value",
+      val: "remote-marker-should-not-apply",
+      col_version: 999,
+      db_version: versionBeforeMarkerApply + 1,
+      site_id: "f".repeat(32),
+      cl: 1,
+      seq: 1,
+    }]);
+    expect(markerApplyResult.appliedCount).toBe(0);
+    expect(markerApplyResult.touchedTables).toEqual([]);
+    expect(dbB.sync.getDbVersion()).toBe(versionBeforeMarkerApply);
+    expect(
+      dbB.get<{ value: string }>("select value from kv where key = ?", [wipeMarker])?.value ?? null,
+    ).toBe(markerValueBeforeApply);
+
     dbAReopened.close();
     dbB.close();
   });

apps/desktop/src/main/services/state/kvDb.ts

@@ -553,10 +553,40 @@ function deleteAllRowsWithoutCrrReplication(
 
 const QUEUE_OVERHAUL_WIPE_MARKER = "queue_landing_state.wiped_for_stacked_overhaul.v1";
 
+function rawCrsqlPrimaryKeyMatchesText(value: unknown, text: string): boolean {
+  if (value === text) return true;
+  if (!(value instanceof Uint8Array)) return false;
+
+  const packed = packedCrsqlPrimaryKey(text);
+  return isSyncScalarBytes(packed)
+    && Buffer.from(value).equals(Buffer.from(packed.base64, "base64"));
+}
+
+function syncScalarPrimaryKeyMatchesText(value: SyncScalar, text: string): boolean {
+  if (value === text) return true;
+
+  const packed = packedCrsqlPrimaryKey(text);
+  return isSyncScalarBytes(value)
+    && isSyncScalarBytes(packed)
+    && value.base64 === packed.base64;
+}
+
+function isLocalOnlyQueueWipeMarkerRawChange(change: { table_name: string; pk: unknown }): boolean {
+  return change.table_name === "kv"
+    && rawCrsqlPrimaryKeyMatchesText(change.pk, QUEUE_OVERHAUL_WIPE_MARKER);
+}
+
+function isLocalOnlyQueueWipeMarkerChange(change: CrsqlChangeRow): boolean {
+  return change.table === "kv"
+    && syncScalarPrimaryKeyMatchesText(change.pk, QUEUE_OVERHAUL_WIPE_MARKER);
+}
+
 /**
  * One-shot local wipe of legacy queue_landing_state on upgrade to the stacked-PR
- * queue overhaul. Must run after migrations and before ensureCrrTables so deletes
- * do not replicate to peers that have not upgraded yet.
+ * queue overhaul. Must run after migrations and before ensureCrrTables so queue
+ * deletes do not replicate to peers that have not upgraded yet. The marker is
+ * stored in kv for compatibility with the original migration, but filtered from
+ * CRDT import/export because it records local upgrade work, not shared state.
  */
 function wipeQueueLandingStateForStackedOverhaulIfNeeded(db: DatabaseSyncType, logger?: Logger): void {
   try {
@@ -2986,17 +3016,19 @@ export async function openKvDb(dbPath: string, logger: Logger): Promise<AdeDb> {
         [version]
       );
 
-      return rows.map((row) => ({
-        table: row.table_name,
-        pk: encodeSyncScalar(row.pk),
-        cid: row.cid,
-        val: encodeSyncScalar(row.val),
-        col_version: Number(row.col_version),
-        db_version: Number(row.db_version),
-        site_id: Buffer.from(row.site_id).toString("hex"),
-        cl: Number(row.cl),
-        seq: Number(row.seq),
-      }));
+      return rows
+        .filter((row) => !isLocalOnlyQueueWipeMarkerRawChange(row))
+        .map((row) => ({
+          table: row.table_name,
+          pk: encodeSyncScalar(row.pk),
+          cid: row.cid,
+          val: encodeSyncScalar(row.val),
+          col_version: Number(row.col_version),
+          db_version: Number(row.db_version),
+          site_id: Buffer.from(row.site_id).toString("hex"),
+          cl: Number(row.cl),
+          seq: Number(row.seq),
+        }));
     },
     applyChanges: (changes: CrsqlChangeRow[]) => {
       if (!crsqliteLoaded) return { appliedCount: 0, dbVersion: 0, touchedTables: [], rebuiltFts: false };
@@ -3005,6 +3037,7 @@ export async function openKvDb(dbPath: string, logger: Logger): Promise<AdeDb> {
       runStatement(db, "begin");
       try {
         for (const rawChange of changes) {
+          if (isLocalOnlyQueueWipeMarkerChange(rawChange)) continue;
           // Skip changes for tables that no longer exist in the schema
           // (e.g. unified_memories removed in #329).
           if (!rawHasTable(db, rawChange.table)) continue;

apps/desktop/src/renderer/components/files/FilesPage.test.tsx

@@ -836,6 +836,74 @@ describe("FilesPage", () => {
     }
   });
 
+  it("preserves oversized dirty tabs when switching lane scopes", async () => {
+    const laneA = "lane-large-a";
+    const laneB = "lane-large-b";
+    useAppStore.setState({
+      selectedLaneId: laneA,
+      lanes: [
+        { id: laneA, name: "Large A", branchRef: "refs/heads/large-a" },
+        { id: laneB, name: "Large B", branchRef: "refs/heads/large-b" },
+      ] as any,
+    });
+    vi.mocked(window.ade.files.listWorkspaces).mockResolvedValue([
+      {
+        id: "primary",
+        kind: "primary",
+        laneId: null,
+        name: "ADE",
+        branchRef: "refs/heads/main",
+        rootPath: projectRoot,
+        isReadOnlyByDefault: false,
+      },
+      {
+        id: "lane-large-a-ws",
+        kind: "worktree",
+        laneId: laneA,
+        name: "Large A",
+        branchRef: "refs/heads/large-a",
+        rootPath: `${projectRoot}/.ade/worktrees/large-a`,
+        isReadOnlyByDefault: false,
+      },
+      {
+        id: "lane-large-b-ws",
+        kind: "worktree",
+        laneId: laneB,
+        name: "Large B",
+        branchRef: "refs/heads/large-b",
+        rootPath: `${projectRoot}/.ade/worktrees/large-b`,
+        isReadOnlyByDefault: false,
+      },
+    ]);
+
+    renderFilesPage({
+      openFilePath: "src/index.ts",
+    });
+
+    await waitForEditorText("value = 1");
+
+    const oversizedDirtyContent = `dirty-start\n${"x".repeat(8 * 1024 * 1024 + 1)}\ndirty-end`;
+    act(() => {
+      latestMockEditor?.setValue(oversizedDirtyContent);
+    });
+    expect(latestMockEditor?.getValue().length).toBe(oversizedDirtyContent.length);
+
+    act(() => {
+      useAppStore.setState({ selectedLaneId: laneB });
+    });
+    await waitFor(() => {
+      expect(screen.getByText(/OPEN A FILE TO START EDITING/i)).toBeTruthy();
+    });
+
+    act(() => {
+      useAppStore.setState({ selectedLaneId: laneA });
+    });
+    await waitFor(() => {
+      expect(latestMockEditor?.getValue().length).toBe(oversizedDirtyContent.length);
+      expect(latestMockEditor?.getValue().endsWith("dirty-end")).toBe(true);
+    });
+  });
+
   it("treats Windows workspace paths case-insensitively for open tabs and watcher events", async () => {
     projectRoot = "C:/Repo";
     resetStore();

apps/ios/ADE/Resources/DatabaseBootstrap.sql

@@ -817,8 +817,6 @@ alter table queue_landing_state add column wait_reason text;
 
 alter table queue_landing_state add column updated_at text;
 
-delete from queue_landing_state;
-
 create table if not exists rebase_dismissed (
       lane_id text not null,
       project_id text not null,