diff --git a/docs/semantic-parity.md b/docs/semantic-parity.md index b9d47c3..088c75c 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 HTTP ranges. 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` 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. | | 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 9d692f6..361faa9 100644 --- a/docs/upstream-harness.md +++ b/docs/upstream-harness.md @@ -223,7 +223,11 @@ 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 receipt proves +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 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/server.zig b/src/internal/server.zig index 0ddc413..25ead18 100644 --- a/src/internal/server.zig +++ b/src/internal/server.zig @@ -945,6 +945,66 @@ const XrpcResponder = struct { if (body.len > 0) self.conn.writeFramed(body) catch return; } + pub fn multipartBytes(self: *XrpcResponder, body: []const u8, etag_hex: []const u8, boundary: *const [60]u8, ranges: anytype) void { + const encoded_len = multipartEncodedLength(boundary, ranges, body.len) orelse + return self.err(500, "InternalError", "multipart response too large"); + self.multipartHeader(etag_hex, boundary, encoded_len); + if (self.observation.code != 206) return; + for (ranges, 0..) |range, i| { + var part_buf: [320]u8 = undefined; + const part = multipartPartHeader(&part_buf, boundary, range.start, range.length, body.len, i == 0) orelse return; + self.conn.writeFramed(part) catch return; + const start: usize = @intCast(range.start); + const end: usize = @intCast(range.start + range.length); + if (end > start) self.conn.writeFramed(body[start..end]) catch return; + } + self.multipartClose(boundary); + } + + pub fn multipartFile(self: *XrpcResponder, io: Io, source: Io.File, etag_hex: []const u8, boundary: *const [60]u8, ranges: anytype, total: u64) void { + const encoded_len = multipartEncodedLength(boundary, ranges, total) orelse + return self.err(500, "InternalError", "multipart response too large"); + self.multipartHeader(etag_hex, boundary, encoded_len); + if (self.observation.code != 206) return; + var copy_buf: [64 * 1024]u8 = undefined; + for (ranges, 0..) |range, i| { + var part_buf: [320]u8 = undefined; + const part = multipartPartHeader(&part_buf, boundary, range.start, range.length, total, i == 0) orelse return; + self.conn.writeFramed(part) catch return; + var offset = range.start; + var remaining = range.length; + while (remaining > 0) { + const want: usize = @intCast(@min(remaining, copy_buf.len)); + const n = source.readPositional(io, &.{copy_buf[0..want]}, offset) catch return; + if (n == 0) return; + self.conn.writeFramed(copy_buf[0..n]) catch return; + offset += n; + remaining -= n; + } + } + self.multipartClose(boundary); + } + + fn multipartHeader(self: *XrpcResponder, etag_hex: []const u8, boundary: *const [60]u8, body_len: u64) void { + var buf: [512]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 }, + ) catch return; + self.observation.code = 206; + self.conn.writeFramed(header) catch { + self.observation.code = 500; + return; + }; + } + + fn multipartClose(self: *XrpcResponder, boundary: *const [60]u8) void { + var buf: [72]u8 = undefined; + const close = std.fmt.bufPrint(&buf, "\r\n--{s}--\r\n", .{boundary}) catch return; + self.conn.writeFramed(close) catch return; + } + /// Stream an immutable sealed file without retaining it per request. The /// upstream Go server reaches the same property through http.ServeContent /// and sendfile; chunking keeps Zig's memory bounded for slow clients. @@ -1022,6 +1082,28 @@ const XrpcResponder = struct { } }; +fn multipartPartHeader(buf: *[320]u8, boundary: *const [60]u8, start: u64, length: u64, total: u64, first: bool) ?[]const u8 { + const end = if (length == 0) start - 1 else start + length - 1; + return std.fmt.bufPrint( + buf, + "{s}--{s}\r\nContent-Range: bytes {d}-{d}/{d}\r\nContent-Type: application/octet-stream\r\n\r\n", + .{ if (first) "" else "\r\n", boundary, start, end, total }, + ) catch null; +} + +fn multipartEncodedLength(boundary: *const [60]u8, ranges: anytype, total: u64) ?u64 { + var encoded: u64 = 0; + for (ranges, 0..) |range, i| { + var buf: [320]u8 = undefined; + const part = multipartPartHeader(&buf, boundary, range.start, range.length, total, i == 0) orelse return null; + encoded = std.math.add(u64, encoded, part.len) catch return null; + encoded = std.math.add(u64, encoded, range.length) catch return null; + } + var close_buf: [72]u8 = undefined; + const close = std.fmt.bufPrint(&close_buf, "\r\n--{s}--\r\n", .{boundary}) catch return null; + return std.math.add(u64, encoded, close.len) catch null; +} + fn statusLine(status: u16) []const u8 { return switch (status) { 200 => "200 OK", diff --git a/src/internal/xrpcapi.zig b/src/internal/xrpcapi.zig index 6060b51..5100ef1 100644 --- a/src/internal/xrpcapi.zig +++ b/src/internal/xrpcapi.zig @@ -7,8 +7,8 @@ //! file is never loaded whole). the segment format is wire-visible contract; //! upstream's client parses these bytes with its own segment package. //! -//! getSegment supports single-range requests (bytes=a-b/a-/-n) with 206 + -//! Content-Range; multi-range is declined (416 falls back to full GET). +//! getSegment and getBlock match upstream's http.ServeContent byte-range +//! contract, including multipart/byteranges and mixed overlap handling. const std = @import("std"); const archive_mod = @import("archive.zig"); @@ -252,10 +252,27 @@ pub const Api = struct { if (if_range) |candidate| { if (!etagMatches(candidate, etag)) break :range; } - const r = parseRange(raw, stat.size) orelse - return respond.rangeNotSatisfiable(stat.size); - respond.file(self.io, file, etag, r.start, r.end, stat.size, true); - return; + 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), + }; + defer self.allocator.free(ranges); + if (sumRangeLengths(ranges) <= stat.size) switch (ranges.len) { + 0 => break :range, + 1 => { + const r = ranges[0]; + respond.file(self.io, file, etag, r.start, r.end(), stat.size, true); + return; + }, + else => { + var entropy: [30]u8 = undefined; + self.io.randomSecure(&entropy) catch + return respond.err(500, "InternalError", "failed to generate multipart boundary"); + const boundary = std.fmt.bytesToHex(entropy, .lower); + respond.multipartFile(self.io, file, etag, &boundary, ranges, stat.size); + return; + }, + }; } respond.file(self.io, file, etag, 0, stat.size - 1, stat.size, false); @@ -309,10 +326,27 @@ pub const Api = struct { if (if_range) |candidate| { if (!etagMatches(candidate, etag)) break :range; } - const r = parseRange(raw, frame.len) orelse - return respond.rangeNotSatisfiable(frame.len); - respond.partial(frame[@intCast(r.start)..@intCast(r.end + 1)], etag, r.start, r.end, frame.len); - return; + 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), + }; + defer self.allocator.free(ranges); + if (sumRangeLengths(ranges) <= frame.len) switch (ranges.len) { + 0 => break :range, + 1 => { + const r = ranges[0]; + respond.partial(frame[@intCast(r.start)..@intCast(r.start + r.length)], etag, r.start, r.end(), frame.len); + return; + }, + else => { + var entropy: [30]u8 = undefined; + self.io.randomSecure(&entropy) catch + return respond.err(500, "InternalError", "failed to generate multipart boundary"); + const boundary = std.fmt.bytesToHex(entropy, .lower); + respond.multipartBytes(frame, etag, &boundary, ranges); + return; + }, + }; } respond.raw(200, frame, etag); } @@ -322,40 +356,78 @@ pub fn checksumHex(buf: *[16]u8, checksum: u64) []const u8 { return std.fmt.bufPrint(buf, "{x:0>16}", .{checksum}) catch unreachable; } -const ByteRange = struct { start: u64, end: u64 }; - -/// parse a single-range "bytes=a-b" / "bytes=a-" / "bytes=-n" header -fn parseRange(raw: []const u8, size: u64) ?ByteRange { - if (size == 0) return null; - if (!std.mem.startsWith(u8, raw, "bytes=")) return null; - const spec = raw["bytes=".len..]; - if (std.mem.indexOfScalar(u8, spec, ',') != null) return null; // multi-range unsupported - const dash = std.mem.indexOfScalar(u8, spec, '-') orelse return null; - const start_str = spec[0..dash]; - const end_str = spec[dash + 1 ..]; - if (start_str.len == 0) { - // suffix range: last n bytes - const n = std.fmt.parseInt(u64, end_str, 10) catch return null; - if (n == 0) return null; - const start = if (n >= size) 0 else size - n; - return .{ .start = start, .end = size - 1 }; +const ByteRange = struct { + start: u64, + length: u64, + + fn end(self: ByteRange) u64 { + // Go's ServeContent preserves a zero-length suffix range as + // `bytes=size-(size-1)/size` with an empty body. + return if (self.length == 0) self.start - 1 else self.start + self.length - 1; + } +}; + +const RangeParseError = error{ InvalidRange, NoOverlap, OutOfMemory }; + +/// Exact shape of Go net/http's parseRange, which pinned upstream calls for +/// both getSegment and getBlock. Non-overlapping members are discarded when +/// another member overlaps; malformed syntax rejects the complete request. +fn parseRanges(allocator: Allocator, raw: []const u8, size: u64) RangeParseError![]ByteRange { + if (!std.mem.startsWith(u8, raw, "bytes=")) return error.InvalidRange; + var ranges: std.ArrayList(ByteRange) = .empty; + errdefer ranges.deinit(allocator); + var no_overlap = false; + var specs = std.mem.splitScalar(u8, raw["bytes=".len..], ','); + while (specs.next()) |spec_raw| { + const spec = std.mem.trim(u8, spec_raw, " \t"); + if (spec.len == 0) continue; + const dash = std.mem.indexOfScalar(u8, spec, '-') orelse return error.InvalidRange; + const start_raw = std.mem.trim(u8, spec[0..dash], " \t"); + const end_raw = std.mem.trim(u8, spec[dash + 1 ..], " \t"); + if (start_raw.len == 0) { + if (end_raw.len == 0 or end_raw[0] == '-') return error.InvalidRange; + const suffix_signed = std.fmt.parseInt(i64, end_raw, 10) catch return error.InvalidRange; + if (suffix_signed < 0) return error.InvalidRange; + const suffix: u64 = @intCast(suffix_signed); + const length = @min(suffix, size); + try ranges.append(allocator, .{ .start = size - length, .length = length }); + continue; + } + const start_signed = std.fmt.parseInt(i64, start_raw, 10) catch return error.InvalidRange; + if (start_signed < 0) return error.InvalidRange; + const start: u64 = @intCast(start_signed); + if (start >= size) { + no_overlap = true; + continue; + } + const length = if (end_raw.len == 0) + size - start + else length: { + const requested_end_signed = std.fmt.parseInt(i64, end_raw, 10) catch return error.InvalidRange; + if (requested_end_signed < 0) return error.InvalidRange; + const requested_end: u64 = @intCast(requested_end_signed); + if (requested_end < start) return error.InvalidRange; + break :length @min(requested_end, size - 1) - start + 1; + }; + try ranges.append(allocator, .{ .start = start, .length = length }); } - const start = std.fmt.parseInt(u64, start_str, 10) catch return null; - if (start >= size) return null; - const end = if (end_str.len == 0) size - 1 else @min(std.fmt.parseInt(u64, end_str, 10) catch return null, size - 1); - if (end < start) return null; - return .{ .start = start, .end = end }; + if (no_overlap and ranges.items.len == 0) return error.NoOverlap; + return ranges.toOwnedSlice(allocator); +} + +fn sumRangeLengths(ranges: []const ByteRange) u64 { + var total: u64 = 0; + for (ranges) |range| total +|= range.length; + return total; } fn etagMatches(raw: []const u8, etag: []const u8) bool { - var it = std.mem.splitScalar(u8, raw, ','); - while (it.next()) |part_raw| { - const part = std.mem.trim(u8, part_raw, " \t"); - if (std.mem.eql(u8, part, "*")) return true; - 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; - } - return false; + // 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); } /// If-None-Match uses weak comparison (RFC 9110 ยง13.1.2), unlike If-Range. @@ -371,19 +443,35 @@ fn ifNoneMatchMatches(raw: []const u8, etag: []const u8) bool { return false; } -test "parseRange forms" { - try std.testing.expectEqual(ByteRange{ .start = 0, .end = 99 }, parseRange("bytes=0-99", 1000).?); - try std.testing.expectEqual(ByteRange{ .start = 500, .end = 999 }, parseRange("bytes=500-", 1000).?); - try std.testing.expectEqual(ByteRange{ .start = 900, .end = 999 }, parseRange("bytes=-100", 1000).?); - try std.testing.expectEqual(ByteRange{ .start = 0, .end = 999 }, parseRange("bytes=0-5000", 1000).?); - try std.testing.expect(parseRange("bytes=1000-", 1000) == null); - try std.testing.expect(parseRange("bytes=0-1,5-9", 1000) == null); +test "parseRanges matches ServeContent forms" { + const allocator = std.testing.allocator; + const Case = struct { raw: []const u8, want: []const ByteRange }; + const cases = [_]Case{ + .{ .raw = "bytes=0-99", .want = &.{.{ .start = 0, .length = 100 }} }, + .{ .raw = "bytes=500-", .want = &.{.{ .start = 500, .length = 500 }} }, + .{ .raw = "bytes=-100", .want = &.{.{ .start = 900, .length = 100 }} }, + .{ .raw = "bytes=-0", .want = &.{.{ .start = 1000, .length = 0 }} }, + .{ .raw = "bytes=0-5000", .want = &.{.{ .start = 0, .length = 1000 }} }, + .{ .raw = "bytes=0-1, 5-9", .want = &.{ .{ .start = 0, .length = 2 }, .{ .start = 5, .length = 5 } } }, + .{ .raw = "bytes=0-1, 1000-", .want = &.{.{ .start = 0, .length = 2 }} }, + .{ .raw = "bytes=", .want = &.{} }, + }; + for (cases) |case| { + const got = try parseRanges(allocator, case.raw, 1000); + defer allocator.free(got); + try std.testing.expectEqualSlices(ByteRange, case.want, got); + } + try std.testing.expectError(error.NoOverlap, parseRanges(allocator, "bytes=1000-", 1000)); + try std.testing.expectError(error.InvalidRange, parseRanges(allocator, "items=0-1", 1000)); + try std.testing.expectError(error.InvalidRange, parseRanges(allocator, "bytes=4-2", 1000)); + try std.testing.expectError(error.InvalidRange, parseRanges(allocator, "bytes=9223372036854775808-", 1000)); } test "strong etag matching" { try std.testing.expect(etagMatches("\"abc:0\"", "abc:0")); - try std.testing.expect(etagMatches("\"other\", \"abc:0\"", "abc:0")); - try std.testing.expect(etagMatches("*", "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")); } diff --git a/tests/backfill_api.py b/tests/backfill_api.py index 88b2133..494ed8f 100644 --- a/tests/backfill_api.py +++ b/tests/backfill_api.py @@ -9,6 +9,7 @@ Set STREAM_BASE_URL and STREAM_DATA_DIR to exercise another loopback candidate without copying or mutating its archive. """ +import email import json, urllib.request, zstandard, hashlib, os, pathlib, sys SERVER = os.environ.get("STREAM_BASE_URL", "http://localhost:6018").rstrip("/") @@ -115,3 +116,100 @@ try: except urllib.error.HTTPError as e: assert e.code == 416 print("range: PASS") + + +def multipart_ranges(url, representation, specs): + raw = ", ".join(f"{start}-{end}" for start, end in specs) + response = urllib.request.urlopen( + urllib.request.Request(url, headers={"Range": "bytes=" + raw}) + ) + payload = response.read() + content_type = response.headers["Content-Type"] + assert response.status == 206 + assert content_type.startswith("multipart/byteranges; boundary=") + assert response.headers.get("Content-Range") is None + assert int(response.headers["Content-Length"]) == len(payload) + message = email.message_from_bytes( + b"Content-Type: " + content_type.encode() + b"\r\nMIME-Version: 1.0\r\n\r\n" + payload + ) + parts = message.get_payload() + assert len(parts) == len(specs) + for part, (start, end) in zip(parts, specs): + assert part.get_content_type() == "application/octet-stream" + assert part["Content-Range"] == f"bytes {start}-{end}/{len(representation)}" + assert part.get_payload(decode=True) == representation[start:end + 1] + + +# Pinned upstream delegates both archive byte endpoints to http.ServeContent. +# Multiple satisfiable ranges therefore produce a real multipart/byteranges +# response rather than 416; exercise the physical file and raw zstd frame. +multipart_ranges( + BASE + f"getSegment?name={s0['name']}", + disk, + [(0, 15), (len(disk) - 16, len(disk) - 1)], +) +multipart_ranges( + BASE + f"getBlock?segment={s0['name']}&blockIndex=0", + frame, + [(0, 7), (len(frame) - 8, len(frame) - 1)], +) + +# ServeContent discards non-overlapping members when another member overlaps, +# and ignores the entire Range header when duplicate ranges ask it to transfer +# more bytes than the representation itself. +mixed = urllib.request.urlopen( + urllib.request.Request( + BASE + f"getSegment?name={s0['name']}", + headers={"Range": f"bytes=0-0, {len(disk) + 10}-"}, + ) +) +assert mixed.status == 206 and mixed.read() == disk[:1] +duplicate = urllib.request.urlopen( + urllib.request.Request( + BASE + f"getSegment?name={s0['name']}", + headers={"Range": f"bytes=0-{len(disk) - 1},0-0"}, + ) +) +assert duplicate.status == 200 and duplicate.read() == disk + +# If-Range uses strong comparison: a wildcard is not a validator. If-None- +# Match uses weak comparison and therefore accepts the weak form of the ETag. +wildcard = urllib.request.urlopen( + urllib.request.Request( + BASE + f"getSegment?name={s0['name']}", + headers={"Range": "bytes=0-0", "If-Range": "*"}, + ) +) +assert wildcard.status == 200 and wildcard.read() == disk +later_match = urllib.request.urlopen( + urllib.request.Request( + BASE + f"getSegment?name={s0['name']}", + headers={ + "Range": "bytes=0-0", + "If-Range": f'"other", "{s0["checksum"]}"', + }, + ) +) +assert later_match.status == 200 and later_match.read() == disk +first_match = urllib.request.urlopen( + urllib.request.Request( + BASE + f"getSegment?name={s0['name']}", + headers={ + "Range": "bytes=0-0", + "If-Range": f'"{s0["checksum"]}", "other"', + }, + ) +) +assert first_match.status == 206 and first_match.read() == disk[:1] +try: + urllib.request.urlopen( + urllib.request.Request( + BASE + f"getSegment?name={s0['name']}", + headers={"If-None-Match": f'W/"{s0["checksum"]}"'}, + ) + ) + sys.exit("expected weak If-None-Match to return 304") +except urllib.error.HTTPError as e: + assert e.code == 304 + +print("servecontent range semantics: PASS")