From 72f033fca49da6b4c2b4619ed63942ca63619673 Mon Sep 17 00:00:00 2001 From: vorotamoroz Date: Sun, 30 Aug 2026 10:23:04 +0000 Subject: [PATCH] Compose replication lifecycle without a legacy module --- src/LiveSyncBaseCore.ts | 6 +- .../testdeno/helpers/compressionDataset.ts | 8 +- .../cli/testdeno/test-benchmark-contract.ts | 5 +- src/modules/core/ModuleReplicator.ts | 424 ----------------- .../core/ModuleReplicator.unit.spec.ts | 445 ------------------ .../replication}/ReplicateResultProcessor.ts | 78 +-- .../ReplicateResultProcessor.unit.spec.ts | 39 +- .../replication/automaticTriggers.ts | 71 +++ .../automaticTriggers.unit.spec.ts} | 76 ++- .../centralCompatibilityRecovery.ts | 194 ++++++++ .../centralCompatibilityRecovery.unit.spec.ts | 229 +++++++++ src/serviceFeatures/replication/index.ts | 110 +++++ src/serviceFeatures/replication/preflight.ts | 65 +++ .../replicationFeature.unit.spec.ts | 126 +++++ 14 files changed, 939 insertions(+), 937 deletions(-) delete mode 100644 src/modules/core/ModuleReplicator.ts delete mode 100644 src/modules/core/ModuleReplicator.unit.spec.ts rename src/{modules/core => serviceFeatures/replication}/ReplicateResultProcessor.ts (91%) rename src/{modules/core => serviceFeatures/replication}/ReplicateResultProcessor.unit.spec.ts (81%) create mode 100644 src/serviceFeatures/replication/automaticTriggers.ts rename src/{modules/core/AutomaticReplicationTriggers.unit.spec.ts => serviceFeatures/replication/automaticTriggers.unit.spec.ts} (78%) create mode 100644 src/serviceFeatures/replication/centralCompatibilityRecovery.ts create mode 100644 src/serviceFeatures/replication/centralCompatibilityRecovery.unit.spec.ts create mode 100644 src/serviceFeatures/replication/index.ts create mode 100644 src/serviceFeatures/replication/preflight.ts create mode 100644 src/serviceFeatures/replication/replicationFeature.unit.spec.ts diff --git a/src/LiveSyncBaseCore.ts b/src/LiveSyncBaseCore.ts index 47a36871..6d2c69a9 100644 --- a/src/LiveSyncBaseCore.ts +++ b/src/LiveSyncBaseCore.ts @@ -26,7 +26,6 @@ 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 { ModuleReplicator } from "./modules/core/ModuleReplicator"; import { ModuleConflictChecker } from "./modules/coreFeatures/ModuleConflictChecker"; import { ModuleConflictResolver } from "./modules/coreFeatures/ModuleConflictResolver"; import { ModuleResolvingMismatchedTweaks } from "./modules/coreFeatures/ModuleResolveMismatchedTweaks"; @@ -37,6 +36,7 @@ import { usePrepareDatabaseForUse } from "@vrtmrz/livesync-commonlib/compat/serv import type { Constructor } from "@vrtmrz/livesync-commonlib/compat/common/utils.type"; import { useReplicationScheduling, type ReplicationSchedulingControl } from "./serviceFeatures/replicationScheduling"; import { createCentralReplicatorProviderDefinitions } from "./common/replicatorProviders"; +import { useReplicationFeature } from "./serviceFeatures/replication"; /** Focused views returned by serviceFeatures which the host may consume during composition. */ export interface LiveSyncCoreFeatureViews { @@ -101,6 +101,9 @@ export class LiveSyncBaseCore< for (const addOn of addOns) { this._registerAddOn(addOn); } + // Preserve the former ModuleReplicator lifecycle-handler order: + // host features and add-ons first, then replication, then legacy modules. + useReplicationFeature(this); this.bindModuleFunctions(); } /** @@ -161,7 +164,6 @@ export class LiveSyncBaseCore< public registerModules(extraModules: AbstractModule[] = []) { this._registerModule(new ModuleLiveSyncMain(this)); this._registerModule(new ModuleConflictChecker(this)); - this._registerModule(new ModuleReplicator(this)); this._registerModule(new ModuleConflictResolver(this)); this._registerModule(new ModuleResolvingMismatchedTweaks(this)); this._registerModule(new ModuleBasicMenu(this)); diff --git a/src/apps/cli/testdeno/helpers/compressionDataset.ts b/src/apps/cli/testdeno/helpers/compressionDataset.ts index 83a63837..308836d8 100644 --- a/src/apps/cli/testdeno/helpers/compressionDataset.ts +++ b/src/apps/cli/testdeno/helpers/compressionDataset.ts @@ -143,8 +143,12 @@ export async function createCompressionBenchmarkDataset(options: { ); await copyRepositoryFile("json", "package.json", "package.json"); await copyRepositoryFile("json", "manifest.json", "manifest.json"); - await copyRepositoryFile("ts", "src/modules/core/ModuleReplicator.ts", "ModuleReplicator.ts"); - await copyRepositoryFile("ts", "src/modules/core/ReplicateResultProcessor.ts", "ReplicateResultProcessor.ts"); + await copyRepositoryFile("ts", "src/serviceFeatures/replication/index.ts", "replicationFeature.ts"); + await copyRepositoryFile( + "ts", + "src/serviceFeatures/replication/ReplicateResultProcessor.ts", + "ReplicateResultProcessor.ts" + ); const markdownBytes = await Deno.readFile(join(repositoryRoot, "docs/settings.md")); const gzipPath = join(datasetRoot, "gz", "settings.md.gz"); diff --git a/src/apps/cli/testdeno/test-benchmark-contract.ts b/src/apps/cli/testdeno/test-benchmark-contract.ts index a2237bb7..5a201fec 100644 --- a/src/apps/cli/testdeno/test-benchmark-contract.ts +++ b/src/apps/cli/testdeno/test-benchmark-contract.ts @@ -79,7 +79,6 @@ Deno.test("benchmark cases record scope and limitations for paper use", () => { ); } }); - Deno.test("CouchDB latency proxy applies half the requested RTT in each direction", async () => { const backendPort = getFreePort(); const proxyPort = getFreePort(); @@ -156,8 +155,8 @@ Deno.test("compression benchmark dataset covers representative file kinds determ "images/quick-setup/guide-quick-setup-first-setup-uri.png", "package.json", "manifest.json", - "src/modules/core/ModuleReplicator.ts", - "src/modules/core/ReplicateResultProcessor.ts", + "src/serviceFeatures/replication/index.ts", + "src/serviceFeatures/replication/ReplicateResultProcessor.ts", ]; try { for (const [index, relativePath] of repositoryFiles.entries()) { diff --git a/src/modules/core/ModuleReplicator.ts b/src/modules/core/ModuleReplicator.ts deleted file mode 100644 index f14321e3..00000000 --- a/src/modules/core/ModuleReplicator.ts +++ /dev/null @@ -1,424 +0,0 @@ -import type PouchDB from "pouchdb-core"; -import { fireAndForget } from "octagonal-wheels/promises"; -import { AbstractModule } from "@/modules/AbstractModule"; -import { Logger, LOG_LEVEL_NOTICE, LOG_LEVEL_INFO, LOG_LEVEL_VERBOSE } from "octagonal-wheels/common/logger"; -import { skipIfDuplicated } from "octagonal-wheels/concurrency/lock"; -import { balanceChunkPurgedDBs } from "@vrtmrz/livesync-commonlib/compat/pouchdb/chunks"; -import { purgeUnreferencedChunks } from "@vrtmrz/livesync-commonlib/compat/pouchdb/chunks"; -import { LiveSyncCouchDBReplicator } from "@vrtmrz/livesync-commonlib/compat/replication/couchdb/LiveSyncReplicator"; -import { - type EntryDoc, - type ObsidianLiveSyncSettings, - type RemoteType, -} from "@vrtmrz/livesync-commonlib/compat/common/types"; - -import { scheduleTask } from "octagonal-wheels/concurrency/task"; -import { EVENT_FILE_SAVED, EVENT_SETTING_SAVED, eventHub } from "@/common/events"; - -import { $msg } from "@/common/translation"; -import type { LiveSyncCore } from "@/main"; -import { ReplicateResultProcessor } from "./ReplicateResultProcessor"; -import { UnresolvedErrorManager } from "@vrtmrz/livesync-commonlib/compat/services/base/UnresolvedErrorManager"; -import { clearHandlers } from "@vrtmrz/livesync-commonlib/compat/replication/SyncParamsHandler"; -import type { NecessaryServices } from "@vrtmrz/livesync-commonlib/compat/interfaces/ServiceModule"; -import { MARK_LOG_NETWORK_ERROR } from "@vrtmrz/livesync-commonlib/compat/services/lib/logUtils"; -import { usesLegacyIndexedDBAdapter } from "@/common/compatibilitySettings.ts"; -import { - CENTRAL_COMPATIBILITY_REJECTION_REASONS, - NO_INTERACTION, - REMOTE_RESOURCE_KINDS, - type ReplicationFailureRequest, -} from "@vrtmrz/livesync-commonlib/replication"; -import { withOwnedRemoteResource } from "@/common/ownedRemoteResource.ts"; - -function isOnlineAndCanReplicate( - errorManager: UnresolvedErrorManager, - host: NecessaryServices<"API", never>, - showMessage: boolean -): Promise { - const errorMessage = "Network is offline"; - if (!host.services.API.isOnline) { - errorManager.showError(errorMessage, showMessage ? LOG_LEVEL_NOTICE : LOG_LEVEL_INFO); - return Promise.resolve(false); - } - errorManager.clearError(errorMessage); - return Promise.resolve(true); -} -/** Refresh and validate the selected central provider's owned Security Seed resource. */ -async function canReplicateWithSecuritySeed( - errorManager: UnresolvedErrorManager, - host: NecessaryServices<"replicator" | "setting", never>, - showMessage: boolean -): Promise { - const currentSettings = host.services.setting.currentSettings(); - const errorMessage = $msg("Replicator.Message.InitialiseFatalError"); - // Showing message is false: that because be shown here. (And it is a fatal error, no way to hide it). - // tagged as network error at beginning for error filtering with NetworkWarningStyles - const ensureMessage = `${MARK_LOG_NETWORK_ERROR}Failed to initialise the encryption key, preventing replication.`; - try { - const resource = await host.services.replicator.createRemoteResource( - REMOTE_RESOURCE_KINDS.SECURITY_SEED, - currentSettings - ); - if (!resource) { - errorManager.showError(errorMessage, showMessage ? LOG_LEVEL_NOTICE : LOG_LEVEL_INFO); - return false; - } - errorManager.clearError(errorMessage); - const seed = await withOwnedRemoteResource(resource, (ownedResource) => ownedResource.read()); - if (seed.length == 0) throw new Error("PBKDF2 salt (Security Seed) is empty"); - } catch (error) { - Logger(error, LOG_LEVEL_VERBOSE); - errorManager.showError(ensureMessage, showMessage ? LOG_LEVEL_NOTICE : LOG_LEVEL_INFO); - return false; - } - errorManager.clearError(ensureMessage); - return true; -} - -export class ModuleReplicator extends AbstractModule { - _replicatorType?: RemoteType; - - processor: ReplicateResultProcessor = new ReplicateResultProcessor(this); - private _unresolvedErrorManager: UnresolvedErrorManager = new UnresolvedErrorManager( - this.core.services.appLifecycle, - this.core.services.context.events - ); - - clearErrors() { - this._unresolvedErrorManager.clearErrors(); - } - - private _normalFileReflectionFilterSignature: string | undefined; - - private getNormalFileReflectionFilterSignature( - settings: Pick< - ObsidianLiveSyncSettings, - | "handleFilenameCaseSensitive" - | "ignoreFiles" - | "maxMTimeForReflectEvents" - | "syncIgnoreRegEx" - | "syncInternalFiles" - | "syncMaxSizeInMB" - | "syncOnlyRegEx" - | "useIgnoreFiles" - > - ): string { - return JSON.stringify({ - handleFilenameCaseSensitive: settings.handleFilenameCaseSensitive ?? false, - ignoreFiles: settings.ignoreFiles ?? "", - maxMTimeForReflectEvents: settings.maxMTimeForReflectEvents ?? 0, - syncIgnoreRegEx: settings.syncIgnoreRegEx ?? "", - syncInternalFiles: settings.syncInternalFiles ?? false, - syncMaxSizeInMB: settings.syncMaxSizeInMB ?? 0, - syncOnlyRegEx: settings.syncOnlyRegEx ?? "", - useIgnoreFiles: settings.useIgnoreFiles ?? false, - }); - } - - private _everyOnloadAfterLoadSettings(): Promise { - this._normalFileReflectionFilterSignature = this.getNormalFileReflectionFilterSignature(this.settings); - eventHub.onEvent(EVENT_FILE_SAVED, () => { - if (this.settings.syncOnSave && !this.core.services.appLifecycle.isSuspended()) { - scheduleTask("perform-replicate-after-save", 250, () => - this.services.replication.replicateUnattendedByEvent({ - trigger: "database-event", - interaction: NO_INTERACTION, - }) - ); - } - }); - eventHub.onEvent(EVENT_SETTING_SAVED, (setting) => { - const previousReflectionFilter = this._normalFileReflectionFilterSignature; - const nextReflectionFilter = this.getNormalFileReflectionFilterSignature(setting); - this._normalFileReflectionFilterSignature = nextReflectionFilter; - if (this.core.settings.suspendParseReplicationResult) { - this.processor.suspend(); - } else { - this.processor.resume(); - } - if (previousReflectionFilter !== undefined && previousReflectionFilter !== nextReflectionFilter) { - fireAndForget(() => this.processor.reprocessStoredDocuments()); - } - }); - - return Promise.resolve(true); - } - - _onBeforeReplicatorPublication(): Promise { - // Clear key-derivation handlers before the candidate Replicator becomes active. - clearHandlers(); - return Promise.resolve(true); - } - - _everyOnDatabaseInitialized(showNotice: boolean): Promise { - fireAndForget(() => this.processor.restoreFromSnapshotOnce()); - return Promise.resolve(true); - } - - async _everyBeforeReplicate(showMessage: boolean): Promise { - await this.processor.restoreFromSnapshotOnce(); - this.clearErrors(); - return true; - } - - /** - * Reconciles an IndexedDB-backed local database after replication reports that the remote was cleaned. - * - * The remote milestone remains a supported compatibility signal. The user can either fetch the remote - * database again, or purge unreferenced local chunks before accepting this device again. - * - * @param showMessage Whether to show the recovery choices as user-facing notices. - * @param setting Detached settings used by the failed attempt. - * @param expectedContext Publication which produced the compatibility rejection. - */ - async cleaned( - showMessage: boolean, - setting: ObsidianLiveSyncSettings, - expectedContext: ReplicationFailureRequest["context"] - ) { - Logger(`The remote database has been cleaned.`, showMessage ? LOG_LEVEL_NOTICE : LOG_LEVEL_INFO); - await skipIfDuplicated("cleanup", async () => { - const count = await purgeUnreferencedChunks(this.localDatabase.localDatabase, true); - const message = `The remote database has been cleaned up. -To synchronize, this device must be also cleaned up. ${count} chunk(s) will be erased from this device. -However, If there are many chunks to be deleted, maybe fetching again is faster. -We will lose the history of this device if we fetch the remote database again. -Even if you choose to clean up, you will see this option again if you exit Obsidian and then synchronise again.`; - const CHOICE_FETCH = "Fetch again"; - const CHOICE_CLEAN = "Cleanup"; - const CHOICE_DISMISS = "Dismiss"; - const ret = await this.core.confirm.confirmWithMessage( - "Cleaned", - message, - [CHOICE_FETCH, CHOICE_CLEAN, CHOICE_DISMISS], - CHOICE_DISMISS, - 30 - ); - if (ret == CHOICE_FETCH) { - await this.core.rebuilder.$performRebuildDB("localOnly"); - } - if (ret == CHOICE_CLEAN) { - await this.services.replicator.runBoundedRemoteActivity( - () => - this.services.replicator.runWithActiveReplicatorContext(async (context) => { - if (context !== expectedContext) return; - const replicator = context.replicator; - if (!(replicator instanceof LiveSyncCouchDBReplicator)) return; - const remoteDB = await replicator.connectRemoteCouchDBWithSetting( - setting, - this.services.API.isMobile(), - true - ); - if (typeof remoteDB == "string") { - Logger(remoteDB, LOG_LEVEL_NOTICE); - return false; - } - - try { - await purgeUnreferencedChunks(this.localDatabase.localDatabase, false); - this.localDatabase.clearCaches(); - // Perform the synchronisation once. - const replicated = await this.services.replicator.runFiniteReplicationActivity( - () => replicator.openOneShotReplication(setting, showMessage, false, "sync", true), - { label: "replication" } - ); - if (replicated) { - await balanceChunkPurgedDBs(this.localDatabase.localDatabase, remoteDB.db); - await purgeUnreferencedChunks(this.localDatabase.localDatabase, false); - this.localDatabase.clearCaches(); - await replicator.markRemoteResolved(setting); - Logger( - "The local database has been cleaned up.", - showMessage ? LOG_LEVEL_NOTICE : LOG_LEVEL_INFO - ); - } else { - Logger( - "Replication has been cancelled. Please try it again.", - showMessage ? LOG_LEVEL_NOTICE : LOG_LEVEL_INFO - ); - } - } finally { - await remoteDB.close(); - } - }), - { label: "database-cleanup" } - ); - } - }); - } - - private async onReplicationFailed(request: ReplicationFailureRequest): Promise { - const { context, interaction, outcome, setting, showMessage } = request; - if (!showMessage) { - // Automatic requests may report the failure, but they must never - // enter tweak, lock, fetch, unlock, or cleanup dialogues. - Logger(`Replication failed on an unattended path.`, LOG_LEVEL_INFO); - return false; - } - if (interaction.kind !== "permitted" || !interaction.permissions.failureRecovery) return false; - const recovery = outcome.recoveryHint; - if (!recovery) return false; - if ( - recovery.reason === CENTRAL_COMPATIBILITY_REJECTION_REASONS.TWEAK_MISMATCH && - recovery.preferredTweakValue - ) { - await this.services.tweakValue.askResolvingMismatched( - recovery.preferredTweakValue, - async (effectiveSetting) => { - let updated = false; - await this.services.replicator.runWithActiveReplicatorContext(async (activeContext) => { - if (activeContext !== context) return; - const candidate = activeContext.replicator as typeof activeContext.replicator & { - setPreferredRemoteTweakSettings?: ( - setting: ObsidianLiveSyncSettings - ) => Promise; - }; - if (typeof candidate.setPreferredRemoteTweakSettings !== "function") return; - await candidate.setPreferredRemoteTweakSettings({ ...effectiveSetting }); - updated = true; - }); - return updated; - } - ); - } else { - if ( - recovery.reason === CENTRAL_COMPATIBILITY_REJECTION_REASONS.NODE_LOCKED || - recovery.reason === CENTRAL_COMPATIBILITY_REJECTION_REASONS.NODE_CLEANED - ) { - if ( - recovery.reason === CENTRAL_COMPATIBILITY_REJECTION_REASONS.NODE_CLEANED && - usesLegacyIndexedDBAdapter(setting) - ) { - await this.cleaned(showMessage, setting, context); - } else { - const message = $msg("Replicator.Dialogue.Locked.Message"); - const CHOICE_FETCH = $msg("Replicator.Dialogue.Locked.Action.Fetch"); - const CHOICE_DISMISS = $msg("Replicator.Dialogue.Locked.Action.Dismiss"); - const CHOICE_UNLOCK = $msg("Replicator.Dialogue.Locked.Action.Unlock"); - const ret = await this.core.confirm.askSelectStringDialogue( - message, - [CHOICE_FETCH, CHOICE_UNLOCK, CHOICE_DISMISS], - { - title: $msg("Replicator.Dialogue.Locked.Title"), - defaultAction: CHOICE_DISMISS, - timeout: 60, - } - ); - if (ret == CHOICE_FETCH) { - this._log($msg("Replicator.Dialogue.Locked.Message.Fetch"), LOG_LEVEL_NOTICE); - await this.core.rebuilder.scheduleFetch(); - this.services.appLifecycle.scheduleRestart(); - return false; - } else if (ret == CHOICE_UNLOCK) { - let unlocked = false; - await this.services.replicator.runWithActiveReplicatorContext(async (activeContext) => { - if (activeContext !== context) return; - const replicator = activeContext.replicator as typeof activeContext.replicator & { - markRemoteResolved(setting: ObsidianLiveSyncSettings): Promise; - }; - if (typeof replicator.markRemoteResolved !== "function") return; - await replicator.markRemoteResolved(setting); - unlocked = true; - }); - if (unlocked) { - this._log($msg("Replicator.Dialogue.Locked.Message.Unlocked"), LOG_LEVEL_NOTICE); - } - return false; - } - } - } - } - // TODO: Check again and true/false return. This will be the result for performReplication. - return false; - } - - // private async _replicateByEvent(): Promise { - // const least = this.settings.syncMinimumInterval; - // if (least > 0) { - // return rateLimitedSharedExecution(KEY_REPLICATION_ON_EVENT, least, async () => { - // return await this.services.replication.replicate(); - // }); - // } - // return await shareRunningResult(`replication`, () => this.services.replication.replicate()); - // } - - _parseReplicationResult(docs: Array>): Promise { - this.processor.enqueueAll(docs); - return Promise.resolve(true); - } - - // _everyBeforeSuspendProcess(): Promise { - // this.core.replicator?.closeReplication(); - // return Promise.resolve(true); - // } - - // private async _replicateAllToServer( - // showingNotice: boolean = false, - // sendChunksInBulkDisabled: boolean = false - // ): Promise { - // if (!this.services.appLifecycle.isReady()) return false; - // if (!(await this.services.replication.onBeforeReplicate(showingNotice))) { - // Logger($msg("Replicator.Message.SomeModuleFailed"), LOG_LEVEL_NOTICE); - // return false; - // } - // if (!sendChunksInBulkDisabled) { - // if (this.core.replicator instanceof LiveSyncCouchDBReplicator) { - // if ( - // (await this.core.confirm.askYesNoDialog("Do you want to send all chunks before replication?", { - // defaultOption: "No", - // timeout: 20, - // })) == "yes" - // ) { - // await this.core.replicator.sendChunks(this.core.settings, undefined, true, 0); - // } - // } - // } - // const ret = await this.core.replicator.replicateAllToServer(this.settings, showingNotice); - // if (ret) return true; - // const checkResult = await this.services.replication.checkConnectionFailure(); - // if (checkResult == "CHECKAGAIN") return await this.services.remote.replicateAllToRemote(showingNotice); - // return !checkResult; - // } - // async _replicateAllFromServer(showingNotice: boolean = false): Promise { - // if (!this.services.appLifecycle.isReady()) return false; - // const ret = await this.core.replicator.replicateAllFromServer(this.settings, showingNotice); - // if (ret) return true; - // const checkResult = await this.services.replication.checkConnectionFailure(); - // if (checkResult == "CHECKAGAIN") return await this.services.remote.replicateAllFromRemote(showingNotice); - // return !checkResult; - // } - - override onBindFunction(core: LiveSyncCore, services: typeof core.services): void { - services.replicator.onBeforeReplicatorPublication.addHandler(this._onBeforeReplicatorPublication.bind(this)); - services.databaseEvents.onDatabaseInitialised.addHandler(this._everyOnDatabaseInitialized.bind(this)); - services.appLifecycle.onSettingLoaded.addHandler(this._everyOnloadAfterLoadSettings.bind(this)); - services.replication.parseSynchroniseResult.addHandler(this._parseReplicationResult.bind(this)); - - // --> These handlers can be separated. - const isOnlineAndCanReplicateWithHost = isOnlineAndCanReplicate.bind(null, this._unresolvedErrorManager, { - services: { - context: services.context, - API: services.API, - }, - serviceModules: {}, - }); - const canReplicateWithSecuritySeedWithHost = canReplicateWithSecuritySeed.bind( - null, - this._unresolvedErrorManager, - { - services: { - context: services.context, - replicator: services.replicator, - setting: services.setting, - }, - serviceModules: {}, - } - ); - services.replication.onBeforeReplicate.addHandler(isOnlineAndCanReplicateWithHost, 10); - services.replication.onPrepareCentralRemoteReplication.addHandler(canReplicateWithSecuritySeedWithHost); - // <-- End of handlers that can be separated. - services.replication.onBeforeReplicate.addHandler(this._everyBeforeReplicate.bind(this), 100); - services.replication.onReplicationFailed.addHandler(this.onReplicationFailed.bind(this)); - } -} diff --git a/src/modules/core/ModuleReplicator.unit.spec.ts b/src/modules/core/ModuleReplicator.unit.spec.ts deleted file mode 100644 index baa15e46..00000000 --- a/src/modules/core/ModuleReplicator.unit.spec.ts +++ /dev/null @@ -1,445 +0,0 @@ -import { describe, expect, it, vi } from "vitest"; -import { createServiceContext } from "@vrtmrz/livesync-commonlib/context"; -import { EVENT_SETTING_SAVED, eventHub } from "@/common/events"; -import type { ObsidianLiveSyncSettings } from "@vrtmrz/livesync-commonlib/compat/common/types"; -import { - CENTRAL_COMPATIBILITY_REJECTION_REASONS, - NO_INTERACTION, - USER_INITIATED_REPLICATION_AUTHORITY, - replicationFailed, -} from "@vrtmrz/livesync-commonlib/replication"; - -const chunkMocks = vi.hoisted(() => ({ - purgeUnreferencedChunks: vi.fn(async (_db: unknown, countOnly: boolean) => (countOnly ? 2 : 0)), - balanceChunkPurgedDBs: vi.fn(async () => undefined), -})); - -vi.mock("@vrtmrz/livesync-commonlib/compat/pouchdb/chunks", () => chunkMocks); -vi.mock("@vrtmrz/livesync-commonlib/compat/replication/couchdb/LiveSyncReplicator", () => ({ - LiveSyncCouchDBReplicator: class {}, -})); - -import { LiveSyncCouchDBReplicator } from "@vrtmrz/livesync-commonlib/compat/replication/couchdb/LiveSyncReplicator"; -import { ModuleReplicator } from "./ModuleReplicator"; - -describe("ModuleReplicator", () => { - it("refreshes the remote Security Seed before replication", async () => { - const read = vi.fn(async () => new Uint8Array([1])); - const dispose = vi.fn(async () => undefined); - const createRemoteResource = vi.fn(async () => ({ read, dispose })); - let prepareCentralRemoteReplication: ((showMessage: boolean) => Promise) | undefined; - const services = { - API: { isOnline: true }, - replicator: { - onBeforeReplicatorPublication: { addHandler: vi.fn() }, - createRemoteResource, - }, - setting: { currentSettings: () => ({}) }, - databaseEvents: { onDatabaseInitialised: { addHandler: vi.fn() } }, - appLifecycle: { onSettingLoaded: { addHandler: vi.fn() } }, - replication: { - parseSynchroniseResult: { addHandler: vi.fn() }, - onBeforeReplicate: { addHandler: vi.fn() }, - onPrepareCentralRemoteReplication: { - addHandler: vi.fn((handler: (showMessage: boolean) => Promise) => { - prepareCentralRemoteReplication = handler; - }), - }, - onReplicationFailed: { addHandler: vi.fn() }, - }, - }; - const module = { - _unresolvedErrorManager: { - showError: vi.fn(), - clearError: vi.fn(), - }, - _onBeforeReplicatorPublication: vi.fn(), - _everyOnDatabaseInitialized: vi.fn(), - _everyOnloadAfterLoadSettings: vi.fn(), - _parseReplicationResult: vi.fn(), - _everyBeforeReplicate: vi.fn(), - onReplicationFailed: vi.fn(), - }; - - ModuleReplicator.prototype.onBindFunction.call(module, {} as never, services as never); - expect(prepareCentralRemoteReplication).toBeDefined(); - - await prepareCentralRemoteReplication!(false); - - expect(createRemoteResource).toHaveBeenCalledWith("security-seed", {}); - expect(read).toHaveBeenCalledOnce(); - expect(dispose).toHaveBeenCalledOnce(); - }); - - it("keeps online and general pre-replication handlers for P2P while skipping central-remote Security Seed preparation", async () => { - const read = vi.fn(async () => new Uint8Array([1])); - const dispose = vi.fn(async () => undefined); - const createRemoteResource = vi.fn(async () => ({ read, dispose })); - const handlers = new Map Promise>(); - const centralRemoteHandlers: Array<(...args: unknown[]) => Promise> = []; - const addHandler = vi.fn((handler: (...args: unknown[]) => Promise, priority?: number) => { - handlers.set(priority ?? 0, handler); - }); - const services = { - API: { isOnline: true }, - replicator: { - onBeforeReplicatorPublication: { addHandler: vi.fn() }, - createRemoteResource, - }, - setting: { currentSettings: () => ({}) }, - databaseEvents: { onDatabaseInitialised: { addHandler: vi.fn() } }, - appLifecycle: { onSettingLoaded: { addHandler: vi.fn() } }, - replication: { - parseSynchroniseResult: { addHandler: vi.fn() }, - onBeforeReplicate: { addHandler }, - onPrepareCentralRemoteReplication: { - addHandler: vi.fn((handler: (...args: unknown[]) => Promise) => { - centralRemoteHandlers.push(handler); - }), - }, - onReplicationFailed: { addHandler: vi.fn() }, - }, - }; - const generalBeforeReplicate = vi.fn(async () => true); - const module = { - _unresolvedErrorManager: { - showError: vi.fn(), - clearError: vi.fn(), - }, - _onBeforeReplicatorPublication: vi.fn(), - _everyOnDatabaseInitialized: vi.fn(), - _everyOnloadAfterLoadSettings: vi.fn(), - _parseReplicationResult: vi.fn(), - _everyBeforeReplicate: generalBeforeReplicate, - onReplicationFailed: vi.fn(), - }; - - ModuleReplicator.prototype.onBindFunction.call(module, {} as never, services as never); - const online = handlers.get(10); - const securitySeed = centralRemoteHandlers[0]; - const general = handlers.get(100); - expect(online).toBeDefined(); - expect(securitySeed).toBeDefined(); - expect(general).toBeDefined(); - - await expect(online!(false)).resolves.toBe(true); - await expect(general!(false)).resolves.toBe(true); - - expect(generalBeforeReplicate).toHaveBeenCalledOnce(); - expect(createRemoteResource).not.toHaveBeenCalled(); - - await expect(securitySeed!(false)).resolves.toBe(true); - expect(createRemoteResource).toHaveBeenCalledOnce(); - expect(read).toHaveBeenCalledOnce(); - expect(dispose).toHaveBeenCalledOnce(); - }); - - it("reprocesses stored documents when the normal-file target filters change", async () => { - eventHub.offAll(); - const settings = { - handleFilenameCaseSensitive: false, - ignoreFiles: ".gitignore", - maxMTimeForReflectEvents: 0, - syncOnlyRegEx: "^E2E/allowed/.*", - syncIgnoreRegEx: "", - syncInternalFiles: false, - syncMaxSizeInMB: 0, - suspendParseReplicationResult: false, - useIgnoreFiles: false, - } as ObsidianLiveSyncSettings; - const services = { - context: createServiceContext(), - API: { - addLog: vi.fn(), - addCommand: vi.fn(), - registerWindow: vi.fn(), - addRibbonIcon: vi.fn(), - registerProtocolHandler: vi.fn(), - }, - appLifecycle: { - getUnresolvedMessages: { addHandler: vi.fn() }, - isSuspended: vi.fn(() => false), - }, - }; - const core = { - _services: services, - services, - settings, - } as any; - const module = new ModuleReplicator(core); - const reprocessStoredDocuments = vi.fn(async () => 1); - Object.assign(module.processor, { reprocessStoredDocuments }); - - try { - await (module as any)._everyOnloadAfterLoadSettings(); - eventHub.emitEvent(EVENT_SETTING_SAVED, { ...settings }); - await Promise.resolve(); - expect(reprocessStoredDocuments).not.toHaveBeenCalled(); - - Object.assign(settings, { syncOnlyRegEx: "" }); - eventHub.emitEvent(EVENT_SETTING_SAVED, { ...settings }); - await vi.waitFor(() => expect(reprocessStoredDocuments).toHaveBeenCalledOnce()); - - settings.syncMaxSizeInMB = 10; - eventHub.emitEvent(EVENT_SETTING_SAVED, { ...settings }); - await vi.waitFor(() => expect(reprocessStoredDocuments).toHaveBeenCalledTimes(2)); - } finally { - eventHub.offAll(); - } - }); - - it("uses the exact failed outcome and permits dialogue only with recovery authority", async () => { - const askResolvingMismatched = vi.fn(async (..._args: unknown[]) => undefined); - const failedSetPreferred = vi.fn(async (_setting: unknown) => undefined); - const failedReplicator = { setPreferredRemoteTweakSettings: failedSetPreferred }; - const replacementSetPreferred = vi.fn(async (_setting: unknown) => undefined); - const replacementReplicator = { - tweakSettingsMismatched: true, - preferredTweakValue: { customChunkSize: 99 }, - setPreferredRemoteTweakSettings: replacementSetPreferred, - }; - const context = { provider: {}, replicator: failedReplicator }; - const replacementContext = { provider: {}, replicator: replacementReplicator }; - const preferredTweakValue = { customChunkSize: 60 }; - const outcome = replicationFailed(new Error("mismatched"), { - reason: CENTRAL_COMPATIBILITY_REJECTION_REASONS.TWEAK_MISMATCH, - preferredTweakValue, - }); - const services = { - context: createServiceContext(), - API: { - addLog: vi.fn(), - addCommand: vi.fn(), - registerWindow: vi.fn(), - addRibbonIcon: vi.fn(), - registerProtocolHandler: vi.fn(), - }, - appLifecycle: { - getUnresolvedMessages: { addHandler: vi.fn() }, - }, - replicator: { - getActiveReplicator: vi.fn(() => replacementReplicator), - runWithActiveReplicatorContext: vi.fn(async (task: (context: unknown) => unknown) => - task(replacementContext) - ), - }, - tweakValue: { askResolvingMismatched }, - }; - const core = { - _services: services, - services, - settings: {}, - } as any; - const module = new ModuleReplicator(core); - - await (module as any).onReplicationFailed({ - context, - setting: {}, - outcome, - showMessage: false, - interaction: NO_INTERACTION, - }); - expect(askResolvingMismatched).not.toHaveBeenCalled(); - - await (module as any).onReplicationFailed({ - context, - setting: {}, - outcome, - showMessage: false, - interaction: { - kind: "permitted", - permissions: { ...USER_INITIATED_REPLICATION_AUTHORITY.permissions, failureRecovery: false }, - }, - }); - expect(askResolvingMismatched).not.toHaveBeenCalled(); - - await (module as any).onReplicationFailed({ - context, - setting: {}, - outcome, - showMessage: true, - interaction: USER_INITIATED_REPLICATION_AUTHORITY, - }); - expect(askResolvingMismatched).toHaveBeenCalledWith(preferredTweakValue, expect.any(Function)); - const updatePreferredRemote = askResolvingMismatched.mock.calls[0][1] as ( - setting: Record - ) => Promise; - await expect(updatePreferredRemote({ customChunkSize: 64 } as any)).resolves.toBe(false); - expect(failedSetPreferred).not.toHaveBeenCalled(); - expect(replacementSetPreferred).not.toHaveBeenCalled(); - }); - - it("writes a mismatch decision only through the still-active failed publication", async () => { - const setPreferredRemoteTweakSettings = vi.fn(async (_setting: unknown) => undefined); - const context = { provider: {}, replicator: { setPreferredRemoteTweakSettings } }; - let updatePreferredRemote: - | ((setting: Record) => Promise) - | undefined; - const askResolvingMismatched = vi.fn( - async (_preferred: unknown, update: (setting: Record) => Promise) => { - updatePreferredRemote = update; - } - ); - const services = { - context: createServiceContext(), - API: { - addLog: vi.fn(), - addCommand: vi.fn(), - registerWindow: vi.fn(), - addRibbonIcon: vi.fn(), - registerProtocolHandler: vi.fn(), - }, - appLifecycle: { getUnresolvedMessages: { addHandler: vi.fn() } }, - replicator: { - runWithActiveReplicatorContext: vi.fn(async (task: (activeContext: unknown) => unknown) => - task(context) - ), - }, - tweakValue: { askResolvingMismatched }, - }; - const module = new ModuleReplicator({ _services: services, services, settings: {} } as any); - - await (module as any).onReplicationFailed({ - context, - setting: {}, - outcome: replicationFailed(new Error("mismatched"), { - reason: CENTRAL_COMPATIBILITY_REJECTION_REASONS.TWEAK_MISMATCH, - preferredTweakValue: { customChunkSize: 60 }, - }), - showMessage: true, - interaction: USER_INITIATED_REPLICATION_AUTHORITY, - }); - - const effectiveSetting = { customChunkSize: 64 }; - await expect(updatePreferredRemote?.(effectiveSetting)).resolves.toBe(true); - expect(setPreferredRemoteTweakSettings).toHaveBeenCalledWith(effectiveSetting); - expect(setPreferredRemoteTweakSettings.mock.calls[0][0]).not.toBe(effectiveSetting); - }); - - it("does not apply an unlock selected for a replaced failed publication", async () => { - const failedMarkResolved = vi.fn(async () => undefined); - const replacementMarkResolved = vi.fn(async () => undefined); - const failedContext = { provider: {}, replicator: { markRemoteResolved: failedMarkResolved } }; - const replacementContext = { provider: {}, replicator: { markRemoteResolved: replacementMarkResolved } }; - const runWithActiveReplicatorContext = vi.fn(async (task: (context: unknown) => unknown) => - task(replacementContext) - ); - const services = { - context: createServiceContext(), - API: { - addLog: vi.fn(), - addCommand: vi.fn(), - registerWindow: vi.fn(), - addRibbonIcon: vi.fn(), - registerProtocolHandler: vi.fn(), - }, - appLifecycle: { - getUnresolvedMessages: { addHandler: vi.fn() }, - scheduleRestart: vi.fn(), - }, - replicator: { runWithActiveReplicatorContext }, - }; - const core = { - _services: services, - services, - settings: {}, - confirm: { - askSelectStringDialogue: vi.fn(async (_message: string, choices: string[]) => choices[1]), - }, - rebuilder: { scheduleFetch: vi.fn() }, - } as any; - const module = new ModuleReplicator(core); - - await (module as any).onReplicationFailed({ - context: failedContext, - setting: {}, - outcome: replicationFailed(new Error("locked"), { - reason: CENTRAL_COMPATIBILITY_REJECTION_REASONS.NODE_LOCKED, - }), - showMessage: true, - interaction: USER_INITIATED_REPLICATION_AUTHORITY, - }); - - expect(runWithActiveReplicatorContext).toHaveBeenCalledOnce(); - expect(failedMarkResolved).not.toHaveBeenCalled(); - expect(replacementMarkResolved).not.toHaveBeenCalled(); - }); -}); - -describe("compatibility: cleaned-remote reconciliation for IndexedDB clients", () => { - it("keeps its finite replication and balancing work inside the shared activity boundary", async () => { - const activityFinished = vi.fn(); - const runBoundedRemoteActivity = vi.fn(async (task: () => unknown) => { - try { - return await task(); - } finally { - activityFinished(); - } - }); - const runFiniteReplicationActivity = vi.fn(async (task: () => unknown) => await task()); - const openOneShotReplication = vi.fn(async () => true); - const remoteDatabase = { - close: vi.fn(async () => undefined), - }; - const close = vi.fn(async () => undefined); - const activeReplicator = Object.assign(new LiveSyncCouchDBReplicator({} as any), { - connectRemoteCouchDBWithSetting: vi.fn(async () => ({ db: remoteDatabase, close })), - openOneShotReplication, - markRemoteResolved: vi.fn(async () => undefined), - }); - const expectedContext = { provider: {}, replicator: activeReplicator }; - const runWithActiveReplicatorContext = vi.fn(async (task: (context: unknown) => unknown) => - task(expectedContext) - ); - const services = { - context: createServiceContext(), - API: { - addLog: vi.fn(), - addCommand: vi.fn(), - registerWindow: vi.fn(), - addRibbonIcon: vi.fn(), - registerProtocolHandler: vi.fn(), - isMobile: vi.fn(() => false), - }, - setting: { saveSettingData: vi.fn(async () => undefined) }, - appLifecycle: { - getUnresolvedMessages: { addHandler: vi.fn() }, - }, - replicator: { - getActiveReplicator: vi.fn(() => activeReplicator), - runBoundedRemoteActivity, - runFiniteReplicationActivity, - runWithActiveReplicatorContext, - }, - }; - const localDatabase = { - localDatabase: {}, - clearCaches: vi.fn(), - }; - const core = { - _services: services, - services, - settings: {}, - localDatabase, - confirm: { confirmWithMessage: vi.fn(async () => "Cleanup") }, - } as any; - const module = new ModuleReplicator(core); - - await module.cleaned(true, {} as ObsidianLiveSyncSettings, expectedContext as never); - - expect(runBoundedRemoteActivity).toHaveBeenCalledWith(expect.any(Function), { - label: "database-cleanup", - }); - expect(runFiniteReplicationActivity).toHaveBeenCalledWith(expect.any(Function), { - label: "replication", - }); - expect(runWithActiveReplicatorContext).toHaveBeenCalledOnce(); - expect(openOneShotReplication).toHaveBeenCalledOnce(); - expect(openOneShotReplication.mock.invocationCallOrder[0]).toBeLessThan( - activityFinished.mock.invocationCallOrder[0] - ); - expect(chunkMocks.balanceChunkPurgedDBs).toHaveBeenCalledOnce(); - expect(close).toHaveBeenCalledOnce(); - expect(close.mock.invocationCallOrder[0]).toBeLessThan(activityFinished.mock.invocationCallOrder[0]); - }); -}); diff --git a/src/modules/core/ReplicateResultProcessor.ts b/src/serviceFeatures/replication/ReplicateResultProcessor.ts similarity index 91% rename from src/modules/core/ReplicateResultProcessor.ts rename to src/serviceFeatures/replication/ReplicateResultProcessor.ts index e6840439..e4d75a20 100644 --- a/src/modules/core/ReplicateResultProcessor.ts +++ b/src/serviceFeatures/replication/ReplicateResultProcessor.ts @@ -6,8 +6,8 @@ import { type EntryLeaf, type LoadedEntry, type MetaEntry, + type ObsidianLiveSyncSettings, } from "@vrtmrz/livesync-commonlib/compat/common/types"; -import type { ModuleReplicator } from "./ModuleReplicator"; import { isChunk } from "@vrtmrz/livesync-commonlib/compat/common/typeUtils"; import { LOG_LEVEL_DEBUG, @@ -28,12 +28,34 @@ import { promiseWithResolvers, type PromiseWithResolvers } from "octagonal-wheel const KV_KEY_REPLICATION_RESULT_PROCESSOR_SNAPSHOT = "replicationResultProcessorSnapshot"; const REPROCESS_BATCH_SIZE = 100; -type LocalApplicationActivityOwner = { - runBoundedLocalApplicationActivity( +type ReplicateResultProcessorSettings = Pick< + ObsidianLiveSyncSettings, + "maxMTimeForReflectEvents" | "suspendParseReplicationResult" +>; +type ReplicateResultProcessorServices = Pick< + LiveSyncBaseCore["services"], + "appLifecycle" | "path" | "replication" | "vault" +>; + +/** + * Narrow collaborators for applying replicated documents. + * + * `requestActiveReplicatorRetirement` starts the owner transition without + * awaiting it. Result application can still be running inside work admitted by + * that owner, so awaiting retirement here could make each side wait for the + * other to finish. + */ +interface ReplicateResultProcessorContext { + readonly currentSettings: () => ReplicateResultProcessorSettings; + readonly keyValueDB: LiveSyncBaseCore["kvDB"]; + readonly localDatabase: LiveSyncBaseCore["localDatabase"]; + readonly requestActiveReplicatorRetirement: () => void; + readonly runLocalApplicationActivity: ( task: () => T | PromiseLike, options?: { label?: string } - ): Promise; -}; + ) => Promise; + readonly services: ReplicateResultProcessorServices; +} type ReplicateResultProcessorState = { queued: PouchDB.Core.ExistingDocument[]; processing: PouchDB.Core.ExistingDocument[]; @@ -52,20 +74,13 @@ export class ReplicateResultProcessor { private logError(e: unknown) { Logger(e, LOG_LEVEL_VERBOSE); } - private replicator: ModuleReplicator; + constructor(private readonly context: ReplicateResultProcessorContext) {} - constructor(replicator: ModuleReplicator) { - this.replicator = replicator; + private get localDatabase() { + return this.context.localDatabase; } - - get localDatabase() { - return this.replicator.core.localDatabase; - } - get services() { - return this.replicator.core.services; - } - get core(): LiveSyncBaseCore { - return this.replicator.core; + private get services() { + return this.context.services; } getPath(entry: AnyEntry): string { @@ -89,9 +104,9 @@ export class ReplicateResultProcessor { public get isSuspended() { return ( this._suspended || - !this.core.services.appLifecycle.isReady || - this.replicator.settings.suspendParseReplicationResult || - this.core.services.appLifecycle.isSuspended() + !this.services.appLifecycle.isReady || + this.context.currentSettings().suspendParseReplicationResult || + this.services.appLifecycle.isSuspended() ); } @@ -104,7 +119,7 @@ export class ReplicateResultProcessor { queued: this._queuedChanges.slice(), processing: this._processingChanges.slice(), } satisfies ReplicateResultProcessorState; - await this.core.kvDB.set(KV_KEY_REPLICATION_RESULT_PROCESSOR_SNAPSHOT, snapshot); + await this.context.keyValueDB.set(KV_KEY_REPLICATION_RESULT_PROCESSOR_SNAPSHOT, snapshot); this.log( `Snapshot taken. Queued: ${snapshot.queued.length}, Processing: ${snapshot.processing.length}`, LOG_LEVEL_DEBUG @@ -126,7 +141,7 @@ export class ReplicateResultProcessor { * Restore from snapshot. */ public async restoreFromSnapshot() { - const snapshot = await this.core.kvDB.get( + const snapshot = await this.context.keyValueDB.get( KV_KEY_REPLICATION_RESULT_PROCESSOR_SNAPSHOT ); if (snapshot) { @@ -231,8 +246,8 @@ export class ReplicateResultProcessor { if (change.type == "versioninfo") { this.log(`Version info document received: ${change._id}`, LOG_LEVEL_VERBOSE); if (change.version > VER) { - // Incompatible version, stop replication. - this.core.replicator.closeReplication(); + // Fence and retire the active publication through its owner. + this.context.requestActiveReplicatorRetirement(); this.log( `Remote database updated to incompatible version. update your Self-hosted LiveSync plugin.`, LOG_LEVEL_NOTICE @@ -277,15 +292,10 @@ export class ReplicateResultProcessor { const activityDone = promiseWithResolvers(); this._processingActivityDone = activityDone; - const activityOwner = this.services.replicator as typeof this.services.replicator & - Partial; - this._processingActivity = ( - activityOwner.runBoundedLocalApplicationActivity - ? activityOwner.runBoundedLocalApplicationActivity(() => activityDone.promise, { - label: "replicated-document-application", - }) - : activityDone.promise - ) + this._processingActivity = this.context + .runLocalApplicationActivity(() => activityDone.promise, { + label: "replicated-document-application", + }) .catch((error) => this.logError(error)) .finally(() => { if (this._processingActivityDone === activityDone) this._processingActivityDone = undefined; @@ -392,7 +402,7 @@ export class ReplicateResultProcessor { try { if (isAnyNote(change)) { const docMtime = change.mtime ?? 0; - const maxMTime = this.replicator.settings.maxMTimeForReflectEvents; + const maxMTime = this.context.currentSettings().maxMTimeForReflectEvents; if (maxMTime > 0 && docMtime > maxMTime) { const docPath = this.getPath(change); this.log( diff --git a/src/modules/core/ReplicateResultProcessor.unit.spec.ts b/src/serviceFeatures/replication/ReplicateResultProcessor.unit.spec.ts similarity index 81% rename from src/modules/core/ReplicateResultProcessor.unit.spec.ts rename to src/serviceFeatures/replication/ReplicateResultProcessor.unit.spec.ts index 61c7af61..08225a37 100644 --- a/src/modules/core/ReplicateResultProcessor.unit.spec.ts +++ b/src/serviceFeatures/replication/ReplicateResultProcessor.unit.spec.ts @@ -1,7 +1,7 @@ import { promiseWithResolvers } from "octagonal-wheels/promises"; import { reactiveSource } from "octagonal-wheels/dataobject/reactive"; import { describe, expect, it, vi } from "vitest"; -import type { EntryDoc } from "@vrtmrz/livesync-commonlib/compat/common/types"; +import { VER, type EntryDoc } from "@vrtmrz/livesync-commonlib/compat/common/types"; import { ReplicateResultProcessor } from "./ReplicateResultProcessor"; function note(id: string): PouchDB.Core.ExistingDocument { @@ -28,6 +28,7 @@ function setup(options: SetupOptions = {}) { const processSynchroniseResult = vi.fn(options.processSynchroniseResult ?? (async () => undefined)); const setSnapshot = vi.fn(options.setSnapshot ?? (async () => undefined)); const runBoundedLocalApplicationActivity = vi.fn(async (task: () => Promise) => await task()); + const onCloseActiveReplication = vi.fn(async () => true); const core = { services: { appLifecycle: { isReady: true, isSuspended: () => false }, @@ -40,7 +41,7 @@ function setup(options: SetupOptions = {}) { processOptionalSynchroniseResult: vi.fn(async () => false), processSynchroniseResult, }, - replicator: { runBoundedLocalApplicationActivity }, + replicator: { onCloseActiveReplication, runBoundedLocalApplicationActivity }, vault: { isTargetFile: vi.fn(async () => true), isFileSizeTooLarge: vi.fn(() => false), @@ -52,16 +53,40 @@ function setup(options: SetupOptions = {}) { getRaw: vi.fn(async (id: string) => ({ _id: id, _rev: "1-test" })), getDBEntryFromMeta: vi.fn(async (entry: object) => ({ ...entry, data: "x" })), }, - replicator: { closeReplication: vi.fn() }, }; const processor = new ReplicateResultProcessor({ - core, - settings: { maxMTimeForReflectEvents: 0, suspendParseReplicationResult: false }, + currentSettings: () => ({ maxMTimeForReflectEvents: 0, suspendParseReplicationResult: false }), + keyValueDB: core.kvDB, + localDatabase: core.localDatabase, + requestActiveReplicatorRetirement: () => { + void onCloseActiveReplication(); + }, + runLocalApplicationActivity: runBoundedLocalApplicationActivity, + services: core.services, } as never); - return { processor, processSynchroniseResult, runBoundedLocalApplicationActivity }; + return { + onCloseActiveReplication, + processor, + processSynchroniseResult, + runBoundedLocalApplicationActivity, + }; } describe("ReplicateResultProcessor", () => { + it("retires active ownership when a newer remote version is observed", async () => { + const { onCloseActiveReplication, processor } = setup(); + const versionInfo = { + _id: "versioninfo", + _rev: "1-test", + type: "versioninfo", + version: VER + 1, + } as unknown as PouchDB.Core.ExistingDocument; + + processor.enqueueAll([versionInfo]); + + await vi.waitFor(() => expect(onCloseActiveReplication).toHaveBeenCalledOnce()); + }); + it("scans normal-file metadata without loading chunk documents and requeues it", async () => { const documents = [ { _id: "first", _rev: "1-a", type: "plain", path: "first.md" }, @@ -71,7 +96,7 @@ describe("ReplicateResultProcessor", () => { yield* documents; }); const processor = new ReplicateResultProcessor({ - core: { localDatabase: { findAllNormalDocs } }, + localDatabase: { findAllNormalDocs }, } as never); const enqueueAll = vi.spyOn(processor, "enqueueAll").mockImplementation(() => undefined); diff --git a/src/serviceFeatures/replication/automaticTriggers.ts b/src/serviceFeatures/replication/automaticTriggers.ts new file mode 100644 index 00000000..5068ab27 --- /dev/null +++ b/src/serviceFeatures/replication/automaticTriggers.ts @@ -0,0 +1,71 @@ +import type { ObsidianLiveSyncSettings } from "@vrtmrz/livesync-commonlib/compat/common/types"; +import { fireAndForget } from "octagonal-wheels/promises"; +import { scheduleTask } from "octagonal-wheels/concurrency/task"; +import { EVENT_FILE_SAVED, EVENT_SETTING_SAVED, eventHub } from "@/common/events"; + +type ReflectionFilterSettings = Pick< + ObsidianLiveSyncSettings, + | "handleFilenameCaseSensitive" + | "ignoreFiles" + | "maxMTimeForReflectEvents" + | "syncIgnoreRegEx" + | "syncInternalFiles" + | "syncMaxSizeInMB" + | "syncOnlyRegEx" + | "useIgnoreFiles" +>; + +interface AutomaticReplicationTriggerContext { + readonly currentSettings: () => ObsidianLiveSyncSettings; + readonly isSuspended: () => boolean; + readonly replicateDatabaseEvent: () => Promise; + readonly reprocessStoredDocuments: () => Promise; + readonly resumeResultApplication: () => void; + readonly suspendResultApplication: () => void; +} + +function normalFileReflectionFilterSignature(settings: ReflectionFilterSettings): string { + return JSON.stringify({ + handleFilenameCaseSensitive: settings.handleFilenameCaseSensitive ?? false, + ignoreFiles: settings.ignoreFiles ?? "", + maxMTimeForReflectEvents: settings.maxMTimeForReflectEvents ?? 0, + syncIgnoreRegEx: settings.syncIgnoreRegEx ?? "", + syncInternalFiles: settings.syncInternalFiles ?? false, + syncMaxSizeInMB: settings.syncMaxSizeInMB ?? 0, + syncOnlyRegEx: settings.syncOnlyRegEx ?? "", + useIgnoreFiles: settings.useIgnoreFiles ?? false, + }); +} + +/** + * Create the settings-loaded handler which installs automatic replication and + * result-application reactions. The returned closure owns the previous filter + * signature; it is private composition state rather than a shared service. + */ +export function createAutomaticReplicationTriggers(context: AutomaticReplicationTriggerContext) { + let reflectionFilterSignature: string | undefined; + + return function initialiseAutomaticReplicationTriggers(): Promise { + reflectionFilterSignature = normalFileReflectionFilterSignature(context.currentSettings()); + eventHub.onEvent(EVENT_FILE_SAVED, () => { + if (context.currentSettings().syncOnSave && !context.isSuspended()) { + scheduleTask("perform-replicate-after-save", 250, () => context.replicateDatabaseEvent()); + } + }); + eventHub.onEvent(EVENT_SETTING_SAVED, (settings) => { + const previousReflectionFilter = reflectionFilterSignature; + const nextReflectionFilter = normalFileReflectionFilterSignature(settings); + reflectionFilterSignature = nextReflectionFilter; + if (settings.suspendParseReplicationResult) { + context.suspendResultApplication(); + } else { + context.resumeResultApplication(); + } + if (previousReflectionFilter !== undefined && previousReflectionFilter !== nextReflectionFilter) { + fireAndForget(() => context.reprocessStoredDocuments()); + } + }); + + return Promise.resolve(true); + }; +} diff --git a/src/modules/core/AutomaticReplicationTriggers.unit.spec.ts b/src/serviceFeatures/replication/automaticTriggers.unit.spec.ts similarity index 78% rename from src/modules/core/AutomaticReplicationTriggers.unit.spec.ts rename to src/serviceFeatures/replication/automaticTriggers.unit.spec.ts index 713cdb5a..7575b8b9 100644 --- a/src/modules/core/AutomaticReplicationTriggers.unit.spec.ts +++ b/src/serviceFeatures/replication/automaticTriggers.unit.spec.ts @@ -4,9 +4,10 @@ import { DEFAULT_SETTINGS, REMOTE_P2P, type FilePathWithPrefix, + type ObsidianLiveSyncSettings, } from "@vrtmrz/livesync-commonlib/compat/common/types"; import { NO_INTERACTION } from "@vrtmrz/livesync-commonlib/replication"; -import { EVENT_FILE_SAVED, eventHub } from "@/common/events"; +import { EVENT_FILE_SAVED, EVENT_SETTING_SAVED, eventHub } from "@/common/events"; const taskMocks = vi.hoisted(() => ({ scheduleTask: vi.fn((_key: string, _delay: number, task: () => unknown) => task()), @@ -14,15 +15,15 @@ const taskMocks = vi.hoisted(() => ({ vi.mock("octagonal-wheels/concurrency/task", () => taskMocks); -import { ModuleConflictResolver } from "../coreFeatures/ModuleConflictResolver"; -import { ModuleObsidianEvents } from "../essentialObsidian/ModuleObsidianEvents"; +import { ModuleConflictResolver } from "@/modules/coreFeatures/ModuleConflictResolver"; +import { ModuleObsidianEvents } from "@/modules/essentialObsidian/ModuleObsidianEvents"; import { createReplicationSchedulingContext, realiseReplicationScheduling, resumeReplicationScheduling, runPeriodicReplication, } from "@/serviceFeatures/replicationScheduling"; -import { ModuleReplicator } from "./ModuleReplicator"; +import { createAutomaticReplicationTriggers } from "./automaticTriggers"; function createApi() { return { @@ -124,24 +125,22 @@ describe("automatic replication triggers while P2P is active", () => { }); it("keeps database-save synchronisation on the event replication boundary", async () => { - const replicateUnattendedByEvent = vi.fn(async () => ({ status: "completed" as const })); + const replicateUnattendedByEvent = vi.fn(async (_request: unknown) => ({ status: "completed" as const })); const settings = p2pSettings({ syncOnSave: true }); - const services = { - appLifecycle: { isSuspended: vi.fn(() => false) }, - replication: { replicateUnattendedByEvent }, - }; - const module = { - core: { services, settings }, - services, - settings, - getNormalFileReflectionFilterSignature: ( - ModuleReplicator.prototype as unknown as { - getNormalFileReflectionFilterSignature: (value: typeof settings) => string; - } - ).getNormalFileReflectionFilterSignature, - }; + const initialise = createAutomaticReplicationTriggers({ + currentSettings: () => settings, + isSuspended: vi.fn(() => false), + replicateDatabaseEvent: () => + replicateUnattendedByEvent({ + trigger: "database-event", + interaction: NO_INTERACTION, + }), + reprocessStoredDocuments: vi.fn(async () => 0), + resumeResultApplication: vi.fn(), + suspendResultApplication: vi.fn(), + }); - await (ModuleReplicator.prototype as any)._everyOnloadAfterLoadSettings.call(module); + await initialise(); eventHub.emitEvent(EVENT_FILE_SAVED); await vi.waitFor(() => expect(replicateUnattendedByEvent).toHaveBeenCalledOnce()); @@ -151,6 +150,43 @@ describe("automatic replication triggers while P2P is active", () => { }); }); + it("reprocesses stored documents when normal-file target filters change", async () => { + const settings = { + ...DEFAULT_SETTINGS, + ignoreFiles: ".gitignore", + syncOnlyRegEx: "^E2E/allowed/.*", + } as ObsidianLiveSyncSettings; + const reprocessStoredDocuments = vi.fn(async () => 1); + const resumeResultApplication = vi.fn(); + const suspendResultApplication = vi.fn(); + const initialise = createAutomaticReplicationTriggers({ + currentSettings: () => settings, + isSuspended: vi.fn(() => false), + replicateDatabaseEvent: vi.fn(async () => undefined), + reprocessStoredDocuments, + resumeResultApplication, + suspendResultApplication, + }); + + await initialise(); + eventHub.emitEvent(EVENT_SETTING_SAVED, { ...settings }); + await Promise.resolve(); + expect(reprocessStoredDocuments).not.toHaveBeenCalled(); + expect(resumeResultApplication).toHaveBeenCalledOnce(); + expect(suspendResultApplication).not.toHaveBeenCalled(); + + eventHub.emitEvent(EVENT_SETTING_SAVED, { ...settings, suspendParseReplicationResult: true }); + expect(suspendResultApplication).toHaveBeenCalledOnce(); + + Object.assign(settings, { syncOnlyRegEx: "" }); + eventHub.emitEvent(EVENT_SETTING_SAVED, { ...settings }); + await vi.waitFor(() => expect(reprocessStoredDocuments).toHaveBeenCalledOnce()); + + settings.syncMaxSizeInMB = 10; + eventHub.emitEvent(EVENT_SETTING_SAVED, { ...settings }); + await vi.waitFor(() => expect(reprocessStoredDocuments).toHaveBeenCalledTimes(2)); + }); + it("keeps editor-save synchronisation on the event replication boundary", async () => { const { module, replicateUnattendedByEvent, save, saveCommand } = createObsidianEventHarness({ syncOnEditorSave: true, diff --git a/src/serviceFeatures/replication/centralCompatibilityRecovery.ts b/src/serviceFeatures/replication/centralCompatibilityRecovery.ts new file mode 100644 index 00000000..28ebd029 --- /dev/null +++ b/src/serviceFeatures/replication/centralCompatibilityRecovery.ts @@ -0,0 +1,194 @@ +import type { ObsidianLiveSyncSettings } from "@vrtmrz/livesync-commonlib/compat/common/types"; +import { LOG_LEVEL_INFO, LOG_LEVEL_NOTICE, Logger } from "octagonal-wheels/common/logger"; +import { skipIfDuplicated } from "octagonal-wheels/concurrency/lock"; +import { balanceChunkPurgedDBs, purgeUnreferencedChunks } from "@vrtmrz/livesync-commonlib/compat/pouchdb/chunks"; +import { LiveSyncCouchDBReplicator } from "@vrtmrz/livesync-commonlib/compat/replication/couchdb/LiveSyncReplicator"; +import { + CENTRAL_COMPATIBILITY_REJECTION_REASONS, + type ReplicationFailureRequest, +} from "@vrtmrz/livesync-commonlib/replication"; +import { $msg } from "@/common/translation"; +import { usesLegacyIndexedDBAdapter } from "@/common/compatibilitySettings"; +import type { LiveSyncBaseCore } from "@/LiveSyncBaseCore"; + +type CentralCompatibilityRecoveryServices = Pick< + LiveSyncBaseCore["services"], + "API" | "appLifecycle" | "replicator" | "tweakValue" +>; + +interface CentralCompatibilityRecoveryContext { + readonly confirm: LiveSyncBaseCore["confirm"]; + readonly localDatabase: LiveSyncBaseCore["localDatabase"]; + readonly rebuilder: LiveSyncBaseCore["rebuilder"]; + readonly services: CentralCompatibilityRecoveryServices; +} + +/** + * Compose central compatibility recovery around the exact failed publication. + * Remote mutations re-admit that publication and become no-ops after a + * replacement; the failure result is never re-read from the current instance. + */ +export function createCentralCompatibilityRecovery(context: CentralCompatibilityRecoveryContext) { + async function reconcileCleanedRemote( + showMessage: boolean, + setting: ObsidianLiveSyncSettings, + expectedContext: ReplicationFailureRequest["context"] + ) { + Logger("The remote database has been cleaned.", showMessage ? LOG_LEVEL_NOTICE : LOG_LEVEL_INFO); + await skipIfDuplicated("cleanup", async () => { + const count = await purgeUnreferencedChunks(context.localDatabase.localDatabase, true); + const message = `The remote database has been cleaned up. +To synchronize, this device must be also cleaned up. ${count} chunk(s) will be erased from this device. +However, If there are many chunks to be deleted, maybe fetching again is faster. +We will lose the history of this device if we fetch the remote database again. +Even if you choose to clean up, you will see this option again if you exit Obsidian and then synchronise again.`; + const CHOICE_FETCH = "Fetch again"; + const CHOICE_CLEAN = "Cleanup"; + const CHOICE_DISMISS = "Dismiss"; + const selected = await context.confirm.confirmWithMessage( + "Cleaned", + message, + [CHOICE_FETCH, CHOICE_CLEAN, CHOICE_DISMISS], + CHOICE_DISMISS, + 30 + ); + if (selected == CHOICE_FETCH) { + await context.rebuilder.$performRebuildDB("localOnly"); + } + if (selected != CHOICE_CLEAN) return; + + await context.services.replicator.runBoundedRemoteActivity( + () => + context.services.replicator.runWithActiveReplicatorContext(async (activeContext) => { + if (activeContext !== expectedContext) return; + const replicator = activeContext.replicator; + if (!(replicator instanceof LiveSyncCouchDBReplicator)) return; + const remoteDatabase = await replicator.connectRemoteCouchDBWithSetting( + setting, + context.services.API.isMobile(), + true + ); + if (typeof remoteDatabase == "string") { + Logger(remoteDatabase, LOG_LEVEL_NOTICE); + return false; + } + + try { + await purgeUnreferencedChunks(context.localDatabase.localDatabase, false); + context.localDatabase.clearCaches(); + const replicated = await context.services.replicator.runFiniteReplicationActivity( + () => replicator.openOneShotReplication(setting, showMessage, false, "sync", true), + { label: "replication" } + ); + if (replicated) { + await balanceChunkPurgedDBs(context.localDatabase.localDatabase, remoteDatabase.db); + await purgeUnreferencedChunks(context.localDatabase.localDatabase, false); + context.localDatabase.clearCaches(); + await replicator.markRemoteResolved(setting); + Logger( + "The local database has been cleaned up.", + showMessage ? LOG_LEVEL_NOTICE : LOG_LEVEL_INFO + ); + } else { + Logger( + "Replication has been cancelled. Please try it again.", + showMessage ? LOG_LEVEL_NOTICE : LOG_LEVEL_INFO + ); + } + } finally { + await remoteDatabase.close(); + } + }), + { label: "database-cleanup" } + ); + }); + } + + async function handleReplicationFailure(request: ReplicationFailureRequest): Promise { + const { context: failedContext, interaction, outcome, setting, showMessage } = request; + if (!showMessage) { + // Automatic requests may report the failure, but must not enter + // tweak, lock, fetch, unlock, or cleanup dialogues. + Logger("Replication failed on an unattended path.", LOG_LEVEL_INFO); + return false; + } + if (interaction.kind !== "permitted" || !interaction.permissions.failureRecovery) return false; + const recovery = outcome.recoveryHint; + if (!recovery) return false; + + if ( + recovery.reason === CENTRAL_COMPATIBILITY_REJECTION_REASONS.TWEAK_MISMATCH && + recovery.preferredTweakValue + ) { + await context.services.tweakValue.askResolvingMismatched( + recovery.preferredTweakValue, + async (effectiveSetting) => { + let updated = false; + await context.services.replicator.runWithActiveReplicatorContext(async (activeContext) => { + if (activeContext !== failedContext) return; + const candidate = activeContext.replicator as typeof activeContext.replicator & { + setPreferredRemoteTweakSettings?: (setting: ObsidianLiveSyncSettings) => Promise; + }; + if (typeof candidate.setPreferredRemoteTweakSettings !== "function") return; + await candidate.setPreferredRemoteTweakSettings({ ...effectiveSetting }); + updated = true; + }); + return updated; + } + ); + return false; + } + + if ( + recovery.reason !== CENTRAL_COMPATIBILITY_REJECTION_REASONS.NODE_LOCKED && + recovery.reason !== CENTRAL_COMPATIBILITY_REJECTION_REASONS.NODE_CLEANED + ) { + return false; + } + if ( + recovery.reason === CENTRAL_COMPATIBILITY_REJECTION_REASONS.NODE_CLEANED && + usesLegacyIndexedDBAdapter(setting) + ) { + await reconcileCleanedRemote(showMessage, setting, failedContext); + return false; + } + + const message = $msg("Replicator.Dialogue.Locked.Message"); + const CHOICE_FETCH = $msg("Replicator.Dialogue.Locked.Action.Fetch"); + const CHOICE_DISMISS = $msg("Replicator.Dialogue.Locked.Action.Dismiss"); + const CHOICE_UNLOCK = $msg("Replicator.Dialogue.Locked.Action.Unlock"); + const selected = await context.confirm.askSelectStringDialogue( + message, + [CHOICE_FETCH, CHOICE_UNLOCK, CHOICE_DISMISS], + { + title: $msg("Replicator.Dialogue.Locked.Title"), + defaultAction: CHOICE_DISMISS, + timeout: 60, + } + ); + if (selected == CHOICE_FETCH) { + Logger($msg("Replicator.Dialogue.Locked.Message.Fetch"), LOG_LEVEL_NOTICE); + await context.rebuilder.scheduleFetch(); + context.services.appLifecycle.scheduleRestart(); + return false; + } + if (selected != CHOICE_UNLOCK) return false; + + let unlocked = false; + await context.services.replicator.runWithActiveReplicatorContext(async (activeContext) => { + if (activeContext !== failedContext) return; + const replicator = activeContext.replicator as typeof activeContext.replicator & { + markRemoteResolved(setting: ObsidianLiveSyncSettings): Promise; + }; + if (typeof replicator.markRemoteResolved !== "function") return; + await replicator.markRemoteResolved(setting); + unlocked = true; + }); + if (unlocked) { + Logger($msg("Replicator.Dialogue.Locked.Message.Unlocked"), LOG_LEVEL_NOTICE); + } + return false; + } + + return Object.freeze({ handleReplicationFailure, reconcileCleanedRemote }); +} diff --git a/src/serviceFeatures/replication/centralCompatibilityRecovery.unit.spec.ts b/src/serviceFeatures/replication/centralCompatibilityRecovery.unit.spec.ts new file mode 100644 index 00000000..83d8fbe5 --- /dev/null +++ b/src/serviceFeatures/replication/centralCompatibilityRecovery.unit.spec.ts @@ -0,0 +1,229 @@ +import { describe, expect, it, vi } from "vitest"; +import type { ObsidianLiveSyncSettings } from "@vrtmrz/livesync-commonlib/compat/common/types"; +import { + CENTRAL_COMPATIBILITY_REJECTION_REASONS, + NO_INTERACTION, + USER_INITIATED_REPLICATION_AUTHORITY, + replicationFailed, +} from "@vrtmrz/livesync-commonlib/replication"; + +const chunkMocks = vi.hoisted(() => ({ + purgeUnreferencedChunks: vi.fn(async (_database: unknown, countOnly: boolean) => (countOnly ? 2 : 0)), + balanceChunkPurgedDBs: vi.fn(async () => undefined), +})); + +vi.mock("@vrtmrz/livesync-commonlib/compat/pouchdb/chunks", () => chunkMocks); +vi.mock("@vrtmrz/livesync-commonlib/compat/replication/couchdb/LiveSyncReplicator", () => ({ + LiveSyncCouchDBReplicator: class {}, +})); + +import { LiveSyncCouchDBReplicator } from "@vrtmrz/livesync-commonlib/compat/replication/couchdb/LiveSyncReplicator"; +import { createCentralCompatibilityRecovery } from "./centralCompatibilityRecovery"; + +describe("central compatibility recovery", () => { + it("uses the exact failed outcome and permits dialogue only with recovery authority", async () => { + const askResolvingMismatched = vi.fn(async (..._arguments: unknown[]) => undefined); + const failedSetPreferred = vi.fn(async (_setting: unknown) => undefined); + const failedReplicator = { setPreferredRemoteTweakSettings: failedSetPreferred }; + const replacementSetPreferred = vi.fn(async (_setting: unknown) => undefined); + const replacementReplicator = { + tweakSettingsMismatched: true, + preferredTweakValue: { customChunkSize: 99 }, + setPreferredRemoteTweakSettings: replacementSetPreferred, + }; + const failedContext = { provider: {}, replicator: failedReplicator }; + const replacementContext = { provider: {}, replicator: replacementReplicator }; + const preferredTweakValue = { customChunkSize: 60 }; + const outcome = replicationFailed(new Error("mismatched"), { + reason: CENTRAL_COMPATIBILITY_REJECTION_REASONS.TWEAK_MISMATCH, + preferredTweakValue, + }); + const recovery = createCentralCompatibilityRecovery({ + confirm: {}, + localDatabase: {}, + rebuilder: {}, + services: { + appLifecycle: {}, + API: {}, + replicator: { + runWithActiveReplicatorContext: vi.fn(async (task: (context: unknown) => unknown) => + task(replacementContext) + ), + }, + tweakValue: { askResolvingMismatched }, + }, + } as never); + + await recovery.handleReplicationFailure({ + context: failedContext, + setting: {}, + outcome, + showMessage: false, + interaction: NO_INTERACTION, + } as never); + expect(askResolvingMismatched).not.toHaveBeenCalled(); + + await recovery.handleReplicationFailure({ + context: failedContext, + setting: {}, + outcome, + showMessage: false, + interaction: { + kind: "permitted", + permissions: { ...USER_INITIATED_REPLICATION_AUTHORITY.permissions, failureRecovery: false }, + }, + } as never); + expect(askResolvingMismatched).not.toHaveBeenCalled(); + + await recovery.handleReplicationFailure({ + context: failedContext, + setting: {}, + outcome, + showMessage: true, + interaction: USER_INITIATED_REPLICATION_AUTHORITY, + } as never); + expect(askResolvingMismatched).toHaveBeenCalledWith(preferredTweakValue, expect.any(Function)); + const updatePreferredRemote = askResolvingMismatched.mock.calls[0][1] as ( + setting: Record + ) => Promise; + await expect(updatePreferredRemote({ customChunkSize: 64 })).resolves.toBe(false); + expect(failedSetPreferred).not.toHaveBeenCalled(); + expect(replacementSetPreferred).not.toHaveBeenCalled(); + }); + + it("writes a mismatch decision only through the still-active failed publication", async () => { + const setPreferredRemoteTweakSettings = vi.fn(async (_setting: unknown) => undefined); + const failedContext = { provider: {}, replicator: { setPreferredRemoteTweakSettings } }; + let updatePreferredRemote: ((setting: Record) => Promise) | undefined; + const askResolvingMismatched = vi.fn( + async (_preferred: unknown, update: (setting: Record) => Promise) => { + updatePreferredRemote = update; + } + ); + const recovery = createCentralCompatibilityRecovery({ + confirm: {}, + localDatabase: {}, + rebuilder: {}, + services: { + appLifecycle: {}, + API: {}, + replicator: { + runWithActiveReplicatorContext: vi.fn(async (task: (context: unknown) => unknown) => + task(failedContext) + ), + }, + tweakValue: { askResolvingMismatched }, + }, + } as never); + + await recovery.handleReplicationFailure({ + context: failedContext, + setting: {}, + outcome: replicationFailed(new Error("mismatched"), { + reason: CENTRAL_COMPATIBILITY_REJECTION_REASONS.TWEAK_MISMATCH, + preferredTweakValue: { customChunkSize: 60 }, + }), + showMessage: true, + interaction: USER_INITIATED_REPLICATION_AUTHORITY, + } as never); + + const effectiveSetting = { customChunkSize: 64 }; + await expect(updatePreferredRemote?.(effectiveSetting)).resolves.toBe(true); + expect(setPreferredRemoteTweakSettings).toHaveBeenCalledWith(effectiveSetting); + expect(setPreferredRemoteTweakSettings.mock.calls[0][0]).not.toBe(effectiveSetting); + }); + + it("does not apply an unlock selected for a replaced failed publication", async () => { + const failedMarkResolved = vi.fn(async () => undefined); + const replacementMarkResolved = vi.fn(async () => undefined); + const failedContext = { provider: {}, replicator: { markRemoteResolved: failedMarkResolved } }; + const replacementContext = { provider: {}, replicator: { markRemoteResolved: replacementMarkResolved } }; + const runWithActiveReplicatorContext = vi.fn(async (task: (context: unknown) => unknown) => + task(replacementContext) + ); + const recovery = createCentralCompatibilityRecovery({ + confirm: { + askSelectStringDialogue: vi.fn(async (_message: string, choices: string[]) => choices[1]), + }, + localDatabase: {}, + rebuilder: {}, + services: { + appLifecycle: { scheduleRestart: vi.fn() }, + API: {}, + replicator: { runWithActiveReplicatorContext }, + tweakValue: {}, + }, + } as never); + + await recovery.handleReplicationFailure({ + context: failedContext, + setting: {}, + outcome: replicationFailed(new Error("locked"), { + reason: CENTRAL_COMPATIBILITY_REJECTION_REASONS.NODE_LOCKED, + }), + showMessage: true, + interaction: USER_INITIATED_REPLICATION_AUTHORITY, + } as never); + + expect(runWithActiveReplicatorContext).toHaveBeenCalledOnce(); + expect(failedMarkResolved).not.toHaveBeenCalled(); + expect(replacementMarkResolved).not.toHaveBeenCalled(); + }); + + it("keeps cleaned-remote replication and balancing inside the shared activity boundary", async () => { + const activityFinished = vi.fn(); + const runBoundedRemoteActivity = vi.fn(async (task: () => unknown) => { + try { + return await task(); + } finally { + activityFinished(); + } + }); + const runFiniteReplicationActivity = vi.fn(async (task: () => unknown) => await task()); + const openOneShotReplication = vi.fn(async () => true); + const remoteDatabase = { close: vi.fn(async () => undefined) }; + const close = vi.fn(async () => undefined); + const activeReplicator = Object.assign(new LiveSyncCouchDBReplicator({} as never), { + connectRemoteCouchDBWithSetting: vi.fn(async () => ({ db: remoteDatabase, close })), + openOneShotReplication, + markRemoteResolved: vi.fn(async () => undefined), + }); + const expectedContext = { provider: {}, replicator: activeReplicator }; + const runWithActiveReplicatorContext = vi.fn(async (task: (context: unknown) => unknown) => + task(expectedContext) + ); + const localDatabase = { localDatabase: {}, clearCaches: vi.fn() }; + const recovery = createCentralCompatibilityRecovery({ + confirm: { confirmWithMessage: vi.fn(async () => "Cleanup") }, + localDatabase, + rebuilder: {}, + services: { + appLifecycle: {}, + API: { isMobile: vi.fn(() => false) }, + replicator: { + runBoundedRemoteActivity, + runFiniteReplicationActivity, + runWithActiveReplicatorContext, + }, + tweakValue: {}, + }, + } as never); + + await recovery.reconcileCleanedRemote(true, {} as ObsidianLiveSyncSettings, expectedContext as never); + + expect(runBoundedRemoteActivity).toHaveBeenCalledWith(expect.any(Function), { + label: "database-cleanup", + }); + expect(runFiniteReplicationActivity).toHaveBeenCalledWith(expect.any(Function), { + label: "replication", + }); + expect(runWithActiveReplicatorContext).toHaveBeenCalledOnce(); + expect(openOneShotReplication).toHaveBeenCalledOnce(); + expect(openOneShotReplication.mock.invocationCallOrder[0]).toBeLessThan( + activityFinished.mock.invocationCallOrder[0] + ); + expect(chunkMocks.balanceChunkPurgedDBs).toHaveBeenCalledOnce(); + expect(close).toHaveBeenCalledOnce(); + expect(close.mock.invocationCallOrder[0]).toBeLessThan(activityFinished.mock.invocationCallOrder[0]); + }); +}); diff --git a/src/serviceFeatures/replication/index.ts b/src/serviceFeatures/replication/index.ts new file mode 100644 index 00000000..78c28441 --- /dev/null +++ b/src/serviceFeatures/replication/index.ts @@ -0,0 +1,110 @@ +import { NO_INTERACTION } from "@vrtmrz/livesync-commonlib/replication"; +import { clearHandlers } from "@vrtmrz/livesync-commonlib/compat/replication/SyncParamsHandler"; +import { UnresolvedErrorManager } from "@vrtmrz/livesync-commonlib/compat/services/base/UnresolvedErrorManager"; +import type { ServiceContext } from "@vrtmrz/livesync-commonlib/context"; +import { fireAndForget } from "octagonal-wheels/promises"; +import type { IMinimumLiveSyncCommands, LiveSyncBaseCore } from "@/LiveSyncBaseCore"; +import { createAutomaticReplicationTriggers } from "./automaticTriggers"; +import { createCentralCompatibilityRecovery } from "./centralCompatibilityRecovery"; +import { createOnlineReplicationPreflight, createSecuritySeedPreflight } from "./preflight"; +import { ReplicateResultProcessor } from "./ReplicateResultProcessor"; + +type LocalApplicationActivityOwner = { + runBoundedLocalApplicationActivity(task: () => T | PromiseLike, options?: { label?: string }): Promise; +}; + +/** + * Compose result application, automatic triggers, preflight, and central + * compatibility recovery around the existing typed Services. + * + * Registration order is observable for equal-priority handlers. The host must + * call this after host serviceFeatures and add-ons are composed, but before + * legacy modules are bound. This preserves the former ModuleReplicator + * lifecycle-handler order without retaining a public module identity. + */ +export function useReplicationFeature( + core: LiveSyncBaseCore +): void { + const { services } = core; + // Obsidian adds an application-activity owner to its ReplicatorService. + // Generic hosts retain the former direct-execution fallback. + const localApplicationActivityOwner = services.replicator as typeof services.replicator & + Partial; + const resultProcessor = new ReplicateResultProcessor({ + currentSettings: () => services.setting.currentSettings(), + keyValueDB: services.keyValueDB.kvDB, + localDatabase: core.localDatabase, + requestActiveReplicatorRetirement: () => { + // Do not await a retirement transition from result application: it + // may be draining the replication work which delivered this item. + fireAndForget(() => services.replicator.onCloseActiveReplication()); + }, + runLocalApplicationActivity: async (task, options) => + localApplicationActivityOwner.runBoundedLocalApplicationActivity + ? await localApplicationActivityOwner.runBoundedLocalApplicationActivity(task, options) + : await task(), + services: { + appLifecycle: services.appLifecycle, + path: services.path, + replication: services.replication, + vault: services.vault, + }, + }); + const unresolvedErrorManager = new UnresolvedErrorManager(services.appLifecycle, services.context.events); + const initialiseAutomaticReplicationTriggers = createAutomaticReplicationTriggers({ + currentSettings: () => services.setting.currentSettings(), + isSuspended: () => services.appLifecycle.isSuspended(), + replicateDatabaseEvent: () => + services.replication.replicateUnattendedByEvent({ + trigger: "database-event", + interaction: NO_INTERACTION, + }), + reprocessStoredDocuments: () => resultProcessor.reprocessStoredDocuments(), + resumeResultApplication: () => resultProcessor.resume(), + suspendResultApplication: () => resultProcessor.suspend(), + }); + const preflightContext = { + services: { + API: services.API, + replicator: services.replicator, + setting: services.setting, + }, + }; + const onlinePreflight = createOnlineReplicationPreflight(unresolvedErrorManager, preflightContext); + const securitySeedPreflight = createSecuritySeedPreflight(unresolvedErrorManager, preflightContext); + const centralCompatibilityRecovery = createCentralCompatibilityRecovery({ + confirm: core.confirm, + localDatabase: core.localDatabase, + rebuilder: core.rebuilder, + services: { + API: services.API, + appLifecycle: services.appLifecycle, + replicator: services.replicator, + tweakValue: services.tweakValue, + }, + }); + + services.replicator.onBeforeReplicatorPublication.addHandler(() => { + // Key-derivation handlers belong to the candidate which is about to + // become active; discard callbacks retained by the previous owner. + clearHandlers(); + return Promise.resolve(true); + }); + services.databaseEvents.onDatabaseInitialised.addHandler(() => { + fireAndForget(() => resultProcessor.restoreFromSnapshotOnce()); + return Promise.resolve(true); + }); + services.appLifecycle.onSettingLoaded.addHandler(initialiseAutomaticReplicationTriggers); + services.replication.parseSynchroniseResult.addHandler((documents) => { + resultProcessor.enqueueAll(documents); + return Promise.resolve(true); + }); + services.replication.onBeforeReplicate.addHandler(onlinePreflight, 10); + services.replication.onPrepareCentralRemoteReplication.addHandler(securitySeedPreflight); + services.replication.onBeforeReplicate.addHandler(async () => { + await resultProcessor.restoreFromSnapshotOnce(); + unresolvedErrorManager.clearErrors(); + return true; + }, 100); + services.replication.onReplicationFailed.addHandler(centralCompatibilityRecovery.handleReplicationFailure); +} diff --git a/src/serviceFeatures/replication/preflight.ts b/src/serviceFeatures/replication/preflight.ts new file mode 100644 index 00000000..6ab3a152 --- /dev/null +++ b/src/serviceFeatures/replication/preflight.ts @@ -0,0 +1,65 @@ +import { $msg } from "@/common/translation"; +import { withOwnedRemoteResource } from "@/common/ownedRemoteResource"; +import { LOG_LEVEL_INFO, LOG_LEVEL_NOTICE, LOG_LEVEL_VERBOSE, Logger } from "octagonal-wheels/common/logger"; +import { MARK_LOG_NETWORK_ERROR } from "@vrtmrz/livesync-commonlib/compat/services/lib/logUtils"; +import { UnresolvedErrorManager } from "@vrtmrz/livesync-commonlib/compat/services/base/UnresolvedErrorManager"; +import { REMOTE_RESOURCE_KINDS } from "@vrtmrz/livesync-commonlib/replication"; +import type { LiveSyncBaseCore } from "@/LiveSyncBaseCore"; + +type ReplicationPreflightServices = Pick; + +interface ReplicationPreflightContext { + readonly services: ReplicationPreflightServices; +} + +/** Return the generic online preflight without inspecting a provider kind. */ +export function createOnlineReplicationPreflight( + errorManager: UnresolvedErrorManager, + context: ReplicationPreflightContext +) { + return function isOnlineAndCanReplicate(showMessage: boolean): Promise { + const errorMessage = "Network is offline"; + if (!context.services.API.isOnline) { + errorManager.showError(errorMessage, showMessage ? LOG_LEVEL_NOTICE : LOG_LEVEL_INFO); + return Promise.resolve(false); + } + errorManager.clearError(errorMessage); + return Promise.resolve(true); + }; +} + +/** + * Return the central-remote Security Seed preflight. The acquired resource is + * owned only for this read and is disposed before the handler settles. + */ +export function createSecuritySeedPreflight( + errorManager: UnresolvedErrorManager, + context: ReplicationPreflightContext +) { + return async function canReplicateWithSecuritySeed(showMessage: boolean): Promise { + const currentSettings = context.services.setting.currentSettings(); + const errorMessage = $msg("Replicator.Message.InitialiseFatalError"); + // This is a fatal preparation error, so the non-interactive path still + // records it while choosing a quieter log level. + const ensureMessage = `${MARK_LOG_NETWORK_ERROR}Failed to initialise the encryption key, preventing replication.`; + try { + const resource = await context.services.replicator.createRemoteResource( + REMOTE_RESOURCE_KINDS.SECURITY_SEED, + currentSettings + ); + if (!resource) { + errorManager.showError(errorMessage, showMessage ? LOG_LEVEL_NOTICE : LOG_LEVEL_INFO); + return false; + } + errorManager.clearError(errorMessage); + const seed = await withOwnedRemoteResource(resource, (ownedResource) => ownedResource.read()); + if (seed.length == 0) throw new Error("PBKDF2 salt (Security Seed) is empty"); + } catch (error) { + Logger(error, LOG_LEVEL_VERBOSE); + errorManager.showError(ensureMessage, showMessage ? LOG_LEVEL_NOTICE : LOG_LEVEL_INFO); + return false; + } + errorManager.clearError(ensureMessage); + return true; + }; +} diff --git a/src/serviceFeatures/replication/replicationFeature.unit.spec.ts b/src/serviceFeatures/replication/replicationFeature.unit.spec.ts new file mode 100644 index 00000000..ed4fe317 --- /dev/null +++ b/src/serviceFeatures/replication/replicationFeature.unit.spec.ts @@ -0,0 +1,126 @@ +import { describe, expect, it, vi } from "vitest"; +import { createServiceContext } from "@vrtmrz/livesync-commonlib/context"; +import { VER, type EntryDoc } from "@vrtmrz/livesync-commonlib/compat/common/types"; +import { promiseWithResolvers } from "octagonal-wheels/promises"; +import { useReplicationFeature } from "./index"; + +type BooleanHandler = (showMessage: boolean) => Promise; +type ParseHandler = (documents: PouchDB.Core.ExistingDocument[]) => Promise; + +function setup(onCloseActiveReplication = vi.fn(async () => true)) { + const read = vi.fn(async () => new Uint8Array([1])); + const dispose = vi.fn(async () => undefined); + const createRemoteResource = vi.fn(async () => ({ read, dispose })); + const beforeReplicateHandlers = new Map(); + const centralRemoteHandlers: BooleanHandler[] = []; + let parseHandler: ParseHandler | undefined; + const services = { + API: { isMobile: vi.fn(() => false), isOnline: true }, + appLifecycle: { + getUnresolvedMessages: { addHandler: vi.fn() }, + isReady: true, + isSuspended: vi.fn(() => false), + onSettingLoaded: { addHandler: vi.fn() }, + }, + context: createServiceContext(), + databaseEvents: { onDatabaseInitialised: { addHandler: vi.fn() } }, + keyValueDB: { + kvDB: { + get: vi.fn(async () => undefined), + set: vi.fn(async () => undefined), + }, + }, + path: { getPath: vi.fn((entry: { path: string }) => entry.path) }, + replication: { + onBeforeReplicate: { + addHandler: vi.fn((handler: BooleanHandler, priority = 0) => { + beforeReplicateHandlers.set(priority, handler); + }), + }, + onPrepareCentralRemoteReplication: { + addHandler: vi.fn((handler: BooleanHandler) => centralRemoteHandlers.push(handler)), + }, + onReplicationFailed: { addHandler: vi.fn() }, + parseSynchroniseResult: { + addHandler: vi.fn((handler: ParseHandler) => { + parseHandler = handler; + }), + }, + replicateUnattendedByEvent: vi.fn(async () => ({ status: "completed" as const })), + }, + replicator: { + createRemoteResource, + onBeforeReplicatorPublication: { addHandler: vi.fn() }, + onCloseActiveReplication, + }, + setting: { currentSettings: vi.fn(() => ({})) }, + tweakValue: {}, + vault: {}, + }; + const core = { + confirm: {}, + localDatabase: {}, + rebuilder: {}, + services, + }; + + useReplicationFeature(core as never); + + return { + beforeReplicateHandlers, + centralRemoteHandlers, + createRemoteResource, + dispose, + get parseHandler() { + return parseHandler; + }, + onCloseActiveReplication, + read, + }; +} + +describe("replication serviceFeature composition", () => { + it("refreshes and disposes the remote Security Seed before central replication", async () => { + const { centralRemoteHandlers, createRemoteResource, dispose, read } = setup(); + + await expect(centralRemoteHandlers[0](false)).resolves.toBe(true); + + expect(createRemoteResource).toHaveBeenCalledWith("security-seed", {}); + expect(read).toHaveBeenCalledOnce(); + expect(dispose).toHaveBeenCalledOnce(); + }); + + it("keeps generic preflight separate from central-remote preparation", async () => { + const { beforeReplicateHandlers, centralRemoteHandlers, createRemoteResource } = setup(); + const online = beforeReplicateHandlers.get(10); + const general = beforeReplicateHandlers.get(100); + + expect(online).toBeDefined(); + expect(general).toBeDefined(); + expect(centralRemoteHandlers).toHaveLength(1); + await expect(online!(false)).resolves.toBe(true); + await expect(general!(false)).resolves.toBe(true); + expect(createRemoteResource).not.toHaveBeenCalled(); + + await expect(centralRemoteHandlers[0](false)).resolves.toBe(true); + expect(createRemoteResource).toHaveBeenCalledOnce(); + }); + + it("requests owner retirement without awaiting the transition from result application", async () => { + const retirement = promiseWithResolvers(); + const onCloseActiveReplication = vi.fn(() => retirement.promise); + const harness = setup(onCloseActiveReplication); + const versionInfo = { + _id: "versioninfo", + _rev: "1-test", + type: "versioninfo", + version: VER + 1, + } as unknown as PouchDB.Core.ExistingDocument; + + expect(harness.parseHandler).toBeDefined(); + await expect(harness.parseHandler!([versionInfo])).resolves.toBe(true); + expect(onCloseActiveReplication).toHaveBeenCalledOnce(); + + retirement.resolve(true); + }); +});