Add support for js.getMsg() via the direct get API #95

lalinsky · 2025-09-03T21:48:44Z

No description provided.

This commit implements the JetStream Direct Get functionality according to ADR-31 and reorganizes the message-related tests into separate, focused files. Key changes: - Added support for JetStream Direct Get API ($JS.API.DIRECT.GET.<stream>) - Unified getMsg() API with GetMsgOptions supporting both legacy and direct get - Added allow_direct field to StreamConfig for enabling direct get on streams - Implemented proper header extraction for direct get responses - Added comprehensive validation for option combinations - Reorganized tests into three focused files: * jetstream_get_msg_test.zig - Legacy API tests * jetstream_get_msg_direct_test.zig - Direct API tests * jetstream_delete_msg_test.zig - Delete/erase operation tests Features implemented: - Direct get from any replica (not just leader) - Support for seq, last_by_subj, and next_by_subj operations - Header preservation in direct get responses - Proper error handling for both APIs - Protocol-mapped field validation 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <[email protected]>

- Fix critical segfault in MessagePool recycling by replacing clearRetainingCapacity() with complete HashMap reset (.{}) in Message.reset() - Move header parsing to occur before processMsg() in parser to eliminate lazy parsing and race conditions - Remove needs_header_parsing boolean flag and associated complexity - Simplify header API functions to remove unnecessary error unions - All direct API tests now pass with message recycling enabled 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <[email protected]>

coderabbitai · 2025-09-03T21:48:50Z

Walkthrough

Adds option-driven JetStream GET APIs including a direct GET path, expands GetMsg request fields and StreamConfig.allow_direct, converts Message header parsing to an explicit parseHeaders flow with numeric status codes, updates parser to parse headers before delivery, and splits/refactors JetStream message tests.

Changes

Cohort / File(s)	Summary
JetStream API & GET flows `src/jetstream.zig`	Introduces `GetMsgOptions` and `GetMsgsOptions`; changes `getMsg` to accept options and dispatch to legacy (`getMsgLegacy`) or direct (`getMsgDirect`) flows; expands `GetMsgRequest` fields; adds `getMsgs` stub; exposes `StreamConfig.allow_direct`; removes `getLastMsg`.
Message internals & headers `src/message.zig`	Replaces string status header constants with numeric `STATUS_*` (u16); adds `status_code` field; removes lazy parsing flag; adds `parseHeaders()`; `setRawHeaders` triggers parse; header accessors made non-throwing; updates encode/reset behavior.
Parser integration `src/parser.zig`	Removes needs-header-parsing flag in MSG handling and calls `msg.parseHeaders()` before `conn.processMsg(msg)`; test expectation updated to reflect non-throwing headerGet.
Message tests / API usage updates `src/message_test.zig`, `tests/headers_test.zig`	Tests adapted to allocator-only `Message.init`, use of setters (`setSubject`, `setReply`, `setPayload`, `setRawHeaders`), explicit `parseHeaders()`, and non-throwing header APIs.
JetStream tests — split and new direct tests `tests/jetstream_get_msg_test.zig`, `tests/jetstream_get_msg_direct_test.zig`, `tests/jetstream_delete_msg_test.zig`, `tests/all_tests.zig`	Adds separate test suites for legacy getMsg, direct getMsg, and delete/erase semantics; updates `all_tests.zig` to import the new modules.
Removed legacy test `tests/jetstream_msg_test.zig`	Deletes the old combined JetStream message test file (replaced by focused suites).

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~60 minutes

Possibly related PRs

Add JetStream message operations #33 — Changes JetStream getMsg flows and message API shapes; directly relevant to GetMsgOptions and direct/legacy dispatch.
Implement shared response subscription for request/reply operations #29 — Adjusts message header/status parsing and no-responders semantics; relevant to Message.parseHeaders() and status_code.
Implement JetStream pull subscriptions with fetch() API #31 — Modifies pull/fetch status handling; related to moving status handling to numeric status codes and updated JetStream status paths.

✨ Finishing Touches

🧪 Generate unit tests

Create PR with unit tests
Post copyable unit tests in a comment
Commit unit tests in branch jetstream-direct-get

🪧 Tips

Chat

There are 3 ways to chat with CodeRabbit:

Review comments: Directly reply to a review comment made by CodeRabbit. Example:
- I pushed a fix in commit <commit_id>, please review it.
- Open a follow-up GitHub issue for this discussion.
Files and specific lines of code (under the "Files changed" tab): Tag @coderabbitai in a new review comment at the desired location with your query.
PR comments: Tag @coderabbitai in a new PR comment to ask questions about the PR branch. For the best results, please provide a very specific query, as very limited context is provided in this mode. Examples:
- @coderabbitai gather interesting stats about this repository and render them as a table. Additionally, render a pie chart showing the language distribution in the codebase.
- @coderabbitai read the files in the src/scheduler package and generate a class diagram using mermaid and a README in the markdown format.

