Skip to content

Commit 8833162

Browse files
committed
fix(clickhouse): retry reads on transient connection errors
ClickHouse reads (query, queryWithStats, queryFast) ran the underlying client.query once and, on any failure, returned a QueryError with no retry. Transient connection errors — most commonly a dead keep-alive socket surfacing as ECONNRESET on the first use of a pooled connection, but also EPIPE/ETIMEDOUT/"socket hang up" — are safe to retry and should not fail the request. - Add isRetryableConnectionError to classify transient connection errors by socket code or message substring, unwrapping a nested cause. Server-side ClickHouseErrors (e.g. SQL errors) are never classified as retryable. - Wrap the read client.query calls in a bounded retry (default 3 attempts, full-jitter exponential backoff). Attempt count and delays are configurable via ClickhouseConfig.connectionRetry so callers and tests can tune or disable the backoff. - Fix the keep-alive config mapping: @clickhouse/client expects snake_case idle_socket_ttl, so the camelCase idleSocketTtl was being silently dropped and the idle-socket TTL never honored. - Add unit tests covering retry-then-success, bounded give-up, non-retryable server errors, and error classification. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01B5mP4ULoVXdrAWmiMQh5bG
1 parent 8257499 commit 8833162

2 files changed

Lines changed: 395 additions & 34 deletions

File tree

