1014 lines
49 KiB
Markdown
1014 lines
49 KiB
Markdown
# Milestone 3: Resolver Transport
|
||
|
||
Goal (PLAN §16 Phase 3): UDP server, TCP server, DoH + DoT upstream clients, upstream pool with
|
||
priority failover, backoff and health. Exit: A/AAAA forwarding over UDP and TCP; health populated.
|
||
|
||
Read first: `AGENTS.md` (values), `specs/research/zig-0.16-api-notes.md` (verified stdlib facts —
|
||
pre-0.16 knowledge is stale and MUST NOT be used), `specs/milestone-1.md` and `specs/milestone-2.md`
|
||
(module conventions and the "As built" notes). The Zig source of truth is `/home/mokhtar/app/zig` at
|
||
tag `0.16.0`. RFCs: 1035 §4.2.2 (TCP length prefix), 7766 (DNS over TCP), 7858 (DoT), 8484 (DoH),
|
||
6891 (EDNS(0)), 4343 (name case-insensitivity), 9619 (QDCOUNT must be 1).
|
||
|
||
## What already exists (do not respecify, import it)
|
||
|
||
- `src/dns/*` — pure wire format. Used here: `packet.parse`, `packet.Packet`, `packet.firstQuestion`,
|
||
`packet.findOptRecord`, `packet.ResponseBuilder`, `packet.setId`, `header.parse`, `header.Flags`,
|
||
`question.Question`, `name.eqlIgnoreCase`, `edns.parseOpt`, `types.Rcode`, `types.Type`.
|
||
- `src/platform/address.zig` — `NetAddress`, `Prefix`, `matchLongest`.
|
||
- `src/platform/tls_client.zig` — `TlsStream` (pinned struct; `init`, `reader`, `writer`, `close`),
|
||
`classify`, `ErrorClass`. DoT builds on this; do not open `std.crypto.tls.Client` directly.
|
||
- `src/platform/tls_server.zig` — server-side TLS. **Not used in this milestone** (local DoH/DoT
|
||
endpoints are Phase 9).
|
||
- `build.zig` — `-Dintegration` (hermetic, loopback only, PR-blocking) and `-Dlive` (leaves the
|
||
machine, manual workflow only) reach test files through `@import("build_options")`.
|
||
|
||
## Sessions
|
||
|
||
```
|
||
S1 (upstream/transport.zig + upstream/health.zig)
|
||
|
|
||
+--> S2 (doh_client) S3 (dot_client) S4 (pool) S5 (handler) [parallel]
|
||
|
|
||
+--> S6 (udp_server + tcp_server)
|
||
|
|
||
+--> S7 (e2e integration test)
|
||
```
|
||
|
||
S1 defines every type the other sessions share, so S2–S5 can be written against the spec alone.
|
||
S6 depends on S5's `Handler` type; S7 depends on everything.
|
||
|
||
The orchestrator — not any session — wires `src/tests.zig` imports and any `build.zig` change. A
|
||
session that needs a build change reports the exact change in its completion report.
|
||
|
||
## Design invariants (all sessions)
|
||
|
||
- The pure core stays pure. `src/dns/` gains nothing. All `std.Io` use lives in `src/server/` and
|
||
`src/upstream/`. `upstream/health.zig` and the validation half of `upstream/transport.zig` are
|
||
themselves pure (timestamps and randomness arrive as parameters), so they are unit-testable
|
||
without a backend.
|
||
- **Failures are classified into three disjoint groups**: peer fault, local resource, cancellation.
|
||
Health and backoff count peer faults ONLY. A local `OutOfMemory` must never mark an upstream sick,
|
||
and `error.Canceled` must never be recorded at all.
|
||
- **QDCOUNT must be exactly 1** (RFC 9619) — for inbound client queries (else FORMERR) and for
|
||
upstream responses (else the response is a peer fault).
|
||
- Every response from an upstream is validated against the request it answers before it reaches a
|
||
client: ID equal, QDCOUNT 1 on both sides, QR set, question name equal **case-insensitively**
|
||
(RFC 4343), qtype and qclass equal.
|
||
- TCP and DoT messages carry a 2-byte big-endian length prefix (RFC 1035 §4.2.2). DoH does not.
|
||
- No stream read or write in 0.16.0 accepts a timeout (verified: `Io.Operation` has no net-stream
|
||
variants). Bound them by running the work under `io.concurrent` / `std.Io.Select` and cancelling
|
||
the loser, exactly as `src/platform/tls_client_integration_test.zig` does. UDP `receive` is the one
|
||
exception — `Socket.receiveTimeout` is implemented on the POSIX Threaded backend (verified:
|
||
`Io/Threaded.zig:2779` handles `net_receive` in the pollable batch path).
|
||
- Never set `net.IpAddress.ConnectOptions.timeout` — the Threaded backend panics
|
||
(`Io/Threaded.zig:12077`).
|
||
- Every dropped datagram, refused connection and failed send increments a named counter. No silent
|
||
drops (AGENTS.md).
|
||
- Unit tests live in-file. Tests that touch loopback sockets live in a separate `*_integration_test.zig`
|
||
file guarded by `if (!build_options.integration) return error.SkipZigTest;`. Tests that leave the
|
||
machine are guarded by `build_options.live`. This mirrors milestone 1 — the guard is a runtime
|
||
return so the body is always compiled and cannot rot.
|
||
|
||
---
|
||
|
||
## Session S1: `src/upstream/transport.zig`, `src/upstream/health.zig`
|
||
|
||
Foundation. No sockets are opened in this session; `transport.zig` names the `Io`-taking interface
|
||
that S2–S4 implement, and everything else in both files is pure.
|
||
|
||
### S1.1 transport.zig — endpoint parsing
|
||
|
||
```zig
|
||
pub const max_message_len = 65535; // RFC 1035 §4.2.2 length prefix is 16-bit
|
||
pub const doh_default_port = 443;
|
||
pub const dot_default_port = 853; // RFC 7858 §3.1
|
||
pub const doh_default_path = "/dns-query"; // RFC 8484 §4.1 well-known template
|
||
|
||
pub const Scheme = enum { doh, dot };
|
||
|
||
/// Borrowed view over the configured URL text; the caller owns the string.
|
||
pub const Endpoint = struct {
|
||
scheme: Scheme,
|
||
url: []const u8, // the original text, for logs and the health API
|
||
host: []const u8, // no brackets, no port; SNI and certificate verification name
|
||
port: u16,
|
||
path: []const u8, // DoH only; always starts with '/'; `doh_default_path` when absent
|
||
|
||
pub const ParseError = error{ UnsupportedScheme, MissingHost, BadPort, BadUrl };
|
||
|
||
/// `https://…` => .doh, `tls://…` => .dot (PLAN §9). Accepts `[v6]:port` bracket form.
|
||
pub fn parse(url: []const u8) ParseError!Endpoint;
|
||
};
|
||
```
|
||
|
||
Hand-written parse (a `std.Uri` round trip would hand back percent-encoded components that need
|
||
re-decoding for one household-scale config value). Rules: reject any other scheme, reject an empty
|
||
host, reject a port that is empty, non-numeric or > 65535, reject a `tls://` URL that carries a path
|
||
other than `/` or nothing.
|
||
|
||
Tests: `https://cloudflare-dns.com/dns-query`, `https://dns.example/x` (path preserved),
|
||
`https://dns.example` (path defaults), `tls://dns.google:853`, `tls://dns.google` (port defaults to
|
||
853), `https://[2606:4700:4700::1111]:8443/dns-query` (host without brackets, port 8443),
|
||
`udp://1.1.1.1:53` → `UnsupportedScheme`, `https://` → `MissingHost`, `https://h:99999/` → `BadPort`.
|
||
|
||
### S1.2 transport.zig — error groups
|
||
|
||
```zig
|
||
/// The upstream misbehaved, timed out, or was unreachable. Only these count against health.
|
||
pub const PeerFault = error{
|
||
ConnectFailed,
|
||
TlsFailed,
|
||
SendFailed,
|
||
ReceiveFailed,
|
||
Timeout,
|
||
BadResponse, // unparseable, not a response, or QDCOUNT != 1
|
||
ResponseMismatch, // ID or question does not match the query
|
||
ResponseTooLarge, // does not fit the caller's buffer
|
||
HttpStatus, // DoH: status other than 200
|
||
HttpContentType, // DoH: content-type other than application/dns-message
|
||
};
|
||
|
||
/// This process ran out of something. Never the upstream's fault.
|
||
pub const LocalResource = error{
|
||
OutOfMemory,
|
||
SystemResources,
|
||
ProcessFdQuotaExceeded,
|
||
SystemFdQuotaExceeded,
|
||
BufferTooSmall, // caller-supplied buffer cannot hold even a query
|
||
Unexpected,
|
||
};
|
||
|
||
pub const Cancellation = error{Canceled};
|
||
|
||
pub const ExchangeError = PeerFault || LocalResource || Cancellation;
|
||
|
||
pub const Group = enum { peer_fault, local_resource, cancellation };
|
||
|
||
/// Exhaustive switch over `ExchangeError` — no `else` arm. A new error member must break
|
||
/// the build here, so no failure can silently land in the wrong group.
|
||
pub fn group(err: ExchangeError) Group;
|
||
```
|
||
|
||
The three sets are disjoint by construction; a test asserts it by walking
|
||
`@typeInfo(PeerFault).error_set.?` and friends and checking no name appears twice.
|
||
|
||
`pub fn mapLocal(err: anyerror) ?ExchangeError` — helper for S2/S3: returns the `LocalResource` or
|
||
`Cancellation` member matching a foreign stdlib error by name, `null` when the caller should treat
|
||
the error as a peer fault. Implemented with an explicit switch over the named errors listed above
|
||
plus `error.Canceled`; this is the ONLY place a foreign error set is folded in.
|
||
|
||
### S1.3 transport.zig — the client interface
|
||
|
||
```zig
|
||
/// A thing that sends one DNS message and returns one validated DNS message.
|
||
/// Implemented by DohClient, DotClient, Pool, and test fakes.
|
||
pub const Client = struct {
|
||
ptr: *anyopaque,
|
||
exchangeFn: *const fn (
|
||
ptr: *anyopaque,
|
||
io: std.Io,
|
||
query: []const u8,
|
||
response_buf: []u8,
|
||
) ExchangeError![]u8,
|
||
|
||
/// Returns a prefix of `response_buf`. The returned message has already passed
|
||
/// `validateResponse` against `query`.
|
||
pub fn exchange(
|
||
self: Client,
|
||
io: std.Io,
|
||
query: []const u8,
|
||
response_buf: []u8,
|
||
) ExchangeError![]u8 {
|
||
return self.exchangeFn(self.ptr, io, query, response_buf);
|
||
}
|
||
};
|
||
```
|
||
|
||
### S1.4 transport.zig — response validation (pure)
|
||
|
||
```zig
|
||
pub const ValidateError = error{ BadResponse, ResponseMismatch };
|
||
|
||
/// RFC 9619: exactly one question on both sides. RFC 4343: names compare case-insensitively,
|
||
/// so a case-mangling (0x20) upstream still matches. Does not inspect the answer section —
|
||
/// content policy is not this layer's business.
|
||
pub fn validateResponse(query: []const u8, response: []const u8) ValidateError!void;
|
||
```
|
||
|
||
Order of checks: response parses (`packet.parse` failure → `BadResponse`); response header
|
||
`flags.qr` is true (else `BadResponse`); response `qdcount == 1` (else `BadResponse`); query parses
|
||
and has `qdcount == 1` (else `BadResponse` — a caller bug, but never a crash); IDs equal (else
|
||
`ResponseMismatch`); question `qtype` and `qclass` equal and `name.eqlIgnoreCase` (else
|
||
`ResponseMismatch`).
|
||
|
||
Tests (hand-built byte fixtures, reuse the shape of `src/dns/packet.zig`'s fixtures): matching pair
|
||
passes; mixed-case question name passes; wrong ID → `ResponseMismatch`; different qtype →
|
||
`ResponseMismatch`; different name → `ResponseMismatch`; QR clear → `BadResponse`; response with
|
||
QDCOUNT 0 → `BadResponse`; response with QDCOUNT 2 → `BadResponse`; truncated garbage →
|
||
`BadResponse`.
|
||
|
||
### S1.5 health.zig — pure health and backoff state
|
||
|
||
```zig
|
||
pub const Config = struct {
|
||
/// Consecutive peer faults before the endpoint is put in backoff.
|
||
failure_threshold: u8 = 2,
|
||
base_backoff_ms: u32 = 500,
|
||
max_backoff_ms: u32 = 60_000,
|
||
};
|
||
|
||
pub const window_len = 32; // rolling success-rate window
|
||
|
||
pub const State = struct {
|
||
consecutive_failures: u32,
|
||
total_successes: u64,
|
||
total_failures: u64,
|
||
last_success_at: ?std.Io.Timestamp,
|
||
last_error_at: ?std.Io.Timestamp,
|
||
last_error_buf: [48]u8,
|
||
last_error_len: u8, // @errorName of the last peer fault, truncated
|
||
backoff_until: ?std.Io.Timestamp,
|
||
window: u32, // bitset, 1 = success, LSB = most recent
|
||
window_filled: u8,
|
||
|
||
pub const init: State = ...;
|
||
|
||
pub fn recordSuccess(self: *State, at: std.Io.Timestamp) void;
|
||
|
||
/// `err_name` is `@errorName` of a PeerFault member. `rand` supplies jitter; the caller
|
||
/// owns the RNG so this stays pure and the test is deterministic.
|
||
pub fn recordFailure(
|
||
self: *State,
|
||
at: std.Io.Timestamp,
|
||
err_name: []const u8,
|
||
cfg: Config,
|
||
rand: u32,
|
||
) void;
|
||
|
||
pub fn available(self: *const State, now: std.Io.Timestamp) bool;
|
||
pub fn successRate(self: *const State) f32; // over the filled part of the window; 1.0 when empty
|
||
pub fn lastError(self: *const State) []const u8;
|
||
};
|
||
```
|
||
|
||
Rules:
|
||
|
||
1. `recordSuccess` clears `consecutive_failures` and `backoff_until`, pushes a 1 into the window,
|
||
and sets `last_success_at` to `@max(existing, at)` by nanoseconds.
|
||
2. `recordFailure` increments `consecutive_failures` and `total_failures`, pushes a 0, copies
|
||
`err_name` (truncated to the buffer), sets `last_error_at` to `@max(existing, at)`. When
|
||
`consecutive_failures >= cfg.failure_threshold`, it computes
|
||
`delay_ms = min(max_backoff_ms, base_backoff_ms << shift)` where
|
||
`shift = min(consecutive_failures - failure_threshold, 20)`, applies jitter
|
||
`jittered = delay_ms/2 + rand % (delay_ms/2 + 1)`, and sets
|
||
`backoff_until = @max(existing, at + jittered)`.
|
||
3. **Out-of-order completions are the normal case**, not an edge case: two concurrent exchanges
|
||
against the same endpoint complete in either order, so `at` can move backwards between calls.
|
||
Every timestamp field therefore updates through `@max` on `.nanoseconds`, and `backoff_until` is
|
||
only ever extended, never shortened. `available` must never compute a negative duration.
|
||
4. `State` carries no lock. The pool owns the mutex.
|
||
|
||
Tests: threshold not reached → `available(now)` stays true; threshold reached → unavailable until
|
||
`backoff_until`, available one nanosecond after; consecutive failures grow the delay and it saturates
|
||
at `max_backoff_ms` (walk 40 failures with `rand = 0`); success resets consecutive count, window and
|
||
backoff; jitter with `rand = 0` and `rand = maxInt(u32)` both land inside `[delay/2, delay]`; an
|
||
out-of-order pair — `recordSuccess(t=100)` then `recordSuccess(t=50)` — leaves `last_success_at` at
|
||
100; `recordFailure(t=50)` after `recordFailure(t=100)` does not shorten `backoff_until`;
|
||
`successRate` over a half-success window is 0.5; `lastError` returns the last recorded name and is
|
||
truncated, not overflowed, by a name longer than the buffer.
|
||
|
||
### S1.6 Acceptance criteria
|
||
|
||
- [ ] `zig test src/upstream/transport.zig` and `zig test src/upstream/health.zig` pass.
|
||
- [ ] `group` compiles with no `else` arm over `ExchangeError`.
|
||
- [ ] The disjointness test over the three error sets passes.
|
||
- [ ] Every listed validation and health case has a test.
|
||
- [ ] `zig fmt --check` clean on both files.
|
||
|
||
---
|
||
|
||
## Session S2: `src/upstream/doh_client.zig` (+ live test)
|
||
|
||
DoH per RFC 8484 over `std.http.Client` (HTTP/1.1; HTTP/2 is permanently out of scope, PLAN §2.2).
|
||
|
||
### S2.1 API
|
||
|
||
```zig
|
||
pub const DohClient = struct {
|
||
http: *std.http.Client, // caller-owned; shared across endpoints, pools connections
|
||
endpoint: transport.Endpoint,
|
||
uri: std.Uri, // built once in init from `endpoint`
|
||
request_buf: []u8, // caller-owned; sendBodyComplete needs a mutable body
|
||
transfer_buf: []u8, // caller-owned; HTTP body transfer buffer
|
||
|
||
pub const InitError = error{BadUrl};
|
||
|
||
pub fn init(
|
||
http: *std.http.Client,
|
||
endpoint: transport.Endpoint,
|
||
request_buf: []u8, // asserted >= 512
|
||
transfer_buf: []u8, // asserted >= 1024
|
||
) InitError!DohClient;
|
||
|
||
pub fn client(self: *DohClient) transport.Client;
|
||
};
|
||
```
|
||
|
||
`self.uri` is derived from `endpoint` with `std.Uri.parse(endpoint.url)`; a parse failure is
|
||
`error.BadUrl`. `endpoint` remains the source of truth for host/port/scheme in logs.
|
||
|
||
### S2.2 Exchange
|
||
|
||
1. `if (query.len > self.request_buf.len) return error.BufferTooSmall;` then copy `query` into
|
||
`request_buf` — `Request.sendBodyComplete` takes `[]u8`, not `[]const u8` (verified,
|
||
`std/http/Client.zig:935`).
|
||
2. ```zig
|
||
var req = try self.http.request(.POST, self.uri, .{
|
||
.keep_alive = true,
|
||
.redirect_behavior = .not_allowed,
|
||
.headers = .{
|
||
.content_type = .{ .override = "application/dns-message" },
|
||
// Compressed bodies would need the decompressing reader; identity keeps the
|
||
// response body byte-exact for `validateResponse`.
|
||
.accept_encoding = .{ .override = "identity" },
|
||
},
|
||
.extra_headers = &.{.{ .name = "accept", .value = "application/dns-message" }},
|
||
});
|
||
defer req.deinit();
|
||
```
|
||
`Request.Headers` has no `accept` field (verified, `std/http/Client.zig:845`), hence
|
||
`extra_headers`.
|
||
3. `try req.sendBodyComplete(self.request_buf[0..query.len]);`
|
||
4. `var resp = try req.receiveHead(&.{});` — an empty redirect buffer is legal with
|
||
`.not_allowed`.
|
||
5. `if (resp.head.status != .ok) return error.HttpStatus;`
|
||
6. Content-type: take `resp.head.content_type orelse return error.HttpContentType`, cut at the
|
||
first `;`, trim ASCII whitespace, `std.ascii.eqlIgnoreCase` against `"application/dns-message"`,
|
||
else `error.HttpContentType`.
|
||
7. `const body = resp.reader(self.transfer_buf);` then fill `response_buf` with repeated
|
||
`readSliceShort` until it returns 0. If the body has not ended when `response_buf` is full →
|
||
`error.ResponseTooLarge`. A `content_length` larger than `response_buf.len` short-circuits to the
|
||
same error.
|
||
8. `try transport.validateResponse(query, response_buf[0..len]);` mapping `ValidateError` members
|
||
straight through (both are `PeerFault` members).
|
||
9. Return `response_buf[0..len]`.
|
||
|
||
The query ID is sent unchanged. RFC 8484 §4.1 suggests ID 0 for HTTP cache friendliness; nxdns puts
|
||
no HTTP cache in this path, and keeping the ID preserves the request/response binding that
|
||
`validateResponse` enforces. State that reason in a comment.
|
||
|
||
Error mapping: one `fn mapError(err: anyerror) transport.ExchangeError` that first tries
|
||
`transport.mapLocal(err)`, and otherwise maps by phase — `request`/connect errors →
|
||
`error.ConnectFailed`, TLS-named errors (`@errorName` starting with `"Tls"` or `"Certificate"`) →
|
||
`error.TlsFailed`, `sendBodyComplete` errors → `error.SendFailed`, `receiveHead` and body-read
|
||
errors → `error.ReceiveFailed`. Do the phase distinction at the call site (each `catch` knows its
|
||
phase), not by guessing from the error name.
|
||
|
||
### S2.3 Tests
|
||
|
||
- In-file unit: `init` with a `tls://` endpoint is rejected before any I/O (assert or `BadUrl`);
|
||
`init` asserts on an undersized `request_buf`; the content-type matcher accepts
|
||
`"application/dns-message"`, `"Application/DNS-Message"` and
|
||
`"application/dns-message; charset=utf-8"`, and rejects `"text/html"` and a missing header.
|
||
Factor the matcher into `fn contentTypeOk(value: ?[]const u8) bool` so it is testable without a
|
||
server.
|
||
- `src/upstream/doh_client_live_test.zig`, guarded by `build_options.live`: POST an A query for
|
||
`example.com` to `https://cloudflare-dns.com/dns-query`, assert the reply validates and carries at
|
||
least one answer record. Whole exchange raced against a 10 s budget with `std.Io.Select`, the
|
||
pattern in `src/platform/tls_client_integration_test.zig`. This host's IPv6 egress is broken (API
|
||
notes) and DoH needs name resolution — if the test stalls on AAAA, report it as an environment
|
||
finding rather than working around it; the test is non-blocking by construction.
|
||
|
||
### S2.4 Acceptance criteria
|
||
|
||
- [ ] `zig test src/upstream/doh_client.zig` passes (unit tests, no network).
|
||
- [ ] `DohClient` satisfies `transport.Client` (a compile-time test takes `client()` and calls
|
||
`exchange` through the interface against a `std.http.Client` that is never driven — verify by
|
||
instantiation, not by running an exchange).
|
||
- [ ] The live test compiles in every build and skips without `-Dlive`.
|
||
- [ ] `zig fmt --check` clean.
|
||
|
||
---
|
||
|
||
## Session S3: `src/upstream/dot_client.zig` (+ live test)
|
||
|
||
DoT per RFC 7858: TLS on port 853, DNS messages framed exactly as over TCP (RFC 1035 §4.2.2).
|
||
|
||
### S3.1 API
|
||
|
||
```zig
|
||
pub const DotClient = struct {
|
||
endpoint: transport.Endpoint,
|
||
gpa: std.mem.Allocator,
|
||
bundle: *std.crypto.Certificate.Bundle, // caller-owned, shared
|
||
bundle_lock: *std.Io.RwLock, // caller-owned, shared
|
||
buffers: Buffers, // caller-owned; one DotClient is used by one task
|
||
|
||
pub const Buffers = struct {
|
||
tls_read: []u8, // >= std.crypto.tls.Client.min_buffer_len
|
||
tls_write: []u8, // >= std.crypto.tls.Client.min_buffer_len
|
||
stream_read: []u8, // >= std.crypto.tls.Client.min_buffer_len
|
||
stream_write: []u8, // >= std.crypto.tls.Client.min_buffer_len
|
||
};
|
||
|
||
pub fn init(
|
||
endpoint: transport.Endpoint,
|
||
gpa: std.mem.Allocator,
|
||
bundle: *std.crypto.Certificate.Bundle,
|
||
bundle_lock: *std.Io.RwLock,
|
||
buffers: Buffers,
|
||
) DotClient; // asserts endpoint.scheme == .dot and every buffer length
|
||
|
||
pub fn client(self: *DotClient) transport.Client;
|
||
};
|
||
```
|
||
|
||
### S3.2 Exchange
|
||
|
||
One TCP connection and one TLS handshake per exchange, closed before returning. Connection reuse is
|
||
not built: at household query rates the pool's per-attempt budget and the failover path stay simple
|
||
and every failure is attributable to one exchange, which is worth more here than the saved round
|
||
trips. Say that in a comment — it is a decision, not a stub.
|
||
|
||
1. Resolve the endpoint to an `std.Io.net.IpAddress` with `net.IpAddress.parse(endpoint.host,
|
||
endpoint.port)`. **A non-literal host is `error.ConnectFailed`** with a log line naming the host:
|
||
name resolution for upstreams is not in this milestone's scope, and a silent fallback would hide
|
||
it. (`tls://dns.google:853` from PLAN §12.1 therefore needs an IP literal in config; note this in
|
||
the completion report so the orchestrator can carry it into the Phase 4 config validator.)
|
||
2. `var stream = try address.connect(io, .{ .mode = .stream });` — no `timeout` field, ever.
|
||
`defer` a close that runs under cancel protection (below).
|
||
3. `var tls_stream: tls_client.TlsStream = undefined;` then `tls_stream.init(io, &stream, bundle,
|
||
bundle_lock, gpa, .{ .host = endpoint.host, .ca = .system, … })`. `TlsStream` is pinned — do not
|
||
copy or move it after init.
|
||
4. Write `[2]u8` big-endian length then `query` to `tls_stream.writer()`, then `flush()`.
|
||
5. Read exactly 2 bytes, then that many bytes, from `tls_stream.reader()`. `len == 0` →
|
||
`error.BadResponse`. `len > response_buf.len` → `error.ResponseTooLarge`. A short read or EOF →
|
||
`error.ReceiveFailed`.
|
||
6. `try transport.validateResponse(query, response_buf[0..len]);`
|
||
7. `tls_stream.close();` then `stream.close(io);`.
|
||
|
||
Cleanup under cancellation: the pool cancels this task on timeout, and the next cancelable `Io` call
|
||
in the `defer` chain would return `error.Canceled` and skip the socket close. Wrap the close path in
|
||
`const prev = io.swapCancelProtection(.blocked); defer _ = io.swapCancelProtection(prev);`
|
||
(verified: `std/Io.zig:1342`) so the socket is always released.
|
||
|
||
Error mapping: `connect` → `error.ConnectFailed`; `TlsStream.init` → `error.TlsFailed` (log
|
||
`tls_client.classify(err)` alongside — that is what milestone 1 built it for); write/flush →
|
||
`error.SendFailed`; read → `error.ReceiveFailed`; `transport.mapLocal` first at every site.
|
||
|
||
### S3.3 Tests
|
||
|
||
- In-file unit: `framePrefix(len: u16) [2]u8` and `parsePrefix(bytes: [2]u8) u16` round-trip,
|
||
including 0 and 65535, big-endian byte order asserted explicitly; `init` asserts on a `.doh`
|
||
endpoint; a non-literal host maps to `error.ConnectFailed` without touching the network (call the
|
||
address-resolution helper directly).
|
||
- `src/upstream/dot_client_live_test.zig`, guarded by `build_options.live`: exchange an A query for
|
||
`example.com` against `tls://1.1.1.1:853` with `endpoint.host` overridden to `"1.1.1.1"` — this
|
||
machine's IPv6 egress is dead, so the documented anycast literal is used, and Cloudflare's
|
||
certificate carries `1.1.1.1` as an IP SAN. If the stdlib verifier rejects an IP-literal SAN, do
|
||
NOT weaken verification: report the finding. Raced against a 10 s budget with `std.Io.Select`.
|
||
|
||
### S3.4 Acceptance criteria
|
||
|
||
- [ ] `zig test src/upstream/dot_client.zig` passes (unit tests, no network).
|
||
- [ ] `DotClient` satisfies `transport.Client`.
|
||
- [ ] The live test compiles in every build and skips without `-Dlive`.
|
||
- [ ] No direct `std.crypto.tls.Client` use — everything goes through `platform/tls_client.zig`.
|
||
- [ ] `zig fmt --check` clean.
|
||
|
||
---
|
||
|
||
## Session S4: `src/upstream/pool.zig`
|
||
|
||
Priority-ordered sequential failover with per-attempt deadline, health tracking and backoff
|
||
(PLAN §9). The pool is itself a `transport.Client`, so the handler sees one interface.
|
||
|
||
### S4.1 API
|
||
|
||
```zig
|
||
pub const Entry = struct {
|
||
endpoint: transport.Endpoint,
|
||
client: transport.Client,
|
||
priority: i32, // lower first (PLAN §11.2 `upstreams.priority`)
|
||
enabled: bool,
|
||
health: health.State,
|
||
};
|
||
|
||
pub const Snapshot = struct {
|
||
url: []const u8,
|
||
enabled: bool,
|
||
available: bool,
|
||
consecutive_failures: u32,
|
||
total_successes: u64,
|
||
total_failures: u64,
|
||
success_rate: f32,
|
||
last_success_at: ?std.Io.Timestamp,
|
||
last_error_at: ?std.Io.Timestamp,
|
||
last_error: []const u8, // borrowed from the entry; valid until the next failure
|
||
backoff_until: ?std.Io.Timestamp,
|
||
};
|
||
|
||
pub const Pool = struct {
|
||
entries: []Entry, // caller-owned, sorted ascending by priority in `init`
|
||
cfg: health.Config,
|
||
attempt_timeout: std.Io.Clock.Duration,
|
||
mutex: std.Io.Mutex,
|
||
rng: std.Random.DefaultPrng,
|
||
|
||
pub fn init(
|
||
entries: []Entry,
|
||
cfg: health.Config,
|
||
attempt_timeout: std.Io.Clock.Duration,
|
||
seed: u64,
|
||
) Pool; // asserts entries.len > 0
|
||
|
||
pub fn client(self: *Pool) transport.Client;
|
||
|
||
pub fn exchange(
|
||
self: *Pool,
|
||
io: std.Io,
|
||
query: []const u8,
|
||
response_buf: []u8,
|
||
) transport.ExchangeError![]u8;
|
||
|
||
/// Copies health into `out` in pool order; returns the number written.
|
||
/// Feeds `GET /api/upstream/health` in Phase 8.
|
||
pub fn snapshot(self: *Pool, io: std.Io, out: []Snapshot) std.Io.Cancelable!usize;
|
||
};
|
||
```
|
||
|
||
`attempt_timeout` uses the `.awake` clock (`.{ .raw = .fromMilliseconds(2000), .clock = .awake }`) so
|
||
a suspended Pi does not burn the budget.
|
||
|
||
### S4.2 Failover algorithm
|
||
|
||
1. Take `now = std.Io.Clock.awake.now(io)`.
|
||
2. **Pass one**: walk `entries` in order; skip `!enabled`; skip entries whose
|
||
`health.available(now)` is false.
|
||
3. **Pass two**, only if pass one attempted nothing: walk again, skipping only `!enabled`. Every
|
||
endpoint being in backoff must not turn into `SERVFAIL` for every client — a probe is better than
|
||
a guaranteed failure, and it is how backoff recovers. Test this explicitly.
|
||
4. For each candidate, run `attempt`:
|
||
- Race `entry.client.exchange(io, query, response_buf)` against
|
||
`attempt_timeout.sleep(io)` using `std.Io.Select` with a two-arm union, then
|
||
`race.cancelDiscard()` on the way out (the milestone-1 pattern). The sleep arm winning yields
|
||
`error.Timeout`.
|
||
- On success: lock the mutex, `entry.health.recordSuccess(completed_at)`, unlock, return the
|
||
slice.
|
||
- On error, switch on `transport.group(err)`:
|
||
- `.peer_fault` → lock, `recordFailure(completed_at, @errorName(err), self.cfg, self.rng.random().int(u32))`,
|
||
unlock, remember it as `last_fault`, continue to the next candidate.
|
||
- `.local_resource` → return the error immediately. Do not record. Do not try another upstream:
|
||
the next one will hit the same wall.
|
||
- `.cancellation` → return `error.Canceled` immediately. Do not record.
|
||
- `completed_at` is `std.Io.Clock.awake.now(io)` taken after the attempt returns.
|
||
5. All candidates exhausted → return `last_fault` (guaranteed non-null: `entries.len > 0` and pass
|
||
two attempts every enabled entry). If every entry is disabled, return `error.ConnectFailed`.
|
||
|
||
`response_buf` is handed to each attempt in turn; a failed attempt may have written into it, so the
|
||
returned slice is only meaningful on success. Note this in a comment.
|
||
|
||
Concurrency: several handler tasks share one `Pool`. The mutex guards health mutation and
|
||
`snapshot` only — never an in-flight exchange, so a slow upstream cannot block bookkeeping. Entry
|
||
`client` values must therefore be safe for concurrent `exchange` calls; `DohClient` and `DotClient`
|
||
own per-instance buffers, so **one `Entry` per concurrent task, or one shared pool with per-entry
|
||
serialization, is a wiring decision for the orchestrator**. Record in the completion report which
|
||
one this session assumed (the specified `Entry` layout assumes each entry's client is used by one
|
||
task at a time; the e2e test in S7 exercises exactly one in-flight query at a time).
|
||
|
||
### S4.3 Tests (in-file, no sockets — fake clients implement `transport.Client`)
|
||
|
||
- Priority order: entry with priority 10 is tried before priority 100.
|
||
- Failover: first fake returns `error.Timeout`, second returns a valid response → the pool returns
|
||
the second one's bytes; first entry's `consecutive_failures == 1`, second's `total_successes == 1`.
|
||
- Backoff skip: drive a fake past `failure_threshold`, then assert the next `exchange` does not call
|
||
that fake (a call counter in the fake) while another entry is available.
|
||
- All-backed-off probe: every entry in backoff → pass two still calls them.
|
||
- `.local_resource` short-circuit: a fake returning `error.OutOfMemory` makes `exchange` return
|
||
`error.OutOfMemory`, the second fake is never called, and the first entry's `total_failures`
|
||
stays 0.
|
||
- `.cancellation` short-circuit: a fake returning `error.Canceled` returns `error.Canceled` and
|
||
records nothing.
|
||
- Attempt timeout: a fake that sleeps longer than `attempt_timeout` yields a recorded
|
||
`error.Timeout` peer fault and moves on. (Needs a live `Io` — use `std.Io.Threaded` inside the
|
||
test; no sockets are involved, so this stays in the default `zig build test`.)
|
||
- All entries disabled → `error.ConnectFailed`.
|
||
- `snapshot` reports the counters set by the preceding cases.
|
||
|
||
### S4.4 Acceptance criteria
|
||
|
||
- [ ] `zig test src/upstream/pool.zig` passes, including the timeout test on a `Threaded` backend.
|
||
- [ ] Every bullet in S4.3 exists as a named test.
|
||
- [ ] `Pool` satisfies `transport.Client`.
|
||
- [ ] `zig fmt --check` clean.
|
||
|
||
---
|
||
|
||
## Session S5: `src/server/handler.zig`
|
||
|
||
The serving handler: bytes in from a listener, validated bytes out. It owns no socket and no clock
|
||
beyond what it passes to the upstream. Phase 7 extends this file with filtering, cache, local
|
||
records and logging — nothing of that appears here.
|
||
|
||
### S5.1 API
|
||
|
||
```zig
|
||
pub const Transport = enum { udp, tcp };
|
||
|
||
pub const Handler = struct {
|
||
upstream: transport.Client, // in production `pool.client()`
|
||
stats: Stats = .{},
|
||
|
||
pub const Stats = struct {
|
||
queries: std.atomic.Value(u64) = .init(0),
|
||
dropped_malformed: std.atomic.Value(u64) = .init(0),
|
||
formerr: std.atomic.Value(u64) = .init(0),
|
||
notimp: std.atomic.Value(u64) = .init(0),
|
||
servfail: std.atomic.Value(u64) = .init(0),
|
||
truncated: std.atomic.Value(u64) = .init(0),
|
||
};
|
||
|
||
pub const Outcome = union(enum) {
|
||
reply: []u8, // a prefix of `response_buf`
|
||
drop, // no response is possible or appropriate
|
||
};
|
||
|
||
/// `response_buf.len` must be >= 512 and <= 65535 (ResponseBuilder asserts the upper bound).
|
||
pub fn handle(
|
||
self: *Handler,
|
||
io: std.Io,
|
||
which: Transport,
|
||
query: []const u8,
|
||
response_buf: []u8,
|
||
) Outcome;
|
||
};
|
||
```
|
||
|
||
`handle` never returns an error: every failure is either a DNS response or a counted drop. That is
|
||
the "every failure mode visible" rule applied to the hot path.
|
||
|
||
### S5.2 Behavior
|
||
|
||
1. `header.parse(query)` fails (< 12 bytes) → `dropped_malformed`, `.drop` (PLAN §6.1: severely
|
||
truncated is a silent drop).
|
||
2. `hdr.flags.qr == true` → a response arriving on a listener port; `dropped_malformed`, `.drop`.
|
||
3. `packet.parse(query)`:
|
||
- `error.Truncated` → `dropped_malformed`, `.drop`.
|
||
- any other `WalkError` → FORMERR reply built from `hdr` with **no** echoed question (the
|
||
question is what failed to parse); `formerr`.
|
||
4. `hdr.flags.opcode != .query` → NOTIMP reply echoing nothing; `notimp`.
|
||
5. `hdr.qdcount != 1` → FORMERR (RFC 9619 makes any other value invalid, in both directions);
|
||
`formerr`.
|
||
6. Forward: `self.upstream.exchange(io, query, response_buf)`.
|
||
- success → `queries` incremented; then the UDP size check in §5.3; `.reply`.
|
||
- `transport.group(err) == .cancellation` → `.drop` (the process is shutting down; nothing to
|
||
say).
|
||
- `.peer_fault` or `.local_resource` → SERVFAIL reply echoing the question; `servfail`.
|
||
7. Every synthesized reply preserves the request ID, opcode and RD bit, sets QR and RA, and echoes
|
||
the question when one parsed — all of that is already `packet.ResponseBuilder.init`'s contract.
|
||
When the query carried an OPT record (`packet.findOptRecord` + `edns.parseOpt`), the reply ends
|
||
with `addOptEcho(opt, opt.do_bit)` so the DO bit passes through (PLAN §6.1).
|
||
|
||
### S5.3 UDP size limit and truncation
|
||
|
||
The upstream answer can exceed what the client will accept over UDP.
|
||
|
||
- `pub fn udpLimit(query_packet: packet.Packet) u16` — the query's OPT `udp_payload_size` clamped to
|
||
`[512, 4096]`, or 512 when there is no OPT record (RFC 1035 §4.2.1 / RFC 6891 §6.2.3).
|
||
- `which == .udp` and the upstream reply is longer than `udpLimit` → build a reply with `TC = 1`,
|
||
RCODE NOERROR, the question echoed, no answer records, OPT echoed when present; increment
|
||
`truncated`. The client then retries over TCP, which is exactly what RFC 1035 §4.2.1 prescribes.
|
||
- `which == .tcp` → no size limit beyond `response_buf`.
|
||
|
||
The upstream reply lands in `response_buf`; building a truncated reply over it needs a second
|
||
buffer. Give `handle` no extra parameter: build the truncated reply in a stack `[512]u8` inside
|
||
`handle` and copy it into `response_buf` before returning. 512 is enough — the message is a header,
|
||
one question and at most one OPT record, and a question longer than ~270 bytes cannot exist
|
||
(name ≤ 255).
|
||
|
||
### S5.4 Tests (in-file, fake upstream implementing `transport.Client`)
|
||
|
||
Build queries with hand-encoded bytes or `packet.ResponseBuilder`; the fake upstream answers by
|
||
copying a fixture and calling `packet.setId`.
|
||
|
||
- A query for `example.com` A over UDP returns the fake's bytes unchanged.
|
||
- A response (QR set) sent to the handler → `.drop`, `dropped_malformed == 1`.
|
||
- An 8-byte query → `.drop`.
|
||
- A query with QDCOUNT 0 → FORMERR reply, question absent, ID preserved.
|
||
- A query with QDCOUNT 2 → FORMERR.
|
||
- A structurally broken question (name pointer loop) → FORMERR, `formerr == 1`.
|
||
- Opcode `update` → NOTIMP.
|
||
- Fake returns `error.Timeout` → SERVFAIL with the question echoed and RD preserved.
|
||
- Fake returns `error.OutOfMemory` → SERVFAIL (not a drop).
|
||
- Fake returns `error.Canceled` → `.drop`.
|
||
- `udpLimit`: no OPT → 512; OPT 1232 → 1232; OPT 200 → 512; OPT 9000 → 4096.
|
||
- Over UDP, a fake reply of 900 bytes with no OPT in the query → TC=1 reply, ancount 0, question
|
||
echoed, `truncated == 1`; the same reply over TCP passes through untouched.
|
||
- DO bit set in the query's OPT → set in the SERVFAIL reply's OPT.
|
||
- Every synthesized reply re-parses with `packet.parse`.
|
||
|
||
### S5.5 Acceptance criteria
|
||
|
||
- [ ] `zig test src/server/handler.zig` passes.
|
||
- [ ] Every bullet in S5.4 exists as a named test.
|
||
- [ ] `handle` has no error union in its return type.
|
||
- [ ] `zig fmt --check` clean.
|
||
|
||
---
|
||
|
||
## Session S6: `src/server/udp_server.zig`, `src/server/tcp_server.zig`
|
||
|
||
Listeners. Both take a `*handler.Handler`, both use a bounded pool of in-flight tasks so one slow
|
||
upstream cannot stall the listener, and both count every failure.
|
||
|
||
### S6.1 udp_server.zig
|
||
|
||
```zig
|
||
pub const max_datagram = 4096; // EDNS ceiling this server advertises
|
||
|
||
pub const Options = struct {
|
||
max_in_flight: u16 = 64,
|
||
};
|
||
|
||
pub const Stats = struct {
|
||
received: std.atomic.Value(u64) = .init(0),
|
||
dropped_oversize: std.atomic.Value(u64) = .init(0), // datagram arrived with flags.trunc
|
||
dropped_no_slot: std.atomic.Value(u64) = .init(0),
|
||
receive_errors: std.atomic.Value(u64) = .init(0),
|
||
send_errors: std.atomic.Value(u64) = .init(0),
|
||
};
|
||
|
||
pub const UdpServer = struct {
|
||
socket: std.Io.net.Socket,
|
||
handler: *handler.Handler,
|
||
slots: []Slot, // allocated in `bind`, freed in `deinit`
|
||
free: std.DynamicBitSetUnmanaged or a u64 free-list — implementer's call, guarded by `mutex`
|
||
mutex: std.Io.Mutex,
|
||
stats: Stats,
|
||
|
||
pub const Slot = struct {
|
||
query: [max_datagram]u8,
|
||
reply: [max_datagram]u8,
|
||
from: std.Io.net.IpAddress,
|
||
len: usize,
|
||
};
|
||
|
||
pub const BindError = std.Io.net.IpAddress.BindError || error{OutOfMemory};
|
||
|
||
pub fn bind(
|
||
gpa: std.mem.Allocator,
|
||
io: std.Io,
|
||
address: std.Io.net.IpAddress,
|
||
h: *handler.Handler,
|
||
options: Options,
|
||
) BindError!UdpServer;
|
||
|
||
/// The kernel-assigned address; port 0 in `bind` resolves here (needed by tests).
|
||
pub fn boundAddress(self: *const UdpServer) std.Io.net.IpAddress;
|
||
|
||
/// Receive loop. Returns when the task is canceled or the socket is closed.
|
||
pub fn serve(self: *UdpServer, io: std.Io) void;
|
||
|
||
pub fn deinit(self: *UdpServer, gpa: std.mem.Allocator, io: std.Io) void;
|
||
};
|
||
```
|
||
|
||
Loop:
|
||
1. `const msg = self.socket.receive(io, slot_query_buf)` into a claimed slot. `error.Canceled` →
|
||
return. Any other error → `receive_errors`, log, continue (a per-datagram error must not kill the
|
||
listener).
|
||
2. `msg.flags.trunc` → `dropped_oversize`, release the slot, continue. A truncated datagram cannot
|
||
be parsed reliably and answering it would be a guess.
|
||
3. No free slot → `dropped_no_slot`, continue. Do not queue: an unbounded queue is the cloudflared
|
||
failure mode (PLAN §1) in a different costume.
|
||
4. `group.concurrent(io, respondOne, .{ self, io, slot_index })` — an `std.Io.Group` owned by
|
||
`serve`. `respondOne` calls `self.handler.handle(io, .udp, query, reply)`, sends `.reply` with
|
||
`self.socket.send(io, &slot.from, bytes)` (a send failure increments `send_errors` and is logged
|
||
at debug, deduplicated by the caller later), then releases the slot.
|
||
5. `serve` awaits the group before returning, under cancel protection so in-flight replies are not
|
||
abandoned mid-send.
|
||
|
||
`deinit` closes the socket and frees the slots. Closing the socket makes a blocked `receive` fail;
|
||
the loop treats that as a stop signal.
|
||
|
||
### S6.2 tcp_server.zig
|
||
|
||
```zig
|
||
pub const Options = struct {
|
||
max_connections: u16 = 64,
|
||
/// RFC 7766 §6.2.3 recommends a few seconds of idle tolerance.
|
||
idle_timeout: std.Io.Clock.Duration = .{ .raw = .fromSeconds(10), .clock = .awake },
|
||
};
|
||
|
||
pub const Stats = struct {
|
||
accepted: std.atomic.Value(u64) = .init(0),
|
||
rejected_at_capacity: std.atomic.Value(u64) = .init(0),
|
||
accept_errors: std.atomic.Value(u64) = .init(0),
|
||
connection_errors: std.atomic.Value(u64) = .init(0),
|
||
idle_timeouts: std.atomic.Value(u64) = .init(0),
|
||
};
|
||
|
||
pub const TcpServer = struct {
|
||
server: std.Io.net.Server,
|
||
handler: *handler.Handler,
|
||
conns: []Conn, // fixed slots, same bounded-capacity rule as UDP
|
||
mutex: std.Io.Mutex,
|
||
stats: Stats,
|
||
|
||
pub const ListenError = std.Io.net.IpAddress.ListenError || error{OutOfMemory};
|
||
|
||
pub fn listen(gpa, io, address, h: *handler.Handler, options: Options) ListenError!TcpServer;
|
||
pub fn boundAddress(self: *const TcpServer) std.Io.net.IpAddress;
|
||
pub fn serve(self: *TcpServer, io: std.Io) void;
|
||
pub fn deinit(self: *TcpServer, gpa: std.mem.Allocator, io: std.Io) void;
|
||
};
|
||
```
|
||
|
||
Accept loop: `self.server.accept(io)`; `error.Canceled` or `error.SocketNotListening` → return
|
||
(`deinit` shuts the listening socket down, which is the documented way to unblock a pending accept —
|
||
API notes). At capacity → close the stream immediately, `rejected_at_capacity`. Otherwise a group
|
||
task per connection.
|
||
|
||
Per connection (RFC 7766 §6.2.1.1 — a connection may carry several queries, handled serially):
|
||
1. Read the 2-byte length prefix. The read is raced against `options.idle_timeout` with
|
||
`std.Io.Select`; the sleep arm winning means `idle_timeouts` and a clean close.
|
||
2. `len == 0` → close (`connection_errors`). `len > transport.max_message_len` is impossible (u16).
|
||
3. Read exactly `len` bytes; a short read → close, `connection_errors`.
|
||
4. `handler.handle(io, .tcp, query, reply)`; `.drop` closes the connection (there is nothing to
|
||
frame); `.reply` writes the 2-byte prefix plus the bytes and flushes.
|
||
5. Loop back to step 1.
|
||
6. Close under cancel protection and release the slot.
|
||
|
||
Buffers: each `Conn` owns `query: [transport.max_message_len]u8` and
|
||
`reply: [transport.max_message_len]u8` plus the stream read/write buffers. That is 128 KiB+ per
|
||
connection slot; with the default 64 slots that is ~8 MiB, inside the PLAN §18 budget. If a session
|
||
finds a materially smaller layout that keeps the 65535-byte ceiling, take it and say so.
|
||
|
||
### S6.3 Tests
|
||
|
||
Unit tests that need no socket go in-file (length-prefix framing helper, capacity accounting).
|
||
The loopback tests live in `src/server/udp_server_integration_test.zig` and
|
||
`src/server/tcp_server_integration_test.zig`, both guarded by
|
||
`if (!build_options.integration) return error.SkipZigTest;`:
|
||
|
||
- UDP: bind `127.0.0.1:0`, run `serve` in an `Io.Group` task, send a query from a second bound
|
||
socket with a fake-upstream handler, assert the reply's ID and question match; then send a 5-byte
|
||
datagram and assert no reply arrives within a short budget and `dropped_malformed` moved.
|
||
- TCP: listen on `127.0.0.1:0`, connect, send two length-prefixed queries on one connection, read
|
||
two length-prefixed replies, assert both match, then close; assert `accepted == 1`.
|
||
- TCP idle timeout: connect, send nothing, assert the server closes the connection and
|
||
`idle_timeouts == 1` (use a short `idle_timeout` in the test options).
|
||
- Both: `deinit` while `serve` is blocked ends `serve` without a hang — the group awaits cleanly.
|
||
|
||
### S6.4 Acceptance criteria
|
||
|
||
- [ ] `zig test src/server/udp_server.zig` and `.../tcp_server.zig` pass (unit tests).
|
||
- [ ] Both integration files compile in every build, skip without `-Dintegration`, and pass with it.
|
||
- [ ] No unbounded queue, no unbounded allocation per datagram or per connection.
|
||
- [ ] `zig fmt --check` clean.
|
||
|
||
---
|
||
|
||
## Session S7: end-to-end integration test
|
||
|
||
`src/server/resolver_integration_test.zig`, guarded by
|
||
`if (!build_options.integration) return error.SkipZigTest;`. Hermetic: loopback sockets and an
|
||
in-process fake upstream, no external network.
|
||
|
||
### S7.1 Fake upstream
|
||
|
||
In this file: a struct implementing `transport.Client` that parses the incoming query, echoes the
|
||
question and appends one A record (`93.184.216.34`, TTL 300) with `packet.ResponseBuilder`, and
|
||
counts calls. A second variant returns a configured `transport.PeerFault` on the first N calls.
|
||
|
||
### S7.2 The test
|
||
|
||
1. `std.Io.Threaded` backend, `io`.
|
||
2. Two `Pool` entries: entry 0 = always-fails fake (priority 10), entry 1 = good fake (priority 20).
|
||
3. `Handler` over `pool.client()`.
|
||
4. `UdpServer` bound to `127.0.0.1:0` and `TcpServer` on `127.0.0.1:0`, both served in one
|
||
`Io.Group`.
|
||
5. Client side: send an A query for `example.com` over UDP; assert the reply parses, ID matches,
|
||
RCODE is NOERROR, ancount is 1, and the A record's rdata is `93.184.216.34`.
|
||
6. Send the same query length-prefixed over TCP; assert the same.
|
||
7. Assert failover happened: entry 0's `consecutive_failures >= 1` and `backoff_until != null`;
|
||
entry 1's `total_successes >= 2`. Read them through `pool.snapshot`.
|
||
8. Send a third query and assert entry 0's call counter did not grow — it is in backoff.
|
||
9. `deinit` both servers, cancel and await the group, `threaded.deinit()` — the test must not hang.
|
||
|
||
### S7.3 Acceptance criteria
|
||
|
||
- [ ] `zig build test -Dintegration` runs this test and it passes.
|
||
- [ ] It skips (not fails) under plain `zig build test`.
|
||
- [ ] It opens no socket outside `127.0.0.1` and reaches no external host.
|
||
- [ ] It completes in bounded time — every wait carries a budget; no unbounded blocking read.
|
||
- [ ] `zig fmt --check` clean.
|
||
|
||
---
|
||
|
||
## Module Layout
|
||
|
||
```
|
||
src/upstream/transport.zig S1 Endpoint, error groups, Client, validateResponse
|
||
src/upstream/health.zig S1 pure health/backoff state
|
||
src/upstream/doh_client.zig S2 RFC 8484 over std.http.Client
|
||
src/upstream/doh_client_live_test.zig S2 -Dlive
|
||
src/upstream/dot_client.zig S3 RFC 7858 over platform/tls_client.zig
|
||
src/upstream/dot_client_live_test.zig S3 -Dlive
|
||
src/upstream/pool.zig S4 priority failover, per-attempt deadline, health
|
||
src/server/handler.zig S5 query -> upstream -> validated reply
|
||
src/server/udp_server.zig S6 UDP/53 listener
|
||
src/server/tcp_server.zig S6 TCP/53 listener, length-prefixed
|
||
src/server/udp_server_integration_test.zig S6 -Dintegration
|
||
src/server/tcp_server_integration_test.zig S6 -Dintegration
|
||
src/server/resolver_integration_test.zig S7 -Dintegration, end to end
|
||
```
|
||
|
||
## File Ownership
|
||
|
||
| Files | Owner | Notes |
|
||
|---|---|---|
|
||
| `src/upstream/transport.zig`, `src/upstream/health.zig` | S1 | frozen after S1 verifies |
|
||
| `src/upstream/doh_client.zig`, `src/upstream/doh_client_live_test.zig` | S2 | |
|
||
| `src/upstream/dot_client.zig`, `src/upstream/dot_client_live_test.zig` | S3 | |
|
||
| `src/upstream/pool.zig` | S4 | |
|
||
| `src/server/handler.zig` | S5 | |
|
||
| `src/server/udp_server.zig`, `src/server/tcp_server.zig`, both `*_server_integration_test.zig` | S6 | |
|
||
| `src/server/resolver_integration_test.zig` | S7 | |
|
||
| `build.zig`, `src/tests.zig` | orchestrator | no session edits these |
|
||
|
||
No session touches milestone 1 or milestone 2 files. A needed change there is reported, not made.
|
||
|
||
## Acceptance Criteria (Milestone 3 Complete)
|
||
|
||
- [ ] `zig build test` exits 0 with every new file wired into `src/tests.zig`.
|
||
- [ ] `zig build test -Dintegration` exits 0: the milestone-1 loopback TLS echo, both listener
|
||
integration tests, and the end-to-end resolver test all pass.
|
||
- [ ] `zig build test -Dintegration -Dlive` exits 0 locally, or the DoH/DoT live failures are
|
||
reported as environment findings with the exact error (a live-network failure is not a gate).
|
||
- [ ] `zig build cross` still produces two statically linked executables.
|
||
- [ ] `transport.group` is exhaustive over `ExchangeError` with no `else` arm.
|
||
- [ ] `pool.zig`'s failover, backoff-skip, all-backed-off probe, local-resource short-circuit and
|
||
cancellation short-circuit each have a passing named test.
|
||
- [ ] `handler.zig`'s FORMERR (QDCOUNT 0 and 2), NOTIMP, SERVFAIL, drop and UDP-truncation paths
|
||
each have a passing named test.
|
||
- [ ] `zig fmt --check` clean repo-wide; GPG-signed lowercase commits.
|
||
|
||
## Anti-Requirements
|
||
|
||
- No caching of any kind (Phase 6). The handler forwards every query.
|
||
- No filtering, blocklists, rules, or blocked-response synthesis (Phase 5).
|
||
- No query logging, no SQLite, no storage, no config file or ZON parsing (Phase 4). Timeouts,
|
||
ports and pool entries arrive as parameters or named constants.
|
||
- No rate limiting (Phase 6) — `src/server/rate_limiter.zig` is not created here.
|
||
- No local DoH or DoT **server** endpoints (Phase 9). Only upstream DoH/DoT clients are in scope;
|
||
`platform/tls_server.zig` is untouched.
|
||
- No local records, no conditional forwarding, no CNAME uncloaking (Phase 5/7).
|
||
- No web UI, no REST API, no `/metrics`. `Pool.snapshot` exists so Phase 8 has something to read;
|
||
it is not exposed anywhere yet.
|
||
- No signal handling, no `src/server/shutdown.zig`, no `main.zig` wiring. Servers stop through
|
||
`deinit` plus group cancellation; process lifetime is Phase 7's problem.
|
||
- No DNS name resolution for upstream hosts — DoT endpoints take IP literals, DoH resolution is
|
||
`std.http.Client`'s business.
|
||
- No HTTP/2, no DoQ, no connection reuse for DoT, no DNSSEC validation (DO bit passes through).
|
||
- No 0x20 query-name randomization. Case-insensitive comparison per RFC 4343 is required; generating
|
||
mixed-case queries is not.
|
||
- No changes to `src/dns/`. If a helper is missing there, report it — do not add transport-aware
|
||
code to the pure core.
|
||
|
||
## As built
|
||
|
||
The implementation matches the spec with these review-driven refinements (three
|
||
review rounds; findings went 9 → 3 → 0):
|
||
|
||
- `pool.zig`: each `Entry` carries a `busy: std.Io.Mutex` held for the whole
|
||
attempt against that entry, so one entry's client (and its buffers/TLS state)
|
||
is never used by two tasks at once. Pass one re-checks `available` with a
|
||
fresh `now` after acquiring `busy` (the wait can outlive the health it was
|
||
admitted under); pass two probes without a re-check by design. Lock order is
|
||
always `busy` then `Pool.mutex`; the pool mutex still guards health and
|
||
`snapshot` only.
|
||
- `health.zig`: staleness rules are symmetric. The newest outcome is the @max
|
||
of `last_success_at`/`last_error_at`. A stale success counts into
|
||
totals/window and @max-updates `last_success_at`, but does not clear
|
||
`consecutive_failures`/`backoff_until`. A stale failure counts into
|
||
totals/window and @max-updates `last_error_at`, but only increments
|
||
`consecutive_failures`/extends backoff when no newer success exists, and
|
||
never overwrites a newer failure's error text.
|
||
- `udp_server.zig`: each slot's reply buffer is
|
||
`[transport.max_message_len]u8` (not `max_datagram`) so an oversized upstream
|
||
reply reaches the handler and becomes a TC=1 reply instead of SERVFAIL
|
||
(64 slots ≈ 4.4 MiB, inside the PLAN §18 budget). A `dropped_handler` stat
|
||
counts `.drop` outcomes from the handler.
|
||
- `tcp_server.zig`: a shutdown flag, set under the slot mutex before the
|
||
active-connection scan, closes the accept-vs-deinit race — a stream accepted
|
||
during shutdown is closed immediately instead of claimed.
|
||
- `handler.zig`: OPT validation walks every record section. OPT in answer or
|
||
authority, more than one OPT, a non-root OPT owner, or an unparseable OPT →
|
||
FORMERR (RFC 6891); only a genuinely absent OPT takes the no-OPT path.
|
||
- `dot_client.zig`: handshake `ReadFailed`/`WriteFailed` unwrap the stream's
|
||
stored cause through `transport.mapLocal` first, and certificate-bundle
|
||
`OutOfMemory` is a local resource, not a peer fault.
|
||
- `transport.zig`: `Endpoint.parse` rejects userinfo/query/fragment delimiters
|
||
(`@`, `?`, `#`) in the authority, and `?`/`#` in a DoH path.
|