fix(reef): stop leaking the inbox loop across channel crash restarts (#110870)

steipete · web-flow · commit e80b0b5aa503 · 2026-07-18T19:59:58.000+01:00
* fix(reef): stop leaking the inbox loop across channel crash restarts

* fix(reef): move channel lifecycle helper to its own module for lint/deadcode gates
diff --git a/extensions/reef/src/channel-lifecycle.ts b/extensions/reef/src/channel-lifecycle.ts
@@ -0,0 +1,50 @@
+import { abortableSleep } from "./transport.js";
+
+const REEF_RECONCILE_INTERVAL_MS = 30_000;
+
+// 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
+// loop keeps reconnecting as this handle, fights the replacement instance for
+// the single relay inbox socket, and drives the relay into rate limiting.
+export async function runReefChannelLifecycle(params: {
+  parentSignal: AbortSignal;
+  startInbox: (signal: AbortSignal) => Promise<void>;
+  reconcile: () => Promise<void>;
+  onReconcileError: (error: unknown) => void;
+  reconcileIntervalMs?: number;
+}): Promise<void> {
+  const lifecycle = new AbortController();
+  const onParentAbort = () => lifecycle.abort();
+  params.parentSignal.addEventListener("abort", onParentAbort, { once: true });
+  if (params.parentSignal.aborted) {
+    // Listeners added after an abort never fire; inherit the abort directly.
+    lifecycle.abort();
+  }
+  const intervalMs = params.reconcileIntervalMs ?? REEF_RECONCILE_INTERVAL_MS;
+  const reconciliationLoop = async () => {
+    while (!lifecycle.signal.aborted) {
+      await abortableSleep(intervalMs, lifecycle.signal);
+      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);
+      }
+    }
+  };
+  const inboxTask = params.startInbox(lifecycle.signal);
+  try {
+    await Promise.all([inboxTask, reconciliationLoop()]);
+  } finally {
+    lifecycle.abort();
+    params.parentSignal.removeEventListener("abort", onParentAbort);
+    // startInbox resolves (never rejects) once its socket work is quiescent,
+    // so no reconnect loop can outlive this account instance.
+    await inboxTask;
+  }
+}
diff --git a/extensions/reef/src/channel.test.ts b/extensions/reef/src/channel.test.ts
@@ -8,8 +8,9 @@ import {
 } from "openclaw/plugin-sdk/plugin-state-test-runtime";
 import { createPluginRuntimeMock } from "openclaw/plugin-sdk/plugin-test-runtime";
 import { defaultRuntime } from "openclaw/plugin-sdk/runtime";
-import { afterEach, beforeEach, describe, expect, it } from "vitest";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import { generateIdentity } from "../protocol/index.js";
+import { runReefChannelLifecycle } from "./channel-lifecycle.js";
 import { reefPlugin } from "./channel.js";
 import { resolveReefConfig } from "./config-schema.js";
 import { resolveReefInboundDispatchContent } from "./inbound.js";
@@ -103,3 +104,96 @@ describe("Reef conversation directory", () => {
     ).resolves.toEqual([{ kind: "user", id: "molty", name: "@molty's agent", handle: "@molty" }]);
   });
 });
+
+describe("Reef channel lifecycle", () => {
+  function hangingInbox() {
+    const seen: AbortSignal[] = [];
+    let settled = false;
+    const startInbox = (signal: AbortSignal) => {
+      seen.push(signal);
+      return new Promise<void>((resolve) => {
+        const done = () => {
+          settled = true;
+          resolve();
+        };
+        if (signal.aborted) {
+          done();
+          return;
+        }
+        signal.addEventListener("abort", done, { once: true });
+      });
+    };
+    return { startInbox, seen, isSettled: () => settled };
+  }
+
+  it("keeps running when a periodic reconcile fails", async () => {
+    const parent = new AbortController();
+    const inbox = hangingInbox();
+    const errors: unknown[] = [];
+    let reconciles = 0;
+    const lifecycle = runReefChannelLifecycle({
+      parentSignal: parent.signal,
+      startInbox: inbox.startInbox,
+      reconcile: async () => {
+        reconciles += 1;
+        throw new Error("rate_limited");
+      },
+      onReconcileError: (error) => errors.push(error),
+      reconcileIntervalMs: 5,
+    });
+    await vi.waitFor(() => {
+      expect(reconciles).toBeGreaterThanOrEqual(2);
+    });
+    expect(errors.length).toBeGreaterThanOrEqual(2);
+    expect(inbox.isSettled()).toBe(false);
+    parent.abort();
+    await lifecycle;
+    expect(inbox.isSettled()).toBe(true);
+  });
+
+  it("tears down the inbox loop before settling when a loop branch throws", async () => {
+    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).
+    const lifecycle = runReefChannelLifecycle({
+      parentSignal: parent.signal,
+      startInbox: inbox.startInbox,
+      reconcile: async () => {
+        throw new Error("boom");
+      },
+      onReconcileError: () => {
+        throw new Error("fatal");
+      },
+      reconcileIntervalMs: 5,
+    });
+    await expect(lifecycle).rejects.toThrow("fatal");
+    // The rejection must not leave the inbox reconnect loop running: its
+    // signal is aborted and its promise has settled before the caller resumes.
+    expect(inbox.seen[0]?.aborted).toBe(true);
+    expect(inbox.isSettled()).toBe(true);
+  });
+});
+
+describe("Reef channel lifecycle abort inheritance", () => {
+  it("settles immediately when the parent signal is already aborted", async () => {
+    const parent = new AbortController();
+    parent.abort();
+    const seen: AbortSignal[] = [];
+    await runReefChannelLifecycle({
+      parentSignal: parent.signal,
+      startInbox: (signal) => {
+        seen.push(signal);
+        return signal.aborted
+          ? Promise.resolve()
+          : new Promise<void>((resolve) => {
+              signal.addEventListener("abort", () => resolve(), { once: true });
+            });
+      },
+      reconcile: async () => {},
+      onReconcileError: () => {},
+      reconcileIntervalMs: 5,
+    });
+    expect(seen[0]?.aborted).toBe(true);
+  });
+});
diff --git a/extensions/reef/src/channel.ts b/extensions/reef/src/channel.ts
@@ -10,6 +10,7 @@ import {
   type ChannelPlugin,
 } from "openclaw/plugin-sdk/core";
 import { createChannelDirectoryAdapter } from "openclaw/plugin-sdk/directory-runtime";
