From ee37c840f3deb2fef30227374663ee5368cde802 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yi=C4=9Fit=20ERDO=C4=9EAN?= Date: Sun, 19 Jul 2026 01:18:20 +0300 Subject: [PATCH] fix(reef): contain startup friend reconcile failures (#110918) * fix(reef): contain startup friend reconcile failures The periodic reconcile treats a transient relay failure as non-fatal, but the startup reconcile was a bare await outside the lifecycle. A relay 429 there rejects startAccount, and the supervisor restart that follows is itself what escalates relay rate limiting -- so Reef gives up permanently after MAX_RESTARTS while the relay is still throttling. Move the startup reconcile into runReefChannelLifecycle so both paths share one failure policy, and activate through an onReady hook to preserve the existing ordering: refresh peer keys, activate, then start the inbox. * fix(reef): harden reconcile recovery Co-authored-by: Yigtwxx * docs(changelog): credit Reef reconcile fix * test(reef): satisfy exact-head validation * test(reef): satisfy lint contract --------- Co-authored-by: Peter Steinberger --- CHANGELOG.md | 1 + extensions/reef/src/channel-lifecycle.ts | 60 ++++++-- extensions/reef/src/channel.test.ts | 183 ++++++++++++++++++++++- extensions/reef/src/channel.ts | 26 +++- extensions/reef/src/transport.test.ts | 57 +++++++ extensions/reef/src/transport.ts | 59 ++++++-- 6 files changed, 357 insertions(+), 29 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e310125df27b..d12b184746c9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -42,6 +42,7 @@ Docs: https://docs.openclaw.ai ### Fixes +- **Reef startup reconciliation:** contain retryable relay failures during startup without supervisor restart loops, while preserving definitive-error and cancellation handling. Thanks @Yigtwxx. - **Codex stale-session replies:** stop model fallback after another gateway supersedes a Codex session generation and deliver a safe retry notice instead of abandoning the message silently. - **Bounded input and provider responses:** cap pasted auth/config input and enforce wall-clock deadlines across generated-media downloads, polling JSON, and failed response details so oversized or slow-drip streams cannot exceed resource budgets (thanks @Pick-cat). - **Cloud worker derived workspace caches:** exclude Python caches, dependency trees, and macOS metadata symmetrically from outbound sync and inbound reconciliation so local cache rewrites cannot fence later cloud results or worker reclaim. diff --git a/extensions/reef/src/channel-lifecycle.ts b/extensions/reef/src/channel-lifecycle.ts index e8d5970fc8e2..829151b496a7 100644 --- a/extensions/reef/src/channel-lifecycle.ts +++ b/extensions/reef/src/channel-lifecycle.ts @@ -1,6 +1,27 @@ import { abortableSleep } from "./transport.js"; const REEF_RECONCILE_INTERVAL_MS = 30_000; +const CONTINUE_AFTER_RECONCILE_ERROR = () => true; +const STOP_AFTER_RECONCILE_ERROR = () => false; + +async function runReconcileStep(params: { + reconcile: () => Promise; + onReconcileError: (error: unknown) => void; + shouldContinueAfterError: (error: unknown) => boolean; + signal: AbortSignal; +}): Promise { + try { + await params.reconcile(); + } catch (error) { + if (params.signal.aborted) { + return; + } + if (!params.shouldContinueAfterError(error)) { + throw error; + } + params.onReconcileError(error); + } +} // One abort scope owns both account loops. If either branch throws, the inbox // loop must be torn down and awaited before startAccount settles: a leaked @@ -11,6 +32,12 @@ export async function runReefChannelLifecycle(params: { startInbox: (signal: AbortSignal) => Promise; reconcile: () => Promise; onReconcileError: (error: unknown) => void; + // Startup may continue only for errors the channel classifies as retryable. + // Periodic reconcile remains best-effort once the account is already active. + shouldContinueAfterStartupReconcileError?: (error: unknown) => boolean; + // Runs after the startup reconcile either refreshes peer keys or reports a + // classified retryable failure, before the inbox can dispatch a turn. + onReady?: () => Promise; reconcileIntervalMs?: number; }): Promise { const lifecycle = new AbortController(); @@ -27,18 +54,33 @@ export async function runReefChannelLifecycle(params: { if (lifecycle.signal.aborted) { return; } - try { - await params.reconcile(); - } catch (error) { - // Transient relay failures (429, network) must not crash the channel: - // the crash-restart cycle re-registers the inbox connection and is - // itself what escalates relay rate limiting. - params.onReconcileError(error); - } + await runReconcileStep({ + ...params, + shouldContinueAfterError: CONTINUE_AFTER_RECONCILE_ERROR, + signal: lifecycle.signal, + }); } }; - const inboxTask = params.startInbox(lifecycle.signal); + // Declared outside the try so the finally can await it even when the startup + // steps below throw before the inbox is started. + let inboxTask: Promise | undefined; try { + if (!lifecycle.signal.aborted) { + await runReconcileStep({ + ...params, + shouldContinueAfterError: + params.shouldContinueAfterStartupReconcileError ?? STOP_AFTER_RECONCILE_ERROR, + signal: lifecycle.signal, + }); + } + if (lifecycle.signal.aborted) { + return; + } + await params.onReady?.(); + if (lifecycle.signal.aborted) { + return; + } + inboxTask = params.startInbox(lifecycle.signal); await Promise.all([inboxTask, reconciliationLoop()]); } finally { lifecycle.abort(); diff --git a/extensions/reef/src/channel.test.ts b/extensions/reef/src/channel.test.ts index 16b00ac2e1e2..db3c694077bf 100644 --- a/extensions/reef/src/channel.test.ts +++ b/extensions/reef/src/channel.test.ts @@ -17,6 +17,16 @@ import { resolveReefInboundDispatchContent } from "./inbound.js"; import { setReefRuntime } from "./runtime.js"; import { openReefTrustStore } from "./trust-store.js"; +function deferred() { + let resolve!: () => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + return { promise, resolve, reject }; +} + describe("Reef inbound dispatch content", () => { it("keeps provenance model-visible without storing it in the transcript body", () => { const content = resolveReefInboundDispatchContent({ @@ -126,6 +136,162 @@ describe("Reef channel lifecycle", () => { return { startInbox, seen, isSettled: () => settled }; } + it("activates and starts the inbox when the startup reconcile fails", async () => { + const parent = new AbortController(); + const inbox = hangingInbox(); + const errors: unknown[] = []; + let reconciles = 0; + // Captured inside onReady so the assertion pins the startup reconcile + // specifically, not "some reconcile eventually failed" once the periodic + // loop has had a chance to run. + let reconcilesAtActivation = -1; + let errorsAtActivation = -1; + const lifecycle = runReefChannelLifecycle({ + parentSignal: parent.signal, + startInbox: inbox.startInbox, + reconcile: async () => { + reconciles += 1; + throw new Error("rate_limited"); + }, + onReconcileError: (error) => errors.push(error), + shouldContinueAfterStartupReconcileError: () => true, + onReady: async () => { + reconcilesAtActivation = reconciles; + errorsAtActivation = errors.length; + }, + reconcileIntervalMs: 5, + }); + await vi.waitFor(() => { + expect(reconcilesAtActivation).toBe(1); + }); + // A relay 429 at startup must not escape startAccount: the supervisor would + // restart the account, and that restart cycle is what escalates the rate + // limiting in the first place. + expect(errorsAtActivation).toBe(1); + expect(inbox.seen).toHaveLength(1); + expect(inbox.isSettled()).toBe(false); + parent.abort(); + await lifecycle; + expect(inbox.isSettled()).toBe(true); + }); + + it("refreshes peer keys before activating and before the inbox starts", async () => { + const parent = new AbortController(); + const inbox = hangingInbox(); + const order: string[] = []; + const lifecycle = runReefChannelLifecycle({ + parentSignal: parent.signal, + startInbox: (signal) => { + order.push("inbox"); + return inbox.startInbox(signal); + }, + reconcile: async () => { + order.push("reconcile"); + }, + onReconcileError: () => {}, + onReady: async () => { + order.push("ready"); + }, + reconcileIntervalMs: 5_000, + }); + await vi.waitFor(() => { + expect(order).toEqual(["reconcile", "ready", "inbox"]); + }); + parent.abort(); + await lifecycle; + }); + + it("rejects startup when the reconcile error is not retryable", async () => { + const parent = new AbortController(); + const inbox = hangingInbox(); + const onReady = vi.fn(async () => {}); + const error = new Error("approval store unavailable"); + await expect( + runReefChannelLifecycle({ + parentSignal: parent.signal, + startInbox: inbox.startInbox, + reconcile: async () => { + throw error; + }, + onReconcileError: () => {}, + shouldContinueAfterStartupReconcileError: () => false, + onReady, + }), + ).rejects.toBe(error); + expect(onReady).not.toHaveBeenCalled(); + expect(inbox.seen).toHaveLength(0); + }); + + it("does not activate when the parent aborts during startup reconcile", async () => { + const parent = new AbortController(); + const inbox = hangingInbox(); + const reconcileStarted = deferred(); + const finishReconcile = deferred(); + const onReady = vi.fn(async () => {}); + const lifecycle = runReefChannelLifecycle({ + parentSignal: parent.signal, + startInbox: inbox.startInbox, + reconcile: async () => { + reconcileStarted.resolve(); + await finishReconcile.promise; + }, + onReconcileError: () => {}, + onReady, + }); + await reconcileStarted.promise; + parent.abort(); + finishReconcile.resolve(); + await lifecycle; + expect(onReady).not.toHaveBeenCalled(); + expect(inbox.seen).toHaveLength(0); + }); + + it("does not reject when startup reconcile fails after the parent aborts", async () => { + const parent = new AbortController(); + const inbox = hangingInbox(); + const reconcileStarted = deferred(); + const finishReconcile = deferred(); + const onReady = vi.fn(async () => {}); + const lifecycle = runReefChannelLifecycle({ + parentSignal: parent.signal, + startInbox: inbox.startInbox, + reconcile: async () => { + reconcileStarted.resolve(); + await finishReconcile.promise; + }, + onReconcileError: () => {}, + onReady, + }); + await reconcileStarted.promise; + parent.abort(); + finishReconcile.reject(new DOMException("aborted", "AbortError")); + await expect(lifecycle).resolves.toBeUndefined(); + expect(onReady).not.toHaveBeenCalled(); + expect(inbox.seen).toHaveLength(0); + }); + + it("does not start the inbox when the parent aborts during activation", async () => { + const parent = new AbortController(); + const inbox = hangingInbox(); + const activationStarted = deferred(); + const finishActivation = deferred(); + const lifecycle = runReefChannelLifecycle({ + parentSignal: parent.signal, + startInbox: inbox.startInbox, + reconcile: async () => {}, + onReconcileError: () => {}, + onReady: async () => { + activationStarted.resolve(); + await finishActivation.promise; + }, + }); + await activationStarted.promise; + parent.abort(); + finishActivation.resolve(); + await lifecycle; + expect(inbox.seen).toHaveLength(0); + }); + it("keeps running when a periodic reconcile fails", async () => { const parent = new AbortController(); const inbox = hangingInbox(); @@ -136,13 +302,15 @@ describe("Reef channel lifecycle", () => { startInbox: inbox.startInbox, reconcile: async () => { reconciles += 1; - throw new Error("rate_limited"); + if (reconciles > 1) { + throw new Error("rate_limited"); + } }, onReconcileError: (error) => errors.push(error), reconcileIntervalMs: 5, }); await vi.waitFor(() => { - expect(reconciles).toBeGreaterThanOrEqual(2); + expect(reconciles).toBeGreaterThanOrEqual(3); }); expect(errors.length).toBeGreaterThanOrEqual(2); expect(inbox.isSettled()).toBe(false); @@ -155,11 +323,18 @@ describe("Reef channel lifecycle", () => { const parent = new AbortController(); const inbox = hangingInbox(); // Simulate a non-transport crash escaping the lifecycle (reconcile errors - // are contained, so throw from the error hook itself). + // are contained, so throw from the error hook itself). The startup + // reconcile succeeds so the failure lands on the periodic loop, with the + // inbox already running and therefore able to leak. + let reconciles = 0; const lifecycle = runReefChannelLifecycle({ parentSignal: parent.signal, startInbox: inbox.startInbox, reconcile: async () => { + reconciles += 1; + if (reconciles === 1) { + return; + } throw new Error("boom"); }, onReconcileError: () => { @@ -194,6 +369,6 @@ describe("Reef channel lifecycle abort inheritance", () => { onReconcileError: () => {}, reconcileIntervalMs: 5, }); - expect(seen[0]?.aborted).toBe(true); + expect(seen).toHaveLength(0); }); }); diff --git a/extensions/reef/src/channel.ts b/extensions/reef/src/channel.ts index 9ea923988f29..2b3bfbc0dfb5 100644 --- a/extensions/reef/src/channel.ts +++ b/extensions/reef/src/channel.ts @@ -33,7 +33,12 @@ import { isRephrasedReefResend } from "./rejection-resend.js"; import { getActiveReef, getOptionalReefRuntime, getReefRuntime, setActiveReef } from "./runtime.js"; import { reefSetupAdapter, reefSetupWizard } from "./setup.js"; import { assertReefIdentityBinding, loadKeys, openStores, ReefInboxCursorStore } from "./state.js"; -import { ReefInboxConnection, ReefTransportClient, createReefWebSocket } from "./transport.js"; +import { + ReefInboxConnection, + ReefTransportClient, + createReefWebSocket, + isRetryableReefRelayFailure, +} from "./transport.js"; import { isReefPairingApprovalToken, openReefTrustStore } from "./trust-store.js"; import type { ReefAccount, ReefIngressMessage } from "./types.js"; @@ -401,12 +406,17 @@ export const reefPlugin: ChannelPlugin = { }); }); }; - // Refresh peer keys before recovery can dispatch an agent turn. Activate - // only after reconciliation, but before that turn can use Reef outbound. - await reconcile(); - setActiveReef({ flow, friends, reviews }); - await receiptNotifier.notifyRejections(trust.pendingOutboundRejections()); - ctx.setStatus({ accountId: "default", running: true, connected: false }); + // Attempt the peer-key refresh before recovery can dispatch an agent + // turn. The lifecycle activates only after that attempt is classified. + // The lifecycle owns both the ordering and the reconcile failure policy. + const activate = async () => { + await receiptNotifier.notifyRejections(trust.pendingOutboundRejections()); + if (ctx.abortSignal.aborted) { + return; + } + setActiveReef({ flow, friends, reviews }); + ctx.setStatus({ accountId: "default", running: true, connected: false }); + }; const inbox = new ReefInboxConnection( transport, (entries) => @@ -472,6 +482,8 @@ export const reefPlugin: ChannelPlugin = { }, onReconcileError: (error) => ctx.log?.error?.(`reef friend reconcile failed: ${String(error)}`), + shouldContinueAfterStartupReconcileError: isRetryableReefRelayFailure, + onReady: activate, }); } finally { ctx.setStatus({ accountId: "default", running: false, connected: false }); diff --git a/extensions/reef/src/transport.test.ts b/extensions/reef/src/transport.test.ts index bab989979f91..d953db3bf1af 100644 --- a/extensions/reef/src/transport.test.ts +++ b/extensions/reef/src/transport.test.ts @@ -10,6 +10,7 @@ import { ReefRelayError, ReefTransportClient, createReefWebSocket, + isRetryableReefRelayFailure, type WebSocketLike, } from "./transport.js"; import type { InboxEntry, ReefKeys, RelayFriend } from "./types.js"; @@ -34,6 +35,62 @@ afterEach(() => { vi.useRealTimers(); }); +describe("isRetryableReefRelayFailure", () => { + it("accepts transient relay responses and timeouts", () => { + expect(isRetryableReefRelayFailure(new ReefRelayError(408, "timeout"))).toBe(true); + expect(isRetryableReefRelayFailure(new ReefRelayError(429, "rate_limited"))).toBe(true); + expect(isRetryableReefRelayFailure(new ReefRelayError(503, "unavailable"))).toBe(true); + expect( + isRetryableReefRelayFailure(Object.assign(new Error("timed out"), { name: "TimeoutError" })), + ).toBe(true); + }); + + it("rejects definitive relay and local failures", () => { + expect(isRetryableReefRelayFailure(new ReefRelayError(401, "unauthorized"))).toBe(false); + expect(isRetryableReefRelayFailure(new Error("approval store unavailable"))).toBe(false); + }); +}); + +describe("ReefTransportClient network failures", () => { + it("normalizes fetch failures without swallowing the cause", async () => { + const cause = new TypeError("fetch failed"); + const client = new ReefTransportClient("https://relay.example", "alice", keys, async () => { + throw cause; + }); + + const error = await client.listFriends().catch((failure: unknown) => failure); + expect(error).toMatchObject({ + name: "ReefRelayUnavailableError", + message: "fetch failed", + cause, + }); + expect(isRetryableReefRelayFailure(error)).toBe(true); + }); + + it("normalizes connection loss while reading a successful response body", async () => { + const cause = new TypeError("terminated"); + const response = new Response( + new ReadableStream({ + start(controller) { + controller.error(cause); + }, + }), + ); + const client = new ReefTransportClient( + "https://relay.example", + "alice", + keys, + async () => response, + ); + + await expect(client.listFriends()).rejects.toMatchObject({ + name: "ReefRelayUnavailableError", + message: "terminated", + cause, + }); + }); +}); + function verifyRelaySignature( signature: string, input: { method: string; path: string; ts: number; bodySha256: string }, diff --git a/extensions/reef/src/transport.ts b/extensions/reef/src/transport.ts index a397fca065f6..4cc4c9b22abe 100644 --- a/extensions/reef/src/transport.ts +++ b/extensions/reef/src/transport.ts @@ -36,6 +36,13 @@ export class ReefRelayError extends Error { } } +class ReefRelayUnavailableError extends Error { + constructor(cause: unknown) { + super(cause instanceof Error ? cause.message : String(cause), { cause }); + this.name = "ReefRelayUnavailableError"; + } +} + export function isDefinitiveReefRegistrationFailure(error: unknown): boolean { return ( error instanceof ReefRelayError && @@ -46,10 +53,38 @@ export function isDefinitiveReefRegistrationFailure(error: unknown): boolean { ); } +export function isRetryableReefRelayFailure(error: unknown): boolean { + if (error instanceof ReefRelayError) { + return error.status === 408 || error.status === 429 || error.status >= 500; + } + return ( + error instanceof ReefRelayUnavailableError || + (error instanceof Error && error.name === "TimeoutError") + ); +} + export function isReefOwnershipRejection(error: unknown): boolean { return error instanceof ReefRelayError && error.message === "unknown_handle"; } +async function readReefRelaySuccessJson(response: Response, signal?: AbortSignal): Promise { + try { + return await readProviderJsonResponse(response, "reef.relay", { + maxBytes: REEF_RELAY_JSON_MAX_BYTES, + }); + } catch (error) { + if (signal?.aborted) { + throw signal.reason; + } + // Undici surfaces socket loss during response-body consumption as a + // TypeError even though fetch already resolved with response headers. + if (error instanceof TypeError) { + throw new ReefRelayUnavailableError(error); + } + throw error; + } +} + export class ReefTransportClient { // Ed25519 is deterministic: identical (method, path, ts, body) requests produce // identical signatures, which collide with the relay's replay key. Keep ts @@ -194,12 +229,20 @@ export class ReefTransportClient { url, }); try { - const response = await this.fetcher(url, { - method, - headers: { ...headers, ...(bytes.length ? { "content-type": "application/json" } : {}) }, - ...(bytes.length ? { body: bytes as BodyInit } : {}), - signal: timeout.signal, - }); + let response: Response; + try { + response = await this.fetcher(url, { + method, + headers: { ...headers, ...(bytes.length ? { "content-type": "application/json" } : {}) }, + ...(bytes.length ? { body: bytes as BodyInit } : {}), + signal: timeout.signal, + }); + } catch (error) { + if (timeout.signal?.aborted) { + throw timeout.signal.reason; + } + throw new ReefRelayUnavailableError(error); + } if (!response.ok) { let message = `relay HTTP ${response.status}`; try { @@ -223,9 +266,7 @@ export class ReefTransportClient { if (response.status === 204) { return undefined as T; } - return await readProviderJsonResponse(response, "reef.relay", { - maxBytes: REEF_RELAY_JSON_MAX_BYTES, - }); + return await readReefRelaySuccessJson(response, timeout.signal); } finally { timeout.cleanup(); }