fix(sync): preserve pull cursor on lock conflict

Propagate retryable import lock conflicts through CLI and Desktop pull adapters so scheduled pulls defer without advancing past unimported messages.

apps/cli/src/sync/adapters.test.ts

@@ -4,7 +4,14 @@ import os from 'node:os'
 import path from 'node:path'
 import test from 'node:test'
 import type { PathProvider } from '@openchatlab/core'
-import { DatabaseManager, DataDirCompatibilityError, raiseDataDirMinRuntimeVersion } from '@openchatlab/node-runtime'
+import {
+  DatabaseManager,
+  DataDirCompatibilityError,
+  IMPORT_IN_PROGRESS_ERROR_KEY,
+  raiseDataDirMinRuntimeVersion,
+  withDataDirImportLock,
+} from '@openchatlab/node-runtime'
+import { streamImport } from '../import/stream-import'
 import { DirectImporter } from './adapters'
 
 const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node')
@@ -57,3 +64,40 @@ test('DirectImporter.sessionExists preserves session DBs when compatibility gate
   )
   assert.equal(fs.existsSync(dbPath), true)
 })
+
+test('DirectImporter marks an incremental import lock conflict as retryable', async (t) => {
+  const root = makeTempDir()
+  const pathProvider = createPathProvider(root)
+  const manager = new DatabaseManager(pathProvider, {
+    nativeBinding,
+    runtime: { version: '0.30.2', kind: 'cli' },
+  })
+  t.after(() => {
+    manager.closeAll()
+    fs.rmSync(root, { recursive: true, force: true })
+  })
+
+  const chatFile = path.join(root, 'chatlab.json')
+  fs.writeFileSync(
+    chatFile,
+    JSON.stringify({
+      chatlab: { version: '0.0.2', exportedAt: 100 },
+      meta: { name: 'Test Session', platform: 'instagram', type: 'private', ownerId: 'u1' },
+      members: [{ platformId: 'u1', accountName: 'Alice' }],
+      messages: [{ sender: 'u1', accountName: 'Alice', timestamp: 101, type: 0, content: 'hi' }],
+    }),
+    'utf-8'
+  )
+
+  const seeded = await streamImport(manager, chatFile, { sessionId: 'local_session' })
+  assert.equal(seeded.success, true)
+
+  const importer = new DirectImporter(manager)
+  const result = await withDataDirImportLock(manager.getUserDataDir(), () =>
+    importer.importFile(chatFile, 'local_session', 'external_session')
+  )
+
+  assert.equal(result.success, false)
+  assert.equal(result.error, IMPORT_IN_PROGRESS_ERROR_KEY)
+  assert.equal(result.retryable, true)
+})

apps/cli/src/sync/adapters.ts

