authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-09-04 09:20:31-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-04 09:20:31-07:00
log7839140e5e93d3f401d7eb45b86cf1a234eb3609
tree40c4972ff05a62b0c9f4c0a4fcc7ace46414168a
parentc7c1dfe4df99edf65a49d8a31b39ef1361f37f44
signature Signed by PGP key B5690EEEBB952194

fix(mobile): preserve saved work after storage read failures (#9710)

5 files changed, 240 insertions(+), 82 deletions(-)

apps/mobile/src/state/thread-outbox-manager.ts+7-11
...@@ -276,17 +276,13 @@ export function createThreadOutboxManager(options: ThreadOutboxManagerOptions) {...@@ -276,17 +276,13 @@ export function createThreadOutboxManager(options: ThreadOutboxManagerOptions) {
276 const revisionsAtRequest = new Map(revisions);276 const revisionsAtRequest = new Map(revisions);
277 return serialize(async () => {277 return serialize(async () => {
278 const persisted = await options.storage.load().catch((cause) => {278 const persisted = await options.storage.load().catch((cause) => {
279 warn(279 throw new ThreadOutboxManagerError({
280 "[thread-outbox] failed to load messages while clearing environment",280 operation: "clear-environment-load",
281 new ThreadOutboxManagerError({281 environmentId,
282 operation: "clear-environment-load",282 threadId: null,
283 environmentId,283 messageId: null,
284 threadId: null,284 cause,
285 messageId: null,285 });
286 cause,
287 }),
288 );
289 return [];
290 });286 });
291 const allMessages = flattenQueuedThreadMessages(287 const allMessages = flattenQueuedThreadMessages(
292 groupQueuedThreadMessages([...persisted, ...currentMessages()]),288 groupQueuedThreadMessages([...persisted, ...currentMessages()]),
apps/mobile/src/state/thread-outbox-storage.ts+11-11
...@@ -80,20 +80,20 @@ export const expoThreadOutboxStorage: ThreadOutboxStorage = {...@@ -80,20 +80,20 @@ export const expoThreadOutboxStorage: ThreadOutboxStorage = {
80 try {80 try {
81 messages.push(decodeQueuedThreadMessage(JSON.parse(await entry.text()) as unknown));81 messages.push(decodeQueuedThreadMessage(JSON.parse(await entry.text()) as unknown));
82 } catch (cause) {82 } catch (cause) {
83 console.warn(83 // A partial queue hides attachment owners from cleanup. Keep all
84 "[thread-outbox] ignored invalid persisted message",84 // records untouched until every persisted message can be read.
85 new ThreadOutboxStorageError({85 throw new ThreadOutboxStorageError({
86 operation: "read-message",86 operation: "read-message",
87 environmentId: null,87 environmentId: null,
88 threadId: null,88 threadId: null,
89 messageId: null,89 messageId: null,
90 fileName: entry.name,90 fileName: entry.name,
91 cause,91 cause,
92 }),92 });
93 );
94 }93 }
95 }94 }
96 } catch (cause) {95 } catch (cause) {
96 if (cause instanceof ThreadOutboxStorageError) throw cause;
97 throw new ThreadOutboxStorageError({97 throw new ThreadOutboxStorageError({
98 operation: "load",98 operation: "load",
99 environmentId: null,99 environmentId: null,
apps/mobile/src/state/thread-outbox.test.ts+95-1
...@@ -8,6 +8,34 @@ import {...@@ -8,6 +8,34 @@ import {
8 ThreadId,8 ThreadId,
9} from "@t3tools/contracts";9} from "@t3tools/contracts";
10import { AtomRegistry } from "effect/unstable/reactivity";10import { AtomRegistry } from "effect/unstable/reactivity";
11import { onTestFinished, vi } from "vite-plus/test";
12
13const outboxFiles = vi.hoisted(() => new Map<string, string | Error>());
14
15vi.mock("expo-file-system", () => {
16 class File {
17 constructor(readonly name: string) {}
18
19 async text(): Promise<string> {
20 const contents = outboxFiles.get(this.name);
21 if (contents instanceof Error) throw contents;
22 if (contents === undefined) throw new Error("Missing file");
23 return contents;
24 }
25 }
26
27 return {
28 File,
29 Directory: class {
30 create() {}
31
32 list() {
33 return Array.from(outboxFiles.keys(), (name) => new File(name));
34 }
35 },
36 Paths: { document: "/documents" },
37 };
38});
1139
12import {40import {
13 decodeQueuedThreadMessage,41 decodeQueuedThreadMessage,
...@@ -24,7 +52,7 @@ import {...@@ -24,7 +52,7 @@ import {
24 type QueuedThreadMessage,52 type QueuedThreadMessage,
25} from "./thread-outbox-model";53} from "./thread-outbox-model";
26import { createThreadOutboxManager, ThreadOutboxManagerError } from "./thread-outbox-manager";54import { createThreadOutboxManager, ThreadOutboxManagerError } from "./thread-outbox-manager";
27import type { ThreadOutboxStorage } from "./thread-outbox-storage";55import { expoThreadOutboxStorage, type ThreadOutboxStorage } from "./thread-outbox-storage";
2856
29function queuedMessage(input: {57function queuedMessage(input: {
30 readonly environmentId?: string;58 readonly environmentId?: string;
...@@ -44,6 +72,72 @@ function queuedMessage(input: {...@@ -44,6 +72,72 @@ function queuedMessage(input: {
44}72}
4573
46describe("thread outbox", () => {74describe("thread outbox", () => {
75 it.each(["read", "json", "schema"] as const)(
76 "does not load a partial outbox after a record %s failure",
77 async (failure) => {
78 onTestFinished(() => outboxFiles.clear());
79 const first = queuedMessage({
80 messageId: "message-1",
81 createdAt: "2026-06-08T10:00:01.000Z",
82 });
83 const second = queuedMessage({
84 messageId: "message-2",
85 createdAt: "2026-06-08T10:00:02.000Z",
86 });
87 outboxFiles.set("message-1.json", JSON.stringify(encodeQueuedThreadMessage(first)));
88 outboxFiles.set(
89 "message-2.json",
90 failure === "read"
91 ? new Error("storage unavailable")
92 : failure === "json"
93 ? "{"
94 : JSON.stringify({ ...second, schemaVersion: 999 }),
95 );
96
97 await expect(expoThreadOutboxStorage.load()).rejects.toMatchObject({
98 operation: "read-message",
99 fileName: "message-2.json",
100 });
101
102 outboxFiles.set("message-2.json", JSON.stringify(encodeQueuedThreadMessage(second)));
103 await expect(expoThreadOutboxStorage.load()).resolves.toEqual([first, second]);
104 },
105 );
106
107 it("preserves queued messages when environment cleanup cannot read the outbox", async () => {
108 const registry = AtomRegistry.make();
109 onTestFinished(() => registry.dispose());
110 const message = queuedMessage({
111 messageId: "message-1",
112 createdAt: "2026-06-08T10:00:01.000Z",
113 });
114 const stored = new Map<MessageId, QueuedThreadMessage>();
115 const manager = createThreadOutboxManager({
116 registry,
117 warn: () => {},
118 storage: {
119 load: async () => {
120 throw new Error("storage unavailable");
121 },
122 write: async (entry) => {
123 stored.set(entry.messageId, entry);
124 },
125 remove: async (entry) => {
126 stored.delete(entry.messageId);
127 },
128 },
129 });
130 await manager.enqueue(message);
131
132 await expect(manager.clearEnvironment(message.environmentId)).rejects.toMatchObject({
133 operation: "clear-environment-load",
134 });
135 expect([...stored.values()]).toEqual([message]);
136 expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({
137 "environment-1:thread-1": [message],
138 });
139 });
140
47 it("groups messages by scoped thread and preserves creation order", () => {141 it("groups messages by scoped thread and preserves creation order", () => {
48 const later = queuedMessage({142 const later = queuedMessage({
49 messageId: "message-2",143 messageId: "message-2",
apps/mobile/src/state/use-composer-drafts.test.ts+66-3
...@@ -9,7 +9,8 @@ import {...@@ -9,7 +9,8 @@ import {
9import { onTestFinished, vi } from "vite-plus/test";9import { onTestFinished, vi } from "vite-plus/test";
1010
11const composerDraftFileMocks = vi.hoisted(() => {11const composerDraftFileMocks = vi.hoisted(() => {
12 let document = "";12 let document = JSON.stringify({ schemaVersion: 1, drafts: {} });
13 let readError: Error | null = null;
13 let writeError: Error | null = null;14 let writeError: Error | null = null;
14 let releaseRead: (() => void) | null = null;15 let releaseRead: (() => void) | null = null;
15 let readBarrier = Promise.resolve();16 let readBarrier = Promise.resolve();
...@@ -33,6 +34,9 @@ const composerDraftFileMocks = vi.hoisted(() => {...@@ -33,6 +34,9 @@ const composerDraftFileMocks = vi.hoisted(() => {
33 setDocument(value: unknown) {34 setDocument(value: unknown) {
34 document = JSON.stringify(value);35 document = JSON.stringify(value);
35 },36 },
37 setReadError(error: Error | null) {
38 readError = error;
39 },
36 setWriteError(error: Error | null) {40 setWriteError(error: Error | null) {
37 writeError = error;41 writeError = error;
38 },42 },
...@@ -50,6 +54,10 @@ const composerDraftFileMocks = vi.hoisted(() => {...@@ -50,6 +54,10 @@ const composerDraftFileMocks = vi.hoisted(() => {
50 },54 },
51 Directory: class {55 Directory: class {
52 create() {}56 create() {}
57
58 list() {
59 return [];
60 }
53 },61 },
54 File: class {62 File: class {
55 exists = true;63 exists = true;
...@@ -61,6 +69,7 @@ const composerDraftFileMocks = vi.hoisted(() => {...@@ -61,6 +69,7 @@ const composerDraftFileMocks = vi.hoisted(() => {
6169
62 async text() {70 async text() {
63 await readBarrier;71 await readBarrier;
72 if (readError) throw readError;
64 return document;73 return document;
65 }74 }
6675
...@@ -156,7 +165,8 @@ const DRAFT: ComposerDraft = {...@@ -156,7 +165,8 @@ const DRAFT: ComposerDraft = {
156afterEach(() => {165afterEach(() => {
157 vi.useRealTimers();166 vi.useRealTimers();
158 resetComposerDraftsLoadState();167 resetComposerDraftsLoadState();
159 composerDraftFileMocks.setDocument("");168 composerDraftFileMocks.setDocument({ schemaVersion: 1, drafts: {} });
169 composerDraftFileMocks.setReadError(null);
160 composerDraftFileMocks.setWriteError(null);170 composerDraftFileMocks.setWriteError(null);
161 composerDraftFileMocks.setNextWriteBarrier(null);171 composerDraftFileMocks.setNextWriteBarrier(null);
162 composerDraftFileMocks.setOnWrite(null);172 composerDraftFileMocks.setOnWrite(null);
...@@ -1043,9 +1053,62 @@ describe("mobile composer drafts", () => {...@@ -1043,9 +1053,62 @@ describe("mobile composer drafts", () => {
1043 });1053 });
1044 });1054 });
10451055
1056 it.each(["read", "decode"] as const)(
1057 "preserves saved drafts and attachment files when the draft %s fails",
1058 async (failure) => {
1059 vi.useFakeTimers();
1060 const file = {
1061 id: "saved-file",
1062 type: "file" as const,
1063 name: "report.pdf",
1064 mimeType: "application/pdf",
1065 sizeBytes: 42,
1066 fileUri: "file:///documents/t3-composer-attachments/report.pdf",
1067 };
1068 composerDraftFileMocks.setDocument({
1069 schemaVersion: failure === "decode" ? 999 : 1,
1070 drafts: { "environment-1:saved": { text: "Saved draft", attachments: [file] } },
1071 });
1072 const original = composerDraftFileMocks.getDocument();
1073 if (failure === "read") {
1074 composerDraftFileMocks.setReadError(new Error("storage unavailable"));
1075 }
1076
1077 await expect(releaseUnusedComposerAttachmentFiles([file])).rejects.toMatchObject({
1078 operation: failure,
1079 });
1080 setComposerDraftText("environment-1:new", "Keep my new edits too");
1081 await expect(flushComposerDrafts()).rejects.toMatchObject({ operation: failure });
1082
1083 expect(composerDraftFileMocks.getDocument()).toBe(original);
1084 expect(composerAttachmentCleanupMocks.remove).not.toHaveBeenCalled();
1085 },
1086 );
1087
1088 it("retries a failed debounced read on final flush without dropping saved drafts or new edits", async () => {
1089 vi.useFakeTimers();
1090 composerDraftFileMocks.setDocument({
1091 schemaVersion: 1,
1092 drafts: { "environment-1:saved": DRAFT },
1093 });
1094 const original = composerDraftFileMocks.getDocument();
1095 composerDraftFileMocks.setReadError(new Error("storage unavailable"));
1096 setComposerDraftText("environment-1:new", "New edits");
1097 await vi.advanceTimersByTimeAsync(200);
1098 expect(composerDraftFileMocks.getDocument()).toBe(original);
1099
1100 composerDraftFileMocks.setReadError(null);
1101 await flushComposerDrafts();
1102
1103 expect(JSON.parse(composerDraftFileMocks.getDocument()).drafts).toEqual({
1104 "environment-1:saved": DRAFT,
1105 "environment-1:new": { text: "New edits", attachments: [] },
1106 });
1107 });
1108
1046 it("serializes environment cleanup after an older queued write", async () => {1109 it("serializes environment cleanup after an older queued write", async () => {
1047 vi.useFakeTimers();1110 vi.useFakeTimers();
1048 composerDraftFileMocks.setDocument(JSON.stringify({ schemaVersion: 1, drafts: {} }));1111 composerDraftFileMocks.setDocument({ schemaVersion: 1, drafts: {} });
1049 composerDraftFileMocks.resetWrites();1112 composerDraftFileMocks.resetWrites();
1050 let releaseFirstWrite!: () => void;1113 let releaseFirstWrite!: () => void;
1051 const firstWriteBarrier = new Promise<void>((resolve) => {1114 const firstWriteBarrier = new Promise<void>((resolve) => {
apps/mobile/src/state/use-composer-drafts.ts+61-56
...@@ -148,11 +148,13 @@ export const composerCloudDraftsAtom = Atom.make<ComposerCloudDraftState>({...@@ -148,11 +148,13 @@ export const composerCloudDraftsAtom = Atom.make<ComposerCloudDraftState>({
148148
149let loadPromise: Promise<void> | null = null;149let loadPromise: Promise<void> | null = null;
150let persistTimer: ReturnType<typeof setTimeout> | null = null;150let persistTimer: ReturnType<typeof setTimeout> | null = null;
151let persistRetryNeeded = false;
151const persistenceQueue = new SerializedAsyncQueue();152const persistenceQueue = new SerializedAsyncQueue();
152153
153/** Resets module-level state between test runs. */154/** Resets module-level state between test runs. */
154export function resetComposerDraftsLoadState(): void {155export function resetComposerDraftsLoadState(): void {
155 loadPromise = null;156 loadPromise = null;
157 persistRetryNeeded = false;
156}158}
157159
158function normalizeDraft(draft: ComposerDraft | undefined): ComposerDraft {160function normalizeDraft(draft: ComposerDraft | undefined): ComposerDraft {
...@@ -265,20 +267,12 @@ async function loadPersistedComposerState(): Promise<...@@ -265,20 +267,12 @@ async function loadPersistedComposerState(): Promise<
265 operation = "decode";267 operation = "decode";
266 return decodePersistedComposerState(JSON.parse(raw) as unknown);268 return decodePersistedComposerState(JSON.parse(raw) as unknown);
267 } catch (cause) {269 } catch (cause) {
268 console.warn(270 throw new ComposerDraftPersistenceError({
269 "[composer-drafts] ignored persisted draft failure",271 operation,
270 new ComposerDraftPersistenceError({272 directory: COMPOSER_DRAFTS_DIRECTORY,
271 operation,273 fileName: COMPOSER_DRAFTS_FILE,
272 directory: COMPOSER_DRAFTS_DIRECTORY,274 cause,
273 fileName: COMPOSER_DRAFTS_FILE,275 });
274 cause,
275 }),
276 );
277 return {
278 drafts: {},
279 stickyModelSelection: null,
280 cloudDrafts: { accountId: null, signedOut: {} },
281 };
282 }276 }
283}277}
284278
...@@ -341,20 +335,26 @@ export async function flushComposerDrafts(): Promise<void> {...@@ -341,20 +335,26 @@ export async function flushComposerDrafts(): Promise<void> {
341 // An edit during an awaited write schedules another debounced write, so335 // An edit during an awaited write schedules another debounced write, so
342 // keep landing snapshots until no debounce is pending after a queue drain.336 // keep landing snapshots until no debounce is pending after a queue drain.
343 do {337 do {
344 while (persistTimer !== null) {338 while (persistTimer !== null || persistRetryNeeded) {
345 clearTimeout(persistTimer);339 if (persistTimer !== null) clearTimeout(persistTimer);
346 persistTimer = null;340 persistTimer = null;
347 await persistenceQueue.run(() =>341 persistRetryNeeded = false;
348 writePersistedComposerState(342 try {
349 appAtomRegistry.get(composerDraftsAtom),343 await persistenceQueue.run(() =>
350 appAtomRegistry.get(stickyComposerModelSelectionAtom),344 writePersistedComposerState(
351 ),345 appAtomRegistry.get(composerDraftsAtom),
352 );346 appAtomRegistry.get(stickyComposerModelSelectionAtom),
347 ),
348 );
349 } catch (error) {
350 persistRetryNeeded = true;
351 throw error;
352 }
353 }353 }
354 // Draining also waits for an already-fired debounce whose write is still354 // Draining also waits for an already-fired debounce whose write is still
355 // gated behind its own hydration await inside the queue.355 // gated behind its own hydration await inside the queue.
356 await persistenceQueue.run(() => Promise.resolve());356 await persistenceQueue.run(() => Promise.resolve());
357 } while (persistTimer !== null);357 } while (persistTimer !== null || persistRetryNeeded);
358}358}
359359
360function signedOutAttachmentOwners() {360function signedOutAttachmentOwners() {
...@@ -534,19 +534,20 @@ function schedulePersistComposerState(): void {...@@ -534,19 +534,20 @@ function schedulePersistComposerState(): void {
534 }534 }
535 persistTimer = setTimeout(() => {535 persistTimer = setTimeout(() => {
536 persistTimer = null;536 persistTimer = null;
537 ensureComposerDraftsLoaded();
538 // The write enters the serialization queue before waiting on hydration,537 // The write enters the serialization queue before waiting on hydration,
539 // so flushComposerDrafts' queue drain cannot resolve ahead of it.538 // so flushComposerDrafts' queue drain cannot resolve ahead of it.
540 void persistenceQueue.run(async () => {539 void persistenceQueue.run(async () => {
541 if (loadPromise !== null) {
542 await loadPromise;
543 }
544 try {540 try {
541 await waitForComposerDraftsLoaded();
545 await writePersistedComposerState(542 await writePersistedComposerState(
546 appAtomRegistry.get(composerDraftsAtom),543 appAtomRegistry.get(composerDraftsAtom),
547 appAtomRegistry.get(stickyComposerModelSelectionAtom),544 appAtomRegistry.get(stickyComposerModelSelectionAtom),
548 );545 );
546 persistRetryNeeded = false;
549 } catch (error) {547 } catch (error) {
548 // A failed debounce has no timer left. A later final flush must retry
549 // these edits after persisted ownership can be read safely.
550 persistRetryNeeded = true;
550 console.warn("[composer-drafts] failed to persist drafts", error);551 console.warn("[composer-drafts] failed to persist drafts", error);
551 // Draft persistence is best-effort; in-memory drafts still keep working.552 // Draft persistence is best-effort; in-memory drafts still keep working.
552 }553 }
...@@ -558,35 +559,39 @@ export function ensureComposerDraftsLoaded(): void {...@@ -558,35 +559,39 @@ export function ensureComposerDraftsLoaded(): void {
558 if (loadPromise !== null) {559 if (loadPromise !== null) {
559 return;560 return;
560 }561 }
561 loadPromise = loadPersistedComposerState()562 const loading = loadPersistedComposerState().then((persisted) => {
562 .then((persisted) => {563 appAtomRegistry.set(composerCloudDraftsAtom, persisted.cloudDrafts);
563 appAtomRegistry.set(composerCloudDraftsAtom, persisted.cloudDrafts);564 if (Object.keys(persisted.drafts).length > 0) {
564 if (Object.keys(persisted.drafts).length > 0) {565 const current = appAtomRegistry.get(composerDraftsAtom);
565 const current = appAtomRegistry.get(composerDraftsAtom);566 appAtomRegistry.set(composerDraftsAtom, {
566 appAtomRegistry.set(composerDraftsAtom, {567 ...persisted.drafts,
567 ...persisted.drafts,568 ...current,
568 ...current,569 });
569 });570 }
570 }571 if (
571 if (572 persisted.stickyModelSelection !== null &&
572 persisted.stickyModelSelection !== null &&573 appAtomRegistry.get(stickyComposerModelSelectionAtom) === null
573 appAtomRegistry.get(stickyComposerModelSelectionAtom) === null574 ) {
574 ) {575 appAtomRegistry.set(stickyComposerModelSelectionAtom, persisted.stickyModelSelection);
575 appAtomRegistry.set(stickyComposerModelSelectionAtom, persisted.stickyModelSelection);576 }
576 }577 });
577 })578 loadPromise = loading;
578 .catch((cause) => {579 // Handle fire-and-forget hook loads without swallowing failures from the
579 console.warn(580 // write and cleanup callers that await this same promise. A later call retries.
580 "[composer-drafts] failed to hydrate drafts",581 void loading.catch((cause) => {
581 new ComposerDraftPersistenceError({582 if (loadPromise === loading) loadPromise = null;
582 operation: "hydrate",583 console.warn(
583 directory: COMPOSER_DRAFTS_DIRECTORY,584 "[composer-drafts] failed to hydrate drafts",
584 fileName: COMPOSER_DRAFTS_FILE,585 cause instanceof ComposerDraftPersistenceError
585 cause,586 ? cause
586 }),587 : new ComposerDraftPersistenceError({
587 );588 operation: "hydrate",
588 // Draft loading is best-effort; in-memory drafts still keep working.589 directory: COMPOSER_DRAFTS_DIRECTORY,
589 });590 fileName: COMPOSER_DRAFTS_FILE,
591 cause,
592 }),
593 );
594 });
590}595}
591596
592/** Wait until persisted drafts have been merged into the in-memory composer state. */597/** Wait until persisted drafts have been merged into the in-memory composer state. */