Unify ObjectResult and ObjectReader for cleaner API and zero-copy streaming #117
Conversation
…eaming This commit refactors the ObjectStore implementation to provide a cleaner, more efficient API by merging ObjectReader functionality directly into ObjectResult. Key improvements: - Unified API: Users now work with ObjectResult directly instead of accessing .reader - Zero-copy streaming: Eliminated buffer copying by holding JetStream message pointers - Reduced coupling: Removed ObjectStore dependency from ObjectResult - Simplified verification: Removed cached verification state for honest verify() calls - Better resource management: Direct message lifecycle management without wrapper overhead Technical changes: - Merged ObjectReader fields and methods into ObjectResult - Replaced ArrayList buffer with current_msg pointer and msg_pos offset - Updated timeout access to use subscription.js.nc.options.timeout_ms - Removed ObjectReader from public API exports - Updated test code to use unified API Performance benefits: - Zero memory copies for streaming data - Reduced memory footprint (no duplicate data storage) - More efficient resource utilization All 223 tests pass, confirming functionality is preserved while achieving significant architectural and performance improvements.
WalkthroughRefactors object store streaming: replaces ObjectReader with ObjectResult, switches from buffered reads to message-based, arena-managed zero-copy streaming, revises init/read/verify/deinit APIs, updates ObjectStore.get/getBytes to use ObjectResult, removes ObjectReader export from root, and adjusts tests to call read/verify on ObjectResult. Changes
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Pre-merge checks and finishing touches✅ Passed checks (3 passed)
✨ Finishing touches🧪 Generate unit tests
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (7)
tests/jetstream_objstore_streaming_test.zig (1)
125-126: Prefer slice over pointer-to-array when passing buffersIdiomatic Zig here is to pass a slice, not
&buffer. While this may compile via coercions, the slice form is clearer.Apply:
- const n = try get_result.read(&buffer); + const n = try get_result.read(buffer[0..]);src/jetstream_objstore.zig (6)
118-130: Fieldallocatorappears unused and ambiguous
allocatoris stored but never read in this type. Also, its meaning is ambiguous vs the arena’s owner allocator. Either remove it or rename it toowner_allocatorand use it indeinit.Apply either:
- Removal:
pub const ObjectResult = struct { info: ObjectInfo, - arena: *std.heap.ArenaAllocator, + arena: *std.heap.ArenaAllocator, // Reading state (zero-copy streaming) subscription: ?*JetStreamSubscription, current_msg: ?*JetStreamMessage, msg_pos: usize, chunk_index: u32, digest: std.crypto.hash.sha2.Sha256, eof: bool, - allocator: std.mem.Allocator, + // allocator removed
- Or rename/repurpose (see deinit comment below).
131-143: Init: pass and store the owner allocator instead (or drop the field)If you keep an allocator in
ObjectResult, it should be the allocator that allocatedarena(the “owner”), not the arena’s allocator. Otherwisedeinitcan’t safely destroy the arena pointer without relying onchild_allocator.Example:
- pub fn init(allocator: std.mem.Allocator, info: ObjectInfo, subscription: ?*JetStreamSubscription, arena: *std.heap.ArenaAllocator) ObjectResult { + pub fn init(owner_allocator: std.mem.Allocator, info: ObjectInfo, subscription: ?*JetStreamSubscription, arena: *std.heap.ArenaAllocator) ObjectResult { return ObjectResult{ .info = info, .arena = arena, .subscription = subscription, .current_msg = null, .msg_pos = 0, .chunk_index = 0, .digest = std.crypto.hash.sha2.Sha256.init(.{}), .eof = false, - .allocator = allocator, + .allocator = owner_allocator, }; }And pass
self.allocator(the owner) at call sites.
145-155: Deinit: destroy arena with the owner allocator (avoid relying onchild_allocator)Destroying the
ArenaAllocatorstruct viachild_allocator.destroy(self.arena)assumes the arena’s child allocator equals the allocator used to allocate the struct. That’s true in your current call sites but brittle.If you keep
owner_allocatoras suggested above:- const child_allocator = self.arena.child_allocator; - self.arena.deinit(); - child_allocator.destroy(self.arena); + const owner = self.allocator; + self.arena.deinit(); + owner.destroy(self.arena);If you drop the stored allocator entirely, ensure all call sites always allocate
arenawith the same allocator used as the arena’s child, or pass the owner allocator intodeinit.
158-169: Minor: early-outsConsider handling
dest.len == 0up front to avoid work and preserve semantics identical to std readers.pub fn read(self: *ObjectResult, dest: []u8) !usize { + if (dest.len == 0) return 0; if (self.eof) return 0; // Early return if no subscription (empty objects) const sub = self.subscription orelse return 0;
171-187: Read: timeout surface vs. chunk expectationsYou increment
chunk_indexupon fetching a message (before copying), and you returnChunkMismatchonly ifchunk_index > info.chunks. If the stream has fewer chunks thaninfo.chunks, callers will seeerror.TimeoutfromnextMsgrather than a domain error. That’s OK if intended, but it shifts error handling to callers.If you want API-consistent errors, map a
Timeouthere toObjectStoreError.ChunkMismatchwhenchunk_index < info.chunks.- const js_msg = try sub.nextMsg(timeout_ms); + const js_msg = sub.nextMsg(timeout_ms) catch |e| { + if (e == error.Timeout and self.chunk_index < self.info.chunks) { + return ObjectStoreError.ChunkMismatch; + } + return e; + };
488-494: Consistent getBytes flowRead-loop + final
verify()keeps the bytes path aligned with the streaming API. Consider surfacing a specialized error if the loop ends early (you already returnChunkMismatch).Also applies to: 501-502
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (3)
src/jetstream_objstore.zig(6 hunks)src/root.zig(0 hunks)tests/jetstream_objstore_streaming_test.zig(2 hunks)
💤 Files with no reviewable changes (1)
- src/root.zig
🧰 Additional context used
🧠 Learnings (1)
📚 Learning: 2025-09-06T20:28:53.180Z
Learnt from: lalinsky
PR: lalinsky/nats.zig#98
File: src/jetstream_kv.zig:235-279
Timestamp: 2025-09-06T20:28:53.180Z
Learning: JetStreamMessage wrappers in nats.zig are allocated on the Message's owned arena, so they are automatically freed when Message.deinit() is called and don't require separate cleanup.
Applied to files:
src/jetstream_objstore.zig
🔇 Additional comments (6)
tests/jetstream_objstore_streaming_test.zig (3)
59-62: Good switch to top-levelreadUsing
get_result.read(buffer[total_read..])matches the new API and keeps the loop simple.
68-68: Good: verify after fully readingCalling
get_result.verify()after consuming the stream is the correct order with the new API.
129-130: LGTM: verify on empty objectVerifying integrity for the empty object is exercised correctly.
src/jetstream_objstore.zig (3)
189-214: Read path looks solidChunked copy, message lifecycle (
msg.deinit()), and EOF setting match the zero-copy intent and the retrieved learning about message-owned arenas.
444-447: Empty-object fast path: OKMarking EOF and skipping subscription for
size == 0is clean and avoids unnecessary allocations.
467-469: Ownership of subscription clarified bydeinitReturning a result that owns and deinits the subscription keeps lifecycle tight.
| /// Verify the complete object integrity | ||
| pub fn verify(self: *ObjectResult) !void { | ||
| const calculated_digest = self.digest.finalResult(); | ||
| var digest_hex: [64]u8 = undefined; | ||
| _ = std.fmt.bufPrint(&digest_hex, "{s}", .{std.fmt.fmtSliceHexLower(&calculated_digest)}) catch unreachable; | ||
|
|
||
| if (!std.mem.eql(u8, &digest_hex, self.info.digest)) { | ||
| return ObjectStoreError.DigestMismatch; | ||
| } | ||
|
|
||
| self.verified = true; | ||
| } | ||
|
|
||
| /// Verify the complete object integrity (public API) | ||
| pub fn verify(self: *ObjectReader) !void { | ||
| try self.autoVerify(); | ||
| } | ||
|
|
||
| /// Check if the object has been verified | ||
| pub fn isVerified(self: *ObjectReader) bool { | ||
| return self.verified; | ||
| } | ||
| }; | ||
|
|
||
| /// ObjectResult provides streaming access to object data | ||
| pub const ObjectResult = struct { | ||
| info: ObjectInfo, | ||
| reader: ObjectReader, | ||
| arena: *std.heap.ArenaAllocator, | ||
|
|
||
| pub fn deinit(self: *ObjectResult) void { | ||
| self.reader.deinit(); | ||
| const child_allocator = self.arena.child_allocator; | ||
| self.arena.deinit(); | ||
| child_allocator.destroy(self.arena); | ||
| } | ||
| }; |
There was a problem hiding this comment.
Fix: use slices (not pointers-to-arrays) in hex formatting and equality; also guard against partial verify
Two issues:
bufPrint(&digest_hex, ...)andeql(&digest_hex, ...)should use slices (digest_hex[0..]) for clarity and to avoid type pitfalls.verify()doesn’t enforce that all chunks were read; calling it mid-stream finalizes the hasher and can’t be resumed. Guard against partial reads.
Apply:
pub fn verify(self: *ObjectResult) !void {
+ // Ensure full object was processed before finalizing digest
+ if (!self.eof or self.current_msg != null or self.chunk_index != self.info.chunks) {
+ return ObjectStoreError.ChunkMismatch;
+ }
const calculated_digest = self.digest.finalResult();
var digest_hex: [64]u8 = undefined;
- _ = std.fmt.bufPrint(&digest_hex, "{s}", .{std.fmt.fmtSliceHexLower(&calculated_digest)}) catch unreachable;
+ _ = std.fmt.bufPrint(digest_hex[0..], "{s}", .{std.fmt.fmtSliceHexLower(&calculated_digest)}) catch unreachable;
- if (!std.mem.eql(u8, &digest_hex, self.info.digest)) {
+ if (!std.mem.eql(u8, digest_hex[0..], self.info.digest)) {
return ObjectStoreError.DigestMismatch;
}
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| /// Verify the complete object integrity | |
| pub fn verify(self: *ObjectResult) !void { | |
| const calculated_digest = self.digest.finalResult(); | |
| var digest_hex: [64]u8 = undefined; | |
| _ = std.fmt.bufPrint(&digest_hex, "{s}", .{std.fmt.fmtSliceHexLower(&calculated_digest)}) catch unreachable; | |
| if (!std.mem.eql(u8, &digest_hex, self.info.digest)) { | |
| return ObjectStoreError.DigestMismatch; | |
| } | |
| self.verified = true; | |
| } | |
| /// Verify the complete object integrity (public API) | |
| pub fn verify(self: *ObjectReader) !void { | |
| try self.autoVerify(); | |
| } | |
| /// Check if the object has been verified | |
| pub fn isVerified(self: *ObjectReader) bool { | |
| return self.verified; | |
| } | |
| }; | |
| /// ObjectResult provides streaming access to object data | |
| pub const ObjectResult = struct { | |
| info: ObjectInfo, | |
| reader: ObjectReader, | |
| arena: *std.heap.ArenaAllocator, | |
| pub fn deinit(self: *ObjectResult) void { | |
| self.reader.deinit(); | |
| const child_allocator = self.arena.child_allocator; | |
| self.arena.deinit(); | |
| child_allocator.destroy(self.arena); | |
| } | |
| }; | |
| /// Verify the complete object integrity | |
| pub fn verify(self: *ObjectResult) !void { | |
| // Ensure full object was processed before finalizing digest | |
| if (!self.eof or self.current_msg != null or self.chunk_index != self.info.chunks) { | |
| return ObjectStoreError.ChunkMismatch; | |
| } | |
| const calculated_digest = self.digest.finalResult(); | |
| var digest_hex: [64]u8 = undefined; | |
| _ = std.fmt.bufPrint(digest_hex[0..], "{s}", .{std.fmt.fmtSliceHexLower(&calculated_digest)}) catch unreachable; | |
| if (!std.mem.eql(u8, digest_hex[0..], self.info.digest)) { | |
| return ObjectStoreError.DigestMismatch; | |
| } | |
| } | |
| }; |
🤖 Prompt for AI Agents
In src/jetstream_objstore.zig around lines 216-226, change the fixed-array
pointer usage to explicit slices when formatting and comparing the hex digest
(use digest_hex[0..] for bufPrint and std.mem.eql) and add a guard to prevent
verify() from being called before the entire object has been read (e.g., check a
completed flag or compare bytes_read against self.info.size and return an
appropriate error like ObjectStoreError.PartialVerify or DigestMismatch if not
complete); ensure the hasher is only finalized in verify() when the object is
fully received so mid-stream calls fail fast rather than corrupting the hasher
state.
1 participant
Summary
This PR refactors the ObjectStore implementation to provide a cleaner, more efficient API by merging ObjectReader functionality directly into ObjectResult.
Key Improvements
ObjectResultdirectly instead of accessing.readerObjectStoredependency fromObjectResultverify()callsTechnical Changes
ObjectReaderfields and methods intoObjectResultArrayListbuffer withcurrent_msgpointer andmsg_posoffsetsubscription.js.nc.options.timeout_msObjectReaderfrom public API exportsPerformance Benefits
API Changes
Before:
After:
Code Review
The changes have been thoroughly reviewed by the code-reviewer agent, receiving a 9/10 quality score with strong approval for:
Test plan
Total: 223/223 tests passing ✅
This refactoring maintains full backward compatibility while delivering significant architectural and performance improvements.