From 56444bb98b6586bf4faba28caac4a5d4e8357436 Mon Sep 17 00:00:00 2001 From: vorotamoroz Date: Thu, 27 Aug 2026 11:16:30 +0000 Subject: [PATCH] Route replication through provider capabilities --- ...plicator_capabilities_03_migration_plan.md | 12 + src/LiveSyncBaseCore.ts | 53 +++- .../cli/commands/daemonCommand.unit.spec.ts | 33 +- src/apps/cli/commands/runCommand.ts | 57 +++- src/features/ConfigSync/PluginPane.svelte | 6 +- .../AutomaticReplicationTriggers.unit.spec.ts | 287 ++++++++++++++++++ src/modules/core/ModulePeriodicProcess.ts | 23 +- .../core/ModuleReplicationLifecycle.ts | 113 +++++++ src/modules/core/ModuleReplicator.ts | 28 +- .../core/ModuleReplicator.unit.spec.ts | 54 ++++ src/modules/core/ModuleReplicatorCouchDB.ts | 50 --- .../core/ModuleReplicatorCouchDB.unit.spec.ts | 89 ------ src/modules/core/ModuleReplicatorMinIO.ts | 18 -- .../core/ReplicationLifecycle.unit.spec.ts | 200 ++++++++++++ src/modules/core/ReplicationScheduling.ts | 49 +++ .../coreFeatures/ModuleConflictResolver.ts | 6 +- .../ModuleConflictResolver.unit.spec.ts | 2 +- src/modules/essential/ModuleBasicMenu.ts | 6 +- .../essential/ModuleBasicMenu.unit.spec.ts | 2 +- .../essentialObsidian/ModuleObsidianEvents.ts | 19 +- .../essentialObsidian/ModuleObsidianMenu.ts | 6 +- .../ModuleInteractiveConflictResolver.ts | 6 +- ...leInteractiveConflictResolver.unit.spec.ts | 4 +- 23 files changed, 918 insertions(+), 205 deletions(-) create mode 100644 src/modules/core/AutomaticReplicationTriggers.unit.spec.ts create mode 100644 src/modules/core/ModuleReplicationLifecycle.ts delete mode 100644 src/modules/core/ModuleReplicatorCouchDB.ts delete mode 100644 src/modules/core/ModuleReplicatorCouchDB.unit.spec.ts delete mode 100644 src/modules/core/ModuleReplicatorMinIO.ts create mode 100644 src/modules/core/ReplicationLifecycle.unit.spec.ts create mode 100644 src/modules/core/ReplicationScheduling.ts diff --git a/docs/adr/2026_08_replicator_capabilities_03_migration_plan.md b/docs/adr/2026_08_replicator_capabilities_03_migration_plan.md index 8deec4f9..bc5eb129 100644 --- a/docs/adr/2026_08_replicator_capabilities_03_migration_plan.md +++ b/docs/adr/2026_08_replicator_capabilities_03_migration_plan.md @@ -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 diff --git a/src/LiveSyncBaseCore.ts b/src/LiveSyncBaseCore.ts index 116d7b06..76f06006 100644 --- a/src/LiveSyncBaseCore.ts +++ b/src/LiveSyncBaseCore.ts @@ -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) => 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)); diff --git a/src/apps/cli/commands/daemonCommand.unit.spec.ts b/src/apps/cli/commands/daemonCommand.unit.spec.ts index 5e448ff0..ccd18c70 100644 --- a/src/apps/cli/commands/daemonCommand.unit.spec.ts +++ b/src/apps/cli/commands/daemonCommand.unit.spec.ts @@ -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 2–7 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"); }); diff --git a/src/apps/cli/commands/runCommand.ts b/src/apps/cli/commands/runCommand.ts index 0c2e320d..eeb5bc51 100644 --- a/src/apps/cli/commands/runCommand.ts +++ b/src/apps/cli/commands/runCommand.ts @@ -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") { diff --git a/src/features/ConfigSync/PluginPane.svelte b/src/features/ConfigSync/PluginPane.svelte index 22e9db7b..cd07fd36 100644 --- a/src/features/ConfigSync/PluginPane.svelte +++ b/src/features/ConfigSync/PluginPane.svelte @@ -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++; diff --git a/src/modules/core/AutomaticReplicationTriggers.unit.spec.ts b/src/modules/core/AutomaticReplicationTriggers.unit.spec.ts new file mode 100644 index 00000000..34920adc --- /dev/null +++ b/src/modules/core/AutomaticReplicationTriggers.unit.spec.ts @@ -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 = {}) { + return { + ...DEFAULT_SETTINGS, + remoteType: REMOTE_P2P, + isConfigured: true, + ...overrides, + }; +} + +function createObsidianEventHarness(settings: Partial) { + 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> = []; + const settingRealisedHandlers: Array<() => Promise> = []; + 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) => 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) => 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()); + }); +}); diff --git a/src/modules/core/ModulePeriodicProcess.ts b/src/modules/core/ModulePeriodicProcess.ts index 9d8027bc..f7cd8486 100644 --- a/src/modules/core/ModulePeriodicProcess.ts +++ b/src/modules/core/ModulePeriodicProcess.ts @@ -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)); diff --git a/src/modules/core/ModuleReplicationLifecycle.ts b/src/modules/core/ModuleReplicationLifecycle.ts new file mode 100644 index 00000000..1e8c4d4b --- /dev/null +++ b/src/modules/core/ModuleReplicationLifecycle.ts @@ -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; + + private async runAfterResume(): Promise { + 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 { + 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)); + } +} diff --git a/src/modules/core/ModuleReplicator.ts b/src/modules/core/ModuleReplicator.ts index 4d2b352a..5bf68965 100644 --- a/src/modules/core/ModuleReplicator.ts +++ b/src/modules/core/ModuleReplicator.ts @@ -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 { + private async onReplicationFailed( + showMessageOrInteraction: boolean | ReplicationInteraction = false, + interaction?: ReplicationInteraction + ): Promise { + // 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 { diff --git a/src/modules/core/ModuleReplicator.unit.spec.ts b/src/modules/core/ModuleReplicator.unit.spec.ts index 668c94bc..9236e6d4 100644 --- a/src/modules/core/ModuleReplicator.unit.spec.ts +++ b/src/modules/core/ModuleReplicator.unit.spec.ts @@ -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", () => { diff --git a/src/modules/core/ModuleReplicatorCouchDB.ts b/src/modules/core/ModuleReplicatorCouchDB.ts deleted file mode 100644 index d96fb5c0..00000000 --- a/src/modules/core/ModuleReplicatorCouchDB.ts +++ /dev/null @@ -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 = {}): Promise { - 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 { - 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)); - } -} diff --git a/src/modules/core/ModuleReplicatorCouchDB.unit.spec.ts b/src/modules/core/ModuleReplicatorCouchDB.unit.spec.ts deleted file mode 100644 index c4296e19..00000000 --- a/src/modules/core/ModuleReplicatorCouchDB.unit.spec.ts +++ /dev/null @@ -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(); - }); -}); diff --git a/src/modules/core/ModuleReplicatorMinIO.ts b/src/modules/core/ModuleReplicatorMinIO.ts deleted file mode 100644 index 68dcc22e..00000000 --- a/src/modules/core/ModuleReplicatorMinIO.ts +++ /dev/null @@ -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 = {}): Promise { - 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)); - } -} diff --git a/src/modules/core/ReplicationLifecycle.unit.spec.ts b/src/modules/core/ReplicationLifecycle.unit.spec.ts new file mode 100644 index 00000000..9e274973 --- /dev/null +++ b/src/modules/core/ReplicationLifecycle.unit.spec.ts @@ -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; + +function createResumeHarness(settings: { + liveSync: boolean; + syncOnStart: boolean; + isConfigured?: boolean; + remoteType?: string; + P2P_Enabled?: boolean; +}) { + const resumeHandlers: ResumeHandler[] = []; + const replicateUnattended = vi.fn(async (): Promise => ({ status: "completed" })); + const startContinuous = vi.fn(async (): Promise => ({ 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, + }); + }); +}); diff --git a/src/modules/core/ReplicationScheduling.ts b/src/modules/core/ReplicationScheduling.ts new file mode 100644 index 00000000..21fc48ac --- /dev/null +++ b/src/modules/core/ReplicationScheduling.ts @@ -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(); + +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; +} diff --git a/src/modules/coreFeatures/ModuleConflictResolver.ts b/src/modules/coreFeatures/ModuleConflictResolver.ts index 85d060a1..0e952dd0 100644 --- a/src/modules/coreFeatures/ModuleConflictResolver.ts +++ b/src/modules/coreFeatures/ModuleConflictResolver.ts @@ -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); diff --git a/src/modules/coreFeatures/ModuleConflictResolver.unit.spec.ts b/src/modules/coreFeatures/ModuleConflictResolver.unit.spec.ts index 18bfb618..630d298e 100644 --- a/src/modules/coreFeatures/ModuleConflictResolver.unit.spec.ts +++ b/src/modules/coreFeatures/ModuleConflictResolver.unit.spec.ts @@ -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), diff --git a/src/modules/essential/ModuleBasicMenu.ts b/src/modules/essential/ModuleBasicMenu.ts index 06d27450..5bedf7b6 100644 --- a/src/modules/essential/ModuleBasicMenu.ts +++ b/src/modules/essential/ModuleBasicMenu.ts @@ -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({ diff --git a/src/modules/essential/ModuleBasicMenu.unit.spec.ts b/src/modules/essential/ModuleBasicMenu.unit.spec.ts index af5871d3..7f670aa0 100644 --- a/src/modules/essential/ModuleBasicMenu.unit.spec.ts +++ b/src/modules/essential/ModuleBasicMenu.unit.spec.ts @@ -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"), diff --git a/src/modules/essentialObsidian/ModuleObsidianEvents.ts b/src/modules/essentialObsidian/ModuleObsidianEvents.ts index d266dd72..5f689bc6 100644 --- a/src/modules/essentialObsidian/ModuleObsidianEvents.ts +++ b/src/modules/essentialObsidian/ModuleObsidianEvents.ts @@ -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); } diff --git a/src/modules/essentialObsidian/ModuleObsidianMenu.ts b/src/modules/essentialObsidian/ModuleObsidianMenu.ts index 911d5c6e..f11d8b9b 100644 --- a/src/modules/essentialObsidian/ModuleObsidianMenu.ts +++ b/src/modules/essentialObsidian/ModuleObsidianMenu.ts @@ -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 { @@ -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); diff --git a/src/modules/features/ModuleInteractiveConflictResolver.ts b/src/modules/features/ModuleInteractiveConflictResolver.ts index 089aa625..27e188ee 100644 --- a/src/modules/features/ModuleInteractiveConflictResolver.ts +++ b/src/modules/features/ModuleInteractiveConflictResolver.ts @@ -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(); @@ -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); diff --git a/src/modules/features/ModuleInteractiveConflictResolver.unit.spec.ts b/src/modules/features/ModuleInteractiveConflictResolver.unit.spec.ts index 82c2a8ef..b3a46f0d 100644 --- a/src/modules/features/ModuleInteractiveConflictResolver.unit.spec.ts +++ b/src/modules/features/ModuleInteractiveConflictResolver.unit.spec.ts @@ -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) }, };