diff --git a/README.md b/README.md index 29a8a3a..4f03a02 100644 --- a/README.md +++ b/README.md @@ -47,6 +47,7 @@ just simulator # local fake atproto network on :7777 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 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 @@ -68,6 +69,19 @@ 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. +`planBackfill` exposes the same CLI controls and defaults as upstream: + +``` +--plan-max-dids=1000 +--plan-max-collections=25 +--plan-max-entries=100000 +--plan-whole-segment-threshold=0.75 +``` + +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. + 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/semantic-parity.md b/docs/semantic-parity.md index e81c1cd..ea05da4 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 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. | +| Archive XRPC | Official Go client plus Stream conformance replay over listSegments/getSegment/getBlock/planBackfill | **closed for wire behavior, cache policy, and planner controls.** 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. Three more production processes consume a physically sealed 3/4-density corpus and prove the distinct DID/collection caps, zero-disabled filters, one-entry pagination, zero-entry-limit unbounded paging, and exact 1.0/0.75 whole-segment threshold boundary. 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. | | 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 cc3332f..db2b562 100644 --- a/docs/upstream-harness.md +++ b/docs/upstream-harness.md @@ -158,6 +158,13 @@ and collection summaries. `Cache-Control: public, max-age=2`. This records upstream's positive-duration ceil behavior independently of the default `public, no-cache` archive receipt. +`just plan-config-contract` builds a real three-segment JSS archive whose +selected-block density is exactly 3/4, then starts three production Stream +processes. It proves the distinct DID/collection caps (including zero disabling +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. + ## 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 9f3fcaa..1714b4b 100644 --- a/justfile +++ b/justfile @@ -109,6 +109,49 @@ archive-contract: STREAM_BASE_URL="http://127.0.0.1:$PORT" \ STREAM_UPSTREAM_REPO='{{upstream}}' python3 tests/archive_client_e2e.py +# Exercise all four non-default planBackfill controls through three real +# production processes over a purpose-built, physically sealed JSS archive. +plan-config-contract: + #!/usr/bin/env bash + set -euo pipefail + DATA=$(mktemp -d) + LOG=$(mktemp) + PORT=${STREAM_PLAN_CONFIG_PORT:-6022} + 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 -- --plan-config-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_BASE_URL="http://127.0.0.1:$PORT" STREAM_PLAN_CONFIG_MODE="$mode" \ + python3 tests/plan_config_contract.py + kill "$pid"; wait "$pid"; unset pid + } + run_case limited --plan-max-dids=1 --plan-max-collections=1 \ + --plan-max-entries=1 --plan-whole-segment-threshold=1 + run_case whole --plan-max-dids=1 --plan-max-collections=1 \ + --plan-max-entries=0 --plan-whole-segment-threshold=0.75 + run_case disabled --plan-max-dids=0 --plan-max-collections=0 \ + --plan-max-entries=0 --plan-whole-segment-threshold=0.75 + if ./zig-out/bin/stream --plan-whole-segment-threshold=0 \ + --data-dir="$DATA" >"$LOG" 2>&1; then + echo "zero planner threshold unexpectedly started" >&2; exit 1 + fi + # 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/xrpcapi.zig b/src/internal/xrpcapi.zig index cfc3211..04a8dae 100644 --- a/src/internal/xrpcapi.zig +++ b/src/internal/xrpcapi.zig @@ -32,11 +32,19 @@ pub const Preconditions = struct { if_range: ?[]const u8 = null, }; +pub const PlanConfig = struct { + max_dids: usize = 1000, + max_collections: usize = 25, + max_entries: usize = 100_000, + whole_segment_threshold: f64 = 0.75, +}; + pub const Api = struct { allocator: Allocator, io: Io, archive: *archive_mod.Archive, segment_cache_max_age_s: u64 = 0, + plan: PlanConfig = .{}, import: ?*import_manager.Manager = null, import_token: []const u8 = "", /// Keep one manifest planning pass in flight so public callers cannot @@ -696,11 +704,6 @@ test "checksumHex is 16 lowercase hex chars" { const gloom = @import("gloom.zig"); const zstd_mod = @import("zstd.zig"); -const whole_segment_threshold = 0.75; -const max_plan_entries = 100_000; // upstream default per-page work-unit limit -const max_plan_dids = 1000; -const max_plan_collections = 25; - pub const PlanRequest = struct { dids: []const []const u8 = &.{}, collections: []const []const u8 = &.{}, @@ -713,7 +716,7 @@ pub fn planBackfill(self: *Api, respond: anytype, body: []const u8) void { defer arena.deinit(); const alloc = arena.allocator(); - const req = parsePlanRequest(alloc, body) catch + const req = parsePlanRequest(alloc, body, self.plan) catch return respond.err(400, "InvalidRequest", "malformed plan request"); self.plan_mutex.lockUncancelable(self.io); @@ -760,10 +763,10 @@ pub fn planBackfill(self: *Api, respond: anytype, body: []const u8) void { if (match_count == 0) continue; const density = @as(f64, @floatFromInt(match_count)) / @as(f64, @floatFromInt(matched.len)); - const whole = density >= whole_segment_threshold; + const whole = density >= self.plan.whole_segment_threshold; var admitted = matched; if (whole) { - if (entries >= max_plan_entries and entries > 0) { + if (self.plan.max_entries > 0 and entries >= self.plan.max_entries and entries > 0) { truncated = true; continue; } @@ -781,7 +784,7 @@ pub fn planBackfill(self: *Api, respond: anytype, body: []const u8) void { } var last = at; while (last + 1 < matched.len and matched[last + 1]) last += 1; - if (entries >= max_plan_entries and entries > 0) { + if (self.plan.max_entries > 0 and entries >= self.plan.max_entries and entries > 0) { truncated = true; break; } @@ -950,7 +953,7 @@ fn writeBlockRanges(s: *std.json.Stringify, matched: []const bool) !void { try s.endArray(); } -fn parsePlanRequest(alloc: Allocator, body: []const u8) !PlanRequest { +fn parsePlanRequest(alloc: Allocator, body: []const u8, config: PlanConfig) !PlanRequest { var req: PlanRequest = .{}; if (body.len == 0) return req; const parsed = try std.json.parseFromSliceLeaky(std.json.Value, alloc, body, .{}); @@ -970,32 +973,32 @@ fn parsePlanRequest(alloc: Allocator, body: []const u8) !PlanRequest { req.dids = try parseStringArray(alloc, parsed.object.get("dids")); req.collections = try parseStringArray(alloc, parsed.object.get("collections")); if (has_after and has_before and req.before_seq <= req.after_seq) return error.Bad; - try validatePlanDids(alloc, req.dids); - try validatePlanCollections(alloc, req.collections); + try validatePlanDids(alloc, req.dids, config.max_dids); + try validatePlanCollections(alloc, req.collections, config.max_collections); return req; } -fn validatePlanDids(alloc: Allocator, dids: []const []const u8) !void { +fn validatePlanDids(alloc: Allocator, dids: []const []const u8, max_dids: usize) !void { var seen: std.StringHashMapUnmanaged(void) = .empty; for (dids) |did| { - const gop = try seen.getOrPut(alloc, did); - if (gop.found_existing) continue; - if (seen.count() > max_plan_dids) return error.Bad; + if (seen.contains(did)) continue; + if (seen.count() == max_dids) return error.Bad; if (@import("zat").Did.parse(did) == null) return error.Bad; + try seen.put(alloc, did, {}); } } -fn validatePlanCollections(alloc: Allocator, collections: []const []const u8) !void { +fn validatePlanCollections(alloc: Allocator, collections: []const []const u8, max_collections: usize) !void { var seen: std.StringHashMapUnmanaged(void) = .empty; for (collections) |collection| { - const gop = try seen.getOrPut(alloc, collection); - if (gop.found_existing) continue; - if (seen.count() > max_plan_collections) return error.Bad; + if (seen.contains(collection)) continue; + if (seen.count() == max_collections) return error.Bad; if (std.mem.endsWith(u8, collection, ".*")) { const head = collection[0 .. collection.len - 2]; const probe = try std.fmt.allocPrint(alloc, "{s}.wildcard", .{head}); if (@import("zat").Nsid.parse(probe) == null) return error.Bad; } else if (@import("zat").Nsid.parse(collection) == null) return error.Bad; + try seen.put(alloc, collection, {}); } } @@ -1025,9 +1028,11 @@ test "plan request rejects invalid filters and windows" { var arena = std.heap.ArenaAllocator.init(std.testing.allocator); defer arena.deinit(); const alloc = arena.allocator(); - try std.testing.expectError(error.Bad, parsePlanRequest(alloc, "{\"afterSeq\":-1}")); - try std.testing.expectError(error.Bad, parsePlanRequest(alloc, "{\"afterSeq\":10,\"beforeSeq\":10}")); - try std.testing.expectError(error.Bad, parsePlanRequest(alloc, "{\"dids\":[\"not-a-did\"]}")); - try std.testing.expectError(error.Bad, parsePlanRequest(alloc, "{\"collections\":[\"not a collection\"]}")); - _ = try parsePlanRequest(alloc, "{\"dids\":[\"did:plc:abc\"],\"collections\":[\"app.bsky.feed.*\"]}"); + try std.testing.expectError(error.Bad, parsePlanRequest(alloc, "{\"afterSeq\":-1}", .{})); + try std.testing.expectError(error.Bad, parsePlanRequest(alloc, "{\"afterSeq\":10,\"beforeSeq\":10}", .{})); + try std.testing.expectError(error.Bad, parsePlanRequest(alloc, "{\"dids\":[\"not-a-did\"]}", .{})); + try std.testing.expectError(error.Bad, parsePlanRequest(alloc, "{\"collections\":[\"not a collection\"]}", .{})); + _ = try parsePlanRequest(alloc, "{\"dids\":[\"did:plc:abc\"],\"collections\":[\"app.bsky.feed.*\"]}", .{}); + try std.testing.expectError(error.Bad, parsePlanRequest(alloc, "{\"dids\":[\"did:plc:a\"]}", .{ .max_dids = 0 })); + try std.testing.expectError(error.Bad, parsePlanRequest(alloc, "{\"collections\":[\"app.bsky.feed.post\"]}", .{ .max_collections = 0 })); } diff --git a/src/main.zig b/src/main.zig index e254661..025f815 100644 --- a/src/main.zig +++ b/src/main.zig @@ -213,6 +213,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 plan_config: xrpcapi.PlanConfig = .{}; var arg_it = init.args.iterate(); _ = arg_it.next(); // program name @@ -245,6 +246,14 @@ 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, "--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=")) { + plan_config.max_collections = try std.fmt.parseInt(usize, arg["--plan-max-collections=".len..], 10); + } else if (std.mem.startsWith(u8, arg, "--plan-max-entries=")) { + plan_config.max_entries = try std.fmt.parseInt(usize, arg["--plan-max-entries=".len..], 10); + } else if (std.mem.startsWith(u8, arg, "--plan-whole-segment-threshold=")) { + plan_config.whole_segment_threshold = try std.fmt.parseFloat(f64, arg["--plan-whole-segment-threshold=".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=")) { @@ -290,6 +299,10 @@ pub fn main(init: std.process.Init.Minimal) !void { return error.BadArgs; } } + if (!(plan_config.whole_segment_threshold > 0 and plan_config.whole_segment_threshold <= 1)) { + log.err("--plan-whole-segment-threshold must be > 0 and <= 1", .{}); + return error.BadArgs; + } if (store_fault_prefix) |prefix| { if (!meta_store.armFault(prefix, store_fault_ordinal)) { log.err("invalid store fault: prefix must be non-empty and ordinal must be positive", .{}); @@ -346,6 +359,7 @@ pub fn main(init: std.process.Init.Minimal) !void { .io = io, .archive = &archive, .segment_cache_max_age_s = segment_cache_max_age_s, + .plan = plan_config, }; hub.xrpc = &xrpc_api; diff --git a/src/write_sample.zig b/src/write_sample.zig index 8dc084e..a1344bd 100644 --- a/src/write_sample.zig +++ b/src/write_sample.zig @@ -127,6 +127,39 @@ pub fn main(init: std.process.Init.Minimal) !void { return; } + // --plan-config-archive : three real sealed segments with a 3/4 + // collection density and two disjoint matching block ranges per segment. + // This lets the production HTTP contract distinguish planner threshold, + // pagination, and unlimited-page configuration without fixture verdicts. + if (std.mem.eql(u8, path, "--plan-config-archive")) { + const dir = args.next() orelse return error.MissingPath; + var a = try archive_mod.Archive.init(allocator, io, dir); + defer a.deinit(); + a.max_events_per_block = 1; + a.writer.max_events_per_block = 1; + a.max_segment_bytes = 1024 * 1024; + var seq: u64 = 1; + for (0..3) |_| { + for (0..4) |at| { + _ = try a.append(.{ + .seq = 0, + .witnessed_at = @intCast(1_700_000_000_000_000 + seq * 1000), + .indexed_at = 0, + .kind = .create, + .did = "did:plc:planconfigaccount", + .collection = if (at == 1) "app.bsky.feed.like" else "app.bsky.feed.post", + .rkey = "3l3qo2vuowo2b", + .rev = "3l3qo2vutsw2b", + .payload = "\xa1\x64\x74\x65\x78\x74\x62\x68\x69", + }, @intCast(seq)); + seq += 1; + } + try a.rotate(); + } + a.close(); + return; + } + var w = try writer_mod.ActiveWriter.init(allocator); defer w.deinit(); w.max_events_per_block = 1000; diff --git a/tests/plan_config_contract.py b/tests/plan_config_contract.py new file mode 100644 index 0000000..648436c --- /dev/null +++ b/tests/plan_config_contract.py @@ -0,0 +1,59 @@ +"""Real-process contract for Jetstream V2 planBackfill operator controls.""" + +import json +import os +import urllib.error +import urllib.request + + +BASE = os.environ["STREAM_BASE_URL"].rstrip("/") + "/xrpc/network.bsky.jetstream.planBackfill" +MODE = os.environ["STREAM_PLAN_CONFIG_MODE"] + + +def post(body): + request = urllib.request.Request( + BASE, + data=json.dumps(body).encode(), + headers={"Content-Type": "application/json"}, + method="POST", + ) + try: + response = urllib.request.urlopen(request) + return response.status, json.load(response) + except urllib.error.HTTPError as error: + return error.code, json.load(error) + + +if MODE == "limited": + status, result = post({"collections": ["app.bsky.feed.post"]}) + assert status == 200 + assert result["sealedTipSeq"] == 12 + assert result["plannedThroughSeq"] == 1 + assert result["stats"]["entries"] == 1 + assert len(result["segments"]) == 1 + assert result["segments"][0]["mode"] == "blocks" + assert result["segments"][0]["blocks"] == [{"first": 0, "last": 0}] + for body in ( + {"dids": ["did:plc:a", "did:plc:b"]}, + {"collections": ["app.bsky.feed.post", "app.bsky.feed.like"]}, + ): + status, error = post(body) + assert status == 400 and error["error"] == "InvalidRequest" +elif MODE == "whole": + status, result = post({"collections": ["app.bsky.feed.post"]}) + assert status == 200 + assert result["plannedThroughSeq"] == result["sealedTipSeq"] == 12 + assert result["stats"]["entries"] == 3 + assert len(result["segments"]) == 3 + assert all(segment["mode"] == "segment" for segment in result["segments"]) +elif MODE == "disabled": + for body in ( + {"dids": ["did:plc:planconfigaccount"]}, + {"collections": ["app.bsky.feed.post"]}, + ): + status, error = post(body) + assert status == 400 and error["error"] == "InvalidRequest" +else: + raise AssertionError(f"unknown mode: {MODE}") + +print(f"plan config {MODE}: PASS")