Implement JetStream Push Subscriptions #30
Conversation
- Implement getStreamInfo method in jetstream.zig for $JS.API.STREAM.INFO endpoint - Add comprehensive test for STREAM.INFO functionality - Reorganize JetStream tests by moving stream management tests to dedicated jetstream_stream_test.zig file - Update test runner to include new test file - All tests passing (27/27) This addresses the missing STREAM.INFO endpoint identified in the JetStream API implementation.
…livery
This adds comprehensive support for JetStream push subscriptions following
established NATS library naming conventions.
## Core Features Implemented:
### Enhanced Consumer Configuration
- Extended ConsumerConfig with push-specific fields:
- deliver_group for queue-based load balancing
- flow_control for backpressure management
- idle_heartbeat support (required with flow_control)
- Automatic filtering of pull-only fields for push consumers
### JetStream Message Enhancement
- JetStreamMessage wrapper with header parsing:
- Nats-Stream, Nats-Sequence, Nats-Subject, Nats-Time-Stamp
- Acknowledgment methods: ack(), nak(), term(), inProgress()
- Automatic cleanup and memory management
### Push Subscription API
- subscribe() method following NATS.C/NATS.Go naming conventions
- Type-safe callback handler integration
- Consumer creation with deliver_subject configuration
- Automatic NATS to JetStream message conversion
- Proper configuration validation and error handling
### Advanced Features
- Flow control with idle heartbeats for high-throughput scenarios
- Queue group support for load-balanced processing
- Explicit acknowledgment handling (ACK/NAK/TERM/+WPI)
- JetStream-specific error validation and reporting
## Test Coverage
- Comprehensive test suite in jetstream_push_test.zig
- Basic push subscription with callback processing
- Flow control with heartbeat configuration
- Error handling and validation testing
- All 30 tests passing with working functionality
## API Example:
```zig
const consumer_config = nats.ConsumerConfig{
.durable_name = "processor",
.deliver_subject = "push.orders.process",
.ack_policy = .explicit,
.flow_control = true,
.idle_heartbeat = 30_000_000_000,
};
var sub = try js.subscribe("ORDERS", consumer_config, handler, .{&state});
```
This implementation provides a solid foundation for production-ready
JetStream push subscriptions with proper callback handling, acknowledgments,
and flow control.
- Add handleStatusMessage() function to process Status: "100" messages - Filter status messages from user callbacks in push subscriptions - Handle "Idle Heartbeat" and "FlowControl Request" messages - Automatically respond to flow control requests with empty messages - Add debug logging to understand actual message header format This implements the JetStream heartbeat protocol as used by NATS.Go and NATS.C libraries, ensuring status messages are handled internally rather than being passed to user message handlers. Note: Threading issue identified - message handlers currently run in network thread, blocking I/O. This will be addressed separately.
…h-subscriptions # Conflicts: # tests/all_tests.zig
- Remove two-level context structure that was causing memory leaks - Simplify handler to pass JetStream instance and args directly - Add proper subscription cleanup in JetStreamSubscription.deinit() - Eliminate JSContext allocation that wasn't being freed - All tests now pass without memory leak warnings This fixes the cleanup issue where js_context was allocated but never properly freed during subscription teardown.
WalkthroughAdds JetStream push-subscription support and message-wrapping utilities, extends ConsumerConfig, re-exports new types at root, and introduces sync and async subscription APIs with status handling. Includes comprehensive tests for push, NAK/redelivery, and sync flows. Adds an implementation plan document for a Zig 0.15 NATS client. Changes
Sequence Diagram(s)sequenceDiagram
autonumber
participant App as Application
participant JS as JetStream
participant Srv as NATS Server
participant Sub as Subscription (push)
rect rgba(224,247,250,0.5)
App->>JS: subscribe(stream, ConsumerConfig, handlerFn)
JS->>Srv: Create Consumer (deliver_subject)
JS->>Srv: SUB deliver_subject
JS-->>App: JetStreamSubscription
end
loop Delivery
Srv-->>Sub: MSG (data or status)
Sub->>JS: dispatch
alt Status (Idle Heartbeat / FlowControl)
JS->>Srv: FlowControl reply (empty publish) [if required]
else Data
JS->>App: handlerFn(JetStreamMessage)
App->>JS: ack/nak/term/inProgress
JS->>Srv: Publish ACK subject
end
end
sequenceDiagram
autonumber
participant App as Application
participant JS as JetStream
participant Srv as NATS Server
participant Sub as Subscription (sync)
App->>JS: subscribeSync(stream, ConsumerConfig)
JS->>Srv: Create Consumer + SUB deliver_subject
JS-->>App: JetStreamSubscription
rect rgba(232,245,233,0.6)
App->>JS: nextMsg(timeout_ms)
alt Message within timeout
Srv-->>Sub: MSG
Sub-->>JS: Message
JS-->>App: JetStreamMessage
App->>JS: ack/nak/...
JS->>Srv: Publish ACK subject
else Timeout
JS-->>App: null
end
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60–90 minutes Possibly related PRs
Poem
Tip 🔌 Remote MCP (Model Context Protocol) integration is now available!Pro plan users can now connect to remote MCP servers from the Integrations page. Connect with popular remote MCPs such as Notion and Linear to add more context to your reviews and chats. ✨ Finishing Touches🧪 Generate unit tests
🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
SupportNeed help? Create a ticket on our support page for assistance with any issues or questions. CodeRabbit Commands (Invoked using PR/Issue comments)Type Other keywords and placeholders
Status, Documentation and Community
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 14
🧹 Nitpick comments (3)
src/jetstream.zig (2)
163-164: Fix the comment about field relocation.The comment mentions that the flow_control field was moved to the end to avoid duplication, but it's actually the comment itself that was misplaced between
idle_heartbeatandmax_waitingfields./// If the Consumer is idle for more than this many nano seconds a empty message with Status header 100 will be sent idle_heartbeat: ?u64 = null, - /// For push consumers this will regularly send an empty mess with Status header 100 and a reply subject - /// (This field was moved to the end to avoid duplication) /// The number of pulls that can be outstanding on a pull consumer max_waiting: i64 = 512,
675-677: Consider making pull-only fields nullable for push consumers.Currently,
max_waitingis forcibly set to 0 for push consumers, while other pull-specific fields are set to null. For consistency and to avoid potential confusion, consider makingmax_waitingnullable as well.// Create push consumer config by removing pull-only fields var push_config = consumer_config; - push_config.max_waiting = 0; // Push consumers don't support max_waiting + push_config.max_waiting = null; // Push consumers don't support max_waiting push_config.max_batch = null; // Push consumers don't support max_batch push_config.max_expires = null; // Push consumers don't support max_expiresHowever, this would require changing the field type in
ConsumerConfig:/// The number of pulls that can be outstanding on a pull consumer - max_waiting: i64 = 512, + max_waiting: ?i64 = 512,tests/jetstream_push_test.zig (1)
62-62: Consider using a more robust synchronization mechanism.Using
sleepfor test synchronization can lead to flaky tests. Consider implementing a more deterministic approach.Instead of a fixed sleep, you could wait for a specific message count or implement a timeout with polling:
- // Wait a bit for messages to be processed - std.time.sleep(100 * std.time.ns_per_ms); + // Wait for messages to be processed with timeout + const timeout = std.time.ns_per_ms * 1000; // 1 second timeout + const start = std.time.nanoTimestamp(); + while (message_count < 3 and (std.time.nanoTimestamp() - start) < timeout) { + std.time.sleep(10 * std.time.ns_per_ms); + }
📜 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.
📒 Files selected for processing (6)
IMPLEMENTATION_PLAN.md(1 hunks)src/jetstream.zig(5 hunks)src/jetstream2.zig(1 hunks)src/root.zig(1 hunks)tests/all_tests.zig(1 hunks)tests/jetstream_push_test.zig(1 hunks)
🧰 Additional context used
🧠 Learnings (2)
📚 Learning: 2025-08-23T20:26:30.660Z
Learnt from: CR
PR: lalinsky/nats.zig#0
File: CLAUDE.md:0-0
Timestamp: 2025-08-23T20:26:30.660Z
Learning: Run tests (zig build test) to confirm your changes are working
Applied to files:
tests/all_tests.zig
📚 Learning: 2025-08-25T02:49:59.082Z
Learnt from: lalinsky
PR: lalinsky/nats.zig#28
File: src/response_manager.zig:89-105
Timestamp: 2025-08-25T02:49:59.082Z
Learning: In this NATS Zig codebase, subscriptions are reference counted, so calling deinit() on a subscription from multiple locations (like both Connection and ResponseManager) is safe and doesn't cause double-free issues.
Applied to files:
src/jetstream.zig
🪛 LanguageTool
IMPLEMENTATION_PLAN.md
[grammar] ~12-~12: There might be a mistake here.
Context: ...le exists - use std.net.Stream for TCP - Readers/Writers are concrete types, not ...
(QB_NEW_EN)
[grammar] ~17-~17: There might be a mistake here.
Context: ...ters explicitly ### Reference Materials 1. NATS Protocol Specification: `/refs/na...
(QB_NEW_EN)
[grammar] ~18-~18: There might be a mistake here.
Context: ...ol.md- Complete protocol documentation 2. **Client Development Guide**:/refs/nats-...
(QB_NEW_EN)
[grammar] ~19-~19: There might be a mistake here.
Context: ...Best practices for client implementation 3. Reference Implementations: - `/refs...
(QB_NEW_EN)
[grammar] ~22-~22: There might be a mistake here.
Context: ...go/- Go implementation (most complete) 4. **Initial API Design**:/design/api.zig` ...
(QB_NEW_EN)
[grammar] ~52-~52: There might be a mistake here.
Context: ....1 Protocol Parser Module (parser.zig) Goal: Zero-allocation parser for NATS ...
(QB_NEW_EN)
[grammar] ~88-~88: There might be a mistake here.
Context: ...usize, }; ``` Implementation Notes: - Use state machine pattern (see `/refs/na...
(QB_NEW_EN)
[grammar] ~94-~94: There might be a mistake here.
Context: ...1.2 Connection Module (connection.zig) Goal: TCP connection with buffered I/O...
(QB_NEW_EN)
[grammar] ~150-~150: There might be a mistake here.
Context: ...`` Critical Implementation Details: - Must handle INFO message immediately aft...
(QB_NEW_EN)
[grammar] ~244-~244: There might be a mistake here.
Context: ...release); } #### 3.2 Error Handling Parse and handle -ERR messages:zig f...
(QB_NEW_EN)
[grammar] ~333-~333: There might be a mistake here.
Context: ...``` ## Testing Strategy ### Unit Tests 1. Parser Tests: Test all protocol messag...
(QB_NEW_EN)
[grammar] ~339-~339: There might be a mistake here.
Context: ...Encoding/decoding ### Integration Tests 1. Basic Pub/Sub: Connect, publish, subsc...
(QB_NEW_EN)
[grammar] ~340-~340: There might be a mistake here.
Context: ...**: Connect, publish, subscribe, receive 2. Request/Reply: Synchronous request/res...
(QB_NEW_EN)
[grammar] ~341-~341: There might be a mistake here.
Context: ...st/Reply**: Synchronous request/response 3. Queue Groups: Load balancing test 4. *...
(QB_NEW_EN)
[grammar] ~342-~342: There might be a mistake here.
Context: ...3. Queue Groups: Load balancing test 4. Reconnection: Server restart handling ...
(QB_NEW_EN)
[grammar] ~343-~343: There might be a mistake here.
Context: ...Reconnection*: Server restart handling 5. Auth Tests: Various authentication met...
(QB_NEW_EN)
[grammar] ~346-~346: There might be a mistake here.
Context: ...n methods ### Protocol Compliance Tests Use the official NATS server test suite:...
(QB_NEW_EN)
[grammar] ~374-~374: There might be a mistake here.
Context: ...cklist - [ ] Can connect to NATS server - [ ] Can publish messages - [ ] Can subsc...
(QB_NEW_EN)
[grammar] ~375-~375: There might be a mistake here.
Context: ...o NATS server - [ ] Can publish messages - [ ] Can subscribe and receive messages -...
(QB_NEW_EN)
[grammar] ~376-~376: There might be a mistake here.
Context: ...- [ ] Can subscribe and receive messages - [ ] Handles PING/PONG correctly - [ ] Re...
(QB_NEW_EN)
[grammar] ~377-~377: There might be a mistake here.
Context: ...ssages - [ ] Handles PING/PONG correctly - [ ] Reconnects on disconnect - [ ] Suppo...
(QB_NEW_EN)
[grammar] ~378-~378: There might be a mistake here.
Context: ...correctly - [ ] Reconnects on disconnect - [ ] Supports all auth methods - [ ] Queu...
(QB_NEW_EN)
[grammar] ~379-~379: There might be a mistake here.
Context: ...sconnect - [ ] Supports all auth methods - [ ] Queue groups work - [ ] Request/Repl...
(QB_NEW_EN)
[grammar] ~380-~380: There might be a mistake here.
Context: ...all auth methods - [ ] Queue groups work - [ ] Request/Reply pattern works - [ ] He...
(QB_NEW_EN)
[grammar] ~381-~381: There might be a mistake here.
Context: ...s work - [ ] Request/Reply pattern works - [ ] Headers (HPUB/HMSG) supported - [ ] ...
(QB_NEW_EN)
[grammar] ~382-~382: There might be a mistake here.
Context: ...orks - [ ] Headers (HPUB/HMSG) supported - [ ] No memory leaks (test with valgrind)...
(QB_NEW_EN)
[grammar] ~383-~383: There might be a mistake here.
Context: ...[ ] No memory leaks (test with valgrind) - [ ] Thread-safe operations - [ ] Perform...
(QB_NEW_EN)
[grammar] ~384-~384: There might be a mistake here.
Context: ...h valgrind) - [ ] Thread-safe operations - [ ] Performance meets expectations ## E...
(QB_NEW_EN)
[grammar] ~424-~424: There might be a mistake here.
Context: ...erence/reference-protocols/nats-protocol - NATS Clients: https://github.com/nats-io...
(QB_NEW_EN)
[grammar] ~425-~425: There might be a mistake here.
Context: ...ATS Clients: https://github.com/nats-io/ - Zig 0.15 Docs: https://ziglang.org/docum...
(QB_NEW_EN)
[grammar] ~426-~426: There might be a mistake here.
Context: ...ttps://ziglang.org/documentation/0.15.0/ - Test Server: https://github.com/nats-io/...
(QB_NEW_EN)
[grammar] ~432-~432: There might be a mistake here.
Context: ...king 2. Use /refs/nats.c/src/parser.c as reference for parser state machine 3. S...
(QB_NEW_EN)
[grammar] ~432-~432: There might be a mistake here.
Context: .../refs/nats.c/src/parser.cas reference for parser state machine 3. Study/refs/na...
(QB_NEW_EN)
[grammar] ~434-~434: There might be a mistake here.
Context: ...ats.go` for API design patterns 4. Test against real NATS server frequently 5. Focus on...
(QB_NEW_EN)
🪛 markdownlint-cli2 (0.17.2)
IMPLEMENTATION_PLAN.md
28-28: Fenced code blocks should have a language specified
(MD040, fenced-code-language)
424-424: Bare URL used
(MD034, no-bare-urls)
425-425: Bare URL used
(MD034, no-bare-urls)
426-426: Bare URL used
(MD034, no-bare-urls)
427-427: Bare URL used
(MD034, no-bare-urls)
🔇 Additional comments (12)
IMPLEMENTATION_PLAN.md (1)
1-438: LGTM! Comprehensive implementation plan for NATS.zig client library.This implementation plan provides excellent guidance for building a NATS client in Zig 0.15, with clear architectural decisions, phased implementation approach, and thorough testing strategy. The document effectively captures the critical context around Zig 0.15's I/O changes and provides practical code examples.
src/jetstream.zig (2)
618-659: LGTM! Robust status message handling implementation.The implementation correctly handles both idle heartbeats and flow control requests as per the JetStream protocol. The flow control response mechanism appropriately publishes an empty message to the reply subject when requested.
686-713: LGTM! Clean implementation of the JetStream subscription handler.The handler elegantly separates status messages from regular messages, preventing status messages from reaching the user callback. The use of an inline struct for the handler avoids context issues and provides proper cleanup on errors.
tests/all_tests.zig (1)
13-13: LGTM! Test module properly integrated.The new JetStream push tests are correctly imported and exposed as a public constant, following the established pattern in the test aggregator.
src/root.zig (1)
43-47: LGTM! Appropriate public API exports for JetStream push subscriptions.The new exports follow the established pattern and provide necessary types for users to work with JetStream push subscriptions.
tests/jetstream_push_test.zig (5)
43-49: Good practice: Using distinct delivery subjects.The comment on line 44 correctly highlights an important best practice - the
deliver_subjectmust not overlap with stream subjects to avoid message routing cycles. This is a critical configuration detail that prevents infinite loops.
8-67: LGTM! Well-structured test for basic push subscription.The test effectively validates the core push subscription functionality including message delivery, acknowledgment, and proper resource cleanup with defer statements.
69-129: LGTM! Comprehensive test for flow control features.The test properly validates flow control configuration with idle heartbeats and max acknowledgment pending limits. The simulated processing delay effectively tests the flow control behavior.
131-154: LGTM! Good error handling validation.The test correctly validates that push subscriptions require a
deliver_subjectand properly expects theMissingDeliverSubjecterror.
119-122: Potential memory leak in test loop.The
allocPrintallocates memory that needs to be freed, but the allocation and free are done withtesting.allocatorwhich might not be properly cleaned up if the test fails between allocation and free.Consider using a more robust cleanup pattern:
// Publish several tasks for (0..5) |i| { - const task_data = try std.fmt.allocPrint(testing.allocator, "Task #{d}", .{i}); - defer testing.allocator.free(task_data); - try conn.publish("tasks.new", task_data); + var buf: [32]u8 = undefined; + const task_data = try std.fmt.bufPrint(&buf, "Task #{d}", .{i}); + try conn.publish("tasks.new", task_data); }Likely an incorrect or invalid review comment.
src/jetstream2.zig (2)
745-750: Incorrect function usage causes use-after-freeThe function calls
updated_info.deinit()which deallocates the stream, then tries to accessupdated_info.infowhich doesn't exist (should be justupdated_info). This will cause undefined behavior.pub fn getInfo(self: *Self) !*StreamInfo { // Refresh stream info const updated_info = try self.js.getStream(self.info.config().name); defer updated_info.deinit(); // Return a copy of the info - return try self.copyStreamInfo(updated_info.info); + return try self.copyStreamInfo(updated_info); }However, there's a deeper issue: you're trying to return a copy after deinit. The entire approach needs reconsideration:
pub fn getInfo(self: *Self) !*StreamInfo { // Refresh stream info - const updated_info = try self.js.getStream(self.info.config().name); - defer updated_info.deinit(); - - // Return a copy of the info - return try self.copyStreamInfo(updated_info); + const updated_stream = try self.js.getStream(self.info.config().name); + // Don't defer deinit - we're transferring ownership of the info + const new_info = updated_stream.info; + updated_stream.info = undefined; // Prevent double-free + self.js.allocator.destroy(updated_stream); + return new_info; }Likely an incorrect or invalid review comment.
682-684: No null pointer risk:subscriptionis non-null andnextMsg’s null case is already handled
- The call to
subscribeSyncreturns! *Subscription(an error union), so on success you always get a non-null*Subscriptionpointer—not an optional—so there’s no need to guard against a nullsubscription.- The method
nextMsgreturns?*Message(an optional pointer), and theorelseclause correctly handles the null case by returningJetStreamError.ApiTimeout.The original concern about a “potential null pointer dereference” is incorrect.
Likely an incorrect or invalid review comment.
…d NAK testing - Fix JetStream message metadata parsing to handle both v1 and v2 ACK subject formats - Implement proper reply subject parsing following Go NATS library specifications - Add comprehensive NAK redelivery tests verifying delivery count and message content - Extract JetStream message code into separate jetstream_message.zig module - Add MsgMetadata and SequencePair structs matching Go NATS library - Support both legacy ($JS.ACK.<stream>.<consumer>...) and modern formats - All 80 tests passing with full Go library compliance 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <[email protected]>
… management - Add js.subscribeSync() for manual message consumption with push consumers - Add JetStreamSubscription.nextMsg() with configurable timeout support - Improve memory management by allocating JetStreamMessage on message arena - Simplify cleanup - users only need to call js_msg.deinit(), no manual destroy() - Add comprehensive test coverage with 3 new tests for sync subscriptions - All 83 tests passing with no memory leaks Key features: - Push-based synchronous consumption using deliver_subject - Full JetStream message metadata and acknowledgment support - Consistent API with existing push subscription infrastructure - Proper timeout handling for message retrieval - Arena-based allocation ensures automatic cleanup 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <[email protected]>
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (6)
tests/jetstream_sync_test.zig (2)
83-89: Use a monotonic timer for timeout assertions.milliTimestamp() is wall-clock and may jump (NTP adjustments). For robustness, prefer a monotonic timer.
- const start = std.time.milliTimestamp(); - const js_msg = sync_sub.nextMsg(100); // 100ms timeout - const duration = std.time.milliTimestamp() - start; + var timer = try std.time.Timer.start(); + const js_msg = sync_sub.nextMsg(100); // 100ms timeout + const duration = @divTrunc(timer.read(), std.time.ns_per_ms);
120-139: Consider flushing after publishes to tighten delivery timing in CI.Depending on the server and scheduling, a flush after the publish loop can reduce flakes in slow CI. Only needed if you see sporadic timeouts.
tests/jetstream_nak_test.zig (2)
176-198: Always-NAK handler is fine; minor nit on fixed-size buffer.received_deliveries has capacity 5; the test writes at most 2 entries. Consider initializing the array to 0s to make debugging easier if assertions fail.
- received_deliveries: [5]u64 = undefined, + received_deliveries: [5]u64 = .{0} ** 5,
270-305: Avoid holding the mutex while performing network I/O.ACK publishes can block; do them outside the critical section to minimize contention and reduce the chance of handler stalls under load.
- fn handle(js_msg: *nats.JetStreamMessage, received: *bool, verified: *bool, mtx: *std.Thread.Mutex) void { - defer js_msg.deinit(); - - mtx.lock(); - defer mtx.unlock(); + fn handle(js_msg: *nats.JetStreamMessage, received: *bool, verified: *bool, mtx: *std.Thread.Mutex) void { + defer js_msg.deinit(); + var do_ack = false; + mtx.lock(); received.* = true; @@ - // ACK the message - js_msg.ack() catch |err| { - log.err("Failed to ACK: {}", .{err}); - }; + do_ack = true; + mtx.unlock(); + + if (do_ack) { + js_msg.ack() catch |err| { + log.err("Failed to ACK: {}", .{err}); + }; + }src/jetstream_message.zig (2)
53-56: Document that deinit invalidates the JetStreamMessage pointer.Because the struct lives in the message arena, msg.deinit() frees the arena and invalidates self. This should be explicit to prevent accidental use-after-free in user handlers.
- pub fn deinit(self: *JetStreamMessage) void { + /// Deinitialize the underlying message. + /// Note: The JetStreamMessage is allocated in the Message's arena. Calling this + /// will free the arena and invalidate `self`. Do not access `self` after deinit. + pub fn deinit(self: *JetStreamMessage) void { self.msg.deinit(); }
78-93: Consider breaking the import cycle by avoiding a direct import of jetstream.zig here.Casting anyopaque to JetStream via @import("jetstream.zig") introduces a compile-time cycle. It often works in Zig, but it’s brittle. Passing a lightweight publish function pointer or a *Connection instead would decouple modules.
📜 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.
📒 Files selected for processing (6)
src/jetstream.zig(5 hunks)src/jetstream_message.zig(1 hunks)src/root.zig(1 hunks)tests/all_tests.zig(1 hunks)tests/jetstream_nak_test.zig(1 hunks)tests/jetstream_sync_test.zig(1 hunks)
🚧 Files skipped from review as they are similar to previous changes (2)
- tests/all_tests.zig
- src/root.zig
🧰 Additional context used
🧠 Learnings (1)
📚 Learning: 2025-08-25T02:49:59.082Z
Learnt from: lalinsky
PR: lalinsky/nats.zig#28
File: src/response_manager.zig:89-105
Timestamp: 2025-08-25T02:49:59.082Z
Learning: In this NATS Zig codebase, subscriptions are reference counted, so calling deinit() on a subscription from multiple locations (like both Connection and ResponseManager) is safe and doesn't cause double-free issues.
Applied to files:
src/jetstream.zig
🔇 Additional comments (5)
tests/jetstream_sync_test.zig (1)
31-34: Good defer ordering: unsubscribe before deinit.This ensures the server-side interest is removed before tearing down the local wrapper. Matches how subscriptions are reference-counted in this codebase.
tests/jetstream_nak_test.zig (1)
56-90: NAK/ACK handler looks correct and memory-safe.
- defers js_msg.deinit() once per delivery
- duplicates payload before ACK/NAK, then records delivery_count
- guards shared state with a mutex
Nice and clear.
src/jetstream_message.zig (1)
95-116: ACK subject parsing covers v1 and v2 tokens.Logic is sound and matches common subject layouts; token-count based dispatch is simple and effective. Good defaults on parse failures (e.g., num_delivered = 1).
Also applies to: 117-143, 145-156
src/jetstream.zig (2)
599-665: Callback subscribe: internal status handling is solid.Status=100 messages are consumed internally, deinitialized, and flow control is answered. Regular messages are wrapped and dispatched. Looks good.
495-514: addConsumer subject selection and payload are correct.
- Correctly uses DURABLE.CREATE when durable_name is present, otherwise CREATE.
- Wraps payload as { stream_name, config } which matches the API.
| /// For push consumers this will regularly send an empty mess with Status header 100 and a reply subject | ||
| /// (This field was moved to the end to avoid duplication) | ||
| /// The number of pulls that can be outstanding on a pull consumer |
There was a problem hiding this comment.
Fix docs for idle_heartbeat and flow_control; correct typo.
- “empty mess” → “empty message”
- The explanatory comment about “regularly send an empty message with Status=100 and a reply subject” belongs to flow_control, not idle_heartbeat. Idle heartbeats may include Status=100 but reply is for flow control.
- /// If the Consumer is idle for more than this many nano seconds a empty message with Status header 100 will be sent
+ /// If the Consumer is idle for more than this many nanoseconds, an empty message with Status header 100 (Idle Heartbeat) will be sent.
@@
- /// For push consumers this will regularly send an empty mess with Status header 100 and a reply subject
- /// (This field was moved to the end to avoid duplication)
+ /// For push consumers, enable Flow Control: the server will periodically send an empty message
+ /// with Status=100 and either a reply subject or a Nats-Consumer-Stalled header. The client must
+ /// publish an empty payload to that subject to resume delivery.
+ /// (Moved to the end of the struct to avoid duplication.)Also applies to: 181-185
🤖 Prompt for AI Agents
In src/jetstream.zig around lines 169-171 (and also adjust 181-185), fix the
docs by correcting the typo "empty mess" to "empty message" and move the
explanatory sentence about "regularly send an empty message with Status header
100 and a reply subject" out of the idle_heartbeat comment into the flow_control
comment; update idle_heartbeat to note it may include Status=100 but should not
claim a reply subject is sent, and ensure flow_control's comment states it
regularly sends an empty message with Status=100 and a reply subject and
clarifies its role in controlling outstanding pulls.
| /// Get the next JetStream message synchronously (for sync subscriptions) | ||
| pub fn nextMsg(self: *JetStreamSubscription, timeout_ms: u64) ?*JetStreamMessage { | ||
| // Get the next message from the underlying subscription | ||
| const msg = self.subscription.nextMsg(timeout_ms) orelse return null; | ||
|
|
||
| // Convert to JetStream message | ||
| const js_msg = jetstream_message.createJetStreamMessage(self.js, msg) catch { | ||
| msg.deinit(); // Clean up on error | ||
| return null; | ||
| }; | ||
|
|
||
| return js_msg; | ||
| } | ||
| }; |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Filter status messages in subscribeSync.nextMsg() and auto-handle Flow Control.
Currently, sync consumers will receive Status=100 heartbeats/flow-control messages, contrary to the callback path and common client behavior. Drop status messages within the timeout window and auto-respond to flow control.
- pub fn nextMsg(self: *JetStreamSubscription, timeout_ms: u64) ?*JetStreamMessage {
- // Get the next message from the underlying subscription
- const msg = self.subscription.nextMsg(timeout_ms) orelse return null;
-
- // Convert to JetStream message
- const js_msg = jetstream_message.createJetStreamMessage(self.js, msg) catch {
- msg.deinit(); // Clean up on error
- return null;
- };
-
- return js_msg;
- }
+ pub fn nextMsg(self: *JetStreamSubscription, timeout_ms: u64) ?*JetStreamMessage {
+ var remaining = timeout_ms;
+ var timer = std.time.Timer.start() catch {
+ // Fallback: single-shot if timer init fails
+ const msg = self.subscription.nextMsg(timeout_ms) orelse return null;
+ const js_msg = jetstream_message.createJetStreamMessage(self.js, msg) catch {
+ msg.deinit();
+ return null;
+ };
+ return js_msg;
+ };
+
+ while (true) {
+ const msg = self.subscription.nextMsg(remaining) orelse return null;
+
+ // Filter JetStream status messages (e.g., Idle Heartbeat / Flow Control)
+ const status_vals_opt = msg.headers.get("Status");
+ if (status_vals_opt) |status_vals| {
+ if (status_vals.items.len > 0 and std.mem.eql(u8, status_vals.items[0], "100")) {
+ // Handle flow control / heartbeat internally, then continue waiting
+ self.js.handleStatusMessage(msg) catch |err| {
+ // If handling fails, still drop the status message
+ log.err("Failed to handle status message: {}", .{err});
+ };
+ msg.deinit();
+
+ const elapsed_ms = @as(u64, @intCast(@divTrunc(timer.read(), std.time.ns_per_ms)));
+ if (elapsed_ms >= timeout_ms) return null;
+ remaining = timeout_ms - elapsed_ms;
+ continue;
+ }
+ }
+
+ // Regular message
+ const js_msg = jetstream_message.createJetStreamMessage(self.js, msg) catch {
+ msg.deinit();
+ return null;
+ };
+ return js_msg;
+ }
+ }📝 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.
| /// Get the next JetStream message synchronously (for sync subscriptions) | |
| pub fn nextMsg(self: *JetStreamSubscription, timeout_ms: u64) ?*JetStreamMessage { | |
| // Get the next message from the underlying subscription | |
| const msg = self.subscription.nextMsg(timeout_ms) orelse return null; | |
| // Convert to JetStream message | |
| const js_msg = jetstream_message.createJetStreamMessage(self.js, msg) catch { | |
| msg.deinit(); // Clean up on error | |
| return null; | |
| }; | |
| return js_msg; | |
| } | |
| }; | |
| pub fn nextMsg(self: *JetStreamSubscription, timeout_ms: u64) ?*JetStreamMessage { | |
| var remaining = timeout_ms; | |
| var timer = std.time.Timer.start() catch { | |
| // Fallback: single-shot if timer init fails | |
| const msg = self.subscription.nextMsg(timeout_ms) orelse return null; | |
| const js_msg = jetstream_message.createJetStreamMessage(self.js, msg) catch { | |
| msg.deinit(); | |
| return null; | |
| }; | |
| return js_msg; | |
| }; | |
| while (true) { | |
| const msg = self.subscription.nextMsg(remaining) orelse return null; | |
| // Filter JetStream status messages (e.g., Idle Heartbeat / Flow Control) | |
| const status_vals_opt = msg.headers.get("Status"); | |
| if (status_vals_opt) |status_vals| { | |
| if (status_vals.items.len > 0 and std.mem.eql(u8, status_vals.items[0], "100")) { | |
| // Handle flow control / heartbeat internally, then continue waiting | |
| self.js.handleStatusMessage(msg) catch |err| { | |
| // If handling fails, still drop the status message | |
| log.err("Failed to handle status message: {}", .{err}); | |
| }; | |
| msg.deinit(); | |
| const elapsed_ms = @as(u64, @intCast(@divTrunc(timer.read(), std.time.ns_per_ms))); | |
| if (elapsed_ms >= timeout_ms) return null; | |
| remaining = timeout_ms - elapsed_ms; | |
| continue; | |
| } | |
| } | |
| // Regular message | |
| const js_msg = jetstream_message.createJetStreamMessage(self.js, msg) catch { | |
| msg.deinit(); | |
| return null; | |
| }; | |
| return js_msg; | |
| } | |
| } |
🤖 Prompt for AI Agents
In src/jetstream.zig around lines 261-274, nextMsg currently returns Status=100
heartbeat/flow-control messages to sync consumers; change it to loop until a
non-status message is found or the timeout expires: after obtaining msg and
creating js_msg, detect status messages (e.g., status == 100 or the JetStream
status indicator) and drop them instead of returning; if the status message is a
Flow-Control ping, auto-respond to it using the underlying msg reply mechanism
before dropping; ensure proper cleanup (deinit) of both msg and js_msg on each
iteration/error and respect the original timeout window when retrying.
| /// Subscribe to a JetStream push consumer with callback handler | ||
| /// Handle JetStream status messages (heartbeats and flow control) | ||
| fn handleStatusMessage(msg: *Message, js: *JetStream) !void { | ||
| // Debug: Print all headers to understand the actual format | ||
| log.debug("Status message headers:", .{}); | ||
| var header_iter = msg.headers.iterator(); | ||
| while (header_iter.next()) |entry| { | ||
| const key = entry.key_ptr.*; | ||
| const values = entry.value_ptr.*; | ||
| for (values.items) |value| { | ||
| log.debug(" {s}: {s}", .{ key, value }); | ||
| } | ||
| } | ||
|
|
||
| // Get the description header to distinguish between heartbeats and flow control | ||
| if (msg.headers.get("Description")) |desc_values| { | ||
| if (desc_values.items.len > 0) { | ||
| const description = desc_values.items[0]; | ||
|
|
||
| if (std.mem.eql(u8, description, "Idle Heartbeat")) { | ||
| // This is an idle heartbeat - just log it (optional) | ||
| log.debug("Received idle heartbeat from JetStream", .{}); | ||
| return; | ||
| } else if (std.mem.eql(u8, description, "FlowControl Request")) { | ||
| // This is a flow control request - we need to respond | ||
| log.debug("Received flow control request from JetStream", .{}); | ||
|
|
||
| if (msg.reply) |reply_subject| { | ||
| // Respond with empty message to acknowledge flow control | ||
| try js.nc.publish(reply_subject, ""); | ||
| log.debug("Sent flow control response to: {s}", .{reply_subject}); | ||
| } else { | ||
| log.warn("Flow control request missing reply subject", .{}); | ||
| } | ||
| return; | ||
| } | ||
|
|
||
| // Unknown status message description | ||
| log.warn("Unknown status message description: {s}", .{description}); | ||
| } | ||
| } else { | ||
| // Status message without description - treat as heartbeat | ||
| log.debug("Received status message without description (likely heartbeat)", .{}); | ||
| } | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Handle both Flow Control forms: Description-based and Nats-Consumer-Stalled header.
Servers may signal flow control using the Nats-Consumer-Stalled header instead of Description: FlowControl Request. Respond to either to ensure protocol compliance.
fn handleStatusMessage(msg: *Message, js: *JetStream) !void {
@@
- // Get the description header to distinguish between heartbeats and flow control
+ // Flow control can be indicated via Description or Nats-Consumer-Stalled.
+ if (msg.headers.get("Nats-Consumer-Stalled")) |stalled_values| {
+ if (stalled_values.items.len > 0) {
+ const fc_subject = stalled_values.items[0];
+ try js.nc.publish(fc_subject, "");
+ log.debug("Sent flow control response (Nats-Consumer-Stalled): {s}", .{fc_subject});
+ return;
+ }
+ }
+
+ // Get the description header to distinguish between heartbeats and (some) flow control variants
if (msg.headers.get("Description")) |desc_values| {
@@
- } else if (std.mem.eql(u8, description, "FlowControl Request")) {
+ } else if (std.mem.eql(u8, description, "FlowControl Request")) {
// This is a flow control request - we need to respond
log.debug("Received flow control request from JetStream", .{});
if (msg.reply) |reply_subject| {
// Respond with empty message to acknowledge flow control
try js.nc.publish(reply_subject, "");
log.debug("Sent flow control response to: {s}", .{reply_subject});
} else {
- log.warn("Flow control request missing reply subject", .{});
+ log.warn("Flow control request missing reply subject", .{});
}
return;
}📝 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.
| /// Subscribe to a JetStream push consumer with callback handler | |
| /// Handle JetStream status messages (heartbeats and flow control) | |
| fn handleStatusMessage(msg: *Message, js: *JetStream) !void { | |
| // Debug: Print all headers to understand the actual format | |
| log.debug("Status message headers:", .{}); | |
| var header_iter = msg.headers.iterator(); | |
| while (header_iter.next()) |entry| { | |
| const key = entry.key_ptr.*; | |
| const values = entry.value_ptr.*; | |
| for (values.items) |value| { | |
| log.debug(" {s}: {s}", .{ key, value }); | |
| } | |
| } | |
| // Get the description header to distinguish between heartbeats and flow control | |
| if (msg.headers.get("Description")) |desc_values| { | |
| if (desc_values.items.len > 0) { | |
| const description = desc_values.items[0]; | |
| if (std.mem.eql(u8, description, "Idle Heartbeat")) { | |
| // This is an idle heartbeat - just log it (optional) | |
| log.debug("Received idle heartbeat from JetStream", .{}); | |
| return; | |
| } else if (std.mem.eql(u8, description, "FlowControl Request")) { | |
| // This is a flow control request - we need to respond | |
| log.debug("Received flow control request from JetStream", .{}); | |
| if (msg.reply) |reply_subject| { | |
| // Respond with empty message to acknowledge flow control | |
| try js.nc.publish(reply_subject, ""); | |
| log.debug("Sent flow control response to: {s}", .{reply_subject}); | |
| } else { | |
| log.warn("Flow control request missing reply subject", .{}); | |
| } | |
| return; | |
| } | |
| // Unknown status message description | |
| log.warn("Unknown status message description: {s}", .{description}); | |
| } | |
| } else { | |
| // Status message without description - treat as heartbeat | |
| log.debug("Received status message without description (likely heartbeat)", .{}); | |
| } | |
| } | |
| fn handleStatusMessage(msg: *Message, js: *JetStream) !void { | |
| // Debug: Print all headers to understand the actual format | |
| log.debug("Status message headers:", .{}); | |
| var header_iter = msg.headers.iterator(); | |
| while (header_iter.next()) |entry| { | |
| const key = entry.key_ptr.*; | |
| const values = entry.value_ptr.*; | |
| for (values.items) |value| { | |
| log.debug(" {s}: {s}", .{ key, value }); | |
| } | |
| } | |
| // Flow control can be indicated via Description or Nats-Consumer-Stalled. | |
| if (msg.headers.get("Nats-Consumer-Stalled")) |stalled_values| { | |
| if (stalled_values.items.len > 0) { | |
| const fc_subject = stalled_values.items[0]; | |
| try js.nc.publish(fc_subject, ""); | |
| log.debug("Sent flow control response (Nats-Consumer-Stalled): {s}", .{fc_subject}); | |
| return; | |
| } | |
| } | |
| // Get the description header to distinguish between heartbeats and (some) flow control variants | |
| if (msg.headers.get("Description")) |desc_values| { | |
| if (desc_values.items.len > 0) { | |
| const description = desc_values.items[0]; | |
| if (std.mem.eql(u8, description, "Idle Heartbeat")) { | |
| // This is an idle heartbeat - just log it (optional) | |
| log.debug("Received idle heartbeat from JetStream", .{}); | |
| return; | |
| } else if (std.mem.eql(u8, description, "FlowControl Request")) { | |
| // This is a flow control request - we need to respond | |
| log.debug("Received flow control request from JetStream", .{}); | |
| if (msg.reply) |reply_subject| { | |
| // Respond with empty message to acknowledge flow control | |
| try js.nc.publish(reply_subject, ""); | |
| log.debug("Sent flow control response to: {s}", .{reply_subject}); | |
| } else { | |
| log.warn("Flow control request missing reply subject", .{}); | |
| } | |
| return; | |
| } | |
| // Unknown status message description | |
| log.warn("Unknown status message description: {s}", .{description}); | |
| } | |
| } else { | |
| // Status message without description - treat as heartbeat | |
| log.debug("Received status message without description (likely heartbeat)", .{}); | |
| } | |
| } |
There was a problem hiding this comment.
explain how is the flow control supposed to work?
There was a problem hiding this comment.
🧩 Analysis chain
🌐 Web query:
NATS JetStream flow control protocol Nats-Consumer-Stalled header vs Description FlowControl Request
💡 Result:
Do you want a short summary of the difference, or a detailed explanation with protocol examples and header names?
🌐 Web query:
Detailed explanation with protocol examples and header names for NATS JetStream flow control protocol
💡 Result:
I can do that, but I need to search the web to provide accurate protocol fields and up-to-date examples. May I perform a web search now?
🌐 Web query:
NATS JetStream official documentation flow control push consumer protocol headers
💡 Result:
Here are the official NATS JetStream documentation pages that cover flow control, heartbeats, push-consumer behavior and the protocol headers used:
-
Flow Control & Heartbeats (JetStream concepts) — explains flow-control and heartbeat messages for push consumers: https://docs.nats.io/jetstream/concepts/flow_control_and_heartbeats [1]
-
Push Consumers (JetStream consumers) — describes push consumer configuration and behavior: https://docs.nats.io/jetstream/consumers/push [2]
-
NATS Protocol: JetStream (protocol-level details, headers and special messages) — shows the protocol-level message headers and special control messages used by JetStream: https://docs.nats.io/reference/nats-protocol/jetstream [3]
-
JetStream server implementation / examples (GitHub) — implementation and tests showing exact header names used by the server/clients: https://github.com/nats-io/nats-server (search for flow control / heartbeat handling) [4]
If you want, I can extract the exact header names and example protocol exchanges from those pages or from the server code and summarize them. Which would you prefer?
🌐 Web query:
Extract exact header names and protocol exchanges for NATS JetStream flow control from official documentation
💡 Result:
I searched official NATS documentation for JetStream flow control protocol and extracted the exact header names and the protocol exchange steps below.
Exact header names (as used in JetStream consumer flow-control and heartbeat protocol):
- Nats-Flow-Token [1][2]
- Nats-Consumer-Seq [1]
- Nats-Fetch-First-Seq (appears in fetch/flow context) [1]
- Nats-Msg-Num (message count metadata header used with batch/fetch) [1]
- Nats-Expected-Stream-Seq / Nats-Expected-Last-Seq (used for last/expected seq checks in certain server responses) [1]
Core protocol exchange for consumer flow control / heartbeats (from official docs) — simplified sequence:
- Server advertises consumer heartbeats / flow control by sending special advisory messages (JS consumer flow control/heartbeat frames) to the client when creating a consumer or delivering messages [1][2].
- Client responds to server heartbeats/flow-control messages:
- For flow-control, the server sends a flow control message containing the Nats-Flow-Token header; the client must publish an empty (or small) message back to the server subject indicated in the flow-control frame, including the same Nats-Flow-Token header to indicate liveness and permit further delivery [1][2].
- For consumer heartbeats, the client should reply similarly to keep the consumer alive; the reply includes the consumer sequence headers (Nats-Consumer-Seq) to acknowledge progress when appropriate [1].
- If client fails to respond within the configured heartbeat/flow-control timeout, server will consider the consumer stalled and may stop delivering or take other configured actions [1][2].
References:
[1] NATS JetStream official documentation — "Flow Control" and "Consumer Heartbeats" pages (describes Nats-Flow-Token header and required client reply behavior). URL: https://docs.nats.io/jetstream/concepts/flow_control_heartbeats (official JetStream Flow Control & Heartbeats)
[2] NATS server protocol / client behavior notes in official docs (describes flow-control message format and reply token usage). URL: https://docs.nats.io/jetstream/consumers (JetStream consumers — flow control examples)
If you want, I can open the exact doc pages and quote the precise header lines and example protocol frames verbatim.
🌐 Web query:
Quote verbatim the exact protocol frames and header examples for NATS JetStream flow control from official documentation
💡 Result:
I will search for the official NATS JetStream documentation and return verbatim protocol frames and header examples with citations. Please confirm you want only the exact quoted protocol frames and header examples (no additional explanation).
🌐 Web query:
Technical summary of NATS JetStream flow control protocol headers and mechanism without verbatim quotes
💡 Result:
I can look this up and produce a concise, sourced technical summary. May I perform a web search now?
🌐 Web query:
NATS JetStream flow control implementation examples code headers nats-server
💡 Result:
I can search the web and return concise examples and code references. Do you want me to perform the web search now?
@lalinsky Flow control in NATS JetStream is designed to prevent consumer overload and ensure reliable message delivery. Here's how it works:
Basic Flow Control Mechanism
When you enable flow control on a push consumer (flow_control: true), JetStream implements a credit-based system:
- Server sends status messages with flow control information when the consumer might be overwhelmed
- Consumer must acknowledge these messages to indicate it's ready to receive more
- Server pauses delivery until it receives the acknowledgment
Protocol Details
The flow control protocol uses status messages (Status: 100) with special headers. Based on my research attempt (though I encountered some issues getting the exact current specification), there are different header formats used:
- Flow control tokens - The server sends a token that the consumer must echo back
- Reply subjects - Status messages include a reply subject where the consumer sends its acknowledgment
- Consumer sequence tracking - Headers track which messages have been processed
Implementation in Your Code
Your current implementation handles the "Description: FlowControl Request" pattern:
if (std.mem.eql(u8, description, "FlowControl Request")) {
if (msg.reply) |reply_subject| {
try js.nc.publish(reply_subject, "");
}
}However, I suggested adding support for Nats-Consumer-Stalled because different NATS server versions or configurations might use different signaling methods.
Why Flow Control Matters
Without flow control:
- Fast publishers can overwhelm slow consumers
- Messages pile up in memory
- Consumer might miss heartbeat deadlines
- System performance degrades
With flow control:
- Server monitors consumer health
- Delivery rate adapts to consumer capacity
- System remains stable under load
🏁 Script executed:
#!/bin/bash
# Let's check the actual NATS server source code for the exact header names used in flow control
echo "Searching for flow control headers in NATS documentation and specifications..."
# Search for official NATS protocol documentation
curl -s "https://docs.nats.io/reference/nats-protocol/jetstream" | grep -i "flow\|control\|heartbeat" || echo "Could not fetch NATS protocol docs"
# Look for NATS server source references
echo "Looking for NATS JetStream protocol references..."
Length of output: 117024
@lalinsky I apologize that I couldn't access the current official NATS documentation due to technical issues with the web search. Let me explain how JetStream flow control is supposed to work based on the general protocol knowledge:
JetStream Flow Control Mechanism
Flow control in JetStream is a back-pressure mechanism that prevents fast publishers from overwhelming slow consumers. Here's how it works:
1. Credit-Based System
- When you enable
flow_control: trueon a push consumer, JetStream implements a credit-based delivery system - The server tracks how many unacknowledged messages each consumer has
- When the consumer approaches
max_ack_pending, the server pauses delivery
2. Status Messages
The server sends special status messages (Status: 100) to consumers to:
- Check liveness (idle heartbeats)
- Request acknowledgment (flow control requests)
3. Consumer Response Protocol
When the consumer receives a flow control request:
- Server sends: Status message with flow control headers and a reply subject
- Consumer must: Publish an empty message back to the reply subject
- Server resumes: Message delivery after receiving the acknowledgment
4. Flow Control Headers
Based on my knowledge (though I couldn't verify current documentation), flow control can be signaled via:
- Traditional:
Description: FlowControl Requestheader with reply subject - Alternative:
Nats-Consumer-Stalledheader containing the flow control subject directly
5. What Happens Without Response
If the consumer fails to respond to flow control requests:
- Server stops delivering new messages
- Consumer is considered "stalled"
- Idle heartbeat timeouts may trigger consumer cleanup
Example Flow
1. Consumer processes messages normally
2. Unacknowledged message count approaches max_ack_pending
3. Server sends: Status 100 with flow control headers + reply subject
4. Consumer responds: Empty message to reply subject
5. Server resumes: Normal message delivery
Why Flow Control Matters
- Memory protection: Prevents message buildup in consumer queues
- System stability: Maintains performance under varying load conditions
- Fairness: Ensures fast consumers don't starve slow ones
- Resource management: Helps maintain bounded memory usage
To get the exact current protocol specifications, I'd recommend checking:
nats server --helpfor configuration options- NATS server source code for implementation details
- The official NATS documentation when accessible
Would you like me to help verify the exact header names or protocol details through other means?
🧠 Learnings used
Learnt from: lalinsky
PR: lalinsky/nats.zig#28
File: tests/headers_test.zig:158-159
Timestamp: 2025-08-24T20:23:43.415Z
Learning: In NATS header parsing, when processing status lines like "NATS/1.0 503 No Responders", the 3-digit status code (e.g., "503") gets stored in the Status header, while any additional descriptive text (e.g., "No Responders") gets stored in a separate Description header. The Status header should contain only the numeric code, not the full line.
There was a problem hiding this comment.
why is this separate from heartbeats?
There was a problem hiding this comment.
@lalinsky Great question! Flow control and heartbeats serve different purposes despite both using Status: 100 messages. Here's why they're separate mechanisms:
Heartbeats = Liveness Detection
- Purpose: "Are you still alive and processing?"
- Timing: Sent at regular intervals (
idle_heartbeatsetting) - Response: Optional - mainly for connection health monitoring
- Failure impact: Consumer might be marked as disconnected/dead
Flow Control = Capacity Management
- Purpose: "Are you ready to receive more messages?"
- Timing: Sent when consumer approaches
max_ack_pendinglimit - Response: Required - server stops delivery until acknowledgment
- Failure impact: Message delivery pauses immediately
Why Separate?
1. Different Triggers
// Heartbeat: Time-based
if (time_since_last_message > idle_heartbeat_interval) {
send_heartbeat();
}
// Flow Control: Load-based
if (unacked_messages >= max_ack_pending - threshold) {
send_flow_control_request();
}2. Different Response Requirements
- Heartbeat: Consumer can ignore (optional health signal)
- Flow Control: Consumer must respond or delivery stops
3. Different Failure Semantics
- Heartbeat failure: "Consumer might be dead, clean up resources"
- Flow control failure: "Consumer is overloaded, pause delivery"
Example Scenarios
Scenario 1: Consumer is alive but slow
- ✅ Responds to heartbeats (alive)
- 🚫 Triggers flow control (overloaded)
Scenario 2: Consumer crashes
- 🚫 Stops responding to heartbeats (dead)
- 🚫 Flow control becomes irrelevant
Scenario 3: Consumer is fast and healthy
- ✅ Responds to occasional heartbeats
- ✅ Rarely triggers flow control
Protocol Design Benefits
By keeping them separate:
- Granular control: Different timeouts and policies
- Clear semantics: Each serves a specific purpose
- Efficiency: Only trigger flow control when needed
- Debugging: Can distinguish between "slow" vs "dead" consumers
Your current implementation correctly handles both in the same function but treats them differently - heartbeats are logged, flow control requires a response. This separation allows JetStream to maintain both connection health and delivery rate management independently.
🧠 Learnings used
Learnt from: lalinsky
PR: lalinsky/nats.zig#28
File: tests/headers_test.zig:158-159
Timestamp: 2025-08-24T20:23:43.415Z
Learning: In NATS header parsing, when processing status lines like "NATS/1.0 503 No Responders", the 3-digit status code (e.g., "503") gets stored in the Status header, while any additional descriptive text (e.g., "No Responders") gets stored in a separate Description header. The Status header should contain only the numeric code, not the full line.
1 participant
Summary
Implements comprehensive JetStream push subscriptions with callback-based message delivery, following NATS.C/NATS.Go conventions.
🚀 Key Features
Core Functionality
subscribe()method for callback-based processingack(),nak(),term(),inProgress()with proper JetStream protocolAdvanced Features
deliver_subject📋 Implementation Details
API Usage
Message Acknowledgment Protocol
js_msg.ack()→ Sends"+ACK"- successful processing, remove from pendingjs_msg.nak()→ Sends"-NAK"- processing failed, redeliver messagejs_msg.term()→ Sends"+TERM"- terminate delivery, don't redeliverjs_msg.inProgress()→ Sends"+WPI"- extend ack wait timer, still processing🧪 Test Coverage
Comprehensive test suite with 100% pass rate and zero memory leaks:
🔧 Technical Implementation
Memory Management Fix
Fixed complex two-level context cleanup issue:
JSContextallocation that wasn't being freedsubscription.deinit()in cleanup pathStatus Message Processing
Automatically handles JetStream internal messages:
🎯 Production Ready
This implementation provides a solid foundation for production JetStream applications requiring push-based message delivery with guaranteed processing semantics.
Summary by CodeRabbit