From cfdc6987751041ae9464fd090eff86fd622d0e67 Mon Sep 17 00:00:00 2001 From: vorotamoroz Date: Thu, 29 Oct 2026 14:56:56 +0000 Subject: [PATCH] Reuse scoped content and progress operations --- src/common/pathCompatibility.ts | 10 +- src/common/utils.ts | 9 - .../CmdConfigSync.applyContent.unit.spec.ts | 275 +++++++++++++++++ ...ConfigSync.scanAllConfigFiles.unit.spec.ts | 279 ++++++++++++++++++ src/features/ConfigSync/CmdConfigSync.ts | 109 ++++--- ...iddenFileSync.attemptProgress.unit.spec.ts | 143 +++++++++ .../CmdHiddenFileSync.progress.unit.spec.ts | 219 ++++++++++++++ .../HiddenFileSync/CmdHiddenFileSync.ts | 67 +++-- .../CmdHiddenFileSync.unit.spec.ts | 79 +++++ .../performBatchWithProgress.ts | 49 +++ .../performBatchWithProgress.unit.spec.ts | 93 ++++++ .../StatusBarDisplay.timeout.unit.spec.ts | 142 +++++++++ src/modules/features/StatusBarDisplay.ts | 81 +++-- .../features/StatusBarDisplay.unit.spec.ts | 119 ++++++++ ...icateResultProcessor.counting.unit.spec.ts | 46 +++ .../replication/ReplicateResultProcessor.ts | 8 +- 16 files changed, 1596 insertions(+), 132 deletions(-) create mode 100644 src/features/ConfigSync/CmdConfigSync.applyContent.unit.spec.ts create mode 100644 src/features/ConfigSync/CmdConfigSync.scanAllConfigFiles.unit.spec.ts create mode 100644 src/features/HiddenFileSync/CmdHiddenFileSync.attemptProgress.unit.spec.ts create mode 100644 src/features/HiddenFileSync/CmdHiddenFileSync.progress.unit.spec.ts create mode 100644 src/features/HiddenFileSync/performBatchWithProgress.ts create mode 100644 src/features/HiddenFileSync/performBatchWithProgress.unit.spec.ts create mode 100644 src/modules/features/StatusBarDisplay.timeout.unit.spec.ts create mode 100644 src/serviceFeatures/replication/ReplicateResultProcessor.counting.unit.spec.ts diff --git a/src/common/pathCompatibility.ts b/src/common/pathCompatibility.ts index 69ad7254..b9f3b1ff 100644 --- a/src/common/pathCompatibility.ts +++ b/src/common/pathCompatibility.ts @@ -1,3 +1,5 @@ +import { findPathComponentsExceedingUtf8Limit as findKitPathComponentsExceedingUtf8Limit } from "octagonal-wheels/path"; + export const ANDROID_LINUX_PATH_COMPONENT_UTF8_WARNING_BOUNDARY = 255; export interface OversizedPathComponent { @@ -5,8 +7,6 @@ export interface OversizedPathComponent { utf8Bytes: number; } -const utf8Encoder = new TextEncoder(); - /** * Return path components which exceed the conservative Android/Linux * compatibility boundary. @@ -18,9 +18,5 @@ export function findPathComponentsExceedingUtf8Limit( path: string, maxBytes: number = ANDROID_LINUX_PATH_COMPONENT_UTF8_WARNING_BOUNDARY ): OversizedPathComponent[] { - return path - .split("/") - .filter((component) => component.length > 0) - .map((component) => ({ component, utf8Bytes: utf8Encoder.encode(component).byteLength })) - .filter(({ utf8Bytes }) => utf8Bytes > maxBytes); + return findKitPathComponentsExceedingUtf8Limit(path, maxBytes); } diff --git a/src/common/utils.ts b/src/common/utils.ts index f4bc621d..fc248ff1 100644 --- a/src/common/utils.ts +++ b/src/common/utils.ts @@ -389,13 +389,4 @@ export async function autosaveCache(db: KeyValueDatabase, mapKey: string): }; } -export function onlyInNTimes(n: number, proc: (progress: number) => unknown) { - let counter = 0; - return function () { - if (counter++ % n == 0) { - proc(counter); - } - }; -} - export { displayRev } from "@vrtmrz/livesync-commonlib/compat/common/utils"; diff --git a/src/features/ConfigSync/CmdConfigSync.applyContent.unit.spec.ts b/src/features/ConfigSync/CmdConfigSync.applyContent.unit.spec.ts new file mode 100644 index 00000000..95e7cab2 --- /dev/null +++ b/src/features/ConfigSync/CmdConfigSync.applyContent.unit.spec.ts @@ -0,0 +1,275 @@ +import { describe, expect, it, vi } from "vitest"; + +vi.mock("@/deps.ts", () => ({ + addIcon: vi.fn(), + diff_match_patch: class DiffMatchPatch {}, + normalizePath: vi.fn((path: string) => path), + parseYaml: vi.fn(), + Platform: {}, +})); +vi.mock("./PluginDialogModal.ts", () => ({ + PluginDialogModal: class PluginDialogModal {}, +})); +vi.mock("@/features/HiddenFileCommon/JsonResolveModal.ts", () => ({ + JsonResolveModal: class JsonResolveModal {}, +})); +vi.mock("@/modules/features/InteractiveConflictResolving/ConflictResolveModal.ts", () => ({ + ConflictResolveModal: class ConflictResolveModal {}, +})); +vi.mock("@/features/LiveSyncCommands.ts", () => ({ + LiveSyncCommands: class LiveSyncCommands { + core!: { services: unknown }; + get services() { + return this.core.services; + } + }, +})); +vi.mock("@/common/types.ts", () => ({ + ICXHeader: "ix:", + PERIODIC_PLUGIN_SWEEP: 60, +})); +vi.mock("@/common/utils.ts", () => ({ + cancelTask: vi.fn(), + EVEN: Symbol("even"), + isCustomisationSyncMetadata: vi.fn(), + isPluginMetadata: vi.fn(), + scheduleTask: vi.fn(), +})); +vi.mock("@/common/PeriodicProcessor.ts", () => ({ + PeriodicProcessor: class PeriodicProcessor {}, +})); +vi.mock("@/common/events.ts", () => ({ + EVENT_REQUEST_OPEN_PLUGIN_SYNC_DIALOG: "open-plugin-sync", + eventHub: { + onEvent: vi.fn(), + }, +})); +vi.mock("@/common/translation", () => ({ + $msg: vi.fn((message: string) => message), +})); +vi.mock("@/common/obsidianCommunityPlugins.ts", () => ({ + getObsidianCommunityPluginManager: vi.fn(), +})); + +import { LOG_LEVEL_NOTICE, LOG_LEVEL_VERBOSE } from "@vrtmrz/livesync-commonlib/compat/common/types"; +import { ConfigSync, type PluginDataExDisplayV2 } from "./CmdConfigSync"; + +const BASE_DIR = ".obsidian"; +const DEVICE_NAME = "test-device"; +const STAT = { mtime: 123456, ctime: 123000 }; + +type ApplyFile = { + filename: string; + datatype: "newnote" | "plain"; + data: string[]; + mtime: number; + ctime: number; +}; + +function makeApplyFile(datatype: ApplyFile["datatype"], data: string[]): ApplyFile { + return { + filename: "plugins/example/data.json", + datatype, + data, + mtime: STAT.mtime, + ctime: STAT.ctime, + }; +} + +function setup(file: ApplyFile, currentText = "before", currentBinary = new ArrayBuffer(0)) { + const path = `${BASE_DIR}/${file.filename}`; + const events: string[] = []; + const storageAccess = { + ensureDir: vi.fn(async (_path: string) => { + events.push("ensure"); + }), + readHiddenFileText: vi.fn(async (_path: string) => { + events.push("read-text"); + return currentText; + }), + readHiddenFileBinary: vi.fn(async (_path: string) => { + events.push("read-binary"); + return currentBinary; + }), + writeHiddenFileAuto: vi.fn( + async (_path: string, _content: string | ArrayBuffer, _stat?: { mtime: number; ctime: number }) => { + events.push("write"); + } + ), + }; + const storeCustomisationFileV2 = vi.fn(async (_path: string, _term: string) => { + events.push("store"); + return true; + }); + const log = vi.fn(); + const configSync = Object.create(ConfigSync.prototype) as ConfigSync; + Object.assign(configSync, { + core: { + services: { + API: { getSystemConfigDir: () => BASE_DIR }, + setting: { getDeviceAndVaultName: () => DEVICE_NAME }, + }, + storageAccess, + }, + _log: log, + storeCustomisationFileV2, + }); + const data = { + displayName: "Example plug-in", + name: "example", + files: [file], + } as unknown as PluginDataExDisplayV2; + return { configSync, data, events, log, path, storageAccess, storeCustomisationFileV2 }; +} + +const changedInputs = [ + { + kind: "text", + file: makeApplyFile("plain", ["replacement text"]), + currentText: "previous text", + expectedText: "replacement text", + }, + { + kind: "binary", + file: makeApplyFile("newnote", ["AQID"]), + currentBinary: new Uint8Array([9]).buffer, + expectedBytes: [1, 2, 3], + }, +] as const; + +const unchangedInputs = [ + { + kind: "text", + file: makeApplyFile("plain", ["unchanged text"]), + currentText: "unchanged text", + }, + { + kind: "binary", + file: makeApplyFile("newnote", ["AQID"]), + currentBinary: new Uint8Array([1, 2, 3]).buffer, + }, +] as const; + +describe("ConfigSync V2 content application", () => { + it.each(changedInputs)("writes changed $kind content, then stores it", async (input) => { + const fixture = setup( + input.file, + input.kind === "text" ? input.currentText : undefined, + input.kind === "binary" ? input.currentBinary : undefined + ); + + await expect(fixture.configSync.applyDataV2(fixture.data)).resolves.toBe(true); + + expect(fixture.storageAccess.ensureDir).toHaveBeenCalledWith(fixture.path); + expect(fixture.storageAccess.writeHiddenFileAuto).toHaveBeenCalledOnce(); + expect(fixture.storageAccess.writeHiddenFileAuto).toHaveBeenCalledWith( + fixture.path, + input.kind === "text" ? input.expectedText : expect.any(ArrayBuffer), + STAT + ); + if (input.kind === "binary") { + const written = fixture.storageAccess.writeHiddenFileAuto.mock.calls[0]?.[1]; + expect(written).toBeInstanceOf(ArrayBuffer); + expect(Array.from(new Uint8Array(written as ArrayBuffer))).toEqual(input.expectedBytes); + } + expect(fixture.events).toEqual([ + "ensure", + input.kind === "text" ? "read-text" : "read-binary", + "write", + "store", + ]); + expect(fixture.storeCustomisationFileV2).toHaveBeenCalledWith(fixture.path, DEVICE_NAME); + }); + + it("writes the file timestamps captured before reading existing content", async () => { + const file = makeApplyFile("plain", ["replacement text"]); + const fixture = setup(file, "previous text"); + fixture.storageAccess.readHiddenFileText.mockImplementation(async () => { + fixture.events.push("read-text"); + file.mtime = STAT.mtime + 1; + file.ctime = STAT.ctime + 1; + return "previous text"; + }); + + await expect(fixture.configSync.applyDataV2(fixture.data)).resolves.toBe(true); + + expect(fixture.storageAccess.writeHiddenFileAuto).toHaveBeenCalledWith(fixture.path, "replacement text", STAT); + }); + + it.each(unchangedInputs)("skips writing and storing unchanged $kind content", async (input) => { + const fixture = setup( + input.file, + input.kind === "text" ? input.currentText : undefined, + input.kind === "binary" ? input.currentBinary : undefined + ); + + await expect(fixture.configSync.applyDataV2(fixture.data)).resolves.toBe(true); + + expect(fixture.storageAccess.writeHiddenFileAuto).not.toHaveBeenCalled(); + expect(fixture.storeCustomisationFileV2).not.toHaveBeenCalled(); + expect(fixture.events).toEqual(["ensure", input.kind === "text" ? "read-text" : "read-binary"]); + expect(fixture.log).toHaveBeenCalledWith( + `The file ${input.file.filename} is already up-to-date`, + LOG_LEVEL_VERBOSE + ); + }); + + it.each([ + { kind: "text", file: makeApplyFile("plain", ["replacement text"]) }, + { kind: "binary", file: makeApplyFile("newnote", ["AQID"]) }, + ])("uses empty $kind content when reading the existing file fails", async ({ kind, file }) => { + const fixture = setup(file); + const readError = new Error("Could not read existing content"); + if (kind === "text") { + fixture.storageAccess.readHiddenFileText.mockRejectedValueOnce(readError); + } else { + fixture.storageAccess.readHiddenFileBinary.mockRejectedValueOnce(readError); + } + + await expect(fixture.configSync.applyDataV2(fixture.data)).resolves.toBe(true); + + expect(fixture.storageAccess.writeHiddenFileAuto).toHaveBeenCalledOnce(); + expect(fixture.storeCustomisationFileV2).toHaveBeenCalledWith(fixture.path, DEVICE_NAME); + expect(fixture.log).toHaveBeenCalledWith(`Could not read the file ${file.filename}`, LOG_LEVEL_VERBOSE); + expect(fixture.log).toHaveBeenCalledWith(readError, LOG_LEVEL_VERBOSE); + }); + + it("awaits the write before storing the applied content", async () => { + const file = makeApplyFile("plain", ["replacement text"]); + const fixture = setup(file, "previous text"); + let finishWrite!: () => void; + let signalWriteStarted!: () => void; + const writeGate = new Promise((resolve) => { + finishWrite = resolve; + }); + const writeStarted = new Promise((resolve) => { + signalWriteStarted = resolve; + }); + fixture.storageAccess.writeHiddenFileAuto.mockImplementation(async () => { + fixture.events.push("write-start"); + signalWriteStarted(); + await writeGate; + fixture.events.push("write-finished"); + }); + + const applying = fixture.configSync.applyDataV2(fixture.data); + await writeStarted; + expect(fixture.storeCustomisationFileV2).not.toHaveBeenCalled(); + + finishWrite(); + await expect(applying).resolves.toBe(true); + expect(fixture.events.slice(-3)).toEqual(["write-start", "write-finished", "store"]); + }); + + it("returns false after a write failure and does not store the file", async () => { + const fixture = setup(makeApplyFile("plain", ["replacement text"]), "previous text"); + const writeError = new Error("Write failed"); + fixture.storageAccess.writeHiddenFileAuto.mockRejectedValueOnce(writeError); + + await expect(fixture.configSync.applyDataV2(fixture.data)).resolves.toBe(false); + + expect(fixture.storeCustomisationFileV2).not.toHaveBeenCalled(); + expect(fixture.log).toHaveBeenCalledWith("Applying Example plug-in.. Failed", LOG_LEVEL_NOTICE); + expect(fixture.log).toHaveBeenCalledWith(writeError, LOG_LEVEL_VERBOSE); + }); +}); diff --git a/src/features/ConfigSync/CmdConfigSync.scanAllConfigFiles.unit.spec.ts b/src/features/ConfigSync/CmdConfigSync.scanAllConfigFiles.unit.spec.ts new file mode 100644 index 00000000..069fdab6 --- /dev/null +++ b/src/features/ConfigSync/CmdConfigSync.scanAllConfigFiles.unit.spec.ts @@ -0,0 +1,279 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +vi.mock("@/deps.ts", () => ({ + addIcon: vi.fn(), + diff_match_patch: class DiffMatchPatch {}, + normalizePath: vi.fn((path: string) => path), + parseYaml: vi.fn(), + Platform: {}, +})); +vi.mock("./PluginDialogModal.ts", () => ({ + PluginDialogModal: class PluginDialogModal {}, +})); +vi.mock("@/features/HiddenFileCommon/JsonResolveModal.ts", () => ({ + JsonResolveModal: class JsonResolveModal {}, +})); +vi.mock("@/modules/features/InteractiveConflictResolving/ConflictResolveModal.ts", () => ({ + ConflictResolveModal: class ConflictResolveModal {}, +})); +vi.mock("@/features/LiveSyncCommands.ts", () => ({ + LiveSyncCommands: class LiveSyncCommands { + core!: { services: unknown; localDatabase: unknown }; + get localDatabase() { + return this.core.localDatabase; + } + get services() { + return this.core.services; + } + }, +})); +vi.mock("@/common/types.ts", () => ({ + ICXHeader: "ix:", + PERIODIC_PLUGIN_SWEEP: 60, +})); +vi.mock("@/common/utils.ts", () => ({ + cancelTask: vi.fn(), + EVEN: Symbol("even"), + isCustomisationSyncMetadata: vi.fn(), + isPluginMetadata: vi.fn(), + scheduleTask: vi.fn(), +})); +vi.mock("@/common/PeriodicProcessor.ts", () => ({ + PeriodicProcessor: class PeriodicProcessor {}, +})); +vi.mock("@/common/events.ts", () => ({ + EVENT_REQUEST_OPEN_PLUGIN_SYNC_DIALOG: "open-plugin-sync", + eventHub: { + onEvent: vi.fn(), + }, +})); +vi.mock("@/common/translation", () => ({ + $msg: vi.fn((message: string) => message), +})); +vi.mock("@/common/obsidianCommunityPlugins.ts", () => ({ + getObsidianCommunityPluginManager: vi.fn(), +})); + +vi.mock("octagonal-wheels/concurrency/scopedIteration", { spy: true }); + +import { LOG_LEVEL_VERBOSE, type FilePath } from "@vrtmrz/livesync-commonlib/compat/common/types"; +import * as iterationModule from "octagonal-wheels/concurrency/scopedIteration"; +import { promiseWithResolvers } from "octagonal-wheels/promises"; +import { ConfigSync } from "./CmdConfigSync"; + +const term = "device"; +const prefix = `ix:${term}/`; +const keyOf = (path: string) => prefix + path; +type Row = { _id: string; path: string }; +const row = (name: string, path = name): Row => ({ _id: keyOf(name), path }); + +function setup(files: string[] = [], rows: Row[] = []) { + const events: string[] = []; + const findEntries = vi.fn(async function* () { + yield* rows; + }); + const scanInternalFiles = vi.fn(async () => files as FilePath[]); + const storeCustomisationFileV2 = vi.fn(async (path: string, _term: string) => { + events.push(`store:${path}`); + }); + const deleteConfigOnDatabase = vi.fn(async (key: string) => { + events.push(`delete:${key}`); + }); + const updatePluginList = vi.fn(async (_showMessage: boolean) => {}); + const log = vi.fn(); + const configSync = Object.create(ConfigSync.prototype) as ConfigSync; + Object.assign(configSync, { + core: { + settings: { usePluginSyncV2: true }, + localDatabase: { findEntries }, + services: { setting: { getDeviceAndVaultName: () => term } }, + }, + scanInternalFiles, + isTargetPath: (path: string) => path !== "ignored", + filenameWithUnifiedKey: (path: string, _term: string) => keyOf(path), + unifiedKeyPrefixOfTerminal: (_term: string) => prefix, + storeCustomisationFileV2, + deleteConfigOnDatabase, + updatePluginList, + _log: log, + }); + return { + configSync, + events, + findEntries, + scanInternalFiles, + storeCustomisationFileV2, + deleteConfigOnDatabase, + updatePluginList, + log, + }; +} + +beforeEach(() => { + vi.clearAllMocks(); +}); + +describe("ConfigSync V2 scan behaviour", () => { + it("finishes existing database entries before storing extra files and retains target filtering", async () => { + const fixture = setup( + ["matched", "extra", "ignored"], + [row("matched"), row("stale"), row("encoded", "encoded%path")] + ); + await fixture.configSync.scanAllConfigFiles(false); + + expect(fixture.events).toEqual(["store:matched", `delete:${keyOf("stale")}`, "store:extra"]); + expect(fixture.storeCustomisationFileV2.mock.calls).toEqual([ + ["matched", term], + ["extra", term], + ]); + expect(fixture.deleteConfigOnDatabase).toHaveBeenCalledExactlyOnceWith(keyOf("stale")); + expect(fixture.findEntries).toHaveBeenCalledExactlyOnceWith(prefix, `${prefix}\u{10ffff}`, { + include_docs: true, + }); + expect(fixture.updatePluginList).toHaveBeenCalledExactlyOnceWith(false); + }); + + it("finishes database enumeration before dispatching database work", async () => { + const fixture = setup(); + const enumerationPaused = promiseWithResolvers(); + const continueEnumeration = promiseWithResolvers(); + fixture.findEntries.mockImplementation(async function* () { + yield row("stale"); + enumerationPaused.resolve(); + await continueEnumeration.promise; + }); + + const scan = fixture.configSync.scanAllConfigFiles(false); + try { + await enumerationPaused.promise; + expect(fixture.deleteConfigOnDatabase).not.toHaveBeenCalled(); + expect(fixture.storeCustomisationFileV2).not.toHaveBeenCalled(); + } finally { + continueEnumeration.resolve(); + await scan; + } + + expect(fixture.deleteConfigOnDatabase).toHaveBeenCalledExactlyOnceWith(keyOf("stale")); + }); + + it("limits actual database operations to ten and completes that batch before the extra batch", async () => { + const fixture = setup( + ["extra"], + Array.from({ length: 11 }, (_, index) => row(String(index))) + ); + const gates = Array.from({ length: 11 }, () => promiseWithResolvers()); + const tenStarted = promiseWithResolvers(); + const lastStarted = promiseWithResolvers(); + let active = 0; + let maximum = 0; + let started = 0; + fixture.deleteConfigOnDatabase.mockImplementation(async (key) => { + const index = Number(key.slice(prefix.length)); + active++; + maximum = Math.max(maximum, active); + started++; + if (started === 10) tenStarted.resolve(); + if (index === 10) lastStarted.resolve(); + try { + await gates[index].promise; + } finally { + active--; + } + }); + const scan = fixture.configSync.scanAllConfigFiles(false); + try { + await tenStarted.promise; + expect(active).toBe(10); + expect(started).toBe(10); + expect(fixture.storeCustomisationFileV2).not.toHaveBeenCalled(); + gates[0].resolve(); + await lastStarted.promise; + expect(active).toBe(10); + expect(fixture.storeCustomisationFileV2).not.toHaveBeenCalled(); + } finally { + gates.forEach((gate) => gate.resolve()); + await scan; + } + expect(maximum).toBe(10); + expect(active).toBe(0); + expect(fixture.deleteConfigOnDatabase).toHaveBeenCalledTimes(11); + expect(fixture.storeCustomisationFileV2).toHaveBeenCalledExactlyOnceWith("extra", term); + }); + + it("logs a database processing error and continues through the extra batch", async () => { + const fixture = setup(["extra"], [row("bad"), row("other")]); + const error = new Error("delete"); + fixture.deleteConfigOnDatabase.mockRejectedValueOnce(error); + await fixture.configSync.scanAllConfigFiles(false); + + expect(fixture.deleteConfigOnDatabase).toHaveBeenCalledTimes(2); + expect(fixture.log).toHaveBeenCalledWith(`scanAllConfigFiles - Error: ${keyOf("bad")}`, LOG_LEVEL_VERBOSE); + expect(fixture.log).toHaveBeenCalledWith(error, LOG_LEVEL_VERBOSE); + expect(fixture.storeCustomisationFileV2).toHaveBeenCalledExactlyOnceWith("extra", term); + expect(fixture.updatePluginList).toHaveBeenCalledOnce(); + }); + + it("retains a matched local file for the extra batch when its first store fails", async () => { + const fixture = setup(["matched"], [row("matched")]); + const error = new Error("first store"); + fixture.storeCustomisationFileV2.mockRejectedValueOnce(error); + await fixture.configSync.scanAllConfigFiles(false); + + expect(fixture.storeCustomisationFileV2.mock.calls).toEqual([ + ["matched", term], + ["matched", term], + ]); + expect(fixture.log).toHaveBeenCalledWith(`scanAllConfigFiles - Error: ${keyOf("matched")}`, LOG_LEVEL_VERBOSE); + expect(fixture.log).toHaveBeenCalledWith(error, LOG_LEVEL_VERBOSE); + expect(fixture.events).toEqual(["store:matched"]); + expect(fixture.updatePluginList).toHaveBeenCalledOnce(); + }); + + it("logs an extra-file processing error and continues with the remaining files", async () => { + const fixture = setup(["bad", "other"]); + const error = new Error("store"); + fixture.storeCustomisationFileV2.mockRejectedValueOnce(error); + await fixture.configSync.scanAllConfigFiles(false); + + expect(fixture.storeCustomisationFileV2.mock.calls).toEqual([ + ["bad", term], + ["other", term], + ]); + expect(fixture.log).toHaveBeenCalledWith("scanAllConfigFiles - Error: bad", LOG_LEVEL_VERBOSE); + expect(fixture.log).toHaveBeenCalledWith(error, LOG_LEVEL_VERBOSE); + expect(fixture.updatePluginList).toHaveBeenCalledOnce(); + }); +}); + +describe("ConfigSync scoped iteration composition", () => { + it("uses bounded child scopes for both batches with the same shared values", async () => { + const fixture = setup(["matched", "extra"], [row("matched"), row("stale")]); + await fixture.configSync.scanAllConfigFiles(false); + + expect(iterationModule.forEachInScope).toHaveBeenCalledTimes(2); + const [database, extra] = vi.mocked(iterationModule.forEachInScope).mock.calls; + expect(database[0]).toBe(extra[0]); + expect(database[0].value).toEqual({ term, localFileMap: new Map([[keyOf("extra"), "extra"]]) }); + expect(database[2]).toEqual({ concurrency: 10 }); + expect(extra[2]).toEqual({ concurrency: 10 }); + expect(fixture.events).toEqual(["store:matched", `delete:${keyOf("stale")}`, "store:extra"]); + }); + + it.each(["database", "extra"])("propagates a %s runner failure outside per-item recovery", async (batch) => { + const fixture = setup(["extra"], [row("stale")]); + const error = new Error("iteration"); + const actual = await vi.importActual( + "octagonal-wheels/concurrency/scopedIteration" + ); + if (batch === "extra") { + vi.mocked(iterationModule.forEachInScope).mockImplementationOnce(actual.forEachInScope); + } + vi.mocked(iterationModule.forEachInScope).mockRejectedValueOnce(error); + await expect(fixture.configSync.scanAllConfigFiles(false)).rejects.toBe(error); + + expect(fixture.deleteConfigOnDatabase).toHaveBeenCalledTimes(batch === "extra" ? 1 : 0); + expect(fixture.storeCustomisationFileV2).not.toHaveBeenCalled(); + expect(fixture.log).not.toHaveBeenCalledWith(error, LOG_LEVEL_VERBOSE); + expect(fixture.updatePluginList).not.toHaveBeenCalled(); + }); +}); diff --git a/src/features/ConfigSync/CmdConfigSync.ts b/src/features/ConfigSync/CmdConfigSync.ts index 68ba9703..1902c590 100644 --- a/src/features/ConfigSync/CmdConfigSync.ts +++ b/src/features/ConfigSync/CmdConfigSync.ts @@ -59,7 +59,8 @@ 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"; -import { Semaphore } from "octagonal-wheels/concurrency/semaphore"; +import { forEachInScope } from "octagonal-wheels/concurrency/scopedIteration"; +import { withScope } from "octagonal-wheels/resource/scope"; import { EVENT_REQUEST_OPEN_PLUGIN_SYNC_DIALOG, eventHub } from "@/common/events.ts"; import { PluginDialogModal } from "./PluginDialogModal.ts"; import { $msg } from "@/common/translation"; @@ -68,6 +69,7 @@ import type { LiveSyncCore } from "@/main.ts"; import { LiveSyncError } from "@vrtmrz/livesync-commonlib/compat/common/LSError"; import type { OptionalSyncFeatureMode } from "@/features/optionalSyncFeatures.ts"; import { getObsidianCommunityPluginManager } from "@/common/obsidianCommunityPlugins.ts"; +import { createContentApplier } from "octagonal-wheels/content"; import type { HandlerOutcome } from "@vrtmrz/livesync-commonlib/compat/services/lib/HandlerUtils"; import type { OptionalFileEventOptions } from "@vrtmrz/livesync-commonlib/compat/services/base/IService"; @@ -285,6 +287,10 @@ export type PluginDataExDisplay = { mtime: number; }; type LoadedEntryPluginDataExFile = LoadedEntry & PluginDataExFile; +type ConfigFileApplicationTarget = { + readonly path: FilePath; + readonly stat: { readonly mtime: number; readonly ctime: number }; +}; function categoryToFolder(category: string, configDir: string = ""): string { switch (category) { @@ -1027,44 +1033,63 @@ export class ConfigSync extends LiveSyncCommands { await this.storeCustomisationFileV2(path, this.services.setting.getDeviceAndVaultName()); } else { const files = data.files; + const applyBinaryContent = createContentApplier< + ConfigFileApplicationTarget, + LoadedEntryPluginDataExFile, + ArrayBuffer, + ArrayBuffer + >({ + read: (target) => this.core.storageAccess.readHiddenFileBinary(target.path), + onReadError: (ex, _path, file) => { + this._log(`Could not read the file ${file.filename}`, LOG_LEVEL_VERBOSE); + this._log(ex, LOG_LEVEL_VERBOSE); + return new ArrayBuffer(0); + }, + decode: (file) => base64ToArrayBuffer(file.data), + isSame: (oldData, content) => isDocContentSame(oldData, content), + write: (target, content) => + this.core.storageAccess.writeHiddenFileAuto(target.path, content, target.stat), + }); + const applyTextContent = createContentApplier< + ConfigFileApplicationTarget, + LoadedEntryPluginDataExFile, + string, + string + >({ + read: (target) => this.core.storageAccess.readHiddenFileText(target.path), + onReadError: (ex, _path, file) => { + this._log(`Could not read the file ${file.filename}`, LOG_LEVEL_VERBOSE); + this._log(ex, LOG_LEVEL_VERBOSE); + return ""; + }, + decode: (file) => getDocData(file.data), + isSame: (oldData, content) => isDocContentSame(oldData, content), + write: (target, content) => + this.core.storageAccess.writeHiddenFileAuto(target.path, content, target.stat), + }); for (const f of files) { // If files have applied, modified time will be updated to the current time. - const stat = { mtime: f.mtime, ctime: f.ctime }; const path = `${baseDir}/${f.filename}` as FilePath; + const target: ConfigFileApplicationTarget = { + path, + stat: { mtime: f.mtime, ctime: f.ctime }, + }; this._log(`Applying ${f.filename} of ${data.displayName || data.name}..`); // const contentEach = createBlob(f.data); await this.core.storageAccess.ensureDir(path); if (f.datatype == "newnote") { - let oldData; - try { - oldData = await this.core.storageAccess.readHiddenFileBinary(path); - } catch (ex) { - this._log(`Could not read the file ${f.filename}`, LOG_LEVEL_VERBOSE); - this._log(ex, LOG_LEVEL_VERBOSE); - oldData = new ArrayBuffer(0); - } - const content = base64ToArrayBuffer(f.data); - if (await isDocContentSame(oldData, content)) { + const result = await applyBinaryContent(target, f); + if (result == "unchanged") { this._log(`The file ${f.filename} is already up-to-date`, LOG_LEVEL_VERBOSE); continue; } - await this.core.storageAccess.writeHiddenFileAuto(path, content, stat); } else { - let oldData; - try { - oldData = await this.core.storageAccess.readHiddenFileText(path); - } catch (ex) { - this._log(`Could not read the file ${f.filename}`, LOG_LEVEL_VERBOSE); - this._log(ex, LOG_LEVEL_VERBOSE); - oldData = ""; - } - const content = getDocData(f.data); - if (await isDocContentSame(oldData, content)) { + const result = await applyTextContent(target, f); + if (result == "unchanged") { this._log(`The file ${f.filename} is already up-to-date`, LOG_LEVEL_VERBOSE); continue; } - await this.core.storageAccess.writeHiddenFileAuto(path, content, stat); } this._log(`Applied ${f.filename} of ${data.displayName || data.name}..`); await this.storeCustomisationFileV2(path, this.services.setting.getDeviceAndVaultName()); @@ -1613,49 +1638,35 @@ export class ConfigSync extends LiveSyncCommands { const entries = this.localDatabase.findEntries(prefix + "", `${prefix}\u{10ffff}`, { include_docs: true, }); - const tasks = [] as (() => Promise)[]; - const concurrency = 10; - const semaphore = Semaphore(concurrency); + const databaseEntries: { _id: string; path: FilePathWithPrefix }[] = []; for await (const item of entries) { - if (item.path.indexOf("%") !== -1) { - continue; - } - tasks.push(async () => { - const releaser = await semaphore.acquire(); + if (item.path.indexOf("%") === -1) databaseEntries.push(item); + } + await withScope({ term, localFileMap }, async (scope) => { + await forEachInScope(scope, databaseEntries, { concurrency: 10 }, async (item, ctx) => { try { const unifiedFilenameWithKey = `${item._id}` as FilePathWithPrefix; - const localPath = localFileMap.get(unifiedFilenameWithKey); + const localPath = ctx.value.localFileMap.get(unifiedFilenameWithKey); if (localPath) { - await this.storeCustomisationFileV2(localPath, term); - localFileMap.delete(unifiedFilenameWithKey); + await this.storeCustomisationFileV2(localPath, ctx.value.term); + ctx.value.localFileMap.delete(unifiedFilenameWithKey); } else { await this.deleteConfigOnDatabase(unifiedFilenameWithKey); } } catch (ex) { this._log(`scanAllConfigFiles - Error: ${item._id}`, LOG_LEVEL_VERBOSE); this._log(ex, LOG_LEVEL_VERBOSE); - } finally { - releaser(); } }); - } - await Promise.all(tasks.map((e) => e())); - // Extra files - const taskExtra = [] as (() => Promise)[]; - for (const [, filePath] of localFileMap) { - taskExtra.push(async () => { - const releaser = await semaphore.acquire(); + await forEachInScope(scope, localFileMap.values(), { concurrency: 10 }, async (filePath, ctx) => { try { - await this.storeCustomisationFileV2(filePath, term); + await this.storeCustomisationFileV2(filePath, ctx.value.term); } catch (ex) { this._log(`scanAllConfigFiles - Error: ${filePath}`, LOG_LEVEL_VERBOSE); this._log(ex, LOG_LEVEL_VERBOSE); - } finally { - releaser(); } }); - } - await Promise.all(taskExtra.map((e) => e())); + }); fireAndForget(() => this.updatePluginList(false)); } else { const files = filesAll diff --git a/src/features/HiddenFileSync/CmdHiddenFileSync.attemptProgress.unit.spec.ts b/src/features/HiddenFileSync/CmdHiddenFileSync.attemptProgress.unit.spec.ts new file mode 100644 index 00000000..02b40969 --- /dev/null +++ b/src/features/HiddenFileSync/CmdHiddenFileSync.attemptProgress.unit.spec.ts @@ -0,0 +1,143 @@ +import { describe, expect, it, vi } from "vitest"; +import { + type DocumentID, + type FilePath, + type FilePathWithPrefix, + type MetaEntry, +} from "@vrtmrz/livesync-commonlib/compat/common/types"; + +vi.mock("@/deps.ts", () => ({})); +vi.mock("@/features/HiddenFileCommon/JsonResolveModal.ts", () => ({ + JsonResolveModal: class JsonResolveModal {}, +})); +vi.mock("@/features/LiveSyncCommands.ts", () => ({ + LiveSyncCommands: class LiveSyncCommands { + plugin!: { app: unknown }; + core!: { services: unknown; settings: unknown }; + get app() { + return this.plugin.app; + } + get services() { + return this.core.services; + } + get settings() { + return this.core.settings; + } + }, +})); +vi.mock("./configureHiddenFileSyncMode.ts", () => ({ + configureHiddenFileSyncMode: vi.fn(), +})); + +import { HiddenFileSync } from "./CmdHiddenFileSync.ts"; + +function createProgress(events: string[] = []) { + return { + log: vi.fn((message: string) => { + if (/^(?:\d+\/\d+|Checking \d+\/\d+)$/.test(message)) events.push(message); + }), + once: vi.fn(), + done: vi.fn(), + }; +} + +function createStoragePaths(count: number): FilePath[] { + return Array.from({ length: count }, (_, index) => `.obsidian/plugins/example/${index}.json` as FilePath); +} + +function createDatabaseEntries(count: number): MetaEntry[] { + return Array.from( + { length: count }, + (_, index) => + ({ + _id: `i:.obsidian/plugins/example/${index}.json` as DocumentID, + _rev: "1-test", + path: `i:.obsidian/plugins/example/${index}.json` as FilePathWithPrefix, + ctime: 1, + mtime: 1, + size: 0, + type: "plain", + datatype: "plain", + children: [], + eden: {}, + deleted: false, + }) as MetaEntry + ); +} + +function createHiddenFileSync(count: number, events: string[] = []) { + const progress = createProgress(events); + const statHidden = vi.fn(async () => { + events.push(`stat:${statHidden.mock.calls.length}`); + return { mtime: 1 }; + }); + const hiddenFileSync = Object.create(HiddenFileSync.prototype) as HiddenFileSync; + Object.assign(hiddenFileSync, { + _progress: vi.fn(() => progress), + _log: vi.fn(), + _fileInfoLastProcessed: new Map(), + _databaseInfoLastProcessed: new Map(), + scanInternalFileNames: vi.fn(async () => createStoragePaths(count)), + getAllDatabaseFiles: vi.fn(async () => createDatabaseEntries(count)), + getPath: vi.fn((file: MetaEntry) => file.path), + updateLastProcessed: vi.fn(), + scanAllStorageChanges: vi.fn(async () => undefined), + scanAllDatabaseChanges: vi.fn(async () => undefined), + resetLastProcessedDatabase: vi.fn(), + resetLastProcessedFile: vi.fn(), + useStorageFiles: vi.fn(async () => false), + useDatabaseFiles: vi.fn(async () => false), + trackStorageFileModification: vi.fn(async () => false), + trackDatabaseFileModification: vi.fn(async () => false), + core: { storageAccess: { statHidden } }, + }); + return { hiddenFileSync, progress, statHidden }; +} + +describe("HiddenFileSync attempt progress", () => { + it("reports applyOfflineChanges attempts at 1 and each 25 items, without a final partial count", async () => { + const events: string[] = []; + const { hiddenFileSync, progress } = createHiddenFileSync(52, events); + + await hiddenFileSync.applyOfflineChanges(false); + + expect(progress.log.mock.calls.map(([message]) => message).filter((message) => /^\d+\/\d+$/.test(message))).toEqual([ + "1/52", + "26/52", + "51/52", + ]); + expect(events.indexOf("51/52")).toBeLessThan(events.indexOf("stat:1")); + }); + + it("reports rebuildMerging attempts at 1 and each 100 items, without a final partial count", async () => { + const events: string[] = []; + const { hiddenFileSync, progress } = createHiddenFileSync(202, events); + + await hiddenFileSync.rebuildMerging(false); + + expect(progress.log.mock.calls.map(([message]) => message).filter((message) => message.startsWith("Checking "))).toEqual([ + "Checking 1/202", + "Checking 101/202", + "Checking 201/202", + ]); + expect(events.indexOf("Checking 101/202")).toBeLessThan(events.indexOf("stat:101")); + }); + + it("preserves stat failures from applyOfflineChanges", async () => { + const { hiddenFileSync, progress, statHidden } = createHiddenFileSync(1); + const error = new Error("stat failed"); + statHidden.mockRejectedValueOnce(error); + + await expect(hiddenFileSync.applyOfflineChanges(false)).rejects.toBe(error); + expect(progress.done).not.toHaveBeenCalled(); + }); + + it("preserves stat failures from rebuildMerging", async () => { + const { hiddenFileSync, progress, statHidden } = createHiddenFileSync(1); + const error = new Error("stat failed"); + statHidden.mockRejectedValueOnce(error); + + await expect(hiddenFileSync.rebuildMerging(false)).rejects.toBe(error); + expect(progress.done).not.toHaveBeenCalled(); + }); +}); diff --git a/src/features/HiddenFileSync/CmdHiddenFileSync.progress.unit.spec.ts b/src/features/HiddenFileSync/CmdHiddenFileSync.progress.unit.spec.ts new file mode 100644 index 00000000..dd301a80 --- /dev/null +++ b/src/features/HiddenFileSync/CmdHiddenFileSync.progress.unit.spec.ts @@ -0,0 +1,219 @@ +import { describe, expect, it, vi } from "vitest"; +import { + LOG_LEVEL_VERBOSE, + type DocumentID, + type FilePath, + type FilePathWithPrefix, + type MetaEntry, +} from "@vrtmrz/livesync-commonlib/compat/common/types"; + +vi.mock("@/deps.ts", () => ({})); +vi.mock("@/features/HiddenFileCommon/JsonResolveModal.ts", () => ({ + JsonResolveModal: class JsonResolveModal {}, +})); +vi.mock("@/features/LiveSyncCommands.ts", () => ({ + LiveSyncCommands: class LiveSyncCommands { + plugin!: { app: unknown }; + core!: { services: unknown; settings: unknown }; + get app() { + return this.plugin.app; + } + get services() { + return this.core.services; + } + get settings() { + return this.core.settings; + } + }, +})); +vi.mock("./configureHiddenFileSyncMode.ts", () => ({ + configureHiddenFileSyncMode: vi.fn(), +})); + +import { HiddenFileSync } from "./CmdHiddenFileSync.ts"; + +function createProgress() { + return { + log: vi.fn(), + once: vi.fn(), + done: vi.fn(), + }; +} + +function createHiddenFileSync() { + const progress = createProgress(); + const log = vi.fn(); + const trackStorageFileModification = vi.fn(async (_file: FilePath) => false); + const trackDatabaseFileModification = vi.fn(async (_path: FilePath) => false); + const isTargetFile = vi.fn(async (_path: FilePath) => true); + const getPath = vi.fn((file: MetaEntry) => file.path); + const hiddenFileSync = Object.create(HiddenFileSync.prototype) as HiddenFileSync; + + Object.assign(hiddenFileSync, { + _progress: vi.fn(() => progress), + _log: log, + trackStorageFileModification, + trackDatabaseFileModification, + isTargetFile, + getPath, + }); + + return { + hiddenFileSync, + progress, + log, + trackStorageFileModification, + trackDatabaseFileModification, + isTargetFile, + }; +} + +function createStoragePaths(count: number): FilePath[] { + return Array.from({ length: count }, (_, index) => `.obsidian/plugins/example/${index}.json` as FilePath); +} + +function createDatabaseEntries(count: number): MetaEntry[] { + return Array.from( + { length: count }, + (_, index) => + ({ + _id: `i:.obsidian/plugins/example/${index}.json` as DocumentID, + _rev: "1-test", + path: `i:.obsidian/plugins/example/${index}.json` as FilePathWithPrefix, + ctime: 1, + mtime: 1, + size: 0, + type: "plain", + datatype: "plain", + children: [], + eden: {}, + deleted: false, + }) as MetaEntry + ); +} + +describe("HiddenFileSync batch progress consumers", () => { + it("counts completed storage modifications even when they return false", async () => { + const { hiddenFileSync, progress, trackStorageFileModification } = createHiddenFileSync(); + const files = createStoragePaths(101); + + await hiddenFileSync.trackScannedStorageChanges(files); + + expect(trackStorageFileModification).toHaveBeenCalledTimes(101); + expect(progress.log.mock.calls).toEqual([["1/101"], ["101/101"]]); + expect(progress.done).toHaveBeenCalledTimes(1); + }); + + it("logs and skips failed storage items when counting progress", async () => { + const { hiddenFileSync, progress, log, trackStorageFileModification } = createHiddenFileSync(); + const files = createStoragePaths(100); + const failingFile = files[50]; + const error = new Error("Storage item failed"); + trackStorageFileModification.mockImplementation(async (file) => { + if (file === failingFile) throw error; + return false; + }); + + await hiddenFileSync.trackScannedStorageChanges(files); + + expect(progress.once).toHaveBeenCalledWith(`Failed to process storage change file:${failingFile}`); + expect(log).toHaveBeenCalledWith(error, LOG_LEVEL_VERBOSE); + expect(progress.log.mock.calls).toEqual([["1/100"]]); + expect(progress.done).toHaveBeenCalledTimes(1); + }); + + it("counts completed database items, including filtered items and false returns", async () => { + const { hiddenFileSync, progress, trackDatabaseFileModification, isTargetFile } = createHiddenFileSync(); + const files = createDatabaseEntries(101); + const filteredPath = ".obsidian/plugins/example/50.json" as FilePath; + isTargetFile.mockImplementation(async (path) => path !== filteredPath); + + await hiddenFileSync.trackScannedDatabaseChange(files); + + expect(isTargetFile).toHaveBeenCalledTimes(101); + expect(trackDatabaseFileModification).toHaveBeenCalledTimes(100); + expect(progress.log.mock.calls).toEqual([["1/101"], ["101/101"]]); + expect(progress.done).toHaveBeenCalledTimes(1); + }); + + it("logs and skips failed database scan items when counting progress", async () => { + const { hiddenFileSync, progress, log, trackDatabaseFileModification } = createHiddenFileSync(); + const files = createDatabaseEntries(101); + const failingPath = ".obsidian/plugins/example/50.json" as FilePath; + const error = new Error("Database scan item failed"); + trackDatabaseFileModification.mockImplementation(async (path) => { + if (path === failingPath) throw error; + return false; + }); + + await hiddenFileSync.trackScannedDatabaseChange(files); + + expect(log).toHaveBeenCalledWith( + expect.stringContaining("Failed to process storage change file:"), + expect.anything() + ); + expect(log).toHaveBeenCalledWith(error, LOG_LEVEL_VERBOSE); + expect(progress.log.mock.calls).toEqual([["1/101"]]); + expect(progress.done).toHaveBeenCalledTimes(1); + }); + + it("uses the database-file progress interval and counts false returns", async () => { + const { hiddenFileSync, progress, trackDatabaseFileModification } = createHiddenFileSync(); + const files = createDatabaseEntries(26); + + await expect(hiddenFileSync.useDatabaseFiles(files)).resolves.toBe(true); + + expect(trackDatabaseFileModification).toHaveBeenCalledTimes(26); + expect(progress.log.mock.calls).toEqual([["Scanning database files..."], ["1/26"], ["26/26"]]); + expect(progress.done).toHaveBeenCalledTimes(1); + }); + + it("logs and skips failed database-file items when counting progress", async () => { + const { hiddenFileSync, progress, log, trackDatabaseFileModification } = createHiddenFileSync(); + const files = createDatabaseEntries(26); + const failingPath = ".obsidian/plugins/example/12.json" as FilePath; + const error = new Error("Database file item failed"); + trackDatabaseFileModification.mockImplementation(async (path) => { + if (path === failingPath) throw error; + return false; + }); + + await expect(hiddenFileSync.useDatabaseFiles(files)).resolves.toBe(true); + + expect(log).toHaveBeenCalledWith(expect.stringContaining("Failed to process database changes:")); + expect(log).toHaveBeenCalledWith(error, LOG_LEVEL_VERBOSE); + expect(progress.log.mock.calls).toEqual([["Scanning database files..."], ["1/26"]]); + expect(progress.done).toHaveBeenCalledTimes(1); + }); + + it("starts all storage work eagerly and calls done only after all promises resolve", async () => { + const { hiddenFileSync, progress, trackStorageFileModification } = createHiddenFileSync(); + const files = createStoragePaths(101); + const resolveModifications: Array<(result: boolean) => void> = []; + trackStorageFileModification.mockImplementation( + () => new Promise((resolve) => resolveModifications.push(resolve)) + ); + + const operation = hiddenFileSync.trackScannedStorageChanges(files); + expect(trackStorageFileModification).toHaveBeenCalledTimes(files.length); + expect(progress.done).not.toHaveBeenCalled(); + + for (const resolveModification of resolveModifications) resolveModification(false); + await operation; + + expect(progress.done).toHaveBeenCalledTimes(1); + }); + + it("does not call done when the item error handler rejects the batch", async () => { + const { hiddenFileSync, progress, trackStorageFileModification } = createHiddenFileSync(); + const itemError = new Error("Storage item failed"); + const handlerError = new Error("Progress error handler failed"); + trackStorageFileModification.mockRejectedValueOnce(itemError); + progress.once.mockImplementation(() => { + throw handlerError; + }); + + await expect(hiddenFileSync.trackScannedStorageChanges(createStoragePaths(1))).rejects.toBe(handlerError); + expect(progress.done).not.toHaveBeenCalled(); + }); +}); diff --git a/src/features/HiddenFileSync/CmdHiddenFileSync.ts b/src/features/HiddenFileSync/CmdHiddenFileSync.ts index 04588bc1..d3911831 100644 --- a/src/features/HiddenFileSync/CmdHiddenFileSync.ts +++ b/src/features/HiddenFileSync/CmdHiddenFileSync.ts @@ -35,7 +35,6 @@ import { getLogLevel, autosaveCache, type MapLike, - onlyInNTimes, BASE_IS_NEW, EVEN, displayRev, @@ -45,6 +44,7 @@ import { serialized, skipIfDuplicated } from "octagonal-wheels/concurrency/lock" import { JsonResolveModal } from "@/features/HiddenFileCommon/JsonResolveModal.ts"; 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 { hiddenFilesEventCount, @@ -57,6 +57,7 @@ import { tryGetFilePath } from "@vrtmrz/livesync-commonlib/compat/common/utils.d import type { HandlerOutcome } from "@vrtmrz/livesync-commonlib/compat/services/lib/HandlerUtils"; import type { OptionalFileEventOptions } from "@vrtmrz/livesync-commonlib/compat/services/base/IService"; import { configureHiddenFileSyncMode, type ConfigureHiddenFileSyncResult } from "./configureHiddenFileSyncMode.ts"; +import { performBatchWithProgress } from "./performBatchWithProgress.ts"; import type { OptionalSyncFeatureMode } from "@/features/optionalSyncFeatures.ts"; import { getObsidianCommunityPluginManager } from "@/common/obsidianCommunityPlugins.ts"; import { $msg } from "@/common/translation"; @@ -571,18 +572,15 @@ export class HiddenFileSync extends LiveSyncCommands { ) { const logLevel = getLogLevel(showNotice); const p = this._progress(`[⚙ Storage -> DB ]\n`, logLevel); - const notifyProgress = onlyInNTimes(100, (progress) => p.log(`${progress}/${processFiles.length}`)); - const processes = processFiles.map(async (file, i) => { - try { - await this.trackStorageFileModification(file, onlyNew, forceWriteAll, includeDeleted); - notifyProgress(); - } catch (ex) { + await performBatchWithProgress(processFiles, 100, { + processItem: (file) => this.trackStorageFileModification(file, onlyNew, forceWriteAll, includeDeleted), + onProgress: (progress) => p.log(`${progress}/${processFiles.length}`), + onItemError: (ex, file) => { p.once(`Failed to process storage change file:${file}`); this._log(ex, LOG_LEVEL_VERBOSE); - } + }, + onDone: () => p.done(), }); - await Promise.all(processes); - p.done(); } async scanAllStorageChanges( showNotice: boolean = false, @@ -1134,9 +1132,8 @@ Offline Changed files: ${processFiles.length}`; ) { const logLevel = getLogLevel(showNotice); const p = this._progress(`[⚙ DB -> Storage ]\n`, logLevel); - const notifyProgress = onlyInNTimes(100, (progress) => p.log(`${progress}/${processFiles.length}`)); - const processes = processFiles.map(async (file) => { - try { + await performBatchWithProgress(processFiles, 100, { + processItem: async (file) => { const path = stripAllPrefixes(this.getPath(file)); if (!(await this.isTargetFile(path))) { this._log( @@ -1153,14 +1150,14 @@ Offline Changed files: ${processFiles.length}`; includeDeletion ); } - notifyProgress(); - } catch (ex) { + }, + onProgress: (progress) => p.log(`${progress}/${processFiles.length}`), + onItemError: (ex, file) => { this._log(`Failed to process storage change file:${tryGetFilePath(file)}`, logLevel); this._log(ex, LOG_LEVEL_VERBOSE); - } + }, + onDone: () => p.done(), }); - await Promise.all(processes); - p.done(); } async applyOfflineChanges(showNotice: boolean) { @@ -1188,9 +1185,14 @@ Untracked database files: ${untrackedDatabase.length} Common untracked files: ${bothUntracked.length}`; p.once(stat); const semaphores = Semaphore(10); - const notifyProgress = onlyInNTimes(25, (progress) => p.log(`${progress}/${bothUntracked.length}`)); + const notifyProgress = createCoalescingNotifier({ + everyCount: 25, + leading: true, + onNotify: (progress) => p.log(`${progress}/${bothUntracked.length}`), + }); + let attempted = 0; const allProcesses = bothUntracked.map(async (file) => { - notifyProgress(); + notifyProgress.update(++attempted); const rel = await semaphores.acquire(); try { const fileStat = await this.core.storageAccess.statHidden(file); @@ -1260,20 +1262,18 @@ Offline Changed files: ${files.length}`; const logLevel = getLogLevel(showNotice); const p = this._progress(`[⚙ Scanning DB -> Storage ]\n`, logLevel); p.log("Scanning database files..."); - const notifyProgress = onlyInNTimes(25, (progress) => p.log(`${progress}/${files.length}`)); - const processFiles = files.map(async (file) => { - try { + await performBatchWithProgress(files, 25, { + processItem: async (file) => { const path = stripAllPrefixes(this.getPath(file)); await this.trackDatabaseFileModification(path, "[Scanning]", true, onlyNew, file); - notifyProgress(); - } catch (ex) { + }, + onProgress: (progress) => p.log(`${progress}/${files.length}`), + onItemError: (ex, file) => { this._log(`Failed to process database changes:${tryGetFilePath(file)}`); this._log(ex, LOG_LEVEL_VERBOSE); - } - return; + }, + onDone: () => p.done(), }); - await Promise.all(processFiles); - p.done(); return true; } @@ -1435,9 +1435,14 @@ Offline Changed files: ${files.length}`; const storageToDatabase = [] as FilePath[]; const databaseToStorage = [] as MetaEntry[]; - const eachProgress = onlyInNTimes(100, (progress) => p.log(`Checking ${progress}/${allFileNames.size}`)); + const eachProgress = createCoalescingNotifier({ + everyCount: 100, + leading: true, + onNotify: (progress) => p.log(`Checking ${progress}/${allFileNames.size}`), + }); + let attempted = 0; for (const file of allFileNames) { - eachProgress(); + eachProgress.update(++attempted); const storageMTime = await this.core.storageAccess.statHidden(file); const mtimeStorage = getComparingMTime(storageMTime); const dbEntry = allDatabaseMap.get(file)!; diff --git a/src/features/HiddenFileSync/CmdHiddenFileSync.unit.spec.ts b/src/features/HiddenFileSync/CmdHiddenFileSync.unit.spec.ts index 9018b222..4b3e5543 100644 --- a/src/features/HiddenFileSync/CmdHiddenFileSync.unit.spec.ts +++ b/src/features/HiddenFileSync/CmdHiddenFileSync.unit.spec.ts @@ -98,6 +98,85 @@ function createHiddenRevisionOperation() { }; } +describe("HiddenFileSync storage loading", () => { + function createStorageReader(content: string | ArrayBuffer) { + const path = ".obsidian/plugins/example/data.json" as FilePath; + const stat = { + ctime: 123, + mtime: 456, + size: typeof content === "string" ? new TextEncoder().encode(content).byteLength : content.byteLength, + type: "file" as const, + }; + const storageAccess = { + statHidden: vi.fn(async (): Promise => stat), + readHiddenFileAuto: vi.fn(async () => content), + }; + const hiddenFileSync = Object.create(HiddenFileSync.prototype) as HiddenFileSync; + Object.assign(hiddenFileSync, { core: { storageAccess } }); + return { hiddenFileSync, storageAccess, path, stat }; + } + + it.each([ + { + kind: "text", + content: "first\r\n日本語 🐙", + expected: new TextEncoder().encode("first\r\n日本語 🐙"), + mime: "text/plain", + }, + { + kind: "binary", + content: new Uint8Array([0, 128, 255]).buffer, + expected: new Uint8Array([0, 128, 255]), + mime: "application/octet-stream", + }, + { kind: "empty text", content: "", expected: new Uint8Array(0), mime: "text/plain" }, + { + kind: "empty binary", + content: new ArrayBuffer(0), + expected: new Uint8Array(0), + mime: "application/octet-stream", + }, + ])("preserves $kind content and its file metadata", async ({ content, expected, mime }) => { + const { hiddenFileSync, storageAccess, path, stat } = createStorageReader(content); + + const file = await hiddenFileSync.loadFileWithInfo(path); + + expect(storageAccess.statHidden).toHaveBeenCalledWith(path); + expect(storageAccess.readHiddenFileAuto).toHaveBeenCalledWith(path); + expect(file).toMatchObject({ name: "data.json", path, stat, isInternal: true, deleted: false }); + expect(file.body).toBeInstanceOf(Blob); + expect(file.body.type).toBe(mime); + expect(new Uint8Array(await file.body.arrayBuffer())).toEqual(expected); + }); + + it("represents a missing file as a deletion without reading storage content", async () => { + const { hiddenFileSync, storageAccess, path } = createStorageReader("unavailable"); + storageAccess.statHidden.mockResolvedValue(null); + + const file = await hiddenFileSync.loadFileWithInfo(path); + + expect(storageAccess.readHiddenFileAuto).not.toHaveBeenCalled(); + expect(file).toMatchObject({ + name: "data.json", + path, + stat: { ctime: 0, mtime: 0, size: 0, type: "file" }, + isInternal: true, + deleted: true, + }); + expect(file.body.type).toBe("application/octet-stream"); + expect(file.body.size).toBe(0); + }); + + it("propagates a read failure for an existing file", async () => { + const { hiddenFileSync, storageAccess, path } = createStorageReader("unavailable"); + const error = new Error("Storage read failed"); + storageAccess.readHiddenFileAuto.mockRejectedValue(error); + + await expect(hiddenFileSync.loadFileWithInfo(path)).rejects.toBe(error); + expect(storageAccess.readHiddenFileAuto).toHaveBeenCalledWith(path); + }); +}); + describe("HiddenFileSync configuration-change notices", () => { it.each(["pushForce", "pullForce", "safe"] as const)( "refuses %s initialisation before processing files while Hidden File Sync is disabled", diff --git a/src/features/HiddenFileSync/performBatchWithProgress.ts b/src/features/HiddenFileSync/performBatchWithProgress.ts new file mode 100644 index 00000000..7c0fd2dd --- /dev/null +++ b/src/features/HiddenFileSync/performBatchWithProgress.ts @@ -0,0 +1,49 @@ +import { createCoalescingNotifier } from "octagonal-wheels/concurrency/coalescingNotifier"; + +interface PerformBatchWithProgressCallbacks { + /** Runs one item. Its resolved value is ignored. */ + processItem: (item: T) => Promise; + /** Receives the number of successfully completed items. */ + onProgress: (completed: number) => void; + /** Handles an item or progress callback error. */ + onItemError: (error: unknown, item: T) => void; + /** Runs once after all per-item promises resolve. */ + onDone: () => void; +} + +/** + * Runs an eager batch and coalesces progress notifications by successful item count. + * @internal + * + * @param items - Items to process concurrently. + * @param eachCount - Number of completed items between progress notifications. + * @param callbacks - Item processing, progress, error, and completion callbacks. + * @returns A promise that resolves with no value after the batch and completion callback finish. + * @throws When notifier configuration is invalid, an error handler throws, or the completion callback throws. + * @remarks The first completed item triggers progress immediately. Work and the progress callback share + * one error boundary; errors are passed to `onItemError`. A rejected item operation does not advance + * the count. The batch has no concurrency limit or retry, and pending progress is not flushed when + * processing ends. + */ +export async function performBatchWithProgress( + items: readonly T[], + eachCount: number, + callbacks: PerformBatchWithProgressCallbacks +): Promise { + const notifier = createCoalescingNotifier({ + everyCount: eachCount, + leading: true, + onNotify: callbacks.onProgress, + }); + let completed = 0; + const processes = items.map(async (item) => { + try { + await callbacks.processItem(item); + notifier.update(++completed); + } catch (error) { + callbacks.onItemError(error, item); + } + }); + await Promise.all(processes); + callbacks.onDone(); +} diff --git a/src/features/HiddenFileSync/performBatchWithProgress.unit.spec.ts b/src/features/HiddenFileSync/performBatchWithProgress.unit.spec.ts new file mode 100644 index 00000000..ba0d7167 --- /dev/null +++ b/src/features/HiddenFileSync/performBatchWithProgress.unit.spec.ts @@ -0,0 +1,93 @@ +import { describe, expect, it, vi } from "vitest"; +import { performBatchWithProgress } from "./performBatchWithProgress.ts"; + +describe("performBatchWithProgress", () => { + it("counts resolved false values, continues after item errors, and does not flush partial progress", async () => { + const error = new Error("Item failed"); + const processItem = vi.fn(async (item: number) => { + if (item === 2) throw error; + return false; + }); + const onProgress = vi.fn(); + const onItemError = vi.fn(); + const onDone = vi.fn(); + + const result = await performBatchWithProgress([1, 2, 3, 4, 5, 6, 7], 2, { + processItem, + onProgress, + onItemError, + onDone, + }); + + expect(result).toBeUndefined(); + expect(processItem).toHaveBeenCalledTimes(7); + expect(onItemError).toHaveBeenCalledWith(error, 2); + expect(onProgress.mock.calls).toEqual([[1], [3], [5]]); + expect(onDone).toHaveBeenCalledTimes(1); + }); + + it("starts every item eagerly and calls onDone only after all item promises resolve", async () => { + const items = [1, 2, 3]; + const resolvers: Array<() => void> = []; + const processItem = vi.fn( + () => + new Promise((resolve) => { + resolvers.push(resolve); + }) + ); + const onProgress = vi.fn(); + const onItemError = vi.fn(); + const onDone = vi.fn(); + + const operation = performBatchWithProgress(items, 100, { processItem, onProgress, onItemError, onDone }); + + expect(processItem).toHaveBeenCalledTimes(items.length); + expect(onDone).not.toHaveBeenCalled(); + for (const resolve of resolvers) resolve(); + await operation; + + expect(onDone).toHaveBeenCalledTimes(1); + }); + + it("routes a progress callback error through the item error handler", async () => { + const error = new Error("Progress callback failed"); + const onProgress = vi.fn(() => { + throw error; + }); + const onItemError = vi.fn(); + const onDone = vi.fn(); + + await performBatchWithProgress(["item"], 100, { + processItem: async () => false, + onProgress, + onItemError, + onDone, + }); + + expect(onItemError).toHaveBeenCalledWith(error, "item"); + expect(onDone).toHaveBeenCalledTimes(1); + }); + + it("rejects when the item error handler throws and skips onDone", async () => { + const itemError = new Error("Item failed"); + const handlerError = new Error("Error handler failed"); + const onItemError = vi.fn(() => { + throw handlerError; + }); + const onDone = vi.fn(); + + await expect( + performBatchWithProgress(["item"], 100, { + processItem: async () => { + throw itemError; + }, + onProgress: vi.fn(), + onItemError, + onDone, + }) + ).rejects.toBe(handlerError); + + expect(onItemError).toHaveBeenCalledWith(itemError, "item"); + expect(onDone).not.toHaveBeenCalled(); + }); +}); diff --git a/src/modules/features/StatusBarDisplay.timeout.unit.spec.ts b/src/modules/features/StatusBarDisplay.timeout.unit.spec.ts new file mode 100644 index 00000000..15bda8f1 --- /dev/null +++ b/src/modules/features/StatusBarDisplay.timeout.unit.spec.ts @@ -0,0 +1,142 @@ +import { reactiveSource } from "octagonal-wheels/dataobject/reactive"; +import { createDeferredAction, type TimeoutScheduler } from "octagonal-wheels/concurrency/deferredAction"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { + STATUS_COUNTER_INACTIVE_LINGER_MS, + createChunkFetchCounterLabel, + createMinimumVisibleActivityCount, + createPaddedCounterLabel, +} from "./StatusBarDisplay.ts"; + +vi.mock("octagonal-wheels/concurrency/deferredAction", { spy: true }); + +function createdControllers() { + return vi.mocked(createDeferredAction).mock.results.map((result) => { + if (result.type !== "return") throw new Error("Expected a real timeout controller"); + return result.value; + }); +} + +function createScheduler() { + const entries: { callback: () => void; delayMs: number; cancel: ReturnType }[] = []; + const schedule = vi.fn((callback, delayMs) => { + const cancel = vi.fn(); + entries.push({ callback, delayMs, cancel }); + return cancel; + }); + return { schedule, entries }; +} + +beforeEach(() => { + vi.clearAllMocks(); + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-10-24T00:00:00Z")); +}); + +afterEach(() => { + vi.clearAllTimers(); + vi.restoreAllMocks(); + vi.useRealTimers(); +}); + +describe("status timeout composition", () => { + it("uses the real shared controller for minimum visibility, resumption, and disposal", () => { + const scheduler = createScheduler(); + const source = reactiveSource(0); + const display = createMinimumVisibleActivityCount(source, 150, scheduler.schedule); + expect(createDeferredAction).toHaveBeenCalledExactlyOnceWith(expect.any(Function), scheduler.schedule); + const controller = createdControllers()[0]; + const schedule = vi.spyOn(controller, "schedule"); + const cancel = vi.spyOn(controller, "cancel"); + const dispose = vi.spyOn(controller, "dispose"); + + source.value = 1; + vi.advanceTimersByTime(50); + source.value = 0; + expect(schedule).toHaveBeenCalledExactlyOnceWith(100); + expect(scheduler.entries[0].delayMs).toBe(100); + source.value = 2; + expect(cancel).toHaveBeenCalledTimes(3); + expect(scheduler.entries[0].cancel).toHaveBeenCalledTimes(1); + scheduler.entries[0].callback(); + expect(display.value).toBe(2); + + source.value = 0; + scheduler.entries[1].callback(); + expect(display.value).toBe(0); + source.value = 3; + source.value = 0; + display.dispose(); + display.dispose(); + scheduler.entries[2].callback(); + expect(display.value).toBe(3); + expect(scheduler.entries[2].cancel).toHaveBeenCalledTimes(1); + expect(dispose).toHaveBeenCalledTimes(1); + }); + + it("uses the shared controller for the counter linger and cancels through it when activity resumes", () => { + const scheduler = createScheduler(); + const source = reactiveSource(0); + const display = createPaddedCounterLabel(source, "A", 25, scheduler.schedule); + expect(createDeferredAction).toHaveBeenCalledExactlyOnceWith(expect.any(Function), scheduler.schedule); + const controller = createdControllers()[0]; + const schedule = vi.spyOn(controller, "schedule"); + const cancel = vi.spyOn(controller, "cancel"); + const dispose = vi.spyOn(controller, "dispose"); + source.value = 1; + source.value = 0; + expect(schedule).toHaveBeenCalledExactlyOnceWith(25); + source.value = 2; + expect(cancel).toHaveBeenCalledTimes(3); + expect(scheduler.entries[0].cancel).toHaveBeenCalledTimes(1); + scheduler.entries[0].callback(); + expect(display.value).toBe(" A\u20072"); + expect(display.count.value).toBe(2); + + source.value = 0; + scheduler.entries[1].callback(); + expect(display.value).toBe(""); + expect(display.count.value).toBeUndefined(); + source.value = 3; + source.value = 0; + display.dispose(); + scheduler.entries[2].callback(); + expect(display.value).toBe(" A\u20070"); + expect(display.count.value).toBe(0); + expect(scheduler.entries[2].cancel).toHaveBeenCalledTimes(1); + expect(dispose).toHaveBeenCalledTimes(1); + }); + + it("passes the scheduler to independent chunk-count controllers and disposes both children", () => { + const scheduler = createScheduler(); + const source = reactiveSource({ initial: 2, retrying: 3 }); + const display = createChunkFetchCounterLabel(source, scheduler.schedule); + expect(display.initialCount.value).toBe(2); + expect(display.retryingCount.value).toBe(3); + expect(createDeferredAction).toHaveBeenCalledTimes(2); + expect(vi.mocked(createDeferredAction).mock.calls.map((args) => args[1])).toEqual([ + scheduler.schedule, + scheduler.schedule, + ]); + const controllers = createdControllers(); + const disposers = controllers.map((controller) => vi.spyOn(controller, "dispose")); + source.value = { initial: 0, retrying: 0 }; + expect(scheduler.entries.map((entry) => entry.delayMs)).toEqual([ + STATUS_COUNTER_INACTIVE_LINGER_MS, + STATUS_COUNTER_INACTIVE_LINGER_MS, + ]); + scheduler.entries[0].callback(); + expect(display.value.replace(/\u2007/g, "")).toBe(" 🔁0"); + expect(display.initialCount.value).toBeUndefined(); + expect(display.retryingCount.value).toBe(0); + display.dispose(); + display.dispose(); + expect(scheduler.entries[0].cancel).not.toHaveBeenCalled(); + expect(scheduler.entries[1].cancel).toHaveBeenCalledTimes(1); + for (const dispose of disposers) expect(dispose).toHaveBeenCalledTimes(1); + scheduler.entries[1].callback(); + expect(display.value.replace(/\u2007/g, "")).toBe(" 🔁0"); + expect(display.initialCount.value).toBeUndefined(); + expect(display.retryingCount.value).toBe(0); + }); +}); diff --git a/src/modules/features/StatusBarDisplay.ts b/src/modules/features/StatusBarDisplay.ts index 9c24a972..cc63aca9 100644 --- a/src/modules/features/StatusBarDisplay.ts +++ b/src/modules/features/StatusBarDisplay.ts @@ -1,7 +1,13 @@ import { reactiveSource, type ReactiveSource, type ReactiveValue } from "octagonal-wheels/dataobject/reactive"; +import { createDeferredAction, type TimeoutScheduler } from "octagonal-wheels/concurrency/deferredAction"; import { compatGlobal } from "@vrtmrz/livesync-commonlib/compat/common/coreEnvFunctions"; +const scheduleStatusTimeout: TimeoutScheduler = (callback, delayMs) => { + const handle = compatGlobal.setTimeout(callback, delayMs); + return () => compatGlobal.clearTimeout(handle); +}; + const STATUS_COUNTER_PADDING = "\u2007".repeat(10); export const STATUS_COUNTER_INACTIVE_LINGER_MS = 3_000; @@ -37,32 +43,32 @@ function asDisposableReactiveValue(value: ReactiveSource, dispose: () => v /** * Mirrors an activity count while keeping each visible period on screen for a * minimum total lifetime. The delay applies only when the source becomes zero. + * + * @param source - The activity count to observe. + * @param minimumVisibleMs - The minimum visible lifetime, clamped to zero. + * @param scheduleTimeout - The scheduler for this display. Defaults to the + * existing compatible host, preserving its timer method receivers. */ export function createMinimumVisibleActivityCount( source: ReactiveValue, - minimumVisibleMs: number + minimumVisibleMs: number, + scheduleTimeout: TimeoutScheduler = scheduleStatusTimeout ): DisposableReactiveValue { const minimumLifetime = Math.max(0, minimumVisibleMs); const displayed = reactiveSource(Math.max(0, source.value)); let visibleSince = displayed.value > 0 ? Date.now() : undefined; - let hideTimer: number | undefined; let disposed = false; - const cancelHide = () => { - if (hideTimer === undefined) return; - compatGlobal.clearTimeout(hideTimer); - hideTimer = undefined; - }; const hideIfIdle = () => { - hideTimer = undefined; if (disposed || Math.max(0, source.value) > 0) return; displayed.value = 0; visibleSince = undefined; }; + const hideTimeout = createDeferredAction(hideIfIdle, scheduleTimeout); const update = () => { if (disposed) return; const nextCount = Math.max(0, source.value); - cancelHide(); + hideTimeout.cancel(); if (nextCount > 0) { if (displayed.value === 0) { visibleSince = Date.now(); @@ -80,7 +86,7 @@ export function createMinimumVisibleActivityCount( if (remaining === 0) { hideIfIdle(); } else { - hideTimer = compatGlobal.setTimeout(hideIfIdle, remaining); + hideTimeout.schedule(remaining); } }; @@ -88,7 +94,7 @@ export function createMinimumVisibleActivityCount( return asDisposableReactiveValue(displayed, () => { if (disposed) return; disposed = true; - cancelHide(); + hideTimeout.dispose(); source.offChanged(update); }); } @@ -96,24 +102,31 @@ export function createMinimumVisibleActivityCount( /** * Formats a counter with a stable width and briefly retains its zero value so * that the completion of queued work remains visible. + * + * @param source - The counter to observe. + * @param mark - The label prefix before the padded count. + * @param inactiveLingerMs - The delay before clearing a zero count, clamped to zero. + * @param scheduleTimeout - The scheduler for this display. Defaults to the + * existing compatible host, preserving its timer method receivers. */ export function createPaddedCounterLabel( source: ReactiveValue, mark: string, - inactiveLingerMs = STATUS_COUNTER_INACTIVE_LINGER_MS + inactiveLingerMs = STATUS_COUNTER_INACTIVE_LINGER_MS, + scheduleTimeout: TimeoutScheduler = scheduleStatusTimeout ): PaddedCounterLabel { const linger = Math.max(0, inactiveLingerMs); const formatted = reactiveSource(""); const displayedCount = reactiveSource(undefined); let maximumLength = 1; - let clearTimer: number | undefined; let disposed = false; - const cancelClear = () => { - if (clearTimer === undefined) return; - compatGlobal.clearTimeout(clearTimer); - clearTimer = undefined; - }; + const clearLabelTimeout = createDeferredAction(() => { + if (disposed) return; + formatted.value = ""; + displayedCount.value = undefined; + maximumLength = 1; + }, scheduleTimeout); const format = (count: number) => { const requiredLength = `${Math.abs(count)}`.length + 1; maximumLength = Math.max(maximumLength, requiredLength); @@ -121,25 +134,19 @@ export function createPaddedCounterLabel( }; const update = () => { if (disposed) return; - cancelClear(); + clearLabelTimeout.cancel(); const count = source.value; formatted.value = format(count); displayedCount.value = count; if (count !== 0) return; - clearTimer = compatGlobal.setTimeout(() => { - clearTimer = undefined; - if (disposed) return; - formatted.value = ""; - displayedCount.value = undefined; - maximumLength = 1; - }, linger); + clearLabelTimeout.schedule(linger); }; source.onChanged(update); const display = asDisposableReactiveValue(formatted, () => { if (disposed) return; disposed = true; - cancelClear(); + clearLabelTimeout.dispose(); source.offChanged(update); }); return { @@ -162,14 +169,28 @@ export function createPaddedCounterLabel( /** * Displays the disjoint initial and retry chunk-fetch counts with the same * padding and inactive linger behaviour as the other status counters. + * + * @param source - The initial and retrying chunk-fetch counts. + * @param scheduleTimeout - The scheduler passed to each independent child display. */ export function createChunkFetchCounterLabel( - source: ReactiveValue<{ initial: number; retrying: number }> + source: ReactiveValue<{ initial: number; retrying: number }>, + scheduleTimeout: TimeoutScheduler = scheduleStatusTimeout ): ChunkFetchCounterLabel { const initialCount = reactiveSource(0); const retryingCount = reactiveSource(0); - const initialLabel = createPaddedCounterLabel(initialCount, "🛄"); - const retryingLabel = createPaddedCounterLabel(retryingCount, "🔁"); + const initialLabel = createPaddedCounterLabel( + initialCount, + "🛄", + STATUS_COUNTER_INACTIVE_LINGER_MS, + scheduleTimeout + ); + const retryingLabel = createPaddedCounterLabel( + retryingCount, + "🔁", + STATUS_COUNTER_INACTIVE_LINGER_MS, + scheduleTimeout + ); const formatted = reactiveSource(`${initialLabel.value}${retryingLabel.value}`); let updatingCounts = false; let disposed = false; diff --git a/src/modules/features/StatusBarDisplay.unit.spec.ts b/src/modules/features/StatusBarDisplay.unit.spec.ts index 5ce8f69e..c704cce4 100644 --- a/src/modules/features/StatusBarDisplay.unit.spec.ts +++ b/src/modules/features/StatusBarDisplay.unit.spec.ts @@ -1,3 +1,4 @@ +import { compatGlobal } from "@vrtmrz/livesync-commonlib/compat/common/coreEnvFunctions"; import { reactive, reactiveSource } from "octagonal-wheels/dataobject/reactive"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; @@ -194,3 +195,121 @@ describe("createChunkFetchCounterLabel", () => { expect(withoutPadding()).toBe(" 🔁0"); }); }); + +describe("status timeout ownership", () => { + beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-10-24T00:00:00Z")); + }); + + afterEach(() => { + vi.restoreAllMocks(); + vi.useRealTimers(); + }); + + it("counts the minimum lifetime from construction when activity is already visible", () => { + const source = reactiveSource(2); + const display = createMinimumVisibleActivityCount(source, 150); + vi.advanceTimersByTime(40); + source.value = 0; + + vi.advanceTimersByTime(109); + expect(display.value).toBe(2); + vi.advanceTimersByTime(1); + expect(display.value).toBe(0); + display.dispose(); + }); + + it("hides immediately for a non-positive minimum lifetime and clamps negative activity", () => { + const source = reactiveSource(-2); + const display = createMinimumVisibleActivityCount(source, -10); + expect(display.value).toBe(0); + source.value = 1; + source.value = -2; + + expect(display.value).toBe(0); + expect(vi.getTimerCount()).toBe(0); + display.dispose(); + }); + + it("keeps a zero-duration counter clear asynchronous", () => { + const source = reactiveSource(0); + const display = createPaddedCounterLabel(source, "A", 0); + source.value = 1; + source.value = 0; + + expect(display.value).toBe(" A\u20070"); + vi.advanceTimersByTime(0); + expect(display.value).toBe(""); + display.dispose(); + }); + + it.each(["activity", "label"] as const)( + "retains timer receivers, zero handles, and idempotent disposal for %s", + (kind) => { + const source = reactiveSource(0); + const offChanged = vi.spyOn(source, "offChanged"); + const setTimer = vi.spyOn(compatGlobal as Window, "setTimeout").mockReturnValue(0); + const clearTimer = vi.spyOn(compatGlobal as Window, "clearTimeout").mockImplementation(() => {}); + const display = + kind === "activity" + ? createMinimumVisibleActivityCount(source, 150) + : createPaddedCounterLabel(source, "A", 150); + + source.value = 1; + source.value = 0; + expect(setTimer).toHaveBeenCalledTimes(1); + expect(setTimer.mock.contexts[0]).toBe(compatGlobal); + expect(setTimer.mock.calls[0][1]).toBe(150); + source.value = 2; + expect(clearTimer).toHaveBeenCalledExactlyOnceWith(0); + source.value = 0; + display.dispose(); + display.dispose(); + source.value = 3; + + expect(setTimer).toHaveBeenCalledTimes(2); + expect(clearTimer).toHaveBeenCalledTimes(2); + expect(clearTimer.mock.calls).toEqual([[0], [0]]); + expect(clearTimer.mock.contexts).toEqual([compatGlobal, compatGlobal]); + expect(offChanged).toHaveBeenCalledTimes(1); + } + ); + + it("keeps pending clears independent between display instances", () => { + const firstSource = reactiveSource(0); + const secondSource = reactiveSource(0); + const first = createPaddedCounterLabel(firstSource, "A", 100); + const second = createPaddedCounterLabel(secondSource, "B", 100); + firstSource.value = 1; + firstSource.value = 0; + secondSource.value = 1; + secondSource.value = 0; + expect(vi.getTimerCount()).toBe(2); + + first.dispose(); + expect(vi.getTimerCount()).toBe(1); + vi.advanceTimersByTime(100); + expect(first.value).toBe(" A\u20070"); + expect(second.value).toBe(""); + second.dispose(); + expect(vi.getTimerCount()).toBe(0); + }); + + it("disposes both child timers and its source subscription for chunk counts", () => { + const source = reactiveSource({ initial: 2, retrying: 3 }); + const offChanged = vi.spyOn(source, "offChanged"); + const display = createChunkFetchCounterLabel(source); + source.value = { initial: 0, retrying: 0 }; + const lastLabel = display.value; + expect(vi.getTimerCount()).toBe(2); + + display.dispose(); + display.dispose(); + expect(vi.getTimerCount()).toBe(0); + vi.advanceTimersByTime(STATUS_COUNTER_INACTIVE_LINGER_MS); + source.value = { initial: 4, retrying: 5 }; + expect(display.value).toBe(lastLabel); + expect(offChanged).toHaveBeenCalledTimes(1); + }); +}); diff --git a/src/serviceFeatures/replication/ReplicateResultProcessor.counting.unit.spec.ts b/src/serviceFeatures/replication/ReplicateResultProcessor.counting.unit.spec.ts new file mode 100644 index 00000000..2a76766c --- /dev/null +++ b/src/serviceFeatures/replication/ReplicateResultProcessor.counting.unit.spec.ts @@ -0,0 +1,46 @@ +import { reactiveSource } from "octagonal-wheels/dataobject/reactive"; +import { promiseWithResolvers } from "octagonal-wheels/promises"; +import { describe, expect, it } from "vitest"; +import { ReplicateResultProcessor } from "./ReplicateResultProcessor"; + +describe("ReplicateResultProcessor.withCounting", () => { + it("counts overlapping callbacks until each settles and preserves results and errors", async () => { + const originalCount = 7; + const count = reactiveSource(originalCount); + const processor = new ReplicateResultProcessor({} as never); + const firstResult = { task: "first" }; + const firstResultGate = promiseWithResolvers(); + const secondResultGate = promiseWithResolvers(); + const firstStarted = promiseWithResolvers(); + const secondStarted = promiseWithResolvers(); + const failure = new Error("second callback failed"); + let firstCountAtStart = 0; + let secondCountAtStart = 0; + + const first = processor.withCounting(async () => { + firstCountAtStart = count.value; + firstStarted.resolve(undefined); + return firstResultGate.promise; + }, count); + await firstStarted.promise; + expect(firstCountAtStart).toBe(originalCount + 1); + expect(count.value).toBe(originalCount + 1); + + const second = processor.withCounting(async () => { + secondCountAtStart = count.value; + secondStarted.resolve(undefined); + return secondResultGate.promise; + }, count); + await secondStarted.promise; + expect(secondCountAtStart).toBe(originalCount + 2); + expect(count.value).toBe(originalCount + 2); + + firstResultGate.resolve(firstResult); + await expect(first).resolves.toBe(firstResult); + expect(count.value).toBe(originalCount + 1); + + secondResultGate.reject(failure); + await expect(second).rejects.toBe(failure); + expect(count.value).toBe(originalCount); + }); +}); diff --git a/src/serviceFeatures/replication/ReplicateResultProcessor.ts b/src/serviceFeatures/replication/ReplicateResultProcessor.ts index 523043d9..43310f34 100644 --- a/src/serviceFeatures/replication/ReplicateResultProcessor.ts +++ b/src/serviceFeatures/replication/ReplicateResultProcessor.ts @@ -25,6 +25,7 @@ import { } from "@vrtmrz/livesync-commonlib/compat/common/logger"; import { fireAndForget, isAnyNote, throttle } from "@vrtmrz/livesync-commonlib/compat/common/utils"; import { Semaphore } from "octagonal-wheels/concurrency/semaphore_v2"; +import { withCounting } from "octagonal-wheels/concurrency/counting"; import { serialized } from "octagonal-wheels/concurrency/lock"; import type { ReactiveSource } from "octagonal-wheels/dataobject/reactive_v2"; import type { LiveSyncBaseCore } from "@/LiveSyncBaseCore"; @@ -221,12 +222,7 @@ export class ReplicateResultProcessor { * @returns result of the procedure */ async withCounting(proc: () => Promise, countValue: ReactiveSource) { - countValue.value++; - try { - return await proc(); - } finally { - countValue.value--; - } + return withCounting(countValue, proc); } /**