Support

Need help? Create a ticket on our support page for assistance with any issues or questions.

CodeRabbit Commands (Invoked using PR/Issue comments)

Type @coderabbitai help to get the list of available commands.

Other keywords and placeholders

Add @coderabbitai ignore or @coderabbit ignore anywhere in the PR description to prevent this PR from being reviewed.
Add @coderabbitai summary to generate the high-level summary at a specific location in the PR description.
Add @coderabbitai anywhere in the PR title to generate the title automatically.

Status, Documentation and Community

Visit our Status Page to check the current availability of CodeRabbit.
Visit our Documentation for detailed information on how to use CodeRabbit.
Join our Discord Community to get help, request features, and share feedback.
Follow us on X/Twitter for updates and announcements.

coderabbitai

Actionable comments posted: 4

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (4)

src/parser.zig (1)
763-783: Remove try from non-throwing headerGet calls in tests
headerGet now returns ?[]const u8 and doesn’t throw—drop try in all test usages:

parser.zig (around line 780):
-            try std.testing.expectEqualStrings("Bar", try msg.headerGet("Foo") orelse "");
+            try std.testing.expectEqualStrings("Bar", msg.headerGet("Foo") orelse "");
message_test.zig (line 70):
-    const missing = try msg.headerGet("Missing");
+    const missing = msg.headerGet("Missing");
message_test.zig (line 166):
-    const result = try msg.headerGet("");
+    const result = msg.headerGet("");
Tests will compile once these try keywords are removed.
src/message_test.zig (2)

18-30: Constructor mismatch vs implementation

Tests call Message.init(allocator, subject, reply, data) and Message.initWithHeaders(...), but the implementation exposes Message.init(allocator) with setters. Either add these convenience constructors or rewrite tests to use setSubject/setReply/setPayload/setRawHeaders + parseHeaders().

I can draft the constructors or update tests — your pick.

79-98: Remove try from non-throwing API calls in tests

In src/message_test.zig:

Line 70: change
const missing = try msg.headerGet("Missing");
to
const missing = msg.headerGet("Missing");

Line 110: change
try testing.expect(try msg.isNoResponders());
to
try testing.expect(msg.isNoResponders());

Line 114: change
try testing.expect(!try msg.isNoResponders());
to
try testing.expect(!msg.isNoResponders());

Line 166: change
const result = try msg.headerGet("");
to
const result = msg.headerGet("");

In src/parser.zig (line 780): change
try std.testing.expectEqualStrings("Bar", try msg.headerGet("Foo") orelse "");
to
try std.testing.expectEqualStrings("Bar", msg.headerGet("Foo") orelse "");
src/jetstream.zig (1)
891-959: Legacy get: call Message helpers for subject and headers.
Replace manual subject duplication with:
-        msg.subject = try arena_allocator.dupe(u8, stored_msg.subject);
+        try msg.setSubject(stored_msg.subject);
And immediately after assigning msg.raw_headers = decoded_headers;, add:
+        if (msg.raw_headers) try msg.parseHeaders();
Consider exposing stored_msg.time (e.g. store the raw timestamp) since RFC3339 parsing is still TODO.

🧹 Nitpick comments (11)

src/message.zig (5)
119-162: parseHeaders: make idempotent or clear map to avoid silent duplication

Repeated calls will append/replace entries in an arena-backed map, leaking capacity within the arena and potentially producing multiple values for the same key. Either short-circuit when headers are already parsed or clear the map first.

