Route replication through provider capabilities

This commit is contained in:
vorotamoroz
2026-08-27 11:16:30 +00:00
parent 7aa41baf08
commit 56444bb98b
23 changed files with 918 additions and 205 deletions
@@ -72,6 +72,18 @@ Replace the factory-registration-only responsibilities of
definitions. Retain a module only for separately identified stateful
behaviour; do not retain an instance merely to add a construction handler.
Serialise active initialisation, replacement, and disposal. Publish the active
provider and Replicator as one context after initialisation, clear that context
before retiring the old adapter, and keep each typed dispatch on one context
snapshot. This is the minimum publication fence for this stage. Waiting for
in-flight adapter work and making acquisitions wait for replacement settlement
remain part of the later active-construction migration.
The lifecycle coordinator coalesces its network work internally, but an
`onResumed` handler settles once that work has been scheduled. It does not hold
later resume consumers until a OneShot transfer or Continuous start has
settled.
At this boundary, existing P2P AutoSync, AutoWatch, and incoming-request
entry points receive the same non-interactive readiness and accepted-peer gate.
The no-interaction authority reaches counterpart RPC authorisation and
+48 -5
View File
@@ -1,7 +1,13 @@
import { LOG_LEVEL_INFO } from "octagonal-wheels/common/logger";
import type PouchDB from "pouchdb-core";
import type { SimpleStore } from "octagonal-wheels/databases/SimpleStoreBase";
import type { HasSettings, ObsidianLiveSyncSettings, EntryDoc } from "@vrtmrz/livesync-commonlib/compat/common/types";
import {
REMOTE_COUCHDB,
REMOTE_MINIO,
type HasSettings,
type ObsidianLiveSyncSettings,
type EntryDoc,
} from "@vrtmrz/livesync-commonlib/compat/common/types";
import { __$checkInstanceBinding } from "@vrtmrz/livesync-commonlib/compat/dev/checks";
import type { Confirm } from "@vrtmrz/livesync-commonlib/compat/interfaces/Confirm";
import type { DatabaseFileAccess } from "@vrtmrz/livesync-commonlib/compat/interfaces/DatabaseFileAccess";
@@ -20,8 +26,7 @@ import type { InjectableServiceHub } from "@vrtmrz/livesync-commonlib/compat/ser
import { AbstractModule } from "./modules/AbstractModule";
import { ModulePeriodicProcess } from "./modules/core/ModulePeriodicProcess";
import { ModuleReplicator } from "./modules/core/ModuleReplicator";
import { ModuleReplicatorCouchDB } from "./modules/core/ModuleReplicatorCouchDB";
import { ModuleReplicatorMinIO } from "./modules/core/ModuleReplicatorMinIO";
import { ModuleReplicationLifecycle } from "./modules/core/ModuleReplicationLifecycle";
import { ModuleConflictChecker } from "./modules/coreFeatures/ModuleConflictChecker";
import { ModuleConflictResolver } from "./modules/coreFeatures/ModuleConflictResolver";
import { ModuleResolvingMismatchedTweaks } from "./modules/coreFeatures/ModuleResolveMismatchedTweaks";
@@ -30,6 +35,15 @@ import type { ServiceModules } from "@vrtmrz/livesync-commonlib/compat/interface
import { ModuleBasicMenu } from "./modules/essential/ModuleBasicMenu";
import { usePrepareDatabaseForUse } from "@vrtmrz/livesync-commonlib/compat/serviceFeatures/prepareDatabaseForUse";
import type { Constructor } from "@vrtmrz/livesync-commonlib/compat/common/utils.type";
import {
CAPABILITY_NOT_APPLICABLE,
defineReplicatorProviderDefinitions,
supportedOpenReplicationContinuous,
supportedOpenReplicationOneShot,
supportedOpenReplicationUnattended,
} from "@vrtmrz/livesync-commonlib/replication";
import { LiveSyncCouchDBReplicator } from "@vrtmrz/livesync-commonlib/compat/replication/couchdb/LiveSyncReplicator";
import { LiveSyncJournalReplicator } from "@vrtmrz/livesync-commonlib/compat/replication/journal/LiveSyncJournalReplicator";
export class LiveSyncBaseCore<
T extends ServiceContext = ServiceContext,
@@ -77,6 +91,7 @@ export class LiveSyncBaseCore<
featuresInitialiser: (core: LiveSyncBaseCore<T, TCommands>) => void
) {
this._services = serviceHub;
this.registerReplicatorProviders();
this._serviceModules = serviceModuleInitialiser(this, serviceHub);
const extraModules = extraModuleInitialiser(this);
this.registerModules(extraModules);
@@ -136,12 +151,40 @@ export class LiveSyncBaseCore<
this.modules.push(module);
}
/** Compose the current central providers before any lifecycle event can acquire one. */
private registerReplicatorProviders() {
const definitions = defineReplicatorProviderDefinitions([REMOTE_COUCHDB, REMOTE_MINIO] as const, {
[REMOTE_COUCHDB]: {
kind: REMOTE_COUCHDB,
diagnosticName: "CouchDB",
isConfigured: (settings) =>
settings.remoteType === REMOTE_COUCHDB &&
!!settings.couchDB_URI?.trim() &&
!!settings.couchDB_DBNAME?.trim(),
create: (_settings) => Promise.resolve(new LiveSyncCouchDBReplicator(this)),
userInitiatedOneShot: supportedOpenReplicationOneShot(),
unattendedOneShot: supportedOpenReplicationUnattended(),
continuous: supportedOpenReplicationContinuous(),
},
[REMOTE_MINIO]: {
kind: REMOTE_MINIO,
diagnosticName: "Object Storage",
isConfigured: (settings) =>
settings.remoteType === REMOTE_MINIO && !!settings.endpoint?.trim() && !!settings.bucket?.trim(),
create: (_settings) => Promise.resolve(new LiveSyncJournalReplicator(this)),
userInitiatedOneShot: supportedOpenReplicationOneShot(),
unattendedOneShot: supportedOpenReplicationUnattended(),
continuous: CAPABILITY_NOT_APPLICABLE,
},
});
this.services.replicator.registerReplicatorProviderDefinitions(definitions);
}
public registerModules(extraModules: AbstractModule[] = []) {
this._registerModule(new ModuleLiveSyncMain(this));
this._registerModule(new ModuleConflictChecker(this));
this._registerModule(new ModuleReplicatorMinIO(this));
this._registerModule(new ModuleReplicatorCouchDB(this));
this._registerModule(new ModuleReplicator(this));
this._registerModule(new ModuleReplicationLifecycle(this));
this._registerModule(new ModuleConflictResolver(this));
this._registerModule(new ModulePeriodicProcess(this));
this._registerModule(new ModuleResolvingMismatchedTweaks(this));
@@ -1,5 +1,6 @@
import { describe, expect, it, vi, beforeEach, afterEach } from "vitest";
import { createServiceContext } from "@vrtmrz/livesync-commonlib/context";
import { NO_INTERACTION } from "@vrtmrz/livesync-commonlib/replication";
import { runCommand } from "./runCommand";
import type { CLIOptions } from "./types";
@@ -18,6 +19,7 @@ vi.mock("@vrtmrz/livesync-commonlib/compat/services/base/UnresolvedErrorManager"
}));
import * as offlineScanner from "@vrtmrz/livesync-commonlib/compat/serviceFeatures/offlineScanner";
import { getReplicationSchedulingControl } from "@/modules/core/ReplicationScheduling";
function createCoreMock() {
const standardIo = {
@@ -38,7 +40,7 @@ function createCoreMock() {
currentSettings: vi.fn(() => ({ liveSync: true, syncOnStart: false })),
},
replication: {
replicate: vi.fn(async () => true),
replicateUnattended: vi.fn(async () => ({ status: "completed" as const })),
},
appLifecycle: {
onUnload: {
@@ -123,6 +125,7 @@ describe("daemon command", () => {
await runCommand(makeDaemonOptions(30), { ...baseContext, core });
expect(setTimeoutSpy).toHaveBeenCalledTimes(1);
expect(getReplicationSchedulingControl(core).externalPolling).toBe(true);
// Interval should be in milliseconds (30s → 30000ms)
expect(setTimeoutSpy).toHaveBeenCalledWith(expect.any(Function), 30000);
});
@@ -194,9 +197,9 @@ describe("daemon command", () => {
it("calls replicate before performFullScan", async () => {
const core = createCoreMock();
const callOrder: string[] = [];
core.services.replication.replicate = vi.fn(async () => {
core.services.replication.replicateUnattended = vi.fn(async () => {
callOrder.push("replicate");
return true;
return { status: "completed" as const };
});
vi.mocked(offlineScanner.performFullScan).mockImplementation(async () => {
callOrder.push("performFullScan");
@@ -206,11 +209,19 @@ describe("daemon command", () => {
await runCommand(makeDaemonOptions(), { ...baseContext, core });
expect(callOrder).toEqual(["replicate", "performFullScan"]);
expect(core.services.replication.replicateUnattended).toHaveBeenCalledWith({
trigger: "daemon",
interaction: NO_INTERACTION,
});
expect(getReplicationSchedulingControl(core).initialOneShotSatisfied).toBe(true);
});
it("returns false when initial replication fails", async () => {
const core = createCoreMock();
core.services.replication.replicate = vi.fn(async () => false);
core.services.replication.replicateUnattended = vi.fn(async () => ({
status: "failed" as const,
error: new Error("initial replication failed"),
}));
vi.mocked(offlineScanner.performFullScan).mockClear();
const result = await runCommand(makeDaemonOptions(), { ...baseContext, core });
@@ -218,6 +229,10 @@ describe("daemon command", () => {
expect(result).toBe(false);
// performFullScan should NOT have been called
expect(offlineScanner.performFullScan).not.toHaveBeenCalled();
expect(core.services.replication.replicateUnattended).toHaveBeenCalledWith({
trigger: "daemon",
interaction: NO_INTERACTION,
});
});
it("polling mode: registers onUnload handler that clears timeout", async () => {
@@ -242,11 +257,11 @@ describe("daemon command", () => {
// startup replicate (call 1) succeeds; poll calls 27 fail; call 8 succeeds.
let callCount = 0;
core.services.replication.replicate = vi.fn(async () => {
core.services.replication.replicateUnattended = vi.fn(async () => {
callCount++;
if (callCount === 1) return true; // initial startup replicate
if (callCount === 1) return { status: "completed" as const }; // initial startup replicate
if (callCount <= 7) throw new Error("network failure");
return true; // recovery
return { status: "completed" as const }; // recovery
});
const baseMs = 30 * 1000;
@@ -297,9 +312,9 @@ describe("daemon command", () => {
// Make replicate succeed on the initial call (startup), then fail on the poll.
let callCount = 0;
core.services.replication.replicate = vi.fn(async () => {
core.services.replication.replicateUnattended = vi.fn(async () => {
callCount++;
if (callCount === 1) return true; // startup replicate
if (callCount === 1) return { status: "completed" as const }; // startup replicate
throw new Error("network failure");
});
+41 -16
View File
@@ -26,6 +26,12 @@ import { fsPromises as fs, path } from "@vrtmrz/livesync-commonlib/node";
import type { LiveSyncCouchDBReplicator } from "@vrtmrz/livesync-commonlib/compat/replication/couchdb/LiveSyncReplicator";
import type { LiveSyncJournalReplicator } from "@vrtmrz/livesync-commonlib/compat/replication/journal/LiveSyncJournalReplicator";
import { writeStderrLine, writeStdoutLine } from "@/apps/cli/cliOutput";
import {
isReplicationCompleted,
NO_INTERACTION,
USER_INITIATED_REPLICATION_AUTHORITY,
} from "@vrtmrz/livesync-commonlib/replication";
import { markInitialOneShotSatisfied, setExternalPollingMode } from "@/modules/core/ReplicationScheduling";
function redactConnectionString(uri: string): string {
return uri.replace(/\/\/([^@/]+)@/u, "//***@");
@@ -95,19 +101,28 @@ export async function runCommand(options: CLIOptions, context: CLICommandContext
if (options.command === "daemon") {
const log = (msg: unknown) => writeStderrLine(standardIo, `[Daemon] ${String(msg)}`);
// The daemon owns its own recurring poller. Suppress the application
// resume starter and generic periodic timer before restoring settings.
setExternalPollingMode(core, !!options.interval);
// Skip the config mismatch dialog — the daemon cannot resolve it interactively
// and the default "Dismiss" action would block replication. The daemon should
// accept whatever configuration the remote has.
await core.services.setting.applyPartial({ disableCheckingConfigMismatch: true }, true);
// 1. Replicate CouchDB → local PouchDB so the mirror scan has content to work with.
log("Replicating from CouchDB...");
const replResult = await core.services.replication.replicate(true);
if (!replResult) {
writeStderrLine(standardIo, "[Daemon] Initial CouchDB replication failed, cannot continue");
// 1. Replicate the configured remote into the local database so the
// mirror scan has content to work with.
log("Replicating from remote...");
const replResult = await core.services.replication.replicateUnattended({
trigger: "daemon",
interaction: NO_INTERACTION,
});
if (!isReplicationCompleted(replResult)) {
writeStderrLine(standardIo, "[Daemon] Initial replication failed, cannot continue");
return false;
}
log("CouchDB replication complete");
markInitialOneShotSatisfied(core);
log("Initial replication complete");
// 2. Mirror scan to reconcile PouchDB ↔ local filesystem.
const errorManager = new UnresolvedErrorManager(core.services.appLifecycle, core.services.context.events);
@@ -129,8 +144,9 @@ export async function runCommand(options: CLIOptions, context: CLICommandContext
true
);
// applySettings fires the full lifecycle: onSuspending → onResumed.
// ModuleReplicatorCouchDB starts continuous replication on onResumed
// via fireAndForget.
// The provider-independent lifecycle coordinator owns any eligible
// Continuous start; the daemon marker suppresses a duplicate
// sync-on-start OneShot.
await core.services.control.applySettings();
// Lifecycle events (onSuspending) may re-enable suspension flags.
// Clear them explicitly after the lifecycle completes. applyPartial
@@ -153,7 +169,13 @@ export async function runCommand(options: CLIOptions, context: CLICommandContext
const poll = async () => {
try {
await core.services.replication.replicate(true);
const result = await core.services.replication.replicateUnattended({
trigger: "daemon",
interaction: NO_INTERACTION,
});
if (!isReplicationCompleted(result)) {
throw new Error(`Daemon polling replication did not complete (${result.status}).`);
}
if (consecutiveFailures > 0) {
consecutiveFailures--;
currentIntervalMs = Math.max(currentIntervalMs / 2, baseIntervalMs);
@@ -182,11 +204,11 @@ export async function runCommand(options: CLIOptions, context: CLICommandContext
return true;
});
} else {
log("LiveSync mode: restoring sync settings and starting _changes feed");
log("LiveSync mode: restoring sync settings and starting continuous synchronisation where supported");
await restoreSyncSettings();
// The applySettings() lifecycle fires onResumed → ModuleReplicatorCouchDB which
// starts continuous replication via fireAndForget(openReplication). Don't call
// openReplication directly — it races with the handler and causes dedup/termination.
// The applySettings() lifecycle fires onResumed → the provider-
// independent lifecycle coordinator, which starts Continuous when
// supported. Do not call a concrete Replicator directly.
log("LiveSync active");
const currentSettings = core.services.setting.currentSettings();
if (!currentSettings.liveSync && !currentSettings.syncOnStart) {
@@ -204,8 +226,11 @@ export async function runCommand(options: CLIOptions, context: CLICommandContext
if (options.command === "sync") {
writeStdoutLine(standardIo, "[Command] sync");
const result = await core.services.replication.replicate(true);
if (!result) {
const result = await core.services.replication.replicateUserInitiated({
trigger: "manual",
interaction: USER_INITIATED_REPLICATION_AUTHORITY,
});
if (!isReplicationCompleted(result)) {
// TODO: Standardise the logic for identifying the cause of replication
// failure so that every reason (locked DB, version mismatch, network
// error, etc.) is surfaced with a CLI-specific actionable message.
@@ -218,7 +243,7 @@ export async function runCommand(options: CLIOptions, context: CLICommandContext
);
}
}
return !!result;
return isReplicationCompleted(result);
}
if (options.command === "p2p-peers") {
+5 -1
View File
@@ -24,6 +24,7 @@
import { LOG_LEVEL_NOTICE, Logger } from "octagonal-wheels/common/logger";
import type { LiveSyncBaseCore } from "@/LiveSyncBaseCore.ts";
import { $msg as translateMessage } from "@/common/translation";
import { USER_INITIATED_REPLICATION_AUTHORITY } from "@vrtmrz/livesync-commonlib/replication";
export let plugin: ObsidianLiveSyncPlugin;
export let core :LiveSyncBaseCore;
// $: core = plugin.core;
@@ -104,7 +105,10 @@
await requestUpdate();
}
async function replicate() {
await core.services.replication.replicate(true);
await core.services.replication.replicateUserInitiated({
trigger: "manual",
interaction: USER_INITIATED_REPLICATION_AUTHORITY,
});
}
function selectAllNewest(selectMode: boolean) {
selectNewestPulse++;
@@ -0,0 +1,287 @@
import { afterEach, describe, expect, it, vi } from "vitest";
import {
AUTO_MERGED,
DEFAULT_SETTINGS,
REMOTE_P2P,
type FilePathWithPrefix,
} from "@vrtmrz/livesync-commonlib/compat/common/types";
import { NO_INTERACTION } from "@vrtmrz/livesync-commonlib/replication";
import { EVENT_FILE_SAVED, eventHub } from "@/common/events";
const taskMocks = vi.hoisted(() => ({
scheduleTask: vi.fn((_key: string, _delay: number, task: () => unknown) => task()),
}));
vi.mock("octagonal-wheels/concurrency/task", () => taskMocks);
import { ModuleConflictResolver } from "../coreFeatures/ModuleConflictResolver";
import { ModuleObsidianEvents } from "../essentialObsidian/ModuleObsidianEvents";
import { ModulePeriodicProcess } from "./ModulePeriodicProcess";
import { ModuleReplicationLifecycle } from "./ModuleReplicationLifecycle";
import { ModuleReplicator } from "./ModuleReplicator";
function createApi() {
return {
addLog: vi.fn(),
addCommand: vi.fn(),
registerWindow: vi.fn(),
addRibbonIcon: vi.fn(),
registerProtocolHandler: vi.fn(),
setInterval: vi.fn(),
clearInterval: vi.fn(),
};
}
function p2pSettings(overrides: Partial<typeof DEFAULT_SETTINGS> = {}) {
return {
...DEFAULT_SETTINGS,
remoteType: REMOTE_P2P,
isConfigured: true,
...overrides,
};
}
function createObsidianEventHarness(settings: Partial<typeof DEFAULT_SETTINGS>) {
const save = vi.fn();
const saveCommand = { callback: save };
const replicateUnattendedByEvent = vi.fn(async () => ({ status: "completed" as const }));
const queueCheckForIfOpen = vi.fn(async () => undefined);
const services = {
API: createApi(),
appLifecycle: {
isReady: vi.fn(() => true),
isSuspended: vi.fn(() => false),
},
conflict: { queueCheckForIfOpen },
control: { hasUnloaded: vi.fn(() => false) },
fileProcessing: { commitPendingFileEvents: vi.fn(async () => true) },
replication: { replicateUnattendedByEvent },
};
const core = {
_services: services,
services,
settings: p2pSettings(settings),
} as any;
const plugin = {
app: {
commands: {
commands: { "editor:save-file": saveCommand },
executeCommandById: vi.fn(),
},
},
} as any;
return {
module: new ModuleObsidianEvents(plugin, core),
queueCheckForIfOpen,
replicateUnattendedByEvent,
save,
saveCommand,
services,
};
}
describe("automatic replication triggers while P2P is active", () => {
afterEach(() => {
eventHub.offAll();
taskMocks.scheduleTask.mockClear();
});
it("keeps periodic synchronisation on the provider-independent replication boundary", async () => {
const replicateUnattended = vi.fn(async () => ({ status: "completed" as const }));
const services = {
API: createApi(),
control: { hasUnloaded: vi.fn(() => false) },
replication: { replicateUnattended },
};
const core = {
_services: services,
services,
settings: p2pSettings({ periodicReplication: true }),
} as any;
const module = new ModulePeriodicProcess(core);
await module.periodicSyncProcessor.process();
expect(replicateUnattended).toHaveBeenCalledOnce();
expect(replicateUnattended).toHaveBeenCalledWith({
trigger: "periodic",
interaction: NO_INTERACTION,
});
});
it("keeps database-save synchronisation on the event replication boundary", async () => {
const replicateUnattendedByEvent = vi.fn(async () => ({ status: "completed" as const }));
const settings = p2pSettings({ syncOnSave: true });
const services = {
appLifecycle: { isSuspended: vi.fn(() => false) },
replication: { replicateUnattendedByEvent },
};
const module = {
core: { services, settings },
services,
settings,
getNormalFileReflectionFilterSignature: (
ModuleReplicator.prototype as unknown as {
getNormalFileReflectionFilterSignature: (value: typeof settings) => string;
}
).getNormalFileReflectionFilterSignature,
};
await (ModuleReplicator.prototype as any)._everyOnloadAfterLoadSettings.call(module);
eventHub.emitEvent(EVENT_FILE_SAVED);
await vi.waitFor(() => expect(replicateUnattendedByEvent).toHaveBeenCalledOnce());
expect(replicateUnattendedByEvent).toHaveBeenCalledWith({
trigger: "database-event",
interaction: NO_INTERACTION,
});
});
it("keeps editor-save synchronisation on the event replication boundary", async () => {
const { module, replicateUnattendedByEvent, save, saveCommand } = createObsidianEventHarness({
syncOnEditorSave: true,
});
module.swapSaveCommand();
saveCommand.callback();
expect(save).toHaveBeenCalledOnce();
await vi.waitFor(() => expect(replicateUnattendedByEvent).toHaveBeenCalledOnce());
expect(replicateUnattendedByEvent).toHaveBeenCalledWith({
trigger: "editor-save",
interaction: NO_INTERACTION,
});
});
it("keeps file-open synchronisation on the event replication boundary", async () => {
const { module, queueCheckForIfOpen, replicateUnattendedByEvent, services } = createObsidianEventHarness({
syncOnFileOpen: true,
});
const file = { path: "opened.md" } as never;
await module.watchWorkspaceOpenAsync(file);
expect(services.fileProcessing.commitPendingFileEvents).toHaveBeenCalledOnce();
expect(replicateUnattendedByEvent).toHaveBeenCalledOnce();
expect(replicateUnattendedByEvent).toHaveBeenCalledWith({
trigger: "file-open",
interaction: NO_INTERACTION,
});
expect(queueCheckForIfOpen).toHaveBeenCalledWith("opened.md");
});
it("keeps post-merge synchronisation on the event replication boundary", async () => {
const replicateUnattendedByEvent = vi.fn(async () => ({ status: "completed" as const }));
const queueCheckFor = vi.fn(async () => undefined);
const path = "merged.md" as FilePathWithPrefix;
const module = {
settings: p2pSettings({ syncAfterMerge: true }),
services: {
appLifecycle: { isSuspended: vi.fn(() => false) },
conflict: { queueCheckFor },
replication: { replicateUnattendedByEvent },
},
checkConflictAndPerformAutoMerge: vi.fn(async () => AUTO_MERGED),
_log: vi.fn(),
};
await (ModuleConflictResolver.prototype as any)._resolveConflict.call(module, path);
expect(replicateUnattendedByEvent).toHaveBeenCalledOnce();
expect(replicateUnattendedByEvent).toHaveBeenCalledWith({
trigger: "merge",
interaction: NO_INTERACTION,
});
expect(queueCheckFor).toHaveBeenCalledWith(path);
});
});
describe("recurring replication scheduling precedence", () => {
afterEach(() => {
eventHub.offAll();
});
function createRecurringSchedulingHarness() {
const resumeHandlers: Array<() => Promise<boolean>> = [];
const settingRealisedHandlers: Array<() => Promise<boolean>> = [];
let resolveContinuous!: (
outcome: { status: "completed" } | { status: "blocked"; reason: "capability-not-applicable" }
) => void;
const startContinuous = vi.fn(
() =>
new Promise<{ status: "completed" } | { status: "blocked"; reason: "capability-not-applicable" }>(
(resolve) => {
resolveContinuous = resolve;
}
)
);
const API = createApi();
const settings = {
...DEFAULT_SETTINGS,
isConfigured: true,
liveSync: true,
syncOnStart: true,
periodicReplication: true,
periodicReplicationInterval: 60,
};
const services = {
API,
appLifecycle: {
isReady: vi.fn(() => true),
isSuspended: vi.fn(() => false),
onResumed: { addHandler: vi.fn((handler: () => Promise<boolean>) => resumeHandlers.push(handler)) },
onSuspending: { addHandler: vi.fn() },
onUnload: { addHandler: vi.fn() },
},
control: { hasUnloaded: vi.fn(() => false) },
replication: {
startContinuous,
replicateUnattended: vi.fn(async () => ({ status: "completed" as const })),
},
setting: {
currentSettings: vi.fn(() => settings),
onBeforeRealiseSetting: { addHandler: vi.fn() },
onSettingRealised: {
addHandler: vi.fn((handler: () => Promise<boolean>) => settingRealisedHandlers.push(handler)),
},
},
};
const core = { _services: services, services, settings } as any;
const lifecycle = new ModuleReplicationLifecycle(core);
const periodic = new ModulePeriodicProcess(core);
lifecycle.onBindFunction(core, services as never);
periodic.onBindFunction(core, services as never);
return {
API,
resolveContinuous: (
outcome: { status: "completed" } | { status: "blocked"; reason: "capability-not-applicable" }
) => resolveContinuous(outcome),
resume: async () => await Promise.all(resumeHandlers.map(async (handler) => await handler())),
realiseSettings: async () =>
await Promise.all(settingRealisedHandlers.map(async (handler) => await handler())),
};
}
it("does not enable the generic periodic timer while Continuous owns recurring synchronisation", async () => {
const harness = createRecurringSchedulingHarness();
await harness.resume();
await harness.realiseSettings();
expect(harness.API.setInterval).not.toHaveBeenCalled();
harness.resolveContinuous({ status: "completed" });
await vi.waitFor(() => expect(harness.API.setInterval).not.toHaveBeenCalled());
});
it("restores the generic periodic timer when Continuous is not applicable", async () => {
const harness = createRecurringSchedulingHarness();
await harness.resume();
await harness.realiseSettings();
harness.resolveContinuous({ status: "blocked", reason: "capability-not-applicable" });
await vi.waitFor(() => expect(harness.API.setInterval).toHaveBeenCalledOnce());
});
});
+22 -1
View File
@@ -1,15 +1,27 @@
import { PeriodicProcessor } from "@/common/PeriodicProcessor";
import type { LiveSyncCore } from "@/main";
import { AbstractModule } from "@/modules/AbstractModule";
import { NO_INTERACTION } from "@vrtmrz/livesync-commonlib/replication";
import { getReplicationSchedulingControl } from "./ReplicationScheduling";
export class ModulePeriodicProcess extends AbstractModule {
periodicSyncProcessor = new PeriodicProcessor(this.core, async () => await this.services.replication.replicate());
private readonly schedulingControl = getReplicationSchedulingControl(this.core);
periodicSyncProcessor = new PeriodicProcessor(this.core, async () => {
await this.services.replication.replicateUnattended({
trigger: "periodic",
interaction: NO_INTERACTION,
});
});
disablePeriodic() {
this.periodicSyncProcessor?.disable();
return Promise.resolve(true);
}
resumePeriodic() {
if (this.schedulingControl.externalPolling || this.schedulingControl.continuousOwnsRecurring) {
void this.disablePeriodic();
return Promise.resolve(true);
}
this.periodicSyncProcessor.enable(
this.settings.periodicReplication ? this.settings.periodicReplicationInterval * 1000 : 0
);
@@ -32,6 +44,15 @@ export class ModulePeriodicProcess extends AbstractModule {
}
override onBindFunction(core: LiveSyncCore, services: typeof core.services): void {
this.schedulingControl.disablePeriodic = () => {
void this.disablePeriodic();
};
this.schedulingControl.refreshPeriodic = () => {
void this.resumePeriodic();
};
if (this.schedulingControl.externalPolling || this.schedulingControl.continuousOwnsRecurring) {
void this.disablePeriodic();
}
services.appLifecycle.onUnload.addHandler(this._allOnUnload.bind(this));
services.setting.onBeforeRealiseSetting.addHandler(this._everyBeforeRealizeSetting.bind(this));
services.setting.onSettingRealised.addHandler(this._everyAfterRealizeSetting.bind(this));
@@ -0,0 +1,113 @@
import { LOG_LEVEL_VERBOSE } from "octagonal-wheels/common/logger";
import {
isReplicationCompleted,
NO_INTERACTION,
type ReplicationOutcome,
} from "@vrtmrz/livesync-commonlib/replication";
import { AbstractModule } from "@/modules/AbstractModule";
import type { LiveSyncCore } from "@/main";
import {
getReplicationSchedulingControl,
markInitialOneShotSatisfied,
setContinuousSchedulingOwnership,
setExternalPollingMode,
} from "./ReplicationScheduling";
function isCapabilityUnavailable(result: ReplicationOutcome): boolean {
return (
result.status === "blocked" &&
(result.reason === "capability-not-applicable" || result.reason === "capability-not-implemented")
);
}
/**
* Coordinates application resume with the active provider's typed roles.
* Provider implementations do not subscribe to the application lifecycle.
*/
export class ModuleReplicationLifecycle extends AbstractModule {
private readonly schedulingControl = getReplicationSchedulingControl(this.core);
private resumePromise?: Promise<boolean>;
private async runAfterResume(): Promise<boolean> {
if (this.schedulingControl.externalPolling) return true;
if (this.services.appLifecycle.isSuspended()) return true;
if (!this.services.appLifecycle.isReady()) return true;
const settings = this.services.setting.currentSettings();
if (!settings.isConfigured) {
setContinuousSchedulingOwnership(this.core, false);
return true;
}
const skipOneShot = this.schedulingControl.initialOneShotSatisfied;
if (settings.liveSync) {
// Reserve recurring ownership before the asynchronous start so a
// later resume handler cannot enable Periodic in the meantime.
setContinuousSchedulingOwnership(this.core, true);
const result = await this.services.replication.startContinuous({
trigger: "resume",
interaction: NO_INTERACTION,
});
if (!isReplicationCompleted(result)) {
setContinuousSchedulingOwnership(this.core, false);
}
// The daemon's initial finite convergence must not suppress a
// supported Continuous start. It only suppresses the fallback
// OneShot when Continuous is unavailable.
this.schedulingControl.initialOneShotSatisfied = false;
if (isCapabilityUnavailable(result) && settings.syncOnStart && !skipOneShot) {
await this.services.replication.replicateUnattended({
trigger: "resume",
interaction: NO_INTERACTION,
});
}
return true;
}
setContinuousSchedulingOwnership(this.core, false);
if (settings.syncOnStart && !skipOneShot) {
await this.services.replication.replicateUnattended({
trigger: "resume",
interaction: NO_INTERACTION,
});
}
this.schedulingControl.initialOneShotSatisfied = false;
return true;
}
private _everyAfterResumeProcess(): Promise<boolean> {
if (!this.resumePromise) {
// The lifecycle event is a short notification boundary. Keep the
// long-running OneShot/Continuous start coalesced internally, but
// let later resume handlers (P2P, periodic scheduling, and other
// modules) continue without waiting for network work to settle.
this.resumePromise = this.runAfterResume()
.catch((error) => {
this._log(error, LOG_LEVEL_VERBOSE);
return true;
})
.finally(() => {
this.resumePromise = undefined;
});
}
return Promise.resolve(true);
}
/**
* Let a CLI daemon own recurring polling without a duplicate lifecycle or
* generic periodic scheduler. This is intentionally narrower than a
* provider or Replicator control API.
*/
setExternalPollingMode(enabled: boolean): void {
setExternalPollingMode(this.core, enabled);
}
/** Mark the daemon's initial finite convergence for the next resume. */
markInitialOneShotSatisfied(): void {
markInitialOneShotSatisfied(this.core);
}
override onBindFunction(core: LiveSyncCore, services: typeof core.services): void {
services.appLifecycle.onResumed.addHandler(this._everyAfterResumeProcess.bind(this));
}
}
+26 -2
View File
@@ -23,6 +23,7 @@ import { clearHandlers } from "@vrtmrz/livesync-commonlib/compat/replication/Syn
import type { NecessaryServices } from "@vrtmrz/livesync-commonlib/compat/interfaces/ServiceModule";
import { MARK_LOG_NETWORK_ERROR } from "@vrtmrz/livesync-commonlib/compat/services/lib/logUtils";
import { usesLegacyIndexedDBAdapter } from "@/common/compatibilitySettings.ts";
import { NO_INTERACTION, type ReplicationInteraction } from "@vrtmrz/livesync-commonlib/replication";
function isOnlineAndCanReplicate(
errorManager: UnresolvedErrorManager,
@@ -108,7 +109,12 @@ export class ModuleReplicator extends AbstractModule {
this._normalFileReflectionFilterSignature = this.getNormalFileReflectionFilterSignature(this.settings);
eventHub.onEvent(EVENT_FILE_SAVED, () => {
if (this.settings.syncOnSave && !this.core.services.appLifecycle.isSuspended()) {
scheduleTask("perform-replicate-after-save", 250, () => this.services.replication.replicateByEvent());
scheduleTask("perform-replicate-after-save", 250, () =>
this.services.replication.replicateUnattendedByEvent({
trigger: "database-event",
interaction: NO_INTERACTION,
})
);
}
});
eventHub.onEvent(EVENT_SETTING_SAVED, (setting) => {
@@ -223,12 +229,30 @@ Even if you choose to clean up, you will see this option again if you exit Obsid
});
}
private async onReplicationFailed(showMessage: boolean = false): Promise<boolean> {
private async onReplicationFailed(
showMessageOrInteraction: boolean | ReplicationInteraction = false,
interaction?: ReplicationInteraction
): Promise<boolean> {
// The typed ReplicationService passes the legacy visibility flag first
// and the authority second. The authority is the source of truth for
// recovery dialogues when it is present; retain the legacy boolean for
// older callers which do not provide one.
const showMessage = interaction
? interaction.kind === "permitted" && interaction.permissions.failureRecovery
: typeof showMessageOrInteraction === "boolean"
? showMessageOrInteraction
: showMessageOrInteraction.kind === "permitted" && showMessageOrInteraction.permissions.failureRecovery;
const activeReplicator = this.services.replicator.getActiveReplicator();
if (!activeReplicator) {
Logger(`No active replicator found`, LOG_LEVEL_INFO);
return false;
}
if (!showMessage) {
// Automatic requests may report the failure, but they must never
// enter tweak, lock, fetch, unlock, or cleanup dialogues.
Logger(`Replication failed on an unattended path.`, LOG_LEVEL_INFO);
return false;
}
if (activeReplicator.tweakSettingsMismatched && activeReplicator.preferredTweakValue) {
await this.services.tweakValue.askResolvingMismatched(activeReplicator.preferredTweakValue);
} else {
@@ -114,6 +114,60 @@ describe("ModuleReplicator", () => {
eventHub.offAll();
}
});
it("only permits recovery dialogue when the authority grants failure recovery", async () => {
const askResolvingMismatched = vi.fn(async () => undefined);
const activeReplicator = {
tweakSettingsMismatched: true,
preferredTweakValue: { customChunkSize: 60 },
};
const services = {
context: createServiceContext(),
API: {
addLog: vi.fn(),
addCommand: vi.fn(),
registerWindow: vi.fn(),
addRibbonIcon: vi.fn(),
registerProtocolHandler: vi.fn(),
},
appLifecycle: {
getUnresolvedMessages: { addHandler: vi.fn() },
},
replicator: { getActiveReplicator: vi.fn(() => activeReplicator) },
tweakValue: { askResolvingMismatched },
};
const core = {
_services: services,
services,
settings: {},
} as any;
const module = new ModuleReplicator(core);
await (module as any).onReplicationFailed(false);
expect(askResolvingMismatched).not.toHaveBeenCalled();
await (module as any).onReplicationFailed(true, {
kind: "permitted",
permissions: {
peerSelection: true,
localPeerAdmission: true,
configurationExchange: true,
failureRecovery: false,
},
});
expect(askResolvingMismatched).not.toHaveBeenCalled();
await (module as any).onReplicationFailed(true, {
kind: "permitted",
permissions: {
peerSelection: true,
localPeerAdmission: true,
configurationExchange: true,
failureRecovery: true,
},
});
expect(askResolvingMismatched).toHaveBeenCalledOnce();
});
});
describe("compatibility: cleaned-remote reconciliation for IndexedDB clients", () => {
@@ -1,50 +0,0 @@
import { fireAndForget } from "octagonal-wheels/promises";
import { REMOTE_MINIO, REMOTE_P2P, type RemoteDBSettings } from "@vrtmrz/livesync-commonlib/compat/common/types";
import { LiveSyncCouchDBReplicator } from "@vrtmrz/livesync-commonlib/compat/replication/couchdb/LiveSyncReplicator";
import type { LiveSyncAbstractReplicator } from "@vrtmrz/livesync-commonlib/compat/replication/LiveSyncAbstractReplicator";
import { AbstractModule } from "@/modules/AbstractModule";
import type { LiveSyncCore } from "@/main";
export class ModuleReplicatorCouchDB extends AbstractModule {
_anyNewReplicator(settingOverride: Partial<RemoteDBSettings> = {}): Promise<LiveSyncAbstractReplicator | false> {
const settings = { ...this.settings, ...settingOverride };
// If new remote types were added, add them here. Do not use `REMOTE_COUCHDB` directly for the safety valve.
if (settings.remoteType == REMOTE_MINIO || settings.remoteType == REMOTE_P2P) {
return Promise.resolve(false);
}
return Promise.resolve(new LiveSyncCouchDBReplicator(this.core));
}
_everyAfterResumeProcess(): Promise<boolean> {
if (this.services.appLifecycle.isSuspended()) return Promise.resolve(true);
if (!this.services.appLifecycle.isReady()) return Promise.resolve(true);
if (this.settings.remoteType != REMOTE_MINIO && this.settings.remoteType != REMOTE_P2P) {
const LiveSyncEnabled = this.settings.liveSync;
const continuous = LiveSyncEnabled;
const eventualOnStart = !LiveSyncEnabled && this.settings.syncOnStart;
// If enabled LiveSync or on start, open replication
if (LiveSyncEnabled || eventualOnStart) {
// And note that we do not open the conflict detection dialogue directly during this process.
// This should be raised explicitly if needed.
fireAndForget(async () => {
const canReplicate = await this.services.replication.isReplicationReady(false);
if (!canReplicate) return;
const openReplication = () =>
this.core.replicator.openReplication(this.settings, continuous, false, false);
if (continuous) {
void openReplication();
} else {
await this.services.replicator.runFiniteReplicationActivity(openReplication, {
label: "replication",
});
}
});
}
}
return Promise.resolve(true);
}
override onBindFunction(core: LiveSyncCore, services: typeof core.services): void {
services.replicator.getNewReplicator.addHandler(this._anyNewReplicator.bind(this));
services.appLifecycle.onResumed.addHandler(this._everyAfterResumeProcess.bind(this));
}
}
@@ -1,89 +0,0 @@
import { describe, expect, it, vi } from "vitest";
import { ModuleReplicatorCouchDB } from "./ModuleReplicatorCouchDB.ts";
function createModule(settings: { liveSync: boolean; syncOnStart: boolean }, isReplicationReady = true) {
const openReplication = vi.fn(async () => true);
const runFiniteReplicationActivity = vi.fn(async (task: () => unknown) => await task());
const services = {
API: {
addLog: vi.fn(),
addCommand: vi.fn(),
registerWindow: vi.fn(),
addRibbonIcon: vi.fn(),
registerProtocolHandler: vi.fn(),
},
appLifecycle: {
isSuspended: vi.fn(() => false),
isReady: vi.fn(() => true),
},
replication: {
isReplicationReady: vi.fn(async () => isReplicationReady),
},
replicator: {
runFiniteReplicationActivity,
},
setting: {
saveSettingData: vi.fn(async () => undefined),
},
};
const core = {
_services: services,
services,
settings: {
remoteType: "",
...settings,
},
replicator: { openReplication },
} as any;
return {
module: new ModuleReplicatorCouchDB(core),
openReplication,
runFiniteReplicationActivity,
};
}
describe("ModuleReplicatorCouchDB resume replication activity", () => {
it("exposes start-up one-shot replication as finite replication activity", async () => {
const { module, openReplication, runFiniteReplicationActivity } = createModule({
liveSync: false,
syncOnStart: true,
});
await module._everyAfterResumeProcess();
await vi.waitFor(() => expect(openReplication).toHaveBeenCalledOnce());
expect(runFiniteReplicationActivity).toHaveBeenCalledWith(expect.any(Function), {
label: "replication",
});
expect(openReplication).toHaveBeenCalledWith(expect.any(Object), false, false, false);
});
it("does not wrap the unbounded continuous channel in another finite activity", async () => {
const { module, openReplication, runFiniteReplicationActivity } = createModule({
liveSync: true,
syncOnStart: false,
});
await module._everyAfterResumeProcess();
await vi.waitFor(() => expect(openReplication).toHaveBeenCalledOnce());
expect(runFiniteReplicationActivity).not.toHaveBeenCalled();
expect(openReplication).toHaveBeenCalledWith(expect.any(Object), true, false, false);
});
it("does not start a one-shot activity when start-up readiness fails", async () => {
const { module, openReplication, runFiniteReplicationActivity } = createModule(
{
liveSync: false,
syncOnStart: true,
},
false
);
await module._everyAfterResumeProcess();
await new Promise((resolve) => setTimeout(resolve, 0));
expect(runFiniteReplicationActivity).not.toHaveBeenCalled();
expect(openReplication).not.toHaveBeenCalled();
});
});
-18
View File
@@ -1,18 +0,0 @@
import { REMOTE_MINIO, type RemoteDBSettings } from "@vrtmrz/livesync-commonlib/compat/common/types";
import { LiveSyncJournalReplicator } from "@vrtmrz/livesync-commonlib/compat/replication/journal/LiveSyncJournalReplicator";
import type { LiveSyncAbstractReplicator } from "@vrtmrz/livesync-commonlib/compat/replication/LiveSyncAbstractReplicator";
import type { LiveSyncCore } from "@/main";
import { AbstractModule } from "@/modules/AbstractModule";
export class ModuleReplicatorMinIO extends AbstractModule {
_anyNewReplicator(settingOverride: Partial<RemoteDBSettings> = {}): Promise<LiveSyncAbstractReplicator | false> {
const settings = { ...this.settings, ...settingOverride };
if (settings.remoteType == REMOTE_MINIO) {
return Promise.resolve(new LiveSyncJournalReplicator(this.core));
}
return Promise.resolve(false);
}
override onBindFunction(core: LiveSyncCore, services: typeof core.services): void {
services.replicator.getNewReplicator.addHandler(this._anyNewReplicator.bind(this));
}
}
@@ -0,0 +1,200 @@
import { describe, expect, it, vi } from "vitest";
import { REMOTE_P2P } from "@vrtmrz/livesync-commonlib/compat/common/types";
import { NO_INTERACTION, type ReplicationOutcome } from "@vrtmrz/livesync-commonlib/replication";
import { createServiceContext } from "@vrtmrz/livesync-commonlib/context";
import { ModuleReplicationLifecycle } from "./ModuleReplicationLifecycle";
import { getReplicationSchedulingControl, setExternalPollingMode } from "./ReplicationScheduling";
type ResumeHandler = () => Promise<boolean>;
function createResumeHarness(settings: {
liveSync: boolean;
syncOnStart: boolean;
isConfigured?: boolean;
remoteType?: string;
P2P_Enabled?: boolean;
}) {
const resumeHandlers: ResumeHandler[] = [];
const replicateUnattended = vi.fn(async (): Promise<ReplicationOutcome> => ({ status: "completed" }));
const startContinuous = vi.fn(async (): Promise<ReplicationOutcome> => ({ status: "completed" }));
const currentSettings = {
isConfigured: true,
periodicReplication: false,
...settings,
};
const services = {
context: createServiceContext(),
API: {
addLog: vi.fn(),
addCommand: vi.fn(),
registerWindow: vi.fn(),
addRibbonIcon: vi.fn(),
registerProtocolHandler: vi.fn(),
isOnline: true,
},
appLifecycle: {
isReady: vi.fn(() => true),
isSuspended: vi.fn(() => false),
onResumed: {
addHandler: vi.fn((handler: ResumeHandler) => resumeHandlers.push(handler)),
},
},
replication: {
replicateUnattended,
startContinuous,
},
setting: {
currentSettings: vi.fn(() => currentSettings),
},
};
const core = {
_services: services,
services,
settings: currentSettings,
} as any;
const module = new ModuleReplicationLifecycle(core);
module.onBindFunction(core, services as never);
return {
core,
module,
replicateUnattended,
startContinuous,
resume: async () => await Promise.all(resumeHandlers.map((handler) => handler())),
};
}
describe("provider-independent replication resume lifecycle", () => {
it("starts one unattended OneShot when sync-on-start is enabled", async () => {
const harness = createResumeHarness({ liveSync: false, syncOnStart: true });
await harness.resume();
expect(harness.replicateUnattended).toHaveBeenCalledOnce();
expect(harness.replicateUnattended).toHaveBeenCalledWith({
trigger: "resume",
interaction: NO_INTERACTION,
});
expect(harness.startContinuous).not.toHaveBeenCalled();
});
it("falls back to sync-on-start when Continuous is not applicable", async () => {
const harness = createResumeHarness({ liveSync: true, syncOnStart: true });
harness.startContinuous.mockResolvedValue({
status: "blocked",
reason: "capability-not-applicable",
});
await harness.resume();
expect(harness.startContinuous).toHaveBeenCalledWith({
trigger: "resume",
interaction: NO_INTERACTION,
});
expect(harness.replicateUnattended).toHaveBeenCalledWith({
trigger: "resume",
interaction: NO_INTERACTION,
});
});
it("starts Continuous without a finite fallback when it is supported", async () => {
const harness = createResumeHarness({ liveSync: true, syncOnStart: true });
await harness.resume();
expect(harness.startContinuous).toHaveBeenCalledOnce();
expect(harness.replicateUnattended).not.toHaveBeenCalled();
});
it("does not fall back after an actual Continuous failure", async () => {
const harness = createResumeHarness({ liveSync: true, syncOnStart: true });
harness.startContinuous.mockResolvedValue({
status: "failed",
error: new Error("connection failed"),
});
await harness.resume();
expect(harness.replicateUnattended).not.toHaveBeenCalled();
});
it("coalesces concurrent resume callbacks", async () => {
const harness = createResumeHarness({ liveSync: false, syncOnStart: true });
let resolveReplication!: (value: { status: "completed" }) => void;
harness.replicateUnattended.mockImplementationOnce(
() => new Promise((resolve) => (resolveReplication = resolve))
);
const first = harness.resume();
const second = harness.resume();
resolveReplication({ status: "completed" });
await Promise.all([first, second]);
expect(harness.replicateUnattended).toHaveBeenCalledOnce();
});
it("does not block later resume handlers while a OneShot is running", async () => {
const harness = createResumeHarness({ liveSync: false, syncOnStart: true });
let resolveReplication!: (value: { status: "completed" }) => void;
harness.replicateUnattended.mockImplementationOnce(
() => new Promise((resolve) => (resolveReplication = resolve))
);
const resumed = harness.resume();
await expect(resumed).resolves.toEqual([true]);
expect(harness.replicateUnattended).toHaveBeenCalledOnce();
resolveReplication({ status: "completed" });
await resumed;
});
it("skips only the daemon-satisfied OneShot while allowing Continuous", async () => {
const harness = createResumeHarness({ liveSync: true, syncOnStart: true });
getReplicationSchedulingControl(harness.core).initialOneShotSatisfied = true;
harness.startContinuous.mockResolvedValue({
status: "blocked",
reason: "capability-not-applicable",
});
await harness.resume();
expect(harness.startContinuous).toHaveBeenCalledOnce();
expect(harness.replicateUnattended).not.toHaveBeenCalled();
expect(getReplicationSchedulingControl(harness.core).initialOneShotSatisfied).toBe(false);
});
it("does not start lifecycle replication while an external poller owns scheduling", async () => {
const harness = createResumeHarness({ liveSync: false, syncOnStart: true });
setExternalPollingMode(harness.core, true);
await harness.resume();
expect(harness.startContinuous).not.toHaveBeenCalled();
expect(harness.replicateUnattended).not.toHaveBeenCalled();
});
it("requests the generic finite fallback for P2P when Continuous is not applicable", async () => {
const harness = createResumeHarness({
remoteType: REMOTE_P2P,
P2P_Enabled: true,
liveSync: true,
syncOnStart: true,
});
harness.startContinuous.mockResolvedValue({
status: "blocked",
reason: "capability-not-applicable",
});
harness.replicateUnattended.mockResolvedValue({
status: "blocked",
reason: "capability-not-implemented",
});
await harness.resume();
expect(harness.startContinuous).toHaveBeenCalledOnce();
expect(harness.replicateUnattended).toHaveBeenCalledWith({
trigger: "resume",
interaction: NO_INTERACTION,
});
});
});
+49
View File
@@ -0,0 +1,49 @@
/**
* Host-owned scheduling state shared by the lifecycle coordinator and the
* CLI daemon. It deliberately contains policy state only; provider choice and
* replication execution remain in ReplicationService.
*/
export interface ReplicationSchedulingControl {
/** The daemon owns recurring polling and suppresses host automation. */
externalPolling: boolean;
/** A Continuous start is pending or accepted and therefore owns recurring synchronisation. */
continuousOwnsRecurring: boolean;
/** The daemon's initial convergence satisfies the next resume OneShot. */
initialOneShotSatisfied: boolean;
/** Registered by the periodic module so the daemon can remove an old timer. */
disablePeriodic?: () => void;
/** Reconcile the periodic timer after recurring ownership changes. */
refreshPeriodic?: () => void;
}
const controls = new WeakMap<object, ReplicationSchedulingControl>();
export function getReplicationSchedulingControl(owner: object): ReplicationSchedulingControl {
let control = controls.get(owner);
if (!control) {
control = {
externalPolling: false,
continuousOwnsRecurring: false,
initialOneShotSatisfied: false,
};
controls.set(owner, control);
}
return control;
}
export function setContinuousSchedulingOwnership(owner: object, ownsRecurring: boolean): void {
const control = getReplicationSchedulingControl(owner);
if (control.continuousOwnsRecurring === ownsRecurring) return;
control.continuousOwnsRecurring = ownsRecurring;
control.refreshPeriodic?.();
}
export function setExternalPollingMode(owner: object, enabled: boolean): void {
const control = getReplicationSchedulingControl(owner);
control.externalPolling = enabled;
if (enabled) control.disablePeriodic?.();
}
export function markInitialOneShotSatisfied(owner: object): void {
getReplicationSchedulingControl(owner).initialOneShotSatisfied = true;
}
@@ -19,6 +19,7 @@ import { stripAllPrefixes, isPlainText } from "@vrtmrz/livesync-commonlib/compat
import { EVENT_CONFLICT_CANCELLED, eventHub } from "@/common/events.ts";
import type { InjectableServiceHub } from "@vrtmrz/livesync-commonlib/compat/services/implements/injectable/InjectableServiceHub";
import type { LiveSyncCore } from "@/main.ts";
import { NO_INTERACTION } from "@vrtmrz/livesync-commonlib/replication";
export class ModuleConflictResolver extends AbstractModule {
private async _resolveConflictByDeletingRev(
@@ -142,7 +143,10 @@ export class ModuleConflictResolver extends AbstractModule {
//auto resolved, but need check again;
if (this.settings.syncAfterMerge && !this.services.appLifecycle.isSuspended()) {
//Wait for the running replication, if not running replication, run it once.
await this.services.replication.replicateByEvent();
await this.services.replication.replicateUnattendedByEvent({
trigger: "merge",
interaction: NO_INTERACTION,
});
}
this._log("[conflict] Automatically merged, but we have to check it again");
await this.services.conflict.queueCheckFor(filename);
@@ -37,7 +37,7 @@ function createModule(files: FilePathWithPrefix[] = []) {
isSuspended: vi.fn(() => false),
},
replication: {
replicateByEvent: vi.fn(async () => true),
replicateUnattendedByEvent: vi.fn(async () => ({ status: "completed" as const })),
},
vault: {
getActiveFilePath: vi.fn(() => undefined),
+5 -1
View File
@@ -4,6 +4,7 @@ import { fireAndForget } from "octagonal-wheels/promises";
import { AbstractModule } from "@/modules/AbstractModule";
import { $msg } from "@/common/translation";
import { copyFileDatabaseInfo } from "@/serviceFeatures/fileDatabaseInfo";
import { USER_INITIATED_REPLICATION_AUTHORITY } from "@vrtmrz/livesync-commonlib/replication";
// Separated Module for basic menu commands, which are not related to obsidian specific features. It is expected to be used in other platforms with minimal changes.
// However, it is odd that it has here at all; it really ought to be in each respective feature. It will likely be moved eventually. Until now, addCommand pointed to Obsidian's version.
export class ModuleBasicMenu extends AbstractModule {
@@ -12,7 +13,10 @@ export class ModuleBasicMenu extends AbstractModule {
id: "livesync-replicate",
name: $msg("Sync now"),
callback: async () => {
await this.services.replication.replicate();
await this.services.replication.replicateUserInitiated({
trigger: "manual",
interaction: USER_INITIATED_REPLICATION_AUTHORITY,
});
},
});
this.addCommand({
@@ -25,7 +25,7 @@ function createFixture() {
registerProtocolHandler: vi.fn(),
},
replication: {
replicate: vi.fn(async () => undefined),
replicateUserInitiated: vi.fn(async () => ({ status: "completed" as const })),
},
vault: {
getActiveFilePath: vi.fn((): string | null => "note.md"),
@@ -14,6 +14,7 @@ import {
} from "@vrtmrz/livesync-commonlib/compat/mock_and_interop/stores";
import type { LiveSyncCore } from "@/main.ts";
import { compatGlobal } from "@vrtmrz/livesync-commonlib/compat/common/coreEnvFunctions";
import { NO_INTERACTION } from "@vrtmrz/livesync-commonlib/replication";
type MutableCommandDefinition = {
callback?: () => void;
@@ -71,7 +72,12 @@ export class ModuleObsidianEvents extends AbstractObsidianModule {
} else {
if (this.settings.syncOnEditorSave) {
this._log("Sync on Editor Save.", LOG_LEVEL_VERBOSE);
fireAndForget(() => this.services.replication.replicateByEvent());
fireAndForget(() =>
this.services.replication.replicateUnattendedByEvent({
trigger: "editor-save",
interaction: NO_INTERACTION,
})
);
}
}
});
@@ -195,11 +201,7 @@ export class ModuleObsidianEvents extends AbstractObsidianModule {
async watchWindowVisibilityAsync() {
if (this.settings.suspendFileWatching) {
if (
this.settings.isConfigured &&
this.services.appLifecycle.isReady() &&
this.hasBoundedActivity()
) {
if (this.settings.isConfigured && this.services.appLifecycle.isReady() && this.hasBoundedActivity()) {
const isHidden = activeWindow.document.hidden;
this.isLastHidden = isHidden;
this.deferredBoundedLifecycle = isHidden ? "suspend-if-hidden" : undefined;
@@ -290,7 +292,10 @@ export class ModuleObsidianEvents extends AbstractObsidianModule {
return;
}
if (this.settings.syncOnFileOpen && !this.services.appLifecycle.isSuspended()) {
await this.services.replication.replicateByEvent();
await this.services.replication.replicateUnattendedByEvent({
trigger: "file-open",
interaction: NO_INTERACTION,
});
}
await this.services.conflict.queueCheckForIfOpen(file.path as FilePathWithPrefix);
}
@@ -2,6 +2,7 @@ import { addIcon } from "@/deps.ts";
import { $msg } from "@/common/translation";
import type { LiveSyncCore } from "@/main.ts";
import { AbstractModule } from "@/modules/AbstractModule.ts";
import { USER_INITIATED_REPLICATION_AUTHORITY } from "@vrtmrz/livesync-commonlib/replication";
// Obsidian specific menu commands.
export class ModuleObsidianMenu extends AbstractModule {
_everyOnloadStart(): Promise<boolean> {
@@ -17,7 +18,10 @@ export class ModuleObsidianMenu extends AbstractModule {
);
this.addRibbonIcon("replicate", $msg("moduleObsidianMenu.replicate"), async () => {
await this.services.replication.replicate(true);
await this.services.replication.replicateUserInitiated({
trigger: "manual",
interaction: USER_INITIATED_REPLICATION_AUTHORITY,
});
}).addClass("livesync-ribbon-replicate");
return Promise.resolve(true);
@@ -18,6 +18,7 @@ import type { LiveSyncCore } from "@/main.ts";
import { EVENT_CONFLICT_CANCELLED, EVENT_ON_UNRESOLVED_ERROR, eventHub } from "@/common/events.ts";
import { $msg } from "@/common/translation.ts";
import type { Editor, MarkdownFileInfo, MarkdownView } from "@/deps.ts";
import { NO_INTERACTION } from "@vrtmrz/livesync-commonlib/replication";
export class ModuleInteractiveConflictResolver extends AbstractObsidianModule {
private postponedConflictEpisodes = new Set<FilePathWithPrefix>();
@@ -182,7 +183,10 @@ export class ModuleInteractiveConflictResolver extends AbstractObsidianModule {
// So we have to run replication if configured.
// TODO: Make this is as a event request
if (this.settings.syncAfterMerge && !this.services.appLifecycle.isSuspended()) {
await this.services.replication.replicateByEvent();
await this.services.replication.replicateUnattendedByEvent({
trigger: "merge",
interaction: NO_INTERACTION,
});
}
// And, check it again.
await this.services.conflict.queueCheckFor(filename);
@@ -77,7 +77,9 @@ function createModule(conflictedRevisions: string[] = ["2-right"]) {
queueCheckFor: vi.fn(async () => undefined),
ensureAllProcessed: vi.fn(async () => true),
},
replication: { replicateByEvent: vi.fn(async () => true) },
replication: {
replicateUnattendedByEvent: vi.fn(async () => ({ status: "completed" as const })),
},
vault: { getActiveFilePath: vi.fn(() => path) },
path: { getPath: vi.fn((entry: { path: FilePathWithPrefix }) => entry.path) },
};