public/obsidian-livesync
1
0
Fork 0
mirror of https://github.com/vrtmrz/obsidian-livesync.git synced 2026-08-30 15:27:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

Compose replication scheduling as a service feature

Browse Source
This commit is contained in:
vorotamoroz
2026-08-28 06:05:29 +00:00
parent fc160ee060
commit f7206b1a6e
15 changed files with 850 additions and 525 deletions
Download Patch File Download Diff File Expand all files Collapse all files
devs.md
+9 -3
View File
@@ -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:
docs/adr/2026_08_replicator_capabilities_01_core_contract.md
+40 -25
View File
@@ -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,
docs/adr/2026_08_replicator_capabilities_03_migration_plan.md
+17 -8
View File
@@ -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;
src/LiveSyncBaseCore.ts
+13 -8
View File
@@ -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<T, TCommands>) => AbstractModule[],
addOnsInitialiser: (core: LiveSyncBaseCore<T, TCommands>) => TCommands[],
featuresInitialiser: (core: LiveSyncBaseCore<T, TCommands>) => void
featuresInitialiser: (core: LiveSyncBaseCore<T, TCommands>, 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),
});
}
}
src/apps/cli/commands/daemonCommand.unit.spec.ts
+27 -15
View File
@@ -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<typeof createCoreMock>) {
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);
src/apps/cli/commands/runCommand.ts
+5 -6
View File
@@ -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<boolean> {
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();
src/apps/cli/commands/types.ts
+3
View File
@@ -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<NodeServiceContext, never>;
/** 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;
src/apps/cli/main.ts
+8 -1
View File
@@ -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<NodeServiceContext, never>, serviceHub: InjectableServiceHub<NodeServiceContext>) => {
@@ -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,
src/modules/core/AutomaticReplicationTriggers.unit.spec.ts
+40 -35
View File
@@ -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<boolean>> = [];
const settingRealisedHandlers: Array<() => Promise<boolean>> = [];
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<boolean>) => resumeHandlers.push(handler)) },
onSuspending: { addHandler: vi.fn() },
onUnload: { addHandler: vi.fn() },
},
control: { hasUnloaded: vi.fn(() => false) },
replication: {
startContinuous,
replicateUnattended: vi.fn(async () => ({ status: "completed" as const })),
},
setting: {
currentSettings: vi.fn(() => settings),
onBeforeRealiseSetting: { addHandler: vi.fn() },
onSettingRealised: {
addHandler: vi.fn((handler: () => Promise<boolean>) => settingRealisedHandlers.push(handler)),
const 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();
},
};
}
src/modules/core/ModulePeriodicProcess.ts
-62
View File
@@ -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<boolean> {
return this.disablePeriodic();
}
private _everyBeforeSuspendProcess(): Promise<boolean> {
return this.disablePeriodic();
}
private _everyAfterResumeProcess(): Promise<boolean> {
return this.resumePeriodic();
}
private _everyAfterRealizeSetting(): Promise<boolean> {
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));
}
}
src/modules/core/ModuleReplicationLifecycle.ts
-113
View File
@@ -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<boolean>;
private async runAfterResume(): Promise<boolean> {
if (this.schedulingControl.externalPolling) return true;
if (this.services.appLifecycle.isSuspended()) return true;
if (!this.services.appLifecycle.isReady()) return true;
const settings = this.services.setting.currentSettings();
if (!settings.isConfigured) {
setContinuousSchedulingOwnership(this.core, false);
return true;
}
const skipOneShot = this.schedulingControl.initialOneShotSatisfied;
if (settings.liveSync) {
// Reserve recurring ownership before the asynchronous start so a
// later resume handler cannot enable Periodic in the meantime.
setContinuousSchedulingOwnership(this.core, true);
const result = await this.services.replication.startContinuous({
trigger: "resume",
interaction: NO_INTERACTION,
});
if (!isReplicationCompleted(result)) {
setContinuousSchedulingOwnership(this.core, false);
}
// The daemon's initial finite convergence must not suppress a
// supported Continuous start. It only suppresses the fallback
// OneShot when Continuous is unavailable.
this.schedulingControl.initialOneShotSatisfied = false;
if (isCapabilityUnavailable(result) && settings.syncOnStart && !skipOneShot) {
await this.services.replication.replicateUnattended({
trigger: "resume",
interaction: NO_INTERACTION,
});
}
return true;
}
setContinuousSchedulingOwnership(this.core, false);
if (settings.syncOnStart && !skipOneShot) {
await this.services.replication.replicateUnattended({
trigger: "resume",
interaction: NO_INTERACTION,
});
}
this.schedulingControl.initialOneShotSatisfied = false;
return true;
}
private _everyAfterResumeProcess(): Promise<boolean> {
if (!this.resumePromise) {
// The lifecycle event is a short notification boundary. Keep the
// long-running OneShot/Continuous start coalesced internally, but
// let later resume handlers (P2P, periodic scheduling, and other
// modules) continue without waiting for network work to settle.
this.resumePromise = this.runAfterResume()
.catch((error) => {
this._log(error, LOG_LEVEL_VERBOSE);
return true;
})
.finally(() => {
this.resumePromise = undefined;
});
}
return Promise.resolve(true);
}
/**
* Let a CLI daemon own recurring polling without a duplicate lifecycle or
* generic periodic scheduler. This is intentionally narrower than a
* provider or Replicator control API.
*/
setExternalPollingMode(enabled: boolean): void {
setExternalPollingMode(this.core, enabled);
}
/** Mark the daemon's initial finite convergence for the next resume. */
markInitialOneShotSatisfied(): void {
markInitialOneShotSatisfied(this.core);
}
override onBindFunction(core: LiveSyncCore, services: typeof core.services): void {
services.appLifecycle.onResumed.addHandler(this._everyAfterResumeProcess.bind(this));
}
}
src/modules/core/ReplicationLifecycle.unit.spec.ts
-200
View File
@@ -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<boolean>;
function createResumeHarness(settings: {
liveSync: boolean;
syncOnStart: boolean;
isConfigured?: boolean;
remoteType?: string;
P2P_Enabled?: boolean;
}) {
const resumeHandlers: ResumeHandler[] = [];
const replicateUnattended = vi.fn(async (): Promise<ReplicationOutcome> => ({ status: "completed" }));
const startContinuous = vi.fn(async (): Promise<ReplicationOutcome> => ({ status: "completed" }));
const currentSettings = {
isConfigured: true,
periodicReplication: false,
...settings,
};
const services = {
context: createServiceContext(),
API: {
addLog: vi.fn(),
addCommand: vi.fn(),
registerWindow: vi.fn(),
addRibbonIcon: vi.fn(),
registerProtocolHandler: vi.fn(),
isOnline: true,
},
appLifecycle: {
isReady: vi.fn(() => true),
isSuspended: vi.fn(() => false),
onResumed: {
addHandler: vi.fn((handler: ResumeHandler) => resumeHandlers.push(handler)),
},
},
replication: {
replicateUnattended,
startContinuous,
},
setting: {
currentSettings: vi.fn(() => currentSettings),
},
};
const core = {
_services: services,
services,
settings: currentSettings,
} as any;
const module = new ModuleReplicationLifecycle(core);
module.onBindFunction(core, services as never);
return {
core,
module,
replicateUnattended,
startContinuous,
resume: async () => await Promise.all(resumeHandlers.map((handler) => handler())),
};
}
describe("provider-independent replication resume lifecycle", () => {
it("starts one unattended OneShot when sync-on-start is enabled", async () => {
const harness = createResumeHarness({ liveSync: false, syncOnStart: true });
await harness.resume();
expect(harness.replicateUnattended).toHaveBeenCalledOnce();
expect(harness.replicateUnattended).toHaveBeenCalledWith({
trigger: "resume",
interaction: NO_INTERACTION,
});
expect(harness.startContinuous).not.toHaveBeenCalled();
});
it("falls back to sync-on-start when Continuous is not applicable", async () => {
const harness = createResumeHarness({ liveSync: true, syncOnStart: true });
harness.startContinuous.mockResolvedValue({
status: "blocked",
reason: "capability-not-applicable",
});
await harness.resume();
expect(harness.startContinuous).toHaveBeenCalledWith({
trigger: "resume",
interaction: NO_INTERACTION,
});
expect(harness.replicateUnattended).toHaveBeenCalledWith({
trigger: "resume",
interaction: NO_INTERACTION,
});
});
it("starts Continuous without a finite fallback when it is supported", async () => {
const harness = createResumeHarness({ liveSync: true, syncOnStart: true });
await harness.resume();
expect(harness.startContinuous).toHaveBeenCalledOnce();
expect(harness.replicateUnattended).not.toHaveBeenCalled();
});
it("does not fall back after an actual Continuous failure", async () => {
const harness = createResumeHarness({ liveSync: true, syncOnStart: true });
harness.startContinuous.mockResolvedValue({
status: "failed",
error: new Error("connection failed"),
});
await harness.resume();
expect(harness.replicateUnattended).not.toHaveBeenCalled();
});
it("coalesces concurrent resume callbacks", async () => {
const harness = createResumeHarness({ liveSync: false, syncOnStart: true });
let resolveReplication!: (value: { status: "completed" }) => void;
harness.replicateUnattended.mockImplementationOnce(
() => new Promise((resolve) => (resolveReplication = resolve))
);
const first = harness.resume();
const second = harness.resume();
resolveReplication({ status: "completed" });
await Promise.all([first, second]);
expect(harness.replicateUnattended).toHaveBeenCalledOnce();
});
it("does not block later resume handlers while a OneShot is running", async () => {
const harness = createResumeHarness({ liveSync: false, syncOnStart: true });
let resolveReplication!: (value: { status: "completed" }) => void;
harness.replicateUnattended.mockImplementationOnce(
() => new Promise((resolve) => (resolveReplication = resolve))
);
const resumed = harness.resume();
await expect(resumed).resolves.toEqual([true]);
expect(harness.replicateUnattended).toHaveBeenCalledOnce();
resolveReplication({ status: "completed" });
await resumed;
});
it("skips only the daemon-satisfied OneShot while allowing Continuous", async () => {
const harness = createResumeHarness({ liveSync: true, syncOnStart: true });
getReplicationSchedulingControl(harness.core).initialOneShotSatisfied = true;
harness.startContinuous.mockResolvedValue({
status: "blocked",
reason: "capability-not-applicable",
});
await harness.resume();
expect(harness.startContinuous).toHaveBeenCalledOnce();
expect(harness.replicateUnattended).not.toHaveBeenCalled();
expect(getReplicationSchedulingControl(harness.core).initialOneShotSatisfied).toBe(false);
});
it("does not start lifecycle replication while an external poller owns scheduling", async () => {
const harness = createResumeHarness({ liveSync: false, syncOnStart: true });
setExternalPollingMode(harness.core, true);
await harness.resume();
expect(harness.startContinuous).not.toHaveBeenCalled();
expect(harness.replicateUnattended).not.toHaveBeenCalled();
});
it("requests the generic finite fallback for P2P when Continuous is not applicable", async () => {
const harness = createResumeHarness({
remoteType: REMOTE_P2P,
P2P_Enabled: true,
liveSync: true,
syncOnStart: true,
});
harness.startContinuous.mockResolvedValue({
status: "blocked",
reason: "capability-not-applicable",
});
harness.replicateUnattended.mockResolvedValue({
status: "blocked",
reason: "capability-not-implemented",
});
await harness.resume();
expect(harness.startContinuous).toHaveBeenCalledOnce();
expect(harness.replicateUnattended).toHaveBeenCalledWith({
trigger: "resume",
interaction: NO_INTERACTION,
});
});
});
src/modules/core/ReplicationScheduling.ts
-49
View File
@@ -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<object, ReplicationSchedulingControl>();
export function getReplicationSchedulingControl(owner: object): ReplicationSchedulingControl {
let control = controls.get(owner);
if (!control) {
control = {
externalPolling: false,
continuousOwnsRecurring: false,
initialOneShotSatisfied: false,
};
controls.set(owner, control);
}
return control;
}
export function setContinuousSchedulingOwnership(owner: object, ownsRecurring: boolean): void {
const control = getReplicationSchedulingControl(owner);
if (control.continuousOwnsRecurring === ownsRecurring) return;
control.continuousOwnsRecurring = ownsRecurring;
control.refreshPeriodic?.();
}
export function setExternalPollingMode(owner: object, enabled: boolean): void {
const control = getReplicationSchedulingControl(owner);
control.externalPolling = enabled;
if (enabled) control.disablePeriodic?.();
}
export function markInitialOneShotSatisfied(owner: object): void {
getReplicationSchedulingControl(owner).initialOneShotSatisfied = true;
}
src/serviceFeatures/replicationScheduling.ts
+339
View File
@@ -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<ReplicationOutcome>;
startContinuous(request: ContinuousReplicationRequest): Promise<ReplicationOutcome>;
timer: ReplicationSchedulingTimer;
log(error: unknown): void;
}
interface ReplicationSchedulingState {
externalPolling: boolean;
continuousOwnsRecurring: boolean;
initialOneShotSatisfied: boolean;
lifecycleAllowsScheduling: boolean;
lifecycleGeneration: number;
resumeOperation: Promise<void> | 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<void> {
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<void> {
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<void>) => 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),
});
}
src/serviceFeatures/replicationScheduling.unit.spec.ts
+349
View File
@@ -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<T>() {
let resolve!: (value: T) => void;
const promise = new Promise<T>((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<ReplicationOutcome> => ({ status: "completed" }));
const startContinuous = vi.fn(async (): Promise<ReplicationOutcome> => ({ 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<ReplicationOutcome>();
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<ReplicationOutcome>();
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<ReplicationOutcome>();
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<ReplicationOutcome>();
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<string, () => Promise<boolean>> = {};
const timer: ReplicationSchedulingTimer = {
enable: vi.fn(),
disable: vi.fn(),
};
let periodicProcess!: () => Promise<void>;
const replicateUnattended = vi.fn(async (): Promise<ReplicationOutcome> => ({ status: "completed" }));
const addHandler = (name: string) =>
vi.fn((handler: () => Promise<boolean>) => {
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<ReplicationOutcome> => ({ 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();
});
});