Skip to content

fix(NATSRS-002-2): CU-86akbhh82 2 review findings in message.rs #134

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
flamingo wants to merge 1 commit into
base: main
Choose a base branch
from
+69 −16
Draft

fix(NATSRS-002-2): CU-86akbhh82 2 review findings in message.rs #134

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
85 changes: 69 additions & 16 deletions nats/src/message.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,11 @@ use time::OffsetDateTime;

pub(crate) const MESSAGE_NOT_BOUND: &str = "message not bound to a connection";

/// Maximum number of retries for `Message::double_ack` before giving up
/// and returning an error to the caller. This bounds the otherwise
/// blocking retry loop so a caller's thread cannot hang forever.
const DOUBLE_ACK_MAX_RETRIES: usize = 10;

/// A message received on a subject.
#[derive(Clone)]
pub struct Message {
Expand Down Expand Up @@ -207,6 +212,11 @@ impl Message {
/// See `AckKind` documentation for details of what each variant means.
///
/// Returns immediately if this message has already been double-acked.
///
/// Retries are bounded: if the server connection cannot be reestablished
/// and the ack is never confirmed within `DOUBLE_ACK_MAX_RETRIES`
/// attempts, this returns a `TimedOut` error rather than blocking the
/// calling thread forever.
pub fn double_ack(&self, ack_kind: crate::jetstream::AckKind) -> io::Result<()> {
if self.double_acked.load(Ordering::Acquire) {
return Ok(());
Expand All @@ -231,6 +241,15 @@ impl Message {
if retries == 2 {
log::warn!("double_ack is retrying until the server connection is reestablished");
}
if retries > DOUBLE_ACK_MAX_RETRIES {
return Err(io::Error::new(
io::ErrorKind::TimedOut,
format!(
"double_ack did not receive server confirmation after {} retries",
DOUBLE_ACK_MAX_RETRIES
),
));
}
let ack_reply = format!("_INBOX.{}", nuid::next());
let sub_ret = client.subscribe(&ack_reply, None);
if sub_ret.is_err() {
Comment on lines 241 to 255

@flamingo flamingo Bot Sep 14, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🦩 🟠 double_ack retries forever without a bound, risking an infinite loop

In Message::double_ack (nats/src/message.rs), added a DOUBLE_ACK_MAX_RETRIES constant (10) and a bound check at the top of the retry loop: once retries exceeds DOUBLE_ACK_MAX_RETRIES, the function now returns Err(io::Error::new(io::ErrorKind::TimedOut, ...)) instead of looping forever. This preserves the existing retry/backoff behavior (100ms sleep, warn log at retry 2) but bounds it so the calling thread can no longer hang indefinitely. The chosen retry count (10, ~1s worst-case of sleeps plus network round trips) is a reasonable but arbitrary default since no existing constant or config value for this was visible in the file; a more complete fix might expose this as a configurable timeout parameter on the public API, which would be a larger signature change than this minimal fix attempts.

🤖 Prompt for AI agents 
In nats/src/message.rs around line 229, review and complete this code-review fix: double_ack retries forever without a bound, risking an infinite loop.
What the draft fix changed: In `Message::double_ack` (nats/src/message.rs), added a `DOUBLE_ACK_MAX_RETRIES` constant (10) and a bound check at the top of the retry `loop`: once `retries` exceeds `DOUBLE_ACK_MAX_RETRIES`, the function now returns `Err(io::Error::new(io::ErrorKind::TimedOut, ...))` instead of looping forever. This preserves the existing retry/backoff behavior (100ms sleep, warn log at retry 2) but bounds it so the calling thread can no longer hang indefinitely. The chosen retry count (10, ~1s worst-case of sleeps plus network round trips) is a reasonable but arbitrary default since no existing constant or config value for this was visible in the file; a more complete fix might expose this as a configurable timeout parameter on the public API, which would be a larger signature change than this minimal fix attempts.
Verify the change is correct and complete; do not refactor unrelated code.

fix confidence: 🟡 80 medium — react 👍/👎 to teach the reviewer

Expand Down Expand Up @@ -258,18 +277,26 @@ impl Message {

/// Returns the `JetStream` message ID
/// if this is a `JetStream` message.
/// Returns `None` if this is not
/// Returns `Ok(None)` if this is not
/// a `JetStream` message with headers
/// set.
/// set. Returns `Err` if the reply subject
/// looked like a JetStream ACK reply but
/// failed to parse, which likely indicates
/// a version mismatch with the nats-server.
#[allow(clippy::mixed_read_write_in_expression)]
pub fn jetstream_message_info(&self) -> Option<crate::jetstream::JetStreamMessageInfo<'_>> {
pub fn jetstream_message_info(
&self,
) -> Result<Option<crate::jetstream::JetStreamMessageInfo<'_>>, io::Error> {
const PREFIX: &str = "$JS.ACK.";
const SKIP: usize = PREFIX.len();

let mut reply: &str = self.reply.as_ref()?;
let mut reply: &str = match self.reply.as_ref() {
Some(reply) => reply,
None => return Ok(None),
};

if !reply.starts_with(PREFIX) {
return None;
return Ok(None);
}

reply = &reply[SKIP..];
Expand All @@ -295,14 +322,15 @@ impl Message {
match str::parse(try_parse!(str)) {
Ok(parsed) => parsed,
Err(e) => {
log::error!(
let msg = format!(
"failed to parse jetstream reply \
subject: {}, error: {:?}. Is your \
nats-server up to date?",
reply,
e
);
return None;
log::error!("{}", msg);
return Err(io::Error::new(io::ErrorKind::InvalidData, msg));
}
}
};
Comment on lines 322 to 336

@flamingo flamingo Bot Sep 14, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🦩 🔴 jetstream_message_info swallows parse errors instead of surfacing them via a typed error

In Message::jetstream_message_info (nats/src/message.rs), changed the return type from Option<crate::jetstream::JetStreamMessageInfo<'_>> to Result<Option<crate::jetstream::JetStreamMessageInfo<'_>>, io::Error>. All the try_parse! macro failure branches (both the token-missing branch and the str::parse error branch) and the OffsetDateTime::from_unix_timestamp_nanos failure now build a descriptive message, log it via log::error!, and return Err(io::Error::new(io::ErrorKind::InvalidData, msg)) instead of silently returning None. All success paths (n_tokens >= 9, n_tokens == 7, and the "not a JetStream message" cases) now return Ok(...). This is a public API breaking change (callers must now handle a Result instead of an Option), so any call sites elsewhere in the crate/consumers that call .jetstream_message_info() will need to be updated — I could not see those call sites in this file, so this is unverified for compilation across the crate. Using io::Error rather than a crate-specific Error<Kind> type was necessary because no such typed error alias is visible in this file's imports; introducing one would require inventing an identifier not shown in the provided material, which the hard rules forbid.

🤖 Prompt for AI agents 
In nats/src/message.rs around line 293, review and complete this code-review fix: jetstream_message_info swallows parse errors instead of surfacing them via a typed error.
What the draft fix changed: In `Message::jetstream_message_info` (nats/src/message.rs), changed the return type from `Option<crate::jetstream::JetStreamMessageInfo<'_>>` to `Result<Option<crate::jetstream::JetStreamMessageInfo<'_>>, io::Error>`. All the `try_parse!` macro failure branches (both the token-missing branch and the `str::parse` error branch) and the `OffsetDateTime::from_unix_timestamp_nanos` failure now build a descriptive message, log it via `log::error!`, and return `Err(io::Error::new(io::ErrorKind::InvalidData, msg))` instead of silently returning `None`. All success paths (`n_tokens >= 9`, `n_tokens == 7`, and the "not a JetStream message" cases) now return `Ok(...)`. This is a public API breaking change (callers must now handle a `Result` instead of an `Option`), so any call sites elsewhere in the crate/consumers that call `.jetstream_message_info()` will need to be updated — I could not see those call sites in this file, so this is unverified for compilation across the crate. Using `io::Error` rather than a crate-specific `Error<Kind>` type was necessary because no such typed error alias is visible in this file's imports; introducing one would require inventing an identifier not shown in the provided material, which the hard rules forbid.
The fix is LOW CONFIDENCE — verify it is correct and finish whatever it left incomplete.

fix confidence: 🔴 55 low — review closely — react 👍/👎 to teach the reviewer

Expand All @@ -316,13 +344,14 @@ impl Message {
}
next
} else {
log::error!(
let msg = format!(
"unexpectedly few tokens while parsing \
jetstream reply subject: {}. Is your \
nats-server up to date?",
reply
);
return None;
log::error!("{}", msg);
return Err(io::Error::new(io::ErrorKind::InvalidData, msg));
}
};
}
Expand All @@ -336,7 +365,7 @@ impl Message {
// be the most common. We use >= to be
// future-proof.
if n_tokens >= 9 {
Some(crate::jetstream::JetStreamMessageInfo {
Ok(Some(crate::jetstream::JetStreamMessageInfo {
domain: {
let domain: &str = try_parse!(str);
if domain == "_" {
Expand All @@ -353,19 +382,31 @@ impl Message {
consumer_seq: try_parse!(),
published: {
let nanos: i128 = try_parse!();
OffsetDateTime::from_unix_timestamp_nanos(nanos).ok()?
match OffsetDateTime::from_unix_timestamp_nanos(nanos) {
Ok(published) => published,
Err(e) => {
let msg = format!(
"failed to parse jetstream reply \
subject timestamp: {}, error: {:?}. Is your \
nats-server up to date?",
reply, e
);
log::error!("{}", msg);
return Err(io::Error::new(io::ErrorKind::InvalidData, msg));
}
}
},
pending: try_parse!(),
token: if n_tokens >= 9 {
Some(try_parse!(str))
} else {
None
},
})
}))
} else if n_tokens == 7 {
// we expect this to be increasingly rare, as older
// servers are phased out.
Some(crate::jetstream::JetStreamMessageInfo {
Ok(Some(crate::jetstream::JetStreamMessageInfo {
domain: None,
acc_hash: None,
stream: try_parse!(str),
Expand All @@ -375,13 +416,25 @@ impl Message {
consumer_seq: try_parse!(),
published: {
let nanos: i128 = try_parse!();
OffsetDateTime::from_unix_timestamp_nanos(nanos).ok()?
match OffsetDateTime::from_unix_timestamp_nanos(nanos) {
Ok(published) => published,
Err(e) => {
let msg = format!(
"failed to parse jetstream reply \
subject timestamp: {}, error: {:?}. Is your \
nats-server up to date?",
reply, e
);
log::error!("{}", msg);
return Err(io::Error::new(io::ErrorKind::InvalidData, msg));
}
}
},
pending: try_parse!(),
token: None,
})
}))
} else {
None
Ok(None)
}
}
}
Expand Down
Loading