Add offline body-CRDT sync E2E coverage

apps/desktop/src/main/database/client.ts

@@ -97,6 +97,10 @@ export function getIndexDatabase(): IndexDb {
   return indexDb
 }
 
+export function isIndexDatabaseInitialized(): boolean {
+  return indexDb !== null
+}
+
 /**
  * Get the raw better-sqlite3 connection for the index database.
  * Used for direct sqlite-vec queries on vec_notes virtual table.

apps/desktop/src/main/database/queries/notes/index.ts

@@ -12,6 +12,7 @@ export {
   bulkInsertNotes,
   clearNoteCache,
   getAllNoteIds,
+  getAllCrdtNoteIds,
   getNotesModifiedAfter,
   type ListNotesOptions
 } from './note-crud'

apps/desktop/src/main/database/queries/notes/note-crud.ts

@@ -198,6 +198,15 @@ export function getAllNoteIds(db: IndexDb): string[] {
     .map((r) => r.id)
 }
 
+export function getAllCrdtNoteIds(db: IndexDb): string[] {
+  return db
+    .select({ id: noteCache.id })
+    .from(noteCache)
+    .where(eq(noteCache.fileType, 'markdown'))
+    .all()
+    .map((r) => r.id)
+}
+
 export function getNotesModifiedAfter(db: IndexDb, date: string): NoteCache[] {
   return db
     .select()

apps/desktop/src/main/index.phase2.test.ts

@@ -3,6 +3,8 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 const appOnMock = vi.fn()
 const whenReadyMock = vi.fn(() => new Promise<void>(() => {}))
 const requestSingleInstanceLockMock = vi.fn(() => true)
+const getPathMock = vi.fn((name: string) => `/mock/${name}`)
+const setPathMock = vi.fn()
 const dotenvConfigMock = vi.fn(() => ({ error: undefined }))
 
 vi.mock('dotenv', () => ({
@@ -39,6 +41,8 @@ vi.mock('electron', () => ({
   app: {
     isPackaged: false,
     getAppPath: vi.fn(() => '/mock/app'),
+    getPath: getPathMock,
+    setPath: setPathMock,
     requestSingleInstanceLock: requestSingleInstanceLockMock,
     on: appOnMock,
     whenReady: whenReadyMock,
@@ -90,6 +94,7 @@ describe('main index phase2 exports', () => {
   beforeEach(() => {
     vi.resetModules()
     vi.clearAllMocks()
+    getPathMock.mockImplementation((name: string) => `/mock/${name}`)
     process.env = { ...ORIGINAL_ENV }
   })
 
@@ -125,6 +130,15 @@ describe('main index phase2 exports', () => {
     expect(module.envConfig.embeddingModel).toBe('embed-test')
   })
 
+  it('skips the single-instance lock for multi-device test launches', async () => {
+    process.env.NODE_ENV = 'test'
+    process.env.MEMRY_DEVICE = 'A'
+
+    await importMainModule()
+
+    expect(requestSingleInstanceLockMock).not.toHaveBeenCalled()
+  })
+
   it('registerOAuthState schedules expiry cleanup at 10 minutes', async () => {
     vi.useFakeTimers()
 

apps/desktop/src/main/index.ts

@@ -27,6 +27,7 @@ import { startSnoozeScheduler, stopSnoozeScheduler, checkDueItemsOnStartup } fro
 import { stopVoiceModel } from './inbox/voice-model'
 import { startReminderScheduler, stopReminderScheduler } from './lib/reminders'
 import { log, createLogger, disableConsoleTransport } from './lib/logger'
+import { registerTestHooks } from './test-hooks'
 import {
   computeSpkiHashFromPem,
   isPinningDisabled,
@@ -58,6 +59,8 @@ const quickCaptureLog = createLogger('QuickCapture')
 const shutdownLog = createLogger('Shutdown')
 const deepLinkLog = createLogger('DeepLink')
 
+registerTestHooks()
+
 for (const signal of ['SIGINT', 'SIGTERM'] as const) {
   process.on(signal, () => {
     disableConsoleTransport()
@@ -314,17 +317,24 @@ function handleDeepLink(url: string): void {
   }
 }
 
-// Windows/Linux: deep links arrive via second-instance event
-const gotTheLock = app.requestSingleInstanceLock()
-if (!gotTheLock) {
-  app.quit()
-} else {
-  app.on('second-instance', (_event, commandLine) => {
-    const deepLinkUrl = commandLine.find((arg) => arg.startsWith('memry://'))
-    if (deepLinkUrl) {
-      handleDeepLink(deepLinkUrl)
-    }
-  })
+const allowMultiInstanceForDeviceTests =
+  process.env.NODE_ENV === 'test' && typeof process.env.MEMRY_DEVICE === 'string'
+
+// Windows/Linux: deep links arrive via second-instance event.
+// Device-scoped E2E runs need two Electron instances side by side, so skip the
+// process-wide lock only for that test harness path.
+if (!allowMultiInstanceForDeviceTests) {
+  const gotTheLock = app.requestSingleInstanceLock()
+  if (!gotTheLock) {
+    app.quit()
+  } else {
+    app.on('second-instance', (_event, commandLine) => {
+      const deepLinkUrl = commandLine.find((arg) => arg.startsWith('memry://'))
+      if (deepLinkUrl) {
+        handleDeepLink(deepLinkUrl)
+      }
+    })
+  }
 }
 
 // This method will be called when Electron has finished

apps/desktop/src/main/sync/crdt-provider.ts

@@ -46,6 +46,7 @@ interface ActiveDoc {
   doc: Y.Doc
   windowIds: Set<number>
   accumulatedBytes: number
+  pendingSnapshotBytes: number
   lastEncodedSize: number
   lastSizeCheckAt: number
   closing?: boolean
@@ -99,6 +100,9 @@ export class CrdtProvider {
     const existing = this.docs.get(noteId)
     if (existing && !existing.closing) {
       if (windowId) existing.windowIds.add(windowId)
+      if (!options?.skipSeed) {
+        await this.seedFromMarkdown(noteId, existing.doc)
+      }
       return existing.doc
     }
 
@@ -107,6 +111,9 @@ export class CrdtProvider {
       const doc = await pending
       const entry = this.docs.get(noteId)
       if (entry && windowId) entry.windowIds.add(windowId)
+      if (!options?.skipSeed) {
+        await this.seedFromMarkdown(noteId, doc)
+      }
       return doc
     }
 
@@ -146,6 +153,7 @@ export class CrdtProvider {
       doc,
       windowIds: new Set(windowId ? [windowId] : []),
       accumulatedBytes: 0,
+      pendingSnapshotBytes: 0,
       lastEncodedSize: 0,
       lastSizeCheckAt: 0
     }
@@ -171,11 +179,12 @@ export class CrdtProvider {
 
     this.flushNetworkBroadcast(noteId)
 
-    if (this.snapshotPushFn && entry.accumulatedBytes > 0) {
+    if (this.snapshotPushFn && entry.pendingSnapshotBytes > 0) {
       const state = Y.encodeStateAsUpdate(entry.doc)
       await this.snapshotPushFn(noteId, state).catch((err) => {
         log.warn('Failed to push snapshot on close', { noteId, error: err })
       })
+      entry.pendingSnapshotBytes = 0
     }
 
     await this.flushDoc(noteId).catch((err) => {
@@ -325,10 +334,12 @@ export class CrdtProvider {
 
     let pushed = 0
     for (const [noteId, entry] of this.docs) {
+      if (entry.pendingSnapshotBytes <= 0) continue
       try {
         const state = Y.encodeStateAsUpdate(entry.doc)
         await this.snapshotPushFn(noteId, state)
         entry.accumulatedBytes = 0
+        entry.pendingSnapshotBytes = 0
         pushed++
         log.info('Pushed server snapshot', { noteId, size: state.byteLength })
       } catch (err) {
@@ -362,7 +373,10 @@ export class CrdtProvider {
       }
 
       // Reset accumulatedBytes BEFORE push so close() won't fire a duplicate push
-      if (entry) entry.accumulatedBytes = 0
+      if (entry) {
+        entry.accumulatedBytes = 0
+        entry.pendingSnapshotBytes = 0
+      }
 
       await this.snapshotPushFn(noteId, state)
       log.info('Pushed snapshot for note', { noteId, size: state.byteLength })
@@ -488,6 +502,9 @@ export class CrdtProvider {
     if (!entry) return
 
     entry.accumulatedBytes += update.byteLength
+    if (origin !== ORIGIN_NETWORK) {
+      entry.pendingSnapshotBytes += update.byteLength
+    }
 
     if (isIpcOrigin(origin)) {
       this.broadcastToWindows(noteId, update, 'ipc', origin.windowId)
@@ -506,6 +523,9 @@ export class CrdtProvider {
 
     if (origin === ORIGIN_NETWORK) {
       recordNetworkUpdate(noteId)
+    }
+
+    if (origin === ORIGIN_NETWORK || isIpcOrigin(origin)) {
       scheduleWriteback(noteId, entry.doc)
     }
   }
@@ -619,8 +639,9 @@ export class CrdtProvider {
     this.compactionBuffers.set(noteId, [])
 
     try {
-      if (this.snapshotPushFn) {
+      if (this.snapshotPushFn && entry.pendingSnapshotBytes > 0) {
         await this.snapshotPushFn(noteId, result.compacted)
+        entry.pendingSnapshotBytes = 0
       }
 
       if (this.persistence) {

apps/desktop/src/main/sync/crdt-queue.ts

@@ -70,6 +70,10 @@ export class CrdtUpdateQueue {
     return count
   }
 
+  getOutstandingCount(): number {
+    return this.getPendingCount() + this.flushingNotes.size
+  }
+
   private flushAll(): void {
     for (const noteId of this.buffers.keys()) {
       this.flushNote(noteId)

apps/desktop/src/main/sync/crdt-writeback.ts

@@ -36,6 +36,31 @@ const pendingTimers = new Map<string, ReturnType<typeof setTimeout>>()
 const ignoredWrites = new Map<string, number>()
 const lastNetworkUpdateMs = new Map<string, number>()
 
+interface WritebackDebugState {
+  pending: boolean
+  scheduledCount: number
+  performedCount: number
+  lastMarkdown: string | null
+  lastError: string | null
+}
+
+const debugState = new Map<string, WritebackDebugState>()
+
+function updateDebugState(noteId: string, patch: Partial<WritebackDebugState>): void {
+  const current = debugState.get(noteId) ?? {
+    pending: false,
+    scheduledCount: 0,
+    performedCount: 0,
+    lastMarkdown: null,
+    lastError: null
+  }
+  debugState.set(noteId, { ...current, ...patch })
+}
+
+export function getWritebackDebugState(noteId: string): WritebackDebugState | null {
+  return debugState.get(noteId) ?? null
+}
+
 function isJournalId(noteId: string): boolean {
   return noteId.startsWith('j') && /^j\d{4}-\d{2}-\d{2}$/.test(noteId)
 }
@@ -87,10 +112,19 @@ function emitToRenderer(channel: string, data: unknown): void {
 export function scheduleWriteback(noteId: string, doc: Y.Doc): void {
   const existing = pendingTimers.get(noteId)
   if (existing) clearTimeout(existing)
+  updateDebugState(noteId, {
+    pending: true,
+    scheduledCount: (debugState.get(noteId)?.scheduledCount ?? 0) + 1,
+    lastError: null
+  })
 
   const timer = setTimeout(() => {
     pendingTimers.delete(noteId)
     performWriteback(noteId, doc).catch((err) => {
+      updateDebugState(noteId, {
+        pending: false,
+        lastError: err instanceof Error ? err.message : String(err)
+      })
       log.error('Write-back failed', { noteId, error: err })
       emitToRenderer('sync:write-back-failed', { noteId })
     })
@@ -108,6 +142,12 @@ export function cancelPendingWritebacks(): void {
 
 async function performWriteback(noteId: string, doc: Y.Doc): Promise<void> {
   const markdown = await yDocToMarkdown(doc)
+  updateDebugState(noteId, {
+    pending: false,
+    performedCount: (debugState.get(noteId)?.performedCount ?? 0) + 1,
+    lastMarkdown: markdown,
+    lastError: null
+  })
   if (markdown === null) {
     log.warn('Conversion returned null, keeping stale file', { noteId })
     return

apps/desktop/src/main/sync/engine.test.ts

@@ -1651,7 +1651,7 @@ describe('SyncEngine', () => {
   })
 
   describe('#given WS reconnect #when handleWsConnected fires', () => {
-    it('#then schedules pull', async () => {
+    it('#then schedules pull and catches up open CRDT notes', async () => {
       const getServerMock = vi.fn().mockResolvedValue({
         items: [],
         deleted: [],
@@ -1660,24 +1660,36 @@ describe('SyncEngine', () => {
       })
       vi.spyOn(await import('./http-client'), 'getFromServer').mockImplementation(getServerMock)
 
-      const deps = createMockDeps(testDb)
+      const deps = createMockDeps(testDb, {
+        crdtProvider: {
+          getOpenNoteIds: vi.fn().mockReturnValue(['note-1', 'note-2'])
+        } as SyncEngineDeps['crdtProvider']
+      })
       const engine = new SyncEngine(deps)
       await engine.start()
 
       getServerMock.mockClear()
 
+      const pullCrdtForNote = vi.fn().mockResolvedValue(undefined)
+      ;(engine as unknown as { crdtSync: { pullCrdtForNote: typeof pullCrdtForNote } }).crdtSync = {
+        pullCrdtForNote
+      }
+
       const pullDone = new Promise<void>((resolve) => {
-        const origPull = engine.pull.bind(engine)
-        engine.pull = async () => {
-          await origPull()
-          resolve()
-        }
+        pullCrdtForNote.mockImplementation(async (noteId: string) => {
+          if (noteId === 'note-2') {
+            resolve()
+          }
+        })
       })
 
       deps.ws.emit('connected')
 
       await pullDone
       expect(getServerMock).toHaveBeenCalled()
+      expect(pullCrdtForNote).toHaveBeenCalledTimes(2)
+      expect(pullCrdtForNote).toHaveBeenCalledWith('note-1')
+      expect(pullCrdtForNote).toHaveBeenCalledWith('note-2')
 
       await engine.stop()
       vi.restoreAllMocks()

apps/desktop/src/main/sync/engine.ts

@@ -548,7 +548,14 @@ export class SyncEngine extends EventEmitter {
 
   private handleWsConnected = (): void => {
     if (!this.stateManager.isPaused()) {
-      this.scheduleSync(() => this.pull())
+      this.scheduleSync(async () => {
+        await this.pull()
+
+        const openNoteIds = this.ctx.deps.crdtProvider?.getOpenNoteIds() ?? []
+        for (const noteId of openNoteIds) {
+          await this.crdtSync.pullCrdtForNote(noteId)
+        }
+      })
     }
   }
 }