From f4d4286917d85b365728132ee1807773fee30cd5 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 2 Jun 2026 09:15:23 +0000
Subject: [PATCH 01/13] Stream produce request topic data in the server

Add streaming readers for the variable-length topic_data array of a
Produce request. reader.topic_data reads a topic_id UUID plus its
partition count, and reader.partition_data reads a partition index plus
the records byte length. The record bytes are left in the stream for the
caller, so an arbitrarily large request is processed one topic/partition
at a time instead of being buffered whole.

The server's produce() now loops over topic_data_size topics and their
partitions, logging each topic/partition and draining the records in
fixed-size chunks (logging the byte count and a short hex preview).

To mirror the streaming readers, the writer's topic_data/partition_data
are reworked from whole-array helpers into per-element ones: the topic
array length stays in produce_req, the partition count moves into the
topic header. Their byte-exact tests become round-trip tests against the
new readers, matching the file's prevailing style. The client now writes
example data for two topics with one and two partitions.

https://claude.ai/code/session_01Sgm6nVAriGJDHANn7yLTnK
---
 client.zig   |  13 +++-
 klog.zig     |  41 ++++++++++++
 protocol.zig | 171 +++++++++++++++++++++++++++++----------------------
 3 files changed, 152 insertions(+), 73 deletions(-)

diff --git a/client.zig b/client.zig
index be73824..0c7e0f9 100644
--- a/client.zig
+++ b/client.zig
@@ -28,7 +28,18 @@ pub fn main(init: std.process.Init) !void {
     const req = proto.ProduceRequest{
         .acks = -1, // ISR
         .timeout_ms = 1000,
-        .topic_data_size = 1,
+        .topic_data_size = 2,
     };
     try writer.produce_req(io_writer, req);
+
+    // Topic one: a single partition.
+    const topic_one = [_]u8{0x11} ** 16;
+    try writer.topic_data(io_writer, topic_one, 1);
+    try writer.partition_data(io_writer, .{ .index = 0, .records = &[_]u8{ 0xca, 0xfe, 0xba, 0xbe } });
+
+    // Topic two: two partitions.
+    const topic_two = [_]u8{0x22} ** 16;
+    try writer.topic_data(io_writer, topic_two, 2);
+    try writer.partition_data(io_writer, .{ .index = 0, .records = &[_]u8{ 0xde, 0xad, 0xbe, 0xef } });
+    try writer.partition_data(io_writer, .{ .index = 1, .records = &[_]u8{ 0x01, 0x02, 0x03 } });
 }
diff --git a/klog.zig b/klog.zig
index 68fd1e5..2708f53 100644
--- a/klog.zig
+++ b/klog.zig
@@ -58,4 +58,45 @@ fn produce(io: Io, io_reader: *Io.Reader, stream: Io.net.Stream) !void {
     var buf: [max_transactional_id]u8 = undefined;
     const req = try reader.produce_req(io_reader, &buf);
     log.debug("produce request: {}", .{req});
+
+    // Read the topic_data array one entry at a time, using the counts and
+    // sizes that prefix each level rather than buffering the whole request.
+    for (0..req.topic_data_size) |_| {
+        const topic = try reader.topic_data(io_reader);
+        log.info("topic {x}: {d} partition(s)", .{ topic.topic_id, topic.partition_count });
+        for (0..topic.partition_count) |_| {
+            const partition = try reader.partition_data(io_reader);
+            try log_records(io_reader, partition);
+        }
+    }
+}
+
+/// Logs a partition's records, consuming them from the stream in
+/// fixed-size chunks so an arbitrarily large records blob never has to be
+/// held in memory at once. Logs the byte count and a short hex preview of
+/// the first chunk.
+fn log_records(io_reader: *Io.Reader, partition: proto.Partition) !void {
+    const size = partition.records_size orelse {
+        log.info("  partition {d}: null records", .{partition.index});
+        return;
+    };
+
+    var chunk: [4096]u8 = undefined;
+    const first_len: usize = @intCast(@min(size, chunk.len));
+    try io_reader.readSliceAll(chunk[0..first_len]);
+    const preview = chunk[0..@min(first_len, 16)];
+    log.info("  partition {d}: {d} record byte(s), first {d}: {x}", .{
+        partition.index,
+        size,
+        preview.len,
+        preview,
+    });
+
+    // Drain the remaining bytes, again one chunk at a time.
+    var remaining = size - first_len;
+    while (remaining > 0) {
+        const take: usize = @intCast(@min(remaining, chunk.len));
+        try io_reader.readSliceAll(chunk[0..take]);
+        remaining -= take;
+    }
 }
diff --git a/protocol.zig b/protocol.zig
index 32befdc..0daab88 100644
--- a/protocol.zig
+++ b/protocol.zig
@@ -57,8 +57,6 @@ pub const ProduceRequest = struct {
     topic_data_size: u64,
 };
 
-// TODO consider client writing the data piece by piece instead
-
 /// topic_id => UUID
 /// partition_data => { index records }
 pub const TopicData = struct {
@@ -73,6 +71,24 @@ pub const PartitionData = struct {
     records: ?[]const u8, // ?[]Record,
 };
 
+/// The fixed part of a topic_data entry, read up front so the partitions
+/// can be streamed one at a time without buffering the whole entry. The
+/// topic array length itself lives in `ProduceRequest.topic_data_size`.
+pub const Topic = struct {
+    topic_id: [16]u8, // UUID
+    /// The number of partition_data entries that follow.
+    partition_count: u64,
+};
+
+/// The fixed part of a partition_data entry. `records_size` is the byte
+/// length of the COMPACT_NULLABLE_RECORDS that follow (null for null
+/// records); the caller consumes those bytes itself rather than buffering
+/// them here.
+pub const Partition = struct {
+    index: i32,
+    records_size: ?u64,
+};
+
 pub const Record = struct {
     // XXX
 };
@@ -197,6 +213,35 @@ pub const reader = struct {
         };
     }
 
+    /// Reads one topic_data entry's fixed part: the topic_id UUID followed
+    /// by the partition_data array length. The caller then reads
+    /// `partition_count` partitions with `partition_data`. There are
+    /// `ProduceRequest.topic_data_size` of these entries in a request.
+    pub fn topic_data(r: *Io.Reader) !Topic {
+        var topic_id: [16]u8 = undefined;
+        try r.readSliceAll(&topic_id);
+        return .{
+            .topic_id = topic_id,
+            .partition_count = try compact_size(r),
+        };
+    }
+
+    /// Reads one partition_data entry's fixed part: the partition index
+    /// followed by the COMPACT_NULLABLE_RECORDS length. The caller then
+    /// consumes `records_size` bytes of records (or none when null) before
+    /// reading the next partition.
+    pub fn partition_data(r: *Io.Reader) !Partition {
+        const index = try r.takeInt(i32, BE);
+        // records => COMPACT_NULLABLE_RECORDS: length N+1, or 0 for null.
+        // compact_size would underflow on the 0 marker, so decode it here
+        // the same way compact_nullable_str handles its null length.
+        const len = try unsigned_varint(r);
+        return .{
+            .index = index,
+            .records_size = if (len == 0) null else len - 1,
+        };
+    }
+
     /// NULLABLE_STRING
     /// Represents a sequence of characters or null. For non-null strings, first the
     /// length N is given as an INT16. Then N bytes follow which are the UTF-8
@@ -269,30 +314,24 @@ pub const writer = struct {
         try unsigned_varint(w, req.topic_data_size + 1);
     }
 
