Use keyed conflict scheduling for normal files

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<ObsidianLiveSyncSettings, "checkConflictOnlyOnOpen">;
 
@@ -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<FilePathWithPrefix, void>(
-        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<void> => {
@@ -76,7 +73,7 @@ export function createConflictCheckingHandlers(dependencies: ConflictCheckingDep
         }
     };
 
-    const ensureAllProcessed = (): Promise<boolean> => conflictQueue.waitForAllProcessed();
+    const ensureAllProcessed = (): Promise<boolean> => conflictQueue.waitForIdle();
 
     return {
         queueCheckForIfOpen,

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<FilePathWithPrefix>();
+        const finishers: Array<() => void> = [];
+        harness.tryAutoMerge.mockImplementation(async (path: FilePathWithPrefix) => {
+            if (firstChecks.has(path)) return { ok: NOT_CONFLICTED };
+            firstChecks.add(path);
+            await new Promise<void>((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();