Make remote and log resource ownership explicit

src/common/ownedRemoteResource.ts

@@ -1,3 +1,5 @@
+import { withOwnedResource, type DisposableResource } from "octagonal-wheels/resource";
+
 /**
  * Run a finite operation with a flow-owned remote resource and release it
  * after either success or failure.
@@ -6,13 +8,9 @@
  * caller's ownership boundary explicit and prevents finite flows from leaking
  * a provider-owned resource when their operation rejects.
  */
-export async function withOwnedRemoteResource<TResource extends { dispose(): Promise<void> }, TResult>(
+export function withOwnedRemoteResource<TResource extends DisposableResource & { dispose(): Promise<void> }, TResult>(
     resource: TResource,
     operation: (ownedResource: TResource) => Promise<TResult>
 ): Promise<TResult> {
-    try {
-        return await operation(resource);
-    } finally {
-        await resource.dispose();
-    }
+    return withOwnedResource(resource, operation);
 }

src/common/ownedRemoteResource.unit.spec.ts

@@ -26,3 +26,47 @@ describe("flow-owned remote resources", () => {
         expect(dispose).toHaveBeenCalledOnce();
     });
 });
+
+describe("owned remote resource completion", () => {
+    it("waits for cleanup and preserves its receiver", async () => {
+        let finish!: () => void;
+        let started!: () => void;
+        const disposing = new Promise<void>((resolve) => {
+            started = resolve;
+        });
+        const cleanup = new Promise<void>((resolve) => {
+            finish = resolve;
+        });
+        const resource = {
+            dispose: vi.fn(function (this: unknown) {
+                expect(this).toBe(resource);
+                started();
+                return cleanup;
+            }),
+        };
+        let settled = false;
+        const result = withOwnedRemoteResource(resource, async () => "done");
+        void result.then(() => {
+            settled = true;
+        });
+        await disposing;
+        expect(settled).toBe(false);
+        finish();
+        await expect(result).resolves.toBe("done");
+        expect(resource.dispose).toHaveBeenCalledOnce();
+    });
+
+    it("keeps finally precedence when both the operation and cleanup fail", async () => {
+        const operationError = new Error("operation");
+        const disposalError = new Error("cleanup");
+        const dispose = vi.fn(async () => {
+            throw disposalError;
+        });
+        await expect(
+            withOwnedRemoteResource({ dispose }, async () => {
+                throw operationError;
+            })
+        ).rejects.toBe(disposalError);
+        expect(dispose).toHaveBeenCalledOnce();
+    });
+});

src/common/replicatorProviders.ts

