From dfb6215e7798f5a13a2ea46934aceb2ad5774c7e Mon Sep 17 00:00:00 2001 From: vorotamoroz Date: Sat, 31 Oct 2026 14:48:57 +0000 Subject: [PATCH] Replace hidden-file conflict processors with scoped queues --- .../CmdHiddenFileSync.queue.unit.spec.ts | 213 +++++++++++++++++- .../HiddenFileSync/CmdHiddenFileSync.ts | 71 +++--- .../HiddenFileSync/HiddenFileConflictQueue.ts | 209 +++++++++++++++++ .../HiddenFileConflictQueue.unit.spec.ts | 194 ++++++++++++++++ 4 files changed, 641 insertions(+), 46 deletions(-) create mode 100644 src/features/HiddenFileSync/HiddenFileConflictQueue.ts create mode 100644 src/features/HiddenFileSync/HiddenFileConflictQueue.unit.spec.ts diff --git a/src/features/HiddenFileSync/CmdHiddenFileSync.queue.unit.spec.ts b/src/features/HiddenFileSync/CmdHiddenFileSync.queue.unit.spec.ts index fbf68283..056e46a7 100644 --- a/src/features/HiddenFileSync/CmdHiddenFileSync.queue.unit.spec.ts +++ b/src/features/HiddenFileSync/CmdHiddenFileSync.queue.unit.spec.ts @@ -3,7 +3,11 @@ import { afterEach, describe, expect, it, vi } from "vitest"; vi.mock("@/deps.ts", () => ({})); vi.mock("@/features/HiddenFileCommon/JsonResolveModal.ts", () => ({ JsonResolveModal: class JsonResolveModal {} })); vi.mock("@/modules/features/ModuleLog.ts", () => ({ MARK_DONE: "" })); -vi.mock("@/common/PeriodicProcessor.ts", () => ({ PeriodicProcessor: class PeriodicProcessor {} })); +vi.mock("@/common/PeriodicProcessor.ts", () => ({ + PeriodicProcessor: class PeriodicProcessor { + disable = vi.fn(); + }, +})); vi.mock("./configureHiddenFileSyncMode.ts", () => ({ configureHiddenFileSyncMode: vi.fn() })); import { @@ -51,7 +55,7 @@ function createHiddenFileSync(localDatabase: object) { afterEach(() => { for (const hiddenFileSync of queueOwners.splice(0)) { - hiddenFileSync.conflictResolutionProcessor.terminateAll(); + hiddenFileSync.conflictResolutionProcessor.dispose(); } vi.restoreAllMocks(); vi.useRealTimers(); @@ -90,7 +94,6 @@ describe("HiddenFileSync conflict queue", () => { let completed = false; hiddenFileSync.queueConflictCheck(path); try { - expect(hiddenFileSync.conflictResolutionProcessor.delay).toBe(10); await waitForCondition( () => hiddenFileSync.conflictResolutionProcessor.remaining === 1, "The initial conflict check was not queued" @@ -168,7 +171,6 @@ describe("HiddenFileSync conflict queue", () => { let completed = false; hiddenFileSync.queueConflictCheck(path); try { - expect(hiddenFileSync.conflictResolutionProcessor.delay).toBe(10); await waitForCondition( () => hiddenFileSync.conflictResolutionProcessor.remaining === 1, "The initial conflict check was not queued" @@ -196,3 +198,206 @@ describe("HiddenFileSync conflict queue", () => { } }); }); + +describe("HiddenFileSync pipeline contracts", () => { + it("starts above the threshold, limits analysis to five, and retains FIFO input", async () => { + vi.useFakeTimers(); + const gates = Array.from({ length: 11 }, () => deferred()); + const paths = gates.map((_, index) => `i:.obsidian/${index}.txt` as FilePathWithPrefix); + const reads: string[] = []; + const getRaw = vi.fn((id: DocumentID) => { + reads.push(id); + return gates[paths.indexOf(id as unknown as FilePathWithPrefix)].promise; + }); + const { hiddenFileSync } = createHiddenFileSync({ getRaw }); + hiddenFileSync.path2id = async (path) => path as unknown as DocumentID; + for (const path of paths.slice(0, 10)) hiddenFileSync.queueConflictCheck(path); + await vi.advanceTimersByTimeAsync(0); + expect(reads).toEqual([]); + hiddenFileSync.queueConflictCheck(paths[10]); + await vi.advanceTimersByTimeAsync(0); + expect(reads).toEqual(paths.slice(0, 5)); + hiddenFileSync.queueConflictCheck(paths[0]); + expect(hiddenFileSync.pendingConflictChecks.size).toBe(11); + gates[0].resolve({ path: paths[0] } as MetaEntry); + await vi.advanceTimersByTimeAsync(0); + expect(reads).toEqual(paths.slice(0, 6)); + gates.forEach((gate, index) => gate.resolve({ path: paths[index] } as MetaEntry)); + await vi.advanceTimersByTimeAsync(100); + expect(await hiddenFileSync.conflictResolutionProcessor.waitForAllProcessed()).toBe(true); + expect(reads).toEqual(paths); + expect(hiddenFileSync.pendingConflictChecks.size).toBe(0); + }); + + it("continues analysis while one manual dialogue is open and serialises the dialogues", async () => { + vi.useFakeTimers(); + const paths = Array.from({ length: 12 }, (_, index) => `i:.obsidian/${index}.json` as FilePathWithPrefix); + const gates = paths.map(() => deferred()); + const localDatabase = { + getRaw: vi.fn(async (id: DocumentID, options?: { revs_info?: boolean }) => + options?.revs_info + ? {} + : { + _id: id, + path: id, + _rev: "2-current", + _conflicts: ["2-other"], + } + ), + getDBEntry: vi.fn(async (path: FilePathWithPrefix) => ({ path }) as LoadedEntry), + managers: { conflictManager: { mergeObject: vi.fn(async () => false) } }, + }; + const { hiddenFileSync } = createHiddenFileSync(localDatabase); + hiddenFileSync.path2id = async (path) => path as unknown as DocumentID; + const opened: FilePathWithPrefix[] = []; + hiddenFileSync.showJSONMergeDialogAndMerge = vi.fn((doc) => { + opened.push(doc.path); + return gates[paths.indexOf(doc.path)].promise; + }); + paths.forEach((path) => hiddenFileSync.queueConflictCheck(path)); + await vi.advanceTimersByTimeAsync(100); + expect(localDatabase.managers.conflictManager.mergeObject).toHaveBeenCalledTimes(12); + expect(opened).toEqual([paths[0]]); + paths.forEach((path) => hiddenFileSync.queueConflictCheck(path)); + expect(hiddenFileSync.pendingConflictChecks.size).toBe(12); + let completed = false; + void hiddenFileSync.conflictResolutionProcessor.waitForAllProcessed().then(() => { + completed = true; + }); + for (let index = 0; index < paths.length; index++) { + expect(opened).toEqual(paths.slice(0, index + 1)); + expect(completed).toBe(false); + gates[index].resolve(false); + await vi.advanceTimersByTimeAsync(100); + } + expect(completed).toBe(true); + expect(hiddenFileSync.pendingConflictChecks.size).toBe(0); + }); + + it("logs an analysis error and continues with later paths", async () => { + vi.useFakeTimers(); + const failure = new Error("Injected read failure"); + const getRaw = vi.fn().mockRejectedValueOnce(failure).mockResolvedValue({}); + const { hiddenFileSync } = createHiddenFileSync({ getRaw }); + hiddenFileSync.path2id = async (path) => path as unknown as DocumentID; + hiddenFileSync.queueConflictCheck("i:bad.txt" as FilePathWithPrefix); + hiddenFileSync.queueConflictCheck("i:good.txt" as FilePathWithPrefix); + await vi.advanceTimersByTimeAsync(100); + expect(await hiddenFileSync.conflictResolutionProcessor.waitForAllProcessed()).toBe(true); + expect(getRaw).toHaveBeenCalledTimes(2); + expect(hiddenFileSync._log).toHaveBeenCalledWith(failure, expect.anything()); + expect(hiddenFileSync.pendingConflictChecks.size).toBe(0); + }); + + it("collects a conflict enumeration before resuming analysis", async () => { + vi.useFakeTimers(); + const gate = deferred(); + const path = "i:.obsidian/scanned.txt" as FilePathWithPrefix; + const getRaw = vi.fn(async () => ({})); + const { hiddenFileSync } = createHiddenFileSync({ + getRaw, + findEntries: async function* () { + await gate.promise; + yield { _id: path, path, _conflicts: ["2-other"] }; + }, + }); + hiddenFileSync.path2id = async (value) => value as unknown as DocumentID; + const result = hiddenFileSync.resolveConflictOnInternalFiles(); + await vi.advanceTimersByTimeAsync(100); + hiddenFileSync.queueConflictCheck(path); + await vi.advanceTimersByTimeAsync(100); + expect(getRaw).not.toHaveBeenCalled(); + gate.resolve(); + await vi.advanceTimersByTimeAsync(100); + await result; + expect(getRaw).toHaveBeenCalledOnce(); + expect(hiddenFileSync.pendingConflictChecks.size).toBe(0); + }); +}); + +describe("HiddenFileSync conflict queue ownership", () => { + it("stops delayed work on unload and rejects later conflict events", async () => { + vi.useFakeTimers(); + const getRaw = vi.fn(async () => ({})); + const { hiddenFileSync } = createHiddenFileSync({ getRaw }); + hiddenFileSync.path2id = async (path) => path as unknown as DocumentID; + const path = "i:.obsidian/queued.txt" as FilePathWithPrefix; + hiddenFileSync.queueConflictCheck(path); + const idle = hiddenFileSync.conflictResolutionProcessor.waitForAllProcessed(); + hiddenFileSync.onunload(); + hiddenFileSync.queueConflictCheck(path); + await vi.advanceTimersByTimeAsync(100); + expect(await idle).toBe(false); + expect(getRaw).not.toHaveBeenCalled(); + expect(hiddenFileSync.pendingConflictChecks.size).toBe(0); + expect(vi.getTimerCount()).toBe(0); + }); + + it("finishes an admitted analysis without handing its result to a new dialogue after unload", async () => { + vi.useFakeTimers(); + const gate = deferred(); + const path = "i:.obsidian/active.json" as FilePathWithPrefix; + const getRaw = vi + .fn() + .mockImplementationOnce(() => gate.promise) + .mockResolvedValue({}); + const getDBEntry = vi.fn(async () => ({ path })); + const mergeObject = vi.fn(async () => false); + const { hiddenFileSync } = createHiddenFileSync({ + getRaw, + getDBEntry, + managers: { conflictManager: { mergeObject } }, + }); + hiddenFileSync.path2id = async () => path as unknown as DocumentID; + hiddenFileSync.queueConflictCheck(path); + await vi.advanceTimersByTimeAsync(10); + expect(getRaw).toHaveBeenCalledOnce(); + hiddenFileSync.onunload(); + expect(await hiddenFileSync.conflictResolutionProcessor.waitForAllProcessed()).toBe(false); + gate.resolve({ + _id: String(path) as DocumentID, + path, + _rev: "2-current", + _conflicts: ["2-other"], + } as MetaEntry); + await vi.advanceTimersByTimeAsync(100); + expect(mergeObject).toHaveBeenCalledOnce(); + expect(getDBEntry).not.toHaveBeenCalled(); + expect(hiddenFileSync.pendingConflictChecks.size).toBe(0); + }); + + it("does not requeue a manual resolution which finishes after unload", async () => { + vi.useFakeTimers(); + const gate = deferred(); + const path = "i:.obsidian/dialogue.json" as FilePathWithPrefix; + const getRaw = vi.fn(async (_id: DocumentID, options?: { revs_info?: boolean }) => + options?.revs_info + ? {} + : { + _id: path, + path, + _rev: "2-current", + _conflicts: ["2-other"], + } + ); + const getDBEntry = vi.fn(async () => ({ path }) as LoadedEntry); + const { hiddenFileSync } = createHiddenFileSync({ + getRaw, + getDBEntry, + managers: { conflictManager: { mergeObject: async () => false } }, + }); + hiddenFileSync.path2id = async () => path as unknown as DocumentID; + hiddenFileSync.showJSONMergeDialogAndMerge = vi.fn(() => gate.promise); + hiddenFileSync.queueConflictCheck(path); + await vi.advanceTimersByTimeAsync(20); + expect(hiddenFileSync.showJSONMergeDialogAndMerge).toHaveBeenCalledOnce(); + hiddenFileSync.onunload(); + expect(await hiddenFileSync.conflictResolutionProcessor.waitForAllProcessed()).toBe(false); + gate.resolve(true); + await vi.advanceTimersByTimeAsync(100); + expect(getRaw).toHaveBeenCalledTimes(2); + expect(getDBEntry).toHaveBeenCalledTimes(2); + expect(hiddenFileSync.pendingConflictChecks.size).toBe(0); + expect(hiddenFileSync.conflictResolutionProcessor.totalRemaining).toBe(0); + }); +}); diff --git a/src/features/HiddenFileSync/CmdHiddenFileSync.ts b/src/features/HiddenFileSync/CmdHiddenFileSync.ts index d3911831..ed31cc49 100644 --- a/src/features/HiddenFileSync/CmdHiddenFileSync.ts +++ b/src/features/HiddenFileSync/CmdHiddenFileSync.ts @@ -45,7 +45,7 @@ import { JsonResolveModal } from "@/features/HiddenFileCommon/JsonResolveModal.t import { LiveSyncCommands } from "@/features/LiveSyncCommands.ts"; import { addPrefix, stripAllPrefixes } from "@vrtmrz/livesync-commonlib/compat/string_and_binary/path"; import { createCoalescingNotifier } from "octagonal-wheels/concurrency/coalescingNotifier"; -import { QueueProcessor } from "octagonal-wheels/concurrency/processor"; +import { HiddenFileConflictQueue } from "./HiddenFileConflictQueue.ts"; import { hiddenFilesEventCount, hiddenFilesProcessingCount, @@ -108,6 +108,8 @@ export class HiddenFileSync extends LiveSyncCommands { } onunload() { this.periodicInternalFileScanProcessor?.disable(); + this.conflictResolutionProcessor.dispose(); + this.pendingConflictChecks.clear(); } onload() { this.services.API.addCommand({ @@ -745,7 +747,7 @@ Offline Changed files: ${processFiles.length}`; queueConflictCheck(path: FilePathWithPrefix) { if (this.pendingConflictChecks.has(path)) return; this.pendingConflictChecks.add(path); - this.conflictResolutionProcessor.enqueue(path); + if (!this.conflictResolutionProcessor.enqueue(path)) this.pendingConflictChecks.delete(path); } finishConflictCheck(path: FilePathWithPrefix) { @@ -802,9 +804,8 @@ Offline Changed files: ${processFiles.length}`; } // check the file again } - conflictResolutionProcessor = new QueueProcessor( - async (paths: FilePathWithPrefix[]) => { - const path = paths[0]; + conflictResolutionProcessor = new HiddenFileConflictQueue( + async (path: FilePathWithPrefix) => { try { // Retrieve data const id = await this.path2id(path, ICHeader); @@ -872,46 +873,32 @@ Offline Changed files: ${processFiles.length}`; return []; } }, - { - suspended: false, - batchSize: 1, - concurrentLimit: 5, - delay: 10, - keepResultUntilDownstreamConnected: true, - yieldThreshold: 10, - pipeTo: new QueueProcessor( - async (results) => { - const { id, doc, path, revA, revB } = results[0]; - const prefixedPath = addPrefix(path, ICHeader); - const docAMerge = await this.localDatabase.getDBEntry(prefixedPath, { rev: revA }); - const docBMerge = await this.localDatabase.getDBEntry(prefixedPath, { rev: revB }); - try { - if (docAMerge != false && docBMerge != false) { - if (await this.showJSONMergeDialogAndMerge(docAMerge, docBMerge)) { - // Again for other conflicted revisions. - this.requeueConflictCheck(path); - } else { - this.finishConflictCheck(path); - } - return; - } else { - // If either revision could not read, force resolving by the newer one. - await this.resolveByNewerEntry(id, path, doc, revA, revB); - } - } catch (ex) { + async (result) => { + const { id, doc, path, revA, revB } = result; + const prefixedPath = addPrefix(path, ICHeader); + const docAMerge = await this.localDatabase.getDBEntry(prefixedPath, { rev: revA }); + const docBMerge = await this.localDatabase.getDBEntry(prefixedPath, { rev: revB }); + try { + if (docAMerge != false && docBMerge != false) { + if (await this.showJSONMergeDialogAndMerge(docAMerge, docBMerge)) { + // Again for other conflicted revisions. + this.requeueConflictCheck(path); + } else { this.finishConflictCheck(path); - throw ex; } - }, - { - suspended: false, - batchSize: 1, - concurrentLimit: 1, - delay: 10, - keepResultUntilDownstreamConnected: false, - yieldThreshold: 10, + return; + } else { + // If either revision could not read, force resolving by the newer one. + await this.resolveByNewerEntry(id, path, doc, revA, revB); } - ), + } catch (ex) { + this.finishConflictCheck(path); + throw ex; + } + }, + (error) => { + this._log("Processor error!"); + this._log(error, LOG_LEVEL_VERBOSE); } ); diff --git a/src/features/HiddenFileSync/HiddenFileConflictQueue.ts b/src/features/HiddenFileSync/HiddenFileConflictQueue.ts new file mode 100644 index 00000000..7a123764 --- /dev/null +++ b/src/features/HiddenFileSync/HiddenFileConflictQueue.ts @@ -0,0 +1,209 @@ +import { compatGlobal, type CompatTimeoutHandle } from "@vrtmrz/livesync-commonlib/compat/common/coreEnvFunctions"; +import { withCancellation } from "octagonal-wheels/concurrency/cancellation"; +import { createNotifier } from "octagonal-wheels/concurrency/notifier"; +import { forEachInScope } from "octagonal-wheels/concurrency/scopedIteration"; +import { createAsyncQueue } from "octagonal-wheels/iterable/asyncQueue"; +import { withOwnedResource } from "octagonal-wheels/resource"; + +/** The scheduling used by one hidden-file conflict stage. */ +class ConflictStage { + private readonly input = createAsyncQueue(); + private readonly reader = this.input[Symbol.asyncIterator](); + private readonly changed = createNotifier(); + private readonly cancellation = new AbortController(); + private timer: CompatTimeoutHandle | undefined; + private ready = false; + private suspended = false; + private stopped = false; + private outstanding = 0; + active = 0; + + constructor( + concurrency: number, + process: (item: T) => Promise, + private readonly notify: () => void, + onError: (error: unknown) => void, + onFailure: (error: unknown) => void + ) { + // The operation owns its input until started callbacks have settled. + // Abort also releases input immediately, before that final cleanup. + void withCancellation(undefined, { signal: this.cancellation.signal }, (scope) => { + const stopInput = () => { + this.input.dispose(); + this.clearTimer(); + this.changed.notify(); + }; + scope.signal.addEventListener("abort", stopInput, { once: true }); + scope.defer(() => scope.signal.removeEventListener("abort", stopInput)); + return withOwnedResource(this.input, () => + forEachInScope(scope, this.availableItems(), { concurrency }, async (entry) => { + while (this.suspended && !this.stopped) await this.changed.wait(); + if (this.stopped) return; + this.active++; + try { + await process(entry.value); + } catch (error) { + try { + onError(error); + } catch (reportingError) { + // Wake a pending input pull before the runner joins + // its other callbacks after an unexpected failure. + onFailure(reportingError); + throw reportingError; + } + } finally { + this.active--; + this.outstanding--; + this.notify(); + } + }) + ); + }).catch(onFailure); + } + + get remaining() { + return this.outstanding; + } + + enqueue(item: T): boolean { + if (this.stopped) return false; + this.input.enqueue(item); + this.outstanding++; + this.schedule(); + this.notify(); + return true; + } + + private clearTimer() { + if (this.timer === undefined) return; + compatGlobal.clearTimeout(this.timer); + this.timer = undefined; + } + + private schedule() { + if (this.stopped || this.suspended || this.input.size === 0) return; + if (this.ready || this.input.size > 10) { + this.clearTimer(); + this.ready = true; + this.changed.notify(); + } else if (this.timer === undefined) { + this.timer = compatGlobal.setTimeout(() => { + this.timer = undefined; + this.ready = true; + this.changed.notify(); + }, 10); + } + } + + private async *availableItems() { + while (!this.stopped) { + while (!this.stopped && (this.suspended || !this.ready || this.input.size === 0)) { + await this.changed.wait(); + } + if (this.stopped) return; + const entry = await this.reader.next(); + if (entry.done) return; + if (this.input.size === 0) { + this.ready = false; + this.clearTimer(); + } + // Outstanding includes this pulled item until its callback finishes. + yield { value: entry.value }; + } + } + + suspend() { + this.suspended = true; + this.ready = false; + this.clearTimer(); + } + + resume() { + if (!this.suspended || this.stopped) return; + this.suspended = false; + this.schedule(); + this.changed.notify(); + } + + dispose() { + if (this.stopped) return; + this.stopped = true; + this.outstanding = this.active; + this.cancellation.abort(); + this.notify(); + } +} + +/** Two hidden-file stages: five analyses, followed by one manual resolution. */ +export class HiddenFileConflictQueue { + private readonly changed = createNotifier(); + private readonly analysis: ConflictStage; + private readonly resolution: ConflictStage; + private stopped = false; + private failure: { error: unknown } | undefined; + + constructor( + analyse: (item: T) => Promise, + resolve: (item: U) => Promise, + onError: (error: unknown) => void + ) { + const notify = () => this.changed.notify(); + const onFailure = (error: unknown) => { + this.failure ??= { error }; + this.dispose(); + }; + this.resolution = new ConflictStage(1, resolve, notify, onError, onFailure); + this.analysis = new ConflictStage( + 5, + async (item) => { + const results = await analyse(item); + for (const result of results) this.resolution.enqueue(result); + }, + notify, + onError, + onFailure + ); + } + + enqueue(item: T): boolean { + return this.analysis.enqueue(item); + } + + get nowProcessing() { + return this.analysis.active; + } + + get remaining() { + return this.analysis.remaining; + } + + get totalRemaining() { + return this.analysis.remaining + this.resolution.remaining; + } + + suspend() { + this.analysis.suspend(); + } + + startPipeline(): this { + this.analysis.resume(); + return this; + } + + /** Includes handoff and self-requeues; disposal is never successful drainage. */ + async waitForAllProcessed(): Promise { + this.startPipeline(); + while (!this.stopped && this.totalRemaining > 0) await this.changed.wait(); + if (this.failure) throw this.failure.error; + return !this.stopped; + } + + /** Stop admission and release pending work without waiting for active dialogue or writes. */ + dispose() { + if (this.stopped) return; + this.stopped = true; + this.analysis.dispose(); + this.resolution.dispose(); + this.changed.notify(); + } +} diff --git a/src/features/HiddenFileSync/HiddenFileConflictQueue.unit.spec.ts b/src/features/HiddenFileSync/HiddenFileConflictQueue.unit.spec.ts new file mode 100644 index 00000000..d4a51f48 --- /dev/null +++ b/src/features/HiddenFileSync/HiddenFileConflictQueue.unit.spec.ts @@ -0,0 +1,194 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { HiddenFileConflictQueue } from "./HiddenFileConflictQueue.ts"; + +const owners: { dispose(): void }[] = []; +function deferred() { + let resolve!: (value: T | PromiseLike) => void; + const promise = new Promise((done) => { + resolve = done; + }); + return { promise, resolve }; +} +function own(queue: HiddenFileConflictQueue) { + owners.push(queue); + return queue; +} + +afterEach(async () => { + owners.splice(0).forEach((owner) => owner.dispose()); + if (vi.isFakeTimers()) await vi.runAllTimersAsync(); + vi.useRealTimers(); +}); + +describe("HiddenFileConflictQueue", () => { + it("awaits repeated handoffs and self-requeues with native timers", async () => { + const seen: number[] = []; + let queue!: HiddenFileConflictQueue; + queue = own( + new HiddenFileConflictQueue( + async (value: number) => [value], + async (value) => { + seen.push(value); + if (value < 3) queue.enqueue(value + 1); + }, + vi.fn() + ) + ); + queue.enqueue(0); + expect(await queue.waitForAllProcessed()).toBe(true); + expect(seen).toEqual([0, 1, 2, 3]); + expect(queue.totalRemaining).toBe(0); + }); + + it("uses the first arrival's ten-millisecond delay in each stage", async () => { + vi.useFakeTimers(); + const analyse = vi.fn(async (value: number) => [value]); + const resolve = vi.fn(async (_value: number) => {}); + const queue = own(new HiddenFileConflictQueue(analyse, resolve, vi.fn())); + queue.enqueue(1); + await vi.advanceTimersByTimeAsync(9); + queue.enqueue(2); + expect(analyse).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1); + expect(analyse.mock.calls).toEqual([[1], [2]]); + expect(resolve).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(9); + expect(resolve).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1); + expect(resolve.mock.calls).toEqual([[1], [2]]); + expect(await queue.waitForAllProcessed()).toBe(true); + expect(queue.totalRemaining).toBe(0); + expect(vi.getTimerCount()).toBe(0); + }); + + it("starts both stages immediately above ten waiting items", async () => { + vi.useFakeTimers(); + const gate = deferred(); + const analyse = vi.fn(async (value: number) => [value]); + const resolve = vi.fn(async (_value: number) => { + await gate.promise; + }); + const queue = own(new HiddenFileConflictQueue(analyse, resolve, vi.fn())); + for (let value = 0; value < 11; value++) queue.enqueue(value); + await vi.advanceTimersByTimeAsync(0); + expect(analyse).toHaveBeenCalledTimes(11); + expect(resolve).toHaveBeenCalledTimes(1); + expect(queue.totalRemaining).toBe(11); + gate.resolve(); + await vi.advanceTimersByTimeAsync(0); + expect(resolve.mock.calls).toEqual(Array.from({ length: 11 }, (_, value) => [value])); + expect(await queue.waitForAllProcessed()).toBe(true); + }); + + it("logs item failures in both stages and continues without implicit retry", async () => { + vi.useFakeTimers(); + const first = new Error("analysis failed"); + const second = new Error("resolution failed"); + const analyse = vi.fn(async (value: number) => { + if (value === 1) throw first; + return [value]; + }); + const resolve = vi.fn(async (value: number) => { + if (value === 2) throw second; + }); + const onError = vi.fn(); + const queue = own(new HiddenFileConflictQueue(analyse, resolve, onError)); + [1, 2, 3].forEach((value) => queue.enqueue(value)); + await vi.advanceTimersByTimeAsync(20); + expect(onError.mock.calls).toEqual([[first], [second]]); + expect(analyse.mock.calls).toEqual([[1], [2], [3]]); + expect(resolve.mock.calls).toEqual([[2], [3]]); + expect(await queue.waitForAllProcessed()).toBe(true); + }); + + it("discards delayed input and releases idle waiters on disposal", async () => { + vi.useFakeTimers(); + const analyse = vi.fn(async (value: number) => [value]); + const resolve = vi.fn(async (_value: number) => {}); + const queue = own(new HiddenFileConflictQueue(analyse, resolve, vi.fn())); + queue.enqueue(1); + const idle = queue.waitForAllProcessed(); + queue.dispose(); + queue.dispose(); + expect(await idle).toBe(false); + expect(queue.enqueue(2)).toBe(false); + expect(queue.totalRemaining).toBe(0); + await vi.advanceTimersByTimeAsync(100); + expect(analyse).not.toHaveBeenCalled(); + expect(resolve).not.toHaveBeenCalled(); + expect(vi.getTimerCount()).toBe(0); + }); + + it("does not wait for active analysis or admit its late handoff and self-requeue after disposal", async () => { + vi.useFakeTimers(); + const gate = deferred(); + const attempted: boolean[] = []; + let finished = 0; + const analyse = vi.fn(async (value: number) => { + await gate.promise; + attempted.push(queue.enqueue(value)); + finished++; + return [value]; + }); + const resolve = vi.fn(async (_value: number) => {}); + const queue = own(new HiddenFileConflictQueue(analyse, resolve, vi.fn())); + for (let value = 0; value < 12; value++) queue.enqueue(value); + await vi.advanceTimersByTimeAsync(0); + expect(analyse).toHaveBeenCalledTimes(5); + const idle = queue.waitForAllProcessed(); + queue.dispose(); + expect(await idle).toBe(false); + expect(finished).toBe(0); + expect(queue.totalRemaining).toBe(5); + gate.resolve(); + await vi.advanceTimersByTimeAsync(100); + expect(finished).toBe(5); + expect(attempted).toEqual([false, false, false, false, false]); + expect(analyse).toHaveBeenCalledTimes(5); + expect(resolve).not.toHaveBeenCalled(); + expect(queue.totalRemaining).toBe(0); + expect(vi.getTimerCount()).toBe(0); + }); + + it("leaves an active dialogue running and discards the next dialogue", async () => { + vi.useFakeTimers(); + const gate = deferred(); + let finished = false; + const resolve = vi.fn(async (_value: number) => { + await gate.promise; + finished = true; + }); + const queue = own(new HiddenFileConflictQueue(async (value: number) => [value], resolve, vi.fn())); + queue.enqueue(1); + queue.enqueue(2); + await vi.advanceTimersByTimeAsync(20); + expect(resolve).toHaveBeenCalledTimes(1); + queue.dispose(); + expect(await queue.waitForAllProcessed()).toBe(false); + expect(finished).toBe(false); + gate.resolve(); + await vi.advanceTimersByTimeAsync(100); + expect(finished).toBe(true); + expect(resolve).toHaveBeenCalledTimes(1); + expect(queue.totalRemaining).toBe(0); + }); + + it("suspends analysis while already handed-off manual work continues", async () => { + vi.useFakeTimers(); + const analyse = vi.fn(async (value: number) => [value]); + const resolve = vi.fn(async (_value: number) => {}); + const queue = own(new HiddenFileConflictQueue(analyse, resolve, vi.fn())); + queue.enqueue(1); + await vi.advanceTimersByTimeAsync(10); + queue.suspend(); + queue.enqueue(2); + await vi.advanceTimersByTimeAsync(100); + expect(analyse.mock.calls).toEqual([[1]]); + expect(resolve.mock.calls).toEqual([[1]]); + queue.startPipeline(); + await vi.advanceTimersByTimeAsync(20); + expect(analyse.mock.calls).toEqual([[1], [2]]); + expect(resolve.mock.calls).toEqual([[1], [2]]); + expect(await queue.waitForAllProcessed()).toBe(true); + }); +});