Own conflict cancellation at the feature boundary

This commit is contained in:
vorotamoroz committed 2026-11-05 08:36:39 +00:00
1 parent 165944fe65
commit 99d9f4c380
6 files changed
+102 -82

No files matched your search

@@ -17,6 +17,7 @@ export interface ConflictCheckingDependencies {
IConflictService, IConflictService,
"getOptionalConflictCheckMethod" | "queueCheckFor" | "resolve" | "resolveByNewest" "getOptionalConflictCheckMethod" | "queueCheckFor" | "resolve" | "resolveByNewest"
>; >;
readonly signal: AbortSignal;
readonly conflictProcessQueueCount: ReactiveSource<number>; readonly conflictProcessQueueCount: ReactiveSource<number>;
readonly cancelActiveResolution: (filename: FilePathWithPrefix) => void; readonly cancelActiveResolution: (filename: FilePathWithPrefix) => void;
readonly currentSettings: () => ConflictCheckingSettings; readonly currentSettings: () => ConflictCheckingSettings;
@@ -25,8 +26,6 @@ export interface ConflictCheckingDependencies {
} }
export interface ConflictCheckingHandlers { export interface ConflictCheckingHandlers {
/** Stop admission and dispatch while allowing started checks to settle. */
readonly dispose: () => void;
readonly queueCheckForIfOpen: (file: FilePathWithPrefix) => Promise<void>; readonly queueCheckForIfOpen: (file: FilePathWithPrefix) => Promise<void>;
readonly queueCheckFor: (file: FilePathWithPrefix) => Promise<void>; readonly queueCheckFor: (file: FilePathWithPrefix) => Promise<void>;
/** Wait for active checks; false means the queue was cancelled during shutdown. */ /** Wait for active checks; false means the queue was cancelled during shutdown. */
@@ -35,7 +34,6 @@ export interface ConflictCheckingHandlers {
/** Create conflict-checking handlers while retaining scheduling state privately. */ /** Create conflict-checking handlers while retaining scheduling state privately. */
export function createConflictCheckingHandlers(dependencies: ConflictCheckingDependencies): ConflictCheckingHandlers { export function createConflictCheckingHandlers(dependencies: ConflictCheckingDependencies): ConflictCheckingHandlers {
let disposed = false;
const conflictQueue = createConflictCheckQueue({ const conflictQueue = createConflictCheckQueue({
async process(filename) { async process(filename) {
try { try {
@@ -55,7 +53,7 @@ export function createConflictCheckingHandlers(dependencies: ConflictCheckingDep
}); });
const queueCheckForIfOpen = async (file: FilePathWithPrefix): Promise<void> => { const queueCheckForIfOpen = async (file: FilePathWithPrefix): Promise<void> => {
if (disposed) return; if (dependencies.signal.aborted) return;
const path = file; const path = file;
if (dependencies.currentSettings().checkConflictOnlyOnOpen) { if (dependencies.currentSettings().checkConflictOnlyOnOpen) {
const activeFile: FilePath | undefined = dependencies.vault.getActiveFilePath(); const activeFile: FilePath | undefined = dependencies.vault.getActiveFilePath();
@@ -68,9 +66,9 @@ export function createConflictCheckingHandlers(dependencies: ConflictCheckingDep
}; };
const queueCheckFor = async (file: FilePathWithPrefix): Promise<void> => { const queueCheckFor = async (file: FilePathWithPrefix): Promise<void> => {
if (disposed) return; if (dependencies.signal.aborted) return;
const optionalConflictResult = await dependencies.conflict.getOptionalConflictCheckMethod(file); const optionalConflictResult = await dependencies.conflict.getOptionalConflictCheckMethod(file);
if (disposed) return; if (dependencies.signal.aborted) return;
if (optionalConflictResult == true) { if (optionalConflictResult == true) {
// The conflict has been resolved by another process. // The conflict has been resolved by another process.
return; return;
@@ -87,16 +85,18 @@ export function createConflictCheckingHandlers(dependencies: ConflictCheckingDep
}; };
const ensureAllProcessed = (): Promise<boolean> => conflictQueue.waitForIdle(); const ensureAllProcessed = (): Promise<boolean> => conflictQueue.waitForIdle();
const dispose = (): void => { const abortQueue = (): void => {
if (disposed) return;
disposed = true;
// Shutdown keeps its existing database-close order. Observe cleanup // Shutdown keeps its existing database-close order. Observe cleanup
// without making the lifecycle wait for an active resolver. // without making the lifecycle wait for an active resolver.
void conflictQueue.abort().catch((error) => dependencies.log(error, LOG_LEVEL_VERBOSE)); void conflictQueue.abort().catch((error) => dependencies.log(error, LOG_LEVEL_VERBOSE));
}; };
if (dependencies.signal.aborted) {
abortQueue();
} else {
dependencies.signal.addEventListener("abort", abortQueue, { once: true });
}
return { return {
dispose,
queueCheckForIfOpen, queueCheckForIfOpen,
queueCheckFor, queueCheckFor,
ensureAllProcessed, ensureAllProcessed,
@@ -1,4 +1,3 @@
import { randomUUID } from "node:crypto";
import { describe, expect, it, vi } from "vitest"; import { describe, expect, it, vi } from "vitest";
import PouchDB from "pouchdb-core"; import PouchDB from "pouchdb-core";
import MemoryAdapter from "pouchdb-adapter-memory"; import MemoryAdapter from "pouchdb-adapter-memory";
@@ -23,11 +22,12 @@ describe("conflict-check shutdown with a real PouchDB handle", () => {
it.each([false, true])( it.each([false, true])(
"reports cancellation and preserves the database outcome when closed before completion: %s", "reports cancellation and preserves the database outcome when closed before completion: %s",
async (closeDuringShutdown) => { async (closeDuringShutdown) => {
const name = `conflict-shutdown-${randomUUID()}`; const name = `conflict-shutdown-${crypto.randomUUID()}`;
const db = new PouchDB<Probe>(name, { adapter: "memory" }); const db = new PouchDB<Probe>(name, { adapter: "memory" });
const started = deferred(); const started = deferred();
const release = deferred(); const release = deferred();
const conflict = new InjectableConflictService(createServiceContext()); const conflict = new InjectableConflictService(createServiceContext());
const lifetime = new AbortController();
const log = vi.fn(); const log = vi.fn();
const resolver = vi.fn(async () => { const resolver = vi.fn(async () => {
started.resolve(); started.resolve();
@@ -38,6 +38,7 @@ describe("conflict-check shutdown with a real PouchDB handle", () => {
conflict.resolve.setHandler(resolver); conflict.resolve.setHandler(resolver);
const checking = createConflictCheckingHandlers({ const checking = createConflictCheckingHandlers({
conflict, conflict,
signal: lifetime.signal,
conflictProcessQueueCount: conflict.conflictProcessQueueCount, conflictProcessQueueCount: conflict.conflictProcessQueueCount,
cancelActiveResolution: vi.fn(), cancelActiveResolution: vi.fn(),
currentSettings: () => ({ checkConflictOnlyOnOpen: false }), currentSettings: () => ({ checkConflictOnlyOnOpen: false }),
@@ -53,8 +54,8 @@ describe("conflict-check shutdown with a real PouchDB handle", () => {
await checking.queueCheckFor("active.md" as FilePathWithPrefix); await checking.queueCheckFor("active.md" as FilePathWithPrefix);
await started.promise; await started.promise;
completion = checking.ensureAllProcessed(); completion = checking.ensureAllProcessed();
checking.dispose(); lifetime.abort();
checking.dispose(); lifetime.abort();
await checking.queueCheckFor("late.md" as FilePathWithPrefix); await checking.queueCheckFor("late.md" as FilePathWithPrefix);
if (closeDuringShutdown) { if (closeDuringShutdown) {
await db.close(); await db.close();
@@ -19,6 +19,7 @@ export type ConflictResolutionHost = NecessaryServices<
/** Compose the host-neutral conflict handlers with the host's borrowed logger. */ /** Compose the host-neutral conflict handlers with the host's borrowed logger. */
export const useConflictResolutionFeature = createServiceFeature((host: ConflictResolutionHost): void => { export const useConflictResolutionFeature = createServiceFeature((host: ConflictResolutionHost): void => {
const { services, serviceModules, log } = host; const { services, serviceModules, log } = host;
const lifetime = new AbortController();
const operations = createConflictResolutionOperations({ const operations = createConflictResolutionOperations({
events: services.context.events, events: services.context.events,
databaseFileAccess: serviceModules.databaseFileAccess, databaseFileAccess: serviceModules.databaseFileAccess,
@@ -39,6 +40,7 @@ export const useConflictResolutionFeature = createServiceFeature((host: Conflict
currentSettings: () => services.setting.currentSettings(), currentSettings: () => services.setting.currentSettings(),
vault: services.vault, vault: services.vault,
log, log,
signal: lifetime.signal,
}); });
services.conflict.queueCheckForIfOpen.setHandler(checking.queueCheckForIfOpen); services.conflict.queueCheckForIfOpen.setHandler(checking.queueCheckForIfOpen);
@@ -49,14 +51,10 @@ export const useConflictResolutionFeature = createServiceFeature((host: Conflict
services.conflict.resolveByNewest.setHandler(operations.resolveByNewest); services.conflict.resolveByNewest.setHandler(operations.resolveByNewest);
services.conflict.resolveAllConflictedFilesByNewerOnes.setHandler(operations.resolveAllConflictedFilesByNewerOnes); services.conflict.resolveAllConflictedFilesByNewerOnes.setHandler(operations.resolveAllConflictedFilesByNewerOnes);
const eventSubscriptions = new AbortController(); const stop = (): void => lifetime.abort();
const dispose = (): void => { services.context.events.onceEvent(EVENT_PLUGIN_UNLOADED, stop, { signal: lifetime.signal });
checking.dispose();
eventSubscriptions.abort();
};
services.context.events.onceEvent(EVENT_PLUGIN_UNLOADED, dispose, { signal: eventSubscriptions.signal });
services.appLifecycle.onUnload.addHandler(() => { services.appLifecycle.onUnload.addHandler(() => {
dispose(); stop();
return Promise.resolve(true); return Promise.resolve(true);
}); });
}); });
@@ -17,8 +17,10 @@ export function useInteractiveConflictResolutionFeature(
createDialogue: ConflictResolveDialogueFactory createDialogue: ConflictResolveDialogueFactory
): void { ): void {
const services = host.services; const services = host.services;
const lifetime = new AbortController();
const operations = createInteractiveConflictResolutionOperations({ const operations = createInteractiveConflictResolutionOperations({
events: services.context.events, events: services.context.events,
signal: lifetime.signal,
databaseFileAccess: host.serviceModules.databaseFileAccess, databaseFileAccess: host.serviceModules.databaseFileAccess,
localDatabase: () => services.database.localDatabase, localDatabase: () => services.database.localDatabase,
confirm: services.UI.confirm, confirm: services.UI.confirm,
@@ -61,24 +63,18 @@ export function useInteractiveConflictResolutionFeature(
}); });
services.appLifecycle.getUnresolvedMessages.addHandler(operations.getActiveConflictMessages); services.appLifecycle.getUnresolvedMessages.addHandler(operations.getActiveConflictMessages);
services.conflict.resolveByUserInteraction.addHandler(operations.resolveByUserInteraction); services.conflict.resolveByUserInteraction.addHandler(operations.resolveByUserInteraction);
const eventSubscriptions = new AbortController();
const dispose = () => {
// Stop the refresh listener before cancellation so that unloading does
// not start a database read which can race with database disposal.
eventSubscriptions.abort();
operations.dispose();
};
services.context.events.onEvent( services.context.events.onEvent(
EVENT_CONFLICT_CANCELLED, EVENT_CONFLICT_CANCELLED,
(filename) => { (filename) => {
operations.invalidateWaitingResolution(filename); operations.invalidateWaitingResolution(filename);
fireAndForget(() => operations.refreshConflictState(filename)); fireAndForget(() => operations.refreshConflictState(filename));
}, },
{ signal: eventSubscriptions.signal } { signal: lifetime.signal }
); );
services.context.events.onceEvent(EVENT_PLUGIN_UNLOADED, dispose, { signal: eventSubscriptions.signal }); const stop = (): void => lifetime.abort();
services.context.events.onceEvent(EVENT_PLUGIN_UNLOADED, stop, { signal: lifetime.signal });
services.appLifecycle.onUnload.addHandler(() => { services.appLifecycle.onUnload.addHandler(() => {
dispose(); stop();
return Promise.resolve(true); return Promise.resolve(true);
}); });
} }
@@ -37,6 +37,7 @@ function createOperations(
overrides: Partial<InteractiveConflictResolutionOperationsDependencies> = {} overrides: Partial<InteractiveConflictResolutionOperationsDependencies> = {}
) { ) {
const context = createServiceContext(); const context = createServiceContext();
const lifetime = new AbortController();
let dialogueResult: unknown = POSTPONED; let dialogueResult: unknown = POSTPONED;
const constructed = { value: 0 }; const constructed = { value: 0 };
const getDBEntry = vi.fn(async (): Promise<false | { _rev: string; _conflicts?: string[] }> => false); const getDBEntry = vi.fn(async (): Promise<false | { _rev: string; _conflicts?: string[] }> => false);
@@ -55,6 +56,7 @@ function createOperations(
const log = vi.fn(); const log = vi.fn();
const operations = createInteractiveConflictResolutionOperations({ const operations = createInteractiveConflictResolutionOperations({
events: context.events, events: context.events,
signal: overrides.signal ?? lifetime.signal,
databaseFileAccess: { databaseFileAccess: {
getConflictedRevs, getConflictedRevs,
storeContent, storeContent,
@@ -83,6 +85,7 @@ function createOperations(
...overrides, ...overrides,
} as unknown as InteractiveConflictResolutionOperationsDependencies); } as unknown as InteractiveConflictResolutionOperationsDependencies);
return { return {
abort: () => lifetime.abort(),
askSelectString, askSelectString,
askInPopup, askInPopup,
constructed, constructed,
@@ -154,11 +157,13 @@ function createControlledDialogueFactory(
function createDialogueConcurrencyHarness() { function createDialogueConcurrencyHarness() {
const context = createServiceContext(); const context = createServiceContext();
const lifetime = new AbortController();
const emitEvent = vi.spyOn(context.events, "emitEvent"); const emitEvent = vi.spyOn(context.events, "emitEvent");
const dialogues: ControlledDialogue[] = []; const dialogues: ControlledDialogue[] = [];
const createDialogue = createControlledDialogueFactory(context, dialogues); const createDialogue = createControlledDialogueFactory(context, dialogues);
const operations = createInteractiveConflictResolutionOperations({ const operations = createInteractiveConflictResolutionOperations({
events: context.events, events: context.events,
signal: lifetime.signal,
databaseFileAccess: { databaseFileAccess: {
getConflictedRevs: vi.fn(async () => ["2-right"]), getConflictedRevs: vi.fn(async () => ["2-right"]),
storeContent: vi.fn(async () => true), storeContent: vi.fn(async () => true),
@@ -183,7 +188,7 @@ function createDialogueConcurrencyHarness() {
createDialogue: createDialogue as ConflictResolveDialogueFactory, createDialogue: createDialogue as ConflictResolveDialogueFactory,
log: vi.fn(), log: vi.fn(),
} as unknown as InteractiveConflictResolutionOperationsDependencies); } as unknown as InteractiveConflictResolutionOperationsDependencies);
return { context, createDialogue, dialogues, emitEvent, operations }; return { abort: () => lifetime.abort(), context, createDialogue, dialogues, emitEvent, operations };
} }
describe("interactive conflict resolution operations", () => { describe("interactive conflict resolution operations", () => {
@@ -218,7 +223,7 @@ describe("interactive conflict resolution operations", () => {
const settled = Promise.allSettled([active, waiting]); const settled = Promise.allSettled([active, waiting]);
const logsBeforeDisposal = sink.mock.calls.length; const logsBeforeDisposal = sink.mock.calls.length;
fixture.operations.dispose(); fixture.abort();
unregisterLog(); unregisterLog();
finish(result); finish(result);
@@ -256,7 +261,7 @@ describe("interactive conflict resolution operations", () => {
await vi.waitFor(() => expect(fixture.getDBEntry).toHaveBeenCalledOnce()); await vi.waitFor(() => expect(fixture.getDBEntry).toHaveBeenCalledOnce());
const logsBeforeDisposal = sink.mock.calls.length; const logsBeforeDisposal = sink.mock.calls.length;
fixture.operations.dispose(); fixture.abort();
finishRead(false); finishRead(false);
await expect(resolving).resolves.toBe(false); await expect(resolving).resolves.toBe(false);
@@ -288,7 +293,7 @@ describe("interactive conflict resolution operations", () => {
const resolving = fixture.operations.resolveByUserInteraction(path, conflict); const resolving = fixture.operations.resolveByUserInteraction(path, conflict);
await vi.waitFor(() => expect(fixture.getDBEntry).toHaveBeenCalledOnce()); await vi.waitFor(() => expect(fixture.getDBEntry).toHaveBeenCalledOnce());
fixture.operations.dispose(); fixture.abort();
finishRead({ _rev: "2-left", _conflicts: ["2-right"] }); finishRead({ _rev: "2-left", _conflicts: ["2-right"] });
await expect(resolving).resolves.toBe(false); await expect(resolving).resolves.toBe(false);
@@ -316,7 +321,7 @@ describe("interactive conflict resolution operations", () => {
await vi.waitFor(() => expect(fixture.storeContent).toHaveBeenCalledOnce()); await vi.waitFor(() => expect(fixture.storeContent).toHaveBeenCalledOnce());
const logsBeforeDisposal = sink.mock.calls.length; const logsBeforeDisposal = sink.mock.calls.length;
fixture.operations.dispose(); fixture.abort();
finishStore(false); finishStore(false);
await expect(resolving).resolves.toBe(false); await expect(resolving).resolves.toBe(false);
@@ -351,7 +356,7 @@ describe("interactive conflict resolution operations", () => {
await vi.waitFor(() => expect(fixture.storeContent).toHaveBeenCalledOnce()); await vi.waitFor(() => expect(fixture.storeContent).toHaveBeenCalledOnce());
const logsBeforeDisposal = sink.mock.calls.length; const logsBeforeDisposal = sink.mock.calls.length;
fixture.operations.dispose(); fixture.abort();
finishStore(true); finishStore(true);
await expect(resolving).resolves.toBe(false); await expect(resolving).resolves.toBe(false);
@@ -383,7 +388,7 @@ describe("interactive conflict resolution operations", () => {
await vi.waitFor(() => expect(fixture.resolveByDeletingRevision).toHaveBeenCalledOnce()); await vi.waitFor(() => expect(fixture.resolveByDeletingRevision).toHaveBeenCalledOnce());
const logsBeforeDisposal = sink.mock.calls.length; const logsBeforeDisposal = sink.mock.calls.length;
fixture.operations.dispose(); fixture.abort();
finishDeletion(MISSING_OR_ERROR); finishDeletion(MISSING_OR_ERROR);
await expect(resolving).resolves.toBe(false); await expect(resolving).resolves.toBe(false);
@@ -417,7 +422,7 @@ describe("interactive conflict resolution operations", () => {
await vi.waitFor(() => expect(fixture.resolveByDeletingRevision).toHaveBeenCalledOnce()); await vi.waitFor(() => expect(fixture.resolveByDeletingRevision).toHaveBeenCalledOnce());
const logsBeforeDisposal = sink.mock.calls.length; const logsBeforeDisposal = sink.mock.calls.length;
fixture.operations.dispose(); fixture.abort();
finishDeletion(AUTO_MERGED); finishDeletion(AUTO_MERGED);
await expect(resolving).resolves.toBe(false); await expect(resolving).resolves.toBe(false);
@@ -447,7 +452,7 @@ describe("interactive conflict resolution operations", () => {
await vi.waitFor(() => expect(fixture.getConflictedRevs).toHaveBeenCalledOnce()); await vi.waitFor(() => expect(fixture.getConflictedRevs).toHaveBeenCalledOnce());
const logsBeforeDisposal = sink.mock.calls.length; const logsBeforeDisposal = sink.mock.calls.length;
fixture.operations.dispose(); fixture.abort();
rejectInspection(new Error("database unavailable")); rejectInspection(new Error("database unavailable"));
await expect(refreshing).resolves.toBeUndefined(); await expect(refreshing).resolves.toBeUndefined();
@@ -483,7 +488,7 @@ describe("interactive conflict resolution operations", () => {
await vi.waitFor(() => expect(fixture.getConflictedRevs).toHaveBeenCalledOnce()); await vi.waitFor(() => expect(fixture.getConflictedRevs).toHaveBeenCalledOnce());
const logsBeforeDisposal = sink.mock.calls.length; const logsBeforeDisposal = sink.mock.calls.length;
fixture.operations.dispose(); fixture.abort();
finishInspection(["2-right"]); finishInspection(["2-right"]);
await expect(refreshing).resolves.toBeUndefined(); await expect(refreshing).resolves.toBeUndefined();
@@ -503,7 +508,7 @@ describe("interactive conflict resolution operations", () => {
const messages = fixture.operations.getActiveConflictMessages(); const messages = fixture.operations.getActiveConflictMessages();
await vi.waitFor(() => expect(fixture.getConflictedRevs).toHaveBeenCalledOnce()); await vi.waitFor(() => expect(fixture.getConflictedRevs).toHaveBeenCalledOnce());
fixture.operations.dispose(); fixture.abort();
finishInspection(["2-right"]); finishInspection(["2-right"]);
await expect(messages).resolves.toEqual([]); await expect(messages).resolves.toEqual([]);
@@ -512,7 +517,7 @@ describe("interactive conflict resolution operations", () => {
it("does not inspect or publish conflict state when called after disposal", async () => { it("does not inspect or publish conflict state when called after disposal", async () => {
const fixture = createOperations(); const fixture = createOperations();
const emit = vi.spyOn(fixture.context.events, "emitEvent"); const emit = vi.spyOn(fixture.context.events, "emitEvent");
fixture.operations.dispose(); fixture.abort();
emit.mockClear(); emit.mockClear();
await expect(fixture.operations.getActiveConflictMessages()).resolves.toEqual([]); await expect(fixture.operations.getActiveConflictMessages()).resolves.toEqual([]);
@@ -543,7 +548,7 @@ describe("interactive conflict resolution operations", () => {
const resolving = fixture.operations.resolveByUserInteraction(path, conflict); const resolving = fixture.operations.resolveByUserInteraction(path, conflict);
await vi.waitFor(() => expect(fixture.getConflictedRevs).toHaveBeenCalledOnce()); await vi.waitFor(() => expect(fixture.getConflictedRevs).toHaveBeenCalledOnce());
const logsBeforeDisposal = sink.mock.calls.length; const logsBeforeDisposal = sink.mock.calls.length;
fixture.operations.dispose(); fixture.abort();
emit.mockClear(); emit.mockClear();
finishInspection(["2-right"]); finishInspection(["2-right"]);
@@ -577,7 +582,7 @@ describe("interactive conflict resolution operations", () => {
await vi.waitFor(() => expect(fixture.resolveByDeletingRevision).toHaveBeenCalledOnce()); await vi.waitFor(() => expect(fixture.resolveByDeletingRevision).toHaveBeenCalledOnce());
expect(fixture.storeContent).toHaveBeenCalledOnce(); expect(fixture.storeContent).toHaveBeenCalledOnce();
const logsBeforeDisposal = sink.mock.calls.length; const logsBeforeDisposal = sink.mock.calls.length;
fixture.operations.dispose(); fixture.abort();
finishDeletion(result); finishDeletion(result);
await expect(resolving).resolves.toBe(false); await expect(resolving).resolves.toBe(false);
@@ -615,7 +620,7 @@ describe("interactive conflict resolution operations", () => {
const resolving = fixture.operations.resolveByUserInteraction(path, conflict); const resolving = fixture.operations.resolveByUserInteraction(path, conflict);
await vi.waitFor(() => expect(fixture.replicateUnattendedByEvent).toHaveBeenCalledOnce()); await vi.waitFor(() => expect(fixture.replicateUnattendedByEvent).toHaveBeenCalledOnce());
fixture.operations.dispose(); fixture.abort();
finishReplication({ status: "completed" }); finishReplication({ status: "completed" });
await expect(resolving).resolves.toBe(false); await expect(resolving).resolves.toBe(false);
@@ -695,8 +700,8 @@ describe("interactive conflict resolution operations", () => {
); );
await Promise.resolve(); await Promise.resolve();
oldInstance.operations.dispose(); oldInstance.abort();
oldInstance.operations.dispose(); oldInstance.abort();
await expect(Promise.all([oldActive, oldWaiting])).resolves.toEqual([false, false]); await expect(Promise.all([oldActive, oldWaiting])).resolves.toEqual([false, false]);
expect(oldInstance.dialogues).toHaveLength(1); expect(oldInstance.dialogues).toHaveLength(1);
@@ -1247,6 +1252,7 @@ describe("interactive conflict resolution feature composition", () => {
expect(completedOnUnload).toBe(true); expect(completedOnUnload).toBe(true);
expect(dialogues).toHaveLength(1); expect(dialogues).toHaveLength(1);
expect(fixture.getConflictedRevs).not.toHaveBeenCalled();
await fixture.handlers.onUnload?.(); await fixture.handlers.onUnload?.();
}); });
@@ -23,6 +23,8 @@ import type { LiveSyncLocalDB } from "@vrtmrz/livesync-commonlib/compat/pouchdb/
import type { LiveSyncEventHub } from "@vrtmrz/livesync-commonlib/context"; import type { LiveSyncEventHub } from "@vrtmrz/livesync-commonlib/context";
import { fireAndForget } from "octagonal-wheels/promises"; import { fireAndForget } from "octagonal-wheels/promises";
import { withLock } from "octagonal-wheels/concurrency/lock"; import { withLock } from "octagonal-wheels/concurrency/lock";
import { withCancellation } from "octagonal-wheels/concurrency/cancellation";
import { withScope } from "octagonal-wheels/resource/scope";
import { NO_INTERACTION } from "@vrtmrz/livesync-commonlib/replication"; import { NO_INTERACTION } from "@vrtmrz/livesync-commonlib/replication";
import { displayRev } from "@vrtmrz/livesync-commonlib/compat/common/utils"; import { displayRev } from "@vrtmrz/livesync-commonlib/compat/common/utils";
import { EVENT_CONFLICT_CANCELLED, EVENT_ON_UNRESOLVED_ERROR } from "@/common/events.ts"; import { EVENT_CONFLICT_CANCELLED, EVENT_ON_UNRESOLVED_ERROR } from "@/common/events.ts";
@@ -38,6 +40,7 @@ import {
export interface InteractiveConflictResolutionOperationsDependencies { export interface InteractiveConflictResolutionOperationsDependencies {
readonly events: Pick<LiveSyncEventHub, "emitEvent">; readonly events: Pick<LiveSyncEventHub, "emitEvent">;
readonly signal: AbortSignal;
readonly databaseFileAccess: Pick<DatabaseFileAccess, "getConflictedRevs" | "storeContent">; readonly databaseFileAccess: Pick<DatabaseFileAccess, "getConflictedRevs" | "storeContent">;
readonly localDatabase: () => Pick<LiveSyncLocalDB, "getDBEntry" | "findAllDocs">; readonly localDatabase: () => Pick<LiveSyncLocalDB, "getDBEntry" | "findAllDocs">;
readonly confirm: Pick<Confirm, "askSelectString" | "askInPopup">; readonly confirm: Pick<Confirm, "askSelectString" | "askInPopup">;
@@ -52,7 +55,6 @@ export interface InteractiveConflictResolutionOperationsDependencies {
} }
export interface InteractiveConflictResolutionOperations { export interface InteractiveConflictResolutionOperations {
readonly dispose: () => void;
readonly invalidateWaitingResolution: (filename: FilePathWithPrefix) => void; readonly invalidateWaitingResolution: (filename: FilePathWithPrefix) => void;
readonly getActiveConflictMessages: () => Promise<string[]>; readonly getActiveConflictMessages: () => Promise<string[]>;
readonly refreshConflictState: (filename: FilePathWithPrefix) => Promise<void>; readonly refreshConflictState: (filename: FilePathWithPrefix) => Promise<void>;
@@ -77,9 +79,18 @@ export function createInteractiveConflictResolutionOperations(
const latestRequestByFilename = new Map<FilePathWithPrefix, number>(); const latestRequestByFilename = new Map<FilePathWithPrefix, number>();
let nextRequestId = 0; let nextRequestId = 0;
let activeDialogue: { filename: FilePathWithPrefix; dialogue: ConflictResolveDialogue } | undefined; let activeDialogue: { filename: FilePathWithPrefix; dialogue: ConflictResolveDialogue } | undefined;
let disposed = false; const clearPendingState = (): void => {
latestRequestByFilename.clear();
postponedConflictEpisodes.clear();
};
if (dependencies.signal.aborted) {
clearPendingState();
} else {
dependencies.signal.addEventListener("abort", clearPendingState, { once: true });
}
const invalidateWaitingResolution = (filename: FilePathWithPrefix): void => { const invalidateWaitingResolution = (filename: FilePathWithPrefix): void => {
if (dependencies.signal.aborted) return;
// An active dialogue consumes this event itself. Removing its request // An active dialogue consumes this event itself. Removing its request
// here would also remove a same-path replacement which emitted the // here would also remove a same-path replacement which emitted the
// event to close that active dialogue. A non-active request is stale // event to close that active dialogue. A non-active request is stale
@@ -88,23 +99,11 @@ export function createInteractiveConflictResolutionOperations(
latestRequestByFilename.delete(filename); latestRequestByFilename.delete(filename);
}; };
const dispose = (): void => {
if (disposed) return;
disposed = true;
latestRequestByFilename.clear();
postponedConflictEpisodes.clear();
const activeFilename = activeDialogue?.filename;
if (activeFilename !== undefined) {
dependencies.events.emitEvent(EVENT_CONFLICT_CANCELLED, activeFilename);
activeDialogue = undefined;
}
};
const getConflictVersionCount = async (filename: FilePathWithPrefix): Promise<number | undefined> => { const getConflictVersionCount = async (filename: FilePathWithPrefix): Promise<number | undefined> => {
if (disposed) return undefined; if (dependencies.signal.aborted) return undefined;
try { try {
const conflictCount = (await dependencies.databaseFileAccess.getConflictedRevs(filename)).length; const conflictCount = (await dependencies.databaseFileAccess.getConflictedRevs(filename)).length;
if (disposed) return undefined; if (dependencies.signal.aborted) return undefined;
return conflictCount === 0 ? 0 : conflictCount + 1; return conflictCount === 0 ? 0 : conflictCount + 1;
} catch (error) { } catch (error) {
dependencies.log(`Could not inspect the conflict state of ${filename}`, LOG_LEVEL_VERBOSE); dependencies.log(`Could not inspect the conflict state of ${filename}`, LOG_LEVEL_VERBOSE);
@@ -114,11 +113,11 @@ export function createInteractiveConflictResolutionOperations(
}; };
const getActiveConflictMessages = async (): Promise<string[]> => { const getActiveConflictMessages = async (): Promise<string[]> => {
if (disposed) return []; if (dependencies.signal.aborted) return [];
const filename = dependencies.vault.getActiveFilePath(); const filename = dependencies.vault.getActiveFilePath();
if (!filename) return []; if (!filename) return [];
const versionCount = await getConflictVersionCount(filename); const versionCount = await getConflictVersionCount(filename);
if (disposed) return []; if (dependencies.signal.aborted) return [];
if (versionCount === 0) { if (versionCount === 0) {
postponedConflictEpisodes.delete(filename); postponedConflictEpisodes.delete(filename);
return []; return [];
@@ -137,9 +136,9 @@ export function createInteractiveConflictResolutionOperations(
}; };
const refreshConflictState = async (filename: FilePathWithPrefix): Promise<void> => { const refreshConflictState = async (filename: FilePathWithPrefix): Promise<void> => {
if (disposed) return; if (dependencies.signal.aborted) return;
const versionCount = await getConflictVersionCount(filename); const versionCount = await getConflictVersionCount(filename);
if (disposed) return; if (dependencies.signal.aborted) return;
if (versionCount === 0) { if (versionCount === 0) {
postponedConflictEpisodes.delete(filename); postponedConflictEpisodes.delete(filename);
} }
@@ -147,9 +146,11 @@ export function createInteractiveConflictResolutionOperations(
}; };
const requestConflictResolution = async (filename: FilePathWithPrefix): Promise<boolean> => { const requestConflictResolution = async (filename: FilePathWithPrefix): Promise<boolean> => {
if (dependencies.signal.aborted) return false;
postponedConflictEpisodes.delete(filename); postponedConflictEpisodes.delete(filename);
dependencies.events.emitEvent(EVENT_ON_UNRESOLVED_ERROR); dependencies.events.emitEvent(EVENT_ON_UNRESOLVED_ERROR);
await dependencies.conflict.queueCheckFor(filename); await dependencies.conflict.queueCheckFor(filename);
if (dependencies.signal.aborted) return false;
return await dependencies.conflict.ensureAllProcessed(); return await dependencies.conflict.ensureAllProcessed();
}; };
@@ -157,22 +158,28 @@ export function createInteractiveConflictResolutionOperations(
filename: FilePathWithPrefix, filename: FilePathWithPrefix,
conflictCheckResult: diff_result conflictCheckResult: diff_result
): Promise<boolean> => { ): Promise<boolean> => {
if (disposed) return false; if (dependencies.signal.aborted) return false;
return await withScope(undefined, async (request) => {
if (dependencies.signal.aborted) return false;
const requestId = ++nextRequestId; const requestId = ++nextRequestId;
if (activeDialogue?.filename === filename) { if (activeDialogue?.filename === filename) {
dependencies.events.emitEvent(EVENT_CONFLICT_CANCELLED, filename); dependencies.events.emitEvent(EVENT_CONFLICT_CANCELLED, filename);
activeDialogue = undefined; activeDialogue = undefined;
} }
latestRequestByFilename.set(filename, requestId); latestRequestByFilename.set(filename, requestId);
request.defer(() => {
if (latestRequestByFilename.get(filename) === requestId) {
latestRequestByFilename.delete(filename);
}
});
// UI for resolving different files should proceed one-by-one. A newer // UI for resolving different files should proceed one-by-one. A newer
// request for the active file replaces its dialogue instead of waiting // request for the active file replaces its dialogue instead of waiting
// behind a comparison which is already stale. // behind a comparison which is already stale.
return await withLock(dialogueSerialisationKey, async () => { return await withLock(dialogueSerialisationKey, async () => {
if (disposed || latestRequestByFilename.get(filename) !== requestId) { if (dependencies.signal.aborted || latestRequestByFilename.get(filename) !== requestId) {
return false; return false;
} }
try {
if (postponedConflictEpisodes.has(filename)) { if (postponedConflictEpisodes.has(filename)) {
dependencies.log(`Merge: Postponed ${filename}`, LOG_LEVEL_VERBOSE); dependencies.log(`Merge: Postponed ${filename}`, LOG_LEVEL_VERBOSE);
dependencies.events.emitEvent(EVENT_ON_UNRESOLVED_ERROR); dependencies.events.emitEvent(EVENT_ON_UNRESOLVED_ERROR);
@@ -183,14 +190,24 @@ export function createInteractiveConflictResolutionOperations(
activeDialogue = { filename, dialogue }; activeDialogue = { filename, dialogue };
let selected: MergeDialogResult; let selected: MergeDialogResult;
try { try {
selected = await withCancellation(dialogue, { signal: dependencies.signal }, async (waiting) => {
const cancelDialogue = (): void => {
if (activeDialogue?.dialogue === dialogue) {
dependencies.events.emitEvent(EVENT_CONFLICT_CANCELLED, filename);
}
};
waiting.signal.addEventListener("abort", cancelDialogue, { once: true });
waiting.defer(() => waiting.signal.removeEventListener("abort", cancelDialogue));
dialogue.open(); dialogue.open();
selected = await dialogue.waitForResult(); if (waiting.isCancelled()) cancelDialogue();
return await dialogue.waitForResult();
});
} finally { } finally {
if (activeDialogue?.dialogue === dialogue) { if (activeDialogue?.dialogue === dialogue) {
activeDialogue = undefined; activeDialogue = undefined;
} }
} }
if (disposed) return false; if (dependencies.signal.aborted) return false;
if (selected === POSTPONED) { if (selected === POSTPONED) {
postponedConflictEpisodes.add(filename); postponedConflictEpisodes.add(filename);
dependencies.events.emitEvent(EVENT_ON_UNRESOLVED_ERROR); dependencies.events.emitEvent(EVENT_ON_UNRESOLVED_ERROR);
@@ -212,7 +229,7 @@ export function createInteractiveConflictResolutionOperations(
if (!testDoc._conflicts || testDoc._conflicts.length === 0) { if (!testDoc._conflicts || testDoc._conflicts.length === 0) {
dependencies.log(`Merge: Nothing to do ${filename}`, LOG_LEVEL_VERBOSE); dependencies.log(`Merge: Nothing to do ${filename}`, LOG_LEVEL_VERBOSE);
await refreshConflictState(filename); await refreshConflictState(filename);
if (disposed) return false; if (dependencies.signal.aborted) return false;
return false; return false;
} }
if ( if (
@@ -224,7 +241,7 @@ export function createInteractiveConflictResolutionOperations(
LOG_LEVEL_INFO LOG_LEVEL_INFO
); );
await refreshConflictState(filename); await refreshConflictState(filename);
if (disposed) return false; if (dependencies.signal.aborted) return false;
await dependencies.conflict.queueCheckFor(filename); await dependencies.conflict.queueCheckFor(filename);
return false; return false;
} }
@@ -286,17 +303,15 @@ export function createInteractiveConflictResolutionOperations(
// And, check it again. // And, check it again.
await dependencies.conflict.queueCheckFor(filename); await dependencies.conflict.queueCheckFor(filename);
return false; return false;
} finally { });
if (latestRequestByFilename.get(filename) === requestId) {
latestRequestByFilename.delete(filename);
}
}
}); });
}; };
const pickFileForResolve = async (notifyIfEmpty = true): Promise<boolean> => { const pickFileForResolve = async (notifyIfEmpty = true): Promise<boolean> => {
if (dependencies.signal.aborted) return false;
const notes: { id: DocumentID; path: FilePathWithPrefix; dispPath: string; mtime: number }[] = []; const notes: { id: DocumentID; path: FilePathWithPrefix; dispPath: string; mtime: number }[] = [];
for await (const doc of dependencies.localDatabase().findAllDocs({ conflicts: true })) { for await (const doc of dependencies.localDatabase().findAllDocs({ conflicts: true })) {
if (dependencies.signal.aborted) return false;
if (!("_conflicts" in doc)) continue; if (!("_conflicts" in doc)) continue;
notes.push({ notes.push({
id: doc._id, id: doc._id,
@@ -305,6 +320,7 @@ export function createInteractiveConflictResolutionOperations(
mtime: doc.mtime, mtime: doc.mtime,
}); });
} }
if (dependencies.signal.aborted) return false;
notes.sort((a, b) => b.mtime - a.mtime); notes.sort((a, b) => b.mtime - a.mtime);
const notesList = notes.map((e) => e.dispPath); const notesList = notes.map((e) => e.dispPath);
if (notesList.length == 0) { if (notesList.length == 0) {
@@ -314,6 +330,7 @@ export function createInteractiveConflictResolutionOperations(
return false; return false;
} }
const target = await dependencies.confirm.askSelectString("File to resolve conflict", notesList); const target = await dependencies.confirm.askSelectString("File to resolve conflict", notesList);
if (dependencies.signal.aborted) return false;
if (target) { if (target) {
const targetItem = notes.find((e) => e.dispPath == target)!; const targetItem = notes.find((e) => e.dispPath == target)!;
await requestConflictResolution(targetItem.path); await requestConflictResolution(targetItem.path);
@@ -324,19 +341,22 @@ export function createInteractiveConflictResolutionOperations(
const allConflictCheck = async (): Promise<void> => { const allConflictCheck = async (): Promise<void> => {
let notifyIfEmpty = true; let notifyIfEmpty = true;
while (await pickFileForResolve(notifyIfEmpty)) { while (!dependencies.signal.aborted && (await pickFileForResolve(notifyIfEmpty))) {
notifyIfEmpty = false; notifyIfEmpty = false;
} }
}; };
const scanStartupIssues = async (): Promise<boolean> => { const scanStartupIssues = async (): Promise<boolean> => {
if (dependencies.signal.aborted) return false;
const notes: { path: string; mtime: number }[] = []; const notes: { path: string; mtime: number }[] = [];
dependencies.log(`Checking conflicted files`, LOG_LEVEL_VERBOSE); dependencies.log(`Checking conflicted files`, LOG_LEVEL_VERBOSE);
try { try {
for await (const doc of dependencies.localDatabase().findAllDocs({ conflicts: true })) { for await (const doc of dependencies.localDatabase().findAllDocs({ conflicts: true })) {
if (dependencies.signal.aborted) return false;
if (!("_conflicts" in doc)) continue; if (!("_conflicts" in doc)) continue;
notes.push({ path: dependencies.path.getPath(doc), mtime: doc.mtime }); notes.push({ path: dependencies.path.getPath(doc), mtime: doc.mtime });
} }
if (dependencies.signal.aborted) return false;
if (notes.length > 0) { if (notes.length > 0) {
dependencies.confirm.askInPopup( dependencies.confirm.askInPopup(
`conflicting-detected-on-safety`, `conflicting-detected-on-safety`,
@@ -367,7 +387,6 @@ export function createInteractiveConflictResolutionOperations(
}; };
return { return {
dispose,
invalidateWaitingResolution, invalidateWaitingResolution,
getActiveConflictMessages, getActiveConflictMessages,
refreshConflictState, refreshConflictState,