@@ -73,7 +73,7 @@ async function runOneShotWithOutcome(
 
 // Manual and unattended wrappers share the provider transfer operation, but
 // keep interaction authority and result presentation explicit at this boundary.
-const couchDBUserInitiatedOneShot: UserInitiatedOneShotRunner = async (instance, setting, request) => {
+const userInitiatedOneShot: UserInitiatedOneShotRunner = async (instance, setting, request) => {
     return await runOneShotWithOutcome(
         instance,
         setting,
@@ -81,20 +81,7 @@ const couchDBUserInitiatedOneShot: UserInitiatedOneShotRunner = async (instance,
     );
 };
 
-const couchDBUnattendedOneShot: UnattendedOneShotRunner = async (instance, setting, request) => {
-    if (request.interaction.kind !== NO_INTERACTION.kind) return replicationBlocked("interaction-required");
-    return await runOneShotWithOutcome(instance, setting, false);
-};
-
-const objectStorageUserInitiatedOneShot: UserInitiatedOneShotRunner = async (instance, setting, request) => {
-    return await runOneShotWithOutcome(
-        instance,
-        setting,
-        request.progressPresentation === REPLICATION_PROGRESS_PRESENTATIONS.NOTICE
-    );
-};
-
-const objectStorageUnattendedOneShot: UnattendedOneShotRunner = async (instance, setting, request) => {
+const unattendedOneShot: UnattendedOneShotRunner = async (instance, setting, request) => {
     if (request.interaction.kind !== NO_INTERACTION.kind) return replicationBlocked("interaction-required");
     return await runOneShotWithOutcome(instance, setting, false);
 };
@@ -136,8 +123,8 @@ export function createCentralReplicatorProviderDefinitions(
                 ),
             },
             centralRemoteAdministration: COUCHDB_CENTRAL_REMOTE_ADMINISTRATION_CAPABILITY,
-            userInitiatedOneShot: supportedCapability(couchDBUserInitiatedOneShot),
-            unattendedOneShot: supportedCapability(couchDBUnattendedOneShot),
+            userInitiatedOneShot: supportedCapability(userInitiatedOneShot),
+            unattendedOneShot: supportedCapability(unattendedOneShot),
             continuous: supportedOpenReplicationContinuous(),
             stopActiveTransfer: supportedStopActiveTransfer(),
         },
@@ -166,8 +153,8 @@ export function createCentralReplicatorProviderDefinitions(
                 [REMOTE_RESOURCE_KINDS.INITIAL_FETCH_FAST_FORWARD]: CAPABILITY_NOT_APPLICABLE,
             },
             centralRemoteAdministration: OBJECT_STORAGE_CENTRAL_REMOTE_ADMINISTRATION_CAPABILITY,
-            userInitiatedOneShot: supportedCapability(objectStorageUserInitiatedOneShot),
-            unattendedOneShot: supportedCapability(objectStorageUnattendedOneShot),
+            userInitiatedOneShot: supportedCapability(userInitiatedOneShot),
+            unattendedOneShot: supportedCapability(unattendedOneShot),
             continuous: CAPABILITY_NOT_APPLICABLE,
             stopActiveTransfer: supportedStopActiveTransfer(),
         },

src/common/replicatorProviders.unit.spec.ts

@@ -209,6 +209,37 @@ describe("central Replicator provider definitions", () => {
         expect(openOneShotReplicationWithOutcome).toHaveBeenNthCalledWith(2, setting, true);
     });
 
+    it.each([REMOTE_COUCHDB, REMOTE_MINIO] as const)(
+        "keeps unattended %s work quiet and rejects interaction before opening replication",
+        async (remoteType) => {
+            const definition = createCentralReplicatorProviderDefinitions({} as never).get(remoteType)!;
+            if (definition.unattendedOneShot.kind !== CAPABILITY_SUPPORT_KINDS.SUPPORTED) {
+                throw new Error("Unattended OneShot is unavailable");
+            }
+            const setting = Object.assign(createNewVaultSettings(), { remoteType });
+            const instance = {
+                initializeDatabaseForReplication: vi.fn(async () => true),
+                openReplication: vi.fn(async () => true),
+                terminateSync: vi.fn(),
+                closeReplication: vi.fn(),
+                openOneShotReplicationWithOutcome: vi.fn(async () => REPLICATION_COMPLETED),
+            };
+            const blocked = await definition.unattendedOneShot.run(instance, setting, {
+                trigger: "resume",
+                interaction: USER_INITIATED_REPLICATION_AUTHORITY,
+            } as never);
+            expect(blocked).toEqual({ status: "blocked", reason: "interaction-required" });
+            expect(instance.openOneShotReplicationWithOutcome).not.toHaveBeenCalled();
+            await expect(
+                definition.unattendedOneShot.run(instance, setting, {
+                    trigger: "resume",
+                    interaction: NO_INTERACTION,
+                })
+            ).resolves.toBe(REPLICATION_COMPLETED);
+            expect(instance.openOneShotReplicationWithOutcome).toHaveBeenCalledExactlyOnceWith(setting, false);
+        }
+    );
+
     it("dispatches central finite work through the declared operation rather than constructor identity", async () => {
         const definitions = createCentralReplicatorProviderDefinitions({} as never);
         const couchDB = definitions.get(REMOTE_COUCHDB)!;

src/common/replicatorResources/shared.ts

@@ -1,3 +1,4 @@
+import { createOwnedResource } from "octagonal-wheels/resource";
 import type { RemoteDBSettings } from "@vrtmrz/livesync-commonlib/compat/common/types";
 
 /**
@@ -13,13 +14,8 @@ export interface ResourceReplicator {
 
 /** Create one idempotent asynchronous disposer for a private Replicator. */
 export function createReplicatorDisposer(replicator: ResourceReplicator): () => Promise<void> {
-    let disposal: Promise<void> | undefined;
-    return () => {
-        if (disposal === undefined) {
-            disposal = Promise.resolve().then(() => replicator.closeReplication());
-        }
-        return disposal;
-    };
+    const resource = createOwnedResource(replicator, (owned) => Promise.resolve().then(() => owned.closeReplication()));
+    return () => Promise.resolve(resource.dispose());
 }
 
 /** Fence a finite resource from later edits to its source settings object. */

src/common/replicatorResources/shared.unit.spec.ts

@@ -0,0 +1,42 @@
+import { describe, expect, it, vi } from "vitest";
+import { createReplicatorDisposer } from "./shared";
+
+describe("Replicator disposal", () => {
+    it("defers closing, preserves the receiver, and retains one completion promise", async () => {
+        let finish!: () => void;
+        const closed = new Promise<void>((resolve) => {
+            finish = resolve;
+        });
+        const replicator = {
+            closeReplication: vi.fn(function (this: unknown) {
+                expect(this).toBe(replicator);
+                return closed;
+            }),
+        };
+        const dispose = createReplicatorDisposer(replicator);
+        const pending = dispose();
+        expect(replicator.closeReplication).not.toHaveBeenCalled();
+        expect(dispose()).toBe(pending);
+        await Promise.resolve();
+        expect(replicator.closeReplication).toHaveBeenCalledOnce();
+        finish();
+        await pending;
+        expect(dispose()).toBe(pending);
+        expect(replicator.closeReplication).toHaveBeenCalledOnce();
+    });
+
+    it("retains synchronous closing failures as one rejected promise", async () => {
+        const error = new Error("close");
+        const replicator = {
+            closeReplication: vi.fn(() => {
+                throw error;
+            }),
+        };
+        const dispose = createReplicatorDisposer(replicator);
+        const pending = dispose();
+        await expect(pending).rejects.toBe(error);
+        expect(dispose()).toBe(pending);
+        await expect(dispose()).rejects.toBe(error);
+        expect(replicator.closeReplication).toHaveBeenCalledOnce();
+    });
+});