Merge main into shared runtime foundations

Keep SyncGuard startup dependencies isolated in unload logging tests.

src/serviceFeatures/replication/ReplicateResultProcessor.ts

@@ -59,6 +59,7 @@ type ReplicateResultProcessorServices = Pick<
  * database would be invalid.
  */
 interface ReplicateResultProcessorContext {
+    readonly onVaultReflectionFailure?: () => void;
     readonly currentSettings: () => ReplicateResultProcessorSettings;
     readonly getKeyValueDB: () => LiveSyncBaseCore["kvDB"];
     readonly getLocalDatabase: () => LiveSyncBaseCore["localDatabase"];
@@ -88,6 +89,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
@@ -379,6 +381,7 @@ export class ReplicateResultProcessor {
         }
         // Enqueue the change
         this._queuedChanges.push(doc);
+        this.reportStatus();
         this.updateProcessingActivity();
         this.triggerTakeSnapshot();
         this.triggerProcessQueue();
@@ -487,6 +490,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 {
@@ -511,6 +517,8 @@ export class ReplicateResultProcessor {
                 releaser = await this._semaphore.acquire();
                 await this._applyToDatabase(doc);
             } catch (e) {
+                // An exception during application must not be mistaken for a fully reflected Vault.
+                this.context.onVaultReflectionFailure?.();
                 this.log(`Error while processing replication result`, LOG_LEVEL_NOTICE);
                 this.logError(e);
             } finally {
@@ -544,6 +552,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;
             }
@@ -619,6 +628,7 @@ export class ReplicateResultProcessor {
                 // Note that this is not a common case.
                 return true;
             } else {
+                this.context.onVaultReflectionFailure?.();
                 this.log(
                     `Failed to get existing document for ${path} (${shortenId(dbDoc._id)}, ${shortenRev(dbDoc._rev)}) `,
                     LOG_LEVEL_NOTICE

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

@@ -37,6 +37,9 @@ function note(id: string): PouchDB.Core.ExistingDocument<EntryDoc> {
 }
 
 type SetupOptions = {
+    onVaultReflectionFailure?: () => void;
+    getRaw?: (id: string) => Promise<unknown>;
+    getDBEntryFromMeta?: (entry: object) => Promise<unknown>;
     applicationReady?: boolean;
     databaseReady?: boolean;
     maxMTimeForReflectEvents?: number;
@@ -57,7 +60,7 @@ function setup(options: SetupOptions = {}) {
     const onCloseActiveReplication = vi.fn(async () => true);
     const isReady = vi.fn(() => options.applicationReady ?? true);
     const isValidPath = vi.fn(options.isValidPath ?? (() => true));
-    const getDBEntryFromMeta = vi.fn(async (entry: object) => ({ ...entry, data: "x" }));
+    const getDBEntryFromMeta = vi.fn(options.getDBEntryFromMeta ?? (async (entry: object) => ({ ...entry, data: "x" })));
     const localPhysicalDatabase = {
         ...(options.databaseId ? { id: vi.fn(async () => options.databaseId) } : {}),
     } as PouchDB.Database<EntryDoc>;
@@ -89,12 +92,14 @@ function setup(options: SetupOptions = {}) {
                     if (options.localVersionInfo === undefined) throw { status: 404 };
                     return options.localVersionInfo;
                 }
+                if (options.getRaw) return await options.getRaw(id);
                 return { _id: id, _rev: "1-test" };
             }),
             getDBEntryFromMeta,
         },
     };
     const processor = new ReplicateResultProcessor({
+        onVaultReflectionFailure: options.onVaultReflectionFailure,
         currentSettings: () => ({
             maxMTimeForReflectEvents: options.maxMTimeForReflectEvents ?? 0,
             suspendParseReplicationResult: false,
@@ -117,12 +122,64 @@ function setup(options: SetupOptions = {}) {
         isValidPath,
         onCloseActiveReplication,
         processor,
+        replicationResultCount: core.services.replication.replicationResultCount,
         processSynchroniseResult,
         runBoundedLocalApplicationActivity,
     };
 }
 
 describe("ReplicateResultProcessor", () => {
+    it("reports an IndexedDB read error as failed reflection", async () => {
+        const onVaultReflectionFailure = vi.fn();
+        const { processor, processSynchroniseResult } = setup({
+            onVaultReflectionFailure,
+            getRaw: async () => { throw new Error("IndexedDB read failed"); },
+        });
+        processor.enqueueAll([note("unreadable")]);
+        await vi.waitFor(() => expect(onVaultReflectionFailure).toHaveBeenCalledOnce());
+        expect(processSynchroniseResult).not.toHaveBeenCalled();
+    });
+
+    it("reports a thrown document reconstruction as failed reflection", async () => {
+        const onVaultReflectionFailure = vi.fn();
+        const { processor } = setup({
+            onVaultReflectionFailure,
+            getDBEntryFromMeta: async () => { throw new Error("Cannot read chunks"); },
+        });
+        processor.enqueueAll([note("missing-chunks")]);
+        await vi.waitFor(() => expect(onVaultReflectionFailure).toHaveBeenCalledOnce());
+    });
+
+    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,