diff --git a/docs/semantic-parity.md b/docs/semantic-parity.md index 088c75c..9243cb5 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` byte-range behavior for both download endpoints: single, suffix, multipart, mixed-overlap, aggregate-size fallback, 416, strong If-Range, and weak If-None-Match paths all cross the real production server. 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.** 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. | | 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 361faa9..d5932ff 100644 --- a/docs/upstream-harness.md +++ b/docs/upstream-harness.md @@ -222,12 +222,16 @@ abort fired, recovered on the same disk, and matched the upstream oracle. `just http-metrics-contract` seeds the same real sealed archive and launches a ReleaseSafe Stream process with loopback-only dependencies. Standard-library -HTTP requests exercise getBlock 200, conditional 304, Range 206, invalid -Range 416, malformed 400, and missing-block 404 responses. The archive contract -additionally drives both getSegment and getBlock through real multipart ranges, -mixed overlapping/non-overlapping members, the aggregate-size fallback, strong -If-Range, and weak If-None-Match. Its red-first receipt observed the former 416 -on the first valid multipart request. The combined receipts prove +HTTP requests exercise getBlock 200, conditional 304, failed-precondition 412, +Range 206, invalid Range 416, malformed 400, and missing-block 404 responses. +The archive contract additionally drives both getSegment and getBlock through real multipart ranges, +mixed overlapping/non-overlapping members, the aggregate-size fallback, and +the different headers and bodies for malformed versus non-overlapping 416s. +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 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/http_time.zig b/src/internal/http_time.zig new file mode 100644 index 0000000..8d34592 --- /dev/null +++ b/src/internal/http_time.zig @@ -0,0 +1,179 @@ +//! HTTP-date parsing and formatting used by archive conditional requests. +//! Go's net/http accepts IMF-fixdate plus the obsolete RFC850 and asctime +//! forms, so this module deliberately implements the same three shapes. + +const std = @import("std"); + +const weekdays = [_][]const u8{ "Sun", "Mon", "Tue", "Wed", "Thu", "Fri", "Sat" }; +const months = [_][]const u8{ "Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec" }; + +pub fn format(buf: *[40]u8, unix_seconds: i64) []const u8 { + const days = @divFloor(unix_seconds, std.time.s_per_day); + const day_seconds: u32 = @intCast(@mod(unix_seconds, std.time.s_per_day)); + const civil = civilFromDays(days); + const weekday: usize = @intCast(@mod(days + 4, 7)); + return std.fmt.bufPrint(buf, "{s}, {d:0>2} {s} {d:0>4} {d:0>2}:{d:0>2}:{d:0>2} GMT", .{ + weekdays[weekday], + @as(u8, @intCast(civil.day)), + months[civil.month - 1], + @as(u32, @intCast(civil.year)), + day_seconds / 3600, + day_seconds / 60 % 60, + day_seconds % 60, + }) catch unreachable; +} + +pub fn parse(raw: []const u8) ?i64 { + if (parseImf(raw)) |value| return value; + if (parseRfc850(raw)) |value| return value; + return parseAsctime(raw); +} + +fn parseImf(raw: []const u8) ?i64 { + if (raw.len < 29 or raw[3] != ',' or raw[4] != ' ' or raw[7] != ' ' or + raw[11] != ' ' or raw[16] != ' ' or raw[19] != ':' or raw[22] != ':' or + !std.mem.eql(u8, afterFraction(raw[25..]) orelse return null, " GMT") or weekdayShort(raw[0..3]) == null) + return null; + return timestamp( + parseDigits(raw[12..16]) orelse return null, + monthNumber(raw[8..11]) orelse return null, + parseDigits(raw[5..7]) orelse return null, + parseDigits(raw[17..19]) orelse return null, + parseDigits(raw[20..22]) orelse return null, + parseDigits(raw[23..25]) orelse return null, + ); +} + +fn parseRfc850(raw: []const u8) ?i64 { + const comma = std.mem.indexOfScalar(u8, raw, ',') orelse return null; + if (weekdayLong(raw[0..comma]) == null) return null; + const tail = raw[comma + 1 ..]; + if (tail.len < 23 or tail[0] != ' ' or tail[3] != '-' or tail[7] != '-' or + tail[10] != ' ' or tail[13] != ':' or tail[16] != ':' or + !std.mem.eql(u8, afterFraction(tail[19..]) orelse return null, " GMT")) return null; + const short_year = parseDigits(tail[8..10]) orelse return null; + const year: u32 = if (short_year >= 69) 1900 + short_year else 2000 + short_year; + return timestamp( + year, + monthNumber(tail[4..7]) orelse return null, + parseDigits(tail[1..3]) orelse return null, + parseDigits(tail[11..13]) orelse return null, + parseDigits(tail[14..16]) orelse return null, + parseDigits(tail[17..19]) orelse return null, + ); +} + +fn parseAsctime(raw: []const u8) ?i64 { + if (raw.len < 24 or raw[3] != ' ' or raw[7] != ' ' or raw[10] != ' ' or + raw[13] != ':' or raw[16] != ':' or + weekdayShort(raw[0..3]) == null) return null; + const year_suffix = afterFraction(raw[19..]) orelse return null; + if (year_suffix.len != 5 or year_suffix[0] != ' ') return null; + const day: u32 = if (raw[8] == ' ') + parseDigits(raw[9..10]) orelse return null + else + parseDigits(raw[8..10]) orelse return null; + return timestamp( + parseDigits(year_suffix[1..]) orelse return null, + monthNumber(raw[4..7]) orelse return null, + day, + parseDigits(raw[11..13]) orelse return null, + parseDigits(raw[14..16]) orelse return null, + parseDigits(raw[17..19]) orelse return null, + ); +} + +/// time.Parse accepts a fractional second even when the selected layout does +/// not contain one. http.ParseTime inherits that behavior for all three date +/// layouts; consume one to nine digits but leave the integer Unix second. +fn afterFraction(raw: []const u8) ?[]const u8 { + if (raw.len == 0 or (raw[0] != '.' and raw[0] != ',')) return raw; + var i: usize = 1; + while (i < raw.len and raw[i] >= '0' and raw[i] <= '9' and i <= 9) : (i += 1) {} + if (i == 1) return null; + return raw[i..]; +} + +fn timestamp(year: u32, month: u32, day: u32, hour: u32, minute: u32, second: u32) ?i64 { + if (hour > 23 or minute > 59 or second > 59) return null; + const days = daysFromCivil(@intCast(year), @intCast(month), @intCast(day)) orelse return null; + return days * std.time.s_per_day + @as(i64, hour) * 3600 + @as(i64, minute) * 60 + second; +} + +fn parseDigits(raw: []const u8) ?u32 { + if (raw.len == 0) return null; + var value: u32 = 0; + for (raw) |c| { + if (c < '0' or c > '9') return null; + value = value * 10 + c - '0'; + } + return value; +} + +fn monthNumber(raw: []const u8) ?u32 { + for (months, 1..) |month, i| if (std.mem.eql(u8, raw, month)) return @intCast(i); + return null; +} + +fn weekdayShort(raw: []const u8) ?usize { + for (weekdays, 0..) |weekday, i| if (std.mem.eql(u8, raw, weekday)) return i; + return null; +} + +fn weekdayLong(raw: []const u8) ?usize { + const long = [_][]const u8{ "Sunday", "Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday" }; + for (long, 0..) |weekday, i| if (std.mem.eql(u8, raw, weekday)) return i; + return null; +} + +fn leap(year: i64) bool { + return @mod(year, 4) == 0 and (@mod(year, 100) != 0 or @mod(year, 400) == 0); +} + +fn daysFromCivil(year_raw: i64, month: u8, day: u8) ?i64 { + if (month < 1 or month > 12 or day < 1) return null; + const lengths = [_]u8{ 31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31 }; + const limit: u8 = lengths[month - 1] + @as(u8, if (month == 2 and leap(year_raw)) 1 else 0); + if (day > limit) return null; + const year = year_raw - @intFromBool(month <= 2); + const era = @divFloor(year, 400); + const yoe = year - era * 400; + const shifted_month: i64 = @as(i64, month) + @as(i64, if (month > 2) -3 else 9); + const doy = @divFloor(153 * shifted_month + 2, 5) + day - 1; + const doe = yoe * 365 + @divFloor(yoe, 4) - @divFloor(yoe, 100) + doy; + return era * 146097 + doe - 719468; +} + +const Civil = struct { year: i64, month: usize, day: i64 }; + +fn civilFromDays(days: i64) Civil { + const z = days + 719468; + const era = @divFloor(z, 146097); + const doe = z - era * 146097; + const yoe = @divFloor(doe - @divFloor(doe, 1460) + @divFloor(doe, 36524) - @divFloor(doe, 146096), 365); + var year = yoe + era * 400; + const doy = doe - (365 * yoe + @divFloor(yoe, 4) - @divFloor(yoe, 100)); + const mp = @divFloor(5 * doy + 2, 153); + const day = doy - @divFloor(153 * mp + 2, 5) + 1; + const month: usize = @intCast(mp + @as(i64, if (mp < 10) 3 else -9)); + year += @intFromBool(month <= 2); + return .{ .year = year, .month = month, .day = day }; +} + +test "formats and parses Go HTTP date forms" { + var buf: [40]u8 = undefined; + try std.testing.expectEqualStrings("Mon, 02 Jan 2006 15:04:05 GMT", format(&buf, 1_136_214_245)); + try std.testing.expectEqual(@as(?i64, 1_136_214_245), parse("Mon, 02 Jan 2006 15:04:05 GMT")); + try std.testing.expectEqual(@as(?i64, 1_136_214_245), parse("Monday, 02-Jan-06 15:04:05 GMT")); + try std.testing.expectEqual(@as(?i64, 1_136_214_245), parse("Mon Jan 2 15:04:05 2006")); + try std.testing.expectEqual(@as(?i64, 0), parse("Thu, 01 Jan 1970 00:00:00 GMT")); + try std.testing.expectEqual(@as(?i64, -1), parse("Wed, 31 Dec 1969 23:59:59 GMT")); + try std.testing.expectEqual(@as(?i64, 784_111_777), parse("Sun, 06 Nov 1994 08:49:37 GMT")); + try std.testing.expectEqual(@as(?i64, 784_111_777), parse("Sunday, 06-Nov-94 08:49:37 GMT")); + try std.testing.expectEqual(@as(?i64, 784_111_777), parse("Sun Nov 6 08:49:37 1994")); + try std.testing.expectEqual(@as(?i64, 784_111_777), parse("Sun, 06 Nov 1994 08:49:37.123456789 GMT")); + try std.testing.expectEqual(@as(?i64, 784_111_777), parse("Sunday, 06-Nov-94 08:49:37,5 GMT")); + try std.testing.expectEqual(@as(?i64, 784_111_777), parse("Sun Nov 6 08:49:37.01 1994")); + try std.testing.expect(parse("Mon, 31 Feb 2006 15:04:05 GMT") == null); + try std.testing.expect(parse("Mon, 02 Jan 2006 15:04:60 GMT") == null); +} diff --git a/src/internal/manifest.zig b/src/internal/manifest.zig index 708d7fa..a678365 100644 --- a/src/internal/manifest.zig +++ b/src/internal/manifest.zig @@ -106,6 +106,7 @@ pub const PlanningSegment = struct { pub const Summary = struct { idx: u64, size: u64, + mtime: Io.Timestamp, header: segment.Header, }; @@ -335,7 +336,7 @@ fn loadMetadata(allocator: Allocator, io: Io, dir: Io.Dir, idx: u64, verify_chec collection_index, ); return .{ - .summary = .{ .idx = idx, .size = stat.size, .header = header }, + .summary = .{ .idx = idx, .size = stat.size, .mtime = stat.mtime, .header = header }, .blocks = resident, }; } diff --git a/src/internal/metrics.zig b/src/internal/metrics.zig index 672c0bd..cc4ad94 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, c416, c500, c503, other }; +pub const HttpCode = enum { c200, c201, c206, c304, c400, c401, c404, c405, c409, c412, c416, c500, 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 }; @@ -358,6 +358,7 @@ fn httpCode(code: u16) HttpCode { 404 => .c404, 405 => .c405, 409 => .c409, + 412 => .c412, 416 => .c416, 500 => .c500, 503 => .c503, @@ -1241,6 +1242,7 @@ fn httpCodeLabel(value: HttpCode) []const u8 { .c404 => "404", .c405 => "405", .c409 => "409", + .c412 => "412", .c416 => "416", .c500 => "500", .c503 => "503", @@ -1297,13 +1299,14 @@ test "getBlock metrics preserve upstream result and served-byte semantics" { stats.observeGetBlock(304, 0, 101); stats.observeGetBlock(400, 0, 102); stats.observeGetBlock(404, 0, 103); - stats.observeGetBlock(416, 0, 104); + stats.observeGetBlock(412, 0, 104); + stats.observeGetBlock(416, 0, 105); try std.testing.expectEqual(3, stats.getblock_requests.getPtrConst(.ok).load(.monotonic)); try std.testing.expectEqual(1, stats.getblock_requests.getPtrConst(.bad_request).load(.monotonic)); try std.testing.expectEqual(1, stats.getblock_requests.getPtrConst(.not_found).load(.monotonic)); - try std.testing.expectEqual(1, stats.getblock_requests.getPtrConst(.error_result).load(.monotonic)); + try std.testing.expectEqual(2, stats.getblock_requests.getPtrConst(.error_result).load(.monotonic)); try std.testing.expectEqual(100, stats.getblock_served_bytes_total.load(.monotonic)); - try std.testing.expectEqual(6, stats.getblock_duration_count.load(.monotonic)); + try std.testing.expectEqual(7, stats.getblock_duration_count.load(.monotonic)); } test "format renders counters" { diff --git a/src/internal/server.zig b/src/internal/server.zig index 25ead18..91c3af3 100644 --- a/src/internal/server.zig +++ b/src/internal/server.zig @@ -16,6 +16,7 @@ const cold = @import("cold.zig"); const disk_space = @import("disk_space.zig"); const filter_mod = @import("filter.zig"); const homepage = @import("homepage.zig"); +const http_time = @import("http_time.zig"); const metrics = @import("metrics.zig"); const process_metrics = @import("process_metrics.zig"); const repo_action_limiter = @import("repo_action_limiter.zig"); @@ -729,9 +730,14 @@ pub const Handler = struct { query, body, headers.get("authorization"), - headers.get("range"), - headers.get("if-none-match"), - headers.get("if-range"), + .{ + .range = headers.get("range"), + .if_match = headers.get("if-match"), + .if_unmodified_since = headers.get("if-unmodified-since"), + .if_none_match = headers.get("if-none-match"), + .if_modified_since = headers.get("if-modified-since"), + .if_range = headers.get("if-range"), + }, )) return; } const status_head = std.mem.eql(u8, method, "HEAD") and std.mem.eql(u8, path, "/status"); @@ -895,6 +901,18 @@ pub const Handler = struct { const XrpcResponder = struct { conn: *websocket.Conn, observation: *HttpObservation, + modified_s: ?i64 = null, + + pub fn archiveModified(self: *XrpcResponder, modified_s: i64) void { + // net/http treats the Unix epoch as an unspecified modification time. + self.modified_s = if (modified_s == 0) null else modified_s; + } + + fn lastModifiedHeader(self: *const XrpcResponder, buf: *[64]u8) []const u8 { + const modified_s = self.modified_s orelse return ""; + var date_buf: [40]u8 = undefined; + return std.fmt.bufPrint(buf, "Last-Modified: {s}\r\n", .{http_time.format(&date_buf, modified_s)}) catch ""; + } pub fn json(self: *XrpcResponder, status: u16, body: []const u8) void { self.observation.code = status; @@ -902,11 +920,13 @@ const XrpcResponder = struct { } pub fn raw(self: *XrpcResponder, status: u16, body: []const u8, etag_hex: []const u8) void { - var buf: [320]u8 = undefined; + var modified_buf: [64]u8 = undefined; + const modified = self.lastModifiedHeader(&modified_buf); + var buf: [384]u8 = undefined; const header = std.fmt.bufPrint( &buf, - "HTTP/1.1 {s}\r\nContent-Type: application/octet-stream\r\nContent-Length: {d}\r\nAccept-Ranges: bytes\r\nETag: \"{s}\"\r\nCache-Control: public, no-cache\r\nConnection: close\r\nServer: stream\r\n\r\n", - .{ statusLine(status), body.len, etag_hex }, + "HTTP/1.1 {s}\r\nContent-Type: application/octet-stream\r\nContent-Length: {d}\r\nAccept-Ranges: bytes\r\nETag: \"{s}\"\r\nCache-Control: public, no-cache\r\n{s}Connection: close\r\nServer: stream\r\n\r\n", + .{ statusLine(status), body.len, etag_hex, modified }, ) catch return; self.observation.code = status; self.conn.writeFramed(header) catch return; @@ -934,11 +954,13 @@ const XrpcResponder = struct { } pub fn partial(self: *XrpcResponder, body: []const u8, etag_hex: []const u8, start: u64, end: u64, total: u64) void { - var buf: [400]u8 = undefined; + var modified_buf: [64]u8 = undefined; + const modified = self.lastModifiedHeader(&modified_buf); + var buf: [464]u8 = undefined; const header = std.fmt.bufPrint( &buf, - "HTTP/1.1 206 Partial Content\r\nContent-Type: application/octet-stream\r\nContent-Length: {d}\r\nContent-Range: bytes {d}-{d}/{d}\r\nAccept-Ranges: bytes\r\nETag: \"{s}\"\r\nCache-Control: public, no-cache\r\nConnection: close\r\nServer: stream\r\n\r\n", - .{ body.len, start, end, total, etag_hex }, + "HTTP/1.1 206 Partial Content\r\nContent-Type: application/octet-stream\r\nContent-Length: {d}\r\nContent-Range: bytes {d}-{d}/{d}\r\nAccept-Ranges: bytes\r\nETag: \"{s}\"\r\nCache-Control: public, no-cache\r\n{s}Connection: close\r\nServer: stream\r\n\r\n", + .{ body.len, start, end, total, etag_hex, modified }, ) catch return; self.observation.code = 206; self.conn.writeFramed(header) catch return; @@ -986,11 +1008,13 @@ const XrpcResponder = struct { } fn multipartHeader(self: *XrpcResponder, etag_hex: []const u8, boundary: *const [60]u8, body_len: u64) void { - var buf: [512]u8 = undefined; + var modified_buf: [64]u8 = undefined; + const modified = self.lastModifiedHeader(&modified_buf); + var buf: [576]u8 = undefined; const header = std.fmt.bufPrint( &buf, - "HTTP/1.1 206 Partial Content\r\nContent-Type: multipart/byteranges; boundary={s}\r\nContent-Length: {d}\r\nAccept-Ranges: bytes\r\nETag: \"{s}\"\r\nCache-Control: public, no-cache\r\nConnection: close\r\nServer: stream\r\n\r\n", - .{ boundary, body_len, etag_hex }, + "HTTP/1.1 206 Partial Content\r\nContent-Type: multipart/byteranges; boundary={s}\r\nContent-Length: {d}\r\nAccept-Ranges: bytes\r\nETag: \"{s}\"\r\nCache-Control: public, no-cache\r\n{s}Connection: close\r\nServer: stream\r\n\r\n", + .{ boundary, body_len, etag_hex, modified }, ) catch return; self.observation.code = 206; self.conn.writeFramed(header) catch { @@ -1013,18 +1037,20 @@ const XrpcResponder = struct { // whole representation (e.g. bytes=0- or an oversized suffix). const partial_response = range_requested; const len = end - start + 1; - var header_buf: [512]u8 = undefined; + var modified_buf: [64]u8 = undefined; + const modified = self.lastModifiedHeader(&modified_buf); + var header_buf: [576]u8 = undefined; const header = if (partial_response) std.fmt.bufPrint( &header_buf, - "HTTP/1.1 206 Partial Content\r\nContent-Type: application/octet-stream\r\nContent-Length: {d}\r\nContent-Range: bytes {d}-{d}/{d}\r\nAccept-Ranges: bytes\r\nETag: \"{s}\"\r\nCache-Control: public, no-cache\r\nConnection: close\r\nServer: stream\r\n\r\n", - .{ len, start, end, total, etag_hex }, + "HTTP/1.1 206 Partial Content\r\nContent-Type: application/octet-stream\r\nContent-Length: {d}\r\nContent-Range: bytes {d}-{d}/{d}\r\nAccept-Ranges: bytes\r\nETag: \"{s}\"\r\nCache-Control: public, no-cache\r\n{s}Connection: close\r\nServer: stream\r\n\r\n", + .{ len, start, end, total, etag_hex, modified }, ) catch return else std.fmt.bufPrint( &header_buf, - "HTTP/1.1 200 OK\r\nContent-Type: application/octet-stream\r\nContent-Length: {d}\r\nAccept-Ranges: bytes\r\nETag: \"{s}\"\r\nCache-Control: public, no-cache\r\nConnection: close\r\nServer: stream\r\n\r\n", - .{ len, etag_hex }, + "HTTP/1.1 200 OK\r\nContent-Type: application/octet-stream\r\nContent-Length: {d}\r\nAccept-Ranges: bytes\r\nETag: \"{s}\"\r\nCache-Control: public, no-cache\r\n{s}Connection: close\r\nServer: stream\r\n\r\n", + .{ len, etag_hex, modified }, ) catch return; self.observation.code = if (partial_response) 206 else 200; self.conn.writeFramed(header) catch return; @@ -1043,14 +1069,36 @@ const XrpcResponder = struct { if (!partial_response) self.observation.served_bytes = len; } - pub fn rangeNotSatisfiable(self: *XrpcResponder, total: u64) void { - var buf: [256]u8 = undefined; + pub fn rangeNotSatisfiable(self: *XrpcResponder, total: u64, no_overlap: bool) void { + const body = if (no_overlap) "invalid range: failed to overlap\n" else "invalid range\n"; + var buf: [384]u8 = undefined; + const header = if (no_overlap) + std.fmt.bufPrint( + &buf, + "HTTP/1.1 416 Range Not Satisfiable\r\nContent-Range: bytes */{d}\r\nContent-Type: text/plain; charset=utf-8\r\nX-Content-Type-Options: nosniff\r\nContent-Length: {d}\r\nConnection: close\r\nServer: stream\r\n\r\n", + .{ total, body.len }, + ) catch return + else + std.fmt.bufPrint( + &buf, + "HTTP/1.1 416 Range Not Satisfiable\r\nContent-Type: text/plain; charset=utf-8\r\nX-Content-Type-Options: nosniff\r\nContent-Length: {d}\r\nConnection: close\r\nServer: stream\r\n\r\n", + .{body.len}, + ) catch return; + self.observation.code = 416; + self.conn.writeFramed(header) catch return; + self.conn.writeFramed(body) catch return; + } + + pub fn preconditionFailed(self: *XrpcResponder, etag_hex: []const u8) void { + var modified_buf: [64]u8 = undefined; + const modified = self.lastModifiedHeader(&modified_buf); + var buf: [384]u8 = undefined; const header = std.fmt.bufPrint( &buf, - "HTTP/1.1 416 Range Not Satisfiable\r\nContent-Range: bytes */{d}\r\nContent-Length: 0\r\nConnection: close\r\nServer: stream\r\n\r\n", - .{total}, + "HTTP/1.1 412 Precondition Failed\r\nContent-Type: application/octet-stream\r\nETag: \"{s}\"\r\nCache-Control: public, no-cache\r\n{s}Content-Length: 0\r\nConnection: close\r\nServer: stream\r\n\r\n", + .{ etag_hex, modified }, ) catch return; - self.observation.code = 416; + self.observation.code = 412; self.conn.writeFramed(header) catch return; } diff --git a/src/internal/xrpcapi.zig b/src/internal/xrpcapi.zig index 5100ef1..9635582 100644 --- a/src/internal/xrpcapi.zig +++ b/src/internal/xrpcapi.zig @@ -15,6 +15,7 @@ const archive_mod = @import("archive.zig"); const segment = @import("segment.zig"); const import_manager = @import("timestamp/manager.zig"); const import_jobs = @import("timestamp/jobs.zig"); +const http_time = @import("http_time.zig"); const time_format = @import("bootstrap/resync.zig"); const zstd = @import("zstd.zig"); @@ -22,6 +23,15 @@ const Io = std.Io; const Allocator = std.mem.Allocator; const log = std.log.scoped(.stream); +pub const Preconditions = struct { + range: ?[]const u8 = null, + if_match: ?[]const u8 = null, + if_unmodified_since: ?[]const u8 = null, + if_none_match: ?[]const u8 = null, + if_modified_since: ?[]const u8 = null, + if_range: ?[]const u8 = null, +}; + pub const Api = struct { allocator: Allocator, io: Io, @@ -34,7 +44,7 @@ pub const Api = struct { /// route an /xrpc/network.bsky.jetstream.* request. returns false when /// the path isn't ours (caller 404s). - pub fn handle(self: *Api, respond: anytype, http_method: []const u8, path: []const u8, query: []const u8, body: []const u8, authorization: ?[]const u8, range_header: ?[]const u8, if_none_match: ?[]const u8, if_range: ?[]const u8) bool { + pub fn handle(self: *Api, respond: anytype, http_method: []const u8, path: []const u8, query: []const u8, body: []const u8, authorization: ?[]const u8, preconditions: Preconditions) bool { const prefix = "/xrpc/network.bsky.jetstream."; if (!std.mem.startsWith(u8, path, prefix)) return false; const method = path[prefix.len..]; @@ -77,13 +87,13 @@ pub const Api = struct { return true; } if (std.mem.eql(u8, method, "getZstdDictionary")) { - self.getZstdDictionary(respond, query, if_none_match); + self.getZstdDictionary(respond, query, preconditions.if_none_match); } else if (std.mem.eql(u8, method, "listSegments")) { self.listSegments(respond, query); } else if (std.mem.eql(u8, method, "getSegment")) { - self.getSegment(respond, query, range_header, if_none_match, if_range); + self.getSegment(respond, query, preconditions); } else if (std.mem.eql(u8, method, "getBlock")) { - self.getBlock(respond, query, range_header, if_none_match, if_range); + self.getBlock(respond, query, preconditions); } else { respond.err(400, "MethodNotImplemented", method); } @@ -223,7 +233,7 @@ pub const Api = struct { respond.json(200, out.written()); } - fn getSegment(self: *Api, respond: anytype, query: []const u8, range_header: ?[]const u8, if_none_match: ?[]const u8, if_range: ?[]const u8) void { + fn getSegment(self: *Api, respond: anytype, query: []const u8, preconditions: Preconditions) void { const name = queryParam(query, "name") orelse return respond.err(400, "InvalidRequest", "missing name"); const idx = archive_mod.parseSegmentIndex(name) orelse @@ -242,19 +252,24 @@ pub const Api = struct { return respond.err(500, "InternalError", "failed to read segment header"); var hex_buf: [16]u8 = undefined; const etag = checksumHex(&hex_buf, header.checksum); - if (if_none_match) |candidate| { - if (ifNoneMatchMatches(candidate, etag)) return respond.notModified(etag); + const modified_s = stat.mtime.toSeconds(); + respond.archiveModified(modified_s); + switch (checkPreconditions(preconditions, etag, modified_s)) { + .proceed => {}, + .not_modified => return respond.notModified(etag), + .failed => return respond.preconditionFailed(etag), } - // single-range requests (bytes=a-b / a- / -n): sealed files are - // immutable so ranges need no If-Range dance beyond the strong ETag - if (range_header) |raw| range: { - if (if_range) |candidate| { - if (!etagMatches(candidate, etag)) break :range; + // ServeContent-style single and multipart ranges. If-Range accepts + // either the strong ETag or the representation's exact mtime. + if (preconditions.range) |raw| range: { + if (preconditions.if_range) |candidate| { + if (!ifRangeMatches(candidate, etag, modified_s)) break :range; } const ranges = parseRanges(self.allocator, raw, stat.size) catch |err| switch (err) { error.OutOfMemory => return respond.err(500, "InternalError", "oom"), - error.InvalidRange, error.NoOverlap => return respond.rangeNotSatisfiable(stat.size), + error.InvalidRange => return respond.rangeNotSatisfiable(stat.size, false), + error.NoOverlap => return respond.rangeNotSatisfiable(stat.size, true), }; defer self.allocator.free(ranges); if (sumRangeLengths(ranges) <= stat.size) switch (ranges.len) { @@ -278,7 +293,7 @@ pub const Api = struct { respond.file(self.io, file, etag, 0, stat.size - 1, stat.size, false); } - fn getBlock(self: *Api, respond: anytype, query: []const u8, range_header: ?[]const u8, if_none_match: ?[]const u8, if_range: ?[]const u8) void { + fn getBlock(self: *Api, respond: anytype, query: []const u8, preconditions: Preconditions) void { const name = queryParam(query, "segment") orelse return respond.err(400, "InvalidRequest", "missing segment"); const block_idx = blk: { @@ -289,7 +304,7 @@ pub const Api = struct { }; const segment_idx = archive_mod.parseSegmentIndex(name) orelse return respond.err(400, "InvalidRequest", "bad segment name"); - if (self.archive.manifest.segmentByIndex(segment_idx) == null) + const manifest_ref = self.archive.manifest.segmentByIndex(segment_idx) orelse return respond.err(404, "SegmentNotFound", name); var file = self.archive.dir.openFile(self.io, name, .{}) catch @@ -319,16 +334,21 @@ pub const Api = struct { var checksum_buf: [16]u8 = undefined; var etag_buf: [32]u8 = undefined; const etag = std.fmt.bufPrint(&etag_buf, "{s}:{d}", .{ checksumHex(&checksum_buf, header.checksum), block_idx }) catch unreachable; - if (if_none_match) |candidate| { - if (ifNoneMatchMatches(candidate, etag)) return respond.notModified(etag); + const modified_s = manifest_ref.mtime.toSeconds(); + respond.archiveModified(modified_s); + switch (checkPreconditions(preconditions, etag, modified_s)) { + .proceed => {}, + .not_modified => return respond.notModified(etag), + .failed => return respond.preconditionFailed(etag), } - if (range_header) |raw| range: { - if (if_range) |candidate| { - if (!etagMatches(candidate, etag)) break :range; + if (preconditions.range) |raw| range: { + if (preconditions.if_range) |candidate| { + if (!ifRangeMatches(candidate, etag, modified_s)) break :range; } const ranges = parseRanges(self.allocator, raw, frame.len) catch |err| switch (err) { error.OutOfMemory => return respond.err(500, "InternalError", "oom"), - error.InvalidRange, error.NoOverlap => return respond.rangeNotSatisfiable(frame.len), + error.InvalidRange => return respond.rangeNotSatisfiable(frame.len, false), + error.NoOverlap => return respond.rangeNotSatisfiable(frame.len, true), }; defer self.allocator.free(ranges); if (sumRangeLengths(ranges) <= frame.len) switch (ranges.len) { @@ -421,26 +441,95 @@ fn sumRangeLengths(ranges: []const ByteRange) u64 { return total; } -fn etagMatches(raw: []const u8, etag: []const u8) bool { - // ServeContent's If-Range path scans only the first entity tag. Trailing - // bytes are ignored after that token, but a later matching tag cannot - // rescue a non-matching first tag. - const part = std.mem.trimStart(u8, raw, " \t"); - return part.len >= etag.len + 2 and part[0] == '"' and part[etag.len + 1] == '"' and - std.mem.eql(u8, part[1 .. etag.len + 1], etag); +const EtagToken = struct { + value: []const u8, + weak: bool, + rest: []const u8, +}; + +/// Exact scanner shape used by Go net/http: trim leading MIME whitespace, +/// accept W/ only with that casing, and stop the list at malformed syntax. +fn scanEtag(raw: []const u8) ?EtagToken { + const trimmed = std.mem.trim(u8, raw, " \t\r\n"); + const weak = std.mem.startsWith(u8, trimmed, "W/"); + const start: usize = if (weak) 2 else 0; + if (trimmed.len < start + 2 or trimmed[start] != '"') return null; + var i = start + 1; + while (i < trimmed.len) : (i += 1) switch (trimmed[i]) { + 0x21, 0x23...0x7e, 0x80...0xff => {}, + '"' => return .{ + .value = trimmed[start + 1 .. i], + .weak = weak, + .rest = trimmed[i + 1 ..], + }, + else => return null, + }; + return null; +} + +fn ifMatchMatches(raw: []const u8, etag: []const u8) bool { + var remaining = raw; + while (true) { + remaining = std.mem.trim(u8, remaining, " \t\r\n"); + if (remaining.len == 0) return false; + if (remaining[0] == ',') { + remaining = remaining[1..]; + continue; + } + if (remaining[0] == '*') return true; + const token = scanEtag(remaining) orelse return false; + if (!token.weak and std.mem.eql(u8, token.value, etag)) return true; + remaining = token.rest; + } } -/// If-None-Match uses weak comparison (RFC 9110 ยง13.1.2), unlike If-Range. +/// If-None-Match uses weak comparison, but otherwise retains Go's scanning +/// behavior: malformed syntax stops evaluation and cannot expose later tags. fn ifNoneMatchMatches(raw: []const u8, etag: []const u8) bool { - var it = std.mem.splitScalar(u8, raw, ','); - while (it.next()) |part_raw| { - var part = std.mem.trim(u8, part_raw, " \t"); - if (std.mem.eql(u8, part, "*")) return true; - if (std.mem.startsWith(u8, part, "W/")) part = part[2..]; - if (part.len == etag.len + 2 and part[0] == '"' and part[part.len - 1] == '"' and - std.mem.eql(u8, part[1 .. part.len - 1], etag)) return true; + var remaining = raw; + while (true) { + remaining = std.mem.trim(u8, remaining, " \t\r\n"); + if (remaining.len == 0) return false; + if (remaining[0] == ',') { + remaining = remaining[1..]; + continue; + } + if (remaining[0] == '*') return true; + const token = scanEtag(remaining) orelse return false; + if (std.mem.eql(u8, token.value, etag)) return true; + remaining = token.rest; } - return false; +} + +const PreconditionResult = enum { proceed, not_modified, failed }; + +fn checkPreconditions(headers: Preconditions, etag: []const u8, modified_s: i64) PreconditionResult { + // RFC 7232 section 6, in the same precedence order as net/http. + const if_match = headers.if_match orelse ""; + if (if_match.len > 0) { + if (!ifMatchMatches(if_match, etag)) return .failed; + } else if (modified_s != 0) { + if (headers.if_unmodified_since) |raw| { + if (http_time.parse(raw)) |date_s| if (modified_s > date_s) return .failed; + } + } + + const if_none_match = headers.if_none_match orelse ""; + if (if_none_match.len > 0) { + if (ifNoneMatchMatches(if_none_match, etag)) return .not_modified; + } else if (modified_s != 0) { + if (headers.if_modified_since) |raw| { + if (http_time.parse(raw)) |date_s| if (modified_s <= date_s) return .not_modified; + } + } + return .proceed; +} + +fn ifRangeMatches(raw: []const u8, etag: []const u8, modified_s: i64) bool { + if (scanEtag(raw)) |token| + return !token.weak and std.mem.eql(u8, token.value, etag); + if (modified_s == 0) return false; + return if (http_time.parse(raw)) |date_s| date_s == modified_s else false; } test "parseRanges matches ServeContent forms" { @@ -468,17 +557,34 @@ test "parseRanges matches ServeContent forms" { } test "strong etag matching" { - try std.testing.expect(etagMatches("\"abc:0\"", "abc:0")); - try std.testing.expect(etagMatches("\"abc:0\", \"other\"", "abc:0")); - try std.testing.expect(!etagMatches("\"other\", \"abc:0\"", "abc:0")); - try std.testing.expect(!etagMatches("*", "abc:0")); - try std.testing.expect(!etagMatches("W/\"abc:0\"", "abc:0")); + try std.testing.expect(ifRangeMatches("\"abc:0\"", "abc:0", 123)); + try std.testing.expect(ifRangeMatches("\"abc:0\", \"other\"", "abc:0", 123)); + try std.testing.expect(!ifRangeMatches("\"other\", \"abc:0\"", "abc:0", 123)); + try std.testing.expect(!ifRangeMatches("*", "abc:0", 123)); + try std.testing.expect(!ifRangeMatches("W/\"abc:0\"", "abc:0", 123)); + try std.testing.expect(ifRangeMatches("Thu, 01 Jan 1970 00:02:03 GMT", "abc:0", 123)); } test "If-None-Match accepts weak validators" { try std.testing.expect(ifNoneMatchMatches("W/\"abc:0\"", "abc:0")); try std.testing.expect(ifNoneMatchMatches("\"other\", W/\"abc:0\"", "abc:0")); try std.testing.expect(!ifNoneMatchMatches("W/\"other\"", "abc:0")); + try std.testing.expect(!ifNoneMatchMatches("bad, W/\"abc:0\"", "abc:0")); +} + +test "If-Match requires a strong matching validator" { + try std.testing.expect(ifMatchMatches("\"other\", \"abc:0\"", "abc:0")); + try std.testing.expect(ifMatchMatches("*", "abc:0")); + try std.testing.expect(!ifMatchMatches("W/\"abc:0\"", "abc:0")); + try std.testing.expect(!ifMatchMatches("bad, \"abc:0\"", "abc:0")); +} + +test "ServeContent precondition precedence" { + try std.testing.expectEqual(PreconditionResult.proceed, checkPreconditions(.{ .if_match = "*", .if_unmodified_since = "Thu, 01 Jan 1970 00:00:00 GMT" }, "abc", 123)); + try std.testing.expectEqual(PreconditionResult.failed, checkPreconditions(.{ .if_unmodified_since = "Thu, 01 Jan 1970 00:00:00 GMT" }, "abc", 123)); + try std.testing.expectEqual(PreconditionResult.not_modified, checkPreconditions(.{ .if_none_match = "W/\"abc\"" }, "abc", 123)); + try std.testing.expectEqual(PreconditionResult.proceed, checkPreconditions(.{ .if_none_match = "\"other\"", .if_modified_since = "Thu, 01 Jan 1970 00:10:00 GMT" }, "abc", 123)); + try std.testing.expectEqual(PreconditionResult.proceed, checkPreconditions(.{ .if_modified_since = "Thu, 01 Jan 1970 00:10:00 GMT" }, "abc", 0)); } fn queryParam(query: []const u8, name: []const u8) ?[]const u8 { diff --git a/tests/backfill_api.py b/tests/backfill_api.py index 494ed8f..158756e 100644 --- a/tests/backfill_api.py +++ b/tests/backfill_api.py @@ -9,7 +9,9 @@ Set STREAM_BASE_URL and STREAM_DATA_DIR to exercise another loopback candidate without copying or mutating its archive. """ +import datetime import email +import email.utils import json, urllib.request, zstandard, hashlib, os, pathlib, sys SERVER = os.environ.get("STREAM_BASE_URL", "http://localhost:6018").rstrip("/") @@ -28,12 +30,16 @@ resp = urllib.request.urlopen(BASE + f"getSegment?name={s0['name']}") body = resp.read() assert resp.headers["ETag"] == f'"{s0["checksum"]}"', resp.headers["ETag"] disk = (DATA_DIR / "segments" / s0["name"]).read_bytes() +disk_stat = (DATA_DIR / "segments" / s0["name"]).stat() +last_modified = email.utils.formatdate(int(disk_stat.st_mtime), usegmt=True) +assert resp.headers["Last-Modified"] == last_modified, resp.headers assert hashlib.sha256(body).digest() == hashlib.sha256(disk).digest(), "served bytes != disk bytes" assert len(body) == s0["sizeBytes"] # getBlock: raw zstd frame that decompresses to a columnar block resp = urllib.request.urlopen(BASE + f"getBlock?segment={s0['name']}&blockIndex=0") frame = resp.read() +assert resp.headers["Last-Modified"] == last_modified, resp.headers block = zstandard.ZstdDecompressor().decompress(frame, max_output_size=1 << 30) n = int.from_bytes(block[:4], "little") assert n > 0, "block event_count" @@ -115,6 +121,8 @@ try: sys.exit("expected 416") except urllib.error.HTTPError as e: assert e.code == 416 + assert e.headers["Content-Range"] == f"bytes */{len(disk)}" + assert e.read() == b"invalid range: failed to overlap\n" print("range: PASS") @@ -211,5 +219,67 @@ try: sys.exit("expected weak If-None-Match to return 304") except urllib.error.HTTPError as e: assert e.code == 304 + assert e.headers.get("Last-Modified") is None -print("servecontent range semantics: PASS") + +def expect_error(url, code, headers): + try: + urllib.request.urlopen(urllib.request.Request(url, headers=headers)) + sys.exit(f"expected HTTP {code} for {url}") + except urllib.error.HTTPError as exc: + assert exc.code == code, (exc.code, exc.read()) + return exc + + +segment_url = BASE + f"getSegment?name={s0['name']}" +block_url = BASE + f"getBlock?segment={s0['name']}&blockIndex=0" +etag = f'"{s0["checksum"]}"' +old_date = email.utils.formatdate(int(disk_stat.st_mtime) - 60, usegmt=True) +future_date = email.utils.formatdate(int(disk_stat.st_mtime) + 60, usegmt=True) +modified_dt = datetime.datetime.fromtimestamp(int(disk_stat.st_mtime), datetime.UTC) +rfc850_date = modified_dt.strftime("%A, %d-%b-%y %H:%M:%S GMT") +asctime_date = modified_dt.strftime("%a %b %e %H:%M:%S %Y") + +# Pinned Go ServeContent evaluates representation preconditions in RFC order. +# If-Match suppresses If-Unmodified-Since, while any present If-None-Match +# suppresses If-Modified-Since even when its validator does not match. +for url in (segment_url, block_url): + assert urllib.request.urlopen(urllib.request.Request(url, headers={"If-Match": "*"})).status == 200 + assert urllib.request.urlopen(urllib.request.Request(url, headers={"If-Match": etag if url == segment_url else f'"{s0["checksum"]}:0"'})).status == 200 + failed_match = expect_error(url, 412, {"If-Match": '"other"'}) + assert failed_match.headers["Last-Modified"] == last_modified + assert failed_match.headers["ETag"] is not None + assert urllib.request.urlopen(urllib.request.Request(url, headers={ + "If-Match": "*", "If-Unmodified-Since": old_date, + })).status == 200 + expect_error(url, 412, {"If-Unmodified-Since": old_date}) + assert urllib.request.urlopen(urllib.request.Request(url, headers={"If-Unmodified-Since": future_date})).status == 200 + expect_error(url, 304, {"If-Modified-Since": last_modified}) + assert urllib.request.urlopen(urllib.request.Request(url, headers={"If-Modified-Since": old_date})).status == 200 + assert urllib.request.urlopen(urllib.request.Request(url, headers={ + "If-None-Match": '"other"', "If-Modified-Since": future_date, + })).status == 200 + +# Go's http.ParseTime accepts its two obsolete HTTP layouts and time.Parse's +# fractional-second extension even though generated Last-Modified is IMF-date. +for compatible_date in (rfc850_date, asctime_date, last_modified.replace(" GMT", ".5 GMT")): + expect_error(segment_url, 304, {"If-Modified-Since": compatible_date}) + +# If-Range accepts either the first strong ETag or a date exactly equal to the +# representation mtime. A stale date makes ServeContent ignore Range. +dated_range = urllib.request.urlopen(urllib.request.Request( + segment_url, headers={"Range": "bytes=0-0", "If-Range": last_modified}, +)) +assert dated_range.status == 206 and dated_range.read() == disk[:1] +stale_range = urllib.request.urlopen(urllib.request.Request( + segment_url, headers={"Range": "bytes=0-0", "If-Range": old_date}, +)) +assert stale_range.status == 200 and stale_range.read() == disk + +# ServeContent distinguishes malformed syntax from a valid range that simply +# does not overlap. Both are 416, but only the latter carries Content-Range. +malformed = expect_error(segment_url, 416, {"Range": "items=0-1"}) +assert malformed.headers.get("Content-Range") is None +assert malformed.read() == b"invalid range\n" + +print("servecontent conditional/range semantics: PASS") diff --git a/tests/http_metrics_contract.py b/tests/http_metrics_contract.py index 64ab3e0..20b40e4 100644 --- a/tests/http_metrics_contract.py +++ b/tests/http_metrics_contract.py @@ -127,6 +127,7 @@ def main() -> None: assert status == 200 and frame etag = headers["etag"] assert request(block_path, headers={"If-None-Match": etag})[0] == 304 + assert request(block_path, headers={"If-Match": '"other"'})[0] == 412 range_status, partial, _ = request(block_path, headers={"Range": "bytes=0-0"}) assert range_status == 206 and partial == frame[:1] assert request(block_path, headers={"Range": "bytes=999999999-"})[0] == 416 @@ -154,6 +155,7 @@ def main() -> None: ("xrpc/", "GET", "304"): 1, ("xrpc/", "GET", "206"): 1, ("xrpc/", "GET", "416"): 1, + ("xrpc/", "GET", "412"): 1, ("xrpc/", "GET", "400"): 2, ("xrpc/", "GET", "404"): 1, ("xrpc/", "POST", "405"): 1, @@ -179,11 +181,11 @@ def main() -> None: ) assert total_http_delta == sum(expected_http.values()), total_http_delta - expected_results = {"ok": 3, "bad_request": 1, "not_found": 1, "error": 1} + expected_results = {"ok": 3, "bad_request": 1, "not_found": 1, "error": 2} for result, expected in expected_results.items(): assert delta(before, after, "jetstream_getblock_requests_total", result=result) == expected assert delta(before, after, "jetstream_getblock_served_bytes_total") == len(frame) - assert delta(before, after, "jetstream_getblock_duration_seconds_count") == 6 + assert delta(before, after, "jetstream_getblock_duration_seconds_count") == 7 # Match upstream's manifest/file race semantics through the real server: # a segment which remains known to the resident manifest but disappears @@ -198,7 +200,7 @@ def main() -> None: print( "HTTP/getBlock metrics PASS: " - f"http={int(total_http_delta)} getBlock=7 full_bytes={len(frame)} websocket=1 storage_race=500" + f"http={int(total_http_delta)} getBlock=8 full_bytes={len(frame)} websocket=1 storage_race=500" )