Merge shared runtime changes onto current main

This commit is contained in:
vorotamoroz committed 2026-11-01 05:14:55 +00:00
commit 6211597d2d
128 files changed
+9204 -1285

No files matched your search

@@ -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<void>((resolve) => {
finishWrite = resolve;
});
const writeStarted = new Promise<void>((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);
});
});
@@ -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),
});
@@ -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");
@@ -0,0 +1,591 @@
import { afterEach, describe, expect, it, vi } from "vitest";
vi.mock("@/deps.ts", () => ({
addIcon: vi.fn(),
diff_match_patch: class DiffMatchPatch {},
normalizePath: (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("@/modules/features/ModuleLog.ts", () => ({ MARK_DONE: "" }));
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 {
disable = vi.fn();
},
}));
vi.mock("@/common/events.ts", () => ({
EVENT_REQUEST_OPEN_PLUGIN_SYNC_DIALOG: "open-plugin-sync",
eventHub: { onEvent: vi.fn() },
}));
vi.mock("@/common/translation", () => ({ $msg: (message: string) => message }));
vi.mock("@/common/obsidianCommunityPlugins.ts", () => ({ getObsidianCommunityPluginManager: vi.fn() }));
import { get } from "svelte/store";
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";
const queueOwners: ConfigSync[] = [];
function deferred() {
let resolve!: () => void;
const promise = new Promise<void>((done) => {
resolve = done;
});
return { promise, resolve };
}
async function waitForCondition(predicate: () => boolean, message: string) {
for (let attempt = 0; attempt < 100; attempt++) {
if (predicate()) return;
await Promise.resolve();
}
throw new Error(message);
}
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<typeof ConfigSync>[0]);
queueOwners.push(configSync);
return configSync;
}
afterEach(async () => {
for (const configSync of queueOwners.splice(0)) {
await configSync.pluginScanProcessor.stop();
}
vi.restoreAllMocks();
pluginList.set([]);
pluginIsEnumerating.set(false);
vi.useRealTimers();
});
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<boolean> }
)._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<ReturnType<typeof fixture.localDatabase.getDBEntry>>;
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;
const data = {
category: "PLUGIN_MAIN",
name: "native-timers",
term: "device",
mtime: 2_000,
files: [{ filename: "main.js", data: ["stored content"], mtime: 2_000, size: 14, hash: "stored-hash" }],
};
let configSync: ConfigSync | undefined;
try {
fixture.settings.usePluginSyncV2 = false;
await expect(
fixture.localDatabase.putDBEntry(optionalFileEntry(path, JSON.stringify(data)))
).resolves.toMatchObject({ ok: true });
vi.spyOn(ConfigSync.prototype, "onBindFunction").mockImplementation(() => {});
configSync = new ConfigSync({
settings: fixture.settings,
localDatabase: fixture.localDatabase,
services: {
...fixture.services,
API: { ...fixture.services.API, addRibbonIcon: vi.fn() },
},
} as unknown as ConstructorParameters<typeof ConfigSync>[0]);
queueOwners.push(configSync);
await configSync.updatePluginList(false);
await expect(configSync.pluginScanProcessor.waitForIdle()).resolves.toBeUndefined();
expect(configSync.pluginList).toEqual([
{
...data,
documentPath: path,
files: [{ ...data.files[0], data: ["stored-hash"] }],
},
]);
expect(get(pluginList)).toEqual(configSync.pluginList);
expect(configSync.pluginScanProcessor.remaining).toBe(0);
} finally {
await configSync?.pluginScanProcessor.stop();
await fixture.dispose();
}
});
it("returns after enumeration, then loads a queued legacy catalogue item after its configured delay", async () => {
vi.useFakeTimers();
pluginList.set([]);
pluginIsEnumerating.set(false);
const path = "ix:device/PLUGIN_MAIN/example" as FilePathWithPrefix;
const catalogueData = {
category: "PLUGIN_MAIN",
name: "example",
term: "device",
mtime: 2_000,
files: [
{ filename: "main.js", data: ["legacy plugin source"], mtime: 2_000, size: 20, hash: "legacy-hash" },
],
};
const entry = { _id: path, path, mtime: 2_000, data: [JSON.stringify(catalogueData)] };
const readGate = deferred();
let loadHasStarted = false;
const localDatabase = {
findEntries: vi.fn(async function* () {
yield entry;
}),
getDBEntry: vi.fn(async () => {
loadHasStarted = true;
await readGate.promise;
return entry;
}),
};
vi.spyOn(ConfigSync.prototype, "onBindFunction").mockImplementation(() => {});
const configSync = new ConfigSync({
settings: { ...DEFAULT_SETTINGS, usePluginSync: true, usePluginSyncV2: false },
localDatabase,
services: {
API: { addLog: vi.fn(), addRibbonIcon: vi.fn() },
path: { getPath: (document: { path: string }) => document.path },
},
} as unknown as ConstructorParameters<typeof ConfigSync>[0]);
queueOwners.push(configSync);
let drained = false;
try {
await configSync.updatePluginList(false);
expect(configSync.pluginScanProcessor.remaining).toBe(1);
expect(configSync.pluginList).toEqual([]);
expect(get(pluginIsEnumerating)).toBe(false);
await vi.advanceTimersByTimeAsync(99);
expect(loadHasStarted).toBe(false);
await vi.advanceTimersByTimeAsync(1);
expect(loadHasStarted).toBe(true);
expect(configSync.pluginScanProcessor.remaining).toBe(1);
const drain = configSync.pluginScanProcessor.waitForIdle();
void drain.then(() => {
drained = true;
});
readGate.resolve();
await vi.advanceTimersByTimeAsync(1_000);
await waitForCondition(() => drained, "The legacy catalogue queue did not drain");
expect(configSync.pluginList).toEqual([
{
...catalogueData,
documentPath: path,
files: [{ ...catalogueData.files[0], data: ["legacy-hash"] }],
},
]);
expect(get(pluginList)).toEqual(configSync.pluginList);
expect(configSync.pluginScanProcessor.remaining).toBe(0);
expect(get(pluginIsEnumerating)).toBe(false);
} finally {
readGate.resolve();
await vi.advanceTimersByTimeAsync(1_000);
}
});
});
@@ -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/scoped.iteration", { spy: true });
import { LOG_LEVEL_VERBOSE, type FilePath } from "@vrtmrz/livesync-commonlib/compat/common/types";
import * as iterationModule from "octagonal-wheels/concurrency/scoped.iteration";
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<void>();
const continueEnumeration = promiseWithResolvers<void>();
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<void>());
const tenStarted = promiseWithResolvers<void>();
const lastStarted = promiseWithResolvers<void>();
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<typeof import("octagonal-wheels/concurrency/scoped.iteration")>(
"octagonal-wheels/concurrency/scoped.iteration"
);
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();
});
});
+107 -125
View File
@@ -55,11 +55,11 @@ 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";
import { Semaphore } from "octagonal-wheels/concurrency/semaphore";
import { forEachInScope } from "octagonal-wheels/concurrency/scoped.iteration";
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,10 +68,12 @@ 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 { 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";
@@ -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) {
@@ -388,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();
}
@@ -417,6 +416,7 @@ export class ConfigSync extends LiveSyncCommands {
pluginList: IPluginDataExDisplay[] = [];
private pluginListGeneration = 0;
private unloaded = false;
showPluginSyncModal() {
if (!this.isThisModuleEnabled()) {
return;
@@ -435,11 +435,18 @@ export class ConfigSync extends LiveSyncCommands {
this.pluginDialog = undefined;
}
}
onunload() {
onunload(): Promise<void> {
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() {
@@ -536,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<PluginDataExDisplay | false> {
@@ -575,77 +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();
pluginScanProcessorV2 = new QueueProcessor(
async (v: AnyEntry[]) => {
const plugin = v[0];
const path = plugin.path || this.getPath(plugin);
const oldEntry = this.pluginList.find((e) => e.documentPath == path);
if (oldEntry && oldEntry.mtime == plugin.mtime) return [];
try {
const pluginData = await this.loadPluginData(path);
if (pluginData) {
let newList = [...this.pluginList];
newList = newList.filter((x) => x.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(ex, LOG_LEVEL_VERBOSE);
}
return [];
onProgress: (remaining) => {
if (this.unloaded) return;
pluginScanningCount.value = remaining;
pluginIsEnumerating.set(remaining !== 0);
},
{
suspended: false,
batchSize: 1,
concurrentLimit: 10,
delay: 100,
yieldThreshold: 10,
maintainDelay: false,
totalRemainingReactiveSource: pluginScanningCount,
}
).startPipeline();
});
filenameToUnifiedKey(path: string, termOverRide?: string) {
const term = termOverRide || this.services.setting.getDeviceAndVaultName();
@@ -768,7 +745,7 @@ export class ConfigSync extends LiveSyncCommands {
async updatePluginListV2(showMessage: boolean, unifiedFilenameWithKey: FilePathWithPrefix): Promise<void> {
const generation = this.pluginListGeneration;
const isCurrent = () => generation === this.pluginListGeneration && this.isThisModuleEnabled();
const isCurrent = () => this.isCurrentPluginList(generation);
try {
this.updatingV2Count++;
pluginV2Progress.set(this.updatingV2Count);
@@ -883,11 +860,9 @@ export class ConfigSync extends LiveSyncCommands {
}
async updatePluginList(showMessage: boolean, updatedDocumentPath?: FilePathWithPrefix): Promise<void> {
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;
@@ -903,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));
@@ -912,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);
@@ -1027,44 +1002,63 @@ export class ConfigSync extends LiveSyncCommands {
await this.storeCustomisationFileV2(path, this.services.setting.getDeviceAndVaultName());
} else {
const files = data.files;
const applyBinaryContent = createModificationHandler<
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 = createModificationHandler<
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());
@@ -1215,6 +1209,7 @@ export class ConfigSync extends LiveSyncCommands {
return true;
}
_everyRealizeSettingSyncMode(): Promise<boolean> {
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();
@@ -1613,49 +1608,35 @@ export class ConfigSync extends LiveSyncCommands {
const entries = this.localDatabase.findEntries(prefix + "", `${prefix}\u{10ffff}`, {
include_docs: true,
});
const tasks = [] as (() => Promise<void>)[];
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<void>)[];
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
@@ -1765,6 +1746,7 @@ export class ConfigSync extends LiveSyncCommands {
this.core.settings.usePluginSync = false;
this.core.settings.autoSweepPlugins = false;
}
this.resetPluginList();
return Promise.resolve(true);
}
+194
View File
@@ -0,0 +1,194 @@
import { compatGlobal, type CompatTimeoutHandle } from "@vrtmrz/livesync-commonlib/compat/common/coreEnvFunctions";
import { createNotifier } from "octagonal-wheels/concurrency/notifier";
import { forEachInScope } from "octagonal-wheels/concurrency/scoped.iteration";
import { createAsyncQueue, type AsyncQueue } from "octagonal-wheels/iterable/asyncQueue";
import { withScope, type Awaitable } from "octagonal-wheels/resource/scope";
interface ConfigScanQueueOptions<T> {
/** The caller logs recoverable item failures and decides whether to continue. */
process: (item: T) => Awaitable<void>;
onProgress?: (remaining: number) => void;
}
interface PendingQueue<T> {
fifo: AsyncQueue<T>;
iterator: AsyncIterator<T>;
}
interface PendingScan<T> {
value: T;
queue: PendingQueue<T>;
}
/** Owns FIFO admission, the initial scan delay, and completion for one catalogue. */
export class ConfigScanQueue<T> {
private pending: PendingQueue<T>;
private pendingCount = 0;
private active = 0;
private ready = false;
private stopped = false;
private stoppingAfterFailure = false;
private readonly failures: unknown[] = [];
private timer?: CompatTimeoutHandle;
private running?: Promise<void>;
private readonly changed = createNotifier();
constructor(private readonly options: ConfigScanQueueOptions<T>) {
this.pending = this.createPendingQueue();
}
private createPendingQueue(): PendingQueue<T> {
const fifo = createAsyncQueue<T>();
return { fifo, iterator: fifo[Symbol.asyncIterator]() };
}
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.fifo.enqueue(value);
this.pendingCount++;
this.reportProgress();
this.schedule();
}
/** Discard unstarted scans; active reads and migrations retain their slots. */
clearPending(): void {
const discarded = this.pending;
this.pending = this.createPendingQueue();
this.pendingCount = 0;
this.ready = false;
this.clearTimer();
discarded.fifo.dispose();
this.reportProgress();
}
/** Stop admission and join active work. A processor must not await its own stop. */
stop(): Promise<void> {
this.stopped = true;
this.clearPending();
return this.waitForIdle();
}
/** Include later arrivals until the queue is idle; this does not close admission. */
async waitForIdle(): Promise<void> {
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<T>): boolean {
if (!this.canDispatch() || this.pending !== entry.queue) return false;
this.pendingCount--;
if (this.pendingCount === 0) this.ready = false;
return true;
}
private async *availableWork() {
while (this.canDispatch()) {
const pending = this.pending;
if (!this.ready) {
if (this.active === 0 && this.pendingCount === 0) return;
await this.changed.wait();
continue;
}
if (pending.fifo.size > 0) {
const entry = await pending.iterator.next();
if (entry.done) {
if (pending !== this.pending) continue;
return;
}
// The callback checks queue identity so clearing a pulled item cancels it.
yield { value: entry.value, queue: pending };
} else if (this.active > 0 || this.pendingCount > 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();
});
}
}
@@ -0,0 +1,400 @@
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { ConfigScanQueue } from "./ConfigScanQueue";
const queues: { stop(): Promise<void> }[] = [];
const releases: (() => void)[] = [];
function gate() {
let resolve!: () => void;
const promise = new Promise<void>((done) => {
resolve = done;
});
releases.push(resolve);
return { promise, resolve };
}
function queue<T>(options: ConstructorParameters<typeof ConfigScanQueue<T>>[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<string>({
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<number>({
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<number>({
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<string>({
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<string>({
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<number>({
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("invalidates prefetched work after clearing and admits replacements alongside active scans", async () => {
const finishFirst = gate();
const finishReplacement = gate();
let scans!: ConfigScanQueue<string>;
const seen: string[] = [];
scans = queue<string>({
process: async (item) => {
seen.push(item);
if (item === "first") {
void Promise.resolve().then(() => Promise.resolve().then(() => scans.clearPending()));
await finishFirst.promise;
} else if (item === "replacement") {
await finishReplacement.promise;
}
},
});
scans.enqueue("first");
scans.enqueue("stale");
await vi.advanceTimersByTimeAsync(100);
expect(seen).toEqual(["first"]);
expect(scans.remaining).toBe(1);
scans.enqueue("replacement");
await vi.advanceTimersByTimeAsync(100);
expect(seen).toEqual(["first", "replacement"]);
expect(scans.remaining).toBe(2);
finishFirst.resolve();
await vi.advanceTimersByTimeAsync(0);
expect(scans.remaining).toBe(1);
finishReplacement.resolve();
await scans.waitForIdle();
expect(scans.remaining).toBe(0);
});
it("keeps the ten-scan limit across clear and replacement work", async () => {
const finish = Array.from({ length: 12 }, () => gate());
const seen: number[] = [];
let active = 0;
let maximum = 0;
const scans = queue<number>({
process: async (item) => {
seen.push(item);
maximum = Math.max(maximum, ++active);
await finish[item].promise;
active--;
},
});
for (let item = 0; item < 10; item++) scans.enqueue(item);
await vi.advanceTimersByTimeAsync(100);
expect(seen).toEqual(Array.from({ length: 10 }, (_, index) => index));
scans.clearPending();
scans.enqueue(10);
scans.enqueue(11);
await vi.advanceTimersByTimeAsync(100);
expect(seen).toEqual(Array.from({ length: 10 }, (_, index) => index));
expect(maximum).toBe(10);
finish[0].resolve();
await vi.advanceTimersByTimeAsync(0);
expect(seen).toEqual(Array.from({ length: 11 }, (_, index) => index));
expect(maximum).toBe(10);
finish.forEach((item) => item.resolve());
await scans.waitForIdle();
expect(seen).toEqual(Array.from({ length: 12 }, (_, index) => index));
expect(maximum).toBe(10);
});
it("stops while waiting for input, joins active work, and rejects later admission", async () => {
const finish = gate();
const seen: number[] = [];
const scans = queue<number>({
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<number>({
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<string>({
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<string>({
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<string>({
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<string>({
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<string>({
process,
onProgress: () => {
throw failure;
},
});
scans.enqueue("pending");
await expect(scans.waitForIdle()).rejects.toBe(failure);
expect(process).not.toHaveBeenCalled();
expect(vi.getTimerCount()).toBe(0);
});
});
@@ -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();
});
});
@@ -90,6 +90,39 @@ afterEach(async () => {
for (const fixture of fixtureOwners.splice(0)) await fixture.dispose();
});
describe("Hidden File Sync initial modification times", () => {
it.each([
{ label: "newer", localMtime: 30_000, expected: "local content" },
{ label: "older", localMtime: 10_000, expected: "database content" },
{ label: "equal", localMtime: 20_000, expected: "database content" },
])("stores only a newer local file when the local timestamp is $label", async ({ localMtime, expected }) => {
const { fixture, hiddenFileSync, storageAccess } = await createHiddenFileSync();
const entry = optionalFileEntry(`i:${path}`, "database content");
entry.mtime = 20_000;
await fixture.localDatabase.putDBEntry(entry);
storageAccess.setFile("local content", localMtime);
await hiddenFileSync.trackStorageFileModification(path, true);
const saved = await fixture.localDatabase.getDBEntry(`i:${path}` as FilePath);
expect(saved).not.toBe(false);
if (saved) {
expect(await getText(saved)).toBe(expected);
expect(saved.mtime).toBe(expected === "local content" ? localMtime : 20_000);
}
});
it("stores a local-only file when only newer changes are requested", async () => {
const { fixture, hiddenFileSync } = await createHiddenFileSync();
await hiddenFileSync.trackStorageFileModification(path, true);
const saved = await fixture.localDatabase.getDBEntry(`i:${path}` as FilePath);
expect(saved).not.toBe(false);
if (saved) expect(await getText(saved)).toBe(content);
});
});
describe("Hidden File Sync optional save outcomes", () => {
it("retries a failed database write when the storage fingerprint is unchanged", async () => {
const { fixture, hiddenFileSync } = await createHiddenFileSync();
@@ -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<boolean>((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();
});
});
@@ -0,0 +1,403 @@
import { afterEach, describe, expect, it, vi } from "vitest";
vi.mock("@/deps.ts", () => ({}));
vi.mock("@/features/HiddenFileCommon/JsonResolveModal.ts", () => ({ JsonResolveModal: class JsonResolveModal {} }));
vi.mock("@/modules/features/ModuleLog.ts", () => ({ MARK_DONE: "" }));
vi.mock("@/common/PeriodicProcessor.ts", () => ({
PeriodicProcessor: class PeriodicProcessor {
disable = vi.fn();
},
}));
vi.mock("./configureHiddenFileSyncMode.ts", () => ({ configureHiddenFileSyncMode: vi.fn() }));
import {
DEFAULT_SETTINGS,
type DocumentID,
type FilePathWithPrefix,
type LoadedEntry,
type MetaEntry,
} from "@vrtmrz/livesync-commonlib/compat/common/types";
import { HiddenFileSync } from "./CmdHiddenFileSync";
const queueOwners: HiddenFileSync[] = [];
function deferred<T = void>() {
let resolve!: (value: T | PromiseLike<T>) => void;
const promise = new Promise<T>((done) => {
resolve = done;
});
return { promise, resolve };
}
async function waitForCondition(predicate: () => boolean, message: string) {
for (let attempt = 0; attempt < 30; attempt++) {
if (predicate()) return;
await Promise.resolve();
}
throw new Error(message);
}
function createHiddenFileSync(localDatabase: object) {
vi.spyOn(HiddenFileSync.prototype, "onBindFunction").mockImplementation(() => {});
const hiddenFileSync = new HiddenFileSync({
settings: {
...DEFAULT_SETTINGS,
syncInternalFiles: true,
syncInternalFileOverwritePatterns: "",
},
localDatabase,
services: { API: { addLog: vi.fn() }, context: { app: {} } },
} as unknown as ConstructorParameters<typeof HiddenFileSync>[0]);
hiddenFileSync._log = vi.fn();
queueOwners.push(hiddenFileSync);
return { hiddenFileSync };
}
afterEach(() => {
for (const hiddenFileSync of queueOwners.splice(0)) {
hiddenFileSync.conflictResolutionProcessor.dispose();
}
vi.restoreAllMocks();
vi.useRealTimers();
});
describe("HiddenFileSync conflict queue", () => {
it("keeps a self-requeued conflict check and waits for its follow-up", async () => {
vi.useFakeTimers();
const path = "i:.obsidian/plugins/example/data.txt" as FilePathWithPrefix;
const id = String(path) as DocumentID;
const conflictDoc = {
_id: id,
_rev: "2-current",
path,
_conflicts: ["2-conflicted"],
} as MetaEntry;
const resolvedDoc = { _id: id, _rev: "3-resolved", path } as MetaEntry;
let rawRead = 0;
const localDatabase = {
findEntries: vi.fn(async function* () {
yield conflictDoc;
}),
getRaw: vi.fn(async () => (rawRead++ === 0 ? conflictDoc : resolvedDoc)),
};
const { hiddenFileSync } = createHiddenFileSync(localDatabase);
hiddenFileSync.path2id = async () => id;
let resolutionCalls = 0;
let requeued = false;
hiddenFileSync.resolveByNewerEntry = vi.fn(async () => {
resolutionCalls++;
hiddenFileSync.requeueConflictCheck(path);
requeued = true;
});
let completed = false;
hiddenFileSync.queueConflictCheck(path);
try {
await waitForCondition(
() => hiddenFileSync.conflictResolutionProcessor.remaining === 1,
"The initial conflict check was not queued"
);
await vi.advanceTimersByTimeAsync(9);
expect(localDatabase.getRaw).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(1);
await waitForCondition(
() =>
requeued &&
hiddenFileSync.conflictResolutionProcessor.nowProcessing === 0 &&
hiddenFileSync.conflictResolutionProcessor.remaining === 1,
"The conflict follow-up was not retained after the first pass"
);
void hiddenFileSync.conflictResolutionProcessor.waitForAllProcessed().then(() => {
completed = true;
});
expect(completed).toBe(false);
expect(localDatabase.getRaw).toHaveBeenCalledTimes(1);
await vi.advanceTimersByTimeAsync(9);
expect(localDatabase.getRaw).toHaveBeenCalledTimes(1);
expect(completed).toBe(false);
await vi.advanceTimersByTimeAsync(1);
await vi.advanceTimersByTimeAsync(1_000);
await waitForCondition(() => completed, "The conflict pipeline did not await its follow-up");
expect(localDatabase.getRaw).toHaveBeenCalledTimes(2);
expect(resolutionCalls).toBe(1);
expect(hiddenFileSync.pendingConflictChecks.has(path)).toBe(false);
expect(hiddenFileSync.conflictResolutionProcessor.remaining).toBe(0);
} finally {
await vi.runAllTimersAsync();
}
});
it("waits for a cancelled manual merge dialogue and then finishes without requeueing", async () => {
vi.useFakeTimers();
const path = "i:.obsidian/plugins/example/data.json" as FilePathWithPrefix;
const id = String(path) as DocumentID;
const conflictDoc = {
_id: id,
_rev: "2-current",
path,
_conflicts: ["2-conflicted"],
} as MetaEntry;
const historyDoc = {} as unknown as MetaEntry;
const loaded = { _id: id, _rev: "2-current", path } as LoadedEntry;
const dialogue = deferred<boolean>();
let dialogueOpened = false;
const localDatabase = {
findEntries: vi.fn(async function* () {
yield conflictDoc;
}),
getRaw: vi.fn(async (_id: DocumentID, options?: { revs_info?: boolean }) =>
options?.revs_info ? historyDoc : conflictDoc
),
getDBEntry: vi.fn(async () => loaded),
managers: {
conflictManager: {
mergeObject: vi.fn(async () => false),
},
},
};
const { hiddenFileSync } = createHiddenFileSync(localDatabase);
hiddenFileSync.path2id = async () => id;
hiddenFileSync.showJSONMergeDialogAndMerge = vi.fn(async () => {
dialogueOpened = true;
return await dialogue.promise;
});
let completed = false;
hiddenFileSync.queueConflictCheck(path);
try {
await waitForCondition(
() => hiddenFileSync.conflictResolutionProcessor.remaining === 1,
"The initial conflict check was not queued"
);
await vi.advanceTimersByTimeAsync(100);
expect(dialogueOpened).toBe(true);
void hiddenFileSync.conflictResolutionProcessor.waitForAllProcessed().then(() => {
completed = true;
});
expect(completed).toBe(false);
expect(hiddenFileSync.conflictResolutionProcessor.totalRemaining).toBeGreaterThan(0);
expect(localDatabase.managers.conflictManager.mergeObject).toHaveBeenCalledOnce();
expect(localDatabase.getDBEntry).toHaveBeenCalledTimes(2);
dialogue.resolve(false);
await vi.advanceTimersByTimeAsync(1_000);
await waitForCondition(() => completed, "The cancelled dialogue did not release the pipeline");
expect(hiddenFileSync.pendingConflictChecks.has(path)).toBe(false);
expect(hiddenFileSync.conflictResolutionProcessor.totalRemaining).toBe(0);
} finally {
dialogue.resolve(false);
await vi.runAllTimersAsync();
}
});
});
describe("HiddenFileSync pipeline contracts", () => {
it("starts above the threshold, limits analysis to five, and retains FIFO input", async () => {
vi.useFakeTimers();
const gates = Array.from({ length: 11 }, () => deferred<MetaEntry>());
const paths = gates.map((_, index) => `i:.obsidian/${index}.txt` as FilePathWithPrefix);
const reads: string[] = [];
const getRaw = vi.fn((id: DocumentID) => {
reads.push(id);
return gates[paths.indexOf(id as unknown as FilePathWithPrefix)].promise;
});
const { hiddenFileSync } = createHiddenFileSync({ getRaw });
hiddenFileSync.path2id = async (path) => path as unknown as DocumentID;
for (const path of paths.slice(0, 10)) hiddenFileSync.queueConflictCheck(path);
await vi.advanceTimersByTimeAsync(0);
expect(reads).toEqual([]);
hiddenFileSync.queueConflictCheck(paths[10]);
await vi.advanceTimersByTimeAsync(0);
expect(reads).toEqual(paths.slice(0, 5));
hiddenFileSync.queueConflictCheck(paths[0]);
expect(hiddenFileSync.pendingConflictChecks.size).toBe(11);
gates[0].resolve({ path: paths[0] } as MetaEntry);
await vi.advanceTimersByTimeAsync(0);
expect(reads).toEqual(paths.slice(0, 6));
gates.forEach((gate, index) => gate.resolve({ path: paths[index] } as MetaEntry));
await vi.advanceTimersByTimeAsync(100);
expect(await hiddenFileSync.conflictResolutionProcessor.waitForAllProcessed()).toBe(true);
expect(reads).toEqual(paths);
expect(hiddenFileSync.pendingConflictChecks.size).toBe(0);
});
it("continues analysis while one manual dialogue is open and serialises the dialogues", async () => {
vi.useFakeTimers();
const paths = Array.from({ length: 12 }, (_, index) => `i:.obsidian/${index}.json` as FilePathWithPrefix);
const gates = paths.map(() => deferred<boolean>());
const localDatabase = {
getRaw: vi.fn(async (id: DocumentID, options?: { revs_info?: boolean }) =>
options?.revs_info
? {}
: {
_id: id,
path: id,
_rev: "2-current",
_conflicts: ["2-other"],
}
),
getDBEntry: vi.fn(async (path: FilePathWithPrefix) => ({ path }) as LoadedEntry),
managers: { conflictManager: { mergeObject: vi.fn(async () => false) } },
};
const { hiddenFileSync } = createHiddenFileSync(localDatabase);
hiddenFileSync.path2id = async (path) => path as unknown as DocumentID;
const opened: FilePathWithPrefix[] = [];
hiddenFileSync.showJSONMergeDialogAndMerge = vi.fn((doc) => {
opened.push(doc.path);
return gates[paths.indexOf(doc.path)].promise;
});
paths.forEach((path) => hiddenFileSync.queueConflictCheck(path));
await vi.advanceTimersByTimeAsync(100);
expect(localDatabase.managers.conflictManager.mergeObject).toHaveBeenCalledTimes(12);
expect(opened).toEqual([paths[0]]);
paths.forEach((path) => hiddenFileSync.queueConflictCheck(path));
expect(hiddenFileSync.pendingConflictChecks.size).toBe(12);
let completed = false;
void hiddenFileSync.conflictResolutionProcessor.waitForAllProcessed().then(() => {
completed = true;
});
for (let index = 0; index < paths.length; index++) {
expect(opened).toEqual(paths.slice(0, index + 1));
expect(completed).toBe(false);
gates[index].resolve(false);
await vi.advanceTimersByTimeAsync(100);
}
expect(completed).toBe(true);
expect(hiddenFileSync.pendingConflictChecks.size).toBe(0);
});
it("logs an analysis error and continues with later paths", async () => {
vi.useFakeTimers();
const failure = new Error("Injected read failure");
const getRaw = vi.fn().mockRejectedValueOnce(failure).mockResolvedValue({});
const { hiddenFileSync } = createHiddenFileSync({ getRaw });
hiddenFileSync.path2id = async (path) => path as unknown as DocumentID;
hiddenFileSync.queueConflictCheck("i:bad.txt" as FilePathWithPrefix);
hiddenFileSync.queueConflictCheck("i:good.txt" as FilePathWithPrefix);
await vi.advanceTimersByTimeAsync(100);
expect(await hiddenFileSync.conflictResolutionProcessor.waitForAllProcessed()).toBe(true);
expect(getRaw).toHaveBeenCalledTimes(2);
expect(hiddenFileSync._log).toHaveBeenCalledWith(failure, expect.anything());
expect(hiddenFileSync.pendingConflictChecks.size).toBe(0);
});
it("collects a conflict enumeration before resuming analysis", async () => {
vi.useFakeTimers();
const gate = deferred();
const path = "i:.obsidian/scanned.txt" as FilePathWithPrefix;
const getRaw = vi.fn(async () => ({}));
const { hiddenFileSync } = createHiddenFileSync({
getRaw,
findEntries: async function* () {
await gate.promise;
yield { _id: path, path, _conflicts: ["2-other"] };
},
});
hiddenFileSync.path2id = async (value) => value as unknown as DocumentID;
const result = hiddenFileSync.resolveConflictOnInternalFiles();
await vi.advanceTimersByTimeAsync(100);
hiddenFileSync.queueConflictCheck(path);
await vi.advanceTimersByTimeAsync(100);
expect(getRaw).not.toHaveBeenCalled();
gate.resolve();
await vi.advanceTimersByTimeAsync(100);
await result;
expect(getRaw).toHaveBeenCalledOnce();
expect(hiddenFileSync.pendingConflictChecks.size).toBe(0);
});
});
describe("HiddenFileSync conflict queue ownership", () => {
it("stops delayed work on unload and rejects later conflict events", async () => {
vi.useFakeTimers();
const getRaw = vi.fn(async () => ({}));
const { hiddenFileSync } = createHiddenFileSync({ getRaw });
hiddenFileSync.path2id = async (path) => path as unknown as DocumentID;
const path = "i:.obsidian/queued.txt" as FilePathWithPrefix;
hiddenFileSync.queueConflictCheck(path);
const idle = hiddenFileSync.conflictResolutionProcessor.waitForAllProcessed();
hiddenFileSync.onunload();
hiddenFileSync.queueConflictCheck(path);
await vi.advanceTimersByTimeAsync(100);
expect(await idle).toBe(false);
expect(getRaw).not.toHaveBeenCalled();
expect(hiddenFileSync.pendingConflictChecks.size).toBe(0);
expect(vi.getTimerCount()).toBe(0);
});
it("finishes an admitted analysis without handing its result to a new dialogue after unload", async () => {
vi.useFakeTimers();
const gate = deferred<MetaEntry>();
const path = "i:.obsidian/active.json" as FilePathWithPrefix;
const getRaw = vi
.fn()
.mockImplementationOnce(() => gate.promise)
.mockResolvedValue({});
const getDBEntry = vi.fn(async () => ({ path }));
const mergeObject = vi.fn(async () => false);
const { hiddenFileSync } = createHiddenFileSync({
getRaw,
getDBEntry,
managers: { conflictManager: { mergeObject } },
});
hiddenFileSync.path2id = async () => path as unknown as DocumentID;
hiddenFileSync.queueConflictCheck(path);
await vi.advanceTimersByTimeAsync(10);
expect(getRaw).toHaveBeenCalledOnce();
hiddenFileSync.onunload();
expect(await hiddenFileSync.conflictResolutionProcessor.waitForAllProcessed()).toBe(false);
gate.resolve({
_id: String(path) as DocumentID,
path,
_rev: "2-current",
_conflicts: ["2-other"],
} as MetaEntry);
await vi.advanceTimersByTimeAsync(100);
expect(mergeObject).toHaveBeenCalledOnce();
expect(getDBEntry).not.toHaveBeenCalled();
expect(hiddenFileSync.pendingConflictChecks.size).toBe(0);
});
it("does not requeue a manual resolution which finishes after unload", async () => {
vi.useFakeTimers();
const gate = deferred<boolean>();
const path = "i:.obsidian/dialogue.json" as FilePathWithPrefix;
const getRaw = vi.fn(async (_id: DocumentID, options?: { revs_info?: boolean }) =>
options?.revs_info
? {}
: {
_id: path,
path,
_rev: "2-current",
_conflicts: ["2-other"],
}
);
const getDBEntry = vi.fn(async () => ({ path }) as LoadedEntry);
const { hiddenFileSync } = createHiddenFileSync({
getRaw,
getDBEntry,
managers: { conflictManager: { mergeObject: async () => false } },
});
hiddenFileSync.path2id = async () => path as unknown as DocumentID;
hiddenFileSync.showJSONMergeDialogAndMerge = vi.fn(() => gate.promise);
hiddenFileSync.queueConflictCheck(path);
await vi.advanceTimersByTimeAsync(20);
expect(hiddenFileSync.showJSONMergeDialogAndMerge).toHaveBeenCalledOnce();
hiddenFileSync.onunload();
expect(await hiddenFileSync.conflictResolutionProcessor.waitForAllProcessed()).toBe(false);
gate.resolve(true);
await vi.advanceTimersByTimeAsync(100);
expect(getRaw).toHaveBeenCalledTimes(2);
expect(getDBEntry).toHaveBeenCalledTimes(2);
expect(hiddenFileSync.pendingConflictChecks.size).toBe(0);
expect(hiddenFileSync.conflictResolutionProcessor.totalRemaining).toBe(0);
});
});
@@ -35,7 +35,6 @@ import {
getLogLevel,
autosaveCache,
type MapLike,
onlyInNTimes,
BASE_IS_NEW,
EVEN,
displayRev,
@@ -45,7 +44,8 @@ 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 { QueueProcessor } from "octagonal-wheels/concurrency/processor";
import { createCoalescingNotifier } from "octagonal-wheels/concurrency/coalescingNotifier";
import { HiddenFileConflictQueue } from "./HiddenFileConflictQueue.ts";
import {
hiddenFilesEventCount,
hiddenFilesProcessingCount,
@@ -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";
@@ -107,6 +108,8 @@ export class HiddenFileSync extends LiveSyncCommands {
}
onunload() {
this.periodicInternalFileScanProcessor?.disable();
this.conflictResolutionProcessor.dispose();
this.pendingConflictChecks.clear();
}
onload() {
this.services.API.addCommand({
@@ -571,18 +574,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,
@@ -688,7 +688,7 @@ Offline Changed files: ${processFiles.length}`;
const dbMTime = getComparingMTime(filesOnDB, includeDeleted);
const diff = compareMTime(storageMTime, dbMTime);
if (diff != TARGET_IS_NEW) {
if (diff != BASE_IS_NEW) {
this._log(`Hidden file:${path} is not new.`, LOG_LEVEL_VERBOSE);
if (filesOnDB && stat) {
// OnlyNew not handles the deletion.
@@ -747,7 +747,7 @@ Offline Changed files: ${processFiles.length}`;
queueConflictCheck(path: FilePathWithPrefix) {
if (this.pendingConflictChecks.has(path)) return;
this.pendingConflictChecks.add(path);
this.conflictResolutionProcessor.enqueue(path);
if (!this.conflictResolutionProcessor.enqueue(path)) this.pendingConflictChecks.delete(path);
}
finishConflictCheck(path: FilePathWithPrefix) {
@@ -804,9 +804,8 @@ Offline Changed files: ${processFiles.length}`;
}
// check the file again
}
conflictResolutionProcessor = new QueueProcessor(
async (paths: FilePathWithPrefix[]) => {
const path = paths[0];
conflictResolutionProcessor = new HiddenFileConflictQueue(
async (path: FilePathWithPrefix) => {
try {
// Retrieve data
const id = await this.path2id(path, ICHeader);
@@ -874,46 +873,32 @@ Offline Changed files: ${processFiles.length}`;
return [];
}
},
{
suspended: false,
batchSize: 1,
concurrentLimit: 5,
delay: 10,
keepResultUntilDownstreamConnected: true,
yieldThreshold: 10,
pipeTo: new QueueProcessor(
async (results) => {
const { id, doc, path, revA, revB } = results[0];
const prefixedPath = addPrefix(path, ICHeader);
const docAMerge = await this.localDatabase.getDBEntry(prefixedPath, { rev: revA });
const docBMerge = await this.localDatabase.getDBEntry(prefixedPath, { rev: revB });
try {
if (docAMerge != false && docBMerge != false) {
if (await this.showJSONMergeDialogAndMerge(docAMerge, docBMerge)) {
// Again for other conflicted revisions.
this.requeueConflictCheck(path);
} else {
this.finishConflictCheck(path);
}
return;
} else {
// If either revision could not read, force resolving by the newer one.
await this.resolveByNewerEntry(id, path, doc, revA, revB);
}
} catch (ex) {
async (result) => {
const { id, doc, path, revA, revB } = result;
const prefixedPath = addPrefix(path, ICHeader);
const docAMerge = await this.localDatabase.getDBEntry(prefixedPath, { rev: revA });
const docBMerge = await this.localDatabase.getDBEntry(prefixedPath, { rev: revB });
try {
if (docAMerge != false && docBMerge != false) {
if (await this.showJSONMergeDialogAndMerge(docAMerge, docBMerge)) {
// Again for other conflicted revisions.
this.requeueConflictCheck(path);
} else {
this.finishConflictCheck(path);
throw ex;
}
},
{
suspended: false,
batchSize: 1,
concurrentLimit: 1,
delay: 10,
keepResultUntilDownstreamConnected: false,
yieldThreshold: 10,
return;
} else {
// If either revision could not read, force resolving by the newer one.
await this.resolveByNewerEntry(id, path, doc, revA, revB);
}
),
} catch (ex) {
this.finishConflictCheck(path);
throw ex;
}
},
(error) => {
this._log("Processor error!");
this._log(error, LOG_LEVEL_VERBOSE);
}
);
@@ -1134,9 +1119,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 +1137,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 +1172,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<number>({
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 +1249,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 +1422,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<number>({
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)!;
@@ -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<typeof stat | null> => 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",
@@ -0,0 +1,209 @@
import { compatGlobal, type CompatTimeoutHandle } from "@vrtmrz/livesync-commonlib/compat/common/coreEnvFunctions";
import { withCancellation } from "octagonal-wheels/concurrency/cancellation";
import { createNotifier } from "octagonal-wheels/concurrency/notifier";
import { forEachInScope } from "octagonal-wheels/concurrency/scoped.iteration";
import { createAsyncQueue } from "octagonal-wheels/iterable/asyncQueue";
import { withOwnedResource } from "octagonal-wheels/resource";
/** The scheduling used by one hidden-file conflict stage. */
class ConflictStage<T> {
private readonly input = createAsyncQueue<T>();
private readonly reader = this.input[Symbol.asyncIterator]();
private readonly changed = createNotifier();
private readonly cancellation = new AbortController();
private timer: CompatTimeoutHandle | undefined;
private ready = false;
private suspended = false;
private stopped = false;
private outstanding = 0;
active = 0;
constructor(
concurrency: number,
process: (item: T) => Promise<unknown>,
private readonly notify: () => void,
onError: (error: unknown) => void,
onFailure: (error: unknown) => void
) {
// The operation owns its input until started callbacks have settled.
// Abort also releases input immediately, before that final cleanup.
void withCancellation(undefined, { signal: this.cancellation.signal }, (scope) => {
const stopInput = () => {
this.input.dispose();
this.clearTimer();
this.changed.notify();
};
scope.signal.addEventListener("abort", stopInput, { once: true });
scope.defer(() => scope.signal.removeEventListener("abort", stopInput));
return withOwnedResource(this.input, () =>
forEachInScope(scope, this.availableItems(), { concurrency }, async (entry) => {
while (this.suspended && !this.stopped) await this.changed.wait();
if (this.stopped) return;
this.active++;
try {
await process(entry.value);
} catch (error) {
try {
onError(error);
} catch (reportingError) {
// Wake a pending input pull before the runner joins
// its other callbacks after an unexpected failure.
onFailure(reportingError);
throw reportingError;
}
} finally {
this.active--;
this.outstanding--;
this.notify();
}
})
);
}).catch(onFailure);
}
get remaining() {
return this.outstanding;
}
enqueue(item: T): boolean {
if (this.stopped) return false;
this.input.enqueue(item);
this.outstanding++;
this.schedule();
this.notify();
return true;
}
private clearTimer() {
if (this.timer === undefined) return;
compatGlobal.clearTimeout(this.timer);
this.timer = undefined;
}
private schedule() {
if (this.stopped || this.suspended || this.input.size === 0) return;
if (this.ready || this.input.size > 10) {
this.clearTimer();
this.ready = true;
this.changed.notify();
} else if (this.timer === undefined) {
this.timer = compatGlobal.setTimeout(() => {
this.timer = undefined;
this.ready = true;
this.changed.notify();
}, 10);
}
}
private async *availableItems() {
while (!this.stopped) {
while (!this.stopped && (this.suspended || !this.ready || this.input.size === 0)) {
await this.changed.wait();
}
if (this.stopped) return;
const entry = await this.reader.next();
if (entry.done) return;
if (this.input.size === 0) {
this.ready = false;
this.clearTimer();
}
// Outstanding includes this pulled item until its callback finishes.
yield { value: entry.value };
}
}
suspend() {
this.suspended = true;
this.ready = false;
this.clearTimer();
}
resume() {
if (!this.suspended || this.stopped) return;
this.suspended = false;
this.schedule();
this.changed.notify();
}
dispose() {
if (this.stopped) return;
this.stopped = true;
this.outstanding = this.active;
this.cancellation.abort();
this.notify();
}
}
/** Two hidden-file stages: five analyses, followed by one manual resolution. */
export class HiddenFileConflictQueue<T, U> {
private readonly changed = createNotifier();
private readonly analysis: ConflictStage<T>;
private readonly resolution: ConflictStage<U>;
private stopped = false;
private failure: { error: unknown } | undefined;
constructor(
analyse: (item: T) => Promise<U[]>,
resolve: (item: U) => Promise<unknown>,
onError: (error: unknown) => void
) {
const notify = () => this.changed.notify();
const onFailure = (error: unknown) => {
this.failure ??= { error };
this.dispose();
};
this.resolution = new ConflictStage(1, resolve, notify, onError, onFailure);
this.analysis = new ConflictStage(
5,
async (item) => {
const results = await analyse(item);
for (const result of results) this.resolution.enqueue(result);
},
notify,
onError,
onFailure
);
}
enqueue(item: T): boolean {
return this.analysis.enqueue(item);
}
get nowProcessing() {
return this.analysis.active;
}
get remaining() {
return this.analysis.remaining;
}
get totalRemaining() {
return this.analysis.remaining + this.resolution.remaining;
}
suspend() {
this.analysis.suspend();
}
startPipeline(): this {
this.analysis.resume();
return this;
}
/** Includes handoff and self-requeues; disposal is never successful drainage. */
async waitForAllProcessed(): Promise<boolean> {
this.startPipeline();
while (!this.stopped && this.totalRemaining > 0) await this.changed.wait();
if (this.failure) throw this.failure.error;
return !this.stopped;
}
/** Stop admission and release pending work without waiting for active dialogue or writes. */
dispose() {
if (this.stopped) return;
this.stopped = true;
this.analysis.dispose();
this.resolution.dispose();
this.changed.notify();
}
}
@@ -0,0 +1,194 @@
import { afterEach, describe, expect, it, vi } from "vitest";
import { HiddenFileConflictQueue } from "./HiddenFileConflictQueue.ts";
const owners: { dispose(): void }[] = [];
function deferred<T = void>() {
let resolve!: (value: T | PromiseLike<T>) => void;
const promise = new Promise<T>((done) => {
resolve = done;
});
return { promise, resolve };
}
function own<T, U>(queue: HiddenFileConflictQueue<T, U>) {
owners.push(queue);
return queue;
}
afterEach(async () => {
owners.splice(0).forEach((owner) => owner.dispose());
if (vi.isFakeTimers()) await vi.runAllTimersAsync();
vi.useRealTimers();
});
describe("HiddenFileConflictQueue", () => {
it("awaits repeated handoffs and self-requeues with native timers", async () => {
const seen: number[] = [];
let queue!: HiddenFileConflictQueue<number, number>;
queue = own(
new HiddenFileConflictQueue(
async (value: number) => [value],
async (value) => {
seen.push(value);
if (value < 3) queue.enqueue(value + 1);
},
vi.fn()
)
);
queue.enqueue(0);
expect(await queue.waitForAllProcessed()).toBe(true);
expect(seen).toEqual([0, 1, 2, 3]);
expect(queue.totalRemaining).toBe(0);
});
it("uses the first arrival's ten-millisecond delay in each stage", async () => {
vi.useFakeTimers();
const analyse = vi.fn(async (value: number) => [value]);
const resolve = vi.fn(async (_value: number) => {});
const queue = own(new HiddenFileConflictQueue(analyse, resolve, vi.fn()));
queue.enqueue(1);
await vi.advanceTimersByTimeAsync(9);
queue.enqueue(2);
expect(analyse).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(1);
expect(analyse.mock.calls).toEqual([[1], [2]]);
expect(resolve).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(9);
expect(resolve).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(1);
expect(resolve.mock.calls).toEqual([[1], [2]]);
expect(await queue.waitForAllProcessed()).toBe(true);
expect(queue.totalRemaining).toBe(0);
expect(vi.getTimerCount()).toBe(0);
});
it("starts both stages immediately above ten waiting items", async () => {
vi.useFakeTimers();
const gate = deferred();
const analyse = vi.fn(async (value: number) => [value]);
const resolve = vi.fn(async (_value: number) => {
await gate.promise;
});
const queue = own(new HiddenFileConflictQueue(analyse, resolve, vi.fn()));
for (let value = 0; value < 11; value++) queue.enqueue(value);
await vi.advanceTimersByTimeAsync(0);
expect(analyse).toHaveBeenCalledTimes(11);
expect(resolve).toHaveBeenCalledTimes(1);
expect(queue.totalRemaining).toBe(11);
gate.resolve();
await vi.advanceTimersByTimeAsync(0);
expect(resolve.mock.calls).toEqual(Array.from({ length: 11 }, (_, value) => [value]));
expect(await queue.waitForAllProcessed()).toBe(true);
});
it("logs item failures in both stages and continues without implicit retry", async () => {
vi.useFakeTimers();
const first = new Error("analysis failed");
const second = new Error("resolution failed");
const analyse = vi.fn(async (value: number) => {
if (value === 1) throw first;
return [value];
});
const resolve = vi.fn(async (value: number) => {
if (value === 2) throw second;
});
const onError = vi.fn();
const queue = own(new HiddenFileConflictQueue(analyse, resolve, onError));
[1, 2, 3].forEach((value) => queue.enqueue(value));
await vi.advanceTimersByTimeAsync(20);
expect(onError.mock.calls).toEqual([[first], [second]]);
expect(analyse.mock.calls).toEqual([[1], [2], [3]]);
expect(resolve.mock.calls).toEqual([[2], [3]]);
expect(await queue.waitForAllProcessed()).toBe(true);
});
it("discards delayed input and releases idle waiters on disposal", async () => {
vi.useFakeTimers();
const analyse = vi.fn(async (value: number) => [value]);
const resolve = vi.fn(async (_value: number) => {});
const queue = own(new HiddenFileConflictQueue(analyse, resolve, vi.fn()));
queue.enqueue(1);
const idle = queue.waitForAllProcessed();
queue.dispose();
queue.dispose();
expect(await idle).toBe(false);
expect(queue.enqueue(2)).toBe(false);
expect(queue.totalRemaining).toBe(0);
await vi.advanceTimersByTimeAsync(100);
expect(analyse).not.toHaveBeenCalled();
expect(resolve).not.toHaveBeenCalled();
expect(vi.getTimerCount()).toBe(0);
});
it("does not wait for active analysis or admit its late handoff and self-requeue after disposal", async () => {
vi.useFakeTimers();
const gate = deferred();
const attempted: boolean[] = [];
let finished = 0;
const analyse = vi.fn(async (value: number) => {
await gate.promise;
attempted.push(queue.enqueue(value));
finished++;
return [value];
});
const resolve = vi.fn(async (_value: number) => {});
const queue = own(new HiddenFileConflictQueue(analyse, resolve, vi.fn()));
for (let value = 0; value < 12; value++) queue.enqueue(value);
await vi.advanceTimersByTimeAsync(0);
expect(analyse).toHaveBeenCalledTimes(5);
const idle = queue.waitForAllProcessed();
queue.dispose();
expect(await idle).toBe(false);
expect(finished).toBe(0);
expect(queue.totalRemaining).toBe(5);
gate.resolve();
await vi.advanceTimersByTimeAsync(100);
expect(finished).toBe(5);
expect(attempted).toEqual([false, false, false, false, false]);
expect(analyse).toHaveBeenCalledTimes(5);
expect(resolve).not.toHaveBeenCalled();
expect(queue.totalRemaining).toBe(0);
expect(vi.getTimerCount()).toBe(0);
});
it("leaves an active dialogue running and discards the next dialogue", async () => {
vi.useFakeTimers();
const gate = deferred();
let finished = false;
const resolve = vi.fn(async (_value: number) => {
await gate.promise;
finished = true;
});
const queue = own(new HiddenFileConflictQueue(async (value: number) => [value], resolve, vi.fn()));
queue.enqueue(1);
queue.enqueue(2);
await vi.advanceTimersByTimeAsync(20);
expect(resolve).toHaveBeenCalledTimes(1);
queue.dispose();
expect(await queue.waitForAllProcessed()).toBe(false);
expect(finished).toBe(false);
gate.resolve();
await vi.advanceTimersByTimeAsync(100);
expect(finished).toBe(true);
expect(resolve).toHaveBeenCalledTimes(1);
expect(queue.totalRemaining).toBe(0);
});
it("suspends analysis while already handed-off manual work continues", async () => {
vi.useFakeTimers();
const analyse = vi.fn(async (value: number) => [value]);
const resolve = vi.fn(async (_value: number) => {});
const queue = own(new HiddenFileConflictQueue(analyse, resolve, vi.fn()));
queue.enqueue(1);
await vi.advanceTimersByTimeAsync(10);
queue.suspend();
queue.enqueue(2);
await vi.advanceTimersByTimeAsync(100);
expect(analyse.mock.calls).toEqual([[1]]);
expect(resolve.mock.calls).toEqual([[1]]);
queue.startPipeline();
await vi.advanceTimersByTimeAsync(20);
expect(analyse.mock.calls).toEqual([[1], [2]]);
expect(resolve.mock.calls).toEqual([[1], [2]]);
expect(await queue.waitForAllProcessed()).toBe(true);
});
});
@@ -0,0 +1,49 @@
import { createCoalescingNotifier } from "octagonal-wheels/concurrency/coalescingNotifier";
interface PerformBatchWithProgressCallbacks<T> {
/** Runs one item. Its resolved value is ignored. */
processItem: (item: T) => Promise<unknown>;
/** 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<T>(
items: readonly T[],
eachCount: number,
callbacks: PerformBatchWithProgressCallbacks<T>
): Promise<void> {
const notifier = createCoalescingNotifier<number>({
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();
}
@@ -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<void>((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();
});
});
+1 -1
View File
@@ -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<void>;
abstract onload(): void | Promise<void>;
_isMainReady() {
@@ -1,4 +1,5 @@
import type { ReviewHarnessScenarioResult } from "./reviewHarnessTypes";
import { createReviewHarnessHeapSampler } from "./reviewHarnessHeapSampling";
export const REVIEW_HARNESS_FILE_CONTENT_HASH_INPUTS = [
{ label: "4 KiB", bytes: 4 * 1024 },
@@ -23,15 +24,6 @@ type BenchmarkPerformance = Pick<Performance, "now"> & {
type HashAlgorithm = "SHA-256" | "xxHash64";
function readHeap(clock: BenchmarkPerformance): number | undefined {
try {
const bytes = clock.memory?.usedJSHeapSize;
return typeof bytes === "number" && Number.isFinite(bytes) && bytes >= 0 ? bytes : undefined;
} catch {
return undefined;
}
}
function summary(samples: readonly number[]): string {
const sorted = [...samples].sort((a, b) => a - b);
const middle = sorted[Math.floor(sorted.length / 2)];
@@ -63,19 +55,13 @@ export async function runReviewHarnessFileContentHashBenchmark(
throw new Error("The xxHash64 initialisation time is invalid.");
}
const before = readHeap(clock);
let highest = before;
const sampleHeap = () => {
const value = readHeap(clock);
if (value !== undefined) highest = Math.max(highest ?? value, value);
return value;
};
const heap = createReviewHarnessHeapSampler(clock);
const inputs: { readonly label: string; readonly bytes: number; readonly content: Blob }[] = [];
for (const { label, bytes } of REVIEW_HARNESS_FILE_CONTENT_HASH_INPUTS) {
await yieldControl();
inputs.push({ label, bytes, content: operations.createInput(bytes) });
sampleHeap();
heap.sample();
}
const hash = (algorithm: HashAlgorithm, content: Blob) =>
@@ -99,7 +85,7 @@ export async function runReviewHarnessFileContentHashBenchmark(
const elapsed = clock.now() - started;
checkHash(algorithm, inputs[0].label, result);
warmupTimes.set(algorithm, elapsed);
sampleHeap();
heap.sample();
}
const observations = [
@@ -124,7 +110,7 @@ export async function runReviewHarnessFileContentHashBenchmark(
throw new Error(`The ${algorithm} measurement returned an invalid duration.`);
}
samples[algorithm].push(elapsed);
sampleHeap();
heap.sample();
}
}
for (const algorithm of ["SHA-256", "xxHash64"] as const) {
@@ -132,16 +118,7 @@ export async function runReviewHarnessFileContentHashBenchmark(
}
}
const after = sampleHeap();
if (highest === undefined) {
observations.push("JavaScript heap: unavailable on this device.");
} else {
const mib = (bytes: number | undefined) =>
bytes === undefined ? "unavailable" : `${(bytes / 1048576).toFixed(2)} MiB`;
observations.push(
`JavaScript heap: before=${mib(before)}; highest sampled=${mib(highest)}; after=${mib(after)}.`
);
}
observations.push(heap.finish());
observations.push(
"Heap samples are approximate, may include other Obsidian work, and are affected by garbage collection. They are neither total app RAM nor a true peak."
);
@@ -74,6 +74,7 @@ describe("Review Harness file content hash measurements", () => {
expect(result.status).toBe("passed");
expect(result.detail).toBe("File content hash measurements completed.");
expect(report).toContain("JavaScript heap: unavailable on this device.");
expect([...f.inputs().contentBySize.keys()]).toEqual(
REVIEW_HARNESS_FILE_CONTENT_HASH_INPUTS.map(({ bytes }) => bytes)
);
@@ -107,6 +108,49 @@ describe("Review Harness file content hash measurements", () => {
expect(report).not.toContain("4096n");
});
it("tracks the sampled maximum and reports a lower final heap sample", async () => {
const f = fixture();
let reads = 0;
const clock = {
now: f.now,
get memory() {
const sample = reads++;
const mebibytes =
sample === 0 ? 2 : sample === 1 ? 5 : sample === 31 ? 1 : 3;
return { usedJSHeapSize: mebibytes * 1048576 };
},
};
const result = await runReviewHarnessFileContentHashBenchmark(f.operations, clock, f.yieldControl);
expect(result.observations).toContain(
"JavaScript heap: before=2.00 MiB; highest sampled=5.00 MiB; after=1.00 MiB."
);
expect(result.observations.join("\n")).toContain("neither total app RAM nor a true peak");
expect(reads).toBe(32);
});
it.each([Number.NaN, Number.POSITIVE_INFINITY, -1, "throws"] as const)(
"keeps timings usable when the heap API returns %s",
async (value) => {
const f = fixture();
const result = await runReviewHarnessFileContentHashBenchmark(
f.operations,
{
now: f.now,
get memory() {
if (value === "throws") throw new Error("Heap API unavailable");
return { usedJSHeapSize: value as number };
},
},
f.yieldControl
);
expect(result.status).toBe("passed");
expect(result.observations).toContain("JavaScript heap: unavailable on this device.");
expect(result.observations.join("\n")).not.toMatch(/NaN|Infinity/u);
}
);
it("uses the same deterministic byte sequence for short and long inputs", async () => {
const shortInput = createFixedFileContentHashInput(4 * 1024);
const longInput = createFixedFileContentHashInput(64 * 1024);
@@ -0,0 +1,34 @@
type HeapPerformance = {
readonly memory?: { readonly usedJSHeapSize: number };
};
function readHeap(clock: HeapPerformance): number | undefined {
try {
const bytes = clock.memory?.usedJSHeapSize;
return typeof bytes === "number" && Number.isFinite(bytes) && bytes >= 0 ? bytes : undefined;
} catch {
return undefined;
}
}
export function createReviewHarnessHeapSampler(clock: HeapPerformance): {
sample(): void;
finish(): string;
} {
const before = readHeap(clock);
let highest = before;
let after = before;
const sample = () => {
after = readHeap(clock);
if (after !== undefined) highest = Math.max(highest ?? after, after);
};
const finish = () => {
sample();
if (highest === undefined) return "JavaScript heap: unavailable on this device.";
const mib = (bytes: number | undefined) =>
bytes === undefined ? "unavailable" : `${(bytes / 1048576).toFixed(2)} MiB`;
return `JavaScript heap: before=${mib(before)}; highest sampled=${mib(highest)}; after=${mib(after)}.`;
};
return { sample, finish };
}
@@ -1,4 +1,5 @@
import type { ReviewHarnessScenarioResult } from "./reviewHarnessTypes";
import { createReviewHarnessHeapSampler } from "./reviewHarnessHeapSampling";
export interface IdBenchmarkOperations {
deriveKey(): Promise<unknown>;
@@ -15,15 +16,6 @@ const SAMPLES = 3;
const BATCH_SIZE = 100;
const WARMUP_COUNT = 32;
function readHeap(clock: BenchmarkPerformance): number | undefined {
try {
const bytes = clock.memory?.usedJSHeapSize;
return typeof bytes === "number" && Number.isFinite(bytes) && bytes >= 0 ? bytes : undefined;
} catch {
return undefined;
}
}
function summary(samples: readonly number[]): string {
const sorted = [...samples].sort((a, b) => a - b);
return `median=${sorted[1].toFixed(2)} ms; range=${sorted[0].toFixed(2)}–${sorted[2].toFixed(2)} ms`;
@@ -34,13 +26,7 @@ export async function runReviewHarnessIdBenchmark(
clock: BenchmarkPerformance = performance,
yieldControl: () => Promise<void> = () => new Promise((resolve) => window.setTimeout(resolve, 0))
): Promise<ReviewHarnessScenarioResult> {
const before = readHeap(clock);
let highest = before;
const sampleHeap = () => {
const value = readHeap(clock);
if (value !== undefined) highest = Math.max(highest ?? value, value);
return value;
};
const heap = createReviewHarnessHeapSampler(clock);
const observations = [
"Fixed synthetic inputs; 3 samples, alternating legacy/independent order; 32 warm-up IDs per sample. Legacy Chunk algorithm: xxhash64.",
"Compute timings include input construction and awaited ID generation. Initialisation, warm-up, and pauses between batches are excluded. This does not measure a Rebuild or remote transfer.",
@@ -51,7 +37,7 @@ export async function runReviewHarnessIdBenchmark(
const started = clock.now();
await operations.deriveKey();
derivationSamples.push(clock.now() - started);
sampleHeap();
heap.sample();
}
observations.push(`ID key derivation at save time: ${summary(derivationSamples)} per derivation.`);
@@ -80,7 +66,7 @@ export async function runReviewHarnessIdBenchmark(
const started = clock.now();
for (let i = batch; i < batch + BATCH_SIZE; i++) await scenario.run(i, independent);
elapsed += clock.now() - started;
sampleHeap();
heap.sample();
}
samples[independent ? 1 : 0].push(elapsed);
}
@@ -92,16 +78,7 @@ export async function runReviewHarnessIdBenchmark(
);
}
}
const after = sampleHeap();
if (highest === undefined) {
observations.push("JavaScript heap: unavailable on this device.");
} else {
const mib = (bytes: number | undefined) =>
bytes === undefined ? "unavailable" : `${(bytes / 1048576).toFixed(2)} MiB`;
observations.push(
`JavaScript heap: before=${mib(before)}; highest sampled=${mib(highest)}; after=${mib(after)}.`
);
}
observations.push(heap.finish());
observations.push(
"Heap samples are approximate, may include other Obsidian work, and are affected by garbage collection. They are neither total app RAM nor a true peak."
);
@@ -73,6 +73,7 @@ describe("Review Harness ID measurements", () => {
"JavaScript heap: before=2.00 MiB; highest sampled=5.00 MiB; after=1.00 MiB."
);
expect(result.observations.join("\n")).toContain("neither total app RAM nor a true peak");
expect(reads).toBe(245);
});
it.each([Number.NaN, Number.POSITIVE_INFINITY, -1, "throws"])(