Skip to content

Commit a5de694

Browse files
steipetesteipete
committed
fix(reef): stop leaking the inbox loop across channel crash restarts
1 parent e04ea79 commit a5de694

4 files changed

Lines changed: 183 additions & 11 deletions

File tree

‎extensions/reef/src/channel.test.ts‎

Lines changed: 95 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -8,9 +8,9 @@ import {
88
} from "openclaw/plugin-sdk/plugin-state-test-runtime";
99
import { createPluginRuntimeMock } from "openclaw/plugin-sdk/plugin-test-runtime";
1010
import { defaultRuntime } from "openclaw/plugin-sdk/runtime";
11-
import { afterEach, beforeEach, describe, expect, it } from "vitest";
11+
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
1212
import { generateIdentity } from "../protocol/index.js";
13-
import { reefPlugin } from "./channel.js";
13+
import { reefPlugin, runReefChannelLifecycle } from "./channel.js";
1414
import { resolveReefConfig } from "./config-schema.js";
1515
import { resolveReefInboundDispatchContent } from "./inbound.js";
1616
import { setReefRuntime } from "./runtime.js";
@@ -103,3 +103,96 @@ describe("Reef conversation directory", () => {
103103
).resolves.toEqual([{ kind: "user", id: "molty", name: "@molty's agent", handle: "@molty" }]);
104104
});
105105
});
106+
107+
describe("Reef channel lifecycle", () => {
108+
function hangingInbox() {
109+
const seen: AbortSignal[] = [];
110+
let settled = false;
111+
const startInbox = (signal: AbortSignal) => {
112+
seen.push(signal);
113+
return new Promise<void>((resolve) => {
114+
const done = () => {
115+
settled = true;
116+
resolve();
117+
};
118+
if (signal.aborted) {
119+
done();
120+
return;
121+
}
122+
signal.addEventListener("abort", done, { once: true });
123+
});
124+
};
125+
return { startInbox, seen, isSettled: () => settled };
126+
}
127+
128+
it("keeps running when a periodic reconcile fails", async () => {
129+
const parent = new AbortController();
130+
const inbox = hangingInbox();
131+
const errors: unknown[] = [];
132+
let reconciles = 0;
133+
const lifecycle = runReefChannelLifecycle({
134+
parentSignal: parent.signal,
135+
startInbox: inbox.startInbox,
136+
reconcile: async () => {
137+
reconciles += 1;
138+
throw new Error("rate_limited");
139+
},
140+
onReconcileError: (error) => errors.push(error),
141+
reconcileIntervalMs: 5,
142+
});
143+
await vi.waitFor(() => {
144+
expect(reconciles).toBeGreaterThanOrEqual(2);
145+
});
146+
expect(errors.length).toBeGreaterThanOrEqual(2);
147+
expect(inbox.isSettled()).toBe(false);
148+
parent.abort();
149+
await lifecycle;
150+
expect(inbox.isSettled()).toBe(true);
151+
});
152+
153+
it("tears down the inbox loop before settling when a loop branch throws", async () => {
154+
const parent = new AbortController();
155+
const inbox = hangingInbox();
156+
// Simulate a non-transport crash escaping the lifecycle (reconcile errors
157+
// are contained, so throw from the error hook itself).
158+
const lifecycle = runReefChannelLifecycle({
159+
parentSignal: parent.signal,
160+
startInbox: inbox.startInbox,
161+
reconcile: async () => {
162+
throw new Error("boom");
163+
},
164+
onReconcileError: () => {
165+
throw new Error("fatal");
166+
},
167+
reconcileIntervalMs: 5,
168+
});
169+
await expect(lifecycle).rejects.toThrow("fatal");
170+
// The rejection must not leave the inbox reconnect loop running: its
171+
// signal is aborted and its promise has settled before the caller resumes.
172+
expect(inbox.seen[0]?.aborted).toBe(true);
173+
expect(inbox.isSettled()).toBe(true);
174+
});
175+
});
176+
177+
describe("Reef channel lifecycle abort inheritance", () => {
178+
it("settles immediately when the parent signal is already aborted", async () => {
179+
const parent = new AbortController();
180+
parent.abort();
181+
const seen: AbortSignal[] = [];
182+
await runReefChannelLifecycle({
183+
parentSignal: parent.signal,
184+
startInbox: (signal) => {
185+
seen.push(signal);
186+
return signal.aborted
187+
? Promise.resolve()
188+
: new Promise<void>((resolve) =>
189+
signal.addEventListener("abort", () => resolve(), { once: true }),
190+
);
191+
},
192+
reconcile: async () => {},
193+
onReconcileError: () => {},
194+
reconcileIntervalMs: 5,
195+
});
196+
expect(seen[0]?.aborted).toBe(true);
197+
});
198+
});

‎extensions/reef/src/channel.ts‎

Lines changed: 56 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -82,6 +82,55 @@ function listTrustedPeerDirectoryEntries(params: {
8282
}));
8383
}
8484

