log: extract RingBuffer from AsyncAppend

Separate the MPMC ring buffer into its own module (RingBuffer.zig):
- Bounded sequence-per-slot queue with push/tryPop/Message.release API
- Cache-line aligned cursors, epoch-based condvar parking
- Independent tests (push/pop, drops-when-full, truncation, FIFO, close, peek)

AsyncAppend is now a thin wrapper (~80 lines):
- Owns a RingBuffer + consumer thread + inner appender forwarding
- No ring buffer logic of its own

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: grapebaba

src/log/AsyncAppend.zig

@@ -1,68 +1,22 @@
 const std = @import("std");
+const ring_buffer = @import("RingBuffer.zig");
+const RingBuffer = ring_buffer.RingBuffer;
 
-/// L1 cache line size. 128 covers x86-64 (64B) and Apple Silicon (128B).
-const cache_line = 128;
+pub const max_message_size = ring_buffer.max_message_size;
 
-/// Maximum bytes per log message slot.
-pub const max_message_size = 4096;
-
-/// A single slot in the ring buffer.
-/// Sequence field is cache-line aligned to prevent false sharing between
-/// adjacent slots (the previous slot's data tail and the next slot's sequence
-/// would otherwise share a cache line).
-const Slot = struct {
-    sequence: std.atomic.Value(usize) align(cache_line),
-    len: usize,
-    data: [max_message_size]u8,
-};
-
-/// Condition-variable waiter with fast-path check.
-/// Only signals when threads are actually parked, avoiding unnecessary syscalls.
-const Waiter = struct {
-    mutex: std.Thread.Mutex = .{},
-    cond: std.Thread.Condition = .{},
-    parked: std.atomic.Value(u32) align(cache_line) = std.atomic.Value(u32).init(0),
-
-    fn notifyOne(self: *Waiter) void {
-        if (self.parked.load(.monotonic) != 0) {
-            self.cond.signal();
-        }
-    }
-
-    fn broadcast(self: *Waiter) void {
-        if (self.parked.load(.monotonic) != 0) {
-            self.cond.broadcast();
-        }
-    }
-};
-
-/// Bounded MPMC ring buffer for async log message passing.
-///
-/// Sequence-per-slot design (LMAX Disruptor / Vyukov style):
-///   - Pre-allocated slots via allocator (zero allocation on hot path)
-///   - Cache-line aligned cursors prevent false sharing
-///   - Epoch-based condvar parking for low-latency wakeup
-///   - Non-blocking producer with drop-on-full policy
-///   - Wrapping arithmetic for correctness over long uptimes
+/// Async appender: non-blocking write via a bounded MPMC ring buffer,
+/// drained by a background consumer thread that forwards to an inner appender.
 pub const AsyncAppend = struct {
-    slots: []Slot,
-    mask: usize,
-    head: std.atomic.Value(usize) align(cache_line) = std.atomic.Value(usize).init(0),
-    tail: std.atomic.Value(usize) align(cache_line) = std.atomic.Value(usize).init(0),
-    epoch: std.atomic.Value(usize) align(cache_line) = std.atomic.Value(usize).init(0),
-    shutdown: std.atomic.Value(bool) align(cache_line) = std.atomic.Value(bool).init(false),
-    waiter: Waiter align(cache_line) = .{},
+    ring: RingBuffer,
     thread: ?std.Thread = null,
     inner_write_fn: *const fn (*anyopaque, []const u8) void,
     inner_ptr: *anyopaque,
-    allocator: std.mem.Allocator,
 
-    /// Pre-allocate ring buffer with `size` slots. Size must be a power of two.
+    /// Create an async appender wrapping `inner` with `size` ring buffer slots.
+    /// Size must be a power of two.
     /// `inner` must be a pointer to a struct with
     /// `pub fn write(self: *T, bytes: []const u8) void`.
     pub fn init(allocator: std.mem.Allocator, size: u32, inner: anytype) !AsyncAppend {
-        std.debug.assert(size > 0 and (size & (size - 1)) == 0);
-
         const T = @TypeOf(inner.*);
         const wrapper = struct {
             fn writeFn(ctx: *anyopaque, bytes: []const u8) void {
@@ -71,18 +25,10 @@ pub const AsyncAppend = struct {
             }
         };
 
-        const slots = try allocator.alloc(Slot, size);
-        for (slots, 0..) |*slot, i| {
-            slot.sequence = std.atomic.Value(usize).init(i);
-            slot.len = 0;
-        }
-
         return .{
-            .slots = slots,
-            .mask = size - 1,
+            .ring = try RingBuffer.init(allocator, size),
             .inner_write_fn = wrapper.writeFn,
             .inner_ptr = @ptrCast(inner),
-            .allocator = allocator,
         };
     }
 
@@ -93,130 +39,68 @@ pub const AsyncAppend = struct {
 
     /// Non-blocking enqueue. Drops message if ring is full.
     pub fn write(self: *AsyncAppend, bytes: []const u8) void {
-        var head = self.head.load(.monotonic);
-        while (true) {
-            const slot = &self.slots[head & self.mask];
-            const seq = slot.sequence.load(.acquire);
-
-            if (seq == head) {
-                if (self.head.cmpxchgWeak(head, head +% 1, .monotonic, .monotonic)) |updated| {
-                    head = updated;
-                    continue;
-                }
-                // Claimed — write payload then publish.
-                const copy_len = @min(bytes.len, max_message_size);
-                @memcpy(slot.data[0..copy_len], bytes[0..copy_len]);
-                slot.len = copy_len;
-                slot.sequence.store(head +% 1, .release);
-                _ = self.epoch.fetchAdd(1, .release);
-                self.waiter.notifyOne();
-                return;
-            }
-
-            // Wrapping diff avoids @intCast trap on long-running systems.
-            const diff = @as(isize, @bitCast(seq -% head));
-            if (diff < 0) {
-                return; // Full — drop message.
-            }
-            head = self.head.load(.monotonic);
-        }
+        _ = self.ring.push(bytes);
     }
 
-    /// Signal shutdown, join consumer thread, drain remaining, free slots.
+    /// Signal shutdown, join consumer thread, drain remaining, free ring.
     pub fn deinit(self: *AsyncAppend) void {
-        self.shutdown.store(true, .release);
-        _ = self.epoch.fetchAdd(1, .release);
-        self.waiter.broadcast();
+        self.ring.close();
         if (self.thread) |t| {
             t.join();
             self.thread = null;
         }
         self.drain();
-        self.allocator.free(self.slots);
+        self.ring.deinit();
     }
 
     fn consumerLoop(self: *AsyncAppend) void {
-        while (!self.shutdown.load(.acquire)) {
-            if (self.tryConsume()) continue;
+        while (!self.ring.isClosed()) {
+            if (self.consumeOne()) continue;
 
             // Spin briefly before parking on condvar.
             var spin: u8 = 0;
             while (spin < 16) : (spin += 1) {
-                if (self.tryConsume()) break;
-                if (self.shutdown.load(.acquire)) return;
+                if (self.consumeOne()) break;
+                if (self.ring.isClosed()) return;
                 std.atomic.spinLoopHint();
             } else {
-                self.park();
+                self.ring.park();
             }
         }
     }
 
-    /// Epoch-based parking: prevents lost wakeups by checking that the
-    /// epoch hasn't changed between deciding to park and actually sleeping.
-    fn park(self: *AsyncAppend) void {
-        const e0 = self.epoch.load(.acquire);
-
-        self.waiter.mutex.lock();
-        defer self.waiter.mutex.unlock();
-
-        _ = self.waiter.parked.fetchAdd(1, .monotonic);
-        defer _ = self.waiter.parked.fetchSub(1, .monotonic);
-
-        while (true) {
-            if (self.shutdown.load(.acquire)) return;
-            if (self.peekReadable()) return;
-            if (self.epoch.load(.acquire) != e0) return;
-            self.waiter.cond.wait(&self.waiter.mutex);
-        }
-    }
-
-    fn peekReadable(self: *AsyncAppend) bool {
-        const tail = self.tail.load(.monotonic);
-        const slot = &self.slots[tail & self.mask];
-        const seq = slot.sequence.load(.acquire);
-        return seq == tail +% 1;
-    }
-
-    fn tryConsume(self: *AsyncAppend) bool {
-        const tail = self.tail.load(.monotonic);
-        const slot = &self.slots[tail & self.mask];
-        const seq = slot.sequence.load(.acquire);
-
-        if (seq == tail +% 1) {
-            if (self.tail.cmpxchgWeak(tail, tail +% 1, .monotonic, .monotonic)) |_| {
-                return false;
-            }
-            self.inner_write_fn(self.inner_ptr, slot.data[0..slot.len]);
-            slot.sequence.store(tail +% self.mask +% 1, .release);
+    fn consumeOne(self: *AsyncAppend) bool {
+        if (self.ring.tryPop()) |msg| {
+            self.inner_write_fn(self.inner_ptr, msg.bytes());
+            msg.release();
             return true;
         }
         return false;
     }
 
     fn drain(self: *AsyncAppend) void {
-        while (self.tryConsume()) {}
+        while (self.ring.tryPop()) |msg| {
+            self.inner_write_fn(self.inner_ptr, msg.bytes());
+            msg.release();
+        }
     }
 };
 
 // ==== 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.last_len = bytes.len;
         _ = self.count.fetchAdd(1, .monotonic);
     }
 };
@@ -277,7 +161,6 @@ test "AsyncAppend truncates long messages" {
     var collector = TestCollector.init();
     var appender = try AsyncAppend.init(std.testing.allocator, 16, &collector);
 
-    // Create a message longer than max_message_size.
     var long_msg: [max_message_size + 100]u8 = undefined;
     @memset(&long_msg, 'A');