feat(log): add AsyncAppend, JsonLayout, and LogfmtLayout

grapebaba · Copilot · GrapeBaBa · commit a68c3547c356 · 2026-04-10T18:43:02.000+08:00
Phase 2 components:
- AsyncAppend: Vyukov bounded MPMC ring buffer with background consumer
  thread, non-blocking enqueue, drop-on-full policy, graceful shutdown+drain
- JsonLayout: JSON lines format with proper string escaping
- LogfmtLayout: key=value format with conditional quoting

Co-authored-by: Copilot &lt;223556219+Copilot@users.noreply.github.com&gt;
diff --git a/src/log/AsyncAppend.zig b/src/log/AsyncAppend.zig
@@ -0,0 +1,234 @@
+const std = @import("std");
+
+/// Maximum bytes per log message slot.
+const max_message_size = 4096;
+
+/// A single slot in the ring buffer.
+const Slot = struct {
+    sequence: std.atomic.Value(usize),
+    len: usize,
+    data: [max_message_size]u8,
+};
+
+/// Vyukov bounded MPMC ring buffer for async log message passing.
+/// `buffer_size_log2` is the log2 of the number of slots (e.g., 10 = 1024 slots).
+pub fn AsyncAppend(comptime buffer_size_log2: comptime_int) type {
+    const buffer_size: usize = 1 << buffer_size_log2;
+    const mask: usize = buffer_size - 1;
+
+    return struct {
+        const Self = @This();
+
+        slots: [buffer_size]Slot,
+        head: std.atomic.Value(usize),
+        tail: std.atomic.Value(usize),
+        shutdown: std.atomic.Value(bool),
+        thread: ?std.Thread,
+        inner_write_fn: *const fn (*anyopaque, []const u8) void,
+        inner_ptr: *anyopaque,
+
+        /// Initialize AsyncAppend wrapping an inner Append.
+        /// `inner` must be a pointer to a struct with `pub fn write(self: *T, bytes: []const u8) void`.
+        pub fn init(inner: anytype) Self {
+            const T = @TypeOf(inner.*);
+            const wrapper = struct {
+                fn writeFn(ctx: *anyopaque, bytes: []const u8) void {
+                    const typed: *T = @ptrCast(@alignCast(ctx));
+                    typed.write(bytes);
+                }
+            };
+
+            var self = Self{
+                .slots = undefined,
+                .head = std.atomic.Value(usize).init(0),
+                .tail = std.atomic.Value(usize).init(0),
+                .shutdown = std.atomic.Value(bool).init(false),
+                .thread = null,
+                .inner_write_fn = wrapper.writeFn,
+                .inner_ptr = @ptrCast(inner),
+            };
+
+            for (0..buffer_size) |i| {
+                self.slots[i].sequence = std.atomic.Value(usize).init(i);
+                self.slots[i].len = 0;
+            }
+
+            return self;
+        }
+
+        /// Start the background consumer thread.
+        pub fn start(self: *Self) void {
+            self.thread = std.Thread.spawn(.{}, consumerLoop, .{self}) catch return;
+        }
+
+        /// Append interface: non-blocking enqueue. Drops message if ring is full.
+        pub fn write(self: *Self, bytes: []const u8) void {
+            var head = self.head.load(.acquire);
+            while (true) {
+                const slot = &self.slots[head & mask];
+                const seq = slot.sequence.load(.acquire);
+
+                const diff = @as(isize, @intCast(seq)) - @as(isize, @intCast(head));
+                if (diff == 0) {
+                    // Slot is available — try to claim it.
+                    if (self.head.cmpxchgWeak(head, head + 1, .acq_rel, .acquire)) |updated| {
+                        head = updated;
+                        continue;
+                    }
+                    // Claimed — write payload.
+                    const copy_len = @min(bytes.len, max_message_size);
+                    @memcpy(slot.data[0..copy_len], bytes[0..copy_len]);
+                    slot.len = copy_len;
+                    // Publish.
+                    slot.sequence.store(head + 1, .release);
+                    return;
+                } else if (diff < 0) {
+                    // Ring is full — drop message.
+                    return;
+                } else {
+                    // Another producer is ahead — reload head.
+                    head = self.head.load(.acquire);
+                }
+            }
+        }
+
+        /// Shutdown: signal, join thread, drain remaining.
+        pub fn deinit(self: *Self) void {
+            self.shutdown.store(true, .release);
+            if (self.thread) |t| {
+                t.join();
+                self.thread = null;
+            }
+            self.drain();
+        }
+
+        fn consumerLoop(self: *Self) void {
+            var idle_spins: u32 = 0;
+            while (!self.shutdown.load(.acquire)) {
+                if (self.tryConsume()) {
+                    idle_spins = 0;
+                } else {
+                    idle_spins += 1;
+                    if (idle_spins > 64) {
+                        std.time.sleep(1_000_000); // 1ms
+                    } else {
+                        std.atomic.spinLoopHint();
+                    }
+                }
+            }
+        }
+
+        fn tryConsume(self: *Self) bool {
+            const tail = self.tail.load(.acquire);
+            const slot = &self.slots[tail & mask];
+            const seq = slot.sequence.load(.acquire);
+
+            const diff = @as(isize, @intCast(seq)) - @as(isize, @intCast(tail + 1));
+            if (diff == 0) {
+                // Slot has data — claim it.
+                if (self.tail.cmpxchgWeak(tail, tail + 1, .acq_rel, .acquire)) |_| {
+                    return false;
+                }
+                self.inner_write_fn(self.inner_ptr, slot.data[0..slot.len]);
+                // Recycle slot.
+                slot.sequence.store(tail + buffer_size, .release);
+                return true;
+            }
+            return false;
+        }
+
+        fn drain(self: *Self) void {
+            while (self.tryConsume()) {}
+        }
+    };
+}
+
+// ==== Tests ====
+
+const TestCollector = struct {
+    count: std.atomic.Value(usize),
+    last_msg: [max_message_size]u8,
+    last_len: usize,
+
+    fn init() TestCollector {
+        return .{
+            .count = std.atomic.Value(usize).init(0),
+            .last_msg = undefined,
+            .last_len = 0,
+        };
+    }
+
+    pub fn write(self: *TestCollector, bytes: []const u8) void {
+        const copy_len = @min(bytes.len, max_message_size);
+        @memcpy(self.last_msg[0..copy_len], bytes[0..copy_len]);
+        self.last_len = copy_len;
+        _ = self.count.fetchAdd(1, .monotonic);
+    }
+};
+
+test "AsyncAppend basic enqueue and drain" {
+    var collector = TestCollector.init();
+    var appender = AsyncAppend(4).init(&collector); // 16 slots
+
+    appender.start();
+
+    appender.write("hello async");
+    appender.write("second message");
+
+    // Give consumer time to process.
+    std.time.sleep(50_000_000); // 50ms
+
+    appender.deinit();
+
+    try std.testing.expect(collector.count.load(.acquire) >= 2);
+}
+
+test "AsyncAppend shutdown drains remaining" {
+    var collector = TestCollector.init();
+    var appender = AsyncAppend(4).init(&collector);
+
+    // Don't start consumer thread — messages queue up.
+    appender.write("msg1");
+    appender.write("msg2");
+    appender.write("msg3");
+
+    // deinit should drain.
+    appender.deinit();
+
+    try std.testing.expectEqual(@as(usize, 3), collector.count.load(.acquire));
+}
+
+test "AsyncAppend drops when full" {
+    var collector = TestCollector.init();
+    var appender = AsyncAppend(2).init(&collector); // 4 slots only
+
+    // Fill beyond capacity without consumer.
+    for (0..10) |i| {
+        var buf: [32]u8 = undefined;
+        const msg = std.fmt.bufPrint(&buf, "msg{d}", .{i}) catch "?";
+        appender.write(msg);
+    }
+
+    // Drain what's in the buffer.
+    appender.deinit();
+
+    // Should have received at most 4 (buffer size).
+    const count = collector.count.load(.acquire);
+    try std.testing.expect(count <= 4);
+    try std.testing.expect(count > 0);
+}
+
+test "AsyncAppend truncates long messages" {
+    var collector = TestCollector.init();
+    var appender = AsyncAppend(4).init(&collector);
+
+    // Create a message longer than max_message_size.
+    var long_msg: [max_message_size + 100]u8 = undefined;
+    @memset(&long_msg, 'A');
+
+    appender.write(&long_msg);
+    appender.deinit();
+
+    try std.testing.expectEqual(@as(usize, 1), collector.count.load(.acquire));
+    try std.testing.expectEqual(@as(usize, max_message_size), collector.last_len);
+}
diff --git a/src/log/JsonLayout.zig b/src/log/JsonLayout.zig
@@ -0,0 +1,161 @@
+const std = @import("std");
+const Record = @import("Record.zig").Record;
+const Level = @import("Level.zig");
+const Attr = @import("Attr.zig").Attr;
+
+/// JSON lines layout. Each record is a single JSON object followed by \n.
+/// Format: {"ts":1234,"level":"info","scope":"fork_choice","msg":"block applied","slot":42}
+pub const JsonLayout = struct {
+    pub fn format(_: *const JsonLayout, record: *const Record, writer: anytype) void {
+        writer.writeAll("{") catch return;
+
+        // Timestamp
+        writer.print("\"ts\":{d}", .{record.timestamp_ms}) catch return;
+
+        // Level
+        writer.writeAll(",\"level\":\"") catch return;
+        writer.writeAll(std.mem.trimRight(u8, Level.asText(record.level), " ")) catch return;
+        writer.writeAll("\"") catch return;
+
+        // Scope
+        if (!std.mem.eql(u8, record.scope_name, "default")) {
+            writer.writeAll(",\"scope\":\"") catch return;
+            writer.writeAll(record.scope_name) catch return;
+            writer.writeAll("\"") catch return;
+        }
+
+        // Message
+        writer.writeAll(",\"msg\":\"") catch return;
+        writeJsonEscaped(writer, record.message);
+        writer.writeAll("\"") catch return;
+
+        // Attributes
+        var iter = record.attrIterator();
+        while (iter.next()) |attr| {
+            writer.writeAll(",\"") catch return;
+            writer.writeAll(attr.key) catch return;
+            writer.writeAll("\":") catch return;
+            writeJsonValue(writer, attr.value);
+        }
+
+        writer.writeAll("}\n") catch return;
+    }
+
+    fn writeJsonValue(writer: anytype, value: Attr.Value) void {
+        switch (value) {
+            .int => |v| writer.print("{d}", .{v}) catch {},
+            .uint => |v| writer.print("{d}", .{v}) catch {},
+            .float => |v| writer.print("{d:.6}", .{v}) catch {},
+            .bool_val => |v| writer.print("{}", .{v}) catch {},
+            .string => |v| {
+                writer.writeByte('"') catch {};
+                writeJsonEscaped(writer, v);
+                writer.writeByte('"') catch {};
+            },
+            .hex_bytes => |v| {
+                writer.writeAll("\"0x") catch {};
+                for (v) |byte| {
+                    writer.print("{x:0>2}", .{byte}) catch {};
+                }
+                writer.writeByte('"') catch {};
+            },
+        }
+    }
+
+    fn writeJsonEscaped(writer: anytype, s: []const u8) void {
+        for (s) |c| {
+            switch (c) {
+                '"' => writer.writeAll("\\\"") catch {},
+                '\\' => writer.writeAll("\\\\") catch {},
+                '\n' => writer.writeAll("\\n") catch {},
+                '\r' => writer.writeAll("\\r") catch {},
+                '\t' => writer.writeAll("\\t") catch {},
+                else => writer.writeByte(c) catch {},
+            }
+        }
+    }
+};
+
+test "JsonLayout basic" {
+    var buf: [512]u8 = undefined;
+    var fbs = std.io.fixedBufferStream(&buf);
+
+    var record = Record{
+        .timestamp_ms = 1234567890,
+        .level = .info,
+        .scope_name = "fork_choice",
+        .message = "block applied",
+    };
+    record.pushEventAttr(Attr.uint("slot", 42));
+
+    const layout = JsonLayout{};
+    layout.format(&record, fbs.writer());
+
+    const output = fbs.getWritten();
+    // Should be valid-ish JSON with expected fields
+    try std.testing.expect(std.mem.indexOf(u8, output, "\"ts\":1234567890") != null);
+    try std.testing.expect(std.mem.indexOf(u8, output, "\"level\":\"info\"") != null);
+    try std.testing.expect(std.mem.indexOf(u8, output, "\"scope\":\"fork_choice\"") != null);
+    try std.testing.expect(std.mem.indexOf(u8, output, "\"msg\":\"block applied\"") != null);
+    try std.testing.expect(std.mem.indexOf(u8, output, "\"slot\":42") != null);
+    try std.testing.expect(output[output.len - 1] == '\n');
+}
+
+test "JsonLayout default scope omitted" {
+    var buf: [512]u8 = undefined;
+    var fbs = std.io.fixedBufferStream(&buf);
+
+    const record = Record{
+        .timestamp_ms = 0,
+        .level = .err,
+        .scope_name = "default",
+        .message = "oops",
+    };
+
+    const layout = JsonLayout{};
+    layout.format(&record, fbs.writer());
+
+    const output = fbs.getWritten();
+    try std.testing.expect(std.mem.indexOf(u8, output, "\"scope\"") == null);
+}
+
+test "JsonLayout escapes special characters" {
+    var buf: [512]u8 = undefined;
+    var fbs = std.io.fixedBufferStream(&buf);
+
+    var record = Record{
+        .timestamp_ms = 0,
+        .level = .warn,
+        .scope_name = "default",
+        .message = "line1\nline2",
+    };
+    record.pushEventAttr(Attr.str("data", "has \"quotes\""));
+
+    const layout = JsonLayout{};
+    layout.format(&record, fbs.writer());
+
+    const output = fbs.getWritten();
+    try std.testing.expect(std.mem.indexOf(u8, output, "line1\\nline2") != null);
+    try std.testing.expect(std.mem.indexOf(u8, output, "has \\\"quotes\\\"") != null);
+}
+
+test "JsonLayout bool and hex values" {
+    var buf: [512]u8 = undefined;
+    var fbs = std.io.fixedBufferStream(&buf);
+
+    var record = Record{
+        .timestamp_ms = 0,
+        .level = .info,
+        .scope_name = "default",
+        .message = "test",
+    };
+    record.pushEventAttr(Attr.boolean("ok", true));
+    record.pushEventAttr(Attr.hex("hash", &[_]u8{ 0xab, 0xcd }));
+
+    const layout = JsonLayout{};
+    layout.format(&record, fbs.writer());
+
+    const output = fbs.getWritten();
+    try std.testing.expect(std.mem.indexOf(u8, output, "\"ok\":true") != null);
+    try std.testing.expect(std.mem.indexOf(u8, output, "\"hash\":\"0xabcd\"") != null);
+}
diff --git a/src/log/LogfmtLayout.zig b/src/log/LogfmtLayout.zig
