History

[codex] Structure mobile thread outbox failures (#3341)

Julius Marminge committed 4 months ago

Commit 8ce627b1 parent 9a78c6f3 files · +208 −30Browse files

Unified · Split

3 files
  1. apps/mobile/src/state/thread-outbox-manager.ts +78 −9
  2. apps/mobile/src/state/thread-outbox-storage.ts +80 −18
  3. apps/mobile/src/state/thread-outbox.test.ts +50 −3
apps/mobile/src/state/thread-outbox-manager.ts+78−9
@@ -1,4 +1,5 @@
1import type { EnvironmentId, MessageId } from "@t3tools/contracts";1import { EnvironmentId, MessageId, ThreadId } from "@t3tools/contracts";
2import * as Schema from "effect/Schema";
2import { Atom, type AtomRegistry } from "effect/unstable/reactivity";3import { Atom, type AtomRegistry } from "effect/unstable/reactivity";
3 4
4import {5import {
@@ -8,6 +9,27 @@ import { · 3 unmodified lines
8} from "./thread-outbox-model";9} from "./thread-outbox-model";
9import type { ThreadOutboxStorage } from "./thread-outbox-storage";10import type { ThreadOutboxStorage } from "./thread-outbox-storage";
10 11
12export class ThreadOutboxManagerError extends Schema.TaggedErrorClass<ThreadOutboxManagerError>()(
13 "ThreadOutboxManagerError",
14 {
15 operation: Schema.Literals([
16 "load",
17 "enqueue",
18 "remove",
19 "clear-environment-load",
20 "clear-environment-remove",
21 ]),
22 environmentId: Schema.NullOr(EnvironmentId),
23 threadId: Schema.NullOr(ThreadId),
24 messageId: Schema.NullOr(MessageId),
25 cause: Schema.Defect(),
26 },
27) {
28 override get message(): string {
29 return `Thread outbox operation ${this.operation} failed for environment ${this.environmentId ?? "unknown"}, thread ${this.threadId ?? "unknown"}, message ${this.messageId ?? "unknown"}.`;
30 }
31}
32
11export interface ThreadOutboxManagerOptions {33export interface ThreadOutboxManagerOptions {
12 readonly registry: AtomRegistry.AtomRegistry;34 readonly registry: AtomRegistry.AtomRegistry;
13 readonly storage: ThreadOutboxStorage;35 readonly storage: ThreadOutboxStorage;
@@ -49,22 +71,51 @@ export function createThreadOutboxManager(options: ThreadOutboxManagerOptions) { · 35 unmodified lines
49 loadPromise = serialize(async () => {71 loadPromise = serialize(async () => {
50 const persistedMessages = await options.storage.load();72 const persistedMessages = await options.storage.load();
51 setMessages([...persistedMessages, ...currentMessages()]);73 setMessages([...persistedMessages, ...currentMessages()]);
52 }).catch((error) => {74 }).catch((cause) => {
53 loadPromise = null;75 loadPromise = null;
54 warn("[thread-outbox] failed to load persisted messages", error);76 warn(
77 "[thread-outbox] failed to load persisted messages",
78 new ThreadOutboxManagerError({
79 operation: "load",
80 environmentId: null,
81 threadId: null,
82 messageId: null,
83 cause,
84 }),
85 );
55 });86 });
56 return loadPromise;87 return loadPromise;
57 };88 };
58 89
59 const enqueue = (message: QueuedThreadMessage): Promise<void> =>90 const enqueue = (message: QueuedThreadMessage): Promise<void> =>
60 serialize(async () => {91 serialize(async () => {
61 await options.storage.write(message);92 try {
93 await options.storage.write(message);
94 } catch (cause) {
95 throw new ThreadOutboxManagerError({
96 operation: "enqueue",
97 environmentId: message.environmentId,
98 threadId: message.threadId,
99 messageId: message.messageId,
100 cause,
101 });
102 }
62 setMessages([...currentMessages(), message]);103 setMessages([...currentMessages(), message]);
63 });104 });
64 105
65 const remove = (message: QueuedThreadMessage): Promise<void> =>106 const remove = (message: QueuedThreadMessage): Promise<void> =>
66 serialize(async () => {107 serialize(async () => {
67 await options.storage.remove(message);108 try {
109 await options.storage.remove(message);
110 } catch (cause) {
111 throw new ThreadOutboxManagerError({
112 operation: "remove",
113 environmentId: message.environmentId,
114 threadId: message.threadId,
115 messageId: message.messageId,
116 cause,
117 });
118 }
68 setMessages(119 setMessages(
69 currentMessages().filter((candidate) => candidate.messageId !== message.messageId),120 currentMessages().filter((candidate) => candidate.messageId !== message.messageId),
70 );121 );
@@ -72,8 +123,17 @@ export function createThreadOutboxManager(options: ThreadOutboxManagerOptions) { · 1 unmodified line
72 123
73 const clearEnvironment = (environmentId: EnvironmentId): Promise<void> =>124 const clearEnvironment = (environmentId: EnvironmentId): Promise<void> =>
74 serialize(async () => {125 serialize(async () => {
75 const persisted = await options.storage.load().catch((error) => {126 const persisted = await options.storage.load().catch((cause) => {
76 warn("[thread-outbox] failed to load messages while clearing environment", error);127 warn(
128 "[thread-outbox] failed to load messages while clearing environment",
129 new ThreadOutboxManagerError({
130 operation: "clear-environment-load",
131 environmentId,
132 threadId: null,
133 messageId: null,
134 cause,
135 }),
136 );
77 return [];137 return [];
78 });138 });
79 const allMessages = flattenQueuedThreadMessages(139 const allMessages = flattenQueuedThreadMessages(
@@ -88,8 +148,17 @@ export function createThreadOutboxManager(options: ThreadOutboxManagerOptions) { · 8 unmodified lines
88 try {148 try {
89 await options.storage.remove(message);149 await options.storage.remove(message);
90 removedMessageIds.add(message.messageId);150 removedMessageIds.add(message.messageId);
91 } catch (error) {151 } catch (cause) {
92 warn("[thread-outbox] failed to clear persisted message", error);152 warn(
153 "[thread-outbox] failed to clear persisted message",
154 new ThreadOutboxManagerError({
155 operation: "clear-environment-remove",
156 environmentId: message.environmentId,
157 threadId: message.threadId,
158 messageId: message.messageId,
159 cause,
160 }),
161 );
93 }162 }
94 }),163 }),
95 );164 );
apps/mobile/src/state/thread-outbox-storage.ts+80−18
@@ -1,4 +1,5 @@
1import type { MessageId } from "@t3tools/contracts";1import { EnvironmentId, MessageId, ThreadId } from "@t3tools/contracts";
2import * as Schema from "effect/Schema";
2 3
3import {4import {
4 decodeQueuedThreadMessage,5 decodeQueuedThreadMessage,
@@ -8,6 +9,22 @@ import { · 3 unmodified lines
8 9
9const THREAD_OUTBOX_DIRECTORY = "thread-outbox";10const THREAD_OUTBOX_DIRECTORY = "thread-outbox";
10 11
12export class ThreadOutboxStorageError extends Schema.TaggedErrorClass<ThreadOutboxStorageError>()(
13 "ThreadOutboxStorageError",
14 {
15 operation: Schema.Literals(["load", "read-message", "write", "remove"]),
16 environmentId: Schema.NullOr(EnvironmentId),
17 threadId: Schema.NullOr(ThreadId),
18 messageId: Schema.NullOr(MessageId),
19 fileName: Schema.NullOr(Schema.String),
20 cause: Schema.Defect(),
21 },
22) {
23 override get message(): string {
24 return `Thread outbox storage operation ${this.operation} failed for environment ${this.environmentId ?? "unknown"}, thread ${this.threadId ?? "unknown"}, message ${this.messageId ?? "unknown"}, file ${this.fileName ?? "unknown"}.`;
25 }
26}
27
11export interface ThreadOutboxStorage {28export interface ThreadOutboxStorage {
12 readonly load: () => Promise<ReadonlyArray<QueuedThreadMessage>>;29 readonly load: () => Promise<ReadonlyArray<QueuedThreadMessage>>;
13 readonly write: (message: QueuedThreadMessage) => Promise<void>;30 readonly write: (message: QueuedThreadMessage) => Promise<void>;
@@ -32,33 +49,78 @@ async function getMessageFile(messageId: MessageId) { · 18 unmodified lines
32 49
33export const expoThreadOutboxStorage: ThreadOutboxStorage = {50export const expoThreadOutboxStorage: ThreadOutboxStorage = {
34 load: async () => {51 load: async () => {
35 const { File } = await import("expo-file-system");
36 const directory = await getOutboxDirectory();
37 const messages: QueuedThreadMessage[] = [];52 const messages: QueuedThreadMessage[] = [];
53 try {
54 const { File } = await import("expo-file-system");
55 const directory = await getOutboxDirectory();
38 56
39 for (const entry of directory.list()) {57 for (const entry of directory.list()) {
40 if (!(entry instanceof File) || !entry.name.endsWith(".json")) {58 if (!(entry instanceof File) || !entry.name.endsWith(".json")) {
41 continue;59 continue;
42 }60 }
43 try {61 try {
44 messages.push(decodeQueuedThreadMessage(JSON.parse(await entry.text()) as unknown));62 messages.push(decodeQueuedThreadMessage(JSON.parse(await entry.text()) as unknown));
45 } catch (error) {63 } catch (cause) {
46 console.warn("[thread-outbox] ignored invalid persisted message", entry.name, error);64 console.warn(
65 "[thread-outbox] ignored invalid persisted message",
66 new ThreadOutboxStorageError({
67 operation: "read-message",
68 environmentId: null,
69 threadId: null,
70 messageId: null,
71 fileName: entry.name,
72 cause,
73 }),
74 );
75 }
47 }76 }
77 } catch (cause) {
78 throw new ThreadOutboxStorageError({
79 operation: "load",
80 environmentId: null,
81 threadId: null,
82 messageId: null,
83 fileName: null,
84 cause,
85 });
48 }86 }
49 return messages;87 return messages;
50 },88 },
51 write: async (message) => {89 write: async (message) => {
52 const file = await getMessageFile(message.messageId);90 const fileName = messageFileName(message.messageId);
53 if (!file.exists) {91 try {
54 file.create({ intermediates: true, overwrite: true });92 const file = await getMessageFile(message.messageId);
93 if (!file.exists) {
94 file.create({ intermediates: true, overwrite: true });
95 }
96 file.write(JSON.stringify(encodeQueuedThreadMessage(message)));
97 } catch (cause) {
98 throw new ThreadOutboxStorageError({
99 operation: "write",
100 environmentId: message.environmentId,
101 threadId: message.threadId,
102 messageId: message.messageId,
103 fileName,
104 cause,
105 });
55 }106 }
56 file.write(JSON.stringify(encodeQueuedThreadMessage(message)));
57 },107 },
58 remove: async (message) => {108 remove: async (message) => {
59 const file = await getMessageFile(message.messageId);109 const fileName = messageFileName(message.messageId);
60 if (file.exists) {110 try {
61 file.delete();111 const file = await getMessageFile(message.messageId);
112 if (file.exists) {
113 file.delete();
114 }
115 } catch (cause) {
116 throw new ThreadOutboxStorageError({
117 operation: "remove",
118 environmentId: message.environmentId,
119 threadId: message.threadId,
120 messageId: message.messageId,
121 fileName,
122 cause,
123 });
62 }124 }
63 },125 },
64};126};
apps/mobile/src/state/thread-outbox.test.ts+50−3
@@ -10,7 +10,7 @@ import { · 9 unmodified lines
10 threadOutboxRetryDelayMs,10 threadOutboxRetryDelayMs,
11 type QueuedThreadMessage,11 type QueuedThreadMessage,
12} from "./thread-outbox-model";12} from "./thread-outbox-model";
13import { createThreadOutboxManager } from "./thread-outbox-manager";13import { createThreadOutboxManager, ThreadOutboxManagerError } from "./thread-outbox-manager";
14import type { ThreadOutboxStorage } from "./thread-outbox-storage";14import type { ThreadOutboxStorage } from "./thread-outbox-storage";
15 15
16function queuedMessage(input: {16function queuedMessage(input: {
@@ -149,9 +149,48 @@ describe("thread outbox", () => { · 132 unmodified lines
149 registry.dispose();149 registry.dispose();
150 });150 });
151 151
152 it("reports structured load failures and permits a retry", async () => {
153 const registry = AtomRegistry.make();
154 const loadCause = new Error("storage unavailable");
155 const warnings: Array<{ message: string; error: unknown }> = [];
156 let loadCalls = 0;
157 const manager = createThreadOutboxManager({
158 registry,
159 storage: {
160 load: async () => {
161 loadCalls += 1;
162 if (loadCalls === 1) throw loadCause;
163 return [];
164 },
165 write: async () => undefined,
166 remove: async () => undefined,
167 },
168 warn: (message, error) => warnings.push({ message, error }),
169 });
170
171 await manager.load();
172 expect(warnings).toEqual([
173 {
174 message: "[thread-outbox] failed to load persisted messages",
175 error: new ThreadOutboxManagerError({
176 operation: "load",
177 environmentId: null,
178 threadId: null,
179 messageId: null,
180 cause: loadCause,
181 }),
182 },
183 ]);
184
185 await manager.load();
186 expect(loadCalls).toBe(2);
187 registry.dispose();
188 });
189
152 it("keeps atom state aligned with durable writes and removals", async () => {190 it("keeps atom state aligned with durable writes and removals", async () => {
153 const registry = AtomRegistry.make();191 const registry = AtomRegistry.make();
154 const stored = new Map<MessageId, QueuedThreadMessage>();192 const stored = new Map<MessageId, QueuedThreadMessage>();
193 const removalCause = new Error("remove failed");
155 let failRemoval = true;194 let failRemoval = true;
156 const storage: ThreadOutboxStorage = {195 const storage: ThreadOutboxStorage = {
157 load: async () => [...stored.values()],196 load: async () => [...stored.values()],
@@ -160,7 +199,7 @@ describe("thread outbox", () => { · 2 unmodified lines
160 },199 },
161 remove: async (message) => {200 remove: async (message) => {
162 if (failRemoval) {201 if (failRemoval) {
163 throw new Error("remove failed");202 throw removalCause;
164 }203 }
165 stored.delete(message.messageId);204 stored.delete(message.messageId);
166 },205 },
@@ -176,7 +215,15 @@ describe("thread outbox", () => { · 9 unmodified lines
176 "environment-1:thread-1": [message],215 "environment-1:thread-1": [message],
177 });216 });
178 217
179 await expect(manager.remove(message)).rejects.toThrow("remove failed");218 await expect(manager.remove(message)).rejects.toEqual(
219 new ThreadOutboxManagerError({
220 operation: "remove",
221 environmentId: message.environmentId,
222 threadId: message.threadId,
223 messageId: message.messageId,
224 cause: removalCause,
225 }),
226 );
180 expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({227 expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({
181 "environment-1:thread-1": [message],228 "environment-1:thread-1": [message],
182 });229 });