perf(mobile): send no longer waits one round trip per saved attachment

apps/mobile/src/lib/attachmentUpload.test.ts

@@ -1,3 +1,4 @@
+import { ASSET_URL_MAX_IN_FLIGHT } from "@t3tools/client-runtime/state/assets";
 import { EnvironmentId, PROVIDER_SEND_TURN_MAX_IMAGE_BYTES } from "@t3tools/contracts";
 import * as Option from "effect/Option";
 import { beforeEach, describe, expect, it, vi } from "vite-plus/test";
@@ -717,6 +718,203 @@ describe("prepareTurnAttachments", () => {
     expect(removeCallsFor("pending-existing-pdf")).toBe(0);
   });
 
+  describe("verifying saved uploads", () => {
+    const savedUpload = (n: number) => ({
+      ...file,
+      id: `file-${n}`,
+      uploadedAttachmentId: `pending-${n}-pdf`,
+      uploadEnvironmentId: environmentId,
+    });
+    const savedUploads = (count: number) => Array.from({ length: count }, (_, i) => savedUpload(i));
+
+    /** Hands every verification to the test as a deferred, keyed by the attachment id it checks. */
+    function holdVerifications() {
+      const held = new Map<string, PromiseWithResolvers<unknown>>();
+      const inFlight = { max: 0, current: 0 };
+      mocks.executeAtomQuery.mockImplementation(
+        async (_registry: unknown, target: unknown): Promise<unknown> => {
+          const { attachmentId } = (target as { input: { resource: { attachmentId: string } } })
+            .input.resource;
+          const deferred = Promise.withResolvers<unknown>();
+          held.set(attachmentId, deferred);
+          inFlight.current += 1;
+          inFlight.max = Math.max(inFlight.max, inFlight.current);
+          try {
+            return await deferred.promise;
+          } finally {
+            inFlight.current -= 1;
+          }
+        },
+      );
+      return {
+        started: () => [...held.keys()],
+        maxInFlight: () => inFlight.max,
+        resolve: (attachmentId: string, result: unknown = { _tag: "Success", value: {} }) =>
+          held.get(attachmentId)!.resolve(result),
+      };
+    }
+    // Lets queued promise callbacks run; the verifications are held, so nothing else can progress.
+    const settle = async () => {
+      for (let turn = 0; turn < 50; turn += 1) await Promise.resolve();
+    };
+    const outcomeOf = (preparing: Promise<unknown>) =>
+      preparing.then(
+        () => "resolved",
+        (error: unknown) => error,
+      );
+    const ids = (prepared: Awaited<ReturnType<typeof prepareTurnAttachments>>) =>
+      prepared.status === "ready" ? prepared.attachments.map((a) => ("id" in a ? a.id : null)) : [];
+
+    it("starts every verification together and keeps attachment order", async () => {
+      const verifications = holdVerifications();
+      const preparing = prepareTurnAttachments({ environmentId, attachments: savedUploads(3) });
+      await settle();
+
+      expect(verifications.started()).toEqual(["pending-0-pdf", "pending-1-pdf", "pending-2-pdf"]);
+      verifications.resolve("pending-2-pdf");
+      verifications.resolve("pending-0-pdf");
+      verifications.resolve("pending-1-pdf");
+      const prepared = await preparing;
+
+      expect(ids(prepared)).toEqual(["pending-0-pdf", "pending-1-pdf", "pending-2-pdf"]);
+      expect(prepared.status === "ready" && prepared.pendingAttachmentIds).toEqual([
+        "pending-0-pdf",
+        "pending-1-pdf",
+        "pending-2-pdf",
+      ]);
+      expect(mocks.upload).not.toHaveBeenCalled();
+    });
+
+    it("keeps at most the asset URL limit of verifications in flight", async () => {
+      const verifications = holdVerifications();
+      const preparing = prepareTurnAttachments({ environmentId, attachments: savedUploads(8) });
+      await settle();
+      expect(verifications.started()).toHaveLength(ASSET_URL_MAX_IN_FLIGHT);
+
+      verifications.resolve("pending-0-pdf");
+      await settle();
+      expect(verifications.started()).toHaveLength(ASSET_URL_MAX_IN_FLIGHT + 1);
+
+      for (let n = 1; n < 8; n += 1) {
+        verifications.resolve(`pending-${n}-pdf`);
+        await settle();
+      }
+      const prepared = await preparing;
+      expect(prepared.status).toBe("ready");
+      expect(verifications.started()).toHaveLength(8);
+      expect(verifications.maxInFlight()).toBe(ASSET_URL_MAX_IN_FLIGHT);
+    });
+
+    it("reports the first failing attachment in order", async () => {
+      mocks.executeAtomQuery.mockImplementation(async (_registry: unknown, target: unknown) => {
+        const { attachmentId } = (target as { input: { resource: { attachmentId: string } } }).input
+          .resource;
+        return attachmentId === "pending-0-pdf"
+          ? { _tag: "Success", value: {} }
+          : { _tag: "Failure", error: new Error(attachmentId) };
+      });
+
+      await expect(
+        prepareTurnAttachments({ environmentId, attachments: savedUploads(3) }),
+      ).rejects.toThrow("pending-1-pdf");
+      expect(mocks.upload).not.toHaveBeenCalled();
+    });
+
+    it("reports the first failing attachment even when a later one fails first", async () => {
+      const verifications = holdVerifications();
+      const outcome = outcomeOf(
+        prepareTurnAttachments({ environmentId, attachments: savedUploads(3) }),
+      );
+      await settle();
+
+      verifications.resolve("pending-2-pdf", { _tag: "Failure", error: new Error("third") });
+      verifications.resolve("pending-1-pdf", { _tag: "Failure", error: new Error("second") });
+      verifications.resolve("pending-0-pdf");
+
+      expect(await outcome).toEqual(new Error("second"));
+    });
+
+    it("does not leak a later verification that rejects after an earlier attachment failed", async () => {
+      mocks.executeAtomQuery.mockImplementation(async (_registry: unknown, target: unknown) => {
+        const { attachmentId } = (target as { input: { resource: { attachmentId: string } } }).input
+          .resource;
+        if (attachmentId === "pending-0-pdf") {
+          return { _tag: "Failure", error: new Error("first") };
+        }
+        throw new Error("transport defect");
+      });
+      await expect(
+        prepareTurnAttachments({ environmentId, attachments: savedUploads(3) }),
+      ).rejects.toThrow("first");
+    });
+
+    it("stops starting queued verifications once an earlier attachment fails", async () => {
+      const verifications = holdVerifications();
+      const outcome = outcomeOf(
+        prepareTurnAttachments({ environmentId, attachments: savedUploads(8) }),
+      );
+      await settle();
+      expect(verifications.started()).toHaveLength(ASSET_URL_MAX_IN_FLIGHT);
+
+      verifications.resolve("pending-0-pdf", { _tag: "Failure", error: new Error("first") });
+      expect(await outcome).toEqual(new Error("first"));
+      for (let n = 1; n < ASSET_URL_MAX_IN_FLIGHT; n += 1) {
+        verifications.resolve(`pending-${n}-pdf`);
+      }
+      await settle();
+
+      // The first freed slot went to the 7th before the loop saw the failure; the 8th never starts.
+      expect(verifications.started()).toHaveLength(ASSET_URL_MAX_IN_FLIGHT + 1);
+    });
+
+    it("verifies later attachments while an earlier expired one uploads, then keeps order", async () => {
+      const verifications = holdVerifications();
+      const uploading = Promise.withResolvers<{ status: number; body: string; headers: object }>();
+      mocks.upload.mockReturnValue(uploading.promise);
+      const preparing = prepareTurnAttachments({ environmentId, attachments: savedUploads(3) });
+      await settle();
+
+      verifications.resolve("pending-0-pdf", {
+        _tag: "Failure",
+        error: { _tag: "AssetAttachmentNotFoundError" },
+      });
+      await vi.waitFor(() => expect(mocks.upload).toHaveBeenCalledOnce());
+      // The upload is still running; the other two checks were already started.
+      expect(verifications.started()).toEqual(["pending-0-pdf", "pending-1-pdf", "pending-2-pdf"]);
+      verifications.resolve("pending-1-pdf");
+      verifications.resolve("pending-2-pdf");
+      uploading.resolve({ status: 204, body: "", headers: {} });
+      const prepared = await preparing;
+
+      expect(ids(prepared)).toEqual([MINTED_ID, "pending-1-pdf", "pending-2-pdf"]);
+      expect(prepared.status === "ready" && prepared.pendingAttachmentIds).toEqual([
+        MINTED_ID,
+        "pending-1-pdf",
+        "pending-2-pdf",
+      ]);
+    });
+
+    it("abandons without uploading when the owner cancels during verification", async () => {
+      const verifications = holdVerifications();
+      const controller = new AbortController();
+      const preparing = prepareTurnAttachments({
+        environmentId,
+        attachments: savedUploads(3),
+        signal: controller.signal,
+      });
+      await settle();
+
+      controller.abort();
+      verifications.resolve("pending-0-pdf");
+      verifications.resolve("pending-1-pdf");
+      verifications.resolve("pending-2-pdf");
+
+      expect(await preparing).toEqual({ status: "abandoned" });
+      expect(mocks.upload).not.toHaveBeenCalled();
+      expect(mocks.runAtomCommand).not.toHaveBeenCalled();
+    });
+  });
+
   it("deletes the minted uploads when the owner abandons the send", async () => {
     const result = await prepareTurnAttachments({
       environmentId,

apps/mobile/src/lib/attachmentUpload.ts

@@ -1,4 +1,4 @@
-import { resolveAssetUrl } from "@t3tools/client-runtime/state/assets";
+import { ASSET_URL_MAX_IN_FLIGHT, resolveAssetUrl } from "@t3tools/client-runtime/state/assets";
 import {
   isAssetAttachmentNotFoundFailure,
   runAttachmentUploadCycle,
@@ -107,6 +107,24 @@ async function releaseCreatedUploadsQuietly(
   }
 }
 
+/** Runs tasks at most `limit` at a time, starting waiting ones in call order. */
+function createConcurrencyLimit(limit: number) {
+  let active = 0;
+  const waiting: Array<() => void> = [];
+  return async <T>(task: () => Promise<T>): Promise<T> => {
+    if (active < limit) active += 1;
+    else await new Promise<void>((resolve) => waiting.push(resolve));
+    try {
+      return await task();
+    } finally {
+      // Hand the slot straight to the next waiter so `active` never exceeds `limit`.
+      const next = waiting.shift();
+      if (next) next();
+      else active -= 1;
+    }
+  };
+}
+
 export interface PreparedTurnAttachments {
   readonly status: "ready";
   readonly attachments: ReadonlyArray<UploadedMobileAttachment>;
@@ -304,8 +322,36 @@ export async function prepareTurnAttachments(input: {
   const controller = new AbortController();
   const abort = () => controller.abort();
   input.signal?.addEventListener("abort", abort, { once: true });
+  // Each saved upload is checked with one round trip. Start them all now, a few at a time like
+  // asset URL requests, and let the loop below read the results in attachment order, so N saved
+  // uploads cost about N / limit round trips instead of N. `stopVerifying` drops the ones that
+  // have not started once the loop has finished or thrown.
+  let stopVerifying = false;
+  const limitVerifications = createConcurrencyLimit(ASSET_URL_MAX_IN_FLIGHT);
+  const verifications = input.attachments.map((attachment) => {
+    if (
+      (attachment.type === "image" && !input.supportsImageUploads) ||
+      attachment.uploadEnvironmentId !== environmentId ||
+      attachment.uploadedAttachmentId === undefined
+    ) {
+      return undefined;
+    }
+    const attachmentId = attachment.uploadedAttachmentId;
+    const verification = limitVerifications(async () => {
+      if (stopVerifying || controller.signal.aborted) throw new Error("Upload cancelled.");
+      return await verifyPersistedAttachmentUpload({
+        registry: appAtomRegistry,
+        createAssetUrl: assetEnvironment.createUrl,
+        environmentId,
+        attachmentId,
+      });
+    });
+    // An earlier attachment can end the loop first; its rejection is read only where the loop awaits it.
+    verification.catch(() => {});
+    return verification;
+  });
   try {
-    for (const attachment of input.attachments) {
+    for (const [index, attachment] of input.attachments.entries()) {
       if (controller.signal.aborted) throw new Error("Upload cancelled.");
       requireUploadAccess();
       if (attachment.type === "image" && !input.supportsImageUploads) {
@@ -317,12 +363,7 @@ export async function prepareTurnAttachments(input: {
         attachment.uploadEnvironmentId === environmentId &&
         attachment.uploadedAttachmentId !== undefined
       ) {
-        const verification = await verifyPersistedAttachmentUpload({
-          registry: appAtomRegistry,
-          createAssetUrl: assetEnvironment.createUrl,
-          environmentId,
-          attachmentId: attachment.uploadedAttachmentId,
-        });
+        const verification = await verifications[index]!;
         if (verification.status === "failed") {
           throw verification.error;
         }
@@ -399,6 +440,7 @@ export async function prepareTurnAttachments(input: {
     if (controller.signal.aborted) return { status: "abandoned" };
     throw error;
   } finally {
+    stopVerifying = true;
     input.signal?.removeEventListener("abort", abort);
   }
 }