feat(turbopg): vendor pg.zig + add io_uring transport (opt-in, Linux)

Vendors karlseguin/pg.zig and its two deps (buffer.zig, metrics.zig)
into the repo so we can stage an io_uring transport without a
separate fork repo. Adds an opt-in per-connection io_uring path
behind a new build flag, and a driver-only A/B bench against
Postgres 18 running in apple/container.

Vendored paths:
  zig/pg/                -- karlseguin/pg.zig @ 7605502 (was a git+url
                            dep, now .path)
  zig/pg-deps-buffer/    -- karlseguin/buffer.zig @ 30f9512 (identical
                            to the previously-cached upstream)
  zig/pg-deps-metrics/   -- karlseguin/metrics.zig @ 13d8706 + a tiny
                            Zig 0.16 compat fix on metric.zig:368
                            (@type(.{.int=...}) -> @int(bits, signed))

New build flag in zig/pg/build.zig:
  -Diouring=true         -- switch pg.Conn transport to io_uring on
                            Linux. Default false, no-op elsewhere.

Transport (zig/pg/src/stream.zig):
  * Per-connection std.os.linux.IoUring, 8 SQEs, no SQPOLL
  * writeAll: IORING_OP_SEND single-shot, submit + copy_cqe
  * read:    IORING_OP_RECV single-shot, submit + copy_cqe
  * connect path reuses PlainStream.connect (blocking getaddrinfo)
  * TLS path unchanged; iouring is skipped when has_openssl=true

Bench harness (bench/turbopg/):
  * bench.zig -- N worker threads, each owns one pg.Conn, hot loop
                 of either 'SELECT 1' or a 50-row generate_series,
                 measures rps for BENCH_DURATION seconds
  * run.sh    -- builds blocking + iouring variants, runs N iters
                 each, prints median rps
  * Containerfile -- debian:bookworm-slim + Zig 0.16.0 aarch64
  * RESULTS.md    -- captured medians from one local A/B run

Results (one local run, apple/container, Linux 6.18.5, 4 threads,
10s per iter, median of 3-5):

  query=SELECT 1   blocking=14,967 rps   iouring=15,178 rps  (+1.4%)
  query=50-row     blocking=13,903 rps   iouring=13,849 rps  (-0.4%)

Both deltas are within run-to-run noise. This is expected for a
per-op submit+wait: we trade one recv() syscall for two
io_uring_enter syscalls, so on a loopback TCP path the per-op cost
is roughly even. The real win requires SQPOLL, batched SEND, and a
cooperative scheduler so one ring drives many connections -- all
explicitly out of scope for this PR (see RESULTS.md).

Per AGENTS.md these numbers are not suitable for release notes,
framework comparison tables, or marketing copy.

