Experiment

[lalinsky]

No description provided.

[coderabbitai]

Important

Review skipped

Auto reviews are disabled on base/target branches other than the default branch.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Walkthrough

Refactors JetStream and KV APIs: adjusts nextMsg error signature, introduces a generic arena-backed Result(T), switches JSON parsing to return Result(T), adds timestamp parsing for messages, redesigns KVWatcher and watch/history/keys APIs, removes KV.destroy, updates message metadata types, and aligns tests with new behaviors.

Changes

Cohort / File(s)	Summary
JetStream core src/jetstream.zig	Imports parseTimestamp and Result(T); changes JetStreamSubscription.nextMsg to return !*JetStreamMessage; replaces pub const Result alias; parseResponse now returns Result(T) from std.json.parse; getMsgLegacy parses time via parseTimestamp.
KV module and watcher src/jetstream_kv.zig	Switches to result.zig Result(T); uses newInbox; KVOperation.fromString now returns error union; adds KVEntry.isDeleted; redesigns KVWatcher (fields, init, and next with timeout returning !?KVEntry); watch/watchAll return KVWatcher (by value) and adds watchMulti; history/keys return arena-backed Result wrappers; removes KV.destroy; updates parsing helpers.
Message models src/jetstream_message.zig, src/message.zig	Makes SequencePair.consumer/stream non-nullable u64 with default 0; adds time: u64 = 0 to Message.
Generic Result wrapper src/result.zig	Adds public generic Result(T) with ArenaAllocator, init(allocator, value), and deinit().
Tests updated for new APIs tests/jetstream_kv_history_test.zig, tests/jetstream_kv_test.zig, tests/jetstream_nak_test.zig	History now via Result wrapper with deinit; watcher.next uses timeout and returns optional; cleanup via kv_manager.deleteBucket; sequence assertions changed to > 0.

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~60 minutes

Possibly related PRs

• Implement KV watch, history and keys functionality (WIP) #97 — Also reworks src/jetstream_kv.zig (watcher, watch/history/keys APIs), overlapping with this KV refactor.
• Change nextMsg() to return error.Timeout instead of null #83 — Adjusts JetStreamSubscription.nextMsg timeout/error behavior, matching this PR’s nextMsg signature change.
• Add JetStream STREAM.INFO API implementation and reorganize tests #25 — Uses parseResponse to return Result(...) for JetStream JSON responses, aligning with the new Result(T) wrapper here.

✨ Finishing Touches

🧪 Generate unit tests

• Create PR with unit tests
• Post copyable unit tests in a comment
• Commit unit tests in branch jetstream-kv-ng

---

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.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[lalinsky]

@coderabbitai review

[coderabbitai]

✅ Actions performed

Review triggered.

Note: CodeRabbit is an incremental review system and does not re-review already reviewed commits. This command is applicable only when automatic reviews are paused.

