From 5f76e6535964449b6b66a93a6dd8cdcc360326c6 Mon Sep 17 00:00:00 2001 From: digua Date: Thu, 21 May 2026 21:06:18 +0800 Subject: [PATCH] fix(sync): prevent race condition when adding new subscriptions When a pullAllSessions was already running, the reloadTimer triggered by adding sessions would be skipped by the source-level concurrency lock, causing newly added sessions to wait until the next scheduled interval (up to 60 minutes) before syncing. Fix: trigger individual per-session pulls via triggerPull(sourceId, sessionId) which bypasses the source-level lock, ensuring immediate sync for newly added sessions regardless of concurrent pulls. Also includes: connection pre-check on data source add, removal of duplicate "browse remote sessions" button, hiding "other" session types, and full resync fix for deleted local sessions. Co-authored-by: Cursor --- apps/cli/src/sync/adapters.ts | 5 ++ apps/cli/src/sync/routes.ts | 3 + apps/desktop/main/ipc/api.ts | 3 + packages/sync/src/pull-engine.ts | 37 ++++++++--- .../Settings/API/DataSourceAddModal.vue | 64 ++++++++++--------- src/i18n/locales/en-US/settings.json | 3 +- src/i18n/locales/ja-JP/settings.json | 3 +- src/i18n/locales/zh-CN/settings.json | 3 +- src/i18n/locales/zh-TW/settings.json | 3 +- src/stores/apiServer.ts | 22 ++++++- 10 files changed, 99 insertions(+), 47 deletions(-) diff --git a/apps/cli/src/sync/adapters.ts b/apps/cli/src/sync/adapters.ts index 0ce7ae90..c38ff5ae 100644 --- a/apps/cli/src/sync/adapters.ts +++ b/apps/cli/src/sync/adapters.ts @@ -77,6 +77,11 @@ export class DirectImporter implements DataImporter { return this.incrementalImportFile(targetSessionId, tempFile) } + if (targetSessionId) { + this.logger.info(`[DirectImporter] Session ${targetSessionId} not found locally, need full resync`) + return { success: false, newMessageCount: 0, needFullResync: true } + } + return this.fullImportFile(tempFile, externalId) } diff --git a/apps/cli/src/sync/routes.ts b/apps/cli/src/sync/routes.ts index aec6ff56..bdb4a5ac 100644 --- a/apps/cli/src/sync/routes.ts +++ b/apps/cli/src/sync/routes.ts @@ -77,6 +77,9 @@ export function registerAutomationRoutes(server: FastifyInstance, ctx: SyncRoute return reply.code(404).send({ error: 'Data source not found' }) } reloadTimer(request.params.id, true) + for (const sess of added) { + pullEngine.triggerPull(request.params.id, sess.id).catch(() => {}) + } return added }) diff --git a/apps/desktop/main/ipc/api.ts b/apps/desktop/main/ipc/api.ts index 3a0c9d79..587319ae 100644 --- a/apps/desktop/main/ipc/api.ts +++ b/apps/desktop/main/ipc/api.ts @@ -178,6 +178,9 @@ export function registerApiHandlers(_ctx: IpcContext): void { (_event, sourceId: string, sessions: Array<{ name: string; remoteSessionId: string }>) => { const added = dsManager.addSessions(sourceId, sessions) reloadTimer(sourceId, true) + for (const sess of added) { + pullEngine.triggerPull(sourceId, sess.id).catch(() => {}) + } return added } ) diff --git a/packages/sync/src/pull-engine.ts b/packages/sync/src/pull-engine.ts index 61aaa8a9..a36863f5 100644 --- a/packages/sync/src/pull-engine.ts +++ b/packages/sync/src/pull-engine.ts @@ -104,6 +104,7 @@ export class PullEngine { private dsManager: DataSourceManager private logger: SyncLogger private isImporting: () => boolean + private pullingSourceIds = new Set() constructor(options: PullEngineOptions) { this.fetcher = options.fetcher @@ -256,8 +257,17 @@ export class PullEngine { } async pullAllSessions(ds: DataSource): Promise { - for (const sess of ds.sessions) { - await this.executePullSession(ds.id, ds, sess) + if (this.pullingSourceIds.has(ds.id)) { + this.logger.info(`[Pull] Skipping pullAllSessions for "${ds.baseUrl}": pull already in progress`) + return + } + this.pullingSourceIds.add(ds.id) + try { + for (const sess of ds.sessions) { + await this.executePullSession(ds.id, ds, sess) + } + } finally { + this.pullingSourceIds.delete(ds.id) } } @@ -272,15 +282,24 @@ export class PullEngine { return { success: result.success, error: result.error } } - const errors: string[] = [] - for (const sess of ds.sessions) { - const result = await this.executePullSession(sourceId, ds, sess) - if (!result.success && result.error) errors.push(`${sess.name}: ${result.error}`) + if (this.pullingSourceIds.has(sourceId)) { + this.logger.info(`[Pull] Skipping triggerPull for source ${sourceId}: pull already in progress`) + return { success: true } } - if (errors.length > 0) { - return { success: false, error: errors.join('; ') } + this.pullingSourceIds.add(sourceId) + try { + const errors: string[] = [] + for (const sess of ds.sessions) { + const result = await this.executePullSession(sourceId, ds, sess) + if (!result.success && result.error) errors.push(`${sess.name}: ${result.error}`) + } + if (errors.length > 0) { + return { success: false, error: errors.join('; ') } + } + return { success: true } + } finally { + this.pullingSourceIds.delete(sourceId) } - return { success: true } } async triggerPullAll(sourceId: string): Promise<{ success: boolean; error?: string }> { diff --git a/src/components/common/Settings/API/DataSourceAddModal.vue b/src/components/common/Settings/API/DataSourceAddModal.vue index 5622b101..a80eaf6a 100644 --- a/src/components/common/Settings/API/DataSourceAddModal.vue +++ b/src/components/common/Settings/API/DataSourceAddModal.vue @@ -33,6 +33,7 @@ const remoteSessions = ref([]) const selectedSessionIds = ref>(new Set()) const discovering = ref(false) const loadingMore = ref(false) +const submitting = ref(false) const discoveryError = ref('') const discoveryKeyword = ref('') const discoveryNextCursor = ref() @@ -75,9 +76,13 @@ watch( { immediate: true } ) +const nonOtherSessions = computed(() => + remoteSessions.value.filter((session) => getSessionTypeSelection(session) !== 'other') +) + const visibleRemoteSessions = computed(() => { - if (activeSessionTypeFilter.value === 'all') return remoteSessions.value - return remoteSessions.value.filter((session) => getSessionTypeSelection(session) === activeSessionTypeFilter.value) + if (activeSessionTypeFilter.value === 'all') return nonOtherSessions.value + return nonOtherSessions.value.filter((session) => getSessionTypeSelection(session) === activeSessionTypeFilter.value) }) const visibleAvailableSessions = computed(() => @@ -96,14 +101,13 @@ const sessionTypeSelectionOptions = computed(() => { value: 'all', label: t('settings.api.dataSources.discovery.typeAll') }, { value: 'private', label: t('settings.api.dataSources.discovery.typePrivate') }, { value: 'group', label: t('settings.api.dataSources.discovery.typeGroup') }, - { value: 'other', label: t('settings.api.dataSources.discovery.typeOther') }, ] as Array<{ value: SessionTypeFilter; label: string }> ).map((option) => ({ ...option, count: option.value === 'all' - ? remoteSessions.value.length - : remoteSessions.value.filter((session) => getSessionTypeSelection(session) === option.value).length, + ? nonOtherSessions.value.length + : nonOtherSessions.value.filter((session) => getSessionTypeSelection(session) === option.value).length, })) ) @@ -201,22 +205,33 @@ async function handleSubmit() { .map((s) => ({ name: s.name, remoteSessionId: s.id })) emit('sessionsAdded', props.manageSource.id, sessions) closeModal() - } else { - const ds = await store.addDataSource({ + return + } + + submitting.value = true + discoveryError.value = '' + try { + await store.fetchRemoteSessions(formData.value.baseUrl, formData.value.token, { limit: 1 }) + } catch (err: any) { + discoveryError.value = err.message || t('settings.api.dataSources.discovery.connectionCheckFailed') + return + } finally { + submitting.value = false + } + + submitting.value = true + try { + await store.addDataSource({ name: formData.value.name || undefined, baseUrl: formData.value.baseUrl, token: formData.value.token, intervalMinutes: formData.value.intervalMinutes, pullLimit: formData.value.pullLimit || undefined, }) - if (ds && selectedSessionIds.value.size > 0) { - const sessions = remoteSessions.value - .filter((s) => selectedSessionIds.value.has(s.id)) - .map((s) => ({ name: s.name, remoteSessionId: s.id })) - await store.addImportSessions(ds.id, sessions) - } emit('sourceAdded') closeModal() + } finally { + submitting.value = false } } @@ -297,19 +312,6 @@ function formatMessageCount(count?: number): string { /> - -
- - - {{ t('settings.api.dataSources.discovery.browse') }} - -
@@ -320,11 +322,11 @@ function formatMessageCount(count?: number): string { {{ discoveryError }} - -
+ +
- {{ t('settings.api.dataSources.discovery.found', { count: remoteSessions.length }) }} + {{ t('settings.api.dataSources.discovery.found', { count: nonOtherSessions.length }) }}
{{ t('settings.api.dataSources.discovery.subscribe', { count: selectedSessionIds.size }) }} - + {{ t('settings.api.dataSources.addBtn') }}
diff --git a/src/i18n/locales/en-US/settings.json b/src/i18n/locales/en-US/settings.json index 720d6c00..8f84db12 100644 --- a/src/i18n/locales/en-US/settings.json +++ b/src/i18n/locales/en-US/settings.json @@ -528,7 +528,8 @@ "loadMore": "Load More", "subscribe": "Subscribe {count} sessions", "messages": "messages", - "error": "Failed to connect to remote server" + "error": "Failed to connect to remote server", + "connectionCheckFailed": "Cannot connect to the remote server. Please check the URL and token." }, "edit": { "title": "Edit Data Source", diff --git a/src/i18n/locales/ja-JP/settings.json b/src/i18n/locales/ja-JP/settings.json index 219f48c5..278cb2ae 100644 --- a/src/i18n/locales/ja-JP/settings.json +++ b/src/i18n/locales/ja-JP/settings.json @@ -527,7 +527,8 @@ "loadMore": "さらに読み込む", "subscribe": "{count} 件を購読", "messages": "件のメッセージ", - "error": "リモートサーバーへの接続に失敗しました" + "error": "リモートサーバーへの接続に失敗しました", + "connectionCheckFailed": "リモートサーバーに接続できません。URL とトークンを確認してください。" }, "noSessions": "インポートセッションがありません。上の「サブスクリプション管理」から追加してください", "edit": { diff --git a/src/i18n/locales/zh-CN/settings.json b/src/i18n/locales/zh-CN/settings.json index ed66c896..6cba2dd4 100644 --- a/src/i18n/locales/zh-CN/settings.json +++ b/src/i18n/locales/zh-CN/settings.json @@ -528,7 +528,8 @@ "loadMore": "加载更多", "subscribe": "订阅 {count} 个会话", "messages": "条消息", - "error": "连接远程服务器失败" + "error": "连接远程服务器失败", + "connectionCheckFailed": "无法连接到远程服务器，请检查地址和令牌是否正确" }, "edit": { "title": "修改配置", diff --git a/src/i18n/locales/zh-TW/settings.json b/src/i18n/locales/zh-TW/settings.json index 3238161d..dc597ba7 100644 --- a/src/i18n/locales/zh-TW/settings.json +++ b/src/i18n/locales/zh-TW/settings.json @@ -527,7 +527,8 @@ "loadMore": "載入更多", "subscribe": "訂閱 {count} 個會話", "messages": "則訊息", - "error": "連線遠端伺服器失敗" + "error": "連線遠端伺服器失敗", + "connectionCheckFailed": "無法連線到遠端伺服器，請檢查地址和權杖是否正確" }, "noSessions": "暫無匯入會話，點擊上方「管理訂閱」新增", "edit": { diff --git a/src/stores/apiServer.ts b/src/stores/apiServer.ts index 6bb3561a..1e9068ea 100644 --- a/src/stores/apiServer.ts +++ b/src/stores/apiServer.ts @@ -131,6 +131,10 @@ function createWebTransport(): ApiTransport { async function fetchJson(url: string, options?: RequestInit): Promise { const resp = await fetch(url, options) + if (!resp.ok) { + const body = await resp.json().catch(() => ({})) + throw new Error((body as any)?.error || `HTTP ${resp.status}`) + } return resp.json() } @@ -379,10 +383,9 @@ export const useApiServerStore = defineStore('apiServer', () => { async function addImportSessions(sourceId: string, sessions: Array<{ name: string; remoteSessionId: string }>) { try { const added = await transport.addImportSessions(sourceId, sessions) + await fetchDataSources() if (added.length > 0) { - await triggerPullAll(sourceId) - } else { - await fetchDataSources() + pollDataSourceUpdates(sourceId) } return added } catch (err) { @@ -391,6 +394,19 @@ export const useApiServerStore = defineStore('apiServer', () => { } } + function pollDataSourceUpdates(sourceId: string, maxAttempts = 10, intervalMs = 3000) { + let attempt = 0 + const timer = setInterval(async () => { + attempt++ + await fetchDataSources() + const ds = dataSources.value.find((s) => s.id === sourceId) + const allDone = ds?.sessions.every((s) => s.lastStatus !== 'idle') + if (allDone || attempt >= maxAttempts) { + clearInterval(timer) + } + }, intervalMs) + } + async function removeImportSession(sourceId: string, sessionId: string) { try { const ok = await transport.removeImportSession(sourceId, sessionId)