Example change:
 pub fn parseHeaders(self: *Self) !void {
-    const raw = self.raw_headers orelse return;
+    const raw = self.raw_headers orelse return;
+    if (self.headers.count() != 0) return; // already parsed or headerSet() used
130-161: Status/Description values rely on raw backing — fine, but document invariant

You’re storing slices into raw_headers (arena-owned). That’s OK, but it assumes raw_headers outlives the header map. Add a brief comment to make that invariant explicit. Alternatively, dupe into the arena for consistency with the rest of the code.

164-184: Consider canonicalizing header keys (case-insensitive semantics)

NATS headers are conventionally case-insensitive (http.Header in Go). Today keys are stored verbatim, so "Content-Type" and "content-type" will coexist. Normalizing (e.g., title-case) on parse avoids surprises.

Sketch:
-const key = std.mem.trim(u8, line[0..colon_pos], " \t");
+var key_buf = std.ArrayList(u8).init(arena_allocator);
+defer key_buf.deinit();
+try key_buf.appendSlice(std.mem.trim(u8, line[0..colon_pos], " \t"));
+// apply canonicalization to key_buf.items
+const key = key_buf.items;
215-222: headerGetAll: returning arena-backed slice is fine — call out lifetime

The returned slice’s lifetime is tied to the message arena. A short note in the docstring would prevent misuse.

264-265: reset: switching to headers = .{} discards table capacity

Given the pool immediately resets the arena afterwards, this is probably fine. If you ever reuse Message without resetting the arena, clearRetainingCapacity() would be cheaper.
tests/jetstream_get_msg_direct_test.zig (1)

279-284: Mixed error types across direct vs legacy paths.

Here you expect error.MessageNotFound, while legacy tests assert nats.JetStreamError.NoMessageFound. If intentional (different surfaces), great; otherwise consider unifying the error mapping for a flatter API.

Would you like a follow-up patch to normalize these to a common error in the public API?

src/jetstream.zig (5)

157-159: StreamConfig.allow_direct added — double-check server compatibility.

Field name matches NATS config. Ensure your test matrix includes a server that supports direct reads; older servers should ignore unknown fields but verify behavior when nulls are sent.

307-330: GetMsgRequest grew a lot; most fields aren’t reachable yet.

Only seq/last_by_subj/next_by_subj are used. Either (a) keep the extra fields but mark them explicitly “reserved, not yet wired” in docs, or (b) defer adding until GetMsgs/direct-batch lands to avoid freezing an unused API.

331-341: GetMsgOptions validation pairs well with getMsg; consider builders for ergonomics.

Optional: add small constructors like GetMsgOptions.bySeq(u64) and .lastBySubj([]const u8) to prevent invalid combos at call sites.

385-389: Public GetMsgsOptions placeholder.

If batch isn’t shipping in this PR, consider making this internal or clearly annotating “experimental” to avoid premature API surface.

988-994: Batch get stub.

Error.NotImplemented is fine. If you want, I can scaffold the ADR-31-compatible request/response and tests behind a feature flag.

📜 Review details

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Pro

💡 Knowledge Base configuration:

MCP integration is disabled by default for public repositories
Jira integration is disabled by default for public repositories
Linear integration is disabled by default for public repositories

You can enable these sources in your CodeRabbit configuration.

📥 Commits

Reviewing files that changed from the base of the PR and between e95a1c5 and a82f666.

📒 Files selected for processing (10)

src/jetstream.zig (7 hunks)
src/message.zig (6 hunks)
src/message_test.zig (3 hunks)
src/parser.zig (1 hunks)
tests/all_tests.zig (1 hunks)
tests/headers_test.zig (3 hunks)
tests/jetstream_delete_msg_test.zig (1 hunks)
tests/jetstream_get_msg_direct_test.zig (1 hunks)
tests/jetstream_get_msg_test.zig (1 hunks)
tests/jetstream_msg_test.zig (0 hunks)

💤 Files with no reviewable changes (1)

tests/jetstream_msg_test.zig

🧰 Additional context used

📓 Path-based instructions (1)

**/*.zig

📄 CodeRabbit inference engine (CLAUDE.md)

Format code using zig fmt before commit

Files:

tests/all_tests.zig
tests/jetstream_get_msg_test.zig
tests/jetstream_get_msg_direct_test.zig
tests/jetstream_delete_msg_test.zig
tests/headers_test.zig
src/parser.zig
src/message_test.zig
src/message.zig
src/jetstream.zig

🧠 Learnings (2)

📚 Learning: 2025-08-25T14:51:33.570Z

Learnt from: lalinsky
PR: lalinsky/nats.zig#31
File: tests/jetstream_pull_test.zig:21-23
Timestamp: 2025-08-25T14:51:33.570Z
Learning: The nats.zig project uses a "tests:beforeEach" test in tests/all_tests.zig that runs before each test to clean up all JetStream streams and consumers, ensuring test isolation. This eliminates the need for unique test resource names as each test starts with a clean JetStream state.

Applied to files:

tests/all_tests.zig
tests/jetstream_get_msg_test.zig
tests/jetstream_get_msg_direct_test.zig
tests/jetstream_delete_msg_test.zig

📚 Learning: 2025-08-29T10:22:22.247Z

Learnt from: CR
PR: lalinsky/nats.zig#0
File: CLAUDE.md:0-0
Timestamp: 2025-08-29T10:22:22.247Z
Learning: Run `zig build test` to confirm your changes are working

Applied to files:

tests/all_tests.zig

🪛 GitHub Actions: CI

tests/jetstream_get_msg_test.zig