| ... | ... | @@ -17,6 +17,7 @@ import { AsyncResult } from "effect/unstable/reactivity"; |
| 17 | 17 | import { useCallback, useEffect, useRef, useState } from "react"; |
| 18 | 18 | import { Alert } from "react-native"; |
| 19 | 19 | |
| 20 | import { createDebugLogger } from "../lib/debugLog"; |
| 20 | 21 | import { scopedThreadKey } from "../lib/scopedEntities"; |
| 21 | 22 | import { buildProjectThreadStartTurnInput } from "../lib/projectThreadStartTurn"; |
| 22 | 23 | import { serializeComposerMessageForServer, uploadedComposerContext } from "../lib/composerContext"; |
| ... | ... | @@ -55,6 +56,7 @@ import { |
| 55 | 56 | type QueuedThreadCreation, |
| 56 | 57 | type QueuedThreadMessage, |
| 57 | 58 | type ThreadOutboxCommandStage, |
| 59 | type ThreadOutboxFailureAction, |
| 58 | 60 | } from "./thread-outbox-model"; |
| 59 | 61 | import { environmentThreadShells, threadEnvironment } from "./threads"; |
| 60 | 62 | import { |
| ... | ... | @@ -82,6 +84,92 @@ import { |
| 82 | 84 | useRemoteConnectionStatus, |
| 83 | 85 | } from "./use-remote-environment-registry"; |
| 84 | 86 | |
| 87 | // Ordinary offline behavior (a socket dropping mid-request, a retryable |
| 88 | // attachment upload failure) must not spam `console.warn` on every backoff |
| 89 | // retry; it goes to the filterable `[t3-thread-outbox]` debug log instead. |
| 90 | // Failures the server decided stay on `console.warn`. |
| 91 | const threadOutboxDebug = createDebugLogger("thread-outbox"); |
| 92 | |
| 93 | /** |
| 94 | * On the queued-request path (settings sync, startTurn) the RPC client |
| 95 | * reports ordinary transport drops as the raw socket/worker reason tags, and |
| 96 | * reserves `RpcClientDefect` for client-side protocol violations and decoding |
| 97 | * failures — unlike the shared config-subscription stream, which |
| 98 | * deliberately re-wraps transport causes under that tag. Defects still retry, |
| 99 | * but they are not ordinary offline behavior and must not hide behind the |
| 100 | * offline debug log. |
| 101 | */ |
| 102 | function isRpcClientDecodeDefect(error: unknown): boolean { |
| 103 | if ( |
| 104 | typeof error !== "object" || |
| 105 | error === null || |
| 106 | !("_tag" in error) || |
| 107 | error._tag !== "RpcClientError" |
| 108 | ) { |
| 109 | return false; |
| 110 | } |
| 111 | const reason: unknown = (error as { readonly reason?: unknown }).reason; |
| 112 | return ( |
| 113 | typeof reason === "object" && |
| 114 | reason !== null && |
| 115 | "_tag" in reason && |
| 116 | reason._tag === "RpcClientDefect" |
| 117 | ); |
| 118 | } |
| 119 | |
| 120 | function isOrdinaryThreadOutboxTransportFailure(error: unknown): boolean { |
| 121 | return shouldRetryThreadOutboxDelivery(error) && !isRpcClientDecodeDefect(error); |
| 122 | } |
| 123 | |
| 124 | /** |
| 125 | * Logs one queued-message delivery failure and returns the retry-or-restore |
| 126 | * decision for the caller. Ordinary transport retries — what an offline |
| 127 | * device or a flapping socket produces on every backoff attempt — go to the |
| 128 | * debug log. Server-decided failures warn. Settings-sync failures always |
| 129 | * resolve to a retry even when the server rejected the command, so the |
| 130 | * error, not the resolved action, must decide the log level there; routing |
| 131 | * every retry to debug could hide a permanently rejected update forever. |
| 132 | */ |
| 133 | function logThreadOutboxDeliveryFailure(input: { |
| 134 | readonly stage: ThreadOutboxCommandStage; |
| 135 | readonly error: unknown; |
| 136 | readonly interrupted: boolean; |
| 137 | readonly context: Record<string, unknown>; |
| 138 | }): ThreadOutboxFailureAction { |
| 139 | const action = resolveThreadOutboxFailureAction({ |
| 140 | stage: input.stage, |
| 141 | error: input.error, |
| 142 | interrupted: input.interrupted, |
| 143 | }); |
| 144 | const details = { ...input.context, stage: input.stage, action }; |
| 145 | const ordinaryTransportRetry = |
| 146 | action === "retry" && |
| 147 | !isRpcClientDecodeDefect(input.error) && |
| 148 | (input.interrupted || |
| 149 | input.stage !== "settings-sync" || |
| 150 | shouldRetryThreadOutboxDelivery(input.error)); |
| 151 | if (ordinaryTransportRetry) { |
| 152 | threadOutboxDebug.log("queued message delivery failed", details); |
| 153 | } else { |
| 154 | console.warn("[thread-outbox] queued message delivery failed", details); |
| 155 | } |
| 156 | return action; |
| 157 | } |
| 158 | |
| 159 | /** Attachment uploads retry like delivery: transport failures are ordinary offline noise. */ |
| 160 | function logThreadOutboxUploadFailure(queuedMessage: QueuedThreadMessage, error: unknown): void { |
| 161 | const context = { |
| 162 | environmentId: queuedMessage.environmentId, |
| 163 | threadId: queuedMessage.threadId, |
| 164 | messageId: queuedMessage.messageId, |
| 165 | }; |
| 166 | if (isOrdinaryThreadOutboxTransportFailure(error)) { |
| 167 | threadOutboxDebug.log("attachment upload failed; retrying", { ...context, error }); |
| 168 | } else { |
| 169 | console.warn("[thread-outbox] failed to upload attachments", { ...context, error }); |
| 170 | } |
| 171 | } |
| 172 | |
| 85 | 173 | function beginDispatchingQueuedMessage(queuedMessageId: MessageId): void { |
| 86 | 174 | appAtomRegistry.set(dispatchingQueuedMessageIdAtom, queuedMessageId); |
| 87 | 175 | } |
| ... | ... | @@ -215,14 +303,13 @@ export async function completeQueuedMessageDelivery( |
| 215 | 303 | ); |
| 216 | 304 | if (!removed) { |
| 217 | 305 | forgetAcknowledgedThreadMessage(queuedMessage); |
| 218 | | console.warn( |
| 219 | | "[thread-outbox] delivered message was edited before cleanup; keeping the newer message", |
| 220 | | { |
| 221 | | environmentId: queuedMessage.environmentId, |
| 222 | | threadId: queuedMessage.threadId, |
| 223 | | messageId: queuedMessage.messageId, |
| 224 | | }, |
| 225 | | ); |
| 306 | // Losing the cleanup race to a user edit is an expected outcome the |
| 307 | // caller handles by keeping the newer message; it is not a warning. |
| 308 | threadOutboxDebug.log("delivered message was edited before cleanup", { |
| 309 | environmentId: queuedMessage.environmentId, |
| 310 | threadId: queuedMessage.threadId, |
| 311 | messageId: queuedMessage.messageId, |
| 312 | }); |
| 226 | 313 | return "edited"; |
| 227 | 314 | } |
| 228 | 315 | return "removed"; |
| ... | ... | @@ -667,18 +754,16 @@ export function useThreadOutboxDrain(): void { |
| 667 | 754 | return null; |
| 668 | 755 | } |
| 669 | 756 | const error = Cause.squash(commandResult.cause); |
| 670 | | const action = resolveThreadOutboxFailureAction({ |
| 757 | const action = logThreadOutboxDeliveryFailure({ |
| 671 | 758 | stage, |
| 672 | 759 | error, |
| 673 | 760 | interrupted: Cause.hasInterruptsOnly(commandResult.cause), |
| 674 | | }); |
| 675 | | console.warn("[thread-outbox] queued message delivery failed", { |
| 676 | | environmentId: queuedMessage.environmentId, |
| 677 | | threadId: queuedMessage.threadId, |
| 678 | | messageId: queuedMessage.messageId, |
| 679 | | stage, |
| 680 | | cause: commandResult.cause, |
| 681 | | action, |
| 761 | context: { |
| 762 | environmentId: queuedMessage.environmentId, |
| 763 | threadId: queuedMessage.threadId, |
| 764 | messageId: queuedMessage.messageId, |
| 765 | cause: commandResult.cause, |
| 766 | }, |
| 682 | 767 | }); |
| 683 | 768 | return { |
| 684 | 769 | action, |
| ... | ... | @@ -772,7 +857,7 @@ export function useThreadOutboxDrain(): void { |
| 772 | 857 | return true; |
| 773 | 858 | } |
| 774 | 859 | } catch (error) { |
| 775 | | console.warn("[thread-outbox] failed to upload attachments", error); |
| 860 | logThreadOutboxUploadFailure(queuedMessage, error); |
| 776 | 861 | if (!shouldRetryThreadOutboxDelivery(error)) { |
| 777 | 862 | return restoreQueuedMessage( |
| 778 | 863 | queuedMessage, |
| ... | ... | @@ -900,7 +985,7 @@ export function useThreadOutboxDrain(): void { |
| 900 | 985 | return true; |
| 901 | 986 | } |
| 902 | 987 | } catch (error) { |
| 903 | | console.warn("[thread-outbox] failed to upload attachments", error); |
| 988 | logThreadOutboxUploadFailure(queuedMessage, error); |
| 904 | 989 | if (!shouldRetryThreadOutboxDelivery(error)) { |
| 905 | 990 | return restoreQueuedMessage( |
| 906 | 991 | queuedMessage, |