fix: keep replicated document application active

Why:
- Large downloaded batches can outlive the finite remote operation and be suspended before every document reaches the Vault.
- The v1.0 activity contract requires local application to remain distinct from remote-operation reporting.

Changes:
- Add an Obsidian-owned local application activity boundary which reuses Wake Lock policy without incrementing remote activity.
- Retain one boundary through queue processing and the final recovery snapshot, releasing and reacquiring it around suspension.
- Log final snapshot failures, preserve processing results, and cover the lifecycle with focused tests and documentation.

src/modules/core/ReplicateResultProcessor.ts

@@ -24,9 +24,16 @@ import type { ReactiveSource } from "octagonal-wheels/dataobject/reactive_v2";
 import type { LiveSyncBaseCore } from "@/LiveSyncBaseCore";
 import { isNotFoundError } from "@vrtmrz/livesync-commonlib/compat/common/utils.doc";
 import type PouchDB from "pouchdb-core";
+import { promiseWithResolvers, type PromiseWithResolvers } from "octagonal-wheels/promises";
 
 const KV_KEY_REPLICATION_RESULT_PROCESSOR_SNAPSHOT = "replicationResultProcessorSnapshot";
 const REPROCESS_BATCH_SIZE = 100;
+type LocalApplicationActivityOwner = {
+    runBoundedLocalApplicationActivity<T>(
+        task: () => T | PromiseLike<T>,
+        options?: { label?: string }
+    ): Promise<T>;
+};
 type ReplicateResultProcessorState = {
     queued: PouchDB.Core.ExistingDocument<EntryDoc>[];
     processing: PouchDB.Core.ExistingDocument<EntryDoc>[];
@@ -67,9 +74,11 @@ export class ReplicateResultProcessor {
 
     public suspend() {
         this._suspended = true;
+        this.updateProcessingActivity();
     }
     public resume() {
         this._suspended = false;
+        this.updateProcessingActivity();
         fireAndForget(() => this.runProcessQueue());
     }
 
@@ -251,6 +260,40 @@ export class ReplicateResultProcessor {
      */
     private _processingChanges: PouchDB.Core.ExistingDocument<EntryDoc>[] = [];
 
+    private _processingActivity?: Promise<void>;
+    private _processingActivityDone?: PromiseWithResolvers<void>;
+
+    private updateProcessingActivity() {
+        if (this.isSuspended) {
+            this._processingActivityDone?.resolve();
+            return;
+        }
+        const hasPendingDocuments = this._queuedChanges.length > 0 || this._processingChanges.length > 0;
+        if (!hasPendingDocuments) {
+            this._processingActivityDone?.resolve();
+            return;
+        }
+        if (this._processingActivity) return;
+
+        const activityDone = promiseWithResolvers<void>();
+        this._processingActivityDone = activityDone;
+        const activityOwner = this.services.replicator as typeof this.services.replicator &
+            Partial<LocalApplicationActivityOwner>;
+        this._processingActivity = (
+            activityOwner.runBoundedLocalApplicationActivity
+                ? activityOwner.runBoundedLocalApplicationActivity(() => activityDone.promise, {
+                      label: "replicated-document-application",
+                  })
+                : activityDone.promise
+        )
+            .catch((error) => this.logError(error))
+            .finally(() => {
+                if (this._processingActivityDone === activityDone) this._processingActivityDone = undefined;
+                this._processingActivity = undefined;
+                this.updateProcessingActivity();
+            });
+    }
+
     /**
      * Enqueue the given document change for processing.
      * @param doc Document change to enqueue
@@ -278,6 +321,7 @@ export class ReplicateResultProcessor {
         }
         // Enqueue the change
         this._queuedChanges.push(doc);
+        this.updateProcessingActivity();
         this.triggerTakeSnapshot();
         this.triggerProcessQueue();
     }
@@ -385,7 +429,19 @@ export class ReplicateResultProcessor {
         } finally {
             // Remove from processing queue
             this._processingChanges = this._processingChanges.filter((e) => e !== change);
-            this.triggerTakeSnapshot();
+            try {
+                if (this._queuedChanges.length === 0 && this._processingChanges.length === 0) {
+                    try {
+                        await this._takeSnapshot();
+                    } catch (error) {
+                        this.logError(error);
+                    }
+                } else {
+                    this.triggerTakeSnapshot();
+                }
+            } finally {
+                this.updateProcessingActivity();
+            }
         }
     }
 

src/modules/core/ReplicateResultProcessor.unit.spec.ts

@@ -1,8 +1,67 @@
+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 { ReplicateResultProcessor } from "./ReplicateResultProcessor";
 
-describe("ReplicateResultProcessor target-filter reprocessing", () => {
+function note(id: string): PouchDB.Core.ExistingDocument<EntryDoc> {
+    return {
+        _id: id,
+        _rev: "1-test",
+        path: `${id}.md`,
+        ctime: 1,
+        mtime: 2,
+        size: 1,
+        children: [],
+        datatype: "plain",
+        type: "plain",
+        eden: {},
+    } as unknown as PouchDB.Core.ExistingDocument<EntryDoc>;
+}
+
+type SetupOptions = {
+    processSynchroniseResult?: (entry: unknown) => Promise<void>;
+    setSnapshot?: (key: string, value: unknown) => Promise<unknown>;
+};
+
+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<void>) => await task());
+    const core = {
+        services: {
+            appLifecycle: { isReady: true, isSuspended: () => false },
+            path: { getPath: (entry: { path: string }) => entry.path },
+            replication: {
+                databaseQueueCount: reactiveSource(0),
+                storageApplyingCount: reactiveSource(0),
+                replicationResultCount: reactiveSource(0),
+                processVirtualDocument: vi.fn(async () => false),
+                processOptionalSynchroniseResult: vi.fn(async () => false),
+                processSynchroniseResult,
+            },
+            replicator: { runBoundedLocalApplicationActivity },
+            vault: {
+                isTargetFile: vi.fn(async () => true),
+                isFileSizeTooLarge: vi.fn(() => false),
+                isValidPath: vi.fn(() => true),
+            },
+        },
+        kvDB: { set: setSnapshot },
+        localDatabase: {
+            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 },
+    } as never);
+    return { processor, processSynchroniseResult, runBoundedLocalApplicationActivity };
+}
+
+describe("ReplicateResultProcessor", () => {
     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" },
@@ -22,4 +81,67 @@ describe("ReplicateResultProcessor target-filter reprocessing", () => {
         expect(enqueueAll).toHaveBeenCalledOnce();
         expect(enqueueAll).toHaveBeenCalledWith(documents);
     });
+
+    it("keeps one local application activity until every replicated document has been applied", async () => {
+        const applying = promiseWithResolvers<void>();
+        let activityFinished = false;
+        const { processor, processSynchroniseResult, runBoundedLocalApplicationActivity } = setup({
+            processSynchroniseResult: async () => applying.promise,
+        });
+        runBoundedLocalApplicationActivity.mockImplementation(async (task: () => Promise<void>) => {
+            await task();
+            activityFinished = true;
+        });
+
+        processor.enqueueAll([note("one"), note("two")]);
+
+        await vi.waitFor(() => expect(processSynchroniseResult).toHaveBeenCalledTimes(2));
+        expect(runBoundedLocalApplicationActivity).toHaveBeenCalledTimes(1);
+        expect(runBoundedLocalApplicationActivity).toHaveBeenCalledWith(expect.any(Function), {
+            label: "replicated-document-application",
+        });
+        expect(activityFinished).toBe(false);
+
+        applying.resolve();
+
+        await vi.waitFor(() => expect(activityFinished).toBe(true));
+    });
+
+    it("settles local application activity when the final recovery snapshot fails", async () => {
+        let activityFinished = false;
+        const { processor, runBoundedLocalApplicationActivity } = setup({
+            setSnapshot: async () => Promise.reject(new Error("snapshot failed")),
+        });
+        runBoundedLocalApplicationActivity.mockImplementation(async (task: () => Promise<void>) => {
+            await task();
+            activityFinished = true;
+        });
+
+        processor.enqueueAll([note("one")]);
+
+        await vi.waitFor(() => expect(activityFinished).toBe(true));
+    });
+
+    it("releases and reacquires local application activity around processing suspension", async () => {
+        const applying = promiseWithResolvers<void>();
+        let completedActivities = 0;
+        const { processor, processSynchroniseResult, runBoundedLocalApplicationActivity } = setup({
+            processSynchroniseResult: async () => applying.promise,
+        });
+        runBoundedLocalApplicationActivity.mockImplementation(async (task: () => Promise<void>) => {
+            await task();
+            completedActivities++;
+        });
+        processor.enqueueAll([note("one")]);
+        await vi.waitFor(() => expect(processSynchroniseResult).toHaveBeenCalledOnce());
+
+        processor.suspend();
+        await vi.waitFor(() => expect(completedActivities).toBe(1));
+
+        processor.resume();
+        await vi.waitFor(() => expect(runBoundedLocalApplicationActivity).toHaveBeenCalledTimes(2));
+
+        applying.resolve();
+        await vi.waitFor(() => expect(completedActivities).toBe(2));
+    });
 });