+import { runReefChannelLifecycle } from "./channel-lifecycle.js";
 import {
   ReefChannelConfigSchema,
   autonomyBudget,
@@ -31,12 +32,7 @@ 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,
-  abortableSleep,
-  createReefWebSocket,
-} from "./transport.js";
+import { ReefInboxConnection, ReefTransportClient, createReefWebSocket } from "./transport.js";
 import { isReefPairingApprovalToken, openReefTrustStore } from "./trust-store.js";
 import type { ReefAccount, ReefIngressMessage } from "./types.js";
 
@@ -454,16 +450,14 @@ export const reefPlugin: ChannelPlugin<ReefAccount> = {
           },
         },
       );
-      const reconciliationLoop = async () => {
-        while (!ctx.abortSignal.aborted) {
-          await abortableSleep(30_000, ctx.abortSignal);
-          if (!ctx.abortSignal.aborted) {
-            await reconcile();
-          }
-        }
-      };
       try {
-        await Promise.all([inbox.start(ctx.abortSignal), reconciliationLoop()]);
+        await runReefChannelLifecycle({
+          parentSignal: ctx.abortSignal,
+          startInbox: (signal) => inbox.start(signal),
+          reconcile,
+          onReconcileError: (error) =>
+            ctx.log?.error?.(`reef friend reconcile failed: ${String(error)}`),
+        });
       } finally {
         ctx.setStatus({ accountId: "default", running: false, connected: false });
       }
diff --git a/src/gateway/server-channels.test.ts b/src/gateway/server-channels.test.ts
@@ -309,6 +309,28 @@ describe("server-channels auto restart", () => {
     expect(startAccount).toHaveBeenCalledTimes(11);
   });
 
+  it("aborts the crashed task's signal before starting its replacement", async () => {
+    const signals: AbortSignal[] = [];
+    const startAccount = vi.fn(async (ctx: ChannelGatewayContext<TestAccount>) => {
+      signals.push(ctx.abortSignal);
+      throw new Error("crash");
+    });
+    installTestRegistry(createTestPlugin({ startAccount }));
+    const manager = createManager();
+
+    await manager.startChannels();
+    await advanceTimersUntil(
+      () => startAccount.mock.calls.length >= 2,
+      "expected a crash-loop restart",
+      { stepMs: 10, maxMs: 500 },
+    );
+
+    // A crashed startAccount can leave background work racing on its signal
+    // (e.g. a reconnect loop). The replacement must never overlap that lifetime.
+    expect(signals[0]?.aborted).toBe(true);
+    expect(signals[1]?.aborted).toBe(false);
+  });
+
   it.each(["resolve", "reject"] as const)(
     "resets the restart counter after a stable run that ends with %s",
     async (outcome) => {
diff --git a/src/gateway/server-channels.ts b/src/gateway/server-channels.ts
@@ -789,6 +789,10 @@ export function createChannelManager(opts: ChannelManagerOptions): ChannelManage
                 if (store.aborts.get(id) === abort) {
                   store.aborts.delete(id);
                 }
+                // The settled task may have left background work racing on this
+                // signal. Abort before the replacement starts so two account
+                // instances can never share a live lifetime.
+                abort.abort();
                 try {
                   await startChannelInternal(channelId, id, {
                     preserveManualStop: true,
@@ -838,6 +842,9 @@ export function createChannelManager(opts: ChannelManagerOptions): ChannelManage
                 if (store.aborts.get(id) === abort) {
                   store.aborts.delete(id);
                 }
+                // See the timed-out-stop restart above: never start the crash
+                // replacement while the predecessor's signal is unaborted.
+                abort.abort();
                 await startChannelInternal(channelId, id, {
                   preserveRestartAttempts: true,
                   preserveManualStop: true,
@@ -853,6 +860,9 @@ export function createChannelManager(opts: ChannelManagerOptions): ChannelManage
               if (store.aborts.get(id) === abort) {
                 store.aborts.delete(id);
               }
+              // Terminal paths (give-up, terminal disconnect, manual stop) end
+              // here without a restart; leave no unaborted lifetime behind.
+              abort.abort();
             });
           function isCurrentTask() {
             return store.tasks.get(id) === trackedPromise;
