From 7ab121ff549c086dcbce834660f9bdad730aac77 Mon Sep 17 00:00:00 2001 From: Pyseph <45090858+Pyseph@users.noreply.github.com> Date: Wed, 4 Nov 2026 10:14:07 +0100 Subject: [PATCH] perf(mobile): verify saved attachment uploads in parallel on delivery prepareTurnAttachments re-verified every already uploaded attachment with its own assets.createUrl round trip, one after another, so a message with N saved uploads waited N round trips before the send command could go out. Start the verifications up front, at most ASSET_URL_MAX_IN_FLIGHT (6) at a time, and let the existing loop read the results in attachment order. Result order, which failing attachment is reported, upload and cancellation handling stay as they were. Verifications that have not started are dropped once the loop ends or the send is cancelled. --- apps/mobile/src/lib/attachmentUpload.test.ts | 198 +++++++++++++++++++ apps/mobile/src/lib/attachmentUpload.ts | 58 +++++- 2 files changed, 248 insertions(+), 8 deletions(-) diff --git a/apps/mobile/src/lib/attachmentUpload.test.ts b/apps/mobile/src/lib/attachmentUpload.test.ts index 54dfa536ff366..e3dff98e2b97a 100644 --- a/apps/mobile/src/lib/attachmentUpload.test.ts +++ b/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>(); + const inFlight = { max: 0, current: 0 }; + mocks.executeAtomQuery.mockImplementation( + async (_registry: unknown, target: unknown): Promise => { + const { attachmentId } = (target as { input: { resource: { attachmentId: string } } }) + .input.resource; + const deferred = Promise.withResolvers(); + 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) => + preparing.then( + () => "resolved", + (error: unknown) => error, + ); + const ids = (prepared: Awaited>) => + 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, diff --git a/apps/mobile/src/lib/attachmentUpload.ts b/apps/mobile/src/lib/attachmentUpload.ts index aef8dd4447513..924fb1f306986 100644 --- a/apps/mobile/src/lib/attachmentUpload.ts +++ b/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 (task: () => Promise): Promise => { + if (active < limit) active += 1; + else await new Promise((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; @@ -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); } }