diff --git a/devs.md b/devs.md index 114f5824..2367f2a5 100644 --- a/devs.md +++ b/devs.md @@ -135,16 +135,22 @@ The application is composed from Services, ServiceModules, serviceFeatures, and - **Service Hub**: the long-lived registry of service contracts. A simple extension, such as a check before replication, belongs in an existing Service handler. - **ServiceModule**: a host-created, long-lived stateful or resource-owning capability shared through the typed `ServiceModules` record. Current examples include storage access, file handling, and database rebuilding. -- **serviceFeature**: a typed composition function which accepts only its declared Services and ServiceModules. It registers lifecycle handlers, commands, UI bindings, or other host glue, and may return a focused view or controller. It is not a runtime registry entry. +- **serviceFeature**: a typed composition function which accepts only its declared Services and ServiceModules. It registers lifecycle handlers, commands, UI bindings, or other host glue, and may return a focused view. It is not a runtime registry entry. - **AbstractModule** and **AbstractObsidianModule**: the legacy application module layer. Existing modules are loaded by the application and bound after the Service graph has been composed; this broad core access is not the preferred dependency boundary for new orchestration. The normal composition order is the Service Hub, replicator-provider registration, ServiceModules, serviceFeatures, add-ons, and finally legacy module binding. A serviceFeature may therefore consume an already constructed ServiceModule. Preferring a serviceFeature for new composition is a dependency-boundary rule, not an initialisation-order rule. -Mutable state is permitted in a serviceFeature. State alone is not a reason to create a ServiceModule or retain an AbstractModule. Separate the component which owns state, transitions, and invariants from the surrounding function which registers lifecycle handlers and connects downstream effects. Give the stateful component narrow collaborators rather than `LiveSyncBaseCore`. Use a ServiceModule when the same operational capability or resource lifecycle must be shared explicitly by several consumers. +Mutable state is permitted in a serviceFeature. State alone is not a reason to create a class, a ServiceModule, or retain an AbstractModule. Prefer one private context, with module-level functions which receive that context, when identity and polymorphism are not part of the contract. Separate the state, transitions, and invariants from the surrounding function which registers lifecycle handlers and connects downstream effects. Give the stateful boundary narrow collaborators rather than `LiveSyncBaseCore`. + +Use a class when stable object identity, replaceable implementations, or an explicit external-resource lifecycle such as serialised ownership, `dispose()`, or `abort()` is part of the contract. Use a ServiceModule when that operational capability or resource lifecycle must also be shared explicitly by several consumers. Do not introduce a class merely to group dependencies or make private functions callable. + +Several narrow views over one lifetime do not require several state owners or a public façade class. One private context may back all of those views, provided that the context remains private and each consumer receives only its declared contract. Keep actual resource owners separate when identity, serialised replacement, abort, retirement, or disposal order is part of their behaviour. + +When a core-owned serviceFeature returns a view needed by one host-specific consumer, pass that view through host composition instead of storing it as a public `LiveSyncBaseCore` property or promoting it to a ServiceModule. The receiving host should inject the view into the narrow command or application context which uses it. Commonlib's `targetFilter.ts` and `prepareDatabaseForUse.ts` demonstrate the intended split: focused factories or operations own their private state and behaviour, while the corresponding `use...` function composes dependencies and registers handlers. The P2P composition follows the same direction at a larger scale by separating durable policy and room-session ownership from host lifecycle and UI wiring. Existing modules do not apply this boundary consistently; improve the affected boundary when changing their behaviour rather than performing an unrelated mechanical conversion. -Use interaction-based, London School unit tests for the composition boundary. Verify collaborator calls, ordering, failure short-circuiting, and handler registration, then test the focused state owner for its transitions and invariants. If a test needs a broad core fixture, deep mock chains, or unrelated Services, treat that friction as a design-review signal before adding more test machinery. +Use interaction-based, London School unit tests for the composition boundary. Verify collaborator calls, ordering, failure short-circuiting, and handler registration, then test the focused state owner for its transitions and invariants. If a test needs a broad core fixture, a large class mock, deep mock chains, or unrelated Services, treat that friction as a design-review signal and consider a private context with narrower functions before adding more test machinery. Legacy modules remain grouped by directory: diff --git a/docs/adr/2026_08_replicator_capabilities_01_core_contract.md b/docs/adr/2026_08_replicator_capabilities_01_core_contract.md index fbe3da8d..74c6e91e 100644 --- a/docs/adr/2026_08_replicator_capabilities_01_core_contract.md +++ b/docs/adr/2026_08_replicator_capabilities_01_core_contract.md @@ -131,38 +131,41 @@ that request can be performed. A provider module must not subscribe to the application resume lifecycle merely because it can construct a transport. Self-hosted LiveSync will compose one LiveSync-owned serviceFeature as the -replication scheduling boundary. The serviceFeature constructs one private -scheduling controller and connects it to `AppLifecycleService`, settings -lifecycle events, and the periodic timer. The controller owns only scheduling -state and transitions: external-poller ownership, Continuous ownership of -recurring work, the daemon's satisfied initial OneShot marker, resume -coalescing, and periodic-timer reconciliation. It does not register handlers -or acquire `LiveSyncBaseCore`. +replication scheduling boundary. The serviceFeature creates one private +scheduling context and passes it to module-level transition functions. The +context contains only scheduling state and narrow collaborators; the functions +implement external-poller ownership, Continuous ownership of recurring work, +the daemon's satisfied initial OneShot marker, resume coalescing, and +periodic-timer reconciliation. They do not register handlers or acquire +`LiveSyncBaseCore`. -The controller receives narrow collaborators for readiness and suspension -queries, current settings, `ReplicationService`, periodic-timer control, and -diagnostic logging. The surrounding serviceFeature owns lifecycle registration -and adapts those Services to the controller. It returns a focused control view -containing only the daemon operations to select external polling and mark the -initial OneShot as satisfied. The host may retain that view for the CLI, but -must not expose the controller's mutable state or recover it from a core-keyed -global or `WeakMap`. +The context receives narrow collaborators for readiness and suspension queries, +current settings, `ReplicationService`, periodic-timer control, and diagnostic +logging. The surrounding serviceFeature owns the context lifetime, lifecycle +registration, and adaptation from those Services. It returns a focused control +view containing only the daemon operations to select external polling and mark +the initial OneShot as satisfied. Core construction passes a frozen bundle of +built-in feature views to host composition, without retaining those views as +public `LiveSyncBaseCore` properties. The CLI injects the scheduling view into +its command context; other hosts may ignore it. No host may expose the context's +mutable state or recover it from a core-keyed global or `WeakMap`. This boundary is not a ServiceModule merely because it owns state. It neither owns a shared external resource nor supplies a general operational capability to several unrelated consumers. If a future consumer needs a stable shared scheduling capability beyond the focused CLI view, that ownership decision -must be reviewed explicitly rather than widening the controller implicitly. +must be reviewed explicitly rather than widening the returned view implicitly. -The scheduling controller uses persisted settings, `ReplicationService`, and -the active support declaration. It does not branch on `remoteType` or use +The scheduling functions use persisted settings, `ReplicationService`, and +the active support declaration. They do not branch on `remoteType` or use `instanceof` as a capability test. Commonlib owns the trigger-aware replication contract; the host owns application lifecycle wiring. The existing `onResumed` event remains the eligible-resume boundary after initial readiness, settings application, and visibility recovery. It is not -redefined as a once-per-process event. The controller coalesces duplicate work -within one lifecycle generation and preserves readiness and suspension gates: +redefined as a once-per-process event. The context-backed functions coalesce +duplicate work within one lifecycle generation and preserve readiness and +suspension gates: - configured Continuous replication starts only through an active Continuous role; @@ -177,7 +180,7 @@ within one lifecycle generation and preserves readiness and suspension gates: `ReplicationService` remains responsible for readiness checks, bounded finite activity, failure processing, and replication timing. It exposes distinct user-initiated and unattended entry points, or a typed request which carries -interaction authority. The scheduling controller never calls a concrete +interaction authority. The scheduling functions never call a concrete Replicator's `openReplication()` directly. `P2P_AutoStart` remains a separate P2P room policy. It is not central @@ -199,17 +202,29 @@ The CLI daemon owns its initial finite convergence before its mirror scan. Restored settings mark that convergence as satisfied for the current lifecycle generation, so `syncOnStart` does not repeat it. In `--interval` mode, the daemon poller is the sole recurring remote-poll scheduler. In changes-feed mode, -the controller starts one configured Continuous session when supported; -otherwise it may enable the configured generic periodic timer. Continuous has -precedence when both are configured. +the scheduling functions start one configured Continuous session when +supported; otherwise they may enable the configured generic periodic timer. +Continuous has precedence when both are configured. -The controller starts resume work synchronously far enough to reserve +The resume function starts work synchronously far enough to reserve Continuous ownership, then lets the lifecycle handler settle without awaiting network completion. Concurrent resume notifications share one internal operation. Periodic reconciliation therefore observes the reservation before it can enable a competing timer. A failed operation is logged and releases the coalescing slot so a later resume can retry. +Coalescing applies only within one observed lifecycle generation. If the +application suspends and resumes while an earlier operation is still settling, +the context retains the newer generation and runs it after the earlier +operation releases the slot. A result from the obsolete generation cannot +change recurring-work ownership or initiate a OneShot fallback for the newer +generation. + +Disabling an interval does not retract a callback which the runtime has already +queued. Each Periodic callback therefore rechecks lifecycle eligibility, +readiness, suspension, configuration, external-poller ownership, and +Continuous ownership immediately before it requests replication. + ### Use a fixed current-provider definition Commonlib defines the canonical current remote kinds, provider contract, 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 1b8686cb..fa35f70f 100644 --- a/docs/adr/2026_08_replicator_capabilities_03_migration_plan.md +++ b/docs/adr/2026_08_replicator_capabilities_03_migration_plan.md @@ -63,8 +63,9 @@ actual journal result. Add the LiveSync-owned replication scheduling serviceFeature, remove the resume handler from `ModuleReplicatorCouchDB`, and route CouchDB Continuous and OneShot Sync plus Object Storage `syncOnStart` through `ReplicationService`. Its private -controller owns scheduling state and transition order; the serviceFeature owns -lifecycle and settings-handler registration. Migrate every automatic caller to +context owns scheduling state, module-level functions implement transitions, +and the serviceFeature owns lifecycle and settings-handler registration. +Migrate every automatic caller to the unattended entry point in this stage, so periodic and event calls cannot fall back to an interactive P2P role. Migrate manual commands to the user-initiated entry point. @@ -81,12 +82,14 @@ 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 scheduling controller coalesces its network work internally, but an +The scheduling context 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. Compose the controller with narrow replication, settings, lifecycle, -timer, and logging collaborators. Return only the daemon-facing control view; -do not retain scheduling state in a core-keyed `WeakMap`. +settled. Pass one private context with narrow replication, settings, lifecycle, +timer, and logging collaborators to module-level functions. Return only the +daemon-facing control view, pass it to host composition, and inject it into the +CLI command context. Do not retain the view as a public `LiveSyncBaseCore` +property or retain scheduling state in a core-keyed `WeakMap`. At this boundary, existing P2P AutoSync, AutoWatch, and incoming-request entry points receive the same non-interactive readiness and accepted-peer gate. @@ -107,9 +110,9 @@ AutoWatch, and accepted incoming-request paths continue with the Stage 2 gate. This is a temporary migration state, not the target matrix in Part 1. Apply and test the CLI scheduling precedence defined in Part 1, so the daemon -and scheduling controller cannot schedule duplicate initial or recurring work. +and scheduling context cannot schedule duplicate initial or recurring work. Replace `ModuleReplicationLifecycle` and the replication-specific -`ModulePeriodicProcess` wiring only after equivalent controller and feature- +`ModulePeriodicProcess` wiring only after equivalent context and feature- binding tests pass. Reuse the existing timer implementation behind a narrow timer port; changing other periodic feature owners is outside this stage. @@ -280,6 +283,12 @@ Cover: - Object Storage `syncOnStart` through resume, including a migrated profile which retains `liveSync: true`; - existing CouchDB Continuous and OneShot paths; +- same-generation resume coalescing, a fresh attempt after a later lifecycle + generation, and rejection of an obsolete generation's OneShot fallback; +- a queued Periodic callback rechecking lifecycle and recurring-work ownership + after its interval has been disabled; +- the daemon's satisfied initial OneShot marker being consumed even when a + Continuous start throws, so a later resume may retry normally; - periodic, database-save, editor-save, file-open, merge, and daemon triggers remaining free of dialogues; - manual P2P and configured peer-targeted flows remaining available; diff --git a/src/LiveSyncBaseCore.ts b/src/LiveSyncBaseCore.ts index 12168bec..2e1025f3 100644 --- a/src/LiveSyncBaseCore.ts +++ b/src/LiveSyncBaseCore.ts @@ -24,9 +24,7 @@ import { useRemoteConfigurationMigration } from "@vrtmrz/livesync-commonlib/comp import type { ServiceContext } from "@vrtmrz/livesync-commonlib/context"; import type { InjectableServiceHub } from "@vrtmrz/livesync-commonlib/compat/services/implements/injectable/InjectableServiceHub"; import { AbstractModule } from "./modules/AbstractModule"; -import { ModulePeriodicProcess } from "./modules/core/ModulePeriodicProcess"; import { ModuleReplicator } from "./modules/core/ModuleReplicator"; -import { ModuleReplicationLifecycle } from "./modules/core/ModuleReplicationLifecycle"; import { ModuleConflictChecker } from "./modules/coreFeatures/ModuleConflictChecker"; import { ModuleConflictResolver } from "./modules/coreFeatures/ModuleConflictResolver"; import { ModuleResolvingMismatchedTweaks } from "./modules/coreFeatures/ModuleResolveMismatchedTweaks"; @@ -46,6 +44,12 @@ import { } 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"; +import { useReplicationScheduling, type ReplicationSchedulingControl } from "./serviceFeatures/replicationScheduling"; + +/** Focused views returned by serviceFeatures which the host may consume during composition. */ +export interface LiveSyncCoreFeatureViews { + readonly replicationScheduling: ReplicationSchedulingControl; +} export class LiveSyncBaseCore< T extends ServiceContext = ServiceContext, @@ -90,15 +94,15 @@ export class LiveSyncBaseCore< ) => ServiceModules, extraModuleInitialiser: (core: LiveSyncBaseCore) => AbstractModule[], addOnsInitialiser: (core: LiveSyncBaseCore) => TCommands[], - featuresInitialiser: (core: LiveSyncBaseCore) => void + featuresInitialiser: (core: LiveSyncBaseCore, coreFeatureViews: LiveSyncCoreFeatureViews) => void ) { this._services = serviceHub; this.registerReplicatorProviders(); this._serviceModules = serviceModuleInitialiser(this, serviceHub); const extraModules = extraModuleInitialiser(this); this.registerModules(extraModules); - this.initialiseServiceFeatures(); - featuresInitialiser(this); + const coreFeatureViews = this.initialiseServiceFeatures(); + featuresInitialiser(this, coreFeatureViews); const addOns = addOnsInitialiser(this); for (const addOn of addOns) { this._registerAddOn(addOn); @@ -190,9 +194,7 @@ export class LiveSyncBaseCore< this._registerModule(new ModuleLiveSyncMain(this)); this._registerModule(new ModuleConflictChecker(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)); this._registerModule(new ModuleBasicMenu(this)); @@ -322,12 +324,15 @@ export class LiveSyncBaseCore< * Initialise ServiceFeatures. * (Please refer `serviceFeatures` for more details) */ - initialiseServiceFeatures() { + initialiseServiceFeatures(): LiveSyncCoreFeatureViews { useTargetFilters(this); // enable target filter feature. usePrepareDatabaseForUse(this); // Migration to multiple remote configurations useRemoteConfigurationMigration(this); + return Object.freeze({ + replicationScheduling: useReplicationScheduling(this), + }); } } diff --git a/src/apps/cli/commands/daemonCommand.unit.spec.ts b/src/apps/cli/commands/daemonCommand.unit.spec.ts index ccd18c70..b99d2176 100644 --- a/src/apps/cli/commands/daemonCommand.unit.spec.ts +++ b/src/apps/cli/commands/daemonCommand.unit.spec.ts @@ -19,7 +19,6 @@ 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 = { @@ -89,6 +88,17 @@ const baseContext = { }, } as any; +function createDaemonContext(core: ReturnType) { + return { + ...baseContext, + core, + replicationScheduling: { + setExternalPollingMode: vi.fn(), + markInitialOneShotSatisfied: vi.fn(), + }, + } as any; +} + describe("daemon command", () => { beforeEach(() => { vi.restoreAllMocks(); @@ -103,7 +113,7 @@ describe("daemon command", () => { const core = createCoreMock(); vi.mocked(offlineScanner.performFullScan).mockResolvedValue(true); - await runCommand(makeDaemonOptions(), { ...baseContext, core }); + await runCommand(makeDaemonOptions(), createDaemonContext(core)); expect(offlineScanner.performFullScan).toHaveBeenCalledTimes(1); }); @@ -112,7 +122,7 @@ describe("daemon command", () => { const core = createCoreMock(); vi.mocked(offlineScanner.performFullScan).mockResolvedValue(false); - const result = await runCommand(makeDaemonOptions(), { ...baseContext, core }); + const result = await runCommand(makeDaemonOptions(), createDaemonContext(core)); expect(result).toBe(false); }); @@ -122,10 +132,11 @@ describe("daemon command", () => { vi.mocked(offlineScanner.performFullScan).mockResolvedValue(true); const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); - await runCommand(makeDaemonOptions(30), { ...baseContext, core }); + const context = createDaemonContext(core); + await runCommand(makeDaemonOptions(30), context); expect(setTimeoutSpy).toHaveBeenCalledTimes(1); - expect(getReplicationSchedulingControl(core).externalPolling).toBe(true); + expect(context.replicationScheduling.setExternalPollingMode).toHaveBeenCalledWith(true); // Interval should be in milliseconds (30s → 30000ms) expect(setTimeoutSpy).toHaveBeenCalledWith(expect.any(Function), 30000); }); @@ -134,7 +145,7 @@ describe("daemon command", () => { const core = createCoreMock(); vi.mocked(offlineScanner.performFullScan).mockResolvedValue(true); - await runCommand(makeDaemonOptions(10), { ...baseContext, core }); + await runCommand(makeDaemonOptions(10), createDaemonContext(core)); expect(core.services.setting.applyPartial).toHaveBeenCalledWith( expect.objectContaining({ suspendFileWatching: false }), @@ -147,7 +158,7 @@ describe("daemon command", () => { const core = createCoreMock(); vi.mocked(offlineScanner.performFullScan).mockResolvedValue(true); - await runCommand(makeDaemonOptions(), { ...baseContext, core }); + await runCommand(makeDaemonOptions(), createDaemonContext(core)); expect(core.services.setting.applyPartial).toHaveBeenCalledWith( expect.objectContaining({ @@ -167,7 +178,7 @@ describe("daemon command", () => { })); vi.mocked(offlineScanner.performFullScan).mockResolvedValue(true); - const result = await runCommand(makeDaemonOptions(), { ...baseContext, core }); + const result = await runCommand(makeDaemonOptions(), createDaemonContext(core)); expect(result).toBe(true); const warningCalls = core.services.context.standardIo.writeStderr.mock.calls.filter( @@ -185,7 +196,7 @@ describe("daemon command", () => { })); vi.mocked(offlineScanner.performFullScan).mockResolvedValue(true); - await runCommand(makeDaemonOptions(), { ...baseContext, core }); + await runCommand(makeDaemonOptions(), createDaemonContext(core)); const warningCalls = core.services.context.standardIo.writeStderr.mock.calls.filter( ([chunk]: [string | Uint8Array]) => @@ -206,14 +217,15 @@ describe("daemon command", () => { return true; }); - await runCommand(makeDaemonOptions(), { ...baseContext, core }); + const context = createDaemonContext(core); + await runCommand(makeDaemonOptions(), context); expect(callOrder).toEqual(["replicate", "performFullScan"]); expect(core.services.replication.replicateUnattended).toHaveBeenCalledWith({ trigger: "daemon", interaction: NO_INTERACTION, }); - expect(getReplicationSchedulingControl(core).initialOneShotSatisfied).toBe(true); + expect(context.replicationScheduling.markInitialOneShotSatisfied).toHaveBeenCalledOnce(); }); it("returns false when initial replication fails", async () => { @@ -224,7 +236,7 @@ describe("daemon command", () => { })); vi.mocked(offlineScanner.performFullScan).mockClear(); - const result = await runCommand(makeDaemonOptions(), { ...baseContext, core }); + const result = await runCommand(makeDaemonOptions(), createDaemonContext(core)); expect(result).toBe(false); // performFullScan should NOT have been called @@ -239,7 +251,7 @@ describe("daemon command", () => { const core = createCoreMock(); vi.mocked(offlineScanner.performFullScan).mockResolvedValue(true); - await runCommand(makeDaemonOptions(10), { ...baseContext, core }); + await runCommand(makeDaemonOptions(10), createDaemonContext(core)); // onUnload handler should have been registered expect(core.services.appLifecycle.onUnload.addHandler).toHaveBeenCalledTimes(1); @@ -267,7 +279,7 @@ describe("daemon command", () => { const baseMs = 30 * 1000; const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); - await runCommand(makeDaemonOptions(30), { ...baseContext, core }); + await runCommand(makeDaemonOptions(30), createDaemonContext(core)); // After runCommand returns the first setTimeout has been scheduled. // setTimeoutSpy.mock.calls[0] is the initial schedule (baseMs). @@ -319,7 +331,7 @@ describe("daemon command", () => { }); const intervalMs = 30 * 1000; - await runCommand(makeDaemonOptions(30), { ...baseContext, core }); + await runCommand(makeDaemonOptions(30), createDaemonContext(core)); // Advance time to trigger the first poll callback and flush its async work. await vi.advanceTimersByTimeAsync(intervalMs); diff --git a/src/apps/cli/commands/runCommand.ts b/src/apps/cli/commands/runCommand.ts index 2d4a4a68..532e1c82 100644 --- a/src/apps/cli/commands/runCommand.ts +++ b/src/apps/cli/commands/runCommand.ts @@ -31,7 +31,6 @@ import { 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, "//***@"); @@ -93,7 +92,7 @@ async function verifyRemoteState( } export async function runCommand(options: CLIOptions, context: CLICommandContext): Promise { - const { databasePath, core, settingsPath } = context; + const { databasePath, core, replicationScheduling, settingsPath } = context; const { standardIo } = core.services.context; const vaultPath = context.vaultPath || databasePath; @@ -103,7 +102,7 @@ export async function runCommand(options: CLIOptions, context: CLICommandContext // The daemon owns its own recurring poller. Suppress the application // resume starter and generic periodic timer before restoring settings. - setExternalPollingMode(core, !!options.interval); + replicationScheduling.setExternalPollingMode(!!options.interval); // Skip the config mismatch dialog — the daemon cannot resolve it interactively // and the default "Dismiss" action would block replication. The daemon should @@ -121,7 +120,7 @@ export async function runCommand(options: CLIOptions, context: CLICommandContext writeStderrLine(standardIo, "[Daemon] Initial replication failed, cannot continue"); return false; } - markInitialOneShotSatisfied(core); + replicationScheduling.markInitialOneShotSatisfied(); log("Initial replication complete"); // 2. Mirror scan to reconcile PouchDB ↔ local filesystem. @@ -144,7 +143,7 @@ export async function runCommand(options: CLIOptions, context: CLICommandContext true ); // applySettings fires the full lifecycle: onSuspending → onResumed. - // The provider-independent lifecycle coordinator owns any eligible + // The provider-independent scheduling feature owns any eligible // Continuous start; the daemon marker suppresses a duplicate // sync-on-start OneShot. await core.services.control.applySettings(); @@ -207,7 +206,7 @@ export async function runCommand(options: CLIOptions, context: CLICommandContext log("LiveSync mode: restoring sync settings and starting continuous synchronisation where supported"); await restoreSyncSettings(); // The applySettings() lifecycle fires onResumed → the provider- - // independent lifecycle coordinator, which starts Continuous when + // independent scheduling feature, which starts Continuous when // supported. Do not call a concrete Replicator directly. log("LiveSync active"); const currentSettings = core.services.setting.currentSettings(); diff --git a/src/apps/cli/commands/types.ts b/src/apps/cli/commands/types.ts index 3b87a51c..8ea537f9 100644 --- a/src/apps/cli/commands/types.ts +++ b/src/apps/cli/commands/types.ts @@ -2,6 +2,7 @@ import { LiveSyncBaseCore } from "@/LiveSyncBaseCore"; import type { ObsidianLiveSyncSettings } from "@vrtmrz/livesync-commonlib/compat/common/types"; import type { NodeServiceContext } from "@/apps/cli/services/NodeServiceContext"; import type { UseP2PReplicatorResult } from "@vrtmrz/livesync-commonlib/p2p"; +import type { ReplicationSchedulingControl } from "@/serviceFeatures/replicationScheduling"; export type CLICommand = | "daemon" @@ -50,6 +51,8 @@ export interface CLICommandContext { databasePath: string; vaultPath: string; core: LiveSyncBaseCore; + /** Host-composition view used only to coordinate daemon-owned recurring work. */ + replicationScheduling: ReplicationSchedulingControl; /** Current-result contract owned by the P2P service feature. */ p2pReplicator?: UseP2PReplicatorResult; settingsPath: string; diff --git a/src/apps/cli/main.ts b/src/apps/cli/main.ts index 240d7694..69d5bfc8 100644 --- a/src/apps/cli/main.ts +++ b/src/apps/cli/main.ts @@ -24,6 +24,7 @@ import { getPathFromUXFileInfo } from "@vrtmrz/livesync-commonlib/compat/common/ import { stripAllPrefixes } from "@vrtmrz/livesync-commonlib/compat/string_and_binary/path"; import { IgnoreRules } from "./serviceModules/IgnoreRules"; import { useP2PReplicatorFeature, type UseP2PReplicatorResult } from "@vrtmrz/livesync-commonlib/p2p"; +import type { ReplicationSchedulingControl } from "@/serviceFeatures/replicationScheduling"; import { createNodeStandardIo, fsPromises as fs, path } from "@vrtmrz/livesync-commonlib/node"; import type { StandardIo } from "@vrtmrz/livesync-commonlib/context"; import { writeStderrLine, writeStdoutLine } from "./cliOutput"; @@ -477,6 +478,7 @@ export async function main( // Create LiveSync core let p2pReplicator: UseP2PReplicatorResult | undefined; + let replicationScheduling: ReplicationSchedulingControl | undefined; const core = new LiveSyncBaseCore( serviceHubInstance, (core: LiveSyncBaseCore, serviceHub: InjectableServiceHub) => { @@ -484,7 +486,8 @@ export async function main( }, (core) => [], () => [], // No add-ons - (core) => { + (core, coreFeatureViews) => { + replicationScheduling = coreFeatureViews.replicationScheduling; // Register P2P replicator feature. p2pReplicator = useP2PReplicatorFeature(core); // Add target filter to prevent internal files are handled @@ -516,6 +519,9 @@ export async function main( } } ); + if (!replicationScheduling) { + throw new Error("Replication scheduling was not provided during core feature composition."); + } // Setup signal handlers for graceful shutdown const shutdown = async (signal: string) => { @@ -622,6 +628,7 @@ export async function main( databasePath, vaultPath, core, + replicationScheduling, p2pReplicator, settingsPath, originalSyncSettings, diff --git a/src/modules/core/AutomaticReplicationTriggers.unit.spec.ts b/src/modules/core/AutomaticReplicationTriggers.unit.spec.ts index 34920adc..713cdb5a 100644 --- a/src/modules/core/AutomaticReplicationTriggers.unit.spec.ts +++ b/src/modules/core/AutomaticReplicationTriggers.unit.spec.ts @@ -16,8 +16,12 @@ 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 { + createReplicationSchedulingContext, + realiseReplicationScheduling, + resumeReplicationScheduling, + runPeriodicReplication, +} from "@/serviceFeatures/replicationScheduling"; import { ModuleReplicator } from "./ModuleReplicator"; function createApi() { @@ -97,11 +101,20 @@ describe("automatic replication triggers while P2P is active", () => { const core = { _services: services, services, - settings: p2pSettings({ periodicReplication: true }), + settings: p2pSettings({ periodicReplication: true, syncOnStart: false }), } as any; - const module = new ModulePeriodicProcess(core); + const context = createReplicationSchedulingContext({ + isReady: vi.fn(() => true), + isSuspended: vi.fn(() => false), + currentSettings: vi.fn(() => core.settings), + replicateUnattended, + startContinuous: vi.fn(async () => ({ status: "completed" as const })), + timer: { enable: vi.fn(), disable: vi.fn() }, + log: vi.fn(), + }); - await module.periodicSyncProcessor.process(); + resumeReplicationScheduling(context); + await runPeriodicReplication(context); expect(replicateUnattended).toHaveBeenCalledOnce(); expect(replicateUnattended).toHaveBeenCalledWith({ @@ -203,8 +216,6 @@ describe("recurring replication scheduling precedence", () => { }); function createRecurringSchedulingHarness() { - const resumeHandlers: Array<() => Promise> = []; - const settingRealisedHandlers: Array<() => Promise> = []; let resolveContinuous!: ( outcome: { status: "completed" } | { status: "blocked"; reason: "capability-not-applicable" } ) => void; @@ -225,42 +236,36 @@ describe("recurring replication scheduling precedence", () => { 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 context = createReplicationSchedulingContext({ + isReady: vi.fn(() => true), + isSuspended: vi.fn(() => false), + currentSettings: vi.fn(() => settings), + startContinuous, + replicateUnattended: vi.fn(async () => ({ status: "completed" as const })), + timer: { + enable: (interval) => { + API.setInterval(vi.fn(), interval); + }, + disable: () => { + API.clearInterval(0); }, }, - }; - 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); + log: vi.fn(), + }); 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())), + resume: async () => { + resumeReplicationScheduling(context); + await Promise.resolve(); + }, + realiseSettings: async () => { + realiseReplicationScheduling(context); + await Promise.resolve(); + }, }; } diff --git a/src/modules/core/ModulePeriodicProcess.ts b/src/modules/core/ModulePeriodicProcess.ts deleted file mode 100644 index f7cd8486..00000000 --- a/src/modules/core/ModulePeriodicProcess.ts +++ /dev/null @@ -1,62 +0,0 @@ -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 { - 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 - ); - return Promise.resolve(true); - } - private _allOnUnload() { - return this.disablePeriodic(); - } - private _everyBeforeRealizeSetting(): Promise { - return this.disablePeriodic(); - } - private _everyBeforeSuspendProcess(): Promise { - return this.disablePeriodic(); - } - private _everyAfterResumeProcess(): Promise { - return this.resumePeriodic(); - } - private _everyAfterRealizeSetting(): Promise { - return this.resumePeriodic(); - } - - 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)); - services.appLifecycle.onSuspending.addHandler(this._everyBeforeSuspendProcess.bind(this)); - services.appLifecycle.onResumed.addHandler(this._everyAfterResumeProcess.bind(this)); - } -} diff --git a/src/modules/core/ModuleReplicationLifecycle.ts b/src/modules/core/ModuleReplicationLifecycle.ts deleted file mode 100644 index 1e8c4d4b..00000000 --- a/src/modules/core/ModuleReplicationLifecycle.ts +++ /dev/null @@ -1,113 +0,0 @@ -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/ReplicationLifecycle.unit.spec.ts b/src/modules/core/ReplicationLifecycle.unit.spec.ts deleted file mode 100644 index 9e274973..00000000 --- a/src/modules/core/ReplicationLifecycle.unit.spec.ts +++ /dev/null @@ -1,200 +0,0 @@ -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 deleted file mode 100644 index 21fc48ac..00000000 --- a/src/modules/core/ReplicationScheduling.ts +++ /dev/null @@ -1,49 +0,0 @@ -/** - * 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/serviceFeatures/replicationScheduling.ts b/src/serviceFeatures/replicationScheduling.ts new file mode 100644 index 00000000..2c598982 --- /dev/null +++ b/src/serviceFeatures/replicationScheduling.ts @@ -0,0 +1,339 @@ +import { LOG_LEVEL_VERBOSE } from "octagonal-wheels/common/logger"; +import type { ObsidianLiveSyncSettings } from "@vrtmrz/livesync-commonlib/compat/common/types"; +import type { NecessaryServices } from "@vrtmrz/livesync-commonlib/compat/interfaces/ServiceModule"; +import { createInstanceLogFunction } from "@vrtmrz/livesync-commonlib/compat/services/lib/logUtils"; +import { + isReplicationCompleted, + NO_INTERACTION, + type ContinuousReplicationRequest, + type ReplicationOutcome, + type UnattendedOneShotRequest, +} from "@vrtmrz/livesync-commonlib/replication"; +import { PeriodicProcessor } from "@/common/PeriodicProcessor"; + +type ReplicationSchedulingSettings = Pick< + ObsidianLiveSyncSettings, + "isConfigured" | "liveSync" | "syncOnStart" | "periodicReplication" | "periodicReplicationInterval" +>; + +/** Timer operations required by the scheduling state owner. */ +export interface ReplicationSchedulingTimer { + enable(intervalMs: number): void; + disable(): void; +} + +/** Daemon-only controls which do not expose mutable scheduling state. */ +export interface ReplicationSchedulingControl { + /** Let an external daemon poller become, or cease to be, the recurring-work owner. */ + setExternalPollingMode(enabled: boolean): void; + /** Consume the next resume-triggered OneShot because the daemon has already converged once. */ + markInitialOneShotSatisfied(): void; +} + +interface ReplicationSchedulingDependencies { + isReady(): boolean; + isSuspended(): boolean; + currentSettings(): ReplicationSchedulingSettings; + replicateUnattended(request: UnattendedOneShotRequest): Promise; + startContinuous(request: ContinuousReplicationRequest): Promise; + timer: ReplicationSchedulingTimer; + log(error: unknown): void; +} + +interface ReplicationSchedulingState { + externalPolling: boolean; + continuousOwnsRecurring: boolean; + initialOneShotSatisfied: boolean; + lifecycleAllowsScheduling: boolean; + lifecycleGeneration: number; + resumeOperation: Promise | undefined; + runningResumeGeneration: number | undefined; + queuedResumeGeneration: number | undefined; +} + +/** Private state and collaborators owned by the replication scheduling serviceFeature. */ +interface ReplicationSchedulingContext { + readonly dependencies: ReplicationSchedulingDependencies; + readonly state: ReplicationSchedulingState; +} + +function isCapabilityUnavailable(result: ReplicationOutcome): boolean { + return ( + result.status === "blocked" && + (result.reason === "capability-not-applicable" || result.reason === "capability-not-implemented") + ); +} + +/** Construct the independently testable context owned by the serviceFeature. */ +export function createReplicationSchedulingContext( + dependencies: ReplicationSchedulingDependencies +): ReplicationSchedulingContext { + return { + dependencies, + state: { + externalPolling: false, + continuousOwnsRecurring: false, + initialOneShotSatisfied: false, + // AppLifecycleService does not expose physical visibility as + // isSuspended(). Keep the observed state in this private context. + lifecycleAllowsScheduling: false, + lifecycleGeneration: 0, + resumeOperation: undefined, + runningResumeGeneration: undefined, + queuedResumeGeneration: undefined, + }, + }; +} + +function canRunPeriodic(context: ReplicationSchedulingContext, settings: ReplicationSchedulingSettings): boolean { + const { dependencies, state } = context; + return ( + state.lifecycleAllowsScheduling && + !state.externalPolling && + !state.continuousOwnsRecurring && + dependencies.isReady() && + !dependencies.isSuspended() && + settings.isConfigured === true && + settings.periodicReplication === true + ); +} + +function reconcilePeriodic(context: ReplicationSchedulingContext): void { + const { dependencies } = context; + const settings = dependencies.currentSettings(); + if (canRunPeriodic(context, settings)) { + dependencies.timer.enable(settings.periodicReplicationInterval * 1000); + } else { + dependencies.timer.disable(); + } +} + +function setContinuousOwnership(context: ReplicationSchedulingContext, ownsRecurring: boolean): void { + const { state } = context; + if (state.continuousOwnsRecurring === ownsRecurring) return; + state.continuousOwnsRecurring = ownsRecurring; + reconcilePeriodic(context); +} + +function isCurrentLifecycleGeneration(context: ReplicationSchedulingContext, generation: number): boolean { + return generation === context.state.lifecycleGeneration; +} + +function canRunResume(context: ReplicationSchedulingContext, generation: number): boolean { + const { dependencies, state } = context; + return ( + isCurrentLifecycleGeneration(context, generation) && + state.lifecycleAllowsScheduling && + !state.externalPolling && + dependencies.isReady() && + !dependencies.isSuspended() + ); +} + +async function runAfterResume(context: ReplicationSchedulingContext, generation: number): Promise { + if (!canRunResume(context, generation)) return; + + const { dependencies, state } = context; + const settings = dependencies.currentSettings(); + if (!settings.isConfigured) { + setContinuousOwnership(context, false); + return; + } + + const skipOneShot = state.initialOneShotSatisfied; + // This marker belongs to one resume attempt. Consume it before any network + // await so an exceptional Continuous start cannot suppress a later retry. + state.initialOneShotSatisfied = false; + if (settings.liveSync) { + setContinuousOwnership(context, true); + let result: ReplicationOutcome; + try { + result = await dependencies.startContinuous({ + trigger: "resume", + interaction: NO_INTERACTION, + }); + } catch (error) { + if (isCurrentLifecycleGeneration(context, generation)) { + setContinuousOwnership(context, false); + } + throw error; + } + if (!isReplicationCompleted(result) && isCurrentLifecycleGeneration(context, generation)) { + setContinuousOwnership(context, false); + } + // A suspend/resume may have started a new lifecycle generation while + // Continuous was settling. Do not let the obsolete result schedule a + // finite fallback for the new generation. + if (isCapabilityUnavailable(result) && canRunResume(context, generation)) { + const currentSettings = dependencies.currentSettings(); + if ( + currentSettings.isConfigured && + currentSettings.liveSync && + currentSettings.syncOnStart && + !skipOneShot + ) { + await dependencies.replicateUnattended({ + trigger: "resume", + interaction: NO_INTERACTION, + }); + } + } + return; + } + + setContinuousOwnership(context, false); + if (settings.syncOnStart && !skipOneShot) { + await dependencies.replicateUnattended({ + trigger: "resume", + interaction: NO_INTERACTION, + }); + } +} + +function scheduleAfterResume(context: ReplicationSchedulingContext): void { + const { dependencies, state } = context; + const requestedGeneration = state.lifecycleGeneration; + if (state.resumeOperation) { + // Duplicate notifications within one generation share the current + // operation. A later lifecycle generation must run after it. + if (state.runningResumeGeneration !== requestedGeneration) { + state.queuedResumeGeneration = requestedGeneration; + } + return; + } + state.runningResumeGeneration = requestedGeneration; + state.resumeOperation = runAfterResume(context, requestedGeneration) + .catch((error: unknown) => { + dependencies.log(error); + }) + .finally(() => { + state.resumeOperation = undefined; + state.runningResumeGeneration = undefined; + const queuedGeneration = state.queuedResumeGeneration; + state.queuedResumeGeneration = undefined; + if (queuedGeneration === state.lifecycleGeneration && state.lifecycleAllowsScheduling) { + scheduleAfterResume(context); + } + }); +} + +/** Schedule eligible work after the application has resumed. */ +export function resumeReplicationScheduling(context: ReplicationSchedulingContext): void { + const { state } = context; + if (!state.lifecycleAllowsScheduling) { + state.lifecycleGeneration += 1; + } + state.lifecycleAllowsScheduling = true; + // runAfterResume executes synchronously until its first await. A Continuous + // request therefore reserves ownership before Periodic is reconciled. + scheduleAfterResume(context); + reconcilePeriodic(context); +} + +/** Stop generic Periodic scheduling before the application suspends. */ +export function suspendReplicationScheduling(context: ReplicationSchedulingContext): void { + context.state.lifecycleAllowsScheduling = false; + context.state.queuedResumeGeneration = undefined; + context.dependencies.timer.disable(); +} + +/** Stop generic Periodic scheduling while settings and provider bindings change. */ +export function prepareReplicationSchedulingForSettings(context: ReplicationSchedulingContext): void { + context.dependencies.timer.disable(); +} + +/** Reconcile generic Periodic scheduling after settings have settled. */ +export function realiseReplicationScheduling(context: ReplicationSchedulingContext): void { + reconcilePeriodic(context); +} + +/** Prevent later timer callbacks from scheduling new work during unload. */ +export function unloadReplicationScheduling(context: ReplicationSchedulingContext): void { + context.state.lifecycleAllowsScheduling = false; + context.state.queuedResumeGeneration = undefined; + context.dependencies.timer.disable(); +} + +/** Execute one timer callback if Periodic still owns recurring work. */ +export async function runPeriodicReplication(context: ReplicationSchedulingContext): Promise { + const { dependencies } = context; + // Clearing an interval does not retract a callback which is already queued. + // Recheck ownership and lifecycle state at execution time. + if (!canRunPeriodic(context, dependencies.currentSettings())) return; + await dependencies.replicateUnattended({ + trigger: "periodic", + interaction: NO_INTERACTION, + }); +} + +/** Declare that an external poller has become, or ceased to be, the recurring-work owner. */ +export function setExternalPollingMode(context: ReplicationSchedulingContext, enabled: boolean): void { + if (context.state.externalPolling === enabled) return; + context.state.externalPolling = enabled; + reconcilePeriodic(context); +} + +/** Consume the next resume-triggered OneShot because the daemon has already converged once. */ +export function markInitialOneShotSatisfied(context: ReplicationSchedulingContext): void { + context.state.initialOneShotSatisfied = true; +} + +type ReplicationSchedulingHost = NecessaryServices< + "API" | "appLifecycle" | "control" | "replication" | "setting", + never +>; + +type ReplicationSchedulingTimerFactory = (process: () => Promise) => ReplicationSchedulingTimer; + +/** + * Compose host lifecycle bindings around one private scheduling context. + * + * The returned view is intentionally limited to daemon scheduling controls. + * @param host Narrow service container used to bind scheduling to the host lifecycle. + * @param createTimer Timer adapter factory, replaceable by focused tests. + * @returns Commands which let the CLI daemon declare its scheduling ownership. + */ +export function useReplicationScheduling( + host: ReplicationSchedulingHost, + createTimer: ReplicationSchedulingTimerFactory = (process) => new PeriodicProcessor(host, process) +): ReplicationSchedulingControl { + const services = host.services; + const log = createInstanceLogFunction("SF:ReplicationScheduling", services.API); + let context!: ReplicationSchedulingContext; + const timer = createTimer(async () => await runPeriodicReplication(context)); + context = createReplicationSchedulingContext({ + isReady: () => services.appLifecycle.isReady(), + isSuspended: () => services.appLifecycle.isSuspended(), + currentSettings: () => services.setting.currentSettings(), + replicateUnattended: (request) => services.replication.replicateUnattended(request), + startContinuous: (request) => services.replication.startContinuous(request), + timer, + log: (error) => log(error, LOG_LEVEL_VERBOSE), + }); + + services.appLifecycle.onUnload.addHandler(() => { + unloadReplicationScheduling(context); + return Promise.resolve(true); + }); + services.setting.onBeforeRealiseSetting.addHandler(() => { + prepareReplicationSchedulingForSettings(context); + return Promise.resolve(true); + }); + services.setting.onSettingRealised.addHandler(() => { + realiseReplicationScheduling(context); + return Promise.resolve(true); + }); + services.appLifecycle.onSuspending.addHandler(() => { + suspendReplicationScheduling(context); + return Promise.resolve(true); + }); + services.appLifecycle.onResumed.addHandler(() => { + resumeReplicationScheduling(context); + return Promise.resolve(true); + }); + + return Object.freeze({ + setExternalPollingMode: (enabled: boolean) => setExternalPollingMode(context, enabled), + markInitialOneShotSatisfied: () => markInitialOneShotSatisfied(context), + }); +} diff --git a/src/serviceFeatures/replicationScheduling.unit.spec.ts b/src/serviceFeatures/replicationScheduling.unit.spec.ts new file mode 100644 index 00000000..154617c3 --- /dev/null +++ b/src/serviceFeatures/replicationScheduling.unit.spec.ts @@ -0,0 +1,349 @@ +import { describe, expect, it, vi } from "vitest"; +import { DEFAULT_SETTINGS } from "@vrtmrz/livesync-commonlib/compat/common/types"; +import { NO_INTERACTION, type ReplicationOutcome } from "@vrtmrz/livesync-commonlib/replication"; +import { + createReplicationSchedulingContext, + markInitialOneShotSatisfied, + resumeReplicationScheduling, + runPeriodicReplication, + setExternalPollingMode, + suspendReplicationScheduling, + useReplicationScheduling, + type ReplicationSchedulingTimer, +} from "./replicationScheduling"; + +function createDeferred() { + let resolve!: (value: T) => void; + const promise = new Promise((res) => { + resolve = res; + }); + return { promise, resolve }; +} + +function createControllerHarness( + overrides: Partial<{ + liveSync: boolean; + syncOnStart: boolean; + periodicReplication: boolean; + periodicReplicationInterval: number; + }> = {} +) { + const settings = { + ...DEFAULT_SETTINGS, + isConfigured: true, + liveSync: false, + syncOnStart: true, + periodicReplication: false, + periodicReplicationInterval: 60, + ...overrides, + }; + const timer: ReplicationSchedulingTimer = { + enable: vi.fn(), + disable: vi.fn(), + }; + const replicateUnattended = vi.fn(async (): Promise => ({ status: "completed" })); + const startContinuous = vi.fn(async (): Promise => ({ status: "completed" })); + const log = vi.fn(); + const context = createReplicationSchedulingContext({ + isReady: vi.fn(() => true), + isSuspended: vi.fn(() => false), + currentSettings: vi.fn(() => settings), + replicateUnattended, + startContinuous, + timer, + log, + }); + return { context, log, replicateUnattended, settings, startContinuous, timer }; +} + +describe("replication scheduling context", () => { + it("starts a configured unattended OneShot without exposing the operation to the lifecycle handler", async () => { + const { context, replicateUnattended } = createControllerHarness(); + + expect(resumeReplicationScheduling(context)).toBeUndefined(); + + await vi.waitFor(() => expect(replicateUnattended).toHaveBeenCalledOnce()); + expect(replicateUnattended).toHaveBeenCalledWith({ + trigger: "resume", + interaction: NO_INTERACTION, + }); + }); + + it("reserves Continuous ownership before reconciling the periodic timer", async () => { + const timeline: string[] = []; + const continuous = createDeferred(); + const { context, startContinuous, timer } = createControllerHarness({ + liveSync: true, + periodicReplication: true, + }); + vi.mocked(timer.disable).mockImplementation(() => { + timeline.push("timer-disabled"); + }); + startContinuous.mockImplementation(() => { + timeline.push("continuous-started"); + return continuous.promise; + }); + + resumeReplicationScheduling(context); + + expect(timeline[0]).toBe("timer-disabled"); + expect(timeline).toContain("continuous-started"); + expect(timer.enable).not.toHaveBeenCalled(); + + continuous.resolve({ status: "completed" }); + await vi.waitFor(() => expect(startContinuous).toHaveBeenCalledOnce()); + }); + + it("restores Periodic and falls back to OneShot when Continuous is not applicable", async () => { + const { context, replicateUnattended, startContinuous, timer } = createControllerHarness({ + liveSync: true, + periodicReplication: true, + periodicReplicationInterval: 45, + }); + startContinuous.mockResolvedValue({ + status: "blocked", + reason: "capability-not-applicable", + }); + + resumeReplicationScheduling(context); + + await vi.waitFor(() => expect(replicateUnattended).toHaveBeenCalledOnce()); + expect(timer.enable).toHaveBeenCalledWith(45_000); + }); + + it("does not run a finite fallback after Continuous starts successfully", async () => { + const { context, replicateUnattended, startContinuous } = createControllerHarness({ liveSync: true }); + + resumeReplicationScheduling(context); + + await vi.waitFor(() => expect(startContinuous).toHaveBeenCalledOnce()); + expect(replicateUnattended).not.toHaveBeenCalled(); + }); + + it("does not run a finite fallback after an actual Continuous failure", async () => { + const { context, replicateUnattended, startContinuous } = createControllerHarness({ liveSync: true }); + startContinuous.mockResolvedValue({ + status: "failed", + error: new Error("connection failed"), + }); + + resumeReplicationScheduling(context); + + await vi.waitFor(() => expect(startContinuous).toHaveBeenCalledOnce()); + expect(replicateUnattended).not.toHaveBeenCalled(); + }); + + it("coalesces concurrent resume notifications", async () => { + const replication = createDeferred(); + const { context, replicateUnattended } = createControllerHarness(); + replicateUnattended.mockImplementation(() => replication.promise); + + resumeReplicationScheduling(context); + resumeReplicationScheduling(context); + + expect(replicateUnattended).toHaveBeenCalledOnce(); + replication.resolve({ status: "completed" }); + await vi.waitFor(() => expect(replicateUnattended).toHaveBeenCalledOnce()); + }); + + it("runs a fresh lifecycle generation instead of applying a stale Continuous fallback", async () => { + const firstContinuous = createDeferred(); + const { context, replicateUnattended, startContinuous, timer } = createControllerHarness({ + liveSync: true, + periodicReplication: true, + }); + startContinuous + .mockImplementationOnce(() => firstContinuous.promise) + .mockResolvedValueOnce({ status: "completed" }); + + resumeReplicationScheduling(context); + suspendReplicationScheduling(context); + resumeReplicationScheduling(context); + firstContinuous.resolve({ + status: "blocked", + reason: "capability-not-applicable", + }); + + await vi.waitFor(() => expect(startContinuous).toHaveBeenCalledTimes(2)); + expect(replicateUnattended).not.toHaveBeenCalled(); + expect(timer.enable).not.toHaveBeenCalled(); + }); + + it("lets the daemon suppress scheduling through the focused control view", async () => { + const { context, replicateUnattended, startContinuous, timer } = createControllerHarness({ + liveSync: true, + periodicReplication: true, + }); + + setExternalPollingMode(context, true); + resumeReplicationScheduling(context); + + expect(timer.disable).toHaveBeenCalled(); + expect(startContinuous).not.toHaveBeenCalled(); + expect(replicateUnattended).not.toHaveBeenCalled(); + }); + + it("consumes the daemon's initial OneShot marker without suppressing a Continuous attempt", async () => { + const { context, replicateUnattended, startContinuous } = createControllerHarness({ liveSync: true }); + startContinuous.mockResolvedValue({ + status: "blocked", + reason: "capability-not-applicable", + }); + + markInitialOneShotSatisfied(context); + resumeReplicationScheduling(context); + + await vi.waitFor(() => expect(startContinuous).toHaveBeenCalledOnce()); + expect(replicateUnattended).not.toHaveBeenCalled(); + }); + + it("consumes the daemon marker even when the first Continuous attempt throws", async () => { + const { context, log, replicateUnattended, startContinuous } = createControllerHarness({ liveSync: true }); + startContinuous.mockRejectedValueOnce(new Error("start failed")).mockResolvedValueOnce({ + status: "blocked", + reason: "capability-not-applicable", + }); + + markInitialOneShotSatisfied(context); + resumeReplicationScheduling(context); + await vi.waitFor(() => expect(log).toHaveBeenCalledOnce()); + + resumeReplicationScheduling(context); + + await vi.waitFor(() => expect(startContinuous).toHaveBeenCalledTimes(2)); + expect(replicateUnattended).toHaveBeenCalledOnce(); + }); + + it("runs the periodic callback through the unattended replication boundary", async () => { + const { context, replicateUnattended } = createControllerHarness({ + syncOnStart: false, + periodicReplication: true, + }); + + resumeReplicationScheduling(context); + + await runPeriodicReplication(context); + + expect(replicateUnattended).toHaveBeenCalledWith({ + trigger: "periodic", + interaction: NO_INTERACTION, + }); + }); + + it("ignores a queued periodic callback before resume and after suspension", async () => { + const { context, replicateUnattended, timer } = createControllerHarness({ + syncOnStart: false, + periodicReplication: true, + }); + + await runPeriodicReplication(context); + expect(replicateUnattended).not.toHaveBeenCalled(); + + resumeReplicationScheduling(context); + expect(timer.enable).toHaveBeenCalledWith(60_000); + + suspendReplicationScheduling(context); + await runPeriodicReplication(context); + expect(replicateUnattended).not.toHaveBeenCalled(); + }); + + it("ignores a queued periodic callback while external polling owns recurring work", async () => { + const { context, replicateUnattended } = createControllerHarness({ + syncOnStart: false, + periodicReplication: true, + }); + resumeReplicationScheduling(context); + + setExternalPollingMode(context, true); + await runPeriodicReplication(context); + + expect(replicateUnattended).not.toHaveBeenCalled(); + }); + + it("ignores a queued periodic callback while Continuous owns recurring work", async () => { + const continuous = createDeferred(); + const { context, replicateUnattended, startContinuous } = createControllerHarness({ + liveSync: true, + syncOnStart: false, + periodicReplication: true, + }); + startContinuous.mockImplementation(() => continuous.promise); + resumeReplicationScheduling(context); + + await runPeriodicReplication(context); + + expect(replicateUnattended).not.toHaveBeenCalled(); + continuous.resolve({ status: "completed" }); + }); +}); + +describe("replication scheduling serviceFeature", () => { + it("binds lifecycle handlers and exposes only daemon scheduling controls", async () => { + const handlers: Record Promise> = {}; + const timer: ReplicationSchedulingTimer = { + enable: vi.fn(), + disable: vi.fn(), + }; + let periodicProcess!: () => Promise; + const replicateUnattended = vi.fn(async (): Promise => ({ status: "completed" })); + const addHandler = (name: string) => + vi.fn((handler: () => Promise) => { + handlers[name] = handler; + return () => undefined; + }); + const services = { + context: {}, + API: { addLog: vi.fn() }, + appLifecycle: { + isReady: vi.fn(() => true), + isSuspended: vi.fn(() => false), + onResumed: { addHandler: addHandler("resumed") }, + onSuspending: { addHandler: addHandler("suspending") }, + onUnload: { addHandler: addHandler("unload") }, + }, + control: { hasUnloaded: vi.fn(() => false) }, + replication: { + replicateUnattended, + startContinuous: vi.fn(async (): Promise => ({ status: "completed" })), + }, + setting: { + currentSettings: vi.fn(() => ({ + ...DEFAULT_SETTINGS, + isConfigured: true, + liveSync: false, + syncOnStart: true, + periodicReplication: true, + })), + onBeforeRealiseSetting: { addHandler: addHandler("before-setting") }, + onSettingRealised: { addHandler: addHandler("setting-realised") }, + }, + }; + + const control = useReplicationScheduling({ services, serviceModules: {} } as never, (process) => { + periodicProcess = process; + return timer; + }); + + expect(Object.keys(control).sort()).toEqual(["markInitialOneShotSatisfied", "setExternalPollingMode"]); + expect(Object.keys(handlers).sort()).toEqual([ + "before-setting", + "resumed", + "setting-realised", + "suspending", + "unload", + ]); + + await expect(handlers.resumed()).resolves.toBe(true); + await vi.waitFor(() => expect(replicateUnattended).toHaveBeenCalledOnce()); + + replicateUnattended.mockClear(); + await periodicProcess(); + expect(replicateUnattended).toHaveBeenCalledWith({ + trigger: "periodic", + interaction: NO_INTERACTION, + }); + + control.setExternalPollingMode(true); + expect(timer.disable).toHaveBeenCalled(); + }); +});