85+
const REEF_RECONCILE_INTERVAL_MS = 30_000;
86+
87+
// One abort scope owns both account loops. If either branch throws, the inbox
88+
// loop must be torn down and awaited before startAccount settles: a leaked
89+
// loop keeps reconnecting as this handle, fights the replacement instance for
90+
// the single relay inbox socket, and drives the relay into rate limiting.
91+
export async function runReefChannelLifecycle(params: {
92+
parentSignal: AbortSignal;
93+
startInbox: (signal: AbortSignal) => Promise<void>;
94+
reconcile: () => Promise<void>;
95+
onReconcileError: (error: unknown) => void;
96+
reconcileIntervalMs?: number;
97+
}): Promise<void> {
98+
const lifecycle = new AbortController();
99+
const onParentAbort = () => lifecycle.abort();
100+
params.parentSignal.addEventListener("abort", onParentAbort, { once: true });
101+
if (params.parentSignal.aborted) {
102+
// Listeners added after an abort never fire; inherit the abort directly.
103+
lifecycle.abort();
104+
}
105+
const intervalMs = params.reconcileIntervalMs ?? REEF_RECONCILE_INTERVAL_MS;
106+
const reconciliationLoop = async () => {
107+
while (!lifecycle.signal.aborted) {
108+
await abortableSleep(intervalMs, lifecycle.signal);
109+
if (lifecycle.signal.aborted) {
110+
return;
111+
}
112+
try {
113+
await params.reconcile();
114+
} catch (error) {
115+
// Transient relay failures (429, network) must not crash the channel:
116+
// the crash-restart cycle re-registers the inbox connection and is
117+
// itself what escalates relay rate limiting.
118+
params.onReconcileError(error);
119+
}
120+
}
121+
};
122+
const inboxTask = params.startInbox(lifecycle.signal);
123+
try {
124+
await Promise.all([inboxTask, reconciliationLoop()]);
125+
} finally {
126+
lifecycle.abort();
127+
params.parentSignal.removeEventListener("abort", onParentAbort);
128+
// startInbox resolves (never rejects) once its socket work is quiescent,
129+
// so no reconnect loop can outlive this account instance.
130+
await inboxTask;
131+
}
132+
}
133+
85134
function replyText(payload: unknown): string {
86135
if (!payload || typeof payload !== "object" || !("text" in payload)) {
87136
return "";
@@ -454,16 +503,14 @@ export const reefPlugin: ChannelPlugin<ReefAccount> = {
454503
},
455504
},
456505
);
457-
const reconciliationLoop = async () => {
458-
while (!ctx.abortSignal.aborted) {
459-
await abortableSleep(30_000, ctx.abortSignal);
460-
if (!ctx.abortSignal.aborted) {
461-
await reconcile();
462-
}
463-
}
464-
};
465506
try {
466-
await Promise.all([inbox.start(ctx.abortSignal), reconciliationLoop()]);
507+
await runReefChannelLifecycle({
508+
parentSignal: ctx.abortSignal,
509+
startInbox: (signal) => inbox.start(signal),
510+
reconcile,
511+
onReconcileError: (error) =>
512+
ctx.log?.error?.(`reef friend reconcile failed: ${String(error)}`),
513+
});
467514
} finally {
468515
ctx.setStatus({ accountId: "default", running: false, connected: false });
469516
}

‎src/gateway/server-channels.test.ts‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -309,6 +309,28 @@ describe("server-channels auto restart", () => {
309309
expect(startAccount).toHaveBeenCalledTimes(11);
310310
});
311311

312+
it("aborts the crashed task's signal before starting its replacement", async () => {
313+
const signals: AbortSignal[] = [];
314+
const startAccount = vi.fn(async (ctx: ChannelGatewayContext<TestAccount>) => {
315+
signals.push(ctx.abortSignal);
316+
throw new Error("crash");
317+
});
318+
installTestRegistry(createTestPlugin({ startAccount }));
319+
const manager = createManager();
320+
321+
await manager.startChannels();
322+
await advanceTimersUntil(
323+
() => startAccount.mock.calls.length >= 2,
324+
"expected a crash-loop restart",
325+
{ stepMs: 10, maxMs: 500 },
326+
);
327+
328+
// A crashed startAccount can leave background work racing on its signal
329+
// (e.g. a reconnect loop). The replacement must never overlap that lifetime.
330+
expect(signals[0]?.aborted).toBe(true);
331+
expect(signals[1]?.aborted).toBe(false);
332+
});
333+
312334
it.each(["resolve", "reject"] as const)(
313335
"resets the restart counter after a stable run that ends with %s",
314336
async (outcome) => {

‎src/gateway/server-channels.ts‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -789,6 +789,10 @@ export function createChannelManager(opts: ChannelManagerOptions): ChannelManage
789789
if (store.aborts.get(id) === abort) {
790790
store.aborts.delete(id);
791791
}
792+
// The settled task may have left background work racing on this
793+
// signal. Abort before the replacement starts so two account
794+
// instances can never share a live lifetime.
795+
abort.abort();
792796
try {
793797
await startChannelInternal(channelId, id, {
794798
preserveManualStop: true,
@@ -838,6 +842,9 @@ export function createChannelManager(opts: ChannelManagerOptions): ChannelManage
838842
if (store.aborts.get(id) === abort) {
839843
store.aborts.delete(id);
840844
}
845+
// See the timed-out-stop restart above: never start the crash
846+
// replacement while the predecessor's signal is unaborted.
847+
abort.abort();
841848
await startChannelInternal(channelId, id, {
842849
preserveRestartAttempts: true,
843850
preserveManualStop: true,
@@ -853,6 +860,9 @@ export function createChannelManager(opts: ChannelManagerOptions): ChannelManage
853860
if (store.aborts.get(id) === abort) {
854861
store.aborts.delete(id);
855862
}
863+
// Terminal paths (give-up, terminal disconnect, manual stop) end
864+
// here without a restart; leave no unaborted lifetime behind.
865+
abort.abort();
856866
});
857867
function isCurrentTask() {
858868
return store.tasks.get(id) === trackedPromise;

0 commit comments

Comments
 (0)