Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114const std = @import("std");const archive_mod = @import("../storage/archive.zig");const observability = @import("observability.zig");const segment = @import("../storage/segment.zig");
const Io = std.Io;
const Capture = struct { body: [128 * 1024]u8 = undefined, len: usize = 0,};
const Collector = struct { fn receive(listener: *Io.net.Server, io: Io, capture: *Capture) !void { var stream = try listener.accept(io); defer stream.close(io); var read_buffer: [4096]u8 = undefined; var write_buffer: [4096]u8 = undefined; var reader = stream.reader(io, &read_buffer); var writer = stream.writer(io, &write_buffer); var server = std.http.Server.init(&reader.interface, &writer.interface); var request = try server.receiveHead(); var body_buffer: [4096]u8 = undefined; const body_reader = request.readerExpectNone(&body_buffer); var body_writer = std.Io.Writer.fixed(&capture.body); _ = try body_reader.streamRemaining(&body_writer); capture.len = body_writer.buffered().len; try request.respond("", .{ .status = .ok, .keep_alive = false }); }};
fn event(seq_hint: u64) segment.Event { return .{ .seq = 0, .witnessed_at = @intCast(1_700_000_000_000_000 + seq_hint), .indexed_at = 0, .kind = .create, .did = "did:plc:tracecontract", .collection = "app.bsky.feed.post", .rkey = "3l3qo2vuowo2b", .rev = "3l3qo2vutsw2b", .payload = "\xa1\x64\x74\x65\x78\x74\x62\x68\x69", };}
test "real archive boundaries export the upstream span tree offline" { const testing = std.testing; var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var listener = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable; defer listener.deinit(io); var capture: Capture = .{}; var collector = try io.concurrent(Collector.receive, .{ &listener, io, &capture }); defer _ = collector.cancel(io) catch {};
var endpoint_buffer: [128]u8 = undefined; const endpoint = try std.fmt.bufPrint( &endpoint_buffer, "OTEL_EXPORTER_OTLP_ENDPOINT=http://127.0.0.1:{d}", .{listener.socket.address.getPort()}, ); var runtime = try observability.Runtime.init(testing.allocator, .{ .io = io, .environment = &.{ endpoint, "OTEL_BSP_SCHEDULE_DELAY=60000" }, .service_name = "stream-tracing-contract", .service_version = "offline", }); defer runtime.deinit();
var process_batch = observability.Span.start( testing.allocator, "ingest/live", "processBatch", observability.root_context, &.{}, ); { 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();
archive.writer.max_events_per_block = 1; archive.max_events_per_block = 1; archive.max_segment_bytes = 1; try archive.enableAsyncFlush(1); var async_row = [_]segment.Event{event(1)}; _ = try archive.appendBatchTraced(&async_row, 1, process_batch.context());
archive.disableAsyncFlush(); archive.max_segment_bytes = archive_mod.default_max_segment_bytes; var sync_row = [_]segment.Event{event(2)}; _ = try archive.appendBatchTraced(&sync_row, 2, process_batch.context()); } process_batch.succeed(); process_batch.deinit();
try testing.expectEqual(@import("otel-api").common.ProcessResult.success, runtime.shutdown(2_000)); try collector.await(io); const body = capture.body[0..capture.len]; for ([_][]const u8{ "processBatch", "commitAsyncFlush", "rotateIfFull", "rotateLocked", "flushAndRotateLocked", "jetstream/ingest/live", "jetstream/ingest", }) |needle| try testing.expect(std.mem.indexOf(u8, body, needle) != null);}