-    /// Writes a COMPACT array of topic_data: the element count as a compact
-    /// array size, followed by each { topic_id (partition_data) } entry.
-    fn topic_data(w: *Io.Writer, arr: []const TopicData) !void {
-        try compact_size(w, arr.len);
-        for (arr) |topic| {
-            try w.writeAll(&topic.topic_id);
-            try partition_data(w, topic.partition_data);
-        }
+    /// Writes one topic_data entry's fixed part: the topic_id UUID followed
+    /// by the partition_data array length. Each of the `partition_count`
+    /// partitions is then written with `partition_data`. The topic array
+    /// length itself is written by `produce_req` (topic_data_size).
+    pub fn topic_data(w: *Io.Writer, topic_id: [16]u8, partition_count: u64) !void {
+        try w.writeAll(&topic_id);
+        try compact_size(w, partition_count);
     }
 
-    /// Writes a COMPACT array of partition_data: the element count as a
-    /// compact array size, followed by each { index records } entry.
-    fn partition_data(w: *Io.Writer, arr: []const PartitionData) !void {
-        try compact_size(w, arr.len);
-        for (arr) |partition| {
-            try w.writeInt(i32, partition.index, BE);
-            // records => COMPACT_NULLABLE_RECORDS: length N+1 then N bytes,
-            // or 0 for null.
-            if (partition.records) |records| {
-                try compact_size(w, records.len);
-                try w.writeAll(records);
-            } else {
-                try compact_size(w, null);
-            }
+    /// Writes one partition_data entry: the index followed by its
+    /// COMPACT_NULLABLE_RECORDS (length N+1 then N bytes, or 0 for null).
+    pub fn partition_data(w: *Io.Writer, partition: PartitionData) !void {
+        try w.writeInt(i32, partition.index, BE);
+        if (partition.records) |records| {
+            try compact_size(w, records.len);
+            try w.writeAll(records);
+        } else {
+            try compact_size(w, null);
         }
     }
 
@@ -448,67 +487,55 @@ test "produce_req round-trip without transactional_id" {
     try testing.expect(read.transactional_id == null);
 }
 
-test "topic_data writes count, ids, and partition_data" {
-    var buf: [64]u8 = undefined;
+test "topic_data round-trip" {
+    var buf: [32]u8 = undefined;
     var w = Io.Writer.fixed(&buf);
-    const topics = [_]TopicData{
-        .{
-            .topic_id = .{ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15 },
-            .partition_data = &[_]PartitionData{
-                .{ .index = 1, .records = &[_]u8{ 0xde, 0xad, 0xbe } },
-                .{ .index = 2, .records = null },
-            },
-        },
-    };
-    try writer.topic_data(&w, &topics);
-
-    try testing.expectEqualSlices(u8, &[_]u8{
-        0x02, // compact array size: 1 topic (N+1)
-        0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, // topic_id UUID
-        0x03, // compact array size: 2 partitions (N+1)
-        0x00, 0x00, 0x00, 0x01, // index 1
-        0x04, 0xde, 0xad, 0xbe, // records: size 3 (N+1) then the bytes
-        0x00, 0x00, 0x00, 0x02, // index 2
-        0x00, // null records
-    }, w.buffered());
+    const topic_id = [16]u8{ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15 };
+    try writer.topic_data(&w, topic_id, 2);
+
+    var r = Io.Reader.fixed(&buf);
+    const read = try reader.topic_data(&r);
+    try testing.expectEqualSlices(u8, &topic_id, &read.topic_id);
+    try testing.expectEqual(@as(u64, 2), read.partition_count);
 }
 
-test "topic_data writes empty array" {
-    var buf: [16]u8 = undefined;
+test "topic_data round-trip with no partitions" {
+    var buf: [32]u8 = undefined;
     var w = Io.Writer.fixed(&buf);
-    const topics = [_]TopicData{};
-    try writer.topic_data(&w, &topics);
+    const topic_id = [_]u8{0xab} ** 16;
+    try writer.topic_data(&w, topic_id, 0);
 
-    // Just the compact array size for zero elements (N+1).
-    try testing.expectEqualSlices(u8, &[_]u8{0x01}, w.buffered());
+    var r = Io.Reader.fixed(&buf);
+    const read = try reader.topic_data(&r);
+    try testing.expectEqualSlices(u8, &topic_id, &read.topic_id);
+    try testing.expectEqual(@as(u64, 0), read.partition_count);
 }
 
-test "partition_data writes count, indices, and records" {
-    var buf: [64]u8 = undefined;
+test "partition_data round-trip with records" {
+    var buf: [16]u8 = undefined;
     var w = Io.Writer.fixed(&buf);
-    const partitions = [_]PartitionData{
-        .{ .index = 1, .records = &[_]u8{ 0xde, 0xad, 0xbe } },
-        .{ .index = 2, .records = null },
-    };
-    try writer.partition_data(&w, &partitions);
-
-    try testing.expectEqualSlices(u8, &[_]u8{
-        0x03, // compact array size: 2 partitions (N+1)
-        0x00, 0x00, 0x00, 0x01, // index 1
-        0x04, 0xde, 0xad, 0xbe, // records: size 3 (N+1) then the bytes
-        0x00, 0x00, 0x00, 0x02, // index 2
-        0x00, // null records
-    }, w.buffered());
+    const records = [_]u8{ 0xde, 0xad, 0xbe };
+    try writer.partition_data(&w, .{ .index = 1, .records = &records });
+
+    var r = Io.Reader.fixed(&buf);
+    const read = try reader.partition_data(&r);
+    try testing.expectEqual(@as(i32, 1), read.index);
+    try testing.expectEqual(@as(?u64, records.len), read.records_size);
+    // The record bytes are left in the stream for the caller to consume.
+    var out: [3]u8 = undefined;
+    try r.readSliceAll(&out);
+    try testing.expectEqualSlices(u8, &records, &out);
 }
 
-test "partition_data writes empty array" {
+test "partition_data round-trip with null records" {
     var buf: [16]u8 = undefined;
     var w = Io.Writer.fixed(&buf);
-    const partitions = [_]PartitionData{};
-    try writer.partition_data(&w, &partitions);
+    try writer.partition_data(&w, .{ .index = 2, .records = null });
 
-    // Just the compact array size for zero elements (N+1).
-    try testing.expectEqualSlices(u8, &[_]u8{0x01}, w.buffered());
+    var r = Io.Reader.fixed(&buf);
+    const read = try reader.partition_data(&r);
+    try testing.expectEqual(@as(i32, 2), read.index);
+    try testing.expectEqual(@as(?u64, null), read.records_size);
 }
 
 test "nullable_str round-trip non-null" {

From 92943fc113e172a555adbbf93d8b0fca3e2bd88a Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 2 Jun 2026 09:31:39 +0000
Subject: [PATCH 02/13] Use structs to write topic/partition data in the client

Make writer.topic_data take a Topic struct so both writer helpers now
take structs (partition_data already took PartitionData), mirroring the
reader side which returns Topic/Partition. The client organizes its
example data in TopicData/PartitionData structs and iterates, deriving
topic_data_size from the array length instead of hardcoding it. No change
to the wire format.

https://claude.ai/code/session_01Sgm6nVAriGJDHANn7yLTnK
---
 client.zig   | 43 ++++++++++++++++++++++++++++++++-----------
 protocol.zig | 17 +++++++++--------
 2 files changed, 41 insertions(+), 19 deletions(-)

diff --git a/client.zig b/client.zig
index 0c7e0f9..defa55c 100644
--- a/client.zig
+++ b/client.zig
@@ -25,21 +25,42 @@ pub fn main(init: std.process.Init) !void {
         .client_id = "test-client",
     };
     try writer.req_header(io_writer, req_header);
+
+    // Example data: topic one with a single partition, topic two with two.
+    const topic1_uuid = [_]u8{0x11} ** 16;
+    const topic2_uuid = [_]u8{0x22} ** 16;
+    const topics = [_]proto.TopicData{
+        .{
+            .topic_id = topic1_uuid,
+            .partition_data = &.{
+                .{ .index = 0, .records = &[_]u8{ 0xca, 0xfe, 0xba, 0xbe } },
+            },
+        },
+        .{
+            .topic_id = topic2_uuid,
+            .partition_data = &.{
+                .{ .index = 0, .records = &[_]u8{ 0xde, 0xad, 0xbe, 0xef } },
+                .{ .index = 1, .records = &[_]u8{ 0x01, 0x02, 0x03 } },
+            },
+        },
+    };
+
     const req = proto.ProduceRequest{
         .acks = -1, // ISR
         .timeout_ms = 1000,
-        .topic_data_size = 2,
+        .topic_data_size = topics.len,
     };
     try writer.produce_req(io_writer, req);
 
-    // Topic one: a single partition.
-    const topic_one = [_]u8{0x11} ** 16;
-    try writer.topic_data(io_writer, topic_one, 1);
-    try writer.partition_data(io_writer, .{ .index = 0, .records = &[_]u8{ 0xca, 0xfe, 0xba, 0xbe } });
-
-    // Topic two: two partitions.
-    const topic_two = [_]u8{0x22} ** 16;
-    try writer.topic_data(io_writer, topic_two, 2);
-    try writer.partition_data(io_writer, .{ .index = 0, .records = &[_]u8{ 0xde, 0xad, 0xbe, 0xef } });
-    try writer.partition_data(io_writer, .{ .index = 1, .records = &[_]u8{ 0x01, 0x02, 0x03 } });
+    // Write each topic header followed by its partitions, mirroring how the
+    // server streams them back out.
+    for (topics) |topic| {
+        try writer.topic_data(io_writer, .{
+            .topic_id = topic.topic_id,
+            .partition_count = topic.partition_data.len,
+        });
+        for (topic.partition_data) |partition| {
+            try writer.partition_data(io_writer, partition);
+        }
+    }
 }
diff --git a/protocol.zig b/protocol.zig
index 0daab88..36e98ef 100644
--- a/protocol.zig
+++ b/protocol.zig
@@ -71,9 +71,10 @@ pub const PartitionData = struct {
     records: ?[]const u8, // ?[]Record,
 };
 
-/// The fixed part of a topic_data entry, read up front so the partitions
-/// can be streamed one at a time without buffering the whole entry. The
-/// topic array length itself lives in `ProduceRequest.topic_data_size`.
+/// The fixed part of a topic_data entry, handled up front so the
+/// partitions can be streamed one at a time without buffering the whole
+/// entry. The topic array length itself lives in
+/// `ProduceRequest.topic_data_size`.
 pub const Topic = struct {
     topic_id: [16]u8, // UUID
     /// The number of partition_data entries that follow.
@@ -318,9 +319,9 @@ pub const writer = struct {
     /// by the partition_data array length. Each of the `partition_count`
     /// partitions is then written with `partition_data`. The topic array
     /// length itself is written by `produce_req` (topic_data_size).
-    pub fn topic_data(w: *Io.Writer, topic_id: [16]u8, partition_count: u64) !void {
-        try w.writeAll(&topic_id);
-        try compact_size(w, partition_count);
+    pub fn topic_data(w: *Io.Writer, topic: Topic) !void {
+        try w.writeAll(&topic.topic_id);
+        try compact_size(w, topic.partition_count);
     }
 
     /// Writes one partition_data entry: the index followed by its
@@ -491,7 +492,7 @@ test "topic_data round-trip" {
     var buf: [32]u8 = undefined;
     var w = Io.Writer.fixed(&buf);
     const topic_id = [16]u8{ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15 };
-    try writer.topic_data(&w, topic_id, 2);
+    try writer.topic_data(&w, .{ .topic_id = topic_id, .partition_count = 2 });
 
     var r = Io.Reader.fixed(&buf);
     const read = try reader.topic_data(&r);
@@ -503,7 +504,7 @@ test "topic_data round-trip with no partitions" {
     var buf: [32]u8 = undefined;
     var w = Io.Writer.fixed(&buf);
     const topic_id = [_]u8{0xab} ** 16;
-    try writer.topic_data(&w, topic_id, 0);
+    try writer.topic_data(&w, .{ .topic_id = topic_id, .partition_count = 0 });
 
     var r = Io.Reader.fixed(&buf);
     const read = try reader.topic_data(&r);

From aa6182454b18a676cda993d4ea3fbd5caf0ab226 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 2 Jun 2026 09:51:01 +0000
Subject: [PATCH 03/13] Use a readable hex-word for the last example payload

Replace the {01 02 03} partition records with feedface so all three
example payloads (cafebabe, deadbeef, feedface) read as recognizable
hex words.

https://claude.ai/code/session_01Sgm6nVAriGJDHANn7yLTnK
---
 client.zig | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/client.zig b/client.zig
index defa55c..c5dd099 100644
--- a/client.zig
+++ b/client.zig
@@ -40,7 +40,7 @@ pub fn main(init: std.process.Init) !void {
             .topic_id = topic2_uuid,
             .partition_data = &.{
                 .{ .index = 0, .records = &[_]u8{ 0xde, 0xad, 0xbe, 0xef } },
-                .{ .index = 1, .records = &[_]u8{ 0x01, 0x02, 0x03 } },
+                .{ .index = 1, .records = &[_]u8{ 0xfe, 0xed, 0xfa, 0xce } },
             },
         },
     };

From b0d1034ea327752c7e80bf3db8fded805aeffda4 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 2 Jun 2026 13:28:09 +0000
Subject: [PATCH 04/13] Share one Topic/Partition struct between reader and
 writer

Delete the writer-only TopicData/PartitionData structs and use the
reader's Topic/Partition for both directions. PartitionData embedded the
variable record bytes, which forced a separate size-only struct for the
streaming reader; moving those bytes out of the struct lets a single
Partition { index, records_size } serve both.

writer.partition_data now writes only the index and the
COMPACT_NULLABLE_RECORDS length prefix; the caller writes the record bytes
itself, mirroring how reader.partition_data returns the size and leaves
the bytes in the stream. The client gains a small write_partition helper
that derives the size from the slice. Wire format is unchanged.

https://claude.ai/code/session_01Sgm6nVAriGJDHANn7yLTnK
---
 client.zig   | 54 ++++++++++++++++++++++------------------------------
 protocol.zig | 46 ++++++++++++++++++--------------------------
 2 files changed, 42 insertions(+), 58 deletions(-)

diff --git a/client.zig b/client.zig
index c5dd099..83bef76 100644
--- a/client.zig
+++ b/client.zig
@@ -26,41 +26,33 @@ pub fn main(init: std.process.Init) !void {
     };
     try writer.req_header(io_writer, req_header);
 
-    // Example data: topic one with a single partition, topic two with two.
-    const topic1_uuid = [_]u8{0x11} ** 16;
-    const topic2_uuid = [_]u8{0x22} ** 16;
-    const topics = [_]proto.TopicData{
-        .{
-            .topic_id = topic1_uuid,
-            .partition_data = &.{
-                .{ .index = 0, .records = &[_]u8{ 0xca, 0xfe, 0xba, 0xbe } },
-            },
-        },
-        .{
-            .topic_id = topic2_uuid,
-            .partition_data = &.{
-                .{ .index = 0, .records = &[_]u8{ 0xde, 0xad, 0xbe, 0xef } },
-                .{ .index = 1, .records = &[_]u8{ 0xfe, 0xed, 0xfa, 0xce } },
-            },
-        },
-    };
-
     const req = proto.ProduceRequest{
         .acks = -1, // ISR
         .timeout_ms = 1000,
-        .topic_data_size = topics.len,
+        .topic_data_size = 2,
     };
     try writer.produce_req(io_writer, req);
 
-    // Write each topic header followed by its partitions, mirroring how the
-    // server streams them back out.
-    for (topics) |topic| {
-        try writer.topic_data(io_writer, .{
-            .topic_id = topic.topic_id,
-            .partition_count = topic.partition_data.len,
-        });
-        for (topic.partition_data) |partition| {
-            try writer.partition_data(io_writer, partition);
-        }
-    }
+    // Example data: topic one with a single partition, topic two with two.
+    // The record bytes live outside the protocol structs, so each partition
+    // is a header write followed by its records.
+    const topic1_uuid = [_]u8{0x11} ** 16;
+    const topic2_uuid = [_]u8{0x22} ** 16;
+
+    try writer.topic_data(io_writer, .{ .topic_id = topic1_uuid, .partition_count = 1 });
+    try write_partition(io_writer, 0, &[_]u8{ 0xca, 0xfe, 0xba, 0xbe });
+
+    try writer.topic_data(io_writer, .{ .topic_id = topic2_uuid, .partition_count = 2 });
+    try write_partition(io_writer, 0, &[_]u8{ 0xde, 0xad, 0xbe, 0xef });
+    try write_partition(io_writer, 1, &[_]u8{ 0xfe, 0xed, 0xfa, 0xce });
+}
+
+/// Writes a partition_data header followed by its record bytes, deriving
+/// the records size from the slice so the two always agree.
+fn write_partition(w: *std.Io.Writer, index: i32, records: ?[]const u8) !void {
+    try proto.writer.partition_data(w, .{
+        .index = index,
+        .records_size = if (records) |r| r.len else null,
+    });
+    if (records) |r| try w.writeAll(r);
 }
diff --git a/protocol.zig b/protocol.zig
index 36e98ef..e853da7 100644
--- a/protocol.zig
+++ b/protocol.zig
@@ -57,24 +57,13 @@ pub const ProduceRequest = struct {
     topic_data_size: u64,
 };
 
-/// topic_id => UUID
-/// partition_data => { index records }
-pub const TopicData = struct {
-    topic_id: [16]u8, // UUID
-    partition_data: []const PartitionData,
-};
-
-/// index => INT32
-/// records => COMPACT_NULLABLE_RECORDS
-pub const PartitionData = struct {
-    index: i32,
-    records: ?[]const u8, // ?[]Record,
-};
-
 /// The fixed part of a topic_data entry, handled up front so the
 /// partitions can be streamed one at a time without buffering the whole
 /// entry. The topic array length itself lives in
 /// `ProduceRequest.topic_data_size`.
+///
+/// topic_id => UUID
+/// partition_data => { index records }
 pub const Topic = struct {
     topic_id: [16]u8, // UUID
     /// The number of partition_data entries that follow.
@@ -83,8 +72,11 @@ pub const Topic = struct {
 
 /// The fixed part of a partition_data entry. `records_size` is the byte
 /// length of the COMPACT_NULLABLE_RECORDS that follow (null for null
-/// records); the caller consumes those bytes itself rather than buffering
-/// them here.
+/// records); those bytes live outside this struct and are written/consumed
+/// by the caller rather than buffered here.
+///
+/// index => INT32
+/// records => COMPACT_NULLABLE_RECORDS
 pub const Partition = struct {
     index: i32,
     records_size: ?u64,
@@ -324,16 +316,14 @@ pub const writer = struct {
         try compact_size(w, topic.partition_count);
     }
 
-    /// Writes one partition_data entry: the index followed by its
-    /// COMPACT_NULLABLE_RECORDS (length N+1 then N bytes, or 0 for null).
-    pub fn partition_data(w: *Io.Writer, partition: PartitionData) !void {
+    /// Writes one partition_data entry's fixed part: the index followed by
+    /// the COMPACT_NULLABLE_RECORDS length prefix (N+1, or 0 for null). The
+    /// caller then writes the `records_size` record bytes itself, mirroring
+    /// how `reader.partition_data` returns the size and leaves the bytes in
+    /// the stream for the caller to consume.
+    pub fn partition_data(w: *Io.Writer, partition: Partition) !void {
         try w.writeInt(i32, partition.index, BE);
-        if (partition.records) |records| {
-            try compact_size(w, records.len);
-            try w.writeAll(records);
-        } else {
-            try compact_size(w, null);
-        }
+        try compact_size(w, partition.records_size);
     }
 
     fn nullable_str(w: *Io.Writer, str: ?[]const u8) !void {
@@ -516,7 +506,9 @@ test "partition_data round-trip with records" {
     var buf: [16]u8 = undefined;
     var w = Io.Writer.fixed(&buf);
     const records = [_]u8{ 0xde, 0xad, 0xbe };
-    try writer.partition_data(&w, .{ .index = 1, .records = &records });
+    // The header carries only the size; the bytes are written separately.
+    try writer.partition_data(&w, .{ .index = 1, .records_size = records.len });
+    try w.writeAll(&records);
 
     var r = Io.Reader.fixed(&buf);
     const read = try reader.partition_data(&r);
@@ -531,7 +523,7 @@ test "partition_data round-trip with records" {
 test "partition_data round-trip with null records" {
     var buf: [16]u8 = undefined;
     var w = Io.Writer.fixed(&buf);
-    try writer.partition_data(&w, .{ .index = 2, .records = null });
+    try writer.partition_data(&w, .{ .index = 2, .records_size = null });
 
     var r = Io.Reader.fixed(&buf);
     const read = try reader.partition_data(&r);

From a89a95ad7b903ac217658542e14b6889ece884d0 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 2 Jun 2026 13:41:55 +0000
Subject: [PATCH 05/13] Exercise the null-records path in the example client

Add a third partition to topic two with null records so the client also
drives the COMPACT_NULLABLE_RECORDS null branch end to end (the server
logs it as "null records").

https://claude.ai/code/session_01Sgm6nVAriGJDHANn7yLTnK
---
 client.zig | 10 ++++++----
 1 file changed, 6 insertions(+), 4 deletions(-)

diff --git a/client.zig b/client.zig
index 83bef76..05dbd96 100644
--- a/client.zig
+++ b/client.zig
@@ -33,18 +33,20 @@ pub fn main(init: std.process.Init) !void {
     };
     try writer.produce_req(io_writer, req);
 
-    // Example data: topic one with a single partition, topic two with two.
-    // The record bytes live outside the protocol structs, so each partition
-    // is a header write followed by its records.
+    // Example data: topic one with a single partition, topic two with three
+    // (the last with null records). The record bytes live outside the
+    // protocol structs, so each partition is a header write followed by its
+    // records.
     const topic1_uuid = [_]u8{0x11} ** 16;
     const topic2_uuid = [_]u8{0x22} ** 16;
 
     try writer.topic_data(io_writer, .{ .topic_id = topic1_uuid, .partition_count = 1 });
     try write_partition(io_writer, 0, &[_]u8{ 0xca, 0xfe, 0xba, 0xbe });
 
-    try writer.topic_data(io_writer, .{ .topic_id = topic2_uuid, .partition_count = 2 });
+    try writer.topic_data(io_writer, .{ .topic_id = topic2_uuid, .partition_count = 3 });
     try write_partition(io_writer, 0, &[_]u8{ 0xde, 0xad, 0xbe, 0xef });
     try write_partition(io_writer, 1, &[_]u8{ 0xfe, 0xed, 0xfa, 0xce });
+    try write_partition(io_writer, 2, null);
 }
 
 /// Writes a partition_data header followed by its record bytes, deriving

From b5035cd08933353e80aa3b742492757af48650c0 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 2 Jun 2026 13:43:31 +0000
Subject: [PATCH 06/13] Name the shared structs TopicData/PartitionData

Rename Topic/Partition to TopicData/PartitionData so the struct names
match the protocol's topic_data/partition_data fields. Pure rename, no
behavior change.

https://claude.ai/code/session_01Sgm6nVAriGJDHANn7yLTnK
---
 klog.zig     |  2 +-
 protocol.zig | 12 ++++++------
 2 files changed, 7 insertions(+), 7 deletions(-)

diff --git a/klog.zig b/klog.zig
index 2708f53..1711bc9 100644
--- a/klog.zig
+++ b/klog.zig
@@ -75,7 +75,7 @@ fn produce(io: Io, io_reader: *Io.Reader, stream: Io.net.Stream) !void {
 /// fixed-size chunks so an arbitrarily large records blob never has to be
 /// held in memory at once. Logs the byte count and a short hex preview of
 /// the first chunk.
-fn log_records(io_reader: *Io.Reader, partition: proto.Partition) !void {
+fn log_records(io_reader: *Io.Reader, partition: proto.PartitionData) !void {
     const size = partition.records_size orelse {
         log.info("  partition {d}: null records", .{partition.index});
         return;
diff --git a/protocol.zig b/protocol.zig
index e853da7..8264c8f 100644
--- a/protocol.zig
+++ b/protocol.zig
@@ -64,7 +64,7 @@ pub const ProduceRequest = struct {
 ///
 /// topic_id => UUID
 /// partition_data => { index records }
-pub const Topic = struct {
+pub const TopicData = struct {
     topic_id: [16]u8, // UUID
     /// The number of partition_data entries that follow.
     partition_count: u64,
@@ -77,7 +77,7 @@ pub const Topic = struct {
 ///
 /// index => INT32
 /// records => COMPACT_NULLABLE_RECORDS
-pub const Partition = struct {
+pub const PartitionData = struct {
     index: i32,
     records_size: ?u64,
 };
@@ -210,7 +210,7 @@ pub const reader = struct {
     /// by the partition_data array length. The caller then reads
     /// `partition_count` partitions with `partition_data`. There are
     /// `ProduceRequest.topic_data_size` of these entries in a request.
-    pub fn topic_data(r: *Io.Reader) !Topic {
+    pub fn topic_data(r: *Io.Reader) !TopicData {
         var topic_id: [16]u8 = undefined;
         try r.readSliceAll(&topic_id);
         return .{
@@ -223,7 +223,7 @@ pub const reader = struct {
     /// followed by the COMPACT_NULLABLE_RECORDS length. The caller then
     /// consumes `records_size` bytes of records (or none when null) before
     /// reading the next partition.
-    pub fn partition_data(r: *Io.Reader) !Partition {
+    pub fn partition_data(r: *Io.Reader) !PartitionData {
         const index = try r.takeInt(i32, BE);
         // records => COMPACT_NULLABLE_RECORDS: length N+1, or 0 for null.
         // compact_size would underflow on the 0 marker, so decode it here
@@ -311,7 +311,7 @@ pub const writer = struct {
     /// by the partition_data array length. Each of the `partition_count`
     /// partitions is then written with `partition_data`. The topic array
     /// length itself is written by `produce_req` (topic_data_size).
-    pub fn topic_data(w: *Io.Writer, topic: Topic) !void {
+    pub fn topic_data(w: *Io.Writer, topic: TopicData) !void {
         try w.writeAll(&topic.topic_id);
         try compact_size(w, topic.partition_count);
     }
@@ -321,7 +321,7 @@ pub const writer = struct {
     /// caller then writes the `records_size` record bytes itself, mirroring
     /// how `reader.partition_data` returns the size and leaves the bytes in
     /// the stream for the caller to consume.
-    pub fn partition_data(w: *Io.Writer, partition: Partition) !void {
+    pub fn partition_data(w: *Io.Writer, partition: PartitionData) !void {
         try w.writeInt(i32, partition.index, BE);
         try compact_size(w, partition.records_size);
     }

From b72b579fd85441dda2021c9b4c90a44ce1b26db3 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 2 Jun 2026 13:46:42 +0000
Subject: [PATCH 07/13] Drop @as coercions from test asserts

expectEqual resolves expected/actual to a common type via peer type
resolution, so bare literals (and records.len / null) suffice in the
asserts, matching the existing expectEqual(3, read.topic_data_size)
style. The remaining @as in unsigned_varint are implementation widening,
not asserts.

https://claude.ai/code/session_01Sgm6nVAriGJDHANn7yLTnK
---
 protocol.zig | 22 +++++++++++-----------
 1 file changed, 11 insertions(+), 11 deletions(-)

diff --git a/protocol.zig b/protocol.zig
index 8264c8f..1774769 100644
--- a/protocol.zig
+++ b/protocol.zig
@@ -377,7 +377,7 @@ test "msg_size round-trip" {
     try writer.msg_size(&w, 1024);
 
     var r = Io.Reader.fixed(&buf);
-    try testing.expectEqual(@as(i32, 1024), try reader.msg_size(&r));
+    try testing.expectEqual(1024, try reader.msg_size(&r));
 }
 
 test "msg_size round-trip zero" {
@@ -386,7 +386,7 @@ test "msg_size round-trip zero" {
     try writer.msg_size(&w, 0);
 
     var r = Io.Reader.fixed(&buf);
-    try testing.expectEqual(@as(i32, 0), try reader.msg_size(&r));
+    try testing.expectEqual(0, try reader.msg_size(&r));
 }
 
 test "req_header round-trip with client_id" {
@@ -487,7 +487,7 @@ test "topic_data round-trip" {
     var r = Io.Reader.fixed(&buf);
     const read = try reader.topic_data(&r);
     try testing.expectEqualSlices(u8, &topic_id, &read.topic_id);
-    try testing.expectEqual(@as(u64, 2), read.partition_count);
+    try testing.expectEqual(2, read.partition_count);
 }
 
 test "topic_data round-trip with no partitions" {
@@ -499,7 +499,7 @@ test "topic_data round-trip with no partitions" {
     var r = Io.Reader.fixed(&buf);
     const read = try reader.topic_data(&r);
     try testing.expectEqualSlices(u8, &topic_id, &read.topic_id);
-    try testing.expectEqual(@as(u64, 0), read.partition_count);
+    try testing.expectEqual(0, read.partition_count);
 }
 
 test "partition_data round-trip with records" {
@@ -512,8 +512,8 @@ test "partition_data round-trip with records" {
 
     var r = Io.Reader.fixed(&buf);
     const read = try reader.partition_data(&r);
-    try testing.expectEqual(@as(i32, 1), read.index);
-    try testing.expectEqual(@as(?u64, records.len), read.records_size);
+    try testing.expectEqual(1, read.index);
+    try testing.expectEqual(records.len, read.records_size);
     // The record bytes are left in the stream for the caller to consume.
     var out: [3]u8 = undefined;
     try r.readSliceAll(&out);
@@ -527,8 +527,8 @@ test "partition_data round-trip with null records" {
 
     var r = Io.Reader.fixed(&buf);
     const read = try reader.partition_data(&r);
-    try testing.expectEqual(@as(i32, 2), read.index);
-    try testing.expectEqual(@as(?u64, null), read.records_size);
+    try testing.expectEqual(2, read.index);
+    try testing.expectEqual(null, read.records_size);
 }
 
 test "nullable_str round-trip non-null" {
@@ -688,9 +688,9 @@ test "unsigned_varint stops at varint boundary" {
     // continuation bit is set.
     const buf = [_]u8{ 0x96, 0x01, 0xff, 0xff };
     var r = Io.Reader.fixed(&buf);
-    try testing.expectEqual(@as(u64, 150), try reader.unsigned_varint(&r));
-    try testing.expectEqual(@as(u8, 0xff), try r.takeByte());
-    try testing.expectEqual(@as(u8, 0xff), try r.takeByte());
+    try testing.expectEqual(150, try reader.unsigned_varint(&r));
+    try testing.expectEqual(0xff, try r.takeByte());
+    try testing.expectEqual(0xff, try r.takeByte());
 }
 
 test "unsigned_varint rejects over-long input" {

From dfb60509b4ae8333f50a43fb30ff40eee57ec195 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 2 Jun 2026 13:53:11 +0000
Subject: [PATCH 08/13] Make reader compact_size nullable and reuse it

The reader's compact_size returned u64 and underflowed on the 0/null
marker, so partition_data decoded COMPACT_NULLABLE_RECORDS inline. Make
compact_size return ?u64 to mirror the writer's compact_size(?u64), and
call it from partition_data (records are nullable) and compact_nullable_str
(dropping its duplicated null decode). The non-nullable COMPACT array
readers (produce_req topic count, topic_data partition count) reject the
null marker as a ProtocolError, which also fixes the previous underflow on
a 0 length there. Adds a compact_size null round-trip test.

https://claude.ai/code/session_01Sgm6nVAriGJDHANn7yLTnK
---
 protocol.zig | 42 ++++++++++++++++++++++++++----------------
 1 file changed, 26 insertions(+), 16 deletions(-)

diff --git a/protocol.zig b/protocol.zig
index 1774769..e034708 100644
--- a/protocol.zig
+++ b/protocol.zig
@@ -202,7 +202,8 @@ pub const reader = struct {
             .transactional_id = try compact_nullable_str(r, buf),
             .acks = try r.takeInt(i16, BE),
             .timeout_ms = try r.takeInt(i32, BE),
-            .topic_data_size = try compact_size(r),
+            // topic_data is a non-nullable COMPACT array; reject the null marker.
+            .topic_data_size = (try compact_size(r)) orelse return Error.ProtocolError,
         };
     }
 
@@ -215,7 +216,8 @@ pub const reader = struct {
         try r.readSliceAll(&topic_id);
         return .{
             .topic_id = topic_id,
-            .partition_count = try compact_size(r),
+            // partition_data is a non-nullable COMPACT array; reject the null marker.
+            .partition_count = (try compact_size(r)) orelse return Error.ProtocolError,
         };
     }
 
@@ -224,14 +226,10 @@ pub const reader = struct {
     /// consumes `records_size` bytes of records (or none when null) before
     /// reading the next partition.
     pub fn partition_data(r: *Io.Reader) !PartitionData {
-        const index = try r.takeInt(i32, BE);
-        // records => COMPACT_NULLABLE_RECORDS: length N+1, or 0 for null.
-        // compact_size would underflow on the 0 marker, so decode it here
-        // the same way compact_nullable_str handles its null length.
-        const len = try unsigned_varint(r);
         return .{
-            .index = index,
-            .records_size = if (len == 0) null else len - 1,
+            .index = try r.takeInt(i32, BE),
+            // records => COMPACT_NULLABLE_RECORDS: null when the length is 0.
+            .records_size = try compact_size(r),
         };
     }
 
@@ -255,20 +253,23 @@ pub const reader = struct {
     /// bytes follow which are the UTF-8 encoding of the character
     /// sequence. A null string is represented with a length of 0.
     fn compact_nullable_str(r: *Io.Reader, buf: []u8) !?[]u8 {
-        const len = try unsigned_varint(r);
-        if (len == 0) return null;
-        const n: usize = @intCast(len - 1);
+        const n: usize = @intCast((try compact_size(r)) orelse return null);
         if (n > buf.len) return Error.BufferTooSmall;
         try r.readSliceAll(buf[0..n]);
         return buf[0..n];
     }
 
     /// Reads the length N from a COMPACT array, encoded as N + 1 in an
-    /// UNSIGNED_VARINT. The same length-prefix scheme is used for compact
-    /// byte sequences (e.g. records), so this applies to both. In protocol
+    /// UNSIGNED_VARINT, or null when the length is 0 — the COMPACT_NULLABLE
+    /// null marker. The same length-prefix scheme is used for compact byte
+    /// sequences (e.g. records), so this applies to both. In protocol
     /// documentation a compact array of T instances is referred to as (T).
-    fn compact_size(r: *Io.Reader) !u64 {
-        return try unsigned_varint(r) - 1;
+    /// Mirrors the writer's `compact_size(?u64)`; non-nullable callers
+    /// reject the null marker.
+    fn compact_size(r: *Io.Reader) !?u64 {
+        const len = try unsigned_varint(r);
+        if (len == 0) return null;
+        return len - 1;
     }
 
     /// Unsigned LEB128, as used by Protocol Buffers.
@@ -643,6 +644,15 @@ test "compact_size round-trip" {
     try testing.expectEqual(size, try reader.compact_size(&r));
 }
 
+test "compact_size round-trip null" {
+    var buf: [16]u8 = undefined;
+    var w = Io.Writer.fixed(&buf);
+    try writer.compact_size(&w, null);
+
+    var r = Io.Reader.fixed(&buf);
+    try testing.expectEqual(null, try reader.compact_size(&r));
+}
+
 test "compact_size encodes null as zero" {
     var buf: [16]u8 = undefined;
     var w = Io.Writer.fixed(&buf);

From 2c51d3debf9d2e48e909dfbf72b1bc24ecd3edbf Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 2 Jun 2026 14:26:57 +0000
Subject: [PATCH 09/13] Write the null-records partition first in the client

Move topic two's null-records partition ahead of the non-null ones so a
partition with records follows it across the records boundary, exercising
that null (zero-length) records don't desync parsing of what comes next.

https://claude.ai/code/session_01Sgm6nVAriGJDHANn7yLTnK
---
 client.zig | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

diff --git a/client.zig b/client.zig
index 05dbd96..3f9c716 100644
--- a/client.zig
+++ b/client.zig
@@ -34,9 +34,9 @@ pub fn main(init: std.process.Init) !void {
     try writer.produce_req(io_writer, req);
 
     // Example data: topic one with a single partition, topic two with three
-    // (the last with null records). The record bytes live outside the
-    // protocol structs, so each partition is a header write followed by its
-    // records.
+    // (the first with null records, so a non-null partition follows it across
+    // the records boundary). The record bytes live outside the protocol
+    // structs, so each partition is a header write followed by its records.
     const topic1_uuid = [_]u8{0x11} ** 16;
     const topic2_uuid = [_]u8{0x22} ** 16;
 
@@ -44,9 +44,9 @@ pub fn main(init: std.process.Init) !void {
     try write_partition(io_writer, 0, &[_]u8{ 0xca, 0xfe, 0xba, 0xbe });
 
     try writer.topic_data(io_writer, .{ .topic_id = topic2_uuid, .partition_count = 3 });
-    try write_partition(io_writer, 0, &[_]u8{ 0xde, 0xad, 0xbe, 0xef });
-    try write_partition(io_writer, 1, &[_]u8{ 0xfe, 0xed, 0xfa, 0xce });
-    try write_partition(io_writer, 2, null);
+    try write_partition(io_writer, 0, null);
+    try write_partition(io_writer, 1, &[_]u8{ 0xde, 0xad, 0xbe, 0xef });
+    try write_partition(io_writer, 2, &[_]u8{ 0xfe, 0xed, 0xfa, 0xce });
 }
 
 /// Writes a partition_data header followed by its record bytes, deriving

From e1e91bafa758dd4bc2de5f6a040c282dc796ad69 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 2 Jun 2026 14:43:42 +0000
Subject: [PATCH 10/13] Make compact_size a generic, exposed helper over the
 slice

Per review: expose compact_size and have it take the slice (generic over
the element type T) rather than a precomputed count, so the client can use
it to write a partition's records length.

writer.compact_size(comptime T, ?[]const T) writes the slice's length as
the COMPACT_NULLABLE prefix (or the 0 null marker); reader.compact_size is
exposed alongside it. The records length prefix moves out of partition_data
into compact_size: PartitionData is now just { index }, write/read
partition_data handle only the index, and the records prefix + bytes are
written/consumed separately (client write_partition, server loop). Topic
and partition array counts, which aren't materialized as slices, write
their length via unsigned_varint(+1) like produce_req already does.

https://claude.ai/code/session_01Sgm6nVAriGJDHANn7yLTnK
---
 client.zig   | 12 +++----
 klog.zig     | 11 +++---
 protocol.zig | 98 ++++++++++++++++++++++++----------------------------
 3 files changed, 58 insertions(+), 63 deletions(-)

diff --git a/client.zig b/client.zig
index 3f9c716..27b8e62 100644
--- a/client.zig
+++ b/client.zig
@@ -49,12 +49,12 @@ pub fn main(init: std.process.Init) !void {
     try write_partition(io_writer, 2, &[_]u8{ 0xfe, 0xed, 0xfa, 0xce });
 }
 
-/// Writes a partition_data header followed by its record bytes, deriving
-/// the records size from the slice so the two always agree.
+/// Writes a partition_data entry: the index, then its records as a
+/// COMPACT_NULLABLE byte sequence (length prefix via compact_size, then the
+/// bytes). compact_size derives the length from the slice, so the prefix and
+/// the bytes always agree.
 fn write_partition(w: *std.Io.Writer, index: i32, records: ?[]const u8) !void {
-    try proto.writer.partition_data(w, .{
-        .index = index,
-        .records_size = if (records) |r| r.len else null,
-    });
+    try proto.writer.partition_data(w, .{ .index = index });
+    try proto.writer.compact_size(w, u8, records);
     if (records) |r| try w.writeAll(r);
 }
diff --git a/klog.zig b/klog.zig
index 1711bc9..fc986e7 100644
--- a/klog.zig
+++ b/klog.zig
@@ -66,7 +66,8 @@ fn produce(io: Io, io_reader: *Io.Reader, stream: Io.net.Stream) !void {
         log.info("topic {x}: {d} partition(s)", .{ topic.topic_id, topic.partition_count });
         for (0..topic.partition_count) |_| {
             const partition = try reader.partition_data(io_reader);
-            try log_records(io_reader, partition);
+            const records_size = try reader.compact_size(io_reader);
+            try log_records(io_reader, partition.index, records_size);
         }
     }
 }
@@ -75,9 +76,9 @@ fn produce(io: Io, io_reader: *Io.Reader, stream: Io.net.Stream) !void {
 /// fixed-size chunks so an arbitrarily large records blob never has to be
 /// held in memory at once. Logs the byte count and a short hex preview of
 /// the first chunk.
-fn log_records(io_reader: *Io.Reader, partition: proto.PartitionData) !void {
-    const size = partition.records_size orelse {
-        log.info("  partition {d}: null records", .{partition.index});
+fn log_records(io_reader: *Io.Reader, index: i32, records_size: ?u64) !void {
+    const size = records_size orelse {
+        log.info("  partition {d}: null records", .{index});
         return;
     };
 
@@ -86,7 +87,7 @@ fn log_records(io_reader: *Io.Reader, partition: proto.PartitionData) !void {
     try io_reader.readSliceAll(chunk[0..first_len]);
     const preview = chunk[0..@min(first_len, 16)];
     log.info("  partition {d}: {d} record byte(s), first {d}: {x}", .{
-        partition.index,
+        index,
         size,
         preview.len,
         preview,
diff --git a/protocol.zig b/protocol.zig
index e034708..a3fdc7e 100644
--- a/protocol.zig
+++ b/protocol.zig
@@ -70,16 +70,15 @@ pub const TopicData = struct {
     partition_count: u64,
 };
 
-/// The fixed part of a partition_data entry. `records_size` is the byte
-/// length of the COMPACT_NULLABLE_RECORDS that follow (null for null
-/// records); those bytes live outside this struct and are written/consumed
-/// by the caller rather than buffered here.
+/// The fixed part of a partition_data entry: just the partition index. The
+/// records that follow are a COMPACT_NULLABLE byte sequence whose length is
+/// read/written with `compact_size`, and whose bytes live outside this
+/// struct, consumed by the caller.
 ///
 /// index => INT32
 /// records => COMPACT_NULLABLE_RECORDS
 pub const PartitionData = struct {
     index: i32,
-    records_size: ?u64,
 };
 
 pub const Record = struct {
@@ -221,16 +220,12 @@ pub const reader = struct {
         };
     }
 
-    /// Reads one partition_data entry's fixed part: the partition index
-    /// followed by the COMPACT_NULLABLE_RECORDS length. The caller then
-    /// consumes `records_size` bytes of records (or none when null) before
-    /// reading the next partition.
+    /// Reads one partition_data entry's fixed part: the partition index.
+    /// The caller then reads the records length with `compact_size` and
+    /// consumes that many record bytes (or none when null) before reading
+    /// the next partition.
     pub fn partition_data(r: *Io.Reader) !PartitionData {
-        return .{
-            .index = try r.takeInt(i32, BE),
-            // records => COMPACT_NULLABLE_RECORDS: null when the length is 0.
-            .records_size = try compact_size(r),
-        };
+        return .{ .index = try r.takeInt(i32, BE) };
     }
 
     /// NULLABLE_STRING
@@ -259,14 +254,14 @@ pub const reader = struct {
         return buf[0..n];
     }
 
-    /// Reads the length N from a COMPACT array, encoded as N + 1 in an
+    /// Reads the length N of a COMPACT array, encoded as N + 1 in an
     /// UNSIGNED_VARINT, or null when the length is 0 — the COMPACT_NULLABLE
     /// null marker. The same length-prefix scheme is used for compact byte
-    /// sequences (e.g. records), so this applies to both. In protocol
-    /// documentation a compact array of T instances is referred to as (T).
-    /// Mirrors the writer's `compact_size(?u64)`; non-nullable callers
+    /// sequences (e.g. partition records), so this applies to both. In
+    /// protocol documentation a compact array of T instances is referred to
+    /// as (T). Mirrors the writer's `compact_size`; non-nullable callers
     /// reject the null marker.
-    fn compact_size(r: *Io.Reader) !?u64 {
+    pub fn compact_size(r: *Io.Reader) !?u64 {
         const len = try unsigned_varint(r);
         if (len == 0) return null;
         return len - 1;
@@ -314,17 +309,17 @@ pub const writer = struct {
     /// length itself is written by `produce_req` (topic_data_size).
     pub fn topic_data(w: *Io.Writer, topic: TopicData) !void {
         try w.writeAll(&topic.topic_id);
-        try compact_size(w, topic.partition_count);
+        // partition_data is a non-nullable COMPACT array; write its length
+        // directly as N+1, the same way produce_req writes topic_data_size.
+        try unsigned_varint(w, topic.partition_count + 1);
     }
 
-    /// Writes one partition_data entry's fixed part: the index followed by
-    /// the COMPACT_NULLABLE_RECORDS length prefix (N+1, or 0 for null). The
-    /// caller then writes the `records_size` record bytes itself, mirroring
-    /// how `reader.partition_data` returns the size and leaves the bytes in
-    /// the stream for the caller to consume.
+    /// Writes one partition_data entry's fixed part: the partition index.
+    /// The caller then writes the records with `compact_size` (the length
+    /// prefix) followed by the bytes, mirroring how `reader.partition_data`
+    /// reads the index and leaves the records for the caller.
     pub fn partition_data(w: *Io.Writer, partition: PartitionData) !void {
         try w.writeInt(i32, partition.index, BE);
-        try compact_size(w, partition.records_size);
     }
 
     fn nullable_str(w: *Io.Writer, str: ?[]const u8) !void {
@@ -338,22 +333,18 @@ pub const writer = struct {
     }
 
     fn compact_nullable_str(w: *Io.Writer, str: ?[]const u8) !void {
-        if (str) |s| {
-            try compact_size(w, s.len);
-            try w.writeAll(s);
-        } else {
-            try compact_size(w, null);
-        }
+        try compact_size(w, u8, str);
+        if (str) |s| try w.writeAll(s);
     }
 
-    /// Writes a compact length N as N+1 in an unsigned varint, or 0 for a
-    /// null value. The N bytes/elements follow for non-null values. The
-    /// produce request's COMPACT arrays — topic_data and partition_data —
-    /// are non-nullable and always pass a length; nullable compact strings
-    /// and records pass null to encode their null form.
-    fn compact_size(w: *Io.Writer, size: ?u64) !void {
-        if (size) |s| {
-            try unsigned_varint(w, s + 1);
+    /// Writes the length N of a COMPACT array as N+1 in an UNSIGNED_VARINT,
+    /// or 0 for a null array (the COMPACT_NULLABLE null marker). Generic
+    /// over the element type T: pass the slice itself — e.g. a partition's
+    /// records bytes or a compact string — rather than a precomputed count.
+    /// The caller writes the N elements/bytes afterwards for non-null values.
+    pub fn compact_size(w: *Io.Writer, comptime T: type, arr: ?[]const T) !void {
+        if (arr) |a| {
+            try unsigned_varint(w, a.len + 1);
         } else {
             try unsigned_varint(w, 0);
         }
@@ -503,33 +494,35 @@ test "topic_data round-trip with no partitions" {
     try testing.expectEqual(0, read.partition_count);
 }
 
-test "partition_data round-trip with records" {
+test "partition round-trip with records" {
     var buf: [16]u8 = undefined;
     var w = Io.Writer.fixed(&buf);
     const records = [_]u8{ 0xde, 0xad, 0xbe };
-    // The header carries only the size; the bytes are written separately.
-    try writer.partition_data(&w, .{ .index = 1, .records_size = records.len });
+    // index, then the records as a length prefix followed by the bytes.
+    try writer.partition_data(&w, .{ .index = 1 });
+    try writer.compact_size(&w, u8, &records);
     try w.writeAll(&records);
 
     var r = Io.Reader.fixed(&buf);
     const read = try reader.partition_data(&r);
     try testing.expectEqual(1, read.index);
-    try testing.expectEqual(records.len, read.records_size);
+    try testing.expectEqual(records.len, try reader.compact_size(&r));
     // The record bytes are left in the stream for the caller to consume.
     var out: [3]u8 = undefined;
     try r.readSliceAll(&out);
     try testing.expectEqualSlices(u8, &records, &out);
 }
 
-test "partition_data round-trip with null records" {
+test "partition round-trip with null records" {
     var buf: [16]u8 = undefined;
     var w = Io.Writer.fixed(&buf);
-    try writer.partition_data(&w, .{ .index = 2, .records_size = null });
+    try writer.partition_data(&w, .{ .index = 2 });
+    try writer.compact_size(&w, u8, null);
 
     var r = Io.Reader.fixed(&buf);
     const read = try reader.partition_data(&r);
     try testing.expectEqual(2, read.index);
-    try testing.expectEqual(null, read.records_size);
+    try testing.expectEqual(null, try reader.compact_size(&r));
 }
 
 test "nullable_str round-trip non-null" {
@@ -637,17 +630,18 @@ test "compact_size round-trip" {
     var buf: [16]u8 = undefined;
     var w = Io.Writer.fixed(&buf);
 
-    const size: u64 = 7;
-    try writer.compact_size(&w, size);
+    // compact_size takes the slice and writes its length; here that is 7.
+    const arr = [_]u8{0} ** 7;
+    try writer.compact_size(&w, u8, &arr);
 
     var r = Io.Reader.fixed(&buf);
-    try testing.expectEqual(size, try reader.compact_size(&r));
+    try testing.expectEqual(arr.len, try reader.compact_size(&r));
 }
 
 test "compact_size round-trip null" {
     var buf: [16]u8 = undefined;
     var w = Io.Writer.fixed(&buf);
-    try writer.compact_size(&w, null);
+    try writer.compact_size(&w, u8, null);
 
     var r = Io.Reader.fixed(&buf);
     try testing.expectEqual(null, try reader.compact_size(&r));
@@ -656,7 +650,7 @@ test "compact_size round-trip null" {
 test "compact_size encodes null as zero" {
     var buf: [16]u8 = undefined;
     var w = Io.Writer.fixed(&buf);
-    try writer.compact_size(&w, null);
+    try writer.compact_size(&w, u8, null);
     try testing.expectEqualSlices(u8, &[_]u8{0x00}, w.buffered());
 }
 

From 77dce96e9e28fbc656ddbb1971c707f6bf0b19a4 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 2 Jun 2026 15:15:07 +0000
Subject: [PATCH 11/13] Revert compact_size to a byte-slice signature

Drop the comptime element-type parameter; the writer's compact_size
takes ?[]const u8 again.
---
 client.zig   |  2 +-
 protocol.zig | 22 +++++++++++-----------
 2 files changed, 12 insertions(+), 12 deletions(-)

diff --git a/client.zig b/client.zig
index 27b8e62..6516168 100644
--- a/client.zig
+++ b/client.zig
@@ -55,6 +55,6 @@ pub fn main(init: std.process.Init) !void {
 /// the bytes always agree.
 fn write_partition(w: *std.Io.Writer, index: i32, records: ?[]const u8) !void {
     try proto.writer.partition_data(w, .{ .index = index });
-    try proto.writer.compact_size(w, u8, records);
+    try proto.writer.compact_size(w, records);
     if (records) |r| try w.writeAll(r);
 }
diff --git a/protocol.zig b/protocol.zig
index a3fdc7e..ce6a601 100644
--- a/protocol.zig
+++ b/protocol.zig
@@ -333,16 +333,16 @@ pub const writer = struct {
     }
 
     fn compact_nullable_str(w: *Io.Writer, str: ?[]const u8) !void {
-        try compact_size(w, u8, str);
+        try compact_size(w, str);
         if (str) |s| try w.writeAll(s);
     }
 
     /// Writes the length N of a COMPACT array as N+1 in an UNSIGNED_VARINT,
-    /// or 0 for a null array (the COMPACT_NULLABLE null marker). Generic
-    /// over the element type T: pass the slice itself — e.g. a partition's
-    /// records bytes or a compact string — rather than a precomputed count.
-    /// The caller writes the N elements/bytes afterwards for non-null values.
-    pub fn compact_size(w: *Io.Writer, comptime T: type, arr: ?[]const T) !void {
+    /// or 0 for a null array (the COMPACT_NULLABLE null marker). Pass the
+    /// slice itself — e.g. a partition's records bytes or a compact string —
+    /// rather than a precomputed count. The caller writes the N bytes
+    /// afterwards for non-null values.
+    pub fn compact_size(w: *Io.Writer, arr: ?[]const u8) !void {
         if (arr) |a| {
             try unsigned_varint(w, a.len + 1);
         } else {
@@ -500,7 +500,7 @@ test "partition round-trip with records" {
     const records = [_]u8{ 0xde, 0xad, 0xbe };
     // index, then the records as a length prefix followed by the bytes.
     try writer.partition_data(&w, .{ .index = 1 });
-    try writer.compact_size(&w, u8, &records);
+    try writer.compact_size(&w, &records);
     try w.writeAll(&records);
 
     var r = Io.Reader.fixed(&buf);
@@ -517,7 +517,7 @@ test "partition round-trip with null records" {
     var buf: [16]u8 = undefined;
     var w = Io.Writer.fixed(&buf);
     try writer.partition_data(&w, .{ .index = 2 });
-    try writer.compact_size(&w, u8, null);
+    try writer.compact_size(&w, null);
 
     var r = Io.Reader.fixed(&buf);
     const read = try reader.partition_data(&r);
@@ -632,7 +632,7 @@ test "compact_size round-trip" {
 
     // compact_size takes the slice and writes its length; here that is 7.
     const arr = [_]u8{0} ** 7;
-    try writer.compact_size(&w, u8, &arr);
+    try writer.compact_size(&w, &arr);
 
     var r = Io.Reader.fixed(&buf);
     try testing.expectEqual(arr.len, try reader.compact_size(&r));
@@ -641,7 +641,7 @@ test "compact_size round-trip" {
 test "compact_size round-trip null" {
     var buf: [16]u8 = undefined;
     var w = Io.Writer.fixed(&buf);
-    try writer.compact_size(&w, u8, null);
+    try writer.compact_size(&w, null);
 
     var r = Io.Reader.fixed(&buf);
     try testing.expectEqual(null, try reader.compact_size(&r));
@@ -650,7 +650,7 @@ test "compact_size round-trip null" {
 test "compact_size encodes null as zero" {
     var buf: [16]u8 = undefined;
     var w = Io.Writer.fixed(&buf);
-    try writer.compact_size(&w, u8, null);
+    try writer.compact_size(&w, null);
     try testing.expectEqualSlices(u8, &[_]u8{0x00}, w.buffered());
 }
 

From dfea8ee1d3391214bc1dfdaa005699de2c32b9c4 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 2 Jun 2026 20:30:46 +0000
Subject: [PATCH 12/13] Carry array/records sizes in every struct, written with
 the struct

Follow the ProduceRequest pattern uniformly: each struct holds its
trailing array's size and writes it when the struct is written, with the
elements/bytes streamed after.

- TopicData.partition_count -> partition_data_size (size suffix).
- PartitionData regains records_size (?u64); partition_data writes the
  index and the COMPACT_NULLABLE records length prefix, and the caller
  writes the record bytes after.

The writer's compact_size is now used only for compact strings.
---
 client.zig   | 16 ++++++------
 klog.zig     |  7 +++--
 protocol.zig | 73 +++++++++++++++++++++++++++++-----------------------
 3 files changed, 52 insertions(+), 44 deletions(-)

diff --git a/client.zig b/client.zig
index 6516168..d255b74 100644
--- a/client.zig
+++ b/client.zig
@@ -40,21 +40,21 @@ pub fn main(init: std.process.Init) !void {
     const topic1_uuid = [_]u8{0x11} ** 16;
     const topic2_uuid = [_]u8{0x22} ** 16;
 
-    try writer.topic_data(io_writer, .{ .topic_id = topic1_uuid, .partition_count = 1 });
+    try writer.topic_data(io_writer, .{ .topic_id = topic1_uuid, .partition_data_size = 1 });
     try write_partition(io_writer, 0, &[_]u8{ 0xca, 0xfe, 0xba, 0xbe });
 
-    try writer.topic_data(io_writer, .{ .topic_id = topic2_uuid, .partition_count = 3 });
+    try writer.topic_data(io_writer, .{ .topic_id = topic2_uuid, .partition_data_size = 3 });
     try write_partition(io_writer, 0, null);
     try write_partition(io_writer, 1, &[_]u8{ 0xde, 0xad, 0xbe, 0xef });
     try write_partition(io_writer, 2, &[_]u8{ 0xfe, 0xed, 0xfa, 0xce });
 }
 
-/// Writes a partition_data entry: the index, then its records as a
-/// COMPACT_NULLABLE byte sequence (length prefix via compact_size, then the
-/// bytes). compact_size derives the length from the slice, so the prefix and
-/// the bytes always agree.
+/// Writes a partition_data entry: the index and records length prefix (both
+/// written by partition_data, the size taken from the slice), then the record
+/// bytes. The bytes live outside the struct, so the size and the bytes are
+/// written from the same slice and always agree.
 fn write_partition(w: *std.Io.Writer, index: i32, records: ?[]const u8) !void {
-    try proto.writer.partition_data(w, .{ .index = index });
-    try proto.writer.compact_size(w, records);
+    const records_size: ?u64 = if (records) |r| @intCast(r.len) else null;
+    try proto.writer.partition_data(w, .{ .index = index, .records_size = records_size });
     if (records) |r| try w.writeAll(r);
 }
diff --git a/klog.zig b/klog.zig
index fc986e7..d9562b1 100644
--- a/klog.zig
+++ b/klog.zig
@@ -63,11 +63,10 @@ fn produce(io: Io, io_reader: *Io.Reader, stream: Io.net.Stream) !void {
     // sizes that prefix each level rather than buffering the whole request.
     for (0..req.topic_data_size) |_| {
         const topic = try reader.topic_data(io_reader);
-        log.info("topic {x}: {d} partition(s)", .{ topic.topic_id, topic.partition_count });
-        for (0..topic.partition_count) |_| {
+        log.info("topic {x}: {d} partition(s)", .{ topic.topic_id, topic.partition_data_size });
+        for (0..topic.partition_data_size) |_| {
             const partition = try reader.partition_data(io_reader);
-            const records_size = try reader.compact_size(io_reader);
-            try log_records(io_reader, partition.index, records_size);
+            try log_records(io_reader, partition.index, partition.records_size);
         }
     }
 }
diff --git a/protocol.zig b/protocol.zig
index ce6a601..4712d03 100644
--- a/protocol.zig
+++ b/protocol.zig
@@ -67,18 +67,21 @@ pub const ProduceRequest = struct {
 pub const TopicData = struct {
     topic_id: [16]u8, // UUID
     /// The number of partition_data entries that follow.
-    partition_count: u64,
+    partition_data_size: u64,
 };
 
-/// The fixed part of a partition_data entry: just the partition index. The
-/// records that follow are a COMPACT_NULLABLE byte sequence whose length is
-/// read/written with `compact_size`, and whose bytes live outside this
-/// struct, consumed by the caller.
+/// The fixed part of a partition_data entry: the partition index and the
+/// size of the records that follow. The records are a COMPACT_NULLABLE byte
+/// sequence whose bytes live outside this struct, consumed by the caller
+/// after the entry is read or written.
 ///
 /// index => INT32
 /// records => COMPACT_NULLABLE_RECORDS
 pub const PartitionData = struct {
     index: i32,
+    /// The number of record bytes that follow, or null for the
+    /// COMPACT_NULLABLE null marker.
+    records_size: ?u64,
 };
 
 pub const Record = struct {
@@ -208,7 +211,7 @@ pub const reader = struct {
 
     /// Reads one topic_data entry's fixed part: the topic_id UUID followed
     /// by the partition_data array length. The caller then reads
-    /// `partition_count` partitions with `partition_data`. There are
+    /// `partition_data_size` partitions with `partition_data`. There are
     /// `ProduceRequest.topic_data_size` of these entries in a request.
     pub fn topic_data(r: *Io.Reader) !TopicData {
         var topic_id: [16]u8 = undefined;
@@ -216,16 +219,18 @@ pub const reader = struct {
         return .{
             .topic_id = topic_id,
             // partition_data is a non-nullable COMPACT array; reject the null marker.
-            .partition_count = (try compact_size(r)) orelse return Error.ProtocolError,
+            .partition_data_size = (try compact_size(r)) orelse return Error.ProtocolError,
         };
     }
 
-    /// Reads one partition_data entry's fixed part: the partition index.
-    /// The caller then reads the records length with `compact_size` and
-    /// consumes that many record bytes (or none when null) before reading
-    /// the next partition.
+    /// Reads one partition_data entry's fixed part: the partition index and
+    /// the records length. The caller then consumes `records_size` record
+    /// bytes (or none when null) before reading the next partition.
     pub fn partition_data(r: *Io.Reader) !PartitionData {
-        return .{ .index = try r.takeInt(i32, BE) };
+        return .{
+            .index = try r.takeInt(i32, BE),
+            .records_size = try compact_size(r),
+        };
     }
 
     /// NULLABLE_STRING
@@ -304,22 +309,29 @@ pub const writer = struct {
     }
 
     /// Writes one topic_data entry's fixed part: the topic_id UUID followed
-    /// by the partition_data array length. Each of the `partition_count`
+    /// by the partition_data array length. Each of the `partition_data_size`
     /// partitions is then written with `partition_data`. The topic array
     /// length itself is written by `produce_req` (topic_data_size).
     pub fn topic_data(w: *Io.Writer, topic: TopicData) !void {
         try w.writeAll(&topic.topic_id);
         // partition_data is a non-nullable COMPACT array; write its length
         // directly as N+1, the same way produce_req writes topic_data_size.
-        try unsigned_varint(w, topic.partition_count + 1);
+        try unsigned_varint(w, topic.partition_data_size + 1);
     }
 
-    /// Writes one partition_data entry's fixed part: the partition index.
-    /// The caller then writes the records with `compact_size` (the length
-    /// prefix) followed by the bytes, mirroring how `reader.partition_data`
-    /// reads the index and leaves the records for the caller.
+    /// Writes one partition_data entry's fixed part: the partition index and
+    /// the records length prefix. The caller then writes the `records_size`
+    /// record bytes (none when null), mirroring how `reader.partition_data`
+    /// reads the index and length and leaves the records for the caller.
     pub fn partition_data(w: *Io.Writer, partition: PartitionData) !void {
         try w.writeInt(i32, partition.index, BE);
+        // records is a COMPACT_NULLABLE byte sequence; write its length as
+        // N+1, or 0 for the null marker. The bytes follow, written by the caller.
+        if (partition.records_size) |n| {
+            try unsigned_varint(w, n + 1);
+        } else {
+            try unsigned_varint(w, 0);
+        }
     }
 
     fn nullable_str(w: *Io.Writer, str: ?[]const u8) !void {
@@ -339,9 +351,8 @@ pub const writer = struct {
 
     /// Writes the length N of a COMPACT array as N+1 in an UNSIGNED_VARINT,
     /// or 0 for a null array (the COMPACT_NULLABLE null marker). Pass the
-    /// slice itself — e.g. a partition's records bytes or a compact string —
-    /// rather than a precomputed count. The caller writes the N bytes
-    /// afterwards for non-null values.
+    /// slice itself — e.g. a compact string — rather than a precomputed
+    /// count. The caller writes the N bytes afterwards for non-null values.
     pub fn compact_size(w: *Io.Writer, arr: ?[]const u8) !void {
         if (arr) |a| {
             try unsigned_varint(w, a.len + 1);
@@ -474,39 +485,38 @@ test "topic_data round-trip" {
     var buf: [32]u8 = undefined;
     var w = Io.Writer.fixed(&buf);
     const topic_id = [16]u8{ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15 };
-    try writer.topic_data(&w, .{ .topic_id = topic_id, .partition_count = 2 });
+    try writer.topic_data(&w, .{ .topic_id = topic_id, .partition_data_size = 2 });
 
     var r = Io.Reader.fixed(&buf);
     const read = try reader.topic_data(&r);
     try testing.expectEqualSlices(u8, &topic_id, &read.topic_id);
-    try testing.expectEqual(2, read.partition_count);
+    try testing.expectEqual(2, read.partition_data_size);
 }
 
 test "topic_data round-trip with no partitions" {
     var buf: [32]u8 = undefined;
     var w = Io.Writer.fixed(&buf);
     const topic_id = [_]u8{0xab} ** 16;
-    try writer.topic_data(&w, .{ .topic_id = topic_id, .partition_count = 0 });
+    try writer.topic_data(&w, .{ .topic_id = topic_id, .partition_data_size = 0 });
 
     var r = Io.Reader.fixed(&buf);
     const read = try reader.topic_data(&r);
     try testing.expectEqualSlices(u8, &topic_id, &read.topic_id);
-    try testing.expectEqual(0, read.partition_count);
+    try testing.expectEqual(0, read.partition_data_size);
 }
 
 test "partition round-trip with records" {
     var buf: [16]u8 = undefined;
     var w = Io.Writer.fixed(&buf);
     const records = [_]u8{ 0xde, 0xad, 0xbe };
-    // index, then the records as a length prefix followed by the bytes.
-    try writer.partition_data(&w, .{ .index = 1 });
-    try writer.compact_size(&w, &records);
+    // index and records length, then the record bytes.
+    try writer.partition_data(&w, .{ .index = 1, .records_size = records.len });
     try w.writeAll(&records);
 
     var r = Io.Reader.fixed(&buf);
     const read = try reader.partition_data(&r);
     try testing.expectEqual(1, read.index);
-    try testing.expectEqual(records.len, try reader.compact_size(&r));
+    try testing.expectEqual(records.len, read.records_size.?);
     // The record bytes are left in the stream for the caller to consume.
     var out: [3]u8 = undefined;
     try r.readSliceAll(&out);
@@ -516,13 +526,12 @@ test "partition round-trip with records" {
 test "partition round-trip with null records" {
     var buf: [16]u8 = undefined;
     var w = Io.Writer.fixed(&buf);
-    try writer.partition_data(&w, .{ .index = 2 });
-    try writer.compact_size(&w, null);
+    try writer.partition_data(&w, .{ .index = 2, .records_size = null });
 
     var r = Io.Reader.fixed(&buf);
     const read = try reader.partition_data(&r);
     try testing.expectEqual(2, read.index);
-    try testing.expectEqual(null, try reader.compact_size(&r));
+    try testing.expectEqual(null, read.records_size);
 }
 
 test "nullable_str round-trip non-null" {

From fe188904c772188cf53c03da5bdc1e408738e20b Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 2 Jun 2026 20:43:55 +0000
Subject: [PATCH 13/13] Split compact_size into a count primitive and a slice
 wrapper

The writer now mirrors the reader: compact_size deals in ?u64 (writing
N+1, or 0 for the null marker), the exact inverse of reader.compact_size.
compact_size_of is a thin wrapper that derives the length from a slice
and defers to it.

This removes the duplicated +1/null-marker encoding: produce_req,
topic_data, and partition_data all write their size fields through the
?u64 primitive (non-nullable counts coerce to the optional), while
compact strings use the slice wrapper.
---
 protocol.zig | 55 +++++++++++++++++++++++++++++++++-------------------
 1 file changed, 35 insertions(+), 20 deletions(-)

diff --git a/protocol.zig b/protocol.zig
index 4712d03..d500455 100644
--- a/protocol.zig
+++ b/protocol.zig
@@ -305,7 +305,8 @@ pub const writer = struct {
         try compact_nullable_str(w, req.transactional_id);
         try w.writeInt(i16, req.acks, BE);
         try w.writeInt(i32, req.timeout_ms, BE);
-        try unsigned_varint(w, req.topic_data_size + 1);
+        // topic_data is a non-nullable COMPACT array; the count writes as N+1.
+        try compact_size(w, req.topic_data_size);
     }
 
     /// Writes one topic_data entry's fixed part: the topic_id UUID followed
@@ -314,9 +315,8 @@ pub const writer = struct {
     /// length itself is written by `produce_req` (topic_data_size).
     pub fn topic_data(w: *Io.Writer, topic: TopicData) !void {
         try w.writeAll(&topic.topic_id);
-        // partition_data is a non-nullable COMPACT array; write its length
-        // directly as N+1, the same way produce_req writes topic_data_size.
-        try unsigned_varint(w, topic.partition_data_size + 1);
+        // partition_data is a non-nullable COMPACT array, like topic_data_size.
+        try compact_size(w, topic.partition_data_size);
     }
 
     /// Writes one partition_data entry's fixed part: the partition index and
@@ -325,13 +325,9 @@ pub const writer = struct {
     /// reads the index and length and leaves the records for the caller.
     pub fn partition_data(w: *Io.Writer, partition: PartitionData) !void {
         try w.writeInt(i32, partition.index, BE);
-        // records is a COMPACT_NULLABLE byte sequence; write its length as
-        // N+1, or 0 for the null marker. The bytes follow, written by the caller.
-        if (partition.records_size) |n| {
-            try unsigned_varint(w, n + 1);
-        } else {
-            try unsigned_varint(w, 0);
-        }
+        // records is a COMPACT_NULLABLE byte sequence; its bytes follow,
+        // written by the caller.
+        try compact_size(w, partition.records_size);
     }
 
     fn nullable_str(w: *Io.Writer, str: ?[]const u8) !void {
@@ -345,22 +341,32 @@ pub const writer = struct {
     }
 
     fn compact_nullable_str(w: *Io.Writer, str: ?[]const u8) !void {
-        try compact_size(w, str);
+        try compact_size_of(w, str);
         if (str) |s| try w.writeAll(s);
     }
 
     /// Writes the length N of a COMPACT array as N+1 in an UNSIGNED_VARINT,
-    /// or 0 for a null array (the COMPACT_NULLABLE null marker). Pass the
-    /// slice itself — e.g. a compact string — rather than a precomputed
-    /// count. The caller writes the N bytes afterwards for non-null values.
-    pub fn compact_size(w: *Io.Writer, arr: ?[]const u8) !void {
-        if (arr) |a| {
-            try unsigned_varint(w, a.len + 1);
+    /// or 0 for null — the COMPACT_NULLABLE null marker. The inverse of the
+    /// reader's `compact_size`, which turns the same encoding back into
+    /// `?u64`. Non-nullable callers pass the count directly (it coerces to
+    /// the optional and never writes the null marker). The caller writes the
+    /// N elements/bytes afterwards.
+    pub fn compact_size(w: *Io.Writer, len: ?u64) !void {
+        if (len) |n| {
+            try unsigned_varint(w, n + 1);
         } else {
             try unsigned_varint(w, 0);
         }
     }
 
+    /// Writes the COMPACT_NULLABLE length prefix of a (possibly null) slice,
+    /// deferring to `compact_size`. Pass the slice itself — e.g. a compact
+    /// string or a partition's records — rather than a precomputed count.
+    pub fn compact_size_of(w: *Io.Writer, arr: ?[]const u8) !void {
+        const len: ?u64 = if (arr) |a| @intCast(a.len) else null;
+        try compact_size(w, len);
+    }
+
     /// Unsigned LEB128, as used by Protocol Buffers.
     /// https://protobuf.dev/programming-guides/encoding/#varints
     fn unsigned_varint(w: *Io.Writer, v: u64) !void {
@@ -638,10 +644,19 @@ test "compact_nullable_str rejects too-small buffer" {
 test "compact_size round-trip" {
     var buf: [16]u8 = undefined;
     var w = Io.Writer.fixed(&buf);
+    try writer.compact_size(&w, 7);
+
+    var r = Io.Reader.fixed(&buf);
+    try testing.expectEqual(7, try reader.compact_size(&r));
+}
+
+test "compact_size_of round-trip" {
+    var buf: [16]u8 = undefined;
+    var w = Io.Writer.fixed(&buf);
 
-    // compact_size takes the slice and writes its length; here that is 7.
+    // compact_size_of takes the slice and writes its length; here that is 7.
     const arr = [_]u8{0} ** 7;
-    try writer.compact_size(&w, &arr);
+    try writer.compact_size_of(&w, &arr);
 
     var r = Io.Reader.fixed(&buf);
     try testing.expectEqual(arr.len, try reader.compact_size(&r));