diff --git a/docs/semantic-parity.md b/docs/semantic-parity.md index 9243cb5..d04b43b 100644 --- a/docs/semantic-parity.md +++ b/docs/semantic-parity.md @@ -14,7 +14,7 @@ this document or `bootstrap-semantic-parity.md` is open. |---|---|---| | Bootstrap, merge, retry | Detailed row-by-row contract and adversity receipt | **closed at the current implementation boundary.** Every row in `bootstrap-semantic-parity.md` is closed; the offline candidate rerun survived all eight lifecycle crash seams. The experiment gate still requires rerunning that oracle against the exact eventual build. | | JSS v1 storage | Upstream-produced fixture, reciprocal segment parsing, checksum/index/gloom vectors | closed for sealed format; rerun reciprocal corpus before experiment | -| Archive XRPC | Official Go client plus Stream conformance replay over listSegments/getSegment/getBlock/planBackfill | **closed.** With the pinned Zig dependencies cached, `just archive-contract` seeds three real sealed JSS segments and runs without network access. The pinned Go client consumed 5,000 ordered whole-segment rows, the exact 2,500-row DID block plan, 50 required identity sentinels for an empty collection match, the 234-row `(1000,1234]` bound, and a duplicate-free archive-to-live cutover at cursor 5000. Direct conformance also proves checksums/ETags, byte-identical segments, raw block decode, error names, DID/collection blooms, plan modes, and the pinned `http.ServeContent` behavior for both download endpoints. Single, suffix, multipart, mixed-overlap, aggregate-size fallback, and the distinct malformed/non-overlap 416 responses cross the real production server. The same receipt proves descriptor- or manifest-derived Last-Modified values, Go's If-Match/If-Unmodified-Since/If-None-Match/If-Modified-Since precedence, strong ETag and exact-date If-Range, and 304/412 representation-header behavior. The date parser accepts all three forms supported by Go's `http.ParseTime`. Sealed reads use the pinned vendored libzstd so per-block content checksums are enforced independently of the header/footer xxh3; mutating only an upstream fixture's block checksum leaves metadata parsing valid and is rejected at decode, matching upstream. The planner uses refcounted manifest-resident headers, block envelopes, DID blooms, and collection summaries; the offline contract hides every JSS file after startup and still requires an exact plan, proving the request path performs no segment-file I/O. The harness rejects any upstream checkout not at the recorded pin. | +| Archive XRPC | Official Go client plus Stream conformance replay over listSegments/getSegment/getBlock/planBackfill | **closed at the default configuration.** With the pinned Zig dependencies cached, `just archive-contract` seeds three real sealed JSS segments and runs without network access. The pinned Go client consumed 5,000 ordered whole-segment rows, the exact 2,500-row DID block plan, 50 required identity sentinels for an empty collection match, the 234-row `(1000,1234]` bound, and a duplicate-free archive-to-live cutover at cursor 5000. Direct conformance also proves checksums/ETags, byte-identical segments, raw block decode, error names, DID/collection blooms, plan modes, and the pinned `http.ServeContent` behavior for both download endpoints. Single, suffix, multipart, mixed-overlap, aggregate-size fallback, and the distinct malformed/non-overlap 416 responses cross the real production server. The same receipt proves descriptor- or manifest-derived Last-Modified values, Go's If-Match/If-Unmodified-Since/If-None-Match/If-Modified-Since precedence, strong ETag and exact-date If-Range, and 304/412 representation-header behavior. The date parser accepts all three forms supported by Go's `http.ParseTime`. Real HTTP receipts also prove the middleware boundary: unknown NSIDs are 501 `MethodNotImplemented`, storage faults are 500 `InternalServerError`, and pre-steady-state XRPC requests receive a JSON 503 `ServiceUnavailable` while operational routes remain reachable. Sealed reads use the pinned vendored libzstd so per-block content checksums are enforced independently of the header/footer xxh3; mutating only an upstream fixture's block checksum leaves metadata parsing valid and is rejected at decode, matching upstream. The planner uses refcounted manifest-resident headers, block envelopes, DID blooms, and collection summaries; the offline contract hides every JSS file after startup and still requires an exact plan, proving the request path performs no segment-file I/O. The harness rejects any upstream checkout not at the recorded pin. Non-default archive cache/configuration controls remain in the explicit configuration-surface audit rather than being implied by this row. | | Subscribe v1/v2 | Wire/filter/cursor/compression unit and local e2e coverage | Cursor parsing/resolution precedes upgrade; both endpoints use the seq/time-us magnitude split, v1 clamps below-floor seqs, and v2 rejects them. Timestamp translation uses bounded positional reads of real sealed JSS headers, block indexes, and one selected block before upgrade; disk/decode/index faults are retryable generic 503s, logged internally and counted as `resolve_failed`. Hot/cold scan, skip, encode, options-update, oversize-parser, clean peer/server close, cursor-resolution, and sustained adversarial-rate boundaries feed canonical metrics, with deterministic offline tests derived from upstream. Subscriber transport sends a ping every 30 seconds, applies a kernel-enforced five-second deadline to every frame write, and sends close code 1001 before intentional server shutdown while keeping transport failure out of clean-disconnect accounting. A failed delivery, cold read, ping, or adversarial-rate check interrupts the server reader and removes the connection instead of leaving a ping-only zombie. V1 now negotiates genuine RFC 7692 deflate with the upstream 128-byte threshold and 32 KiB context takeover; v1 custom zstd and all v2 connections explicitly decline it. V2 dictionary negotiation, exact pinned dictionary download/validators, and v2 binary delivery remain independent of the frozen v1 dictionary contract; pinned official Go clients have decoded both compression families offline. | | Sync 1.1 live verification | Upstream requires durable per-DID chain/hosting state, MST inversion, op-CID consistency, rev replay/future guards, default acceptance of legacy-shaped commits, transparent whole-repo repair, and Atmos delivery scheduling | **closed at the implementation boundary.** Diff-CAR/MST inversion (including upstream's narrow default lenient carve-out), post-state op-CID proof, signed inner/outer consistency, exact decimal size/future-rev/replay gates, exact legacy-shape detection with default `LegacyAccept`, two-phase durable chain/hosting state, and account/identity replay ratchets are implemented. Recoverable decode/inversion/duplicate-path/op-CID/chain failures route to repair; signature failures and outer/inner producer-integrity mismatches bypass it, and no failed event is archived. `prepareRepo` retains and authenticates the canonical signed complete head without a second multi-gigabyte parse/walk. The live path now uses Atmos's 32 worker slots, one worker-held FIFO chain per DID, a 64-pending drop-oldest boundary, completion-order batches of 50/500 ms across DIDs, and `min(inflight)-1` cursor watermarks. Dropped and silent events leave the inflight set without inventing a delivery batch; cursor advancement remains coupled to durable archive boundaries. Offline receipts use real encoded frames, real RocksDB verifier state, and actual worker blocking rather than fixture verdicts or scheduler mocks. See `live-scheduler.md`. | | Sync 1.1 resync ordering | Upstream serializes per DID, drops stale async repairs, and emits a sync tombstone plus authoritative replacements | **closed at the implementation boundary.** The live coordinator has 32 real fetch workers, a 64-job queue, a five-minute fetch budget plus the 1,000 B/s-for-30-seconds slow-transfer guard, a bounded 16,384-entry 5/minute-per-DID limiter, 2,048 pending commits with drop-oldest overflow, fetch outside the DID lane, authenticated apply with older/equal-contradictory head rejection, ordered pending replay, and trigger-ticket-gated outbox delivery. Commit divergence schedules async repair; `#sync` divergence attempts inline repair and queues one retry only for transient failure, preserving the original relay cursor only on inline success. Matching Atmos's gate ownership, a divergent `#sync` repairs before envelope/inner validation; an invalid envelope rev then durably advances only the authenticated chain and drops the original tombstone plus every replacement row as one `live/invalid_rev` event. Valid repairs emit `sync` then bounded 1,024-row `create_resync` batches and stage fetched/pending chain checkpoints under the archive lock before any matching fsync. The shared hot tail is an amortized physical deque: mass replacement eviction cannot shift the full 256 MiB log once per row, and a 20,000-entry burst regression locks that boundary. Wire encoding uses a per-row scratch arena before the tail takes its durable copy, matching upstream's bounded append/publish lifetime instead of retaining v1 and v2 JSON for the entire fetched repository. A real 4,096-record MST repair now completes under a 1 MiB repair-scratch cap; the prior ownership model deterministically fails the same test with `ArchiveAppendFailed`. A separate pinned-simulator saturation receipt drives 96 real chain breaks through the ReleaseSafe process, observes the exact 32-active/64-queued Atmos envelope, completes all 96 real CAR/MST repairs, and converges the physical archive to independent ground truth. Loopback-only tests additionally exercise real DID resolution/signature, buffering while a response is held open, slow-transfer cancellation, durable RocksDB promotion/cursor coupling, v2 tail publication, and sealed JSS row order. | diff --git a/docs/upstream-harness.md b/docs/upstream-harness.md index d5932ff..cfe32fc 100644 --- a/docs/upstream-harness.md +++ b/docs/upstream-harness.md @@ -231,7 +231,12 @@ It verifies Last-Modified against the physical segment mtime and exercises If-Match, If-Unmodified-Since, If-None-Match, If-Modified-Since, and ETag/date If-Range with Go's precondition precedence. The first red receipt lacked Last-Modified; the earlier multipart receipt observed 416 on the first valid -multipart request. The combined receipts prove +multipart request. Unknown NSIDs traverse the same real route and must return +501 `MethodNotImplemented`; deleting a manifest-resident segment proves both +byte endpoints use the canonical 500 `InternalServerError` envelope. A +separate real-server test holds the serving gate closed and requires the JSON +503 `ServiceUnavailable` envelope instead of Stream's former plain-text 503. +The combined receipts prove upstream's `ok`, `bad_request`, `not_found`, and `error` partition; duration cardinality; and that served bytes increase by exactly one completely written 200 frame, never by partial or conditional responses. It also sends a real diff --git a/src/internal/metrics.zig b/src/internal/metrics.zig index cc4ad94..c62a891 100644 --- a/src/internal/metrics.zig +++ b/src/internal/metrics.zig @@ -28,7 +28,7 @@ pub const HttpHandler = enum { xrpc, }; pub const HttpMethod = enum { get, head, post, put, patch, delete, options, connect, trace, other }; -pub const HttpCode = enum { c200, c201, c206, c304, c400, c401, c404, c405, c409, c412, c416, c500, c503, other }; +pub const HttpCode = enum { c200, c201, c206, c304, c400, c401, c404, c405, c409, c412, c416, c500, c501, c503, other }; pub const GetBlockResult = enum { ok, not_found, bad_request, error_result }; pub const StoreOp = enum { get, set, delete, batch_commit }; pub const StoreStatus = enum { ok, notfound, error_result }; @@ -361,6 +361,7 @@ fn httpCode(code: u16) HttpCode { 412 => .c412, 416 => .c416, 500 => .c500, + 501 => .c501, 503 => .c503, else => .other, }; @@ -1245,6 +1246,7 @@ fn httpCodeLabel(value: HttpCode) []const u8 { .c412 => "412", .c416 => "416", .c500 => "500", + .c501 => "501", .c503 => "503", .other => "other", }; diff --git a/src/internal/server.zig b/src/internal/server.zig index 91c3af3..f5fb33e 100644 --- a/src/internal/server.zig +++ b/src/internal/server.zig @@ -717,6 +717,8 @@ pub const Handler = struct { const ops_path = std.mem.eql(u8, path, "/healthz") or std.mem.eql(u8, path, "/metrics") or std.mem.eql(u8, path, "/status") or std.mem.eql(u8, path, "/") or import_ops or dictionary_path; if (!hub.serving.load(.acquire) and !ops_path) { observation.code = 503; + if (std.mem.startsWith(u8, path, "/xrpc/")) + return respond(conn, "503 Service Unavailable", "application/json", "{\"error\":\"ServiceUnavailable\",\"message\":\"bootstrap in progress\"}"); return respond(conn, "503 Service Unavailable", "text/plain", "bootstrap in progress\n"); } if (hub.xrpc) |api| { @@ -1115,13 +1117,15 @@ const XrpcResponder = struct { pub fn err(self: *XrpcResponder, status: u16, name: []const u8, message: []const u8) void { var buf: [512]u8 = undefined; - const body = std.fmt.bufPrint(&buf, "{{\"error\":\"{s}\",\"message\":\"{s}\"}}", .{ name, message }) catch return; + const wire_name = if (status == 500) "InternalServerError" else name; + const body = std.fmt.bufPrint(&buf, "{{\"error\":\"{s}\",\"message\":\"{s}\"}}", .{ wire_name, message }) catch return; const status_line: []const u8 = switch (status) { 401 => "401 Unauthorized", 404 => "404 Not Found", 405 => "405 Method Not Allowed", 409 => "409 Conflict", 500 => "500 Internal Server Error", + 501 => "501 Not Implemented", 503 => "503 Service Unavailable", else => "400 Bad Request", }; @@ -1415,6 +1419,56 @@ test "cursor resolution is pre-upgrade and preserves v1-v2 floor semantics" { try std.testing.expectError(error.InvalidCursor, resolveCursor(&hub, true, "-1")); } +test "XRPC readiness gate returns the upstream JSON service error" { + const testing = std.testing; + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var tail = tail_mod.Tail.init(testing.allocator, io, 1 << 20); + defer tail.deinit(); + var stats: metrics.Stats = .{}; + var hub: Hub = .{ + .allocator = testing.allocator, + .io = io, + .tail = &tail, + .stats = &stats, + .serving = .init(false), + }; + defer hub.deinit(); + + var listener = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable; + defer listener.deinit(io); + const port = listener.socket.address.getPort(); + var ws_server = try websocket.Server(Handler).init(testing.allocator, io, .{ + .port = port, + .address = "127.0.0.1", + .max_conn = 4, + .handshake = .{ .max_size = 16 * 1024, .max_headers = 64 }, + }); + defer ws_server.deinit(); + const Fixture = struct { + fn run(server: *websocket.Server(Handler), server_listener: *Io.net.Server, fixture_hub: *Hub) !void { + server.runIo(server_listener, fixture_hub); + } + }; + var future = try io.concurrent(Fixture.run, .{ &ws_server, &listener, &hub }); + defer _ = future.cancel(io) catch {}; + + var url_buf: [256]u8 = undefined; + const url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}/xrpc/network.bsky.jetstream.listSegments", .{port}); + var transport = zat.HttpTransport.init(io, testing.allocator); + transport.keep_alive = false; + defer transport.deinit(); + var response = try transport.fetch(.{ .url = url, .max_response_size = 4096, .capture_response_headers = true }); + defer response.deinit(testing.allocator); + try testing.expectEqual(std.http.Status.service_unavailable, response.status); + try testing.expectEqualStrings("application/json", response.oauth.content_type.?); + const parsed = try std.json.parseFromSlice(std.json.Value, testing.allocator, response.body, .{}); + defer parsed.deinit(); + try testing.expectEqualStrings("ServiceUnavailable", parsed.value.object.get("error").?.string); + try testing.expect(std.mem.indexOf(u8, parsed.value.object.get("message").?.string, "bootstrap in progress") != null); +} + test "accounts HTTP route verifies real archive state and rate limits by port-free peer IP" { const testing = std.testing; var threaded: Io.Threaded = .init(testing.allocator, .{}); diff --git a/src/internal/xrpcapi.zig b/src/internal/xrpcapi.zig index 9635582..3baf80e 100644 --- a/src/internal/xrpcapi.zig +++ b/src/internal/xrpcapi.zig @@ -95,7 +95,7 @@ pub const Api = struct { } else if (std.mem.eql(u8, method, "getBlock")) { self.getBlock(respond, query, preconditions); } else { - respond.err(400, "MethodNotImplemented", method); + respond.err(501, "MethodNotImplemented", method); } return true; } diff --git a/tests/backfill_api.py b/tests/backfill_api.py index 158756e..20f9166 100644 --- a/tests/backfill_api.py +++ b/tests/backfill_api.py @@ -56,6 +56,15 @@ for url, want in [ except urllib.error.HTTPError as e: assert json.load(e)["error"] == want +# xrpcserver's public middleware owns unknown NSIDs: 501 with the canonical +# MethodNotImplemented envelope, not an ordinary InvalidRequest-shaped 400. +try: + urllib.request.urlopen(BASE + "doesNotExist") + sys.exit("expected unknown XRPC method to return 501") +except urllib.error.HTTPError as e: + assert e.code == 501 + assert json.load(e)["error"] == "MethodNotImplemented" + print(f"backfill api: PASS ({len(segs)} segments, seg0 {s0['eventCount']} events, block0 {n} events)") # --- planBackfill --- diff --git a/tests/http_metrics_contract.py b/tests/http_metrics_contract.py index 20b40e4..547512c 100644 --- a/tests/http_metrics_contract.py +++ b/tests/http_metrics_contract.py @@ -133,7 +133,7 @@ def main() -> None: assert request(block_path, headers={"Range": "bytes=999999999-"})[0] == 416 assert request("/xrpc/network.bsky.jetstream.getBlock")[0] == 400 assert request(f"/xrpc/network.bsky.jetstream.getBlock?segment={name}&blockIndex=999999")[0] == 404 - assert request("/xrpc/network.bsky.jetstream.notImplemented")[0] == 400 + assert request("/xrpc/network.bsky.jetstream.notImplemented")[0] == 501 assert request(block_path, method="POST")[0] == 405 assert request("/")[0] == 200 assert request("/status", method="HEAD")[0] == 200 @@ -156,8 +156,9 @@ def main() -> None: ("xrpc/", "GET", "206"): 1, ("xrpc/", "GET", "416"): 1, ("xrpc/", "GET", "412"): 1, - ("xrpc/", "GET", "400"): 2, + ("xrpc/", "GET", "400"): 1, ("xrpc/", "GET", "404"): 1, + ("xrpc/", "GET", "501"): 1, ("xrpc/", "POST", "405"): 1, ("root", "GET", "200"): 1, ("status", "HEAD", "200"): 1, @@ -192,8 +193,12 @@ def main() -> None: # from disk is an internal storage failure, not SegmentNotFound. (DATA_DIR / "segments" / segment["name"]).unlink() segment_path = f"/xrpc/network.bsky.jetstream.getSegment?name={name}" - assert request(segment_path)[0] == 500 - assert request(block_path)[0] == 500 + segment_failure = request(segment_path) + block_failure = request(block_path) + assert segment_failure[0] == 500 + assert block_failure[0] == 500 + assert json.loads(segment_failure[1])["error"] == "InternalServerError" + assert json.loads(block_failure[1])["error"] == "InternalServerError" failed = scrape() assert delta(after, failed, "jetstream_http_request_duration_seconds_count", handler="xrpc/", method="GET", code="500") == 2 assert delta(after, failed, "jetstream_getblock_requests_total", result="error") == 1