diff --git a/src/serviceFeatures/conflictResolution/checker.ts b/src/serviceFeatures/conflictResolution/checker.ts index 9882761d..d90263a3 100644 --- a/src/serviceFeatures/conflictResolution/checker.ts +++ b/src/serviceFeatures/conflictResolution/checker.ts @@ -1,5 +1,6 @@ import { LOG_LEVEL_NOTICE, + LOG_LEVEL_VERBOSE, type FilePath, type FilePathWithPrefix, type ObsidianLiveSyncSettings, @@ -7,7 +8,7 @@ import { import type { IConflictService, IVaultService } from "@vrtmrz/livesync-commonlib/compat/services/base/IService"; import type { LogFunction } from "@vrtmrz/livesync-commonlib/compat/services/lib/logUtils"; import type { ReactiveSource } from "octagonal-wheels/dataobject/reactive"; -import { QueueProcessor } from "octagonal-wheels/concurrency/processor"; +import { createConflictCheckQueue } from "@vrtmrz/livesync-commonlib/conflicts"; export type ConflictCheckingSettings = Pick; @@ -30,25 +31,21 @@ export interface ConflictCheckingHandlers { /** Create conflict-checking handlers while retaining scheduling state privately. */ export function createConflictCheckingHandlers(dependencies: ConflictCheckingDependencies): ConflictCheckingHandlers { - const conflictQueue = new QueueProcessor( - async (filenames: FilePathWithPrefix[]) => { - const filename = filenames[0]; - return await dependencies.conflict.resolve(filename); + const conflictQueue = createConflictCheckQueue({ + async process(filename) { + try { + await dependencies.conflict.resolve(filename); + } catch (error) { + dependencies.log(`Conflict check failed: ${filename}`); + dependencies.log(error, LOG_LEVEL_VERBOSE); + } + }, + // The resolver serialises shared user interactions; independent checks + // and automatic merges can use the remaining execution slots. + concurrency: 10, + onProgress(remaining) { + dependencies.conflictProcessQueueCount.value = remaining; }, - { - suspended: false, - batchSize: 1, - // No need to limit concurrency to `1` here, subsequent process will handle it, - // and some cases do not need to be synchronised (for example, auto-merge). - // Global concurrency is limited by the resolver with the UI. - concurrentLimit: 10, - delay: 0, - keepResultUntilDownstreamConnected: false, - totalRemainingReactiveSource: dependencies.conflictProcessQueueCount, - } - ).replaceEnqueueProcessor((queue, newEntity) => { - const newQueue = [...queue].filter((entry) => entry != newEntity); - return [...newQueue, newEntity]; }); const queueCheckForIfOpen = async (file: FilePathWithPrefix): Promise => { @@ -76,7 +73,7 @@ export function createConflictCheckingHandlers(dependencies: ConflictCheckingDep } }; - const ensureAllProcessed = (): Promise => conflictQueue.waitForAllProcessed(); + const ensureAllProcessed = (): Promise => conflictQueue.waitForIdle(); return { queueCheckForIfOpen, diff --git a/src/serviceFeatures/conflictResolution/conflictResolution.unit.spec.ts b/src/serviceFeatures/conflictResolution/conflictResolution.unit.spec.ts index b0aa10cf..57cde733 100644 --- a/src/serviceFeatures/conflictResolution/conflictResolution.unit.spec.ts +++ b/src/serviceFeatures/conflictResolution/conflictResolution.unit.spec.ts @@ -6,6 +6,7 @@ import { DEFAULT_SETTINGS, LOG_LEVEL_INFO, LOG_LEVEL_NOTICE, + LOG_LEVEL_VERBOSE, MISSING_OR_ERROR, NOT_CONFLICTED, type FilePath, @@ -814,6 +815,56 @@ describe("conflict resolution serviceFeature", () => { } }); + it("continues checking accepted and later paths after one resolver rejects", async () => { + const harness = createHarness(); + const failure = new Error("Conflict check failed"); + const resolve = vi.fn().mockRejectedValueOnce(failure).mockResolvedValue(undefined); + harness.conflict.resolve.setHandler(resolve, true); + const first = "failed.md" as FilePathWithPrefix; + const second = "accepted.md" as FilePathWithPrefix; + const later = "later.md" as FilePathWithPrefix; + + await Promise.all([harness.conflict.queueCheckFor(first), harness.conflict.queueCheckFor(second)]); + await expect(harness.conflict.ensureAllProcessed()).resolves.toBe(true); + await harness.conflict.queueCheckFor(later); + await expect(harness.conflict.ensureAllProcessed()).resolves.toBe(true); + + expect(resolve.mock.calls.map(([path]) => path)).toEqual([first, second, later]); + expect(harness.addLog).toHaveBeenCalledWith( + `[SF:ConflictResolution]${MARK_LOG_SEPARATOR} Conflict check failed: ${first}`, + LOG_LEVEL_INFO, + "" + ); + expect(harness.addLog).toHaveBeenCalledWith(failure, LOG_LEVEL_VERBOSE, ""); + expect(harness.conflict.conflictProcessQueueCount.value).toBe(0); + }); + + it("drains requeued automatic checks when every resolver slot is occupied", async () => { + const harness = createHarness({ settings: { syncAfterMerge: false } }); + const paths = Array.from({ length: 10 }, (_, index) => `requeue-${index}.md` as FilePathWithPrefix); + const firstChecks = new Set(); + const finishers: Array<() => void> = []; + harness.tryAutoMerge.mockImplementation(async (path: FilePathWithPrefix) => { + if (firstChecks.has(path)) return { ok: NOT_CONFLICTED }; + firstChecks.add(path); + await new Promise((resolve) => { + finishers.push(resolve); + }); + return { ok: AUTO_MERGED }; + }); + + await Promise.all(paths.map((path) => harness.conflict.queueCheckFor(path))); + await vi.waitFor(() => expect(finishers).toHaveLength(10)); + const completion = harness.conflict.ensureAllProcessed(); + for (const finish of finishers) finish(); + + await expect(completion).resolves.toBe(true); + for (const path of paths) { + expect(harness.tryAutoMerge.mock.calls.filter(([checked]) => checked === path)).toHaveLength(2); + } + expect(harness.conflict.conflictProcessQueueCount.value).toBe(0); + }); + it("honours optional conflict handlers before entering the check queue", async () => { const path = "optional.md" as FilePathWithPrefix; const harness = createHarness();