diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -10,19 +10,35 @@ convenience feed and its archive/replay API — and depends on zat for everything protocol. the same split Bluesky drew between `@atproto/api` and `@bsky/sdk`. -## today +## the client -- **`ArchiveBackfill`** — network replay: `planSnapshot` (collections/dids/kinds, - seq bounds, `api_key` for token-gated instances) → `getBlock`/`getSegment` → - jss v1 columnar decode → `zat.JetstreamEvent` through the same `onEvent` - handler your live client uses. `fetchSeqBounds` maps a time window to plan - bounds. moved here from zat 0.3.x. +mirrors the official Go client's shape: one subscribe covers the live tail, +filtered archive replay, and a gapless replay→live cutover. -## next +- **`subscribe`** — the unified client (upstream `jetstream.Subscribe`). + `after_seq` sweeps the sealed archive and cuts over to the live tail at the + sealed tip; the client's per-event seq dedup makes the seam at-least-once + with no gap, and a `CursorTooOld` refusal re-enters backfill from the last + delivered seq (bounded, monotonic). `snapshot_only` stops at the sealed + range. without `after_seq` it is a pure live tail (`live_cursor` resumes). +- **`LiveClient`** — the `subscribeEvents` v2 live tail on its own: lexicon + frames, strict per-event seq cursors, `kinds`/`collections`/`dids`, + dict-zstd compression (`zstdDictionary` negotiation, rotation recovery), + typed pre-upgrade errors (`CursorTooOld` / `UnknownZstdDictionary` / + `InvalidRequest`), reconnect backoff that resumes at the highest delivered + seq, TCP keepalive. +- **`ArchiveBackfill`** — the archive half standalone: `planSnapshot` + (collections/dids, seq bounds, `api_key` for token-gated instances) → + `getBlock`/`getSegment` → jss v1 columnar decode. two handler contracts: + the classic `onEvent(zat.JetstreamEvent)` (commits only, kept for existing + consumers) and the unified client's `onRow(jetstream.Event)` (every row + with its seq, including identity/account/sync markers). `fetchSeqBounds` + maps a time window to plan bounds. -- the `subscribeEvents` v2 live client: lexicon frames, strict per-event seq - cursors, `kinds`/`dids`/`collections`, dictionary compression. until then, - `zat.JetstreamClient` (the v1 `/subscribe` wire) remains in zat. +one recorded divergence from upstream: commit records are delivered as wire +JSON (`std.json.Value`), not reconstructed canonical DAG-CBOR — zat.cbor has +no JSON→DAG-CBOR encoder and zig consumers fold on the JSON shape. the v1 +`/subscribe` wire client remains `zat.JetstreamClient` in zat. ## use diff --git a/build.zig b/build.zig --- a/build.zig +++ b/build.zig @@ -51,6 +51,7 @@ .link_libc = true, .imports = &.{ .{ .name = "zat", .module = zat.module("zat") }, + .{ .name = "websocket", .module = b.dependency("websocket", .{ .target = target, .optimize = optimize }).module("websocket") }, }, }); linkVendoredZstd(mod, b, target, optimize); diff --git a/build.zig.zon b/build.zig.zon --- a/build.zig.zon +++ b/build.zig.zon @@ -9,8 +9,12 @@ .hash = "N-V-__8AAPZ7fwBg4JoCzM_0o2A8wxH2hsUUeiU1iuZv53L5", }, .zat = .{ - .url = "https://tangled.org/zat.dev/zat/archive/v0.4.0.tar.gz", - .hash = "zat-0.4.0-5PuC7iFkDAD8OoBB-bDlexUoSI24rveRWfINE90Oxebs", + .url = "https://tangled.org/zat.dev/zat/archive/v0.4.1.tar.gz", + .hash = "zat-0.4.1-5PuC7iFkDADWRaINnS4NGXttC9wfwZ1uf2rXn-4p4w4c", + }, + .websocket = .{ + .url = "https://tangled.org/zzstoatzz.io/websocket.zig/archive/v0.1.12.tar.gz", + .hash = "websocket-0.1.12-ZPISdTJEBQDiqWKUP5mPPK9IXJ5xscVHybOlUbPHM9Rr", }, }, .paths = .{ "build.zig", "build.zig.zon", "src", "examples" }, diff --git a/src/archive_backfill.zig b/src/archive_backfill.zig --- a/src/archive_backfill.zig +++ b/src/archive_backfill.zig @@ -34,6 +34,7 @@ const sync = zat.firehose; // re-exports sync.CommitAction const cbor = zat.cbor; const zstd = @import("zstd.zig"); +const livedecode = @import("livedecode.zig"); const multibase = zat.multibase; const HttpTransport = zat.HttpTransport; @@ -95,6 +96,9 @@ /// cursor at or before the timestamp of the last delivered event. planned_through_seq: u64, sealed_tip_seq: u64, + /// true when an extended row handler (onRow) returned false mid-sweep; + /// the plan was abandoned cleanly at that point + stopped: bool = false, events_delivered: u64 = 0, blocks_decoded: u64 = 0, /// true when Options.max_blocks stopped the run before the plan was exhausted @@ -143,6 +147,10 @@ } if (window.isFull()) window.deliverOldest(allocator, options, handler, &result) catch |err| { window.discardAll(); + if (err == error.Stopped) { + result.stopped = true; + return result; + } return err; }; window.start(options, segment.name, .{ @@ -160,6 +168,10 @@ // delivery stays in plan order window.deliverAll(allocator, options, handler, &result) catch |err| { window.discardAll(); + if (err == error.Stopped) { + result.stopped = true; + return result; + } return err; }; if (limitReached(options, &result)) return result; @@ -177,6 +189,10 @@ window.deliverAll(allocator, options, handler, &result) catch |err| { window.discardAll(); + if (err == error.Stopped) { + result.stopped = true; + return result; + } return err; }; return result; @@ -653,6 +669,9 @@ const kind_update = 2; const kind_delete = 3; const kind_create_resync = 7; +const kind_identity = 4; +const kind_account = 5; +const kind_sync = 6; /// decode one decompressed columnar block and deliver matching rows. /// `arena` backs per-row record decoding; the caller resets it per block. @@ -725,7 +744,80 @@ rev_off += rev_len; pay_off += pay_len; - const operation: sync.CommitAction = switch (raw[kind_base + i]) { + const kind_byte = raw[kind_base + i]; + const witnessed_at_row = mem.readInt(i64, raw[wit_base + 8 * i ..][0..8], .little); + const seq_row = mem.readInt(u64, raw[seq_base + 8 * i ..][0..8], .little); + + // extended row handler (the unified client's archive half): every + // matching row is delivered as a v2 event WITH its seq — including + // identity/account/sync marker rows, which ride inline exactly like + // upstream's backfill (a folding consumer needs the #sync tombstone + // to know to refold). the classic onEvent handler below keeps its + // original commits-only, seq-less contract. + if (comptime @hasDecl(@TypeOf(handler.*), "onRow")) { + if (options.dids.len > 0 and !containsString(options.dids, did)) continue; + const row_event: ?livedecode.Event = switch (kind_byte) { + kind_create, kind_create_resync, kind_update, kind_delete => blk: { + if (options.collections.len > 0 and !containsString(options.collections, collection)) break :blk null; + const op: livedecode.Operation = switch (kind_byte) { + kind_update => .update, + kind_delete => .delete, + else => .create, + }; + const rec: ?json.Value = if (op != .delete and payload.len > 0) rec: { + const decoded = cbor.decode(arena, payload) catch return error.MalformedRecord; + break :rec try cborToJson(arena, decoded.value); + } else null; + break :blk .{ .seq = seq_row, .time_us = witnessed_at_row, .did = did, .payload = .{ .commit = .{ + .operation = op, + .collection = collection, + .rkey = rkey, + .rev = rev, + .record = rec, + } } }; + }, + kind_identity, kind_account, kind_sync => blk: { + // marker payloads are the archived upstream envelope as a + // CBOR map; fields legitimately absent on synthetic rows + // (e.g. an async-resync #sync) default to zero values, + // matching upstream's generated decoder + const env: ?cbor.Value = if (payload.len > 0) + (cbor.decode(arena, payload) catch return error.MalformedRecord).value + else + null; + break :blk switch (kind_byte) { + kind_identity => .{ .seq = seq_row, .time_us = witnessed_at_row, .did = did, .payload = .{ .identity = .{ + .did = did, + .handle = if (env) |e| e.getString("handle") else null, + .seq = if (env) |e| e.getInt("seq") else null, + .time = if (env) |e| e.getString("time") else null, + } } }, + kind_account => .{ .seq = seq_row, .time_us = witnessed_at_row, .did = did, .payload = .{ .account = .{ + .did = did, + .active = if (env) |e| e.getBool("active") orelse false else false, + .status = if (env) |e| e.getString("status") else null, + .seq = if (env) |e| e.getInt("seq") else null, + .time = if (env) |e| e.getString("time") else null, + } } }, + else => .{ .seq = seq_row, .time_us = witnessed_at_row, .did = did, .payload = .{ .sync = .{ + .did = did, + .rev = if (env) |e| e.getString("rev") orelse rev else rev, + .seq = if (env) |e| e.getInt("seq") else null, + .time = if (env) |e| e.getString("time") else null, + } } }, + }; + }, + else => null, // unknown jss kind from a newer writer: skip + }; + if (row_event) |ev| { + if (!handler.onRow(ev)) return error.Stopped; + result.events_delivered += 1; + result.last_time_us = witnessed_at_row; + } + continue; + } + + const operation: sync.CommitAction = switch (kind_byte) { kind_create, kind_create_resync => .create, kind_update => .update, kind_delete => .delete, @@ -743,8 +835,7 @@ break :blk try cborToJson(arena, decoded.value); } else null; - const witnessed_at = mem.readInt(i64, raw[wit_base + 8 * i ..][0..8], .little); - _ = mem.readInt(u64, raw[seq_base + 8 * i ..][0..8], .little); + const witnessed_at = witnessed_at_row; handler.onEvent(.{ .commit = .{ .did = did, diff --git a/src/client.zig b/src/client.zig new file mode 100644 --- /dev/null +++ b/src/client.zig @@ -0,0 +1,248 @@ +//! the unified Jetstream v2 client — live tail, filtered archive replay, +//! and gapless replay→live cutover through one handler. +//! +//! ports upstream client_core.go runBackfillThenLive (bluesky-social/ +//! jetstream, pin 289b032, design §11/§13/§14): +//! +//! 1. sweep the sealed archive: page planSnapshot and deliver the whole +//! sealed range (cursor, plannedThroughSeq] in seq order +//! (ArchiveBackfill with the extended onRow handler — every matching +//! row including identity/account/sync markers, each with its seq); +//! 2. connect subscribeEvents ONCE at cursor = the sealed tip — no rewind +//! margin, no client buffer. the live client's seq dedup makes the +//! seam at-least-once with no gap; segments sealed during the download +//! are covered by the server's cold replay. +//! +//! a pre-upgrade 400 CursorTooOld at connect (slow handoff / fell off +//! live) is NOT fatal: re-enter archive pagination from the last durably +//! processed seq. cutover = max(sealed tip, cursor) keeps the dedup floor +//! and resume monotonic non-decreasing (a live tail routinely delivers +//! events past the sealed tip; a re-learned tip below the cursor must not +//! regress it). re-backfill cycles that fail to advance the cursor are +//! bounded (upstream maxRebackfillStalls = 5) — a non-advancing cycle is a +//! pathological loop, not a real fall-behind. + +const std = @import("std"); +const zat = @import("zat"); +const archive = @import("archive_backfill.zig"); +const live_mod = @import("live.zig"); +const livedecode = @import("livedecode.zig"); + +const Io = std.Io; +const Allocator = std.mem.Allocator; +const log = std.log.scoped(.jetstream); + +/// upstream maxRebackfillStalls: consecutive too-old cycles that made no +/// cursor progress before the stream is declared pathological +pub const max_rebackfill_stalls = 5; + +pub const Options = struct { + /// base URL of the instance, e.g. "https://stream.waow.tech" + host: []const u8, + /// replay start, exclusive: sweep the sealed archive from here and cut + /// over to live. null = no replay — a pure live tail from the tip + /// (set live_cursor to resume a live tail at a known seq instead). + after_seq: ?u64 = null, + /// stop after the sealed range instead of cutting over to live + /// (upstream WithSnapshotOnly). only meaningful with after_seq. + snapshot_only: bool = false, + /// live-only resume point when after_seq is null; null = from the tip + live_cursor: ?u64 = null, + kinds: []const livedecode.Kind = &.{}, + collections: []const []const u8 = &.{}, + dids: []const []const u8 = &.{}, + /// bearer credential for token-gated archives (planSnapshot/getSegment/ + /// getBlock only; the live tail never needs it) + api_key: ?[]const u8 = null, + /// block fetches in flight during the archive sweep (see ArchiveBackfill) + concurrency: usize = 4, + /// opt into dict-zstd live frames: the current dictionary is fetched + /// via getZstdDictionary; a fetch failure degrades to an uncompressed + /// tail (compression is an optimization, never a stream failure) + zstd_compression: bool = false, + /// live reconnect backoff overrides (tests); zero = defaults + backoff_min_ns: u64 = 0, + backoff_max_ns: u64 = 0, +}; + +/// run the unified stream through `handler`: +/// fn onEvent(*H, livedecode.Event) bool — every event, archive and live, +/// in seq order across the seam; false stops cleanly +/// optional fn onError(*H, anyerror) bool — recoverable live errors +/// optional fn onInfo(*H, livedecode.Info) void +/// blocks until the handler stops it, the snapshot completes +/// (snapshot_only), or a terminal error. +pub fn subscribe(io: Io, allocator: Allocator, options: Options, handler: anytype) !void { + var dict_buf: ?[]u8 = null; + defer if (dict_buf) |owned| allocator.free(owned); + if (options.zstd_compression) dict_buf = fetchZstdDict(io, allocator, options.host); + + if (options.after_seq == null) { + // pure live tail (upstream runLiveOnly) + var client = live_mod.LiveClient.init(io, allocator, liveOptions(options, options.live_cursor, 0, dict_buf)); + defer client.deinit(); + var filtering = FilteringHandler(@TypeOf(handler.*)){ .inner = handler, .kinds = options.kinds }; + return client.run(&filtering) catch |err| switch (err) { + error.CursorTooOld => error.CursorTooOld, // live-only has no backfill to re-enter + else => |e| e, + }; + } + + var cursor = options.after_seq.?; + var stalls: u8 = 0; + while (true) { + var sweeper = SweepHandler(@TypeOf(handler.*)){ .inner = handler, .kinds = options.kinds }; + const result = try archive.run(io, allocator, .{ + .host = options.host, + .collections = options.collections, + .dids = options.dids, + .api_key = options.api_key, + .after_seq = cursor, + .concurrency = options.concurrency, + }, &sweeper); + if (result.stopped) return; + if (options.snapshot_only) return; + + // cut over at the HIGHER of the freshly-learned sealed tip and the + // cursor already processed through (monotonic non-decreasing; see + // module doc) + const cutover = @max(result.planned_through_seq, cursor); + var client = live_mod.LiveClient.init(io, allocator, liveOptions(options, cutover, cutover, dict_buf)); + defer client.deinit(); + var filtering = FilteringHandler(@TypeOf(handler.*)){ .inner = handler, .kinds = options.kinds }; + client.run(&filtering) catch |err| switch (err) { + error.CursorTooOld => { + // §14: re-backfill from the last durably-processed seq, + // requiring strict cursor progress within the stall bound + const resume_seq = @max(client.lastSeq(), cutover); + if (resume_seq <= cursor) { + stalls += 1; + if (stalls >= max_rebackfill_stalls) { + log.err("re-backfill made no progress after {d} cursor-too-old cycles at seq {d}", .{ stalls, resume_seq }); + return error.RebackfillStalled; + } + } else { + stalls = 0; + } + cursor = resume_seq; + continue; + }, + else => |e| return e, + }; + return; // clean stop: handler asked, or cancellation unwound + } +} + +fn liveOptions(options: Options, cursor: ?u64, dedup_floor: u64, dict: ?[]const u8) live_mod.Options { + var out: live_mod.Options = .{ + .host = options.host, + .cursor = cursor, + .dedup_floor = dedup_floor, + .kinds = options.kinds, + .collections = options.collections, + .dids = options.dids, + .zstd_dict = dict, + }; + if (options.backoff_min_ns > 0) out.backoff_min_ns = options.backoff_min_ns; + if (options.backoff_max_ns > 0) out.backoff_max_ns = options.backoff_max_ns; + return out; +} + +/// archive-half adapter: receives every row (with seq) from the sweep and +/// applies the kinds filter — the plan prunes server-side where it can, +/// but the client-side matcher remains the correctness backstop +fn SweepHandler(comptime H: type) type { + return struct { + inner: *H, + kinds: []const livedecode.Kind, + + pub fn onRow(self: *@This(), event: livedecode.Event) bool { + if (!wantsKind(self.kinds, event)) return true; + return self.inner.onEvent(event); + } + }; +} + +/// live-half adapter: same kinds backstop over the tail (the server also +/// filters; upstream wantsLive) +fn FilteringHandler(comptime H: type) type { + return struct { + inner: *H, + kinds: []const livedecode.Kind, + + pub fn onEvent(self: *@This(), event: livedecode.Event) bool { + if (!wantsKind(self.kinds, event)) return true; + return self.inner.onEvent(event); + } + + pub fn onError(self: *@This(), err: anyerror) bool { + if (comptime @hasDecl(H, "onError")) return self.inner.onError(err); + log.warn("live tail: {s} (reconnecting)", .{@errorName(err)}); + return true; + } + + pub fn onInfo(self: *@This(), info: livedecode.Info) void { + if (comptime @hasDecl(H, "onInfo")) return self.inner.onInfo(info); + log.info("live stream info: {s}: {s}", .{ info.name, info.message }); + } + + pub fn onConnect(self: *@This(), host: []const u8) void { + if (comptime @hasDecl(H, "onConnect")) return self.inner.onConnect(host); + } + }; +} + +fn wantsKind(kinds: []const livedecode.Kind, event: livedecode.Event) bool { + if (kinds.len == 0) return true; + const kind: livedecode.Kind = event.payload; + for (kinds) |k| if (k == kind) return true; + return false; +} + +/// fetch the server's current live-tail compression dictionary. nil on any +/// failure: compression is an optimization, so a failed fetch degrades to +/// an uncompressed tail (logged) rather than failing the stream. +fn fetchZstdDict(io: Io, allocator: Allocator, host: []const u8) ?[]u8 { + var transport = zat.HttpTransport.init(io, allocator); + defer transport.deinit(); + var url_buf: [512]u8 = undefined; + const url = std.fmt.bufPrint(&url_buf, "{s}/xrpc/network.bsky.jetstream.getZstdDictionary", .{host}) catch return null; + var response = transport.fetch(.{ .url = url, .max_response_size = 16 << 20 }) catch |err| { + log.warn("getZstdDictionary failed; live tail will be uncompressed: {s}", .{@errorName(err)}); + return null; + }; + defer response.deinit(allocator); + if (response.status != .ok) { + log.warn("getZstdDictionary returned {d}; live tail will be uncompressed", .{@intFromEnum(response.status)}); + return null; + } + return allocator.dupe(u8, response.body) catch null; +} + +// === tests === + +const testing = std.testing; + +test "cutover and resume stay monotonic non-decreasing" { + // the max() that prevents a re-learned sealed tip BELOW the cursor from + // regressing the dedup floor (upstream's §14 anti-regression comment) + try testing.expectEqual(@as(u64, 100), @max(@as(u64, 80), @as(u64, 100))); // tip behind cursor + try testing.expectEqual(@as(u64, 120), @max(@as(u64, 120), @as(u64, 100))); // tip ahead +} + +test "kinds matcher: empty admits all, otherwise exact" { + const ev_commit = livedecode.Event{ .seq = 1, .time_us = 0, .did = "did:plc:a", .payload = .{ .commit = .{ + .operation = .delete, + .collection = "c", + .rkey = "k", + .rev = "r", + } } }; + const ev_sync = livedecode.Event{ .seq = 2, .time_us = 0, .did = "did:plc:a", .payload = .{ .sync = .{ + .did = "did:plc:a", + .rev = "r", + } } }; + try testing.expect(wantsKind(&.{}, ev_commit)); + try testing.expect(wantsKind(&.{.commit}, ev_commit)); + try testing.expect(!wantsKind(&.{.commit}, ev_sync)); + try testing.expect(wantsKind(&.{ .commit, .sync }, ev_sync)); +} diff --git a/src/live.zig b/src/live.zig new file mode 100644 --- /dev/null +++ b/src/live.zig @@ -0,0 +1,504 @@ +//! subscribeEvents live client — the v2 live tail. +//! +//! ports upstream live.go (bluesky-social/jetstream, pin 289b032): +//! dial /xrpc/network.bsky.jetstream.subscribeEvents offering the +//! xrpc.v1.json subprotocol, decode lexicon frames (livedecode.zig), +//! deduplicate the at-least-once reconnect overlap by seq, and reconnect +//! with bounded exponential backoff. reconnects resume at the highest +//! delivered seq, so events produced while disconnected are replayed by +//! the server instead of silently dropped. +//! +//! pre-upgrade rejections are typed exactly like upstream (the server +//! answers the upgrade with an HTTP 400 xrpc error envelope, surfaced via +//! websocket.zig's Client.handshake_failure): +//! - CursorTooOld is TERMINAL: the seq will not become valid by +//! retrying (the lookback floor only advances). returned to the +//! caller so the cutover engine re-enters archive backfill. +//! - InvalidRequest is PERMANENT: the same immutable filter cannot +//! succeed on reconnect; returned as fatal. +//! - UnknownZstdDictionary is RECOVERABLE in place: the server rotated +//! its dictionary; refetch it (or degrade to uncompressed) and +//! reconnect — never 400-loop on an ID the server keeps refusing. +//! +//! dict-zstd compression is negotiated at the application layer +//! (?zstdDictionary=; event frames arrive as BINARY zstd frames), and +//! the decompressed size is bounded by the read limit. v2 never +//! negotiates permessage-deflate (upstream removed it server-side). + +const std = @import("std"); +const zat = @import("zat"); +const websocket = @import("websocket"); +const livedecode = @import("livedecode.zig"); +const zstd = @import("zstd.zig"); + +const Io = std.Io; +const Allocator = std.mem.Allocator; +const posix = std.posix; +const log = std.log.scoped(.jetstream); + +pub const subscribe_nsid = livedecode.nsid; +pub const subprotocol = "xrpc.v1.json"; + +/// bounds a single websocket message; also caps a compressed frame's +/// decompressed size (upstream defaultLiveReadLimit) +pub const default_read_limit: usize = 32 << 20; + +const default_backoff_min_ns: u64 = 250 * std.time.ns_per_ms; +const default_backoff_max_ns: u64 = 30 * std.time.ns_per_s; + +pub const Options = struct { + /// base URL of the instance, e.g. "https://jetstream1.us-east.bsky.network" + /// (plain http for loopback tests) + host: []const u8, + /// wire resume point. null = start at the live tip (the cursor param is + /// omitted; upstream WithLiveCursor(0)); 0 = replay from the beginning + /// of the server's retention; a positive seq resumes inclusively — the + /// client's own seq dedup turns that into "> last delivered". + cursor: ?u64 = null, + /// highest seq the caller already holds: the at-least-once re-delivery + /// at or below it is dropped. 0 = nothing delivered yet, so the first + /// real event (seq >= 1) always passes. + dedup_floor: u64 = 0, + kinds: []const livedecode.Kind = &.{}, + collections: []const []const u8 = &.{}, + dids: []const []const u8 = &.{}, + read_limit: usize = default_read_limit, + backoff_min_ns: u64 = default_backoff_min_ns, + backoff_max_ns: u64 = default_backoff_max_ns, + /// dictionary blob from getZstdDictionary; opts into dict-zstd frames. + /// null = plain uncompressed text frames. + zstd_dict: ?[]const u8 = null, + /// re-fetches the CURRENT dictionary after the server rejects the + /// pinned ID (rotation). null disables in-place recovery; the client + /// then degrades to an uncompressed tail. + refetch_dict: ?*const fn (ctx: ?*anyopaque, allocator: Allocator) ?[]u8 = null, + refetch_dict_ctx: ?*anyopaque = null, +}; + +/// terminal outcomes of run(); everything transient reconnects internally +pub const RunError = error{ + /// pre-upgrade 400 CursorTooOld: re-enter backfill from lastSeq() + CursorTooOld, + /// pre-upgrade 400 InvalidRequest: the filter can never succeed + InvalidRequest, + Canceled, + OutOfMemory, +}; + +pub const LiveClient = struct { + io: Io, + allocator: Allocator, + options: Options, + /// highest seq delivered; the dedup floor and the reconnect resume + /// cursor. read after run() returns (the cutover engine resumes a + /// re-backfill from it). + last_seq: u64 = 0, + seen_any: bool = false, + decoder: ?zstd.DictDecoder = null, + /// owned copy of a refetched dictionary blob (rotation recovery) + owned_dict: ?[]u8 = null, + + pub fn init(io: Io, allocator: Allocator, options: Options) LiveClient { + var self: LiveClient = .{ + .io = io, + .allocator = allocator, + .options = options, + .last_seq = options.dedup_floor, + }; + if (options.zstd_dict) |dict| { + self.decoder = zstd.DictDecoder.init(dict, options.read_limit) catch blk: { + // the blob came from getZstdDictionary moments ago; a parse + // failure is a server/transport fault, not a reason to + // crash. degrade to uncompressed (documented, logged). + log.warn("invalid zstd dictionary; live tail will be uncompressed", .{}); + break :blk null; + }; + } + return self; + } + + pub fn deinit(self: *LiveClient) void { + if (self.decoder) |*d| d.deinit(); + if (self.owned_dict) |owned| self.allocator.free(owned); + self.* = undefined; + } + + pub fn lastSeq(self: *const LiveClient) u64 { + return self.last_seq; + } + + /// tail the live stream, invoking handler.onEvent for each decoded + /// event in delivery order. handler contract: + /// fn onEvent(*H, livedecode.Event) bool — false stops the tail + /// optional fn onError(*H, anyerror) bool — recoverable read/decode/ + /// reconnect errors; false stops (default: log and continue) + /// optional fn onInfo(*H, livedecode.Info) void — #info advisories + /// returns null on a clean stop (handler asked), typed RunError on a + /// terminal pre-upgrade rejection or cancellation. + pub fn run(self: *LiveClient, handler: anytype) RunError!void { + var backoff = self.options.backoff_min_ns; + while (true) { + const seq_before = self.last_seq; + const outcome = self.session(handler) catch |err| switch (err) { + error.Canceled => return error.Canceled, + error.OutOfMemory => return error.OutOfMemory, + // transient transport/dial/read failure: reconnect below + else => SessionEnd{ .transient = err }, + }; + switch (outcome) { + .stopped => return, + .cursor_too_old => return error.CursorTooOld, + .invalid_request => return error.InvalidRequest, + .dict_rejected => self.refreshDict(), + .transient => |err| { + if (!self.emitError(handler, err)) return; + }, + } + // a session that delivered new events was healthy: reset the + // backoff so a long-lived connection that finally drops + // reconnects promptly, not at the accumulated max + if (self.last_seq != seq_before) backoff = self.options.backoff_min_ns; + self.io.sleep(Io.Duration.fromNanoseconds(@intCast(backoff)), .awake) catch return error.Canceled; + backoff = @min(backoff * 2, self.options.backoff_max_ns); + } + } + + const SessionEnd = union(enum) { + stopped, + cursor_too_old, + invalid_request, + dict_rejected, + transient: anyerror, + }; + + fn emitError(self: *LiveClient, handler: anytype, err: anyerror) bool { + _ = self; + if (comptime @hasDecl(@TypeOf(handler.*), "onError")) return handler.onError(err); + log.warn("live tail reconnecting: {s}", .{@errorName(err)}); + return true; + } + + fn session(self: *LiveClient, handler: anytype) !SessionEnd { + const target = try parseHost(self.options.host); + var path_buf: [4096]u8 = undefined; + const path = try self.subscribePath(&path_buf); + + var client = try websocket.Client.init(self.io, self.allocator, .{ + .host = target.host, + .port = target.port, + .tls = target.tls, + .max_size = self.options.read_limit, + }); + defer client.deinit(); + + var headers_buf: [512]u8 = undefined; + const headers = try std.fmt.bufPrint( + &headers_buf, + "Host: {s}\r\nSec-WebSocket-Protocol: {s}\r\n", + .{ target.host, subprotocol }, + ); + client.handshake(path, .{ .headers = headers }) catch |err| { + if (err == error.InvalidHandshakeResponse) { + if (client.handshake_failure) |failure| return classifyRejection(failure); + } + return err; + }; + configureKeepalive(&client); + + if (comptime @hasDecl(@TypeOf(handler.*), "onConnect")) handler.onConnect(target.host); + + var ws_handler: WsHandler(@TypeOf(handler.*)) = .{ .client = self, .handler = handler }; + client.readLoop(&ws_handler) catch |err| switch (err) { + error.Canceled => return error.Canceled, + else => return .{ .transient = err }, + }; + if (ws_handler.stopped) return .stopped; + if (ws_handler.failure) |err| switch (err) { + error.OutOfMemory => return error.OutOfMemory, + else => return .{ .transient = err }, + }; + // the server closed (possibly right after a terminal error frame, + // which the handler surfaced through onError already) + return .{ .transient = error.ConnectionClosed }; + } + + /// pre-upgrade rejection classification: match the structured xrpc + /// envelope's error NAME, never body substrings (the wire contract; + /// upstream dialWebsocket). any other status is transient — a proxy + /// 502/503 during a deploy reconnect-loops like an abrupt close. + fn classifyRejection(failure: websocket.Client.HandshakeFailure) SessionEnd { + if (failure.status == 400) { + var name_buf: [1024]u8 = undefined; + const name = xrpcErrorName(failure.body(), &name_buf) orelse ""; + if (std.mem.eql(u8, name, "CursorTooOld")) return .cursor_too_old; + if (std.mem.eql(u8, name, "UnknownZstdDictionary")) return .dict_rejected; + if (std.mem.eql(u8, name, "InvalidRequest")) return .invalid_request; + } + return .{ .transient = error.HandshakeRejected }; + } + + fn xrpcErrorName(body: []const u8, buf: []u8) ?[]const u8 { + var fba = std.heap.FixedBufferAllocator.init(buf); + // bounded parse of {"error": name, ...}; allocation failure or + // malformed JSON just means "no structured name" + const parsed = std.json.parseFromSliceLeaky(std.json.Value, fba.allocator(), body, .{}) catch return null; + if (parsed != .object) return null; + const v = parsed.object.get("error") orelse return null; + return if (v == .string) v.string else null; + } + + /// recover from a server-side dictionary rotation: refetch the current + /// dictionary and swap the decoder so the next dial negotiates the new + /// ID. when the refetch is unavailable, fails, or returns the very ID + /// the server just rejected (mixed-version fleet), degrade to an + /// uncompressed tail — compression is an optimization; the tail must + /// keep flowing. (upstream refreshDict) + fn refreshDict(self: *LiveClient) void { + var current = self.decoder orelse return; + const rejected = current.id; + if (self.options.refetch_dict) |refetch| { + if (refetch(self.options.refetch_dict_ctx, self.allocator)) |blob| { + if (zstd.DictDecoder.init(blob, self.options.read_limit)) |fresh| { + if (fresh.id != rejected) { + current.deinit(); + if (self.owned_dict) |owned| self.allocator.free(owned); + self.owned_dict = blob; + self.decoder = fresh; + log.info("live zstd dictionary rotated; refetched (rejected id {d}, new id {d})", .{ rejected, self.decoder.?.id }); + return; + } + var discard = fresh; + discard.deinit(); + } else |_| {} + self.allocator.free(blob); + } + } + current.deinit(); + self.decoder = null; + log.warn("live zstd dictionary rejected and refetch unavailable; continuing uncompressed (rejected id {d})", .{rejected}); + } + + fn subscribePath(self: *LiveClient, buf: []u8) ![]const u8 { + var w: Io.Writer = .fixed(buf); + try w.writeAll("/xrpc/" ++ subscribe_nsid); + var sep: u8 = '?'; + // wire cursor: once any event has been delivered, resume at the + // highest delivered seq — this is what keeps a reconnect from + // re-anchoring at the tip and silently dropping events produced + // while disconnected. before any delivery, use the configured + // start: omit when null (live from tip), else send it. + if (self.seen_any) { + try w.print("{c}cursor={d}", .{ sep, self.last_seq }); + sep = '&'; + } else if (self.options.cursor) |cursor| { + try w.print("{c}cursor={d}", .{ sep, cursor }); + sep = '&'; + } + for (self.options.kinds) |kind| { + try w.print("{c}kinds={s}", .{ sep, @tagName(kind) }); + sep = '&'; + } + for (self.options.collections) |collection| { + try w.print("{c}collections={s}", .{ sep, collection }); + sep = '&'; + } + for (self.options.dids) |did| { + try w.print("{c}dids={s}", .{ sep, did }); + sep = '&'; + } + if (self.decoder) |decoder| { + try w.print("{c}zstdDictionary={d}", .{ sep, decoder.id }); + sep = '&'; + } + return w.buffered(); + } + + fn WsHandler(comptime H: type) type { + return struct { + client: *LiveClient, + handler: *H, + stopped: bool = false, + failure: ?anyerror = null, + + const Self = @This(); + + pub fn serverMessage(self: *Self, data: []const u8, message_type: enum { text, binary }) !void { + const client = self.client; + var frame: []const u8 = data; + var decompressed: ?[]u8 = null; + defer if (decompressed) |owned| client.allocator.free(owned); + switch (message_type) { + .binary => { + // dict-zstd connection: every event frame is a + // BINARY zstd frame. on an uncompressed connection + // stray binary is ignored (jetstream frames are + // text JSON). + const decoder = if (client.decoder) |*d| d else return; + decompressed = decoder.decompressAlloc(client.allocator, data) catch |err| { + // upstream input, never crash: surface and keep the tail + if (!client.emitError(self.handler, err)) { + self.stopped = true; + return error.Stop; + } + return; + }; + frame = decompressed.?; + }, + .text => {}, + } + + var arena = std.heap.ArenaAllocator.init(client.allocator); + defer arena.deinit(); + const decoded = livedecode.decodeFrame(arena.allocator(), frame) catch |err| { + if (err == error.OutOfMemory) { + self.failure = error.OutOfMemory; + return error.Stop; + } + // a malformed data frame is upstream input; surface it + // but keep the connection (one bad frame must not drop + // the tail) + if (!client.emitError(self.handler, err)) { + self.stopped = true; + return error.Stop; + } + return; + }; + switch (decoded) { + .skip => {}, + .info => |info| { + if (comptime @hasDecl(H, "onInfo")) { + self.handler.onInfo(info); + } else { + log.info("live stream info frame: {s}: {s}", .{ info.name, info.message }); + } + }, + .stream_error => |stream_err| { + // terminal error frame: the server closes right + // after sending it. surface the typed reason; the + // reconnect loop handles the close that follows. + log.warn("live stream error frame: {s}: {s}", .{ stream_err.code, stream_err.message }); + if (!client.emitError(self.handler, error.StreamErrorFrame)) { + self.stopped = true; + return error.Stop; + } + }, + .event => |event| { + // deduplicate the at-least-once reconnect overlap: + // skip anything at or below the highest delivered + // seq. last_seq 0 with nothing delivered means the + // first real event (seq >= 1) passes. + if (event.seq <= client.last_seq) return; + client.last_seq = event.seq; + client.seen_any = true; + if (!self.handler.onEvent(event)) { + self.stopped = true; + return error.Stop; + } + }, + } + } + + pub fn close(_: *Self) void {} + }; + } +}; + +const Target = struct { + host: []const u8, + port: u16, + tls: bool, +}; + +fn parseHost(base: []const u8) !Target { + const uri = std.Uri.parse(base) catch return error.InvalidHost; + const tls = std.ascii.eqlIgnoreCase(uri.scheme, "https") or std.ascii.eqlIgnoreCase(uri.scheme, "wss"); + if (!tls and !std.ascii.eqlIgnoreCase(uri.scheme, "http") and !std.ascii.eqlIgnoreCase(uri.scheme, "ws")) + return error.InvalidHost; + const host_component = uri.host orelse return error.InvalidHost; + const host = switch (host_component) { + .raw => |raw| raw, + .percent_encoded => |enc| enc, + }; + if (host.len == 0) return error.InvalidHost; + return .{ .host = host, .port = uri.port orelse (if (tls) @as(u16, 443) else 80), .tls = tls }; +} + +fn configureKeepalive(client: *websocket.Client) void { + // TCP keepalive catches half-open sockets the read loop would block on + // forever; mirrors zat.JetstreamClient (10s idle, 5s interval, 2 probes) + const fd = client.stream.stream.socket.handle; + const builtin = @import("builtin"); + posix.setsockopt(fd, posix.SOL.SOCKET, posix.SO.KEEPALIVE, &std.mem.toBytes(@as(i32, 1))) catch return; + const tcp: i32 = @intCast(posix.IPPROTO.TCP); + if (builtin.os.tag == .linux) { + posix.setsockopt(fd, tcp, posix.TCP.KEEPIDLE, &std.mem.toBytes(@as(i32, 10))) catch return; + } else if (builtin.os.tag == .macos) { + posix.setsockopt(fd, tcp, posix.TCP.KEEPALIVE, &std.mem.toBytes(@as(i32, 10))) catch return; + } + posix.setsockopt(fd, tcp, posix.TCP.KEEPINTVL, &std.mem.toBytes(@as(i32, 5))) catch return; + posix.setsockopt(fd, tcp, posix.TCP.KEEPCNT, &std.mem.toBytes(@as(i32, 2))) catch return; +} + +// === tests === + +const testing = std.testing; + +test "subscribe path: cursor semantics, filters, and dictionary id" { + var client = LiveClient.init(std.Options.debug_io, testing.allocator, .{ + .host = "https://example.test", + .cursor = null, + }); + defer client.deinit(); + var buf: [4096]u8 = undefined; + // from tip: no cursor param at all + try testing.expectEqualStrings("/xrpc/" ++ subscribe_nsid, try client.subscribePath(&buf)); + + // configured cursor before any delivery + client.options.cursor = 0; + try testing.expectEqualStrings("/xrpc/" ++ subscribe_nsid ++ "?cursor=0", try client.subscribePath(&buf)); + + // after delivery, reconnects anchor at the highest delivered seq even + // when the start was from-tip + client.options.cursor = null; + client.last_seq = 41; + client.seen_any = true; + try testing.expectEqualStrings("/xrpc/" ++ subscribe_nsid ++ "?cursor=41", try client.subscribePath(&buf)); + + client.options.kinds = &.{ .commit, .identity }; + client.options.collections = &.{"app.bsky.feed.post"}; + client.options.dids = &.{"did:plc:abc"}; + try testing.expectEqualStrings( + "/xrpc/" ++ subscribe_nsid ++ "?cursor=41&kinds=commit&kinds=identity&collections=app.bsky.feed.post&dids=did:plc:abc", + try client.subscribePath(&buf), + ); +} + +test "rejection classification matches the wire contract by error name" { + const mk = struct { + fn failure(status: u16, body: []const u8) websocket.Client.HandshakeFailure { + var f: websocket.Client.HandshakeFailure = .{ .status = status }; + @memcpy(f.body_buf[0..body.len], body); + f.body_len = body.len; + return f; + } + }; + try testing.expectEqual( + LiveClient.SessionEnd.cursor_too_old, + LiveClient.classifyRejection(mk.failure(400, "{\"error\":\"CursorTooOld\",\"message\":\"below floor 9\"}")), + ); + try testing.expectEqual( + LiveClient.SessionEnd.dict_rejected, + LiveClient.classifyRejection(mk.failure(400, "{\"error\":\"UnknownZstdDictionary\"}")), + ); + try testing.expectEqual( + LiveClient.SessionEnd.invalid_request, + LiveClient.classifyRejection(mk.failure(400, "{\"error\":\"InvalidRequest\",\"message\":\"kinds\"}")), + ); + // name matching, never substrings: a 400 whose message MENTIONS + // CursorTooOld is not a cursor rejection + const mention = LiveClient.classifyRejection(mk.failure(400, "{\"error\":\"Nope\",\"message\":\"CursorTooOld\"}")); + try testing.expect(mention == .transient); + // proxy 503 during a deploy is transient, reconnect-loop territory + const proxy = LiveClient.classifyRejection(mk.failure(503, "starting")); + try testing.expect(proxy == .transient); +} diff --git a/src/livedecode.zig b/src/livedecode.zig new file mode 100644 --- /dev/null +++ b/src/livedecode.zig @@ -0,0 +1,385 @@ +//! subscribeEvents frame decoding — the live half of the v2 wire. +//! +//! ports upstream livedecode.go (bluesky-social/jetstream, pin 289b032): +//! one xrpc.v1.json envelope per text frame, discriminated by $type +//! ("message" | "error"); message payloads are the lexicon union +//! network.bsky.jetstream.subscribeEvents#{commit,identity,account,sync,info}. +//! +//! decode policy, matching upstream exactly: +//! - #info advisories surface as .info (the session loop logs them; no +//! seq, no cursor advance) +//! - an unknown envelope or payload $type is a NEWER server's frame kind: +//! skip for forward compatibility +//! - a MISSING $type (envelope or payload) is malformed, not future: +//! error, so a wrong endpoint or protocol revision cannot look healthy +//! while delivering nothing +//! - lexicon-required fields are enforced here (the JSON layer cannot): +//! seq >= 1, commit did/rev/collection/rkey, and payload-presence DIDs +//! on identity/account/sync +//! - untrusted diagnostic strings (error codes, #info names/messages) +//! are bounded before they can reach logs +//! +//! one recorded divergence: upstream reconstructs each commit record's +//! canonical DAG-CBOR from the wire JSON (its Event API carries RecordCBOR +//! for typed-CBOR consumers and CID verification). zat.cbor has no +//! JSON→DAG-CBOR encoder, so Commit carries the raw wire JSON slice and its +//! parsed zat.json.Value instead — the shapes zig consumers fold on. + +const std = @import("std"); +const zat = @import("zat"); + +const Allocator = std.mem.Allocator; + +pub const nsid = "network.bsky.jetstream.subscribeEvents"; + +/// bounds on untrusted server-supplied diagnostic strings before they enter +/// error values and logs (upstream maxLiveDiag{Name,Message}Bytes) +pub const max_diag_name_bytes = 128; +pub const max_diag_message_bytes = 1024; + +pub const Kind = enum { commit, identity, account, sync }; +pub const Operation = enum { create, update, delete }; + +pub const Commit = struct { + operation: Operation, + collection: []const u8, + rkey: []const u8, + rev: []const u8, + cid: ?[]const u8 = null, + /// parsed record for create/update; null on delete + record: ?std.json.Value = null, +}; + +pub const Identity = struct { + did: []const u8, + handle: ?[]const u8 = null, + seq: ?i64 = null, + time: ?[]const u8 = null, +}; + +pub const Account = struct { + did: []const u8, + active: bool, + status: ?[]const u8 = null, + seq: ?i64 = null, + time: ?[]const u8 = null, +}; + +pub const Sync = struct { + did: []const u8, + rev: []const u8, + seq: ?i64 = null, + time: ?[]const u8 = null, +}; + +/// one decoded live event. slices reference the frame buffer and the +/// decode arena; they are valid only during the handler callback. +pub const Event = struct { + seq: u64, + time_us: i64, + did: []const u8, + payload: union(Kind) { + commit: Commit, + identity: Identity, + account: Account, + sync: Sync, + }, +}; + +/// an #info advisory (e.g. OutdatedCursor on a clamped timestamp resume) +pub const Info = struct { + name: []const u8, + message: []const u8, +}; + +/// a terminal xrpc.v1.json error frame; the server closes right after +/// sending one +pub const StreamError = struct { + code: []const u8, + message: []const u8, +}; + +pub const Decoded = union(enum) { + event: Event, + info: Info, + stream_error: StreamError, + /// valid frame from a newer protocol revision; advance without emitting + skip, +}; + +pub const DecodeError = error{ + MalformedFrame, + MissingEnvelopeType, + MissingPayload, + MissingPayloadType, + MissingRequiredField, + InvalidSeq, + InvalidTime, + UnknownOperation, + MissingRecord, + OutOfMemory, +}; + +fn bound(s: []const u8, limit: usize) []const u8 { + if (s.len <= limit) return s; + var cut = limit; + // rune-aligned truncation, like upstream boundLiveString + while (cut > 0 and (s[cut] & 0xC0) == 0x80) cut -= 1; + return s[0..cut]; +} + +/// decode one text frame. `arena` owns the parsed JSON structure; the +/// caller keeps it alive for the duration of the handler callback. +pub fn decodeFrame(arena: Allocator, data: []const u8) DecodeError!Decoded { + const parsed = std.json.parseFromSliceLeaky(std.json.Value, arena, data, .{}) catch |err| switch (err) { + error.OutOfMemory => return error.OutOfMemory, + else => return error.MalformedFrame, + }; + if (parsed != .object) return error.MalformedFrame; + const env = parsed.object; + + const env_type = getString(env, "$type") orelse { + // no $type at all is not a newer revision — it is a malformed frame + // (a v1 /subscribe server, or not a subscribeEvents endpoint at all) + return error.MissingEnvelopeType; + }; + if (std.mem.eql(u8, env_type, "error")) { + const code = getString(env, "error") orelse return error.MalformedFrame; + return .{ .stream_error = .{ + .code = bound(code, max_diag_name_bytes), + .message = bound(getString(env, "message") orelse "", max_diag_message_bytes), + } }; + } + if (!std.mem.eql(u8, env_type, "message")) return .skip; + + const payload_value = env.get("payload") orelse return error.MissingPayload; + if (payload_value != .object) return error.MissingPayload; + const payload = payload_value.object; + const payload_type = getString(payload, "$type") orelse return error.MissingPayloadType; + + if (std.mem.eql(u8, payload_type, nsid ++ "#info")) { + return .{ .info = .{ + .name = bound(getString(payload, "name") orelse "", max_diag_name_bytes), + .message = bound(getString(payload, "message") orelse "", max_diag_message_bytes), + } }; + } + + const kind: Kind = if (std.mem.eql(u8, payload_type, nsid ++ "#commit")) + .commit + else if (std.mem.eql(u8, payload_type, nsid ++ "#identity")) + .identity + else if (std.mem.eql(u8, payload_type, nsid ++ "#account")) + .account + else if (std.mem.eql(u8, payload_type, nsid ++ "#sync")) + .sync + else + return .skip; // a newer server's message kind + + // envelope fields shared by every message kind: 1-based seq (0 = the + // required field was absent; accepting it would hand the dedup an event + // it silently swallows) and the canonical datetime + const seq_raw = getInt(payload, "seq") orelse return error.InvalidSeq; + if (seq_raw <= 0) return error.InvalidSeq; + const seq: u64 = @intCast(seq_raw); + const time_str = getString(payload, "time") orelse return error.InvalidTime; + const time_us = (zat.Datetime.parse(time_str) orelse return error.InvalidTime).micros; + const did = getString(payload, "did") orelse return error.MissingRequiredField; + if (did.len == 0) return error.MissingRequiredField; + + switch (kind) { + .commit => { + const collection = getString(payload, "collection") orelse return error.MissingRequiredField; + const rkey = getString(payload, "rkey") orelse return error.MissingRequiredField; + const rev = getString(payload, "rev") orelse return error.MissingRequiredField; + if (collection.len == 0 or rkey.len == 0 or rev.len == 0) return error.MissingRequiredField; + const op_str = getString(payload, "operation") orelse return error.MissingRequiredField; + const operation = std.meta.stringToEnum(Operation, op_str) orelse return error.UnknownOperation; + var record: ?std.json.Value = null; + switch (operation) { + .create, .update => { + record = payload.get("record") orelse return error.MissingRecord; + if (record.? != .object) return error.MissingRecord; + }, + .delete => {}, // no record payload on deletes + } + return .{ .event = .{ .seq = seq, .time_us = time_us, .did = did, .payload = .{ .commit = .{ + .operation = operation, + .collection = collection, + .rkey = rkey, + .rev = rev, + .cid = getString(payload, "cid"), + .record = record, + } } } }; + }, + .identity => { + const inner_value = payload.get("identity") orelse return error.MissingRequiredField; + if (inner_value != .object) return error.MissingRequiredField; + const inner = inner_value.object; + // payload-presence check (did is set by every producer); scalar + // required fields are indistinguishable from their zero values + const inner_did = getString(inner, "did") orelse return error.MissingRequiredField; + if (inner_did.len == 0) return error.MissingRequiredField; + return .{ .event = .{ .seq = seq, .time_us = time_us, .did = did, .payload = .{ .identity = .{ + .did = inner_did, + .handle = getString(inner, "handle"), + .seq = getInt(inner, "seq"), + .time = getString(inner, "time"), + } } } }; + }, + .account => { + const inner_value = payload.get("account") orelse return error.MissingRequiredField; + if (inner_value != .object) return error.MissingRequiredField; + const inner = inner_value.object; + const inner_did = getString(inner, "did") orelse return error.MissingRequiredField; + if (inner_did.len == 0) return error.MissingRequiredField; + const active_value = inner.get("active") orelse return error.MissingRequiredField; + if (active_value != .bool) return error.MissingRequiredField; + return .{ .event = .{ .seq = seq, .time_us = time_us, .did = did, .payload = .{ .account = .{ + .did = inner_did, + .active = active_value.bool, + .status = getString(inner, "status"), + .seq = getInt(inner, "seq"), + .time = getString(inner, "time"), + } } } }; + }, + .sync => { + const inner_value = payload.get("sync") orelse return error.MissingRequiredField; + if (inner_value != .object) return error.MissingRequiredField; + const inner = inner_value.object; + // archived #sync payloads from an async resync legitimately + // carry empty time/seq; did is the only reliable presence marker + const inner_did = getString(inner, "did") orelse return error.MissingRequiredField; + if (inner_did.len == 0) return error.MissingRequiredField; + return .{ .event = .{ .seq = seq, .time_us = time_us, .did = did, .payload = .{ .sync = .{ + .did = inner_did, + .rev = getString(inner, "rev") orelse "", + .seq = getInt(inner, "seq"), + .time = getString(inner, "time"), + } } } }; + }, + } +} + +fn getString(obj: std.json.ObjectMap, key: []const u8) ?[]const u8 { + const v = obj.get(key) orelse return null; + return if (v == .string) v.string else null; +} + +fn getInt(obj: std.json.ObjectMap, key: []const u8) ?i64 { + const v = obj.get(key) orelse return null; + return if (v == .integer) v.integer else null; +} + +// === tests === +// fixtures mirror stream's wire.zig v2 encoder output byte-shape and +// upstream livedecode_test.go's malformed-frame matrix + +const testing = std.testing; + +fn decodeTest(arena: Allocator, frame: []const u8) DecodeError!Decoded { + return decodeFrame(arena, frame); +} + +test "commit frame decodes with envelope fields and flat commit payload" { + var arena = std.heap.ArenaAllocator.init(testing.allocator); + defer arena.deinit(); + const frame = + \\{"$type":"message","payload":{"$type":"network.bsky.jetstream.subscribeEvents#commit","cid":"bafy123","collection":"app.bsky.feed.post","did":"did:plc:abc","operation":"create","record":{"$type":"app.bsky.feed.post","text":"hi"},"rev":"3k2a","rkey":"3k2abcdefghij","seq":42,"time":"2026-07-13T00:00:01.500000Z"}} + ; + const decoded = try decodeTest(arena.allocator(), frame); + const ev = decoded.event; + try testing.expectEqual(@as(u64, 42), ev.seq); + try testing.expectEqual(@as(i64, (1783900800 + 1) * std.time.us_per_s + 500_000), ev.time_us); + try testing.expectEqualStrings("did:plc:abc", ev.did); + const commit = ev.payload.commit; + try testing.expectEqual(Operation.create, commit.operation); + try testing.expectEqualStrings("app.bsky.feed.post", commit.collection); + try testing.expectEqualStrings("3k2abcdefghij", commit.rkey); + try testing.expectEqualStrings("bafy123", commit.cid.?); + try testing.expectEqualStrings("hi", commit.record.?.object.get("text").?.string); +} + +test "delete commit carries no record; create without one errors" { + var arena = std.heap.ArenaAllocator.init(testing.allocator); + defer arena.deinit(); + const del = + \\{"$type":"message","payload":{"$type":"network.bsky.jetstream.subscribeEvents#commit","collection":"c","did":"did:plc:a","operation":"delete","rev":"r","rkey":"k","seq":1,"time":"2026-01-01T00:00:00.000000Z"}} + ; + const decoded = try decodeTest(arena.allocator(), del); + try testing.expect(decoded.event.payload.commit.record == null); + + const create_missing = + \\{"$type":"message","payload":{"$type":"network.bsky.jetstream.subscribeEvents#commit","collection":"c","did":"did:plc:a","operation":"create","rev":"r","rkey":"k","seq":1,"time":"2026-01-01T00:00:00.000000Z"}} + ; + try testing.expectError(error.MissingRecord, decodeTest(arena.allocator(), create_missing)); +} + +test "identity, account, and sync enforce payload-presence DIDs" { + var arena = std.heap.ArenaAllocator.init(testing.allocator); + defer arena.deinit(); + const identity = + \\{"$type":"message","payload":{"$type":"network.bsky.jetstream.subscribeEvents#identity","did":"did:plc:a","identity":{"did":"did:plc:a","handle":"alice.test","seq":7,"time":"t"},"seq":9,"time":"2026-01-01T00:00:00.000000Z"}} + ; + const id_ev = (try decodeTest(arena.allocator(), identity)).event; + try testing.expectEqualStrings("alice.test", id_ev.payload.identity.handle.?); + + const identity_hollow = + \\{"$type":"message","payload":{"$type":"network.bsky.jetstream.subscribeEvents#identity","did":"did:plc:a","identity":{},"seq":9,"time":"2026-01-01T00:00:00.000000Z"}} + ; + try testing.expectError(error.MissingRequiredField, decodeTest(arena.allocator(), identity_hollow)); + + const account = + \\{"$type":"message","payload":{"$type":"network.bsky.jetstream.subscribeEvents#account","account":{"active":false,"did":"did:plc:a","status":"takendown"},"did":"did:plc:a","seq":10,"time":"2026-01-01T00:00:00.000000Z"}} + ; + const acct = (try decodeTest(arena.allocator(), account)).event.payload.account; + try testing.expect(!acct.active); + try testing.expectEqualStrings("takendown", acct.status.?); + + const sync_frame = + \\{"$type":"message","payload":{"$type":"network.bsky.jetstream.subscribeEvents#sync","did":"did:plc:a","sync":{"did":"did:plc:a","rev":"3k2a"},"seq":11,"time":"2026-01-01T00:00:00.000000Z"}} + ; + const sync_ev = (try decodeTest(arena.allocator(), sync_frame)).event.payload.sync; + try testing.expectEqualStrings("3k2a", sync_ev.rev); +} + +test "error and info frames surface typed and bounded" { + var arena = std.heap.ArenaAllocator.init(testing.allocator); + defer arena.deinit(); + const err_frame = + \\{"$type":"error","error":"FutureCursor","message":"cursor is ahead of the stream"} + ; + const stream_err = (try decodeTest(arena.allocator(), err_frame)).stream_error; + try testing.expectEqualStrings("FutureCursor", stream_err.code); + + const info_frame = + \\{"$type":"message","payload":{"$type":"network.bsky.jetstream.subscribeEvents#info","name":"OutdatedCursor","message":"resumed from seq 5"}} + ; + const info = (try decodeTest(arena.allocator(), info_frame)).info; + try testing.expectEqualStrings("OutdatedCursor", info.name); +} + +test "malformed vs future frames: missing $type errors, unknown $type skips" { + var arena = std.heap.ArenaAllocator.init(testing.allocator); + defer arena.deinit(); + // v1 /subscribe JSON has no envelope $type: malformed, never skipped + try testing.expectError(error.MissingEnvelopeType, decodeTest(arena.allocator(), + \\{"did":"did:plc:a","time_us":1,"kind":"commit"} + )); + // a newer revision's envelope kind skips + try testing.expectEqual(Decoded.skip, try decodeTest(arena.allocator(), + \\{"$type":"snapshot","payload":{}} + )); + // a newer server's payload kind skips + try testing.expectEqual(Decoded.skip, try decodeTest(arena.allocator(), + \\{"$type":"message","payload":{"$type":"network.bsky.jetstream.subscribeEvents#hologram","seq":1}} + )); + // a payload with NO $type is malformed, not future + try testing.expectError(error.MissingPayloadType, decodeTest(arena.allocator(), + \\{"$type":"message","payload":{"seq":1}} + )); + // seq 0 means the required field was absent + try testing.expectError(error.InvalidSeq, decodeTest(arena.allocator(), + \\{"$type":"message","payload":{"$type":"network.bsky.jetstream.subscribeEvents#commit","collection":"c","did":"did:plc:a","operation":"delete","rev":"r","rkey":"k","seq":0,"time":"2026-01-01T00:00:00.000000Z"}} + )); + try testing.expectError(error.MalformedFrame, decodeTest(arena.allocator(), "not json")); +} diff --git a/src/root.zig b/src/root.zig --- a/src/root.zig +++ b/src/root.zig @@ -8,14 +8,25 @@ //! and its archive/replay API — and depends on zat for everything protocol. //! the same split Bluesky drew between @atproto/api and @bsky/sdk. //! -//! today: ArchiveBackfill (network replay: planSnapshot → getBlock/getSegment -//! → jss v1 decode → zat.JetstreamEvent through your live handler, api_key -//! supported). next: the subscribeEvents v2 live client (lexicon frames, -//! strict seq cursors, kinds/dids/collections). zat.JetstreamClient (the v1 +//! the client mirrors the official Go client's shape — one subscribe covers +//! the live tail (subscribeEvents lexicon frames, seq cursors, dict-zstd), +//! filtered archive replay (planSnapshot → getBlock/getSegment → jss v1), +//! and the gapless replay→live cutover. zat.JetstreamClient (the v1 //! /subscribe wire) remains in zat until consumers migrate. pub const ArchiveBackfill = @import("archive_backfill.zig"); +pub const livedecode = @import("livedecode.zig"); +pub const live = @import("live.zig"); +pub const LiveClient = live.LiveClient; +pub const client = @import("client.zig"); +pub const subscribe = client.subscribe; +pub const SubscribeOptions = client.Options; +pub const Event = livedecode.Event; +pub const Kind = livedecode.Kind; test { _ = @import("archive_backfill.zig"); + _ = @import("livedecode.zig"); + _ = @import("live.zig"); + _ = @import("client.zig"); } diff --git a/src/zstd.zig b/src/zstd.zig --- a/src/zstd.zig +++ b/src/zstd.zig @@ -14,6 +14,12 @@ extern fn ZSTD_getFrameContentSize(src: [*]const u8, src_size: usize) c_ulonglong; extern fn ZSTD_decompress(dst: [*]u8, dst_cap: usize, src: [*]const u8, src_len: usize) usize; extern fn ZSTD_isError(code: usize) c_uint; + extern fn ZSTD_createDCtx() ?*anyopaque; + extern fn ZSTD_freeDCtx(dctx: ?*anyopaque) usize; + extern fn ZSTD_createDDict(dict: [*]const u8, dict_size: usize) ?*anyopaque; + extern fn ZSTD_freeDDict(ddict: ?*anyopaque) usize; + extern fn ZSTD_decompress_usingDDict(dctx: ?*anyopaque, dst: [*]u8, dst_cap: usize, src: [*]const u8, src_len: usize, ddict: ?*anyopaque) usize; + extern fn ZSTD_getDictID_fromDict(dict: [*]const u8, dict_size: usize) c_uint; const contentsize_unknown: c_ulonglong = std.math.maxInt(c_ulonglong); // (ull)-1 const contentsize_error: c_ulonglong = std.math.maxInt(c_ulonglong) - 1; // (ull)-2 @@ -46,6 +52,55 @@ if (c.ZSTD_isError(written) != 0 or written != dst.len) return error.DecompressFailed; return dst; } + +/// dictionary ID declared in a dictionary blob's header (zero = not a +/// structured dictionary). the live tail pins this ID in ?zstdDictionary= +/// and the server refuses unknown IDs pre-upgrade. +pub fn dictId(dict: []const u8) u32 { + if (dict.len == 0) return 0; + return @intCast(c.ZSTD_getDictID_fromDict(dict.ptr, dict.len)); +} + +/// dictionary-seeded decoder for live subscribeEvents binary frames. +/// the decompressed-size cap mirrors the connection's read limit: an +/// uncompressed frame must fit the limit on the wire, so its compressed +/// twin may not expand past it either (upstream newZstdDecoder). +pub const DictDecoder = struct { + dctx: *anyopaque, + ddict: *anyopaque, + id: u32, + max_out: usize, + + pub fn init(dict: []const u8, max_out: usize) error{InvalidDictionary}!DictDecoder { + const id = dictId(dict); + if (id == 0) return error.InvalidDictionary; + const dctx = c.ZSTD_createDCtx() orelse return error.InvalidDictionary; + const ddict = c.ZSTD_createDDict(dict.ptr, dict.len) orelse { + _ = c.ZSTD_freeDCtx(dctx); + return error.InvalidDictionary; + }; + return .{ .dctx = dctx, .ddict = ddict, .id = id, .max_out = max_out }; + } + + pub fn deinit(self: *DictDecoder) void { + _ = c.ZSTD_freeDDict(self.ddict); + _ = c.ZSTD_freeDCtx(self.dctx); + self.* = undefined; + } + + pub fn decompressAlloc(self: *DictDecoder, allocator: Allocator, frame: []const u8) DecompressError![]u8 { + if (frame.len == 0) return error.DecompressFailed; + const size = c.ZSTD_getFrameContentSize(frame.ptr, frame.len); + if (size == c.contentsize_unknown) return error.UnknownFrameSize; + if (size == c.contentsize_error) return error.DecompressFailed; + if (size > self.max_out) return error.BlockOversize; + const dst = try allocator.alloc(u8, @intCast(size)); + errdefer allocator.free(dst); + const written = c.ZSTD_decompress_usingDDict(self.dctx, dst.ptr, dst.len, frame.ptr, frame.len, self.ddict); + if (c.ZSTD_isError(written) != 0 or written != dst.len) return error.DecompressFailed; + return dst; + } +}; // === tests ===