From 98e9f070e591037c047eccd4eb231ea64f4b6bf7 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Thu, 13 Aug 2026 14:45:29 -0500 Subject: [PATCH] re-pin to v0.2.0 (the Jetstream v2 release) and port kinds filtering upstream shipped officially 2026-08-13 (Bluesky Protocol Services). the waypoint moves d4dd2f0 -> 289b032 (v0.2.0); the server-side delta is PR #336, end-to-end kinds filtering, ported here for planSnapshot: a kinds mask on the request (unknown values and collections+kinds-excluding-commit are request errors, upstream's exact message), block selection admits sentinel $-collections per-kind instead of unconditionally (kinds=commit skips marker-only blocks; the marker-safe default keeps its pre-#336 meaning bit-for-bit), and segments with no collection index still fail open (stream-only: our archive has pre-index segments; upstream's cannot). subscribeEvents' kinds axis was already ported with the 0015 re-pin. differential oracle passes against upstream's released client (one overlay rename: WithBackfillOnly -> WithSnapshotOnly). Co-Authored-By: Claude Fable 5 --- docs/semantic-parity.md | 2 +- docs/upstream-harness.md | 5 +- src/internal/serve/xrpcapi.zig | 269 +++++++++++++++++- tests/archive_client_e2e.py | 2 +- tests/differential_oracle.py | 2 +- tests/environment_contract.py | 6 +- .../stream_differential_test.go | 4 +- 7 files changed, 269 insertions(+), 21 deletions(-) diff --git a/docs/semantic-parity.md b/docs/semantic-parity.md index e1bef63..29edaea 100644 --- a/docs/semantic-parity.md +++ b/docs/semantic-parity.md @@ -7,7 +7,7 @@ prove today; its status reflects current evidence only. > `network.bsky.jetstream.subscribeEvents` (proposal-0015 lexicon frames), > renamed `planBackfill` to `planSnapshot`, removed `recordCbor`, added a v1 > envelope `cursor`, and retrained the v2 zstd dictionary. Stream ported the -> new contract against pin `d4dd2f0127e8934d125b2a26a90fe88239ca98a2`; rows +> new contract against pin `289b0328c2e1a0ccf8c870cb45de0b2397de19fb`; rows > below still describe the f29815c-era audit and are being re-proved through > the gate (the differential oracle and archive-client e2e now run upstream's > d4dd2f0 client against Stream). One deliberate non-port: upstream's diff --git a/docs/upstream-harness.md b/docs/upstream-harness.md index cc96892..98285f0 100644 --- a/docs/upstream-harness.md +++ b/docs/upstream-harness.md @@ -2,7 +2,10 @@ condensed study of `bluesky-social/jetstream`, written 2026-07-08. -**Upstream pin: `d4dd2f0127e8934d125b2a26a90fe88239ca98a2`** (2026-08-12). The +**Upstream pin: `289b0328c2e1a0ccf8c870cb45de0b2397de19fb`** (v0.2.0, the +2026-08-13 official release — Bluesky Protocol Services / Jetstream v2 with +network replay). Server-side delta from the prior pin `d4dd2f0` is PR #336 +(end-to-end kinds filtering); the rest is client/SDK/docs. The suites below verify against that commit — see lines citing it further down. This is the pin most Stream docs share; note `jss-format-v1.md` and `jss-seal-spec.md` are instead on `0c45f29` (2026-07-08), 56 commits earlier. diff --git a/src/internal/serve/xrpcapi.zig b/src/internal/serve/xrpcapi.zig index a3b5d7f..1945339 100644 --- a/src/internal/serve/xrpcapi.zig +++ b/src/internal/serve/xrpcapi.zig @@ -49,6 +49,12 @@ pub const Api = struct { plan: PlanConfig = .{}, import: ?*import_manager.Manager = null, import_token: []const u8 = "", + /// Gates the archive endpoints when non-empty, mirroring the hosted + /// Bluesky instances (Jetstream v2, 2026-08-13): live tail and + /// getZstdDictionary stay open, archive requests need a bearer key. + /// Empty = open archive — the operator's opt-in, since gating an + /// already-public surface must be a deliberate flip, not a default. + archive_api_key: []const u8 = "", /// Keep one manifest planning pass in flight so public callers cannot /// multiply CPU and response allocation with concurrent requests. plan_mutex: Io.Mutex = .init, @@ -85,6 +91,14 @@ pub const Api = struct { self.getImportStatus(respond, query); return true; } + const archive_method = std.mem.eql(u8, method, "planSnapshot") or + std.mem.eql(u8, method, "listSegments") or + std.mem.eql(u8, method, "getSegment") or + std.mem.eql(u8, method, "getBlock"); + if (archive_method and self.archive_api_key.len > 0 and !authorized(self.archive_api_key, authorization)) { + respond.err(401, "AuthRequired", "invalid bearer credential"); + return true; + } if (std.mem.eql(u8, method, "planSnapshot")) { if (!is_post) { respond.err(405, "InvalidRequest", "planSnapshot is a procedure (POST)"); @@ -884,10 +898,31 @@ const zstd_mod = @import("../storage/zstd.zig"); pub const PlanRequest = struct { dids: []const []const u8 = &.{}, collections: []const []const u8 = &.{}, + /// kind mask, upstream's manifest.KindMask: zero means omitted/match-all, + /// preserving the pre-kind-filter contract (v0.2.0, PR #336) + kinds: u8 = 0, after_seq: u64 = 0, before_seq: u64 = std.math.maxInt(u64), }; +pub const kind_commit: u8 = 1; +pub const kind_identity: u8 = 2; +pub const kind_account: u8 = 4; +pub const kind_sync: u8 = 8; + +fn kindsAdmit(mask: u8, kind: u8) bool { + return mask == 0 or mask & kind != 0; +} + +/// upstream manifest.sentinelKind: which public kind a DID-marker sentinel +/// collection carries (0 = a real record collection) +fn sentinelKindMask(nsid: []const u8) u8 { + if (std.mem.eql(u8, nsid, "$account")) return kind_account; + if (std.mem.eql(u8, nsid, "$identity")) return kind_identity; + if (std.mem.eql(u8, nsid, "$sync")) return kind_sync; + return 0; +} + /// Render the listSegments body. Failures propagate so the caller answers /// 500; returning without responding would drop the request entirely. fn renderSegments( @@ -956,6 +991,9 @@ pub fn planSnapshot(self: *Api, respond: anytype, body: []const u8) void { // request is permanently bad and it will never retry the range. const req = parsePlanRequest(alloc, body, self.plan) catch |err| switch (err) { error.OutOfMemory => return respond.err(500, "InternalError", "oom"), + // upstream's exact message: the request is well-formed but the + // combination can never match anything, which the caller must learn + error.NeverApplies => return respond.err(400, "InvalidRequest", "collections filter can never apply: kinds excludes commit"), else => return respond.err(400, "InvalidRequest", "malformed plan request"), }; @@ -1113,27 +1151,54 @@ fn matchBlocks(alloc: Allocator, sealed: *const @import("../storage/manifest.zig if (info.event_count == 0 or info.max_seq <= req.after_seq or info.min_seq > req.before_seq) matched[i] = false; } - if (req.collections.len > 0) { + // Upstream collectionCandidatesForSegment (v0.2.0, PR #336): the + // kinds/collections axes prune blocks together. A sentinel $-collection + // is admitted only when its kind is (previously: always) — kinds=commit + // now skips marker-only blocks; the marker-safe default (kinds omitted) + // still selects them. Real collections are admitted only while commit + // is; a collections filter that excludes commit is rejected at parse. + // A segment with no collection index at all (sealed before the index + // existed) fails open, as before: kind/collection pruning needs the + // metadata to prune soundly. + if ((req.kinds != 0 or req.collections.len > 0) and sealed.collection_index.entries.len > 0) { const ci = sealed.collection_index; - // wanted collection ids; wildcards match archived collections by - // prefix. sentinel $-collections always match (DID-level markers - // must reach every filtered backfill — upstream §5/#142) - var wanted = try alloc.alloc(bool, ci.entries.len); + const collection_match_all = req.collections.len == 0; + const all_real = collection_match_all and kindsAdmit(req.kinds, kind_commit); + const included = try alloc.alloc(bool, ci.entries.len); + const real = try alloc.alloc(bool, ci.entries.len); + var has_real = false; + var any_included = false; for (ci.entries, 0..) |entry, id| { - wanted[id] = entry.nsid.len > 0 and entry.nsid[0] == '$'; - if (wanted[id]) continue; - for (req.collections) |c| { - if (std.mem.endsWith(u8, c, ".*")) { - if (std.mem.startsWith(u8, entry.nsid, c[0 .. c.len - 1])) wanted[id] = true; - } else if (std.mem.eql(u8, entry.nsid, c)) wanted[id] = true; + included[id] = false; + const smask = sentinelKindMask(entry.nsid); + real[id] = smask == 0; + if (smask != 0) { + included[id] = kindsAdmit(req.kinds, smask); + } else if (kindsAdmit(req.kinds, kind_commit)) { + has_real = true; + if (!collection_match_all) { + for (req.collections) |c| { + if (std.mem.endsWith(u8, c, ".*")) { + if (std.mem.startsWith(u8, entry.nsid, c[0 .. c.len - 1])) included[id] = true; + } else if (std.mem.eql(u8, entry.nsid, c)) included[id] = true; + } + } } + if (included[id]) any_included = true; } + // upstream candidates.empty(): nothing this segment holds can match + const empty = !any_included and (!all_real or !has_real); for (matched, 0..) |*m, block| { if (!m.*) continue; - if (ci.entries.len == 0 or ci.blockRow(block) == null) continue; // missing metadata fails open + if (empty) { + m.* = false; + continue; + } + if (ci.blockRow(block) == null) continue; // missing bitmask row fails open m.* = false; - for (wanted, 0..) |w, id| { - if (w and ci.blockHasCollection(block, id)) { + for (included, 0..) |inc, id| { + const admit = inc or (all_real and real[id]); + if (admit and ci.blockHasCollection(block, id)) { m.* = true; break; } @@ -1225,12 +1290,36 @@ fn parsePlanRequest(alloc: Allocator, body: []const u8, config: PlanConfig) !Pla } req.dids = try parseStringArray(alloc, parsed.object.get("dids")); req.collections = try parseStringArray(alloc, parsed.object.get("collections")); + req.kinds = try parsePlanKinds(try parseStringArray(alloc, parsed.object.get("kinds"))); if (has_after and has_before and req.before_seq <= req.after_seq) return error.Bad; try validatePlanDids(alloc, req.dids, config.max_dids); try validatePlanCollections(alloc, req.collections, config.max_collections); + // upstream plansnapshot.go: a collections filter that can never apply + // (kinds set and excluding commit) is a request error, not an empty plan + if (req.collections.len > 0 and req.kinds != 0 and req.kinds & kind_commit == 0) + return error.NeverApplies; return req; } +/// upstream validatePlanKinds: at most 4 values (the enum's size), each one +/// of commit/identity/account/sync; empty means match-all (mask 0) +fn parsePlanKinds(values: []const []const u8) !u8 { + if (values.len > 4) return error.Bad; + var mask: u8 = 0; + for (values) |v| { + if (std.mem.eql(u8, v, "commit")) { + mask |= kind_commit; + } else if (std.mem.eql(u8, v, "identity")) { + mask |= kind_identity; + } else if (std.mem.eql(u8, v, "account")) { + mask |= kind_account; + } else if (std.mem.eql(u8, v, "sync")) { + mask |= kind_sync; + } else return error.Bad; + } + return mask; +} + fn validatePlanDids(alloc: Allocator, dids: []const []const u8, max_dids: usize) !void { var seen: std.StringHashMapUnmanaged(void) = .empty; for (dids) |did| { @@ -1362,6 +1451,78 @@ test "exact collection reporting does not fail open" { try std.testing.expect(ci.blockRow(1) == null); } +// Mirrors upstream plan_test.go's kind-mask cases (v0.2.0, PR #336): the +// kinds axis prunes blocks through the same collection index, sentinels are +// admitted per-kind instead of unconditionally, and the marker-safe default +// (kinds omitted) keeps its pre-#336 meaning. +test "plan kinds axis prunes sentinel and real blocks like upstream" { + var arena = std.heap.ArenaAllocator.init(std.testing.allocator); + defer arena.deinit(); + const alloc = arena.allocator(); + + const blocks = [_]segment.BlockIndexEntry{ oneBlock(1), oneBlock(2), oneBlock(3) }; + var entries = [_]segment.CollectionEntry{ + .{ .nsid = "$account", .count = 0 }, + .{ .nsid = "$sync", .count = 0 }, + .{ .nsid = "app.bsky.feed.post", .count = 2 }, + }; + // block0: $account only; block1: post only; block2: $sync + post + var bitmasks = [_]u8{ 0b001, 0b100, 0b110 }; + const ci: segment.CollectionIndex = .{ + .entries = &entries, + .bitmasks = &bitmasks, + .bitmask_len = 1, + .buffer = &.{}, + }; + const sealed = planFixture(&blocks, &ci); + + // kinds=commit skips marker-only blocks (the new capability) + const commits = try matchBlocks(alloc, &sealed, .{ .kinds = kind_commit }); + try std.testing.expectEqualSlices(bool, &.{ false, true, true }, commits); + + // kinds=account selects only the account-marker block + const accounts = try matchBlocks(alloc, &sealed, .{ .kinds = kind_account }); + try std.testing.expectEqualSlices(bool, &.{ true, false, false }, accounts); + + // kinds omitted + collections: marker-safe default still admits every + // sentinel (pre-#336 behavior preserved bit-for-bit) + const legacy = try matchBlocks(alloc, &sealed, .{ .collections = &.{"app.bsky.feed.post"} }); + try std.testing.expectEqualSlices(bool, &.{ true, true, true }, legacy); + + // kinds=commit,account + a collections filter that matches nothing: + // only the admitted sentinel's block survives ($sync stays excluded) + const narrow = try matchBlocks(alloc, &sealed, .{ + .kinds = kind_commit | kind_account, + .collections = &.{"other.example.thing"}, + }); + try std.testing.expectEqualSlices(bool, &.{ true, false, false }, narrow); +} + +test "plan kinds validation mirrors upstream plansnapshot.go" { + var arena = std.heap.ArenaAllocator.init(std.testing.allocator); + defer arena.deinit(); + const alloc = arena.allocator(); + const config: PlanConfig = .{}; + + // collections + kinds excluding commit can never apply — request error + try std.testing.expectError(error.NeverApplies, parsePlanRequest(alloc, + \\{"collections":["app.bsky.feed.post"],"kinds":["account"]} + , config)); + // unknown kind value + try std.testing.expectError(error.Bad, parsePlanRequest(alloc, + \\{"kinds":["bogus"]} + , config)); + // more values than the enum holds + try std.testing.expectError(error.Bad, parsePlanRequest(alloc, + \\{"kinds":["commit","commit","commit","commit","commit"]} + , config)); + // valid: kinds with commit alongside collections; duplicates fold + const ok = try parsePlanRequest(alloc, + \\{"collections":["app.bsky.feed.post"],"kinds":["commit","commit","sync"]} + , config); + try std.testing.expectEqual(@as(u8, kind_commit | kind_sync), ok.kinds); +} + test "matchBlocks propagates allocation failure instead of planning a false negative" { const blocks = [_]segment.BlockIndexEntry{oneBlock(1)}; const ci: segment.CollectionIndex = .{ @@ -1395,12 +1556,92 @@ const PlanResponder = struct { fn written(self: *const PlanResponder) []const u8 { return self.body[0..self.body_len]; } + // full-route tests reach the dictionary/archive paths; record the code + fn dictionary(self: *PlanResponder, _: []const u8, _: u32, _: []const u8, not_modified: bool) void { + self.code = if (not_modified) 304 else 200; + } + fn archiveModified(_: *PlanResponder, _: i64) void {} + fn archiveCacheMaxAge(_: *PlanResponder, _: u64) void {} + fn notModified(self: *PlanResponder, _: []const u8) void { + self.code = 304; + } + fn preconditionFailed(self: *PlanResponder, _: []const u8) void { + self.code = 412; + } + fn rangeNotSatisfiable(self: *PlanResponder, _: usize, _: bool) void { + self.code = 416; + } + fn partial(self: *PlanResponder, _: []const u8, _: []const u8, _: u64, _: u64, _: usize) void { + self.code = 206; + } + fn multipartBytes(self: *PlanResponder, _: []const u8, _: []const u8, _: []const u8, _: []const ByteRange) void { + self.code = 206; + } + fn raw(self: *PlanResponder, code: u16, _: []const u8, _: []const u8) void { + self.code = code; + } + fn file(self: *PlanResponder, _: Io, _: Io.File, _: []const u8, _: u64, _: u64, _: u64, _: bool) void { + self.code = 200; + } + fn multipartFile(self: *PlanResponder, _: Io, _: Io.File, _: []const u8, _: []const u8, _: []const ByteRange, _: u64) void { + self.code = 206; + } }; // An allocation failure anywhere inside planning must never surface as a 200 // carrying a smaller plan. A client cannot distinguish that from "no matching // data" and would skip the missing range forever. Sweeping the fail index // keeps this honest as the allocation sequence changes. +// Mirrors the hosted Bluesky instances' edge contract (Jetstream v2, +// 2026-08-13): archive endpoints 401 without the configured bearer key; +// the live tail and the dictionary stay open. An EMPTY key leaves the +// archive open — gating a public surface is the operator's deliberate flip, +// the opposite default from the import token's secure-by-default 401. +test "archive api key gates plan/list/segment/block and nothing else" { + const testing = std.testing; + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var path_buffer: [Io.Dir.max_path_bytes]u8 = undefined; + const path_len = try tmp.dir.realPath(io, &path_buffer); + var archive = try archive_mod.Archive.init(testing.allocator, io, path_buffer[0..path_len]); + defer archive.deinit(); + + var api: Api = .{ .allocator = testing.allocator, .io = io, .archive = &archive, .archive_api_key = "sekrit" }; + + const gated = [_]struct { m: []const u8, p: []const u8 }{ + .{ .m = "POST", .p = "/xrpc/network.bsky.jetstream.planSnapshot" }, + .{ .m = "GET", .p = "/xrpc/network.bsky.jetstream.listSegments" }, + .{ .m = "GET", .p = "/xrpc/network.bsky.jetstream.getSegment" }, + .{ .m = "GET", .p = "/xrpc/network.bsky.jetstream.getBlock" }, + }; + for (gated) |g| { + var no_auth: PlanResponder = .{}; + _ = api.handle(&no_auth, g.m, g.p, "", "{}", null, .{}); + try testing.expectEqual(@as(u16, 401), no_auth.code); + var bad: PlanResponder = .{}; + _ = api.handle(&bad, g.m, g.p, "", "{}", "Bearer wrong", .{}); + try testing.expectEqual(@as(u16, 401), bad.code); + var good: PlanResponder = .{}; + _ = api.handle(&good, g.m, g.p, "", "{}", "Bearer sekrit", .{}); + try testing.expect(good.code != 401); + } + + // the dictionary stays open, exactly like upstream's hosted edge + var dict: PlanResponder = .{}; + _ = api.handle(&dict, "GET", "/xrpc/network.bsky.jetstream.getZstdDictionary", "", "", null, .{}); + try testing.expectEqual(@as(u16, 200), dict.code); + + // empty key = open archive (today's public contract, unchanged) + var open_api: Api = .{ .allocator = testing.allocator, .io = io, .archive = &archive }; + var open: PlanResponder = .{}; + _ = open_api.handle(&open, "GET", "/xrpc/network.bsky.jetstream.listSegments", "", "", null, .{}); + try testing.expect(open.code != 401); +} + test "planBackfill never answers 200 with a short plan under allocation failure" { const testing = std.testing; var threaded: Io.Threaded = .init(testing.allocator, .{}); diff --git a/tests/archive_client_e2e.py b/tests/archive_client_e2e.py index ae48bdc..3d03782 100644 --- a/tests/archive_client_e2e.py +++ b/tests/archive_client_e2e.py @@ -19,7 +19,7 @@ import subprocess import urllib.request -UPSTREAM_PIN = "d4dd2f0127e8934d125b2a26a90fe88239ca98a2" +UPSTREAM_PIN = "289b0328c2e1a0ccf8c870cb45de0b2397de19fb" UPSTREAM = pathlib.Path( os.environ.get( "STREAM_UPSTREAM_REPO", diff --git a/tests/differential_oracle.py b/tests/differential_oracle.py index 3323eb7..42dd65d 100644 --- a/tests/differential_oracle.py +++ b/tests/differential_oracle.py @@ -15,7 +15,7 @@ import subprocess import tempfile -PIN = "d4dd2f0127e8934d125b2a26a90fe88239ca98a2" +PIN = "289b0328c2e1a0ccf8c870cb45de0b2397de19fb" ROOT = Path(__file__).resolve().parents[1] UPSTREAM = Path( os.environ.get( diff --git a/tests/environment_contract.py b/tests/environment_contract.py index c20236e..d94566d 100644 --- a/tests/environment_contract.py +++ b/tests/environment_contract.py @@ -17,7 +17,7 @@ import urllib.request BIN = "./zig-out/bin/stream" -PIN = "d4dd2f0127e8934d125b2a26a90fe88239ca98a2" +PIN = "289b0328c2e1a0ccf8c870cb45de0b2397de19fb" ROOT = pathlib.Path(__file__).resolve().parents[1] # STREAM_UPSTREAM_REPO is the name the justfile, scripts/admit, and the other # oracles already use; this file was the only one reading JETSTREAM_UPSTREAM, @@ -63,6 +63,10 @@ def exact_pinned_map_case() -> None: # upstream exactly. stream_only = { "JETSTREAM_REBLOOM_SWEEP": "rebloom-sweep", + # Archive bearer gate: upstream's OSS server has no archive auth (the + # hosted Bluesky instances gate at their edge); stream carries the + # gate in-process. Empty = open. + "JETSTREAM_ARCHIVE_API_KEY": "archive-api-key", } # Upstream variables Stream knowingly has NOT ported: the PDS-direct # fleet bootstrap (upstream specs/notes/2026-08-03-pds-direct-backfill- diff --git a/tests/upstream_oracle/stream_differential_test.go b/tests/upstream_oracle/stream_differential_test.go index 27cf1c6..eb6b768 100644 --- a/tests/upstream_oracle/stream_differential_test.go +++ b/tests/upstream_oracle/stream_differential_test.go @@ -34,7 +34,7 @@ import ( "github.com/stretchr/testify/require" ) -const streamDifferentialPin = "d4dd2f0127e8934d125b2a26a90fe88239ca98a2" +const streamDifferentialPin = "289b0328c2e1a0ccf8c870cb45de0b2397de19fb" type synchronizedBuffer struct { mu sync.Mutex @@ -982,7 +982,7 @@ func collectStreamOracleBackfill(t *testing.T, baseURL string, through uint64) [ client, err := jetstream.Subscribe(baseURL, jetstream.WithAfterSeq(0), jetstream.WithBeforeSeq(through), - jetstream.WithBackfillOnly(), + jetstream.WithSnapshotOnly(), jetstream.WithBatchSize(32), ) require.NoError(t, err) -- knotmirror 0.0.1