mirror/browser
1
0
Fork 0
mirror of https://github.com/lightpanda-io/browser.git synced 2026-07-31 17:55:59 -04:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #2814 from lightpanda-io/telemetry_thread

Browse Source 
telemetry: Move telemetry worker to its own thread
This commit is contained in:
Karl Seguin
2026-06-26 08:55:26 +08:00
committed by GitHub
parent 05986de731 13eb35720a
commit b473b019cb
4 changed files with 195 additions and 316 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
src/main.zig
View File

@@ -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;

276
src/network/Network.zig
View File

@@ -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);
}

8
src/network/http.zig
View File

@@ -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));
}

212
src/telemetry/lightpanda.zig
View File

@@ -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 {