From b3125d971a5732188cb9bacf9ab6af874f850e40 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Tue, 21 Jul 2026 18:27:47 -0500 Subject: [PATCH] match upstream cursor lookback --- README.md | 10 +++ docs/upstream-harness.md | 9 +++ justfile | 36 ++++++++++ src/internal/manifest.zig | 40 +++++++++++ src/internal/server.zig | 42 +++++++++-- src/main.zig | 18 +++-- tests/cursor_lookback_contract.py | 115 ++++++++++++++++++++++++++++++ 7 files changed, 260 insertions(+), 10 deletions(-) create mode 100644 tests/cursor_lookback_contract.py diff --git a/README.md b/README.md index 4f03a02..9203ac3 100644 --- a/README.md +++ b/README.md @@ -48,6 +48,7 @@ just run-sim # stream against it just e2e # python wire checks just archive-contract # archive XRPCs + resident manifest + pinned Go client just plan-config-contract # non-default planner limits/threshold, real process +just cursor-lookback-contract # v1/v2 replay-window behavior, real processes just status-contract # durable host rows + public HTTP view, offline just oracle # lifecycle crash + RocksDB fault recovery matrix just powerloss-image # one-time Linux NBD/ext4 tool-image bootstrap @@ -82,6 +83,15 @@ Zero disables non-empty DID or collection filters for the corresponding filter cap. For `--plan-max-entries`, zero instead means one unbounded page; the threshold must be greater than zero and at most one. +Subscriber replay matches upstream's 36-hour default lookback. A v1 sequence +cursor older than the retained window is conservatively clamped to the first +eligible sealed segment; v2 rejects the same cursor before WebSocket upgrade +with a `cursor too old` response. Timestamp cursors are clamped for both +protocols. Operators may change the window with a Go-style duration such as +`--cursor-lookback=72h`; zero makes both endpoints pure-live. The offline +`just cursor-lookback-contract` receipt exercises all three modes against real +sealed JSS through production Stream processes. + 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/docs/upstream-harness.md b/docs/upstream-harness.md index db2b562..0b9a2dc 100644 --- a/docs/upstream-harness.md +++ b/docs/upstream-harness.md @@ -165,6 +165,15 @@ filters), one-entry work-unit pagination, zero-entry-limit unbounded paging, and the 1.0-versus-0.75 whole-segment threshold boundary through public `planBackfill` responses. +`just cursor-lookback-contract` seeds three real sealed segments and starts +three production Stream processes. With the upstream 36-hour default it proves +that v2 rejects an old sequence cursor before upgrade while v1 clamps to the +oldest eligible segment. A widened window replays from the requested sequence; +`--cursor-lookback=0` upgrades without replay and records the disabled cursor +mode. Timestamp cursors use the same conservative segment floor and clamp on +both protocols. The receipt speaks raw WebSocket framing with Python's standard +library and needs no network access. + ## pinned differential oracle (2026-07-20) `just differential-oracle` refuses to run unless the upstream checkout is at diff --git a/justfile b/justfile index 1714b4b..25bcafe 100644 --- a/justfile +++ b/justfile @@ -152,6 +152,42 @@ plan-config-contract: echo "zero planner threshold unexpectedly started" >&2; exit 1 fi +# Prove default, widened, and disabled replay windows over real sealed JSS via +# pre-upgrade HTTP and actual WebSocket delivery. +cursor-lookback-contract: + #!/usr/bin/env bash + set -euo pipefail + DATA=$(mktemp -d) + LOG=$(mktemp) + PORT=${STREAM_CURSOR_LOOKBACK_PORT:-6023} + cleanup() { + if [[ -n "${pid:-}" ]]; then kill "$pid" 2>/dev/null || true; wait "$pid" 2>/dev/null || true; fi + rm -rf "$DATA" "$LOG" + } + trap cleanup EXIT + python3 -c 'import socket,sys; s=socket.socket(); s.bind(("127.0.0.1",int(sys.argv[1]))); s.close()' "$PORT" + zig build -Doptimize=ReleaseSafe + zig build write-sample -- --archive "$DATA" + run_case() { + local mode=$1; shift + ./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 & + pid=$! + for _ in {1..100}; do + if curl -fsS "http://127.0.0.1:$PORT/healthz" >/dev/null 2>&1; then break; fi + if ! kill -0 "$pid" 2>/dev/null; then cat "$LOG" >&2; exit 1; fi + sleep 0.1 + done + STREAM_CURSOR_LOOKBACK_PORT="$PORT" STREAM_CURSOR_LOOKBACK_MODE="$mode" \ + python3 tests/cursor_lookback_contract.py + kill "$pid"; wait "$pid"; unset pid + } + run_case default + run_case wide --cursor-lookback=100000h + run_case disabled --cursor-lookback=0 + # Seed the real RocksDB repo/host schema, reopen it through the production # binary, and exercise the public host/account views without network access. status-contract: diff --git a/src/internal/manifest.zig b/src/internal/manifest.zig index a678365..d542657 100644 --- a/src/internal/manifest.zig +++ b/src/internal/manifest.zig @@ -110,6 +110,11 @@ pub const Summary = struct { header: segment.Header, }; +pub const LookbackFloor = struct { + seq: u64 = 0, + witnessed_at: i64 = 0, +}; + const Metadata = struct { summary: Summary, blocks: *ResidentBlocks, @@ -164,6 +169,22 @@ pub const Manifest = struct { return result; } + /// Oldest segment conservatively retained by the cursor lookback. This + /// mirrors upstream: choose the first segment whose max witnessed time is + /// inside the window, or the freshest segment when the entire archive is + /// older. Returning its min bound may expose one extra segment, never less. + pub fn lookbackFloor(self: *Manifest, now_us: i64, lookback_ns: u64) LookbackFloor { + self.mu.lockUncancelable(self.io); + defer self.mu.unlock(self.io); + if (self.segments.items.len == 0) return .{}; + const lookback_us: i64 = @intCast(lookback_ns / std.time.ns_per_us); + const floor_time = now_us -| lookback_us; + const selected = for (self.segments.items) |metadata| { + if (metadata.summary.header.max_witnessed_at >= floor_time) break metadata.summary.header; + } else self.segments.items[self.segments.items.len - 1].summary.header; + return .{ .seq = selected.min_seq, .witnessed_at = selected.min_witnessed_at }; + } + pub fn segmentByIndex(self: *Manifest, idx: u64) ?Summary { self.mu.lockUncancelable(self.io); defer self.mu.unlock(self.io); @@ -460,6 +481,25 @@ test "resident manifest loads sealed metadata and measures real lookups" { try std.testing.expectEqual(@as(u64, 1), stats.manifest_block_index_cache_misses_total.load(.monotonic)); } +test "lookback floor is conservative and falls back to freshest segment" { + var threaded: Io.Threaded = .init(std.testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + var dir = try tmp.dir.createDirPathOpen(io, "segments", .{ .open_options = .{ .iterate = true } }); + defer dir.close(io); + try writeSegment(std.testing.allocator, io, dir, 0, &.{ 1, 2 }, true); + try writeSegment(std.testing.allocator, io, dir, 1, &.{ 3, 4 }, true); + try writeSegment(std.testing.allocator, io, dir, 2, &.{ 5, 6 }, true); + var manifest = try Manifest.init(std.testing.allocator, io, dir, null); + defer manifest.deinit(); + + try std.testing.expectEqual(@as(u64, 5), manifest.lookbackFloor(100, 55 * std.time.ns_per_us).seq); + try std.testing.expectEqual(@as(u64, 1), manifest.lookbackFloor(100, 150 * std.time.ns_per_us).seq); + try std.testing.expectEqual(@as(u64, 5), manifest.lookbackFloor(100, 10 * std.time.ns_per_us).seq); +} + test "block index handle remains valid across verified refresh" { var threaded: Io.Threaded = .init(std.testing.allocator, .{}); defer threaded.deinit(); diff --git a/src/internal/server.zig b/src/internal/server.zig index bf9514c..b51342d 100644 --- a/src/internal/server.zig +++ b/src/internal/server.zig @@ -99,6 +99,9 @@ pub const Hub = struct { bootstrap_enabled: bool = false, compaction_enabled: bool = false, retry_enabled: bool = false, + /// Upstream cursor replay window. Zero makes both endpoints pure-live and + /// classifies an otherwise-valid cursor as disabled. + cursor_lookback_ns: u64 = 36 * std.time.ns_per_hour, /// serving gate: false until the bootstrap lifecycle reaches /// steady_state — subscribe + xrpc return 503 (upstream contract); /// /healthz and /metrics stay reachable @@ -459,8 +462,19 @@ pub const Handler = struct { finishHttp(hub, http_started, http_handler, 503); return error.Close; } - if (err == error.CursorTooOld) + if (err == error.CursorTooOld) { _ = hub.stats.subscribe_cursor_requests.getPtr(.too_old).fetchAdd(1, .monotonic); + const requested = std.fmt.parseInt(i64, queryParam(query, "cursor").?, 10) catch 0; + const floor = if (hub.archive) |archive| + archive.manifest.lookbackFloor(Io.Timestamp.now(hub.io, .real).toMicroseconds(), hub.cursor_lookback_ns).seq + else + 0; + var body_buf: [192]u8 = undefined; + const body = std.fmt.bufPrint(&body_buf, "subscribe: cursor too old: cursor {d} below lookback floor {d}; re-backfill from your last seq\n", .{ requested, floor }) catch "subscribe: cursor too old\n"; + respond(conn, "400 Bad Request", "text/plain; charset=utf-8", body); + finishHttp(hub, http_started, http_handler, 400); + return error.Close; + } finishHttp(hub, http_started, http_handler, 400); return error.InvalidRequest; }; @@ -1308,8 +1322,10 @@ fn queryParam(query: []const u8, name: []const u8) ?[]const u8 { fn resolveCursor(hub: *Hub, v2: bool, raw_param: ?[]const u8) !CursorPlan { const raw = raw_param orelse return .{}; if (raw.len == 0) return .{}; + if (hub.cursor_lookback_ns == 0) return .{ .mode = .disabled }; const parsed = std.fmt.parseInt(i64, raw, 10) catch return error.InvalidCursor; if (parsed < 0) return error.InvalidCursor; + const now_us = Io.Timestamp.now(hub.io, .real).toMicroseconds(); // Both endpoints use the magnitude split. V2 differs only by rejecting // a seq cursor below the retained floor; v1 preserves its clamp. @@ -1327,6 +1343,14 @@ fn resolveCursor(hub: *Hub, v2: bool, raw_param: ?[]const u8) !CursorPlan { seq = 1; clamped = true; } + if (hub.archive) |archive| { + const floor = archive.manifest.lookbackFloor(now_us, hub.cursor_lookback_ns).seq; + if (floor != 0 and seq < floor) { + if (v2) return error.CursorTooOld; + seq = floor; + clamped = true; + } + } const gauges = hub.tail.gauges(); if (gauges.entries != 0) if (hub.tail.indexForSeq(seq)) |idx| return .{ .start_idx = idx, .mode = if (clamped) .clamped else .seq }; @@ -1340,21 +1364,27 @@ fn resolveCursor(hub: *Hub, v2: bool, raw_param: ?[]const u8) !CursorPlan { return .{ .start_idx = gauges.base, .mode = .clamped }; } - const now_us = Io.Timestamp.now(hub.io, .real).toMicroseconds(); if (parsed > now_us) return .{ .mode = .clamped }; if (hub.archive) |archive| { const resolved = cold.resolveTimeToSeq(hub.allocator, hub.io, archive, parsed) catch |err| { log.err("timestamp cursor resolution failed for {d}: {s}", .{ parsed, @errorName(err) }); return error.CursorResolveFailed; }; - if (hub.tail.indexForSeq(resolved.seq)) |idx| return .{ + var resolved_seq = resolved.seq; + var clamped = resolved.clamped; + const floor = archive.manifest.lookbackFloor(now_us, hub.cursor_lookback_ns).seq; + if (floor != 0 and resolved_seq < floor) { + resolved_seq = floor; + clamped = true; + } + if (hub.tail.indexForSeq(resolved_seq)) |idx| return .{ .start_idx = idx, - .mode = if (resolved.clamped) .clamped else .time_us, + .mode = if (clamped) .clamped else .time_us, }; return .{ .start_idx = hub.tail.gauges().base, - .cold_from_seq = resolved.seq, - .mode = if (resolved.clamped) .clamped else .time_us, + .cold_from_seq = resolved_seq, + .mode = if (clamped) .clamped else .time_us, }; } var mode: metrics.CursorMode = .time_us; diff --git a/src/main.zig b/src/main.zig index 025f815..b16acc1 100644 --- a/src/main.zig +++ b/src/main.zig @@ -90,9 +90,9 @@ test "explicit backfill repo parsing matches upstream selection rules" { 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 { +/// Parse the non-negative subset of Go's time.ParseDuration used by upstream's +/// duration flags. Like Go, fractional values are truncated below 1 ns. +fn parseDurationNanoseconds(raw: []const u8) !u64 { if (raw.len == 0) return error.InvalidDuration; var i: usize = 0; if (raw[i] == '+') { @@ -158,7 +158,12 @@ fn parseDurationCeilSeconds(raw: []const u8) !u64 { total += component; } if (tokens == 0) return error.InvalidDuration; - return @intCast((total + std.time.ns_per_s - 1) / std.time.ns_per_s); + return @intCast(total); +} + +fn parseDurationCeilSeconds(raw: []const u8) !u64 { + const nanoseconds = try parseDurationNanoseconds(raw); + return (nanoseconds + std.time.ns_per_s - 1) / std.time.ns_per_s; } test "segment cache duration matches Go duration and ceil policy" { @@ -174,6 +179,7 @@ test "segment cache duration matches Go duration and ceil policy" { try testing.expectError(error.InvalidDuration, parseDurationCeilSeconds("-1s")); try testing.expectError(error.InvalidDuration, parseDurationCeilSeconds("1")); try testing.expectError(error.InvalidDuration, parseDurationCeilSeconds("wat")); + try testing.expectEqual(@as(u64, 1_500_000_000), try parseDurationNanoseconds("1.5s")); } pub fn main(init: std.process.Init.Minimal) !void { @@ -213,6 +219,7 @@ pub fn main(init: std.process.Init.Minimal) !void { var store_fault_ordinal: u64 = 1; var segment_fault_spec: ?[]const u8 = null; var segment_cache_max_age_s: u64 = 0; + var cursor_lookback_ns: u64 = 36 * std.time.ns_per_hour; var plan_config: xrpcapi.PlanConfig = .{}; var arg_it = init.args.iterate(); @@ -246,6 +253,8 @@ pub fn main(init: std.process.Init.Minimal) !void { 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, "--cursor-lookback=")) { + cursor_lookback_ns = try parseDurationNanoseconds(arg["--cursor-lookback=".len..]); } else if (std.mem.startsWith(u8, arg, "--plan-max-dids=")) { plan_config.max_dids = try std.fmt.parseInt(usize, arg["--plan-max-dids=".len..], 10); } else if (std.mem.startsWith(u8, arg, "--plan-max-collections=")) { @@ -335,6 +344,7 @@ pub fn main(init: std.process.Init.Minimal) !void { .plc_url = plc_url, .data_dir = data_dir, .repo_action_rate_limits = repo_action_rate_limits, + .cursor_lookback_ns = cursor_lookback_ns, }; defer hub.deinit(); diff --git a/tests/cursor_lookback_contract.py b/tests/cursor_lookback_contract.py new file mode 100644 index 0000000..9ae14a3 --- /dev/null +++ b/tests/cursor_lookback_contract.py @@ -0,0 +1,115 @@ +"""Real WebSocket contract for Jetstream V2 cursor-lookback policy.""" + +import base64 +import hashlib +import json +import os +import socket +import urllib.request + + +HOST = "127.0.0.1" +PORT = int(os.environ["STREAM_CURSOR_LOOKBACK_PORT"]) +MODE = os.environ["STREAM_CURSOR_LOOKBACK_MODE"] + + +def handshake(path): + connection = socket.create_connection((HOST, PORT), timeout=2) + connection.settimeout(2) + key = base64.b64encode(os.urandom(16)).decode() + connection.sendall( + ( + f"GET {path} HTTP/1.1\r\n" + f"Host: {HOST}:{PORT}\r\n" + "Upgrade: websocket\r\n" + "Connection: Upgrade\r\n" + f"Sec-WebSocket-Key: {key}\r\n" + "Sec-WebSocket-Version: 13\r\n\r\n" + ).encode() + ) + response = b"" + while b"\r\n\r\n" not in response: + response += connection.recv(4096) + head, buffered = response.split(b"\r\n\r\n", 1) + lines = head.decode().split("\r\n") + headers = {} + for line in lines[1:]: + name, value = line.split(":", 1) + headers[name.lower()] = value.strip() + if lines[0].startswith("HTTP/1.1 101"): + expected = base64.b64encode( + hashlib.sha1((key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").encode()).digest() + ).decode() + assert headers["sec-websocket-accept"] == expected + return connection, lines[0], headers, buffered + + +def read_exact(connection, buffered, length): + while len(buffered) < length: + buffered += connection.recv(length - len(buffered)) + return buffered[:length], buffered[length:] + + +def read_frame(connection, buffered): + header, buffered = read_exact(connection, buffered, 2) + opcode = header[0] & 0x0F + length = header[1] & 0x7F + assert header[1] & 0x80 == 0, "server frames must not be masked" + if length == 126: + raw, buffered = read_exact(connection, buffered, 2) + length = int.from_bytes(raw, "big") + elif length == 127: + raw, buffered = read_exact(connection, buffered, 8) + length = int.from_bytes(raw, "big") + payload, buffered = read_exact(connection, buffered, length) + assert opcode in (1, 2) + return payload, buffered + + +if MODE == "default": + connection, status, headers, buffered = handshake("/subscribe-v2?cursor=1") + try: + assert status.startswith("HTTP/1.1 400") + length = int(headers["content-length"]) + body, _ = read_exact(connection, buffered, length) + text = body.decode() + assert "subscribe: cursor too old" in text + assert "cursor 1 below lookback floor 4001" in text + finally: + connection.close() + + connection, status, _, buffered = handshake("/subscribe?cursor=1") + try: + assert status.startswith("HTTP/1.1 101") + payload, _ = read_frame(connection, buffered) + event = json.loads(payload) + assert event["time_us"] == 1_700_000_004_001_000 + finally: + connection.close() +elif MODE == "wide": + connection, status, _, buffered = handshake("/subscribe-v2?cursor=1") + try: + assert status.startswith("HTTP/1.1 101") + payload, _ = read_frame(connection, buffered) + event = json.loads(payload) + assert event["seq"] == 1 + finally: + connection.close() +elif MODE == "disabled": + connection, status, _, buffered = handshake("/subscribe-v2?cursor=1") + try: + assert status.startswith("HTTP/1.1 101") + connection.settimeout(0.2) + try: + unexpected = buffered or connection.recv(1) + except TimeoutError: + unexpected = b"" + assert unexpected == b"", "disabled cursor unexpectedly replayed archive data" + metrics = urllib.request.urlopen(f"http://{HOST}:{PORT}/metrics").read().decode() + assert 'stream_subscribe_cursor_requests_total{mode="disabled"} 1' in metrics + finally: + connection.close() +else: + raise AssertionError(f"unknown mode: {MODE}") + +print(f"cursor lookback {MODE}: PASS") -- 2.51.2