authorgravatar for 71498452+StiensWout@users.noreply.github.comWout Stiens <71498452+StiensWout@users.noreply.github.com> 2026-09-22 11:25:14+02:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-22 02:25:14-07:00
logaff9318bf46beaf05cc7155b428d3f0b8711efd2
treee6c97ee915fb37744a49afb6d030115a7baff2ca
parentb5a0f810108d42ca8635b5a3d75a6e885bb3a254
signature Signed by PGP key B5690EEEBB952194

fix(web): retry failed attachment uploads after reconnect (#10338)

3 files changed, 145 insertions(+), 2 deletions(-)

apps/mobile/src/lib/composerAttachmentUploadQueue.test.ts+25
...@@ -105,6 +105,31 @@ describe("composer attachment upload queue", () => {...@@ -105,6 +105,31 @@ describe("composer attachment upload queue", () => {
105 queue.dispose();105 queue.dispose();
106 });106 });
107107
108 it("retries a failed attachment when the worker restores requests after reconnect", async () => {
109 let states: Readonly<Record<string, ComposerAttachmentUploadState>> = {};
110 const upload = vi.fn().mockRejectedValueOnce(new Error("Disconnected")).mockResolvedValue(true);
111 const queue = createComposerAttachmentUploadQueue({
112 upload,
113 onChange: (next) => {
114 states = next;
115 },
116 });
117 const local = request("failed-upload");
118 const key = composerAttachmentUploadKey(environmentId, local.attachment.id);
119 queue.sync([local]);
120 await queue.settled();
121 expect(states[key]?.status).toBe("failed");
122 queue.sync([local]);
123 await queue.settled();
124 expect(upload).toHaveBeenCalledTimes(1);
125 queue.sync([]);
126 queue.sync([local]);
127 await queue.settled();
128 expect(upload).toHaveBeenCalledTimes(2);
129 expect(states[key]?.status).toBe("ready");
130 queue.dispose();
131 });
132
108 it("ignores a late completion after removal or environment switch", async () => {133 it("ignores a late completion after removal or environment switch", async () => {
109 const gate = Promise.withResolvers<boolean>();134 const gate = Promise.withResolvers<boolean>();
110 const started = Promise.withResolvers<void>();135 const started = Promise.withResolvers<void>();
apps/web/src/lib/attachmentUploadQueue.test.ts+84-1
...@@ -1,5 +1,7 @@...@@ -1,5 +1,7 @@
1import { EnvironmentId } from "@t3tools/contracts";1import { EnvironmentId } from "@t3tools/contracts";
2import { afterEach, beforeEach, describe, expect, it, vi } from "vite-plus/test";2import { afterEach, beforeEach, describe, expect, it, vi } from "vite-plus/test";
3import { Atom, AsyncResult } from "effect/unstable/reactivity";
4import { appAtomRegistry } from "../rpc/atomRegistry";
35
4import {6import {
5 composerFileNeedsReattach,7 composerFileNeedsReattach,
...@@ -10,6 +12,7 @@ import {...@@ -10,6 +12,7 @@ import {
10} from "../composerDraftStore";12} from "../composerDraftStore";
1113
12const mocks = vi.hoisted(() => ({14const mocks = vi.hoisted(() => ({
15 connectionStateAtom: vi.fn(),
13 createAssetUrl: vi.fn(),16 createAssetUrl: vi.fn(),
14 createUploadUrl: Symbol("create-upload-url"),17 createUploadUrl: Symbol("create-upload-url"),
15 executeAtomQuery: vi.fn(),18 executeAtomQuery: vi.fn(),
...@@ -24,7 +27,14 @@ vi.mock("@t3tools/client-runtime/state/runtime", () => ({...@@ -24,7 +27,14 @@ vi.mock("@t3tools/client-runtime/state/runtime", () => ({
24 squashAtomCommandFailure: (result: { readonly error: unknown }) => result.error,27 squashAtomCommandFailure: (result: { readonly error: unknown }) => result.error,
25}));28}));
2629
27vi.mock("../rpc/atomRegistry", () => ({ appAtomRegistry: {} }));30vi.mock("../rpc/atomRegistry", async () => {
31 const { AtomRegistry } = await import("effect/unstable/reactivity");
32 return { appAtomRegistry: AtomRegistry.make() };
33});
34
35vi.mock("../connection/catalog", () => ({
36 environmentCatalog: { stateAtom: mocks.connectionStateAtom },
37}));
2838
29vi.mock("../state/assets", () => ({39vi.mock("../state/assets", () => ({
30 assetEnvironment: { createUrl: mocks.createAssetUrl },40 assetEnvironment: { createUrl: mocks.createAssetUrl },
...@@ -140,8 +150,22 @@ function makeFile(id: string): ComposerFileAttachment {...@@ -140,8 +150,22 @@ function makeFile(id: string): ComposerFileAttachment {
140 };150 };
141}151}
142152
153const connectionStates = Atom.family((_environmentId: EnvironmentId) =>
154 Atom.make(AsyncResult.success({ phase: "connected" })),
155);
156
157function setConnected(environmentId: EnvironmentId, connected: boolean) {
158 appAtomRegistry.set(
159 connectionStates(environmentId),
160 AsyncResult.success({ phase: connected ? "connected" : "backoff" }),
161 );
162}
163
143describe("attachmentUploadQueue", () => {164describe("attachmentUploadQueue", () => {
144 beforeEach(() => {165 beforeEach(() => {
166 mocks.connectionStateAtom.mockImplementation(connectionStates);
167 setConnected(firstEnvironment, true);
168 setConnected(secondEnvironment, true);
145 TestXmlHttpRequest.requests = [];169 TestXmlHttpRequest.requests = [];
146 mocks.createAssetUrl.mockReset();170 mocks.createAssetUrl.mockReset();
147 mocks.createAssetUrl.mockImplementation((target: unknown) => target);171 mocks.createAssetUrl.mockImplementation((target: unknown) => target);
...@@ -183,6 +207,65 @@ describe("attachmentUploadQueue", () => {...@@ -183,6 +207,65 @@ describe("attachmentUploadQueue", () => {
183 vi.unstubAllGlobals();207 vi.unstubAllGlobals();
184 });208 });
185209
210 it.each([false, true])(
211 "retries a failed file once after reconnect, including a late HTTP failure: %s",
212 async (lateFailure) => {
213 const image = makeFile("reconnect");
214 startAttachmentUpload({ environmentId: firstEnvironment, image });
215 await Promise.resolve();
216 const firstSettled = awaitAttachmentUploads([image.id]);
217 setConnected(firstEnvironment, false);
218 if (lateFailure) setConnected(firstEnvironment, true);
219 TestXmlHttpRequest.requests[0]!.complete(503);
220 await firstSettled;
221 if (!lateFailure) setConnected(firstEnvironment, true);
222 await Promise.resolve();
223 await Promise.resolve();
224 expect(TestXmlHttpRequest.requests).toHaveLength(2);
225 const retrySettled = awaitAttachmentUploads([image.id]);
226 TestXmlHttpRequest.requests[1]!.complete(503);
227 await retrySettled;
228 setConnected(firstEnvironment, true);
229 await Promise.resolve();
230 expect(TestXmlHttpRequest.requests).toHaveLength(2);
231 expect(readAttachmentUpload(image.id)?.status).toBe("failed");
232
233 setConnected(firstEnvironment, false);
234 setConnected(firstEnvironment, true);
235 await Promise.resolve();
236 await Promise.resolve();
237 const finalSettled = awaitAttachmentUploads([image.id]);
238 TestXmlHttpRequest.requests[2]!.complete();
239 await finalSettled;
240 expect(
241 getUploadedAttachments({ environmentId: firstEnvironment, images: [image] }),
242 ).not.toBeNull();
243 setConnected(firstEnvironment, false);
244 setConnected(firstEnvironment, true);
245 await Promise.resolve();
246 expect(TestXmlHttpRequest.requests).toHaveLength(3);
247 },
248 );
249
250 it("does not retry for another environment or after the attachment is removed", async () => {
251 const image = makeFile("removed");
252 startAttachmentUpload({ environmentId: firstEnvironment, image });
253 await Promise.resolve();
254 const settled = awaitAttachmentUploads([image.id]);
255 TestXmlHttpRequest.requests[0]!.complete(503);
256 await settled;
257 setConnected(secondEnvironment, false);
258 setConnected(secondEnvironment, true);
259 await Promise.resolve();
260 expect(TestXmlHttpRequest.requests).toHaveLength(1);
261 setConnected(firstEnvironment, false);
262 setConnected(firstEnvironment, true);
263 releaseAttachmentUpload(image.id);
264 await Promise.resolve();
265 expect(TestXmlHttpRequest.requests).toHaveLength(1);
266 expect(readAttachmentUpload(image.id)).toBeUndefined();
267 });
268
186 it("uploads images immediately and sends attachment references", async () => {269 it("uploads images immediately and sends attachment references", async () => {
187 const image = {270 const image = {
188 ...makeImage("image-1"),271 ...makeImage("image-1"),
apps/web/src/lib/attachmentUploadQueue.ts+36-1
...@@ -12,6 +12,8 @@ import {...@@ -12,6 +12,8 @@ import {
12 type PersistedAttachmentVerification,12 type PersistedAttachmentVerification,
13} from "@t3tools/client-runtime/state/attachments";13} from "@t3tools/client-runtime/state/attachments";
14import { create } from "zustand";14import { create } from "zustand";
15import * as Option from "effect/Option";
16import { AsyncResult } from "effect/unstable/reactivity";
1517
16import {18import {
17 DraftId,19 DraftId,
...@@ -21,6 +23,7 @@ import {...@@ -21,6 +23,7 @@ import {
21 type ComposerThreadTarget,23 type ComposerThreadTarget,
22} from "../composerDraftStore";24} from "../composerDraftStore";
23import { appAtomRegistry } from "../rpc/atomRegistry";25import { appAtomRegistry } from "../rpc/atomRegistry";
26import { environmentCatalog } from "../connection/catalog";
24import { assetEnvironment } from "../state/assets";27import { assetEnvironment } from "../state/assets";
25import { attachmentEnvironment } from "../state/attachments";28import { attachmentEnvironment } from "../state/attachments";
26import { readPreparedConnection } from "../state/session";29import { readPreparedConnection } from "../state/session";
...@@ -58,8 +61,10 @@ interface UploadJob {...@@ -58,8 +61,10 @@ interface UploadJob {
58 attachmentId: string | null;61 attachmentId: string | null;
59 cancelled: boolean;62 cancelled: boolean;
60 abort: (() => void) | null;63 abort: (() => void) | null;
64 stopWatchingConnection: () => void;
61}65}
6266
67// Failed jobs retain their source and connection subscription until retry or release.
63const jobsByImageId = new Map<string, UploadJob>();68const jobsByImageId = new Map<string, UploadJob>();
64const queue: UploadJob[] = [];69const queue: UploadJob[] = [];
65const activeUploadsByEnvironment = new Map<EnvironmentId, number>();70const activeUploadsByEnvironment = new Map<EnvironmentId, number>();
...@@ -358,7 +363,11 @@ function pumpUploads(): void {...@@ -358,7 +363,11 @@ function pumpUploads(): void {
358 }363 }
359 })364 })
360 .finally(() => {365 .finally(() => {
361 if (jobsByImageId.get(job.image.id) === job) {366 if (
367 jobsByImageId.get(job.image.id) === job &&
368 readAttachmentUpload(job.image.id)?.status !== "failed"
369 ) {
370 job.stopWatchingConnection();
362 jobsByImageId.delete(job.image.id);371 jobsByImageId.delete(job.image.id);
363 }372 }
364 const remaining = (activeUploadsByEnvironment.get(job.environmentId) ?? 1) - 1;373 const remaining = (activeUploadsByEnvironment.get(job.environmentId) ?? 1) - 1;
...@@ -427,9 +436,34 @@ export function startAttachmentUpload(input: {...@@ -427,9 +436,34 @@ export function startAttachmentUpload(input: {
427 attachmentId: null,436 attachmentId: null,
428 cancelled: false,437 cancelled: false,
429 abort: null,438 abort: null,
439 stopWatchingConnection: () => {},
430 };440 };
431441
432 jobsByImageId.set(input.image.id, job);442 jobsByImageId.set(input.image.id, job);
443 const connectionAtom = environmentCatalog.stateAtom(job.environmentId);
444 const isConnected = () =>
445 Option.exists(
446 AsyncResult.value(appAtomRegistry.get(connectionAtom)),
447 (state) => state.phase === "connected",
448 );
449 let wasConnected = isConnected();
450 job.stopWatchingConnection = appAtomRegistry.subscribe(connectionAtom, () => {
451 const connected = isConnected();
452 const reconnected = connected && !wasConnected;
453 wasConnected = connected;
454 if (!reconnected) return;
455 // The HTTP failure can arrive after the socket has already reconnected.
456 // Wait for that attempt, then retry only if this job still owns the file.
457 void job.settled.then(() => {
458 if (
459 jobsByImageId.get(job.image.id) === job &&
460 readAttachmentUpload(job.image.id)?.status === "failed" &&
461 isConnected()
462 ) {
463 retryAttachmentUpload(input);
464 }
465 });
466 });
433 queue.push(job);467 queue.push(job);
434 setUploadState(input.image.id, {468 setUploadState(input.image.id, {
435 status: "uploading",469 status: "uploading",
...@@ -451,6 +485,7 @@ function cancelAttachmentUpload(imageId: string): void {...@@ -451,6 +485,7 @@ function cancelAttachmentUpload(imageId: string): void {
451 return;485 return;
452 }486 }
453 job.cancelled = true;487 job.cancelled = true;
488 job.stopWatchingConnection();
454 jobsByImageId.delete(imageId);489 jobsByImageId.delete(imageId);
455 const queuedIndex = queue.indexOf(job);490 const queuedIndex = queue.indexOf(job);
456 if (queuedIndex !== -1) {491 if (queuedIndex !== -1) {