Generated with [Devin](https://cli.devin.ai/docs)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

Author: justrach

bench/turbopg/.gitignore

@@ -0,0 +1,5 @@
+# Built artifacts and per-iter bench outputs
+zig-out/
+.zig-cache/
+zig-pkg/
+results/

bench/turbopg/Containerfile

@@ -0,0 +1,25 @@
+# Build + run the turbopg driver-only A/B bench against a Postgres 18
+# container. Per AGENTS.md: results from this container are NOT suitable
+# for release notes or comparison tables.
+FROM debian:bookworm-slim
+
+ENV DEBIAN_FRONTEND=noninteractive
+ENV PATH="/opt/zig:${PATH}"
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    ca-certificates curl xz-utils postgresql-client python3 \
+ && rm -rf /var/lib/apt/lists/*
+
+# Zig 0.16.0 aarch64-linux (matches host arch in apple/container on M-series).
+RUN mkdir -p /opt && cd /opt \
+ && curl -fsSL https://ziglang.org/download/0.16.0/zig-aarch64-linux-0.16.0.tar.xz -o zig.tar.xz \
+ && tar -xJf zig.tar.xz \
+ && mv zig-aarch64-linux-0.16.0 zig \
+ && rm zig.tar.xz
+
+WORKDIR /work
+
+COPY run.sh /usr/local/bin/run.sh
+RUN chmod +x /usr/local/bin/run.sh
+
+ENTRYPOINT ["/usr/local/bin/run.sh"]

bench/turbopg/RESULTS.md

@@ -0,0 +1,125 @@
+# turbopg / pg.zig — blocking vs io_uring transport A/B
+
+> **Scope:** the only code path that differs between the two builds is
+> the `Stream` struct in `zig/pg/src/stream.zig`. Everything else
+> (wire-protocol codec, `Conn`, `Pool`, result decoding) is identical.
+> The io_uring path is **one ring per connection**, single in-flight
+> SQE per `read` / `writeAll` (submit + `copy_cqe`). **No scheduler, no
+> SQPOLL, no multi-shot, no registered fds.** The real async win
+> needs all of those on top. Per `AGENTS.md`, do not cite these
+> numbers in release notes, framework comparison tables, or
+> marketing copy.
+
+## Environment
+
+- Apple `container` CLI (two Linux microVMs on macOS, shared vnic)
+- DB container: `postgres:18`, trust auth, default shm, `192.168.64.14`
+- Bench container: `debian:bookworm-slim` + Zig 0.16.0 aarch64-linux
+- Kernel (both VMs): `Linux 6.18.5 aarch64`
+- pg.zig build: `-Doptimize=ReleaseFast`
+- Workload: N worker threads, each owns one `pg.Conn`, hot loop of a
+  single query shape for the duration
+- Network: container-to-container via the default `container` network
+
+## Variants
+
+| label      | build flag         | transport used           |
+|------------|--------------------|--------------------------|
+| `blocking` | `-Diouring=false`  | `PlainStream` (read / send via libc syscalls) |
+| `iouring`  | `-Diouring=true`   | `IoUringStream` (one ring per conn, `IORING_OP_SEND` / `IORING_OP_RECV`, submit + `copy_cqe`) |
+
+Both variants share the same connect path, auth path, and `Reader`.
+
+## Workloads
+
+| id | SQL                                          | notes |
+|----|----------------------------------------------|-------|
+| 1  | `SELECT 1`                                   | smallest round trip |
+| 2  | `SELECT id FROM generate_series(1, 50) AS id` | 50 rows back per query, ~300 B response |
+
+## Results
+
+4 worker threads, 10 s per run.
+
+### query=1  (`SELECT 1`), median of 3
+
+| variant  | median rps | min     | max     |
+|----------|-----------:|--------:|--------:|
+| blocking |  14,967.12 | 14,878  | 15,140  |
+| iouring  |  15,178.03 | 15,127  | 15,230  |
+
+Δ: **+1.4 %**, well within run-to-run noise with n=3.
+
+### query=2  (`SELECT` generate_series(1,50)), median of 5
+
+| variant  | median rps | min     | max     |
+|----------|-----------:|--------:|--------:|
+| blocking |  13,903.37 | 13,770  | 14,001  |
+| iouring  |  13,849.02 | 13,752  | 13,929  |
+
+Δ: **−0.4 %**, again within noise.
+
+## What this tells us
+
+1. Per-connection single-SQE io_uring is **roughly a wash** on a
+   driver that was already blocking-sync. Expected: we trade one
+   `recv()` syscall for one `io_uring_enter(submit)` +
+   `io_uring_enter(wait_cqe)`, so the per-op syscall cost is
+   approximately even. Kernel fastpath for small receives on a local
+   TCP loop is already very fast.
+2. On q=2 (bigger response, more bytes per recv) io_uring trends
+   slightly slower — consistent with the extra ring bookkeeping
+   overhead showing up once the per-op cost matters at all.
+3. No regressions, no query errors. The abstraction and the ring
+   plumbing work correctly for the full `Conn` lifetime (connect,
+   startup, simple query, extended query, close).
+
+## What would actually move the needle
+
+The next items (deliberately **not** in this PR):
+
+- **SQPOLL** so submitting no longer needs an `io_uring_enter`
+  syscall in the common case.
+- **Batched send**: queue up the parse/bind/describe/execute/sync
+  packets into one `IORING_OP_SEND` instead of the current per-packet
+  writes.
+- **A cooperative scheduler** so one ring drives N connections
+  concurrently and a waiting query yields the thread rather than
+  blocking on `copy_cqe`. This is the real win and turns this from a
+  neutral change into an actual throughput improvement.
+
+## Caveats (read these)
+
+- 3–5 iterations, 10 s each, one client, one DB. Enough to catch
+  big regressions, not enough to publish percentage claims.
+- `postgres:18` with default config, no tuning, `trust` auth.
+- Apple `container` runs each container in its own microVM; cross-VM
+  network adds a real-ish TCP path but results will not match a
+  co-located production setup.
+- No TLS. The io_uring path is plaintext-only in this PR;
+  `-Dopenssl_lib_name=...` still picks TLS + the old blocking socket.
+
+## Reproducing
+
+```bash
+# 1. Start Postgres 18
+container run -d --name pg18 -e POSTGRES_HOST_AUTH_METHOD=trust postgres:18
+
+# 2. Build the bench image once
+container build -t turbopg-bench \
+    -f bench/turbopg/Containerfile bench/turbopg
+
+# 3. Find the pg18 IP (field ADDR in `container ls`)
+PG_IP=$(container ls | awk '$1=="pg18" {print $6}' | cut -d/ -f1)
+
+# 4. Run
+container run --rm -m 4G -c 4 \
+    -e PGHOST="$PG_IP" \
+    -e BENCH_QUERY=1 \
+    -e BENCH_ITERS=5 \
+    -v "$PWD":/work \
+    turbopg-bench
+```
+
+Override `BENCH_QUERY` (1 or 2), `BENCH_THREADS`, `BENCH_DURATION`,
+`BENCH_ITERS` as needed.

bench/turbopg/bench.zig

@@ -0,0 +1,167 @@
+//! Throughput bench for the vendored pg.zig driver.
+//!
+//! Spawns N worker threads, each running a hot loop of either:
+//!   * `SELECT 1`
+//!   * `SELECT generate_series(1,50) AS id`
+//!
+//! against a single Postgres instance. Each thread owns its own
+//! `pg.Conn` (no pool contention, no scheduler), so the only thing
+//! varying between the blocking-build and the iouring-build is the
+//! stream transport in `zig/pg/src/stream.zig`.
+//!
+//! Output is plain text:
+//!
+//!   transport=<blocking|iouring> threads=N duration=Ds query=<id>
+//!     queries=Q rows=R rps=X.YZ
+//!
+//! Per AGENTS.md, do NOT cite these numbers in release notes.
+
+const std = @import("std");
+const pg = @import("pg");
+
+const Args = struct {
+    host: []const u8 = "127.0.0.1",
+    port: u16 = 5432,
+    user: []const u8 = "postgres",
+    database: []const u8 = "postgres",
+    threads: usize = 4,
+    duration_s: u64 = 10,
+    query_id: u8 = 1,
+    label: []const u8 = "blocking",
+
+    fn fromEnv(_: std.mem.Allocator) !Args {
+        var a: Args = .{};
+        if (getenv("PGHOST")) |v| a.host = v;
+        if (getenv("PGPORT")) |v| a.port = try std.fmt.parseInt(u16, v, 10);
+        if (getenv("PGUSER")) |v| a.user = v;
+        if (getenv("PGDATABASE")) |v| a.database = v;
+        if (getenv("BENCH_THREADS")) |v| a.threads = try std.fmt.parseInt(usize, v, 10);
+        if (getenv("BENCH_DURATION")) |v| a.duration_s = try std.fmt.parseInt(u64, v, 10);
+        if (getenv("BENCH_QUERY")) |v| a.query_id = try std.fmt.parseInt(u8, v, 10);
+        if (getenv("BENCH_LABEL")) |v| a.label = v;
+        return a;
+    }
+};
+
+fn nowNs() i128 {
+    var ts: std.posix.timespec = undefined;
+    _ = std.posix.system.clock_gettime(.MONOTONIC, &ts);
+    return @as(i128, ts.sec) * 1_000_000_000 + @as(i128, ts.nsec);
+}
+
+fn sleepSeconds(secs: u64) void {
+    var ts: std.posix.timespec = .{ .sec = @intCast(secs), .nsec = 0 };
+    _ = std.posix.system.nanosleep(&ts, null);
+}
+
+fn getenv(key: []const u8) ?[]const u8 {
+    // libc getenv; each worker thread only reads it, never mutates.
+    var buf: [128]u8 = undefined;
+    const key_z = std.fmt.bufPrintZ(&buf, "{s}", .{key}) catch return null;
+    const raw = std.c.getenv(key_z.ptr) orelse return null;
+    return std.mem.span(raw);
+}
+
+const ThreadStats = struct {
+    queries: u64 = 0,
+    rows: u64 = 0,
+    err_count: u64 = 0,
+};
+
+fn workerLoop(
+    args: *const Args,
+    stop_flag: *std.atomic.Value(bool),
+    stats: *ThreadStats,
+) !void {
+    const allocator = std.heap.smp_allocator;
+
+    var conn = try pg.Conn.open(allocator, .{
+        .host = args.host,
+        .port = args.port,
+    });
+    defer conn.deinit();
+
+    try conn.auth(.{
+        .username = args.user,
+        .database = args.database,
+        .timeout = 10_000,
+    });
+
+    const sql = switch (args.query_id) {
+        1 => "SELECT 1",
+        2 => "SELECT id FROM generate_series(1, 50) AS id",
+        else => "SELECT 1",
+    };
+
+    while (!stop_flag.load(.acquire)) {
+        var result = conn.query(sql, .{}) catch |err| {
+            stats.err_count += 1;
+            if (stats.err_count > 10) return err;
+            continue;
+        };
+        defer result.deinit();
+
+        while (result.next() catch null) |_| {
+            stats.rows += 1;
+        }
+        stats.queries += 1;
+    }
+}
+
+fn workerEntry(
+    args: *const Args,
+    stop_flag: *std.atomic.Value(bool),
+    stats: *ThreadStats,
+) void {
+    workerLoop(args, stop_flag, stats) catch |err| {
+        std.debug.print("worker error: {s}\n", .{@errorName(err)});
+    };
+}
+
+pub fn main() !void {
+    const allocator = std.heap.smp_allocator;
+
+    const args = try Args.fromEnv(allocator);
+
+    std.debug.print(
+        "[bench] transport={s} threads={d} duration={d}s query={d} host={s}:{d}\n",
+        .{ args.label, args.threads, args.duration_s, args.query_id, args.host, args.port },
+    );
+
+    var stop_flag = std.atomic.Value(bool).init(false);
+
+    const stats = try allocator.alloc(ThreadStats, args.threads);
+    @memset(stats, .{});
+    defer allocator.free(stats);
+
+    const threads = try allocator.alloc(std.Thread, args.threads);
+    defer allocator.free(threads);
+
+    const t_start = nowNs();
+    for (threads, 0..) |*t, i| {
+        t.* = try std.Thread.spawn(.{}, workerEntry, .{ &args, &stop_flag, &stats[i] });
+    }
+
+    sleepSeconds(args.duration_s);
+
+    stop_flag.store(true, .release);
+    for (threads) |t| t.join();
+    const t_end = nowNs();
+
+    var total_q: u64 = 0;
+    var total_r: u64 = 0;
+    var total_e: u64 = 0;
+    for (stats) |s| {
+        total_q += s.queries;
+        total_r += s.rows;
+        total_e += s.err_count;
+    }
+    const elapsed_s: f64 = @as(f64, @floatFromInt(t_end - t_start)) / 1e9;
+    const rps: f64 = @as(f64, @floatFromInt(total_q)) / elapsed_s;
+    const rows_ps: f64 = @as(f64, @floatFromInt(total_r)) / elapsed_s;
+
+    std.debug.print(
+        "[bench] result transport={s} queries={d} rows={d} errors={d} elapsed={d:.2}s rps={d:.2} rows_ps={d:.2}\n",
+        .{ args.label, total_q, total_r, total_e, elapsed_s, rps, rows_ps },
+    );
+}

bench/turbopg/build.zig

@@ -0,0 +1,33 @@
+const std = @import("std");
+
+pub fn build(b: *std.Build) void {
+    const target = b.standardTargetOptions(.{});
+    const optimize = b.standardOptimizeOption(.{});
+    const iouring = b.option(bool, "iouring", "Use io_uring transport") orelse false;
+
+    const pg_dep = b.dependency("pg", .{
+        .target = target,
+        .optimize = optimize,
+        .iouring = iouring,
+    });
+
+    const exe = b.addExecutable(.{
+        .name = "turbopg-bench",
+        .root_module = b.createModule(.{
+            .target = target,
+            .optimize = optimize,
+            .root_source_file = b.path("bench.zig"),
+            .link_libc = true,
+            .imports = &.{
+                .{ .name = "pg", .module = pg_dep.module("pg") },
+            },
+        }),
+    });
+
+    b.installArtifact(exe);
+    const run_cmd = b.addRunArtifact(exe);
+    run_cmd.step.dependOn(b.getInstallStep());
+    if (b.args) |args| run_cmd.addArgs(args);
+    const run_step = b.step("run", "Run the bench");
+    run_step.dependOn(&run_cmd.step);
+}

bench/turbopg/build.zig.zon

@@ -0,0 +1,11 @@
+.{
+    .name = .turbopg_bench,
+    .version = "0.0.0",
+    .fingerprint = 0x86359a9f291096a0,
+    .paths = .{""},
+    .dependencies = .{
+        .pg = .{
+            .path = "../../zig/pg",
+        },
+    },
+}