Skip to content

Commit 87ee2e8

Browse files
steipeteSunnyShu0925
steipete
and
SunnyShu0925
committed
fix(outbound): retry proven pre-connect failures
Co-authored-by: 0668000539 <[email protected]>
1 parent 6ed2aaf commit 87ee2e8

12 files changed

Lines changed: 713 additions & 13 deletions

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@ Docs: https://docs.openclaw.ai
2828
- **CJK Markdown emphasis:** render adjacent Chinese, Japanese, and Korean emphasis punctuation through the shared Markdown pipeline instead of leaking literal markers across channels. (#101230, #101120) Thanks @nicknmorty.
2929
- **Codex yielded native subagents:** keep the parent app-server subscription and shared client alive until yielded native subagent completion delivery settles, preventing lost wakeups and leaked one-shot cleanup.
3030
- **Delivery recovery pacing:** pace eligible outbound and restart-continuation replays after gateway startup so outage backlogs do not burst into channel rate limits, while preserving the wall-clock recovery budget. (#101118, #101058) Thanks @ZengWen-DT.
31+
- **Outbound pre-connect recovery:** clear stale platform-send evidence atomically when a connect or DNS failure proves no request was sent, allowing queued Discord and other channel messages to replay after connectivity returns without weakening the unknown-send duplicate guard. (#101024, #100979) Thanks @SunnyShu0925.
3132
- **Discord streamed finals:** send completion replies as fresh messages so inactive channels become unread, while preserving targeted mentions without escalating `@everyone` or `@here`. (#99711, #99662) Thanks @davelutztx.
3233
- **OpenAI-compatible SSE parsing:** recognize event streams mislabeled as JSON without prepending a second `data:` prefix, preserving valid streamed responses from non-conforming providers. (#96503) Thanks @ZengWen-DT.
3334
- **LM Studio embedding preload:** honor model- and provider-level context-window limits when preloading embedding models, preventing avoidable GPU out-of-memory failures. (#100750) Thanks @zak-li, @ZOOWH, and @hxz398.

‎src/infra/delivery-recovery.shared.ts‎

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,81 @@
11
import { sleep } from "../utils/sleep.js";
2+
import { collectErrorGraphCandidates, extractErrorCode } from "./errors.js";
3+
import { getRetryAttemptErrors } from "./retry.js";
24

35
const RECOVERY_BACKOFF_MS: readonly number[] = [5_000, 25_000, 120_000, 600_000];
46
export const RECOVERY_REPLAY_SPACING_MS = 250;
57

8+
const PRE_CONNECT_ERROR_CODES = new Set([
9+
"ECONNREFUSED",
10+
"ENOTFOUND",
11+
"EAI_AGAIN",
12+
"ENETDOWN",
13+
"ENETUNREACH",
14+
"EHOSTUNREACH",
15+
]);
16+
const TRANSPORT_ERROR_CODE_RE =
17+
/^(?:E(?:AI_|CONN|NET|HOST|ADDR|PIPE|TIMEDOUT|SOCKET)|UND_ERR_|ERR_(?:NETWORK|HTTP2|QUIC|TLS|SSL))/;
18+
19+
function isProvenPreConnectCandidate(candidate: unknown): boolean {
20+
const code = extractErrorCode(candidate)?.trim().toUpperCase();
21+
if (code === "UND_ERR_CONNECT_TIMEOUT" || code === "UND_ERR_DNS_RESOLVE_FAILED") {
22+
return true;
23+
}
24+
if (!code || !PRE_CONNECT_ERROR_CODES.has(code) || !candidate || typeof candidate !== "object") {
25+
return false;
26+
}
27+
const syscall = (candidate as { syscall?: unknown }).syscall;
28+
return syscall === "connect" || syscall === "getaddrinfo";
29+
}
30+
31+
function nestedErrorCandidates(current: Record<string, unknown>): unknown[] {
32+
const retryAttempts = getRetryAttemptErrors(current);
33+
if (isProvenPreConnectCandidate(current)) {
34+
return retryAttempts ? [...retryAttempts] : [];
35+
}
36+
const nested = [current.cause, current.original, current.error, current.reason];
37+
if (Array.isArray(current.errors)) {
38+
nested.push(...current.errors);
39+
}
40+
const nestedObjects = nested.filter(
41+
(candidate) => candidate !== null && typeof candidate === "object",
42+
);
43+
return retryAttempts ? [...retryAttempts, ...nestedObjects] : nestedObjects;
44+
}
45+
46+
export function isPreConnectNetworkError(err: unknown): boolean {
47+
let foundPreConnectProof = false;
48+
for (const candidate of collectErrorGraphCandidates(err, nestedErrorCandidates)) {
49+
const code = extractErrorCode(candidate)?.trim().toUpperCase();
50+
if (isProvenPreConnectCandidate(candidate)) {
51+
foundPreConnectProof = true;
52+
continue;
53+
}
54+
const nested =
55+
candidate && typeof candidate === "object"
56+
? nestedErrorCandidates(candidate as Record<string, unknown>)
57+
: [];
58+
const isPreConnectAggregateSummary =
59+
candidate !== null &&
60+
typeof candidate === "object" &&
61+
Array.isArray((candidate as { errors?: unknown }).errors) &&
62+
code !== undefined &&
63+
PRE_CONNECT_ERROR_CODES.has(code);
64+
// Wrapper nodes may carry neutral SDK codes. Every transport leaf must still
65+
// prove pre-connect failure; Node AggregateError summary codes are accepted
66+
// only after their children are traversed and independently prove the same.
67+
if (
68+
nested.length === 0 ||
69+
(code &&
70+
!isPreConnectAggregateSummary &&
71+
(PRE_CONNECT_ERROR_CODES.has(code) || TRANSPORT_ERROR_CODE_RE.test(code)))
72+
) {
73+
return false;
74+
}
75+
}
76+
return foundPreConnectProof;
77+
}
78+
679
export function computeBackoffMs(retryCount: number): number {
780
if (retryCount <= 0) {
881
return 0;

‎src/infra/outbound/deliver.queue-integration.test.ts‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -171,4 +171,58 @@ describe("deliverOutboundPayloads queue integration: mid-batch failure with send
171171
expect(entry.recoveryState).toBe("send_attempt_started");
172172
expect(entry.lastError).toContain("first payload send failed");
173173
});
174+
175+
it("replays an entry after a proven pre-connect failure clears send evidence", async () => {
176+
process.env.OPENCLAW_STATE_DIR = tmpDir;
177+
const connectError = Object.assign(new Error("connect ECONNREFUSED"), {
178+
code: "ECONNREFUSED",
179+
syscall: "connect",
180+
});
181+
const sendMatrix = vi.fn().mockRejectedValueOnce(connectError);
182+
183+
await expect(
184+
deliverOutboundPayloads({
185+
cfg: {} as OpenClawConfig,
186+
channel: "matrix",
187+
to: "!room:example",
188+
payloads: [{ text: "first" }],
189+
deps: { matrix: sendMatrix },
190+
queuePolicy: "required",
191+
}),
192+
).rejects.toThrow("ECONNREFUSED");
193+
194+
const beforeDrain = await loadPendingDeliveries(tmpDir);
195+
expect(beforeDrain).toHaveLength(1);
196+
expect(beforeDrain[0]).toMatchObject({
197+
retryCount: 1,
198+
lastError: expect.stringContaining("ECONNREFUSED"),
199+
});
200+
expect(beforeDrain[0]?.recoveryState).toBeUndefined();
201+
expect(beforeDrain[0]?.platformSendStartedAt).toBeUndefined();
202+
203+
const recoverySendMatrix = vi
204+
.fn()
205+
.mockRejectedValueOnce(connectError)
206+
.mockResolvedValueOnce({ messageId: "recovered" });
207+
const deliver = vi.fn<DeliverFn>(async (params) =>
208+
deliverOutboundPayloads({
209+
...params,
210+
deps: { matrix: recoverySendMatrix },
211+
}),
212+
);
213+
await drainMatrixReconnect({ deliver, stateDir: tmpDir });
214+
215+
expect(deliver).toHaveBeenCalledTimes(1);
216+
const afterRepeatedFailure = await loadPendingDeliveries(tmpDir);
217+
expect(afterRepeatedFailure).toHaveLength(1);
218+
expect(afterRepeatedFailure[0]?.retryCount).toBe(2);
219+
expect(afterRepeatedFailure[0]?.recoveryState).toBeUndefined();
220+
expect(afterRepeatedFailure[0]?.platformSendStartedAt).toBeUndefined();
221+
222+
await drainMatrixReconnect({ deliver, stateDir: tmpDir });
223+
224+
expect(deliver).toHaveBeenCalledTimes(2);
225+
expect(recoverySendMatrix).toHaveBeenCalledTimes(2);
226+
expect(await loadPendingDeliveries(tmpDir)).toHaveLength(0);
227+
});
174228
});

0 commit comments

Comments
 (0)