fix(prometheus): flaky metrics server

build.zig

@@ -89,6 +89,9 @@ pub fn build(b: *Build) !void {
     const zig_cli_dep = b.dependency("zig-cli", dep_opts);
     const zig_cli_mod = zig_cli_dep.module("zig-cli");
 
+    const httpz_dep = b.dependency("httpz", dep_opts);
+    const httpz_mod = httpz_dep.module("httpz");
+
     const zstd_dep = b.dependency("zstd", dep_opts);
     const zstd_mod = zstd_dep.module("zstd");
 
@@ -124,6 +127,7 @@ pub fn build(b: *Build) !void {
     sig_mod.addImport("base58", base58_mod);
     sig_mod.addImport("zig-cli", zig_cli_mod);
     sig_mod.addImport("secp256k1", secp256k1_mod);
+    sig_mod.addImport("httpz", httpz_mod);
     sig_mod.addImport("zstd", zstd_mod);
 
     sig_mod.addImport("poseidon", poseidon_mod);
@@ -154,6 +158,7 @@ pub fn build(b: *Build) !void {
 
     sig_exe.root_module.addImport("xev", xev_mod);
     sig_exe.root_module.addImport("base58", base58_mod);
+    sig_exe.root_module.addImport("httpz", httpz_mod);
     sig_exe.root_module.addImport("zig-cli", zig_cli_mod);
     sig_exe.root_module.addImport("zig-network", zig_network_mod);
     sig_exe.root_module.addImport("zstd", zstd_mod);
@@ -184,6 +189,7 @@ pub fn build(b: *Build) !void {
 
     unit_tests_exe.root_module.addImport("xev", xev_mod);
     unit_tests_exe.root_module.addImport("base58", base58_mod);
+    unit_tests_exe.root_module.addImport("httpz", httpz_mod);
     unit_tests_exe.root_module.addImport("zig-network", zig_network_mod);
     unit_tests_exe.root_module.addImport("zstd", zstd_mod);
     unit_tests_exe.root_module.addImport("poseidon", poseidon_mod);
@@ -213,6 +219,7 @@ pub fn build(b: *Build) !void {
     fuzz_exe.root_module.addImport("base58", base58_mod);
     fuzz_exe.root_module.addImport("secp256k1", secp256k1_mod);
     fuzz_exe.root_module.addImport("zig-network", zig_network_mod);
+    fuzz_exe.root_module.addImport("httpz", httpz_mod);
     fuzz_exe.root_module.addImport("zstd", zstd_mod);
     switch (config.blockstore_db) {
         .rocksdb => fuzz_exe.root_module.addImport("rocksdb", rocksdb_mod),
@@ -244,6 +251,7 @@ pub fn build(b: *Build) !void {
     benchmark_exe.root_module.addImport("secp256k1", secp256k1_mod);
     benchmark_exe.root_module.addImport("base58", base58_mod);
     benchmark_exe.root_module.addImport("zig-network", zig_network_mod);
+    benchmark_exe.root_module.addImport("httpz", httpz_mod);
     benchmark_exe.root_module.addImport("zstd", zstd_mod);
     benchmark_exe.root_module.addImport("prettytable", pretty_table_mod);
     switch (config.blockstore_db) {

build.zig.zon

@@ -15,6 +15,10 @@
             .url = "https://github.com/sam701/zig-cli/archive/8c7a798c0f7fa0358d7ab41106fc872fca4cd995.tar.gz",
             .hash = "1220c008492d9460c3be2b209600a948181e6efb3bf0d79a1633def499632e708f4b",
         },
+        .httpz = .{
+            .url = "https://github.com/karlseguin/http.zig/archive/79dad0f0cc652830cd8e49bf3e73aa77155ad4b2.tar.gz",
+            .hash = "1220b8a918dfcee4fc8326ec337776e2ffd3029511c35f6b96d10aa7be98ca2faf99",
+        },
         .zstd = .{
             .url = "git+https://github.com/Syndica/zstd.zig#5095f011c1183aa67d696172795440d6a33732c9",
             .hash = "122030ebe280b73693963a67ed656226a67b7f00a0a05665155da00c9fcdee90de88",

src/prometheus/http.zig

@@ -1,6 +1,5 @@
-const builtin = @import("builtin");
 const std = @import("std");
-const sig = @import("../sig.zig");
+const httpz = @import("httpz");
 
 const Registry = @import("registry.zig").Registry;
 const globalRegistry = @import("registry.zig").globalRegistry;
@@ -11,58 +10,39 @@
     registry: *Registry(.{}),
     port: u16,
 ) !void {
-    const our_ip = std.net.Address.initIp4(.{ 0, 0, 0, 0 }, port);
-    var tcp = try our_ip.listen(.{
-        .force_nonblocking = true,
-        .reuse_address = true,
-    });
-    defer tcp.deinit();
-
-    while (true) {
-        const conn = tcp.accept() catch |err| switch (err) {
-            error.WouldBlock => continue,
-            else => |e| return e,
-        };
-
-        // TODO: unify this with the code for the RPC server
-        if (comptime builtin.target.isDarwin()) set_flags: {
-            const FlagsInt = @typeInfo(std.posix.O).Struct.backing_integer.?;
-            var flags_int: FlagsInt =
-                @intCast(try std.posix.fcntl(conn.stream.handle, std.posix.F.GETFL, 0));
-            const flags: *std.posix.O =
-                std.mem.bytesAsValue(std.posix.O, std.mem.asBytes(&flags_int));
-            if (flags.NONBLOCK == false and flags.CLOEXEC == true) break :set_flags;
-            flags.NONBLOCK = false;
-            flags.CLOEXEC = true;
-            _ = try std.posix.fcntl(conn.stream.handle, std.posix.F.SETFL, flags_int);
-        }
+    const endpoint = MetricsEndpoint{
+        .allocator = allocator,
+        .registry = registry,
+    };
+    var server = try httpz.ServerCtx(*const MetricsEndpoint, *const MetricsEndpoint).init(
+        allocator,
+        .{ .port = port, .address = "0.0.0.0" },
+        &endpoint,
+    );
+    var router = server.router();
+    router.get("/metrics", getMetrics);
+    return server.listen();
+}
 
-        var read_buffer: [4096]u8 = undefined;
-        var http_server = std.http.Server.init(conn, &read_buffer);
-        var request = http_server.receiveHead() catch continue;
+const MetricsEndpoint = struct {
+    allocator: std.mem.Allocator,
+    registry: *Registry(.{}),
+};
 
-        if (request.head.method != .GET or
-            !std.mem.eql(u8, request.head.target, "/metrics") //
-        ) {
-            try request.respond("", .{
-                .status = .not_found,
-                .keep_alive = false,
-            });
-            continue;
-        }
+/// Initializes the global registry. Returns error if registry was already initialized.
+/// Spawns a thread to serve the metrics over http on the given port.
+pub fn spawnMetrics(gpa_allocator: std.mem.Allocator, port: u16) !std.Thread {
+    const registry = globalRegistry();
+    return std.Thread.spawn(.{}, servePrometheus, .{ gpa_allocator, registry, port });
+}
 
-        var send_buffer: [4096]u8 = undefined;
-        var response = request.respondStreaming(.{
-            .send_buffer = &send_buffer,
-            .respond_options = .{
-                .status = .ok,
-                .keep_alive = true,
-                .extra_headers = &.{.{ .name = "Content-Type", .value = "text/plain" }},
-            },
-        });
-        try registry.write(allocator, response.writer());
-        try response.end();
-    }
+pub fn getMetrics(
+    self: *const MetricsEndpoint,
+    _: *httpz.Request,
+    response: *httpz.Response,
+) !void {
+    response.content_type = .TEXT; // expected by prometheus
+    try self.registry.write(self.allocator, response.writer());
 }
 
 /// Runs a test prometheus endpoint with dummy data.