From bca2f2dcc6c2f77a1af586a49bee5d21f3e97f5c Mon Sep 17 00:00:00 2001 From: Artem Ovchinnikov <53350796+Akilaydin@users.noreply.github.com> Date: Wed, 21 Oct 2026 21:25:16 +0300 Subject: [PATCH] fix(replication): publish pending counts and reflection failures --- src/common/events.ts | 3 ++ .../replication/ReplicateResultProcessor.ts | 7 ++++ .../ReplicateResultProcessor.unit.spec.ts | 33 +++++++++++++++++++ src/serviceFeatures/replication/index.ts | 3 +- 4 files changed, 45 insertions(+), 1 deletion(-) diff --git a/src/common/events.ts b/src/common/events.ts index d358f997..220c420e 100644 --- a/src/common/events.ts +++ b/src/common/events.ts @@ -4,6 +4,8 @@ import { createLiveSyncEventHub } from "@vrtmrz/livesync-commonlib/context"; export const EVENT_PLUGIN_LOADED = "plugin-loaded"; export const EVENT_PLUGIN_UNLOADED = "plugin-unloaded"; export const EVENT_FILE_SAVED = "file-saved"; +/** A received note could not be reflected from the local database into the Vault. */ +export const EVENT_VAULT_REFLECTION_FAILED = "vault-reflection-failed"; export const EVENT_LEAF_ACTIVE_CHANGED = "leaf-active-changed"; export const EVENT_REQUEST_OPEN_SETTINGS = "request-open-settings"; @@ -28,6 +30,7 @@ declare global { interface LSEvents { [EVENT_PLUGIN_LOADED]: undefined; [EVENT_PLUGIN_UNLOADED]: undefined; + [EVENT_VAULT_REFLECTION_FAILED]: undefined; [EVENT_REQUEST_OPEN_PLUGIN_SYNC_DIALOG]: undefined; [EVENT_REQUEST_OPEN_SETTINGS]: undefined; [EVENT_REQUEST_RELOAD_SETTING_TAB]: undefined; diff --git a/src/serviceFeatures/replication/ReplicateResultProcessor.ts b/src/serviceFeatures/replication/ReplicateResultProcessor.ts index 523043d9..075ebd46 100644 --- a/src/serviceFeatures/replication/ReplicateResultProcessor.ts +++ b/src/serviceFeatures/replication/ReplicateResultProcessor.ts @@ -58,6 +58,7 @@ type ReplicateResultProcessorServices = Pick< * database would be invalid. */ interface ReplicateResultProcessorContext { + readonly onVaultReflectionFailure?: () => void; readonly currentSettings: () => ReplicateResultProcessorSettings; readonly getKeyValueDB: () => LiveSyncBaseCore["kvDB"]; readonly getLocalDatabase: () => LiveSyncBaseCore["localDatabase"]; @@ -87,6 +88,7 @@ export class ReplicateResultProcessor { Logger(e, LOG_LEVEL_VERBOSE); } private reportVaultReflectionFailure(entry: MetaEntry, cause?: unknown) { + this.context.onVaultReflectionFailure?.(); this.log( `Live replication could not reflect ${this.getPath(entry)} from the local database to the Vault; this path remains eligible for a later Vault scan.`, LOG_LEVEL_VERBOSE @@ -383,6 +385,7 @@ export class ReplicateResultProcessor { } // Enqueue the change this._queuedChanges.push(doc); + this.reportStatus(); this.updateProcessingActivity(); this.triggerTakeSnapshot(); this.triggerProcessQueue(); @@ -491,6 +494,9 @@ export class ReplicateResultProcessor { } finally { // Remove from processing queue this._processingChanges = this._processingChanges.filter((e) => e !== change); + // Queue visibility must not wait for persisting a recovery snapshot. Consumers + // use this count to decide whether the reflected Vault is ready for interaction. + this.reportStatus(); try { if (this._queuedChanges.length === 0 && this._processingChanges.length === 0) { try { @@ -548,6 +554,7 @@ export class ReplicateResultProcessor { : await this.localDatabase.getDBEntryFromMeta({ ...dbDoc }, false, true); if (!doc) { // Failed to gather content + this.context.onVaultReflectionFailure?.(); this.log(`Failed to gather content of ${docNote}`, LOG_LEVEL_NOTICE); return; } diff --git a/src/serviceFeatures/replication/ReplicateResultProcessor.unit.spec.ts b/src/serviceFeatures/replication/ReplicateResultProcessor.unit.spec.ts index 5ed7bf31..c1bf5680 100644 --- a/src/serviceFeatures/replication/ReplicateResultProcessor.unit.spec.ts +++ b/src/serviceFeatures/replication/ReplicateResultProcessor.unit.spec.ts @@ -37,6 +37,7 @@ function note(id: string): PouchDB.Core.ExistingDocument { } type SetupOptions = { + onVaultReflectionFailure?: () => void; applicationReady?: boolean; databaseReady?: boolean; maxMTimeForReflectEvents?: number; @@ -95,6 +96,7 @@ function setup(options: SetupOptions = {}) { }, }; const processor = new ReplicateResultProcessor({ + onVaultReflectionFailure: options.onVaultReflectionFailure, currentSettings: () => ({ maxMTimeForReflectEvents: options.maxMTimeForReflectEvents ?? 0, suspendParseReplicationResult: false, @@ -117,12 +119,43 @@ function setup(options: SetupOptions = {}) { isValidPath, onCloseActiveReplication, processor, + replicationResultCount: core.services.replication.replicationResultCount, processSynchroniseResult, runBoundedLocalApplicationActivity, }; } describe("ReplicateResultProcessor", () => { + it("publishes pending received documents synchronously even if snapshot persistence stalls", () => { + const { processor, replicationResultCount, processSynchroniseResult } = setup({ + applicationReady: false, + setSnapshot: async () => new Promise(() => undefined), + }); + processor.enqueueAll([note("pending")]); + expect(replicationResultCount.value).toBe(1); + expect(processSynchroniseResult).not.toHaveBeenCalled(); + }); + + it("clears the received-document counter before awaiting an unresponsive snapshot write", async () => { + const { processor, replicationResultCount, processSynchroniseResult } = setup({ + setSnapshot: async () => new Promise(() => undefined), + }); + processor.enqueueAll([note("pending")]); + expect(replicationResultCount.value).toBe(1); + await vi.waitFor(() => expect(processSynchroniseResult).toHaveBeenCalledOnce()); + await vi.waitFor(() => expect(replicationResultCount.value).toBe(0)); + }); + + it("reports a failed Vault reflection instead of treating an empty queue as success", async () => { + const onVaultReflectionFailure = vi.fn(); + const { processor, processSynchroniseResult } = setup({ + onVaultReflectionFailure, + processSynchroniseResult: async () => false, + }); + processor.enqueueAll([note("unreflected")]); + await vi.waitFor(() => expect(processSynchroniseResult).toHaveBeenCalledTimes(1)); + await vi.waitFor(() => expect(onVaultReflectionFailure).toHaveBeenCalledTimes(1)); + }); it("does not add a permanent application block when snapshot recovery fails", async () => { const { processor } = setup({ getSnapshot: async () => { diff --git a/src/serviceFeatures/replication/index.ts b/src/serviceFeatures/replication/index.ts index 37eb5f52..473c7bb7 100644 --- a/src/serviceFeatures/replication/index.ts +++ b/src/serviceFeatures/replication/index.ts @@ -4,7 +4,7 @@ import { UnresolvedErrorManager } from "@vrtmrz/livesync-commonlib/compat/servic import type { ServiceContext } from "@vrtmrz/livesync-commonlib/context"; import { fireAndForget } from "octagonal-wheels/promises"; import type { IMinimumLiveSyncCommands, LiveSyncBaseCore } from "@/LiveSyncBaseCore"; -import { EVENT_APPLICATION_READY } from "@/common/events"; +import { EVENT_APPLICATION_READY, EVENT_VAULT_REFLECTION_FAILED } from "@/common/events"; import { createAutomaticReplicationTriggers } from "./automaticTriggers"; import { createCentralCompatibilityRecovery } from "./centralCompatibilityRecovery"; import { createOnlineReplicationPreflight, createSecuritySeedPreflight } from "./preflight"; @@ -38,6 +38,7 @@ export function useReplicationFeature services.context.events.emitEvent(EVENT_VAULT_REFLECTION_FAILED), currentSettings: () => services.setting.currentSettings(), getKeyValueDB: () => services.keyValueDB.kvDB, getLocalDatabase: () => core.localDatabase,