fix(replication): publish pending counts and reflection failures

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;

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;
             }

src/serviceFeatures/replication/ReplicateResultProcessor.unit.spec.ts

@@ -37,6 +37,7 @@ function note(id: string): PouchDB.Core.ExistingDocument<EntryDoc> {
 }
 
 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<never>(() => 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<never>(() => 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 () => {

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<TContext extends ServiceContext, TCommands
         ? services.replicator
         : undefined;
     const resultProcessor = new ReplicateResultProcessor({
+        onVaultReflectionFailure: () => services.context.events.emitEvent(EVENT_VAULT_REFLECTION_FAILED),
         currentSettings: () => services.setting.currentSettings(),
         getKeyValueDB: () => services.keyValueDB.kvDB,
         getLocalDatabase: () => core.localDatabase,