From de8c2148d1615de9d320abacad5fadb776e1d8c8 Mon Sep 17 00:00:00 2001 From: vorotamoroz Date: Sat, 31 Oct 2026 13:14:25 +0000 Subject: [PATCH] Replace ConfigSync queue and discard stale catalogue scans --- src/LiveSyncBaseCore.ts | 2 +- .../CmdConfigSync.catalogue.unit.spec.ts | 2 +- .../CmdConfigSync.command.unit.spec.ts | 8 +- .../CmdConfigSync.queue.unit.spec.ts | 427 +++++++++++++++++- src/features/ConfigSync/CmdConfigSync.ts | 91 ++-- src/features/ConfigSync/ConfigScanQueue.ts | 176 ++++++++ .../ConfigSync/ConfigScanQueue.unit.spec.ts | 331 ++++++++++++++ src/features/LiveSyncCommands.ts | 2 +- updates.md | 3 + 9 files changed, 983 insertions(+), 59 deletions(-) create mode 100644 src/features/ConfigSync/ConfigScanQueue.ts create mode 100644 src/features/ConfigSync/ConfigScanQueue.unit.spec.ts diff --git a/src/LiveSyncBaseCore.ts b/src/LiveSyncBaseCore.ts index 777bc164..6255cedf 100644 --- a/src/LiveSyncBaseCore.ts +++ b/src/LiveSyncBaseCore.ts @@ -308,7 +308,7 @@ export class LiveSyncBaseCore< } export interface IMinimumLiveSyncCommands { - onunload(): void; + onunload(): void | Promise; onload(): void | Promise; constructor: { name: string }; } diff --git a/src/features/ConfigSync/CmdConfigSync.catalogue.unit.spec.ts b/src/features/ConfigSync/CmdConfigSync.catalogue.unit.spec.ts index 06b04919..ddfce2b0 100644 --- a/src/features/ConfigSync/CmdConfigSync.catalogue.unit.spec.ts +++ b/src/features/ConfigSync/CmdConfigSync.catalogue.unit.spec.ts @@ -84,7 +84,7 @@ async function catalogue() { pluginListGeneration: 0, loadedManifest_mTime: new Map(), updatingV2Count: 0, - pluginScanProcessor: { clearQueue: vi.fn(), enqueue: vi.fn() }, + pluginScanProcessor: { clearPending: vi.fn(), enqueue: vi.fn() }, _log: vi.fn(), storeCustomisationFileV2: vi.fn(async () => true), }); diff --git a/src/features/ConfigSync/CmdConfigSync.command.unit.spec.ts b/src/features/ConfigSync/CmdConfigSync.command.unit.spec.ts index 5f524f8a..6124c4bd 100644 --- a/src/features/ConfigSync/CmdConfigSync.command.unit.spec.ts +++ b/src/features/ConfigSync/CmdConfigSync.command.unit.spec.ts @@ -91,7 +91,7 @@ describe("ConfigSync commands", () => { expect(showPluginSyncModal).toHaveBeenCalledOnce(); }); - it("cancels the pending configuration Notice before releasing its owned UI", () => { + it("cancels the pending configuration Notice before releasing its owned UI", async () => { const notices = { hide: vi.fn() }; const periodicPluginSweepProcessor = { disable: vi.fn() }; const configSync = Object.create(ConfigSync.prototype) as ConfigSync; @@ -102,9 +102,13 @@ describe("ConfigSync commands", () => { }, }, periodicPluginSweepProcessor, + pluginList: [], + pluginListGeneration: 0, + loadedManifest_mTime: new Map(), + pluginScanProcessor: { clearPending: vi.fn(), stop: vi.fn(async () => {}) }, }); - configSync.onunload(); + await configSync.onunload(); expect(cancelTask).toHaveBeenCalledWith("config-sync:updated-configuration"); expect(notices.hide).toHaveBeenCalledWith("config-sync:updated-configuration"); diff --git a/src/features/ConfigSync/CmdConfigSync.queue.unit.spec.ts b/src/features/ConfigSync/CmdConfigSync.queue.unit.spec.ts index bf93190b..98f4000d 100644 --- a/src/features/ConfigSync/CmdConfigSync.queue.unit.spec.ts +++ b/src/features/ConfigSync/CmdConfigSync.queue.unit.spec.ts @@ -20,7 +20,11 @@ vi.mock("@/common/utils.ts", () => ({ isPluginMetadata: vi.fn(), scheduleTask: vi.fn(), })); -vi.mock("@/common/PeriodicProcessor.ts", () => ({ PeriodicProcessor: class PeriodicProcessor {} })); +vi.mock("@/common/PeriodicProcessor.ts", () => ({ + PeriodicProcessor: class PeriodicProcessor { + disable = vi.fn(); + }, +})); vi.mock("@/common/events.ts", () => ({ EVENT_REQUEST_OPEN_PLUGIN_SYNC_DIALOG: "open-plugin-sync", eventHub: { onEvent: vi.fn() }, @@ -29,7 +33,12 @@ vi.mock("@/common/translation", () => ({ $msg: (message: string) => message })); vi.mock("@/common/obsidianCommunityPlugins.ts", () => ({ getObsidianCommunityPluginManager: vi.fn() })); import { get } from "svelte/store"; -import { DEFAULT_SETTINGS, type FilePathWithPrefix } from "@vrtmrz/livesync-commonlib/compat/common/types"; +import { getDocData } from "@vrtmrz/livesync-commonlib/compat/common/utils"; +import { + DEFAULT_SETTINGS, + type AnyEntry, + type FilePathWithPrefix, +} from "@vrtmrz/livesync-commonlib/compat/common/types"; import { createOptionalFileSyncDatabase, optionalFileEntry } from "../../../test/fixtures/optionalFileSyncDatabase"; import { ConfigSync, pluginIsEnumerating, pluginList } from "./CmdConfigSync"; @@ -44,16 +53,63 @@ function deferred() { } async function waitForCondition(predicate: () => boolean, message: string) { - for (let attempt = 0; attempt < 30; attempt++) { + for (let attempt = 0; attempt < 100; attempt++) { if (predicate()) return; await Promise.resolve(); } throw new Error(message); } -afterEach(() => { +function queueEntry(path: string, data: unknown = {}): AnyEntry { + const documentPath = path as FilePathWithPrefix; + return { + _id: documentPath, + path: documentPath, + mtime: 2_000, + data: [JSON.stringify(data)], + } as unknown as AnyEntry; +} + +function legacyCatalogueEntry(path: string, name: string, mtime = 2_000): AnyEntry { + return queueEntry(path, { + category: "PLUGIN_MAIN", + name, + term: "device", + mtime, + files: [ + { + filename: `plugins/${name}/main.js`, + data: [`legacy source for ${name}`], + mtime, + size: name.length, + hash: `hash-${name}`, + }, + ], + }); +} + +function createQueueConfigSync( + settings: typeof DEFAULT_SETTINGS, + localDatabase: unknown, + services: unknown = { + API: { addLog: vi.fn(), addRibbonIcon: vi.fn() }, + path: { getPath: (entry: { path: string }) => entry.path }, + context: { notices: { hide: vi.fn() } }, + } +) { + vi.spyOn(ConfigSync.prototype, "onBindFunction").mockImplementation(() => {}); + const configSync = new ConfigSync({ + settings, + localDatabase, + services, + } as unknown as ConstructorParameters[0]); + queueOwners.push(configSync); + return configSync; +} + +afterEach(async () => { for (const configSync of queueOwners.splice(0)) { - configSync.pluginScanProcessor.terminateAll(); + await configSync.pluginScanProcessor.stop(); } vi.restoreAllMocks(); pluginList.set([]); @@ -62,6 +118,356 @@ afterEach(() => { }); describe("ConfigSync legacy catalogue queue", () => { + it("starts duplicate entries in FIFO order with ten concurrent items", async () => { + const settings = { ...DEFAULT_SETTINGS, usePluginSync: true, usePluginSyncV2: true }; + const configSync = createQueueConfigSync(settings, {}); + const paths = [ + "ix:device/PLUGIN_MAIN/first.md", + "ix:device/PLUGIN_MAIN/second.md", + "ix:device/PLUGIN_MAIN/duplicate.md", + "ix:device/PLUGIN_MAIN/third.md", + "ix:device/PLUGIN_MAIN/duplicate.md", + "ix:device/PLUGIN_MAIN/fourth.md", + "ix:device/PLUGIN_MAIN/fifth.md", + "ix:device/PLUGIN_MAIN/sixth.md", + "ix:device/PLUGIN_MAIN/seventh.md", + "ix:device/PLUGIN_MAIN/eighth.md", + "ix:device/PLUGIN_MAIN/ninth.md", + "ix:device/PLUGIN_MAIN/tenth.md", + ]; + const gate = deferred(); + const started: string[] = []; + let active = 0; + let maximumActive = 0; + vi.spyOn(configSync, "migrateV1ToV2").mockImplementation(async (_showMessage, entry) => { + started.push(entry.path); + active++; + maximumActive = Math.max(maximumActive, active); + await gate.promise; + active--; + }); + + try { + for (const path of paths) + configSync.pluginScanProcessor.enqueue({ entry: queueEntry(path), generation: 0 }); + await waitForCondition(() => started.length === 10, "The first ten queue items did not start"); + + expect(started).toEqual(paths.slice(0, 10)); + expect(maximumActive).toBe(10); + + gate.resolve(); + await expect(configSync.pluginScanProcessor.waitForIdle()).resolves.toBeUndefined(); + expect(started).toEqual(paths); + } finally { + gate.resolve(); + } + }); + + it("continues with queued items after a migration throws", async () => { + const settings = { ...DEFAULT_SETTINGS, usePluginSync: true, usePluginSyncV2: true }; + const configSync = createQueueConfigSync(settings, {}); + const paths = Array.from({ length: 11 }, (_, index) => `ix:device/PLUGIN_MAIN/item-${index}.md`); + const gate = deferred(); + const started: string[] = []; + vi.spyOn(configSync, "migrateV1ToV2").mockImplementation(async (_showMessage, entry) => { + started.push(entry.path); + if (entry.path === paths[0]) throw new Error("migration failed"); + if (paths.slice(1, 10).includes(entry.path)) await gate.promise; + }); + + try { + for (const path of paths) + configSync.pluginScanProcessor.enqueue({ entry: queueEntry(path), generation: 0 }); + await waitForCondition(() => started.length === 10, "The first ten queue items did not start"); + await waitForCondition(() => started.length === 11, "The item after the failed migration did not start"); + + gate.resolve(); + await expect(configSync.pluginScanProcessor.waitForIdle()).resolves.toBeUndefined(); + expect(started).toEqual(paths); + } finally { + gate.resolve(); + } + }); + + it("does not restore an old read result after reloading a newer catalogue", async () => { + vi.useFakeTimers(); + const oldPath = "ix:device/PLUGIN_MAIN/old" as FilePathWithPrefix; + const newerPath = "ix:device/PLUGIN_MAIN/newer" as FilePathWithPrefix; + const oldEntry = legacyCatalogueEntry(oldPath, "old"); + const newerEntry = legacyCatalogueEntry(newerPath, "newer", 3_000); + const oldRead = deferred(); + let scanCount = 0; + let oldReadStarted = false; + const localDatabase = { + findEntries: vi.fn(async function* () { + for (const entry of scanCount++ === 0 ? [oldEntry] : [newerEntry]) yield entry; + }), + getDBEntry: vi.fn(async (path: FilePathWithPrefix) => { + if (path === oldPath) { + oldReadStarted = true; + await oldRead.promise; + return oldEntry; + } + return newerEntry; + }), + }; + const settings = { ...DEFAULT_SETTINGS, usePluginSync: true, usePluginSyncV2: false }; + const configSync = createQueueConfigSync(settings, localDatabase); + + try { + await configSync.updatePluginList(false); + await vi.advanceTimersByTimeAsync(100); + await waitForCondition(() => oldReadStarted, "The old catalogue read did not start"); + + await configSync.reloadPluginList(false); + await vi.advanceTimersByTimeAsync(100); + expect(configSync.pluginList.map((entry) => entry.name)).toEqual(["newer"]); + + oldRead.resolve(); + await expect(configSync.pluginScanProcessor.waitForIdle()).resolves.toBeUndefined(); + expect(configSync.pluginList.map((entry) => entry.name)).toEqual(["newer"]); + expect(get(pluginList).map((entry) => entry.name)).toEqual(["newer"]); + } finally { + oldRead.resolve(); + await vi.advanceTimersByTimeAsync(100); + } + }); + + it.each(["catalogue update", "settings application", "temporary suspension"] as const)( + "does not restore an active legacy read after disabling through %s and re-enabling", + async (route) => { + vi.useFakeTimers(); + const oldPath = "ix:device/PLUGIN_MAIN/old" as FilePathWithPrefix; + const oldEntry = legacyCatalogueEntry(oldPath, "old"); + const oldRead = deferred(); + let oldReadStarted = false; + let scanCount = 0; + const localDatabase = { + findEntries: vi.fn(async function* () { + if (scanCount++ === 0) yield oldEntry; + }), + getDBEntry: vi.fn(async () => { + oldReadStarted = true; + await oldRead.promise; + return oldEntry; + }), + }; + const settings = { ...DEFAULT_SETTINGS, usePluginSync: true, usePluginSyncV2: false }; + const configSync = createQueueConfigSync(settings, localDatabase); + + try { + await configSync.updatePluginList(false); + await vi.advanceTimersByTimeAsync(100); + await waitForCondition(() => oldReadStarted, "The active legacy read did not start"); + + settings.usePluginSync = false; + if (route === "catalogue update") { + await configSync.updatePluginList(false); + } else if (route === "settings application") { + await configSync._everyRealizeSettingSyncMode(); + } else { + await ( + configSync as unknown as { _allSuspendExtraSync(): Promise } + )._allSuspendExtraSync(); + } + settings.usePluginSync = true; + await configSync.updatePluginList(false); + + oldRead.resolve(); + await expect(configSync.pluginScanProcessor.waitForIdle()).resolves.toBeUndefined(); + expect(configSync.pluginList).toEqual([]); + expect(get(pluginList)).toEqual([]); + } finally { + oldRead.resolve(); + await vi.advanceTimersByTimeAsync(100); + } + } + ); + + it("stops admission on unload and joins a read without changing the replacement instance's list", async () => { + vi.useFakeTimers(); + const entry = legacyCatalogueEntry("ix:device/PLUGIN_MAIN/old", "old"); + const read = deferred(); + const localDatabase = { + findEntries: vi.fn(async function* () { + yield entry; + }), + getDBEntry: vi.fn(async () => { + await read.promise; + return entry; + }), + }; + const settings = { ...DEFAULT_SETTINGS, usePluginSync: true, usePluginSyncV2: false }; + const configSync = createQueueConfigSync(settings, localDatabase); + try { + await configSync.updatePluginList(false); + await vi.advanceTimersByTimeAsync(100); + expect(localDatabase.getDBEntry).toHaveBeenCalledOnce(); + let stopped = false; + const done = configSync.onunload().then(() => { + stopped = true; + }); + const replacement = { name: "replacement" } as (typeof configSync.pluginList)[number]; + pluginList.set([replacement]); + await configSync.updatePluginList(false); + await configSync.reloadPluginList(false); + await vi.advanceTimersByTimeAsync(100); + expect(localDatabase.findEntries).toHaveBeenCalledOnce(); + expect(stopped).toBe(false); + expect(configSync.pluginScanProcessor.remaining).toBe(1); + read.resolve(); + await done; + expect(configSync.pluginList).toEqual([]); + expect(get(pluginList)).toEqual([replacement]); + expect(configSync.pluginScanProcessor.remaining).toBe(0); + } finally { + read.resolve(); + await vi.advanceTimersByTimeAsync(100); + } + }); + + it("cancels unstarted legacy reads when the first ten are blocked and the module is disabled", async () => { + const paths = Array.from({ length: 12 }, (_, index) => `ix:device/PLUGIN_MAIN/item-${index}`); + const entries = paths.map((path, index) => legacyCatalogueEntry(path, `item-${index}`)); + const gates = entries.slice(0, 10).map(() => deferred()); + const started: string[] = []; + const localDatabase = { + findEntries: vi.fn(async function* () { + for (const entry of entries) yield entry; + }), + getDBEntry: vi.fn(async (path: FilePathWithPrefix) => { + const index = paths.indexOf(path); + started.push(path); + if (index >= 0 && index < gates.length) await gates[index].promise; + return entries[index]; + }), + }; + const settings = { ...DEFAULT_SETTINGS, usePluginSync: true, usePluginSyncV2: false }; + const configSync = createQueueConfigSync(settings, localDatabase); + + try { + await configSync.updatePluginList(false); + await waitForCondition(() => started.length === 10, "The first ten legacy reads did not start"); + + settings.usePluginSync = false; + await configSync.updatePluginList(false); + for (const gate of gates) gate.resolve(); + await expect(configSync.pluginScanProcessor.waitForIdle()).resolves.toBeUndefined(); + + expect(started).toEqual(paths.slice(0, 10)); + } finally { + for (const gate of gates) gate.resolve(); + } + }); + + it("finishes a started V1-to-V2 migration but does not start pending migrations after disable", async () => { + const fixture = await createOptionalFileSyncDatabase(); + const v1Path = "ix:device/PLUGIN_MAIN/migrate.md" as FilePathWithPrefix; + const v2Path = "ix:device/PLUGIN_MAIN/migrate%main.js" as FilePathWithPrefix; + const source = "module.exports = 'migrated';"; + const data = { + category: "PLUGIN_MAIN", + name: "migrate", + term: "device", + mtime: 2_000, + files: [ + { + filename: "plugins/migrate/main.js", + data: [source], + mtime: 2_000, + size: source.length, + hash: "migration-source-hash", + }, + ], + }; + const settings = fixture.settings; + const configSync = createQueueConfigSync(settings, fixture.localDatabase, { + ...fixture.services, + API: { ...fixture.services.API, addRibbonIcon: vi.fn() }, + }); + const pendingGate = deferred(); + const putGate = deferred(); + const migrationFinished = deferred(); + const started: string[] = []; + let v2PutStarted = false; + + try { + await expect( + fixture.localDatabase.putDBEntry(optionalFileEntry(v1Path, JSON.stringify(data))) + ).resolves.toMatchObject({ ok: true }); + + const loadedLegacy = { + ...optionalFileEntry(v1Path, JSON.stringify(data)), + data: [JSON.stringify(data)], + } as unknown as Awaited>; + let migrationRead = false; + const read = fixture.localDatabase.getDBEntry.bind(fixture.localDatabase); + vi.spyOn(fixture.localDatabase, "getDBEntry").mockImplementation(async (path, ...args) => { + if (path === v1Path && !migrationRead) { + migrationRead = true; + return loadedLegacy; + } + return await read(path, ...args); + }); + const migrate = configSync.migrateV1ToV2.bind(configSync); + vi.spyOn(configSync, "migrateV1ToV2").mockImplementation(async (showMessage, entry) => { + started.push(entry.path); + if (entry.path === v1Path) { + try { + await migrate(showMessage, entry); + } finally { + migrationFinished.resolve(); + } + return; + } + await pendingGate.promise; + }); + const put = fixture.localDatabase.putDBEntry.bind(fixture.localDatabase); + vi.spyOn(fixture.localDatabase, "putDBEntry").mockImplementation(async (...args) => { + if ((args[0] as { path?: string }).path === v2Path) { + v2PutStarted = true; + await putGate.promise; + } + return await put(...args); + }); + + const paths = [ + v1Path, + ...Array.from( + { length: 10 }, + (_, index) => `ix:device/PLUGIN_MAIN/pending-${index}.md` as FilePathWithPrefix + ), + ]; + for (const path of paths) + configSync.pluginScanProcessor.enqueue({ entry: queueEntry(path), generation: 0 }); + await waitForCondition( + () => started.length === 10, + `The first ten migrations did not start: ${started.join(", ")}` + ); + await waitForCondition(() => v2PutStarted, "The migration did not reach its write"); + + settings.usePluginSync = false; + await configSync.updatePluginList(false); + putGate.resolve(); + await migrationFinished.promise; + expect.soft(started).toEqual(paths.slice(0, 10)); + + pendingGate.resolve(); + await expect(configSync.pluginScanProcessor.waitForIdle()).resolves.toBeUndefined(); + expect.soft(started).toEqual(paths.slice(0, 10)); + + const migrated = await fixture.localDatabase.getDBEntry(v2Path, undefined, false, false); + expect(migrated).not.toBe(false); + if (migrated !== false) expect(getDocData(migrated.data)).toContain(source); + await expect(fixture.localDatabase.getDBEntry(v1Path)).resolves.toBe(false); + } finally { + putGate.resolve(); + pendingGate.resolve(); + await configSync.pluginScanProcessor.waitForIdle(); + await fixture.dispose(); + } + }); + it("loads legacy catalogue contents through the database with native timers", async () => { const fixture = await createOptionalFileSyncDatabase(); const path = "ix:device/PLUGIN_MAIN/native-timers" as FilePathWithPrefix; @@ -89,7 +495,7 @@ describe("ConfigSync legacy catalogue queue", () => { } as unknown as ConstructorParameters[0]); queueOwners.push(configSync); await configSync.updatePluginList(false); - expect(await configSync.pluginScanProcessor.waitForAllProcessed(2_000)).toBe(true); + await expect(configSync.pluginScanProcessor.waitForIdle()).resolves.toBeUndefined(); expect(configSync.pluginList).toEqual([ { ...data, @@ -100,7 +506,7 @@ describe("ConfigSync legacy catalogue queue", () => { expect(get(pluginList)).toEqual(configSync.pluginList); expect(configSync.pluginScanProcessor.remaining).toBe(0); } finally { - configSync?.pluginScanProcessor.terminateAll(); + await configSync?.pluginScanProcessor.stop(); await fixture.dispose(); } }); @@ -148,7 +554,6 @@ describe("ConfigSync legacy catalogue queue", () => { try { await configSync.updatePluginList(false); - expect(configSync.pluginScanProcessor.delay).toBe(100); expect(configSync.pluginScanProcessor.remaining).toBe(1); expect(configSync.pluginList).toEqual([]); expect(get(pluginIsEnumerating)).toBe(false); @@ -159,9 +564,9 @@ describe("ConfigSync legacy catalogue queue", () => { await vi.advanceTimersByTimeAsync(1); expect(loadHasStarted).toBe(true); expect(configSync.pluginScanProcessor.remaining).toBe(1); - const drain = configSync.pluginScanProcessor.waitForAllProcessed(); - void drain.then((result) => { - drained = result; + const drain = configSync.pluginScanProcessor.waitForIdle(); + void drain.then(() => { + drained = true; }); readGate.resolve(); diff --git a/src/features/ConfigSync/CmdConfigSync.ts b/src/features/ConfigSync/CmdConfigSync.ts index e6f61f43..55b11869 100644 --- a/src/features/ConfigSync/CmdConfigSync.ts +++ b/src/features/ConfigSync/CmdConfigSync.ts @@ -55,7 +55,6 @@ import { stripAllPrefixes } from "@vrtmrz/livesync-commonlib/compat/string_and_b import { cancelTask, EVEN, isCustomisationSyncMetadata, isPluginMetadata, scheduleTask } from "@/common/utils.ts"; import { PeriodicProcessor } from "@/common/PeriodicProcessor.ts"; import { JsonResolveModal } from "@/features/HiddenFileCommon/JsonResolveModal.ts"; -import { QueueProcessor } from "octagonal-wheels/concurrency/processor"; import { pluginScanningCount } from "@vrtmrz/livesync-commonlib/compat/mock_and_interop/stores"; import { base64ToArrayBuffer, base64ToString } from "octagonal-wheels/binary/base64"; import { ConflictResolveModal } from "@/modules/features/InteractiveConflictResolving/ConflictResolveModal.ts"; @@ -74,6 +73,7 @@ import { createModificationHandler } from "octagonal-wheels/modification"; import type { HandlerOutcome } from "@vrtmrz/livesync-commonlib/compat/services/lib/HandlerUtils"; import type { OptionalFileEventOptions } from "@vrtmrz/livesync-commonlib/compat/services/base/IService"; import { PendingConfigSaves } from "./PendingConfigSaves"; +import { ConfigScanQueue } from "./ConfigScanQueue"; const d = "\u200b"; const d2 = "\n"; @@ -394,13 +394,6 @@ export type PluginDataEx = { }; export class ConfigSync extends LiveSyncCommands { - constructor(core: LiveSyncCore) { - super(core); - pluginScanningCount.onChanged((e) => { - const total = e.value; - pluginIsEnumerating.set(total != 0); - }); - } get configDir() { return this.core.services.API.getSystemConfigDir(); } @@ -423,6 +416,7 @@ export class ConfigSync extends LiveSyncCommands { pluginList: IPluginDataExDisplay[] = []; private pluginListGeneration = 0; + private unloaded = false; showPluginSyncModal() { if (!this.isThisModuleEnabled()) { return; @@ -441,11 +435,18 @@ export class ConfigSync extends LiveSyncCommands { this.pluginDialog = undefined; } } - onunload() { + onunload(): Promise { + this.unloaded = true; + this.resetPluginList(); + const stopped = this.pluginScanProcessor.stop(); + pluginScanningCount.value = 0; + pluginIsEnumerating.set(false); + cancelTask("updatePluginListV2"); cancelTask(UPDATED_CONFIGURATION_NOTICE_KEY); this.hidePluginSyncModal(); this.periodicPluginSweepProcessor?.disable(); this.services.context.notices.hide(UPDATED_CONFIGURATION_NOTICE_KEY); + return stopped; } addRibbonIcon = this.services.API.addRibbonIcon.bind(this.services.API); onload() { @@ -542,11 +543,21 @@ export class ConfigSync extends LiveSyncCommands { q?.toggleClass("sls-hidden", !this.isThisModuleEnabled()); return Promise.resolve(true); } - async reloadPluginList(showMessage: boolean) { + private isCurrentPluginList(generation: number): boolean { + return !this.unloaded && generation === this.pluginListGeneration && this.isThisModuleEnabled(); + } + + private resetPluginList(): void { this.pluginListGeneration++; + this.pluginScanProcessor.clearPending(); this.pluginList = []; this.loadedManifest_mTime.clear(); pluginList.set(this.pluginList); + } + + async reloadPluginList(showMessage: boolean) { + if (this.unloaded) return; + this.resetPluginList(); await this.updatePluginList(showMessage); } async loadPluginData(path: FilePathWithPrefix): Promise { @@ -581,43 +592,37 @@ export class ConfigSync extends LiveSyncCommands { return false; } - pluginScanProcessor = new QueueProcessor( - async (v: AnyEntry[]) => { - const plugin = v[0]; - if (this.useV2) { - await this.migrateV1ToV2(false, plugin); - return []; - } - const path = plugin.path || this.getPath(plugin); - const oldEntry = this.pluginList.find((e) => e.documentPath == path); - if (oldEntry && oldEntry.mtime == plugin.mtime) return []; + pluginScanProcessor = new ConfigScanQueue<{ entry: AnyEntry; generation: number }>({ + process: async ({ entry: plugin, generation }) => { + if (!this.isCurrentPluginList(generation)) return; try { + if (this.useV2) { + // Once started, migration owns its writes through completion. + await this.migrateV1ToV2(false, plugin); + return; + } + const path = plugin.path || this.getPath(plugin); + const oldEntry = this.pluginList.find((entry) => entry.documentPath == path); + if (oldEntry && oldEntry.mtime == plugin.mtime) return; const pluginData = await this.loadPluginData(path); + if (!this.isCurrentPluginList(generation)) return; if (pluginData) { - let newList = [...this.pluginList]; - newList = newList.filter((x) => x.documentPath != pluginData.documentPath); + const newList = this.pluginList.filter((entry) => entry.documentPath != pluginData.documentPath); newList.push(pluginData); this.pluginList = newList; pluginList.set(newList); } - // Failed to load - return []; } catch (ex) { - this._log(`Something happened at enumerating customization :${path}`, LOG_LEVEL_NOTICE); + this._log(`Something happened at enumerating customization :${plugin.path}`, LOG_LEVEL_NOTICE); this._log(ex, LOG_LEVEL_VERBOSE); } - return []; }, - { - suspended: false, - batchSize: 1, - concurrentLimit: 10, - delay: 100, - yieldThreshold: 10, - maintainDelay: false, - totalRemainingReactiveSource: pluginScanningCount, - } - ).startPipeline(); + onProgress: (remaining) => { + if (this.unloaded) return; + pluginScanningCount.value = remaining; + pluginIsEnumerating.set(remaining !== 0); + }, + }); filenameToUnifiedKey(path: string, termOverRide?: string) { const term = termOverRide || this.services.setting.getDeviceAndVaultName(); @@ -740,7 +745,7 @@ export class ConfigSync extends LiveSyncCommands { async updatePluginListV2(showMessage: boolean, unifiedFilenameWithKey: FilePathWithPrefix): Promise { const generation = this.pluginListGeneration; - const isCurrent = () => generation === this.pluginListGeneration && this.isThisModuleEnabled(); + const isCurrent = () => this.isCurrentPluginList(generation); try { this.updatingV2Count++; pluginV2Progress.set(this.updatingV2Count); @@ -855,11 +860,9 @@ export class ConfigSync extends LiveSyncCommands { } async updatePluginList(showMessage: boolean, updatedDocumentPath?: FilePathWithPrefix): Promise { + if (this.unloaded) return; if (!this.isThisModuleEnabled()) { - this.pluginListGeneration++; - this.pluginScanProcessor.clearQueue(); - this.pluginList = []; - pluginList.set(this.pluginList); + this.resetPluginList(); return; } const generation = this.pluginListGeneration; @@ -875,7 +878,7 @@ export class ConfigSync extends LiveSyncCommands { }) : this.localDatabase.findEntries(ICXHeader + "", `${ICXHeader}\u{10ffff}`, { include_docs: true }); for await (const v of plugins) { - if (generation !== this.pluginListGeneration || !this.isThisModuleEnabled()) return; + if (!this.isCurrentPluginList(generation)) return; if (v.deleted || v._deleted) continue; if (v.path.indexOf("%") !== -1) { fireAndForget(() => this.updatePluginListV2(showMessage, v.path)); @@ -884,7 +887,7 @@ export class ConfigSync extends LiveSyncCommands { const path = v.path || this.getPath(v); if (updatedDocumentPath && updatedDocumentPath != path) continue; - this.pluginScanProcessor.enqueue(v); + this.pluginScanProcessor.enqueue({ entry: v, generation }); } } finally { pluginIsEnumerating.set(false); @@ -1206,6 +1209,7 @@ export class ConfigSync extends LiveSyncCommands { return true; } _everyRealizeSettingSyncMode(): Promise { + if (!this.isThisModuleEnabled()) this.resetPluginList(); // applySettings calls onResuming after this handler when synchronisation is active. // That handler scans and restarts the sweep; doing so here would scan twice. this.periodicPluginSweepProcessor?.disable(); @@ -1742,6 +1746,7 @@ export class ConfigSync extends LiveSyncCommands { this.core.settings.usePluginSync = false; this.core.settings.autoSweepPlugins = false; } + this.resetPluginList(); return Promise.resolve(true); } diff --git a/src/features/ConfigSync/ConfigScanQueue.ts b/src/features/ConfigSync/ConfigScanQueue.ts new file mode 100644 index 00000000..dd960dd0 --- /dev/null +++ b/src/features/ConfigSync/ConfigScanQueue.ts @@ -0,0 +1,176 @@ +import { compatGlobal, type CompatTimeoutHandle } from "@vrtmrz/livesync-commonlib/compat/common/coreEnvFunctions"; +import { createNotifier } from "octagonal-wheels/concurrency/notifier"; +import { forEachInScope } from "octagonal-wheels/concurrency/scopedIteration"; +import { withScope, type Awaitable } from "octagonal-wheels/resource/scope"; + +interface ConfigScanQueueOptions { + /** The caller logs recoverable item failures and decides whether to continue. */ + process: (item: T) => Awaitable; + onProgress?: (remaining: number) => void; +} + +type PendingScan = { value: T }; + +/** Owns FIFO admission, the initial scan delay, and completion for one catalogue. */ +export class ConfigScanQueue { + private pending: (PendingScan | undefined)[] = []; + private head = 0; + private active = 0; + private ready = false; + private stopped = false; + private stoppingAfterFailure = false; + private readonly failures: unknown[] = []; + private timer?: CompatTimeoutHandle; + private running?: Promise; + private readonly changed = createNotifier(); + + constructor(private readonly options: ConfigScanQueueOptions) {} + + private get pendingCount() { + return this.pending.length - this.head; + } + + get remaining(): number { + return this.pendingCount + this.active; + } + + enqueue(value: T): void { + if (!this.canDispatch()) throw new Error("The customisation scan queue is closed."); + this.pending.push({ value }); + this.reportProgress(); + this.schedule(); + } + + /** Discard unstarted scans; active reads and migrations retain their slots. */ + clearPending(): void { + this.pending = []; + this.head = 0; + this.ready = false; + this.clearTimer(); + this.reportProgress(); + } + + /** Stop admission and join active work. A processor must not await its own stop. */ + stop(): Promise { + this.stopped = true; + this.clearPending(); + return this.waitForIdle(); + } + + /** Include later arrivals until the queue is idle; this does not close admission. */ + async waitForIdle(): Promise { + this.schedule(); + while (this.running || this.timer !== undefined) await this.changed.wait(); + if (this.failures.length === 1) throw this.failures[0]; + if (this.failures.length > 1) { + throw Object.assign(new Error("Customisation scans failed."), { errors: [...this.failures] }); + } + } + + private canDispatch(): boolean { + return !this.stopped && !this.stoppingAfterFailure; + } + + private clearTimer(): void { + if (this.timer === undefined) return; + compatGlobal.clearTimeout(this.timer); + this.timer = undefined; + } + + private recordFailure(error: unknown): void { + this.stoppingAfterFailure = true; + this.failures.push(error); + this.clearTimer(); + this.changed.notify(); + } + + private reportProgress(): void { + if (this.failures.length === 0) { + try { + this.options.onProgress?.(this.remaining); + } catch (error) { + this.recordFailure(error); + } + } + this.changed.notify(); + } + + private claim(entry: PendingScan): boolean { + if (!this.canDispatch() || this.pending[this.head] !== entry) return false; + this.pending[this.head++] = undefined; + if (this.pendingCount === 0) { + this.pending = []; + this.head = 0; + this.ready = false; + } else if (this.head >= 1024 && this.head * 2 >= this.pending.length) { + this.pending = this.pending.slice(this.head); + this.head = 0; + } + return true; + } + + private async *availableWork() { + while (this.canDispatch()) { + const entry = this.pending[this.head]; + if (this.ready && entry) { + // The callback claims the item so clearing a pulled item also cancels it. + yield entry; + } else if (this.active > 0) { + // Keep free slots available to later arrivals while a slow scan finishes. + await this.changed.wait(); + } else { + return; + } + } + } + + private schedule(): void { + if (!this.canDispatch() || this.pendingCount === 0) return; + if (!this.ready) { + if (this.pendingCount > 10) { + this.ready = true; + this.clearTimer(); + } else if (this.timer === undefined) { + this.timer = compatGlobal.setTimeout(() => { + this.timer = undefined; + this.ready = true; + this.schedule(); + this.changed.notify(); + }, 100); + } + } + if (!this.ready || this.running) return; + this.running = Promise.resolve() + .then(() => + withScope(undefined, (parent) => + forEachInScope(parent, this.availableWork(), { concurrency: 10 }, async (entry, child) => { + if (!this.claim(entry)) return; + this.active++; + child.defer(() => { + this.active--; + this.reportProgress(); + }); + this.reportProgress(); + if (!this.canDispatch()) return; + try { + await this.options.process(entry.value); + } catch (error) { + this.stoppingAfterFailure = true; + this.clearTimer(); + this.changed.notify(); + throw error; + } + }) + ) + ) + .then( + (): void => undefined, + (error: unknown) => this.recordFailure(error) + ) + .finally(() => { + this.running = undefined; + this.schedule(); + this.changed.notify(); + }); + } +} diff --git a/src/features/ConfigSync/ConfigScanQueue.unit.spec.ts b/src/features/ConfigSync/ConfigScanQueue.unit.spec.ts new file mode 100644 index 00000000..c1182d2b --- /dev/null +++ b/src/features/ConfigSync/ConfigScanQueue.unit.spec.ts @@ -0,0 +1,331 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { ConfigScanQueue } from "./ConfigScanQueue"; + +const queues: { stop(): Promise }[] = []; +const releases: (() => void)[] = []; + +function gate() { + let resolve!: () => void; + const promise = new Promise((done) => { + resolve = done; + }); + releases.push(resolve); + return { promise, resolve }; +} + +function queue(options: ConstructorParameters>[0]) { + const result = new ConfigScanQueue(options); + queues.push(result); + return result; +} + +beforeEach(() => vi.useFakeTimers()); +afterEach(async () => { + for (const release of releases.splice(0)) release(); + await Promise.all(queues.splice(0).map((item) => item.stop().catch(() => {}))); + vi.useRealTimers(); +}); + +describe("ConfigScanQueue", () => { + it("waits from the first arrival without extending the delay for later arrivals", async () => { + const seen: string[] = []; + const scans = queue({ + process: (item) => { + seen.push(item); + }, + }); + scans.enqueue("first"); + await vi.advanceTimersByTimeAsync(50); + scans.enqueue("second"); + await vi.advanceTimersByTimeAsync(49); + expect(seen).toEqual([]); + await vi.advanceTimersByTimeAsync(1); + await scans.waitForIdle(); + expect(seen).toEqual(["first", "second"]); + expect(vi.getTimerCount()).toBe(0); + }); + + it("bypasses the delay only when pending work exceeds ten items", async () => { + const seen: number[] = []; + const scans = queue({ + process: (item) => { + seen.push(item); + }, + }); + for (let item = 0; item < 10; item++) scans.enqueue(item); + await vi.advanceTimersByTimeAsync(0); + expect(seen).toEqual([]); + scans.enqueue(10); + await vi.advanceTimersByTimeAsync(0); + await scans.waitForIdle(); + expect(seen).toEqual(Array.from({ length: 11 }, (_, index) => index)); + expect(vi.getTimerCount()).toBe(0); + }); + + it("keeps duplicate items in FIFO order and admits at most ten active scans", async () => { + const finish = Array.from({ length: 13 }, () => gate()); + const input = Array.from({ length: 13 }, (_, index) => index % 3); + const seen: number[] = []; + let active = 0; + let maximum = 0; + const progress: number[] = []; + const scans = queue({ + process: async (item) => { + const index = seen.length; + seen.push(item); + maximum = Math.max(maximum, ++active); + await finish[index].promise; + active--; + }, + onProgress: (remaining) => progress.push(remaining), + }); + input.forEach((item) => scans.enqueue(item)); + await vi.advanceTimersByTimeAsync(0); + expect(seen).toEqual(input.slice(0, 10)); + expect(scans.remaining).toBe(13); + finish[3].resolve(); + await vi.advanceTimersByTimeAsync(0); + expect(seen).toEqual(input.slice(0, 11)); + finish.forEach((item) => item.resolve()); + await scans.waitForIdle(); + expect(seen).toEqual(input); + expect(maximum).toBe(10); + expect(scans.remaining).toBe(0); + expect(progress[progress.length - 1]).toBe(0); + }); + + it("starts a later arrival in a free slot before an earlier slow scan finishes", async () => { + const slow = gate(); + const seen: string[] = []; + const scans = queue({ + process: async (item) => { + seen.push(item); + if (item === "slow") await slow.promise; + }, + }); + scans.enqueue("slow"); + await vi.advanceTimersByTimeAsync(100); + let idle = false; + const done = scans.waitForIdle().then(() => { + idle = true; + }); + scans.enqueue("later"); + await vi.advanceTimersByTimeAsync(100); + expect(seen).toEqual(["slow", "later"]); + expect(idle).toBe(false); + slow.resolve(); + await done; + expect(scans.remaining).toBe(0); + }); + + it("clears delayed scans and gives replacement work its own delay", async () => { + const seen: string[] = []; + const scans = queue({ + process: (item) => { + seen.push(item); + }, + }); + scans.enqueue("old"); + await vi.advanceTimersByTimeAsync(50); + scans.clearPending(); + expect(scans.remaining).toBe(0); + expect(vi.getTimerCount()).toBe(0); + await scans.waitForIdle(); + scans.enqueue("new"); + await vi.advanceTimersByTimeAsync(99); + expect(seen).toEqual([]); + await vi.advanceTimersByTimeAsync(1); + await scans.waitForIdle(); + expect(seen).toEqual(["new"]); + }); + + it("discards all unstarted work while active scans remain tracked", async () => { + const finish = gate(); + const seen: number[] = []; + const scans = queue({ + process: async (item) => { + seen.push(item); + await finish.promise; + }, + }); + for (let item = 0; item < 50; item++) scans.enqueue(item); + await vi.advanceTimersByTimeAsync(0); + expect(seen).toHaveLength(10); + scans.clearPending(); + expect(scans.remaining).toBe(10); + let idle = false; + const done = scans.waitForIdle().then(() => { + idle = true; + }); + await vi.advanceTimersByTimeAsync(0); + expect(idle).toBe(false); + finish.resolve(); + await done; + expect(seen).toEqual(Array.from({ length: 10 }, (_, index) => index)); + expect(scans.remaining).toBe(0); + }); + + it("stops while waiting for input, joins active work, and rejects later admission", async () => { + const finish = gate(); + const seen: number[] = []; + const scans = queue({ + process: async (item) => { + seen.push(item); + await finish.promise; + }, + }); + scans.enqueue(1); + await vi.advanceTimersByTimeAsync(100); + scans.enqueue(2); + let stopped = false; + const done = scans.stop().then(() => { + stopped = true; + }); + expect(() => scans.enqueue(3)).toThrow("closed"); + expect(vi.getTimerCount()).toBe(0); + await vi.advanceTimersByTimeAsync(0); + expect(stopped).toBe(false); + finish.resolve(); + await done; + expect(seen).toEqual([1]); + expect(scans.remaining).toBe(0); + await scans.stop(); + }); + + it("stops before the first scan without leaving a timer", async () => { + const process = vi.fn(); + const scans = queue({ process }); + scans.enqueue("pending"); + await scans.stop(); + await vi.advanceTimersByTimeAsync(100); + expect(process).not.toHaveBeenCalled(); + expect(vi.getTimerCount()).toBe(0); + }); + + it("observes an unhandled item failure after other started work settles", async () => { + const fail = gate(); + const finish = gate(); + const failure = new Error("read failed"); + const seen: number[] = []; + const scans = queue({ + process: async (item) => { + seen.push(item); + if (item === 0) { + await fail.promise; + throw failure; + } + await finish.promise; + }, + }); + for (let item = 0; item < 12; item++) scans.enqueue(item); + await vi.advanceTimersByTimeAsync(0); + let error: unknown; + const done = scans.waitForIdle().catch((reason: unknown) => { + error = reason; + }); + fail.resolve(); + await vi.advanceTimersByTimeAsync(0); + expect(error).toBeUndefined(); + expect(seen).toHaveLength(10); + expect(() => scans.enqueue(12)).toThrow("closed"); + finish.resolve(); + await done; + expect(error).toBe(failure); + expect(scans.remaining).toBe(2); + }); + + it("includes work enqueued by an active scan in an existing idle wait", async () => { + const seen: string[] = []; + const scans = queue({ + process: (item) => { + seen.push(item); + if (item === "first") scans.enqueue("follow-up"); + }, + }); + scans.enqueue("first"); + let idle = false; + const done = scans.waitForIdle().then(() => { + idle = true; + }); + await vi.advanceTimersByTimeAsync(100); + expect(seen).toEqual(["first"]); + expect(idle).toBe(false); + await vi.advanceTimersByTimeAsync(100); + await done; + expect(seen).toEqual(["first", "follow-up"]); + expect(vi.getTimerCount()).toBe(0); + }); + + it("wakes an input wait when its only active scan fails", async () => { + const finish = gate(); + const failure = new Error("read failed"); + const scans = queue({ + process: async () => { + await finish.promise; + throw failure; + }, + }); + scans.enqueue("only"); + await vi.advanceTimersByTimeAsync(100); + const done = expect(scans.waitForIdle()).rejects.toBe(failure); + finish.resolve(); + await done; + expect(scans.remaining).toBe(0); + }); + + it("rejects an existing idle wait if observing completion fails", async () => { + const finish = gate(); + const failure = new Error("observer failed"); + const scans = queue({ + process: () => finish.promise, + onProgress: (remaining) => { + if (remaining === 0) throw failure; + }, + }); + scans.enqueue("active"); + await vi.advanceTimersByTimeAsync(100); + const done = expect(scans.waitForIdle()).rejects.toBe(failure); + finish.resolve(); + await done; + expect(scans.remaining).toBe(0); + }); + + it("retains both an observer failure and a later failure from active work", async () => { + const finish = gate(); + const observerFailure = new Error("observer failed"); + const readFailure = new Error("read failed"); + const scans = queue({ + process: async () => { + await finish.promise; + throw readFailure; + }, + onProgress: (remaining) => { + if (remaining === 2) throw observerFailure; + }, + }); + scans.enqueue("active"); + await vi.advanceTimersByTimeAsync(100); + scans.enqueue("pending"); + const done = expect(scans.waitForIdle()).rejects.toMatchObject({ + message: "Customisation scans failed.", + errors: [observerFailure, readFailure], + }); + finish.resolve(); + await done; + }); + + it("wakes an idle waiter when the progress observer fails", async () => { + const failure = new Error("observer failed"); + const process = vi.fn(); + const scans = queue({ + process, + onProgress: () => { + throw failure; + }, + }); + scans.enqueue("pending"); + await expect(scans.waitForIdle()).rejects.toBe(failure); + expect(process).not.toHaveBeenCalled(); + expect(vi.getTimerCount()).toBe(0); + }); +}); diff --git a/src/features/LiveSyncCommands.ts b/src/features/LiveSyncCommands.ts index 3f1ec783..d3467549 100644 --- a/src/features/LiveSyncCommands.ts +++ b/src/features/LiveSyncCommands.ts @@ -43,7 +43,7 @@ export abstract class LiveSyncCommands { this._log = createInstanceLogFunction(this.constructor.name, this.services.API); // __$checkInstanceBinding(this); } - abstract onunload(): void; + abstract onunload(): void | Promise; abstract onload(): void | Promise; _isMainReady() { diff --git a/updates.md b/updates.md index e7aa80f8..2b43202c 100644 --- a/updates.md +++ b/updates.md @@ -14,6 +14,9 @@ Earlier releases remain available in the 1.0 release history, the 1.0 preview hi ### Fixes +- Reloading the Customisation Sync catalogue no longer allows an earlier legacy-format scan to put outdated items back into the list. + - Reloading or disabling Customisation Sync discards unstarted scans and ignores stale read results. Migrations to per-file storage which have already started are allowed to finish. + - Buttons in confirmation and setup dialogues now align with the content above them. Confirmation dialogue buttons also have space for keyboard focus outlines. - Mobile confirmation dialogues now retain Obsidian's standard content spacing. Text and password prompts stay above the software keyboard.