[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 (1)

tests/jetstream_nak_test.zig (1)

281-285: Fix log format: using optional placeholder {?} with non-optional values

sequence.consumer, sequence.stream, and num_pending are u64 now; {?} will fail format-time. Use {}.

Apply:

-            log.info("- Consumer sequence: {?}", .{js_msg.metadata.sequence.consumer});
-            log.info("- Stream sequence: {?}", .{js_msg.metadata.sequence.stream});
+            log.info("- Consumer sequence: {}", .{js_msg.metadata.sequence.consumer});
+            log.info("- Stream sequence: {}", .{js_msg.metadata.sequence.stream});
-            log.info("- Pending count: {?}", .{js_msg.metadata.num_pending});
+            log.info("- Pending count: {}", .{js_msg.metadata.num_pending});

🧹 Nitpick comments (6)

tests/jetstream_kv_test.zig (1)

32-34: Prefer deferred cleanup for buckets

Defer deletion so the bucket is cleaned up even if the test bails early. This keeps resource lifecycle tight and avoids ordering footguns.

-    // Clean up
-    try kv_manager.deleteBucket(bucket_name);
+    // Clean up
+    defer kv_manager.deleteBucket(bucket_name) catch {};

src/result.zig (1)

27-37: Nit: init doesn’t actually fail anymore

After the change above, init has no fallible path. Consider returning Self instead of Allocator.Error!Self and drop try at call sites.

-        pub fn init(allocator: std.mem.Allocator, value: T) std.mem.Allocator.Error!Self {
+        pub fn init(allocator: std.mem.Allocator, value: T) Self {
             return Self{
                 .arena = std.heap.ArenaAllocator.init(allocator),
                 .value = value,
             };
         }

tests/jetstream_kv_history_test.zig (1)

180-199: Watcher.next() assertions: prefer checking the error branch explicitly.

Comparing an error-union variable to error tags can be brittle. Pattern-match the error branch to assert Timeout/QueueEmpty.

Apply:

-    const result = watcher.next(1000);
-    try testing.expect(result == error.Timeout or result == error.QueueEmpty);
+    const res = watcher.next(1000);
+    if (res) |_| {
+        try testing.expect(false); // expected a timeout/empty after completion marker
+    } else |err| {
+        try testing.expect(err == error.Timeout or err == error.QueueEmpty);
+    }

src/jetstream_kv.zig (3)

200-228: KVWatcher.init(): config is reasonable, but one nuance.

headers_only mirrors meta_only — good. Using filter_subjects is correct for multi-pattern. Note: init_pending from consumer_info is best-effort; you also derive it from first message which is fine as fallback.

If you want to reduce surprises, consider a specific initial_count strategy: prefer first-message delta when include_history=false, else consumer_info.num_pending.

---

632-639: Minor: drop the unconditional debug log of subjects in production builds.

Log spam in hot paths makes tests noisy. Consider gating behind a verbose flag or removing.

-        log.debug("subject: {s}", .{msg.subject});
+        if (@import("builtin").mode == .Debug) {
+            log.debug("subject: {s}", .{msg.subject});
+        }

---

686-729: history(): sensible arena ownership; one tiny improvement.

Using nc.options.timeout_ms as a total budget is fine. Optionally accept an override timeout for history to avoid coupling to connection defaults.

Happy to draft an overload if you want one.

📜 Review details

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between 54016ec and d9da51a.

📒 Files selected for processing (8)

• src/jetstream.zig (4 hunks)
• src/jetstream_kv.zig (8 hunks)
• src/jetstream_message.zig (1 hunks)
• src/message.zig (1 hunks)
• src/result.zig (1 hunks)
• tests/jetstream_kv_history_test.zig (2 hunks)
• tests/jetstream_kv_test.zig (1 hunks)
• tests/jetstream_nak_test.zig (1 hunks)

🧰 Additional context used

🧠 Learnings (1)

📚 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/jetstream_nak_test.zig
• tests/jetstream_kv_test.zig

🔇 Additional comments (14)

tests/jetstream_nak_test.zig (1)

293-295: Good switch to numeric checks

> 0 is the right replacement after making sequences non-optional.

tests/jetstream_kv_history_test.zig (1)

46-50: Good move to arena-backed Result for history.

The wrapper + deinit() contract is clear, and per-entry cleanup via entry.deinit() is correct.

src/jetstream.zig (4)

27-29: Imports look right.

Pulling in parseTimestamp and the project-wide Result wrapper aligns this module with the new API.

---

654-667: parseResponse(): consistent Result(T) wrapping with arena hand-off.

Looks good. Callers must remember to result.deinit(); your tests reflect that.

---

939-946: Legacy GetMsg: timestamp parsing added.

Setting msg.time via parseTimestamp(stored_msg.time) is the right fix; downstream code can now rely on msg.time.

---

593-600: Restrict catch to only error.Timeout in examples/benchmarks
In examples/replier.zig (line 35), benchmarks/subscriber.zig (line 42) and benchmarks/echo_server.zig (line 41), you’re using a generic catch continue which silences all errors. Change to explicit matching—continue only on error.Timeout and propagate any other error.

⛔ Skipped due to learnings

Learnt from: lalinsky
PR: lalinsky/nats.zig#60
File: src/queue.zig:390-393
Timestamp: 2025-08-30T07:40:28.186Z
Learning: In the nats.zig project, the maintainer prefers to rely on Zig's built-in integer overflow detection rather than implementing manual saturated conversions for timeout calculations (timeout_ms * std.time.ns_per_ms). Zig automatically fails on integer overflow in Debug and ReleaseSafe modes.

src/jetstream_kv.zig (8)

21-21: Module imports updated to local Result and newInbox.

All good; this decouples KV from jetstream.zig’s old alias and avoids implicit Inbox creation paths.

Also applies to: 29-29

---

54-59: KVOperation.fromString(): fail-fast on unknown values.

Returning error.InvalidOperation is preferable to silently defaulting. Callers already “try” it.

---

118-124: isDeleted() helper improves readability.

This simplifies KeyNotFound checks elsewhere.

---

395-401: getRawEntry(): good mapping for MessageNotFound -> KeyNotFound.

errdefer msg.deinit() and parseMessage() ownership are consistent.

---

408-409: fromString() errors are propagated here correctly.

Keeps bad headers from masquerading as PUT.

---

586-593: watch/watchMulti/watchAll: APIs look good; lifetime of temporary subjects is safe.

Subjects are serialized into JSON during addConsumer; freeing the local copies after init is fine.

Also applies to: 596-617, 620-625

---

733-763: keys(): uses completion marker correctly and dedup via StringHashMap.

Allocation split (arena for keys, self.allocator for map) is reasonable and avoids double free.

Also applies to: 769-781

---

802-813: keysWithFilters(): mirrors keys() well.

Same notes as keys(); looks good.

Also applies to: 815-831, 839-850