diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -60,6 +60,14 @@ RocksDB, zstd, xxHash, and websocket.zig's zlib 1.3.2; it does not depend on a system zlib installation. +Sealed archive downloads use upstream's cache policy. The default is +`Cache-Control: public, no-cache`; operators may set a positive Go-style +duration such as `--segment-cache-max-age=15m`. As upstream does, Stream rounds +positive durations up to whole seconds in `max-age`. This is most useful when +compaction is disabled or the cache lifetime is comfortably shorter than the +configured compaction interval, because compaction can replace a segment while +retaining its name. + The strict power-loss tier is deliberately separate from ordinary unit and process tests. It cross-builds the production ReleaseSafe Linux binary, runs RocksDB and JSS on ext4 over a kernel NBD device, kills the block backend before diff --git a/justfile b/justfile --- a/justfile +++ b/justfile @@ -65,7 +65,7 @@ ./zig-out/bin/stream --port="$PORT" --data-dir="$DATA" \ --upstream=ws://127.0.0.1:17999 --plc=http://127.0.0.1:17999 \ --relay-http=http://127.0.0.1:17999 --compaction-interval=0 \ - --retry-interval=0 --no-verify >"$LOG" 2>&1 & + --retry-interval=0 --segment-cache-max-age=1500ms --no-verify >"$LOG" 2>&1 & pid=$! for _ in {1..100}; do if curl -fsS "http://127.0.0.1:$PORT/healthz" >/dev/null 2>&1; then break; fi diff --git a/docs/semantic-parity.md b/docs/semantic-parity.md --- a/docs/semantic-parity.md +++ b/docs/semantic-parity.md @@ -14,7 +14,7 @@ |---|---|---| | 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 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. | +| Archive XRPC | Official Go client plus Stream conformance replay over listSegments/getSegment/getBlock/planBackfill | **closed for wire behavior and archive cache policy.** 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`. A separate production-process receipt passes `--segment-cache-max-age=1500ms` and requires upstream's ceil-to-seconds `public, max-age=2` response; the default remains `public, no-cache`, and 416 clears the representation cache header. 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. Remaining non-default planner limits stay 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 --- a/docs/upstream-harness.md +++ b/docs/upstream-harness.md @@ -153,6 +153,11 @@ segment data instead of using its refcounted resident envelopes, DID blooms, and collection summaries. +`just http-metrics-contract` launches the real candidate with +`--segment-cache-max-age=1500ms` and requires `getBlock` to return +`Cache-Control: public, max-age=2`. This records upstream's positive-duration +ceil behavior independently of the default `public, no-cache` archive receipt. + ## pinned differential oracle (2026-07-20) `just differential-oracle` refuses to run unless the upstream checkout is at diff --git a/src/main.zig b/src/main.zig --- a/src/main.zig +++ b/src/main.zig @@ -90,6 +90,92 @@ try testing.expectError(error.InvalidBackfillRepo, parseBackfillRepos(testing.allocator, "not-a-did")); } +/// Parse the non-negative subset of Go's time.ParseDuration accepted by +/// upstream's --segment-cache-max-age, then apply its ceil-to-seconds policy. +fn parseDurationCeilSeconds(raw: []const u8) !u64 { + if (raw.len == 0) return error.InvalidDuration; + var i: usize = 0; + if (raw[i] == '+') { + i += 1; + if (i == raw.len) return error.InvalidDuration; + } else if (raw[i] == '-') return error.InvalidDuration; + if (std.mem.eql(u8, raw[i..], "0")) return 0; + + const Unit = struct { name: []const u8, nanoseconds: u64 }; + const units = [_]Unit{ + .{ .name = "ns", .nanoseconds = 1 }, + .{ .name = "us", .nanoseconds = std.time.ns_per_us }, + .{ .name = "µs", .nanoseconds = std.time.ns_per_us }, + .{ .name = "μs", .nanoseconds = std.time.ns_per_us }, + .{ .name = "ms", .nanoseconds = std.time.ns_per_ms }, + .{ .name = "s", .nanoseconds = std.time.ns_per_s }, + .{ .name = "m", .nanoseconds = std.time.ns_per_min }, + .{ .name = "h", .nanoseconds = std.time.ns_per_hour }, + }; + const limit: u128 = std.math.maxInt(i64); + var total: u128 = 0; + var tokens: usize = 0; + while (i < raw.len) : (tokens += 1) { + var whole: u128 = 0; + var digits: usize = 0; + while (i < raw.len and std.ascii.isDigit(raw[i])) : (i += 1) { + const digit = raw[i] - '0'; + if (whole > (limit - digit) / 10) return error.DurationOverflow; + whole = whole * 10 + digit; + digits += 1; + } + + var fraction: u128 = 0; + var fraction_scale: u128 = 1; + if (i < raw.len and raw[i] == '.') { + i += 1; + while (i < raw.len and std.ascii.isDigit(raw[i])) : (i += 1) { + digits += 1; + // Nanosecond resolution never needs more than 18 decimal + // places for the largest supported unit; Go truncates below + // one nanosecond as well. + if (fraction_scale < 1_000_000_000_000_000_000) { + fraction = fraction * 10 + (raw[i] - '0'); + fraction_scale *= 10; + } + } + } + if (digits == 0) return error.InvalidDuration; + + var unit: ?Unit = null; + for (units) |candidate| { + if (std.mem.startsWith(u8, raw[i..], candidate.name)) { + unit = candidate; + i += candidate.name.len; + break; + } + } + const selected = unit orelse return error.InvalidDuration; + if (whole > limit / selected.nanoseconds) return error.DurationOverflow; + var component = whole * selected.nanoseconds; + component += (fraction * selected.nanoseconds) / fraction_scale; + if (component > limit - total) return error.DurationOverflow; + total += component; + } + if (tokens == 0) return error.InvalidDuration; + return @intCast((total + std.time.ns_per_s - 1) / std.time.ns_per_s); +} + +test "segment cache duration matches Go duration and ceil policy" { + const testing = std.testing; + try testing.expectEqual(@as(u64, 0), try parseDurationCeilSeconds("0")); + try testing.expectEqual(@as(u64, 1), try parseDurationCeilSeconds("500ms")); + try testing.expectEqual(@as(u64, 2), try parseDurationCeilSeconds("1500ms")); + try testing.expectEqual(@as(u64, 3600), try parseDurationCeilSeconds("1h")); + try testing.expectEqual(@as(u64, 5400), try parseDurationCeilSeconds("1h30m")); + try testing.expectEqual(@as(u64, 1), try parseDurationCeilSeconds(".1s")); + try testing.expectEqual(@as(u64, 1), try parseDurationCeilSeconds("1us")); + try testing.expectEqual(@as(u64, 1), try parseDurationCeilSeconds("1µs")); + try testing.expectError(error.InvalidDuration, parseDurationCeilSeconds("-1s")); + try testing.expectError(error.InvalidDuration, parseDurationCeilSeconds("1")); + try testing.expectError(error.InvalidDuration, parseDurationCeilSeconds("wat")); +} + pub fn main(init: std.process.Init.Minimal) !void { const allocator = std.heap.c_allocator; @@ -126,6 +212,7 @@ var store_fault_prefix: ?[]const u8 = null; var store_fault_ordinal: u64 = 1; var segment_fault_spec: ?[]const u8 = null; + var segment_cache_max_age_s: u64 = 0; var arg_it = init.args.iterate(); _ = arg_it.next(); // program name @@ -156,6 +243,8 @@ compaction_interval_s = try std.fmt.parseInt(u64, arg["--compaction-interval=".len..], 10); } else if (std.mem.startsWith(u8, arg, "--retry-interval=")) { retry_interval_s = try std.fmt.parseInt(u64, arg["--retry-interval=".len..], 10); + } else if (std.mem.startsWith(u8, arg, "--segment-cache-max-age=")) { + segment_cache_max_age_s = try parseDurationCeilSeconds(arg["--segment-cache-max-age=".len..]); } else if (std.mem.startsWith(u8, arg, "--timestamp-import-token=")) { timestamp_import_token = arg["--timestamp-import-token=".len..]; } else if (std.mem.startsWith(u8, arg, "--timestamp-import-dir=")) { @@ -252,7 +341,12 @@ hot_tail.enableDurabilityGate(archive.committed_seq.load(.acquire)); log.info("archive at {s}/segments (next seq {d})", .{ data_dir, archive.next_seq }); hub.archive = &archive; - var xrpc_api: xrpcapi.Api = .{ .allocator = allocator, .io = io, .archive = &archive }; + var xrpc_api: xrpcapi.Api = .{ + .allocator = allocator, + .io = io, + .archive = &archive, + .segment_cache_max_age_s = segment_cache_max_age_s, + }; hub.xrpc = &xrpc_api; var meta = try meta_store.Store.openWithStats(allocator, io, data_dir, &stats); diff --git a/tests/http_metrics_contract.py b/tests/http_metrics_contract.py --- a/tests/http_metrics_contract.py +++ b/tests/http_metrics_contract.py @@ -120,17 +120,27 @@ segment = listing["segments"][0] name = urllib.parse.quote(segment["name"]) + status, sealed, headers = request( + f"/xrpc/network.bsky.jetstream.getSegment?name={name}" + ) + assert status == 200 and sealed + assert headers["cache-control"] == "public, max-age=2" + before = scrape() block_path = f"/xrpc/network.bsky.jetstream.getBlock?segment={name}&blockIndex=0" status, frame, headers = request(block_path) assert status == 200 and frame + assert headers["cache-control"] == "public, max-age=2" etag = headers["etag"] - assert request(block_path, headers={"If-None-Match": etag})[0] == 304 - assert request(block_path, headers={"If-Match": '"other"'})[0] == 412 + status, _, headers = request(block_path, headers={"If-None-Match": etag}) + assert status == 304 and headers["cache-control"] == "public, max-age=2" + status, _, headers = request(block_path, headers={"If-Match": '"other"'}) + assert status == 412 and headers["cache-control"] == "public, max-age=2" 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 + status, _, headers = request(block_path, headers={"Range": "bytes=999999999-"}) + assert status == 416 and "cache-control" not in headers 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] == 501 diff --git a/src/internal/server.zig b/src/internal/server.zig --- a/src/internal/server.zig +++ b/src/internal/server.zig @@ -904,10 +904,20 @@ conn: *websocket.Conn, observation: *HttpObservation, modified_s: ?i64 = null, + cache_max_age_s: u64 = 0, 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; + } + + pub fn archiveCacheMaxAge(self: *XrpcResponder, max_age_s: u64) void { + self.cache_max_age_s = max_age_s; + } + + fn cacheControl(self: *const XrpcResponder, buf: *[64]u8) []const u8 { + if (self.cache_max_age_s == 0) return "public, no-cache"; + return std.fmt.bufPrint(buf, "public, max-age={d}", .{self.cache_max_age_s}) catch "public, no-cache"; } fn lastModifiedHeader(self: *const XrpcResponder, buf: *[64]u8) []const u8 { @@ -924,11 +934,13 @@ pub fn raw(self: *XrpcResponder, status: u16, body: []const u8, etag_hex: []const u8) void { var modified_buf: [64]u8 = undefined; const modified = self.lastModifiedHeader(&modified_buf); + var cache_buf: [64]u8 = undefined; + const cache = self.cacheControl(&cache_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\n{s}Connection: close\r\nServer: stream\r\n\r\n", - .{ statusLine(status), body.len, etag_hex, modified }, + "HTTP/1.1 {s}\r\nContent-Type: application/octet-stream\r\nContent-Length: {d}\r\nAccept-Ranges: bytes\r\nETag: \"{s}\"\r\nCache-Control: {s}\r\n{s}Connection: close\r\nServer: stream\r\n\r\n", + .{ statusLine(status), body.len, etag_hex, cache, modified }, ) catch return; self.observation.code = status; self.conn.writeFramed(header) catch return; @@ -958,11 +970,13 @@ pub fn partial(self: *XrpcResponder, body: []const u8, etag_hex: []const u8, start: u64, end: u64, total: u64) void { var modified_buf: [64]u8 = undefined; const modified = self.lastModifiedHeader(&modified_buf); + var cache_buf: [64]u8 = undefined; + const cache = self.cacheControl(&cache_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\n{s}Connection: close\r\nServer: stream\r\n\r\n", - .{ body.len, start, end, total, etag_hex, modified }, + "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: {s}\r\n{s}Connection: close\r\nServer: stream\r\n\r\n", + .{ body.len, start, end, total, etag_hex, cache, modified }, ) catch return; self.observation.code = 206; self.conn.writeFramed(header) catch return; @@ -1012,11 +1026,13 @@ fn multipartHeader(self: *XrpcResponder, etag_hex: []const u8, boundary: *const [60]u8, body_len: u64) void { var modified_buf: [64]u8 = undefined; const modified = self.lastModifiedHeader(&modified_buf); + var cache_buf: [64]u8 = undefined; + const cache = self.cacheControl(&cache_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\n{s}Connection: close\r\nServer: stream\r\n\r\n", - .{ boundary, body_len, etag_hex, modified }, + "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: {s}\r\n{s}Connection: close\r\nServer: stream\r\n\r\n", + .{ boundary, body_len, etag_hex, cache, modified }, ) catch return; self.observation.code = 206; self.conn.writeFramed(header) catch { @@ -1041,18 +1057,20 @@ const len = end - start + 1; var modified_buf: [64]u8 = undefined; const modified = self.lastModifiedHeader(&modified_buf); + var cache_buf: [64]u8 = undefined; + const cache = self.cacheControl(&cache_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\n{s}Connection: close\r\nServer: stream\r\n\r\n", - .{ len, start, end, total, etag_hex, modified }, + "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: {s}\r\n{s}Connection: close\r\nServer: stream\r\n\r\n", + .{ len, start, end, total, etag_hex, cache, 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\n{s}Connection: close\r\nServer: stream\r\n\r\n", - .{ len, etag_hex, modified }, + "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: {s}\r\n{s}Connection: close\r\nServer: stream\r\n\r\n", + .{ len, etag_hex, cache, modified }, ) catch return; self.observation.code = if (partial_response) 206 else 200; self.conn.writeFramed(header) catch return; @@ -1094,22 +1112,26 @@ pub fn preconditionFailed(self: *XrpcResponder, etag_hex: []const u8) void { var modified_buf: [64]u8 = undefined; const modified = self.lastModifiedHeader(&modified_buf); + var cache_buf: [64]u8 = undefined; + const cache = self.cacheControl(&cache_buf); var buf: [384]u8 = undefined; const header = std.fmt.bufPrint( &buf, - "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 }, + "HTTP/1.1 412 Precondition Failed\r\nContent-Type: application/octet-stream\r\nETag: \"{s}\"\r\nCache-Control: {s}\r\n{s}Content-Length: 0\r\nConnection: close\r\nServer: stream\r\n\r\n", + .{ etag_hex, cache, modified }, ) catch return; self.observation.code = 412; self.conn.writeFramed(header) catch return; } pub fn notModified(self: *XrpcResponder, etag_hex: []const u8) void { + var cache_buf: [64]u8 = undefined; + const cache = self.cacheControl(&cache_buf); var buf: [256]u8 = undefined; const header = std.fmt.bufPrint( &buf, - "HTTP/1.1 304 Not Modified\r\nETag: \"{s}\"\r\nCache-Control: public, no-cache\r\nConnection: close\r\nServer: stream\r\n\r\n", - .{etag_hex}, + "HTTP/1.1 304 Not Modified\r\nETag: \"{s}\"\r\nCache-Control: {s}\r\nConnection: close\r\nServer: stream\r\n\r\n", + .{ etag_hex, cache }, ) catch return; self.observation.code = 304; self.conn.writeFramed(header) catch return; diff --git a/src/internal/xrpcapi.zig b/src/internal/xrpcapi.zig --- a/src/internal/xrpcapi.zig +++ b/src/internal/xrpcapi.zig @@ -36,6 +36,7 @@ allocator: Allocator, io: Io, archive: *archive_mod.Archive, + segment_cache_max_age_s: u64 = 0, import: ?*import_manager.Manager = null, import_token: []const u8 = "", /// Keep one manifest planning pass in flight so public callers cannot @@ -254,6 +255,7 @@ const etag = checksumHex(&hex_buf, header.checksum); const modified_s = stat.mtime.toSeconds(); respond.archiveModified(modified_s); + respond.archiveCacheMaxAge(self.segment_cache_max_age_s); switch (checkPreconditions(preconditions, etag, modified_s)) { .proceed => {}, .not_modified => return respond.notModified(etag), @@ -336,6 +338,7 @@ const etag = std.fmt.bufPrint(&etag_buf, "{s}:{d}", .{ checksumHex(&checksum_buf, header.checksum), block_idx }) catch unreachable; const modified_s = manifest_ref.mtime.toSeconds(); respond.archiveModified(modified_s); + respond.archiveCacheMaxAge(self.segment_cache_max_age_s); switch (checkPreconditions(preconditions, etag, modified_s)) { .proceed => {}, .not_modified => return respond.notModified(etag),