karlseguin.tngl.sh
Native PostgreSQL driver / client for Zig
Propagate error.Canceled instead of swallowing it (#125)
On an async std.Io a cancelled read/write surfaces as ReadFailed/WriteFailed
with the real error.Canceled hidden in the reader's .err, which we dropped.
Worse, a read error kicked off a readyForQuery drain that blocked until the
query finished, so a query deadline never worked.
Recover the real error in readStream/writeStream, and only drain after a PG
error (Conn.recoverFromError still re-raises a cancellation). drain() now
skips poisoned connections, and the teardown Terminate writes are shielded
from cancellation. Includes a test that cancels an in-flight pg_sleep and
checks it fails fast with error.Canceled.
author
Lukáš Lalinský
committer
GitHub
date
(Jul 9, 2026, 10:37 AM +0800)
commit
c9213c2a
c9213c2a0f9942e76ad856efbd2e5847f153b55e
parent
a4e12641
a4e126413decee1ba141db7dfd37c84b6cc385af
+61
-16
+36
-8
src/conn.zig
+36
-8
src/conn.zig
···
198
198
allocator.free(self._param_oids);
199
199
self._result_state.deinit(allocator);
200
200
201
-
// try to send a Terminate to the DB
202
-
self.write(&.{ 'X', 0, 0, 0, 4 }) catch {};
201
+
lib.sendTerminate(&self._stream, self._io);
203
202
lib.freeSSLContext(self._ssl_ctx);
204
203
self._stream.close();
205
204
···
401
400
var affected: ?i64 = null;
402
401
while (true) {
403
402
const msg = self.read() catch |err| {
404
-
if (err == error.PG) {
405
-
self.readyForQuery() catch {};
406
-
}
403
+
if (err == error.PG) try self.recoverFromError();
407
404
return err;
408
405
};
409
406
switch (msg.type) {
···
456
453
try self.write(buf.string());
457
454
while (true) {
458
455
const msg = self.read() catch |err| {
459
-
if (state != .fail and err == error.PG) {
460
-
self.readyForQuery() catch {};
461
-
}
456
+
if (state != .fail and err == error.PG) try self.recoverFromError();
462
457
return err;
463
458
};
464
459
switch (msg.type) {
···
563
558
if (msg.type != 'Z') {
564
559
return self.unexpectedDBMessage();
565
560
}
561
+
}
562
+
563
+
// Drain the trailing ReadyForQuery after a server error so the connection
564
+
// stays usable. Best-effort, but never swallow a cancellation.
565
+
pub fn recoverFromError(self: *Conn) error{Canceled}!void {
566
+
self.readyForQuery() catch |err| {
567
+
if (err == error.Canceled) return error.Canceled;
568
+
};
566
569
}
567
570
};
568
571
···
1953
1956
var conn = try t.connect(.{ .tls = Conn.Opts.TLS{ .verify_full = "tests/root.crt" }, .username = "pgz_user_ssl", .password = "pgz_user_ssl_pw" });
1954
1957
defer conn.deinit();
1955
1958
}
1959
+
}
1960
+
1961
+
test "Conn: query is cancelable" {
1962
+
const S = struct {
1963
+
fn sleepQuery(c: *Conn) !void {
1964
+
var result = try c.query("select pg_sleep(3)", .{});
1965
+
result.deinit();
1966
+
}
1967
+
};
1968
+
1969
+
var conn = try t.connect(.{});
1970
+
defer conn.deinit();
1971
+
1972
+
// Run the query concurrently, let it reach its blocking read, then cancel.
1973
+
var future = try t.io.concurrent(S.sleepQuery, .{&conn});
1974
+
try t.io.sleep(.fromMilliseconds(50), .awake);
1975
+
1976
+
const start = std.Io.Clock.Timestamp.now(t.io, .awake);
1977
+
const result = future.cancel(t.io);
1978
+
const elapsed_ms = start.untilNow(t.io).raw.toMilliseconds();
1979
+
1980
+
try t.expectError(error.Canceled, result);
1981
+
try t.expectEqual(true, elapsed_ms < 1500); // prompt, not blocked until pg_sleep ends
1982
+
try t.expectEqual(Conn.State.fail, conn._state);
1983
+
try t.expectError(error.ConnectionBusy, conn.exec("select 1", .{}));
1956
1984
}
1957
1985
1958
1986
test "PG: cached query" {
+1
src/lib.zig
+1
src/lib.zig
···
14
14
pub const Stmt = @import("stmt.zig").Stmt;
15
15
pub const Pool = @import("pool.zig").Pool;
16
16
pub const Stream = @import("stream.zig").Stream;
17
+
pub const sendTerminate = @import("stream.zig").sendTerminate;
17
18
pub const metrics = @import("metrics.zig");
18
19
pub const has_openssl = build_config.openssl;
19
20
pub const SSLCtx = if (has_openssl) openssl.SSL_CTX else void;
+1
-2
src/listener.zig
+1
-2
src/listener.zig
+3
-1
src/result.zig
+3
-1
src/result.zig
···
60
60
// and returning an error union in deinit is a pain for the caller.
61
61
pub fn drain(self: *Result) !void {
62
62
var conn = self._conn;
63
-
if (conn._state == .idle) {
63
+
// Only an in-flight query has anything to drain; reading in any other
64
+
// state (e.g. a poisoned connection) would block.
65
+
if (conn._state != .query) {
64
66
return;
65
67
}
66
68
+2
-2
src/stmt.zig
+2
-2
src/stmt.zig
···
167
167
// If Parse fails, then the server won't reply to our other messages
168
168
// (i.e. Describe) and it'l immediately send a ReadyForQuery.
169
169
const msg = conn.read() catch |err| {
170
-
conn.readyForQuery() catch {};
170
+
if (err == error.PG) try conn.recoverFromError();
171
171
return err;
172
172
};
173
173
···
341
341
342
342
{
343
343
const msg = conn.read() catch |err| {
344
-
conn.readyForQuery() catch {};
344
+
if (err == error.PG) try conn.recoverFromError();
345
345
return err;
346
346
};
347
347
if (msg.type != '2') {
+18
-3
src/stream.zig
+18
-3
src/stream.zig
···
321
321
var vecs: [1][]u8 = .{buf};
322
322
var reader = stream.reader(io, &.{});
323
323
const r = &reader.interface;
324
-
return try r.readVec(&vecs);
324
+
return r.readVec(&vecs) catch |err| switch (err) {
325
+
error.ReadFailed => return reader.err orelse err,
326
+
else => return err,
327
+
};
325
328
}
326
329
327
330
fn writeStream(stream: Io.net.Stream, io: Io, data: []const u8) !void {
328
331
var buf: [1024]u8 = undefined;
329
332
var writer = stream.writer(io, &buf);
330
333
const w = &writer.interface;
331
-
try w.writeAll(data);
332
-
try w.flush();
334
+
w.writeAll(data) catch |err| switch (err) {
335
+
error.WriteFailed => return writer.err orelse err,
336
+
};
337
+
w.flush() catch |err| switch (err) {
338
+
error.WriteFailed => return writer.err orelse err,
339
+
};
340
+
}
341
+
342
+
// Sends a best-effort Terminate ('X') message, shielded from cancellation so
343
+
// teardown can't be interrupted.
344
+
pub fn sendTerminate(stream: *Stream, io: Io) void {
345
+
const prev = io.swapCancelProtection(.blocked);
346
+
defer _ = io.swapCancelProtection(prev);
347
+
stream.writeAll(&.{ 'X', 0, 0, 0, 4 }) catch {};
333
348
}
334
349
335
350
fn isHostName(host: []const u8) bool {