Lines changed: 222 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,222 @@
1+
import { ClickHouseError } from "@clickhouse/client";
2+
import { describe, expect, it, vi } from "vitest";
3+
import { z } from "zod";
4+
import { ClickhouseClient, isRetryableConnectionError } from "./client.js";
5+
6+
/**
7+
* Unit tests for the connection-retry behaviour of read queries. These tests do
8+
* NOT require a live ClickHouse server: the underlying `@clickhouse/client`
9+
* query method is lazy (it does not connect at construction time), so we spy on
10+
* `client.client.query` and simulate transient connection errors.
11+
*/
12+
13+
function connectionResetError(message = "read ECONNRESET", code = "ECONNRESET") {
14+
return Object.assign(new Error(message), { code });
15+
}
16+
17+
function fakeResultSet(rows: Array<Record<string, unknown>>) {
18+
return {
19+
query_id: "test-query-id",
20+
response_headers: {} as Record<string, string>,
21+
json: async () => rows,
22+
} as any;
23+
}
24+
25+
function newClient(overrides?: Partial<ConstructorParameters<typeof ClickhouseClient>[0]>) {
26+
return new ClickhouseClient({
27+
name: "test",
28+
url: "http://localhost:8123",
29+
// Zero delays so retry tests run instantly.
30+
connectionRetry: { maxAttempts: 3, minDelayMs: 0, maxDelayMs: 0 },
31+
logLevel: "error",
32+
...overrides,
33+
});
34+
}
35+
36+
const schema = z.object({ value: z.number() });
37+
38+
describe("ClickhouseClient read retries", () => {
39+
it("retries a read that fails once with ECONNRESET and then succeeds", async () => {
40+
const client = newClient();
41+
42+
let calls = 0;
43+
const spy = vi.spyOn(client.client, "query").mockImplementation((async () => {
44+
calls++;
45+
if (calls === 1) {
46+
throw connectionResetError();
47+
}
48+
return fakeResultSet([{ value: 42 }]);
49+
}) as any);
50+
51+
const runQuery = client.query({
52+
name: "retry-once",
53+
query: "SELECT 42 AS value",
54+
schema,
55+
});
56+
57+
const [error, result] = await runQuery({});
58+
59+
expect(error).toBeNull();
60+
expect(result).toEqual([{ value: 42 }]);
61+
expect(spy).toHaveBeenCalledTimes(2);
62+
});
63+
64+
it("gives up after the configured number of attempts on repeated connection resets", async () => {
65+
const client = newClient({ connectionRetry: { maxAttempts: 3, minDelayMs: 0, maxDelayMs: 0 } });
66+
67+
const spy = vi.spyOn(client.client, "query").mockImplementation((async () => {
68+
throw connectionResetError();
69+
}) as any);
70+
71+
const runQuery = client.query({
72+
name: "always-reset",
73+
query: "SELECT 1 AS value",
74+
schema,
75+
});
76+
77+
const [error, result] = await runQuery({});
78+
79+
expect(result).toBeNull();
80+
expect(error).not.toBeNull();
81+
expect(error?.name).toBe("QueryError");
82+
expect(error?.message).toContain("ECONNRESET");
83+
// maxAttempts total (1 initial + 2 retries)
84+
expect(spy).toHaveBeenCalledTimes(3);
85+
});
86+
87+
it("does NOT retry a server-side ClickHouseError (e.g. SQL error)", async () => {
88+
const client = newClient();
89+
90+
const spy = vi.spyOn(client.client, "query").mockImplementation((async () => {
91+
throw new ClickHouseError({
92+
code: "62",
93+
type: "SYNTAX_ERROR",
94+
message: "Syntax error near 'SELCT'",
95+
});
96+
}) as any);
97+
98+
const runQuery = client.query({
99+
name: "sql-error",
100+
query: "SELCT 1",
101+
schema,
102+
});
103+
104+
const [error, result] = await runQuery({});
105+
106+
expect(result).toBeNull();
107+
expect(error?.name).toBe("QueryError");
108+
// Server errors must fail immediately, without retrying.
109+
expect(spy).toHaveBeenCalledTimes(1);
110+
});
111+
112+
it("does NOT retry a generic non-connection error", async () => {
113+
const client = newClient();
114+
115+
const spy = vi.spyOn(client.client, "query").mockImplementation((async () => {
116+
throw new Error("something unrelated went wrong");
117+
}) as any);
118+
119+
const runQuery = client.query({
120+
name: "generic-error",
121+
query: "SELECT 1 AS value",
122+
schema,
123+
});
124+
125+
const [error, result] = await runQuery({});
126+
127+
expect(result).toBeNull();
128+
expect(error?.name).toBe("QueryError");
129+
expect(spy).toHaveBeenCalledTimes(1);
130+
});
131+
132+
it("retries queryWithStats and queryFast on connection resets", async () => {
133+
// queryWithStats
134+
const statsClient = newClient();
135+
let statsCalls = 0;
136+
vi.spyOn(statsClient.client, "query").mockImplementation((async () => {
137+
statsCalls++;
138+
if (statsCalls === 1) throw connectionResetError("socket hang up", undefined as any);
139+
return fakeResultSet([{ value: 7 }]);
140+
}) as any);
141+
142+
const runWithStats = statsClient.queryWithStats({
143+
name: "stats-retry",
144+
query: "SELECT 7 AS value",
145+
schema,
146+
});
147+
const [statsError, statsResult] = await runWithStats({});
148+
expect(statsError).toBeNull();
149+
expect(statsResult?.rows).toEqual([{ value: 7 }]);
150+
expect(statsCalls).toBe(2);
151+
152+
// queryFast
153+
const fastClient = newClient();
154+
let fastCalls = 0;
155+
vi.spyOn(fastClient.client, "query").mockImplementation((async () => {
156+
fastCalls++;
157+
if (fastCalls === 1) throw connectionResetError();
158+
return {
159+
query_id: "fast-id",
160+
response_headers: {},
161+
stream: async function* () {
162+
yield [{ json: () => [99] }];
163+
},
164+
} as any;
165+
}) as any);
166+
167+
const runFast = fastClient.queryFast<{ value: number }, {}>({
168+
name: "fast-retry",
169+
query: "SELECT 99 AS value",
170+
columns: ["value"],
171+
});
172+
const [fastError, fastResult] = await runFast({});
173+
expect(fastError).toBeNull();
174+
expect(fastResult).toEqual([{ value: 99 }]);
175+
expect(fastCalls).toBe(2);
176+
});
177+
});
178+
179+
describe("isRetryableConnectionError", () => {
180+
it("classifies ECONNRESET (by code) as retryable", () => {
181+
expect(isRetryableConnectionError(connectionResetError("read ECONNRESET", "ECONNRESET"))).toBe(
182+
true
183+
);
184+
});
185+
186+
it("classifies EPIPE (by code) as retryable", () => {
187+
expect(isRetryableConnectionError(connectionResetError("write EPIPE", "EPIPE"))).toBe(true);
188+
});
189+
190+
it("classifies ETIMEDOUT (by code) as retryable", () => {
191+
expect(isRetryableConnectionError(connectionResetError("timeout", "ETIMEDOUT"))).toBe(true);
192+
});
193+
194+
it("classifies 'socket hang up' message (no code) as retryable", () => {
195+
expect(isRetryableConnectionError(new Error("socket hang up"))).toBe(true);
196+
});
197+
198+
it("classifies an ECONNRESET substring in the message as retryable", () => {
199+
expect(isRetryableConnectionError(new Error("Unable to query clickhouse: read ECONNRESET"))).toBe(
200+
true
201+
);
202+
});
203+
204+
it("classifies a wrapped connection error (via cause) as retryable", () => {
205+
const wrapped = new Error("outer failure");
206+
(wrapped as any).cause = connectionResetError();
207+
expect(isRetryableConnectionError(wrapped)).toBe(true);
208+
});
209+
210+
it("does NOT classify a server-side ClickHouseError as retryable", () => {
211+
const chError = new ClickHouseError({
212+
code: "241",
213+
type: "MEMORY_LIMIT_EXCEEDED",
214+
message: "Memory limit exceeded",
215+
});
216+
expect(isRetryableConnectionError(chError)).toBe(false);
217+
});
218+
219+
it("does NOT classify a generic error as retryable", () => {
220+
expect(isRetryableConnectionError(new Error("something unrelated"))).toBe(false);
221+
});
222+
});

0 commit comments

Comments
 (0)