From 5be001fea57f13951ec43de9dd07940fc8128ad3 Mon Sep 17 00:00:00 2001 From: digua Date: Mon, 13 Jul 2026 00:53:34 +0800 Subject: [PATCH] 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 | 46 +++++++++++++++++- apps/cli/src/sync/adapters.ts | 10 +++- apps/desktop/main/api/adapters.ts | 9 ++++ packages/sync/src/pull-engine.test.ts | 69 +++++++++++++++++++++++++++ packages/sync/src/pull-engine.ts | 24 +++++----- packages/sync/src/types.ts | 2 + 6 files changed, 147 insertions(+), 13 deletions(-) diff --git a/apps/cli/src/sync/adapters.test.ts b/apps/cli/src/sync/adapters.test.ts index 01a22eee..be84875a 100644 --- a/apps/cli/src/sync/adapters.test.ts +++ b/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) +}) diff --git a/apps/cli/src/sync/adapters.ts b/apps/cli/src/sync/adapters.ts index fe3600bf..68de909a 100644 --- a/apps/cli/src/sync/adapters.ts +++ b/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) { diff --git a/apps/desktop/main/api/adapters.ts b/apps/desktop/main/api/adapters.ts index bc73c4f6..3684963b 100644 --- a/apps/desktop/main/api/adapters.ts +++ b/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 } } diff --git a/packages/sync/src/pull-engine.test.ts b/packages/sync/src/pull-engine.test.ts index 3a22e540..5cdab321 100644 --- a/packages/sync/src/pull-engine.test.ts +++ b/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 }> = [] + 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 }> = [] + 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() diff --git a/packages/sync/src/pull-engine.ts b/packages/sync/src/pull-engine.ts index 66b4c560..ad646312 100644 --- a/packages/sync/src/pull-engine.ts +++ b/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( - baseUrl: string, - sess: ImportSession, - tempFile: string - ): Promise<{ - success: boolean - newMessageCount: number - sessionId?: string - error?: string - needFullResync?: boolean - }> { + private async importTempFile(baseUrl: string, sess: ImportSession, tempFile: string): Promise { 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 { 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), diff --git a/packages/sync/src/types.ts b/packages/sync/src/types.ts index 73d9d1e7..6725cca0 100644 --- a/packages/sync/src/types.ts +++ b/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 {