@@ -14,7 +14,7 @@ import type { HttpFetcher, DataImporter, SyncNotifier, ImportResult, FetchParams
 import { NOOP_LOGGER } from '@openchatlab/sync'
 import { buildPullUrl } from '@openchatlab/sync'
 import type { DatabaseManager } from '@openchatlab/node-runtime'
-import { DataDirCompatibilityError } from '@openchatlab/node-runtime'
+import { DataDirCompatibilityError, IMPORT_IN_PROGRESS_ERROR_KEY } from '@openchatlab/node-runtime'
 import { streamImport, incrementalImport } from '../import/stream-import'
 
 function getTempFilePath(ext: string): string {
@@ -120,6 +120,10 @@ export class DirectImporter implements DataImporter {
       if (result.error === 'error.session_not_found' || result.error?.includes('no such table')) {
         return { success: false, newMessageCount: 0, sessionId, needFullResync: true }
       }
+      if (result.error === IMPORT_IN_PROGRESS_ERROR_KEY) {
+        this.logger.info(`[DirectImporter] Incremental import deferred: another import is in progress`)
+        return { success: false, newMessageCount: 0, sessionId, error: result.error, retryable: true }
+      }
 
       return { success: false, newMessageCount: 0, sessionId, error: result.error }
     } catch (err: any) {
@@ -137,6 +141,10 @@ export class DirectImporter implements DataImporter {
         this.logger.info(`[DirectImporter] Full import OK: +${newMessageCount} messages`)
         return { success: true, newMessageCount, sessionId: result.sessionId ?? externalId }
       }
+      if (result.error === IMPORT_IN_PROGRESS_ERROR_KEY) {
+        this.logger.info(`[DirectImporter] Full import deferred: another import is in progress`)
+        return { success: false, newMessageCount: 0, error: result.error, retryable: true }
+      }
 
       return { success: false, newMessageCount: 0, error: result.error }
     } catch (err: any) {

apps/desktop/main/api/adapters.ts

@@ -15,6 +15,7 @@ import { openDatabase } from '../database/core'
 import * as worker from '../worker/workerManager'
 import { getPathProvider } from '../path-context'
 import { assertDesktopDataDirCompatible, getDesktopAppVersion } from '../runtime-compat'
+import { IMPORT_IN_PROGRESS_ERROR_KEY } from '@openchatlab/node-runtime'
 import type { HttpFetcher, DataImporter, SyncNotifier, ImportResult, FetchParams, SyncLogger } from '@openchatlab/sync'
 import { buildPullUrl, NOOP_LOGGER } from '@openchatlab/sync'
 
@@ -136,6 +137,10 @@ export class WorkerImporter implements DataImporter {
       this.logger.warn(`[Pull] Session ${sessionId} not found or schema invalid, need full resync`)
       return { success: false, newMessageCount: 0, sessionId, needFullResync: true }
     }
+    if (result.error === IMPORT_IN_PROGRESS_ERROR_KEY) {
+      this.logger.info(`[Pull] Incremental import deferred: another import is in progress`)
+      return { success: false, newMessageCount: 0, sessionId, error: result.error, retryable: true }
+    }
     this.logger.error(`[Pull] Incremental import failed: ${result.error}`)
     return { success: false, newMessageCount: 0, sessionId, error: result.error }
   }
@@ -148,6 +153,10 @@ export class WorkerImporter implements DataImporter {
       this.logger.info(`[Pull] streamImport OK: session=${result.sessionId}, messages=${msgCount}`)
       return { success: true, newMessageCount: msgCount, sessionId: result.sessionId }
     }
+    if (result.error === IMPORT_IN_PROGRESS_ERROR_KEY) {
+      this.logger.info(`[Pull] Full import deferred: another import is in progress`)
+      return { success: false, newMessageCount: 0, error: result.error, retryable: true }
+    }
     this.logger.error(`[Pull] streamImport failed: ${result.error}`)
     return { success: false, newMessageCount: 0, error: result.error }
   }

packages/sync/src/pull-engine.test.ts

@@ -215,6 +215,75 @@ describe('PullEngine', () => {
     assert.deepEqual(checkedSessionIds, ['local_session_b'])
   })
 
+  it('keeps the pull cursor stable when an import is deferred by another writer', async () => {
+    const session = createSession()
+    const dataSource = createDataSource()
+    dataSource.sessions = [session]
+    const page = writeTempJson({
+      chatlab: { version: '0.0.2', exportedAt: 100 },
+      meta: { name: 'Test Session', platform: 'test', type: 'group' },
+      members: [{ platformId: 'u1', accountName: 'Alice' }],
+      messages: [{ sender: 'u1', timestamp: 101, type: 0, content: 'hi' }],
+      sync: { hasMore: false, nextSince: 101 },
+    })
+    const sessionUpdates: Array<{ sessionId: string; updates: Partial<ImportSession> }> = []
+    const engine = createEngine({
+      files: [page],
+      dataSource,
+      sessionUpdates,
+      importResult: {
+        success: false,
+        newMessageCount: 0,
+        error: 'error.import_in_progress',
+        retryable: true,
+      },
+    })
+
+    const result = await withImmediateTimers(() => engine.executePullSession(dataSource.id, dataSource, session))
+
+    assert.equal(result.success, false)
+    assert.equal(result.error, 'error.import_in_progress')
+    assert.deepEqual(sessionUpdates, [])
+  })
+
+  it('keeps the pull cursor stable when a retry-page import is deferred', async () => {
+    const session = createSession()
+    const dataSource = createDataSource()
+    dataSource.sessions = [session]
+    const emptyInitialPage = writeTempJson({
+      chatlab: { version: '0.0.2', exportedAt: 100 },
+      meta: { name: 'Test Session', platform: 'test', type: 'group' },
+      members: [],
+      messages: [],
+      sync: { hasMore: false, nextSince: 100 },
+    })
+    const retryPage = writeTempJson({
+      chatlab: { version: '0.0.2', exportedAt: 100 },
+      meta: { name: 'Test Session', platform: 'test', type: 'group' },
+      members: [{ platformId: 'u1', accountName: 'Alice' }],
+      messages: [{ sender: 'u1', timestamp: 101, type: 0, content: 'x'.repeat(1200) }],
+      sync: { hasMore: false, nextSince: 101 },
+    })
+    const sessionUpdates: Array<{ sessionId: string; updates: Partial<ImportSession> }> = []
+    const engine = createEngine({
+      files: [emptyInitialPage, retryPage],
+      dataSource,
+      sessionUpdates,
+      importResult: {
+        success: false,
+        newMessageCount: 0,
+        error: 'error.import_in_progress',
+        retryable: true,
+      },
+    })
+
+    const result = await withImmediateTimers(() => engine.executePullSession(dataSource.id, dataSource, session))
+
+    assert.equal(result.success, false)
+    assert.equal(result.error, 'error.import_in_progress')
+    assert.deepEqual(sessionUpdates, [])
+  })
+
   it('reports retry import failure instead of marking the pull successful', async () => {
     const session = createSession()
     const dataSource = createDataSource()

packages/sync/src/pull-engine.ts

@@ -17,6 +17,7 @@ import type {
   SyncLogger,
   FetchParams,
   SyncMeta,
+  ImportResult,
   PullSessionResult,
   PullProgress,
 } from './types'
@@ -223,17 +224,7 @@ export class PullEngine {
     return Array.from(this.progressMap.values())
   }
 
-  private async importTempFile(
+  private async importTempFile(baseUrl: string, sess: ImportSession, tempFile: string): Promise<ImportResult> {
-    baseUrl: string,
-    sess: ImportSession,
-    tempFile: string
-  ): Promise<{
-    success: boolean
-    newMessageCount: number
-    sessionId?: string
-    error?: string
-    needFullResync?: boolean
-  }> {
     let targetId = sess.targetSessionId
     if (!targetId) {
       const derived = deriveLocalSessionId(baseUrl, sess.remoteSessionId)
@@ -247,6 +238,13 @@ export class PullEngine {
     return this.importer.importFile(tempFile, targetId || undefined, externalId)
   }
 
+  private deferRetryableImport(sess: ImportSession, error?: string): PullSessionResult {
+    const errMsg = error || 'Import deferred'
+    this.logger.info(`[Pull] Deferring "${sess.name}": ${errMsg}`)
+    this.markProgressDone(sess.id)
+    return { success: false, newMessageCount: 0, error: errMsg }
+  }
+
   async executePullSession(sourceId: string, ds: DataSource, sess: ImportSession): Promise<PullSessionResult> {
     const currentDs = this.dsManager.get(sourceId)
     if (!currentDs || !currentDs.sessions.some((s) => s.id === sess.id)) {
@@ -342,6 +340,8 @@ export class PullEngine {
                 }
 
                 if (!retryResult.success) {
+                  if (retryResult.retryable) return this.deferRetryableImport(sess, retryResult.error)
+
                   const errMsg = retryResult.error || 'Import failed'
                   this.dsManager.updateSession(sourceId, sess.id, {
                     lastPullAt: Math.floor(Date.now() / 1000),
@@ -421,6 +421,8 @@ export class PullEngine {
           }
 
           if (!result.success) {
+            if (result.retryable) return this.deferRetryableImport(sess, result.error)
+
             const errMsg = result.error || 'Import failed'
             this.dsManager.updateSession(sourceId, sess.id, {
               lastPullAt: Math.floor(Date.now() / 1000),

packages/sync/src/types.ts

@@ -93,6 +93,8 @@ export interface ImportResult {
   sessionId?: string
   error?: string
   needFullResync?: boolean
+  /** The source page was not imported and should be retried without advancing the pull cursor. */
+  retryable?: boolean
 }
 
 export interface PullSessionResult {