diff --git a/src/main.zig b/src/main.zig index 6495406e9..439c3a36b 100644 --- a/src/main.zig +++ b/src/main.zig @@ -184,9 +184,7 @@ fn run(allocator: Allocator, main_arena: Allocator) !void { } var worker_thread = try std.Thread.spawn(.{}, fetchThread, .{ app, &ft, urls, fetch_opts }); - defer worker_thread.join(); - - app.network.run(); + worker_thread.join(); }, .mcp => |opts| { log.info(.mcp, "starting server", .{}); @@ -207,7 +205,12 @@ fn run(allocator: Allocator, main_arena: Allocator) !void { var worker_thread = try std.Thread.spawn(.{}, mcpThread, .{ allocator, app }); defer worker_thread.join(); - app.network.run(); + // mcp talks over stdio on mcpThread. Only run the CDP accept/read + // loop when an optional CDP server was started; otherwise the main + // thread just waits for the worker. + if (cdp_server != null) { + app.network.run(); + } }, .agent => |opts| { log.info(.app, "starting agent", .{}); @@ -230,9 +233,7 @@ fn run(allocator: Allocator, main_arena: Allocator) !void { &cancelled, &sig_bridge, }); - defer worker_thread.join(); - - app.network.run(); + worker_thread.join(); } if (cancelled) return error.UserCancelled; diff --git a/src/network/Network.zig b/src/network/Network.zig index cbc1b454f..dde76146e 100644 --- a/src/network/Network.zig +++ b/src/network/Network.zig @@ -81,8 +81,6 @@ pub const CdpLink = struct { // Number of fixed pollfds entries (wakeup pipe + listener). const PSEUDO_POLLFDS = 2; -const MAX_TICK_CALLBACKS = 16; - allocator: Allocator, app: *App, @@ -110,17 +108,6 @@ wakeup_pipe: [2]posix.fd_t = .{ -1, -1 }, shutdown: std.atomic.Value(bool) = .init(false), -// Multi is a heavy structure that can consume up to 2MB of RAM. -// Currently, Network is used sparingly, and we only create it on demand. -// When Network becomes truly shared, it should become a regular field. -multi: ?*libcurl.CurlM = null, -submission_mutex: std.Thread.Mutex = .{}, -submission_queue: DoublyLinkedList = .{}, - -callbacks: [MAX_TICK_CALLBACKS]TickCallback = undefined, -callbacks_len: usize = 0, -callbacks_mutex: std.Thread.Mutex = .{}, - // Registered CDP read endpoints. Producer-side (the worker doing // register/unregister) and consumer-side (this thread's run loop) are // serialized by cdp_mutex. cdp_unregister signals when a link @@ -149,11 +136,6 @@ cdp_start: usize, /// Optional IP filter for blocking requests to private/internal networks (--block-private-networks). ip_filter: ?*IpFilter = null, -const TickCallback = struct { - ctx: *anyopaque, - fun: *const fn (*anyopaque) void, -}; - fn globalInit(allocator: Allocator) void { // Only route curl's own allocations through our allocator in Debug, so the // leak detector sees them. In Release it'd just wrap c_allocator (curl's @@ -178,19 +160,12 @@ pub fn init(allocator: Allocator, app: *App, config: *const Config) !Network { const pipe = try posix.pipe2(.{ .NONBLOCK = true, .CLOEXEC = true }); - // IMPORTANT: This is a bit messy, and it exists specifically because - // self.multi is optional. self.multi is optional so that, when telemetry is - // disabled, we don't need the overhead of a multi. If self.multi wasn't - // optional, then we wouldn't need to use posix.poll, we could use - // curl_multi_poll. This is to do in a follow up. - - // The structure is: 0 is wakeup, 1 is listener, rest for curl fds: - // [0] wakeup pipe - // [1] listener - // [PSEUDO_POLLFDS .. + httpMaxConcurrent] curl multi fds - // [.. + maxConnections] CDP socket fds + // pollfds layout: + // [0] wakeup pipe + // [1] listener + // [PSEUDO_POLLFDS .. + max_cdp] CDP socket fds const max_cdp = config.maxConnections(); - const pollfds = try allocator.alloc(posix.pollfd, PSEUDO_POLLFDS + config.httpMaxConcurrent() + max_cdp); + const pollfds = try allocator.alloc(posix.pollfd, PSEUDO_POLLFDS + max_cdp); errdefer allocator.free(pollfds); const cdp_poll_snapshot = try allocator.alloc(?*CdpLink, max_cdp); @@ -262,7 +237,7 @@ pub fn init(allocator: Allocator, app: *App, config: *const Config) !Network { .pollfds = pollfds, .wakeup_pipe = pipe, .cdp_poll_snapshot = cdp_poll_snapshot, - .cdp_start = PSEUDO_POLLFDS + config.httpMaxConcurrent(), + .cdp_start = PSEUDO_POLLFDS, .available = available, .connections = connections, @@ -281,10 +256,6 @@ pub fn init(allocator: Allocator, app: *App, config: *const Config) !Network { } pub fn deinit(self: *Network) void { - if (self.multi) |multi| { - libcurl.curl_multi_cleanup(multi) catch {}; - } - for (&self.wakeup_pipe) |*fd| { if (fd.* >= 0) { posix.close(fd.*); @@ -368,30 +339,6 @@ pub fn unbind(self: *Network) void { self.wakeupPoll(); } -pub fn onTick(self: *Network, ctx: *anyopaque, callback: *const fn (*anyopaque) void) void { - self.callbacks_mutex.lock(); - defer self.callbacks_mutex.unlock(); - - lp.assert(self.callbacks_len < MAX_TICK_CALLBACKS, "too many ticks", .{}); - - self.callbacks[self.callbacks_len] = .{ - .ctx = ctx, - .fun = callback, - }; - self.callbacks_len += 1; - - self.wakeupPoll(); -} - -pub fn fireTicks(self: *Network) void { - self.callbacks_mutex.lock(); - defer self.callbacks_mutex.unlock(); - - for (self.callbacks[0..self.callbacks_len]) |*callback| { - callback.fun(callback.ctx); - } -} - // Hand a CDP WebSocket's read side over to the main network thread. The caller // owns the link and must keep it alive until unregisterCdp is called. // The caller must not read from the socket. @@ -609,15 +556,15 @@ fn shutdownCdpLinks(self: *Network) void { pub fn run(self: *Network) void { var drain_buf: [64]u8 = undefined; - var running_handles: c_int = 0; const poll_fd = &self.pollfds[0]; const listen_fd = &self.pollfds[1]; - // Please note that receiving a shutdown command does not terminate all connections. - // When gracefully shutting down a server, we at least want to send the remaining - // telemetry, but we stop accepting new connections. It is the responsibility - // of external code to terminate its requests upon shutdown. + // Receiving a shutdown command does not terminate existing connections: we + // stop accepting new ones but leave in-flight requests to external code to + // terminate. This loop only services the listener and the CDP read sockets; + // page fetches run on per-worker HttpClient multis and telemetry on its own + // thread, so nothing here drives libcurl. while (true) { if (self.listener != null and !self.accept.load(.acquire)) { posix.close(self.listener.?.socket); @@ -625,43 +572,10 @@ pub fn run(self: *Network) void { self.pollfds[1] = .{ .fd = -1, .events = 0, .revents = 0 }; } - self.drainQueue(); - - if (self.multi) |multi| { - // Kickstart newly added handles (DNS/connect) so that - // curl registers its sockets before we poll. - libcurl.curl_multi_perform(multi, &running_handles) catch |err| { - lp.log.err(.app, "curl perform", .{ .err = err }); - }; - - self.preparePollFds(multi); - } - self.prepareCdpPollFds(); - // for ontick to work, you need to wake up periodically - const timeout = blk: { - const min_timeout = 250; // 250ms - if (self.multi == null) { - break :blk min_timeout; - } - - // curl_multi_timeout reports -1 when curl has no timeout - // preference (idle) and 0 when it wants to be serviced - // immediately. Treat both as "no curl-imposed deadline" and - // fall back to min_timeout — otherwise @min(min_timeout, -1) - // would be -1, i.e. poll() blocks forever, starving onTick - // (telemetry's periodic flush) and removing the safety net - // that bounds any missed wakeup to min_timeout. - const curl_timeout = self.getCurlTimeout(); - if (curl_timeout <= 0) { - break :blk min_timeout; - } - - break :blk @min(min_timeout, curl_timeout); - }; - - _ = posix.poll(self.pollfds, timeout) catch |err| { + // wait until we get a CDP message or a signal on the wakeup pipe + _ = posix.poll(self.pollfds, -1) catch |err| { lp.log.err(.app, "poll", .{ .err = err }); continue; }; @@ -679,35 +593,14 @@ pub fn run(self: *Network) void { self.acceptConnections(); } - if (self.multi) |multi| { - // Drive transfers and process completions. - libcurl.curl_multi_perform(multi, &running_handles) catch |err| { - lp.log.err(.app, "curl perform", .{ .err = err }); - }; - self.processCompletions(multi); - } - self.processCdpEvents(); - self.fireTicks(); - if (self.shutdown.load(.acquire)) { - // Drain any live CDP links so their workers can exit (issue #2510). - // Idempotent — no-op once drained, safe to call every iteration + // Drain any live CDP links so their workers can exit (issue #2510), + // then stop. Page fetches and telemetry don't run on this loop, so + // there is nothing else to flush here. self.shutdownCdpLinks(); - - if (running_handles == 0) { - // Check if fireTicks submitted new requests (e.g. telemetry - // flush). If so, continue the loop to drain and send them - // before exiting. - self.submission_mutex.lock(); - const has_pending = self.submission_queue.first != null; - self.submission_mutex.unlock(); - - if (!has_pending) { - break; - } - } + break; } } @@ -725,46 +618,10 @@ pub fn run(self: *Network) void { } } -pub fn submitRequest(self: *Network, conn: *http.Connection) void { - self.submission_mutex.lock(); - self.submission_queue.append(&conn.node); - self.submission_mutex.unlock(); - self.wakeupPoll(); -} - fn wakeupPoll(self: *Network) void { _ = posix.write(self.wakeup_pipe[1], &.{1}) catch {}; } -fn drainQueue(self: *Network) void { - self.submission_mutex.lock(); - defer self.submission_mutex.unlock(); - - if (self.submission_queue.first == null) return; - - const multi = self.multi orelse blk: { - const m = libcurl.curl_multi_init() orelse { - lp.assert(false, "curl multi init failed", .{}); - unreachable; - }; - self.multi = m; - break :blk m; - }; - - while (self.submission_queue.popFirst()) |node| { - const conn: *http.Connection = @fieldParentPtr("node", node); - conn.setPrivate(conn) catch |err| { - lp.log.err(.app, "curl set private", .{ .err = err }); - self.releaseConnection(conn); - continue; - }; - libcurl.curl_multi_add_handle(multi, conn._easy) catch |err| { - lp.log.err(.app, "curl multi add", .{ .err = err }); - self.releaseConnection(conn); - }; - } -} - pub fn stop(self: *Network) void { self.shutdown.store(true, .release); self.wakeupPoll(); @@ -800,68 +657,6 @@ fn acceptConnections(self: *Network) void { } } -fn preparePollFds(self: *Network, multi: *libcurl.CurlM) void { - // Only the curl slice — NOT through to the end of pollfds. The CDP - // socket fds live in [cdp_start..] and are owned by - // prepareCdpPollFds, which only rebuilds them when cdp_dirty is set - // (a steady-state optimization). Slicing to the end here would - // @memset those fds to -1 every iteration once a multi exists (which - // happens as soon as telemetry sends its first request), silently - // dropping every live CDP socket from the poll set — Network then - // never reads another CDP message (#2508) nor observes peer - // EOF/shutdown (#2507). - const curl_fds = self.pollfds[PSEUDO_POLLFDS..self.cdp_start]; - @memset(curl_fds, .{ .fd = -1, .events = 0, .revents = 0 }); - - var fd_count: c_uint = 0; - const wait_fds: []libcurl.CurlWaitFd = @ptrCast(curl_fds); - libcurl.curl_multi_waitfds(multi, wait_fds, &fd_count) catch |err| { - lp.log.err(.app, "curl waitfds", .{ .err = err }); - }; -} - -fn getCurlTimeout(self: *Network) i32 { - const multi = self.multi orelse return -1; - var timeout_ms: c_long = -1; - libcurl.curl_multi_timeout(multi, &timeout_ms) catch return -1; - return @intCast(@min(timeout_ms, std.math.maxInt(i32))); -} - -fn processCompletions(self: *Network, multi: *libcurl.CurlM) void { - var msgs_in_queue: c_int = 0; - while (libcurl.curl_multi_info_read(multi, &msgs_in_queue)) |msg| { - switch (msg.data) { - .done => |maybe_err| { - if (maybe_err) |err| { - lp.log.warn(.app, "curl transfer error", .{ .err = err }); - } - }, - else => continue, - } - - const easy: *libcurl.Curl = msg.easy_handle; - var ptr: *anyopaque = undefined; - libcurl.curl_easy_getinfo(easy, .private, &ptr) catch - lp.assert(false, "curl getinfo private", .{}); - const conn: *http.Connection = @ptrCast(@alignCast(ptr)); - - libcurl.curl_multi_remove_handle(multi, easy) catch {}; - self.releaseConnection(conn); - } -} - -comptime { - if (@sizeOf(posix.pollfd) != @sizeOf(libcurl.CurlWaitFd)) { - @compileError("pollfd and CurlWaitFd size mismatch"); - } - if (@offsetOf(posix.pollfd, "fd") != @offsetOf(libcurl.CurlWaitFd, "fd") or - @offsetOf(posix.pollfd, "events") != @offsetOf(libcurl.CurlWaitFd, "events") or - @offsetOf(posix.pollfd, "revents") != @offsetOf(libcurl.CurlWaitFd, "revents")) - { - @compileError("pollfd and CurlWaitFd layout mismatch"); - } -} - pub fn getConnection(self: *Network) ?*http.Connection { self.conn_mutex.lock(); defer self.conn_mutex.unlock(); @@ -1003,40 +798,3 @@ fn loadCerts(allocator: Allocator) !libcurl.CurlBlob { .flags = 0, }; } - -const testing = @import("../testing.zig"); - -test "Network: preparePollFds leaves the CDP fd region untouched" { - // Regression for #2507 / #2508. Once a multi exists (telemetry creates - // one in optimized builds), preparePollFds runs every loop iteration. - // It rebuilds only the curl slice [PSEUDO_POLLFDS..cdp_start]; the CDP - // region [cdp_start..] is owned by prepareCdpPollFds, which keeps its - // entries across iterations and only rebuilds when cdp_dirty is set. - // A slice that ran to the end of pollfds @memset those CDP sockets to - // -1, silently dropping every live CDP connection from the poll set — - // so Network stopped reading CDP messages (#2508) and never observed - // peer EOF/shutdown (#2507). curl global is initialized by the test - // harness (App.init -> Network.init). - const multi = libcurl.curl_multi_init() orelse return error.FailedToInitMulti; - defer libcurl.curl_multi_cleanup(multi) catch {}; - - const curl_slots = 4; - const cdp_slots = 3; - var pollfds: [PSEUDO_POLLFDS + curl_slots + cdp_slots]posix.pollfd = undefined; - @memset(&pollfds, .{ .fd = -1, .events = 0, .revents = 0 }); - - // preparePollFds only reads self.pollfds and self.cdp_start. - var nw: Network = undefined; - nw.pollfds = &pollfds; - nw.cdp_start = PSEUDO_POLLFDS + curl_slots; - - // Two live CDP sockets parked in the CDP region, mimicking the steady - // state between cdp_dirty rebuilds. - pollfds[nw.cdp_start] = .{ .fd = 4242, .events = posix.POLL.IN, .revents = 0 }; - pollfds[nw.cdp_start + 1] = .{ .fd = 4243, .events = posix.POLL.IN, .revents = 0 }; - - nw.preparePollFds(multi); - - try testing.expectEqual(@as(posix.fd_t, 4242), pollfds[nw.cdp_start].fd); - try testing.expectEqual(@as(posix.fd_t, 4243), pollfds[nw.cdp_start + 1].fd); -} diff --git a/src/network/http.zig b/src/network/http.zig index 718d57b56..481a991a6 100644 --- a/src/network/http.zig +++ b/src/network/http.zig @@ -668,6 +668,14 @@ pub const Connection = struct { return self.getResponseCode(); } + // Synchronous transfer that adds no request headers. request() injects the + // browser User-Agent / sec-ch-ua machinery meant for page fetches; callers + // that manage their own connection (telemetry) use this leaner path. + pub fn perform(self: *const Connection) !u16 { + try libcurl.curl_easy_perform(self._easy); + return self.getResponseCode(); + } + pub fn wsStartFrame(self: *const Connection, frame_type: libcurl.WsFrameType, size: usize) !void { try libcurl.curl_ws_start_frame(self._easy, frame_type, @intCast(size)); } diff --git a/src/telemetry/lightpanda.zig b/src/telemetry/lightpanda.zig index 7662d27ac..2ab1eb08f 100644 --- a/src/telemetry/lightpanda.zig +++ b/src/telemetry/lightpanda.zig @@ -5,33 +5,55 @@ const build_config = @import("build_config"); const App = @import("../App.zig"); const Config = @import("../Config.zig"); -const telemetry = @import("telemetry.zig"); + +const http = @import("../network/http.zig"); const Network = @import("../network/Network.zig"); +const telemetry = @import("telemetry.zig"); + const log = lp.log; const Allocator = std.mem.Allocator; +const IS_DEBUG = builtin.mode == .Debug; -const BUFFER_SIZE = 1024; -const MAX_BODY_SIZE = 500 * 1024; // 500KB server limit +const MAX_PENDING = 4096; // hard cap: drop + count beyond this (reported via buffer_overflow) +const RECLAIM_CAPACITY = 64; // reclaim the drain buffer once a burst grows it past this +const MAX_BODY_SIZE = 500 * 1024; // 500KB +const REQUEST_TIMEOUT_MS = 5000; const URL = "https://telemetry.lightpanda.io"; const LightPanda = @This(); allocator: Allocator, network: *Network, -writer: std.Io.Writer.Allocating, -/// Protects concurrent producers in send(). -mutex: std.Thread.Mutex = .{}, +// Owned and used only by the sender thread; never touched by producers. +writer: std.Io.Writer.Allocating, iid: ?[36]u8 = null, run_mode: Config.RunMode = .serve, interactive: bool = false, -head: std.atomic.Value(usize) = .init(0), -tail: std.atomic.Value(usize) = .init(0), -dropped: std.atomic.Value(u32) = .init(0), -buffer: [BUFFER_SIZE]telemetry.Event = undefined, +// `mutex` guards the ring buffer (head/tail/dropped), `running`, and the lazy +// `thread` creation. The sender thread blocks on `cond` while idle. +mutex: std.Thread.Mutex = .{}, +cond: std.Thread.Condition = .{}, + +// The sender thread is spawned on the first event, so a process that emits no +// telemetry pays for none of this +thread: ?std.Thread = null, +running: bool = true, + +// Pending events. Producers append under `mutex`; the sender thread swaps the +// whole list out in O(1) and drains it. The list grows from empty under load +// and is reclaimed afterward, so memory tracks actual queue depth and never +// pre-commits a fixed count × sizeOf(Event). Events must be self-contained +// (inline storage, no borrowed slices) since they outlive the send() call on +// the sender thread. Past MAX_PENDING (or on alloc failure) we drop and bump +// `dropped`, which is reported as a buffer_overflow event on the next +// successful POST and only cleared then — our signal that the cap is too low or +// the endpoint is failing. +pending: std.ArrayList(telemetry.Event) = .empty, +dropped: u32 = 0, pub fn init(self: *LightPanda, app: *App, iid: ?[36]u8, run_mode: Config.RunMode, interactive: bool) !void { self.* = .{ @@ -42,11 +64,19 @@ pub fn init(self: *LightPanda, app: *App, iid: ?[36]u8, run_mode: Config.RunMode .network = &app.network, .writer = std.Io.Writer.Allocating.init(app.allocator), }; - - self.network.onTick(@ptrCast(self), flushCallback); } pub fn deinit(self: *LightPanda) void { + if (self.thread) |thread| { + self.mutex.lock(); + self.running = false; + self.mutex.unlock(); + self.cond.signal(); + // The thread drains anything still queued before returning, so this is + // also our graceful flush-on-shutdown. Bounded by REQUEST_TIMEOUT_MS. + thread.join(); + } + self.pending.deinit(self.allocator); self.writer.deinit(); } @@ -54,60 +84,142 @@ pub fn send(self: *LightPanda, raw_event: telemetry.Event) !void { self.mutex.lock(); defer self.mutex.unlock(); - const t = self.tail.load(.monotonic); - const h = self.head.load(.acquire); - if (t - h >= BUFFER_SIZE) { - _ = self.dropped.fetchAdd(1, .monotonic); + if (self.thread == null) { + // First event: bring the sender thread online. If it can't spawn, drop + // the event rather than blocking the producer. + self.thread = std.Thread.spawn(.{}, run, .{self}) catch |err| { + log.warn(.telemetry, "thread spawn", .{ .err = err }); + return; + }; + } + + if (self.pending.items.len >= MAX_PENDING) { + self.dropped +|= 1; return; } - self.buffer[t % BUFFER_SIZE] = raw_event; - self.tail.store(t + 1, .release); -} - -fn flushCallback(ctx: *anyopaque) void { - const self: *LightPanda = @ptrCast(@alignCast(ctx)); - self.postEvent() catch |err| { - log.warn(.telemetry, "flush error", .{ .err = err }); - }; -} - -fn postEvent(self: *LightPanda) !void { - const conn = self.network.getConnection() orelse { + self.pending.append(self.allocator, raw_event) catch { + // Allocation failure growing the queue: drop and count, same as a cap hit. + self.dropped +|= 1; return; }; - errdefer self.network.releaseConnection(conn); + self.cond.signal(); +} - const h = self.head.load(.monotonic); - const t = self.tail.load(.acquire); - const dropped = self.dropped.swap(0, .monotonic); - - if (h == t and dropped == 0) { - self.network.releaseConnection(conn); +fn run(self: *LightPanda) void { + // The connection is created, owned, and torn down entirely on this thread; + // the network thread never sees it (Transport == .none). + var conn = http.Connection.init(self.network.ca_blob, self.network.config, self.network.ip_filter) catch |err| { + // Essentially OOM — the process is already in trouble. The thread + // handle stays set so send() won't respawn; events drop at the cap. + log.warn(.telemetry, "connection init", .{ .err = err }); return; + }; + defer conn.deinit(); + + // Static for the lifetime of the connection. + conn.setURL(URL) catch |err| return log.warn(.telemetry, "set url", .{ .err = err }); + conn.setMethod(.POST) catch |err| return log.warn(.telemetry, "set method", .{ .err = err }); + conn.setTimeout(REQUEST_TIMEOUT_MS) catch |err| return log.warn(.telemetry, "set timeout", .{ .err = err }); + + // Consumer-owned buffer, swapped with `pending` each cycle: producers always + // get an empty list back and we own the batch lock-free across the POST. + var batch: std.ArrayList(telemetry.Event) = .empty; + defer batch.deinit(self.allocator); + + self.mutex.lock(); + defer self.mutex.unlock(); + while (true) { + while (self.pending.items.len == 0) { + if (self.running == false) { + return; + } + self.cond.wait(&self.mutex); + } + + // Swap the batch out and release the lock for the (blocking) serialize + + // POST so producers never wait on the network. + std.mem.swap(std.ArrayList(telemetry.Event), &self.pending, &batch); + const dropped = self.dropped; + self.dropped = 0; + self.mutex.unlock(); + + var sent: usize = 0; + self.postEvents(&conn, batch.items, dropped, &sent) catch |err| { + log.warn(.telemetry, "postEvents", .{ .err = err, .events = batch.items.len, .dropped = dropped }); + }; + const lost: u32 = @intCast(batch.items.len - sent); + + // batch is owned by this thread and can be mutated without the lock + if (batch.capacity > RECLAIM_CAPACITY) { + batch.clearAndFree(self.allocator); + } else { + batch.clearRetainingCapacity(); + } + + self.mutex.lock(); + // Re-fold whatever didn't reach the wire back into `dropped`, so the next + // successful POST still reports it — our only failure signal. + self.dropped +|= lost; + if (sent == 0) { + // if nothing was sent, than we didn't get to send the buffer_overflow + // message, so whatever was dropped is still dropped. + self.dropped +|= dropped; + } } - errdefer _ = self.dropped.fetchAdd(dropped, .monotonic); +} - self.writer.clearRetainingCapacity(); - - if (dropped > 0) { +fn postEvents(self: *LightPanda, conn: *http.Connection, events: []const telemetry.Event, dropped: u32, sent: *usize) !void { + // The overflow report rides ahead of the first body; the rest of that body + // and any subsequent ones are filled from `events` below. + const has_overflow = dropped > 0; + if (has_overflow) { _ = try self.writeEvent(.{ .buffer_overflow = .{ .dropped = dropped } }); } - var sent: usize = 0; - for (h..t) |i| { - const fit = try self.writeEvent(self.buffer[i % BUFFER_SIZE]); - if (!fit) break; + var queued: usize = 0; + for (events) |event| { + if (try self.writeEvent(event) == false) { + try self.flush(conn); + sent.* += queued; + queued = 0; - sent += 1; + // now re-write the message that didn't fit + const fit = try self.writeEvent(event); + if (comptime IS_DEBUG) { + std.debug.assert(fit); + } + if (fit == false) { + // we have a single event that can never be serialized. This + // should never happen, but I'm not willing to crash on that + // belief. + continue; + } + } + queued += 1; } - try conn.setURL(URL); - try conn.setMethod(.POST); - try conn.setBody(self.writer.written()); + if (self.writer.written().len != 0) { + try self.flush(conn); + sent.* += queued; + } +} - self.head.store(h + sent, .release); - self.network.submitRequest(conn); +fn flush(self: *LightPanda, conn: *http.Connection) !void { + self._flush(conn) catch |err| { + log.warn(.telemetry, "flush", .{ .err = err, .size = self.writer.written().len }); + return err; + }; +} + +fn _flush(self: *LightPanda, conn: *http.Connection) !void { + defer self.writer.clearRetainingCapacity(); + + try conn.setBody(self.writer.written()); + const status = try conn.perform(); + if (status != 200) { + return error.ServerError; + } } fn writeEvent(self: *LightPanda, event: telemetry.Event) !bool {