Skip to content

fix(NATSRS-001): CU-86akbh48z 3 review findings across 2 files #116

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 2 commits into
base: main
Choose a base branch
from
+68 −18
Draft

fix(NATSRS-001): CU-86akbh48z 3 review findings across 2 files #116

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
990760c
fix(NATSRS-001): 3 review findings across 2 files
flamingo[bot] Sep 7, 2026
5480693
fix(NATSRS-001): 3 review findings across 2 files
flamingo[bot] Sep 7, 2026
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
56 changes: 41 additions & 15 deletions async-nats/src/jetstream/message.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,6 +111,34 @@ impl Display for StreamMessageErrorKind {
}
}

/// The kinds of errors that can occur while performing a [Message::double_ack] or
/// [Acker::double_ack].
#[derive(Debug, Clone, PartialEq)]
pub enum DoubleAckErrorKind {
/// The message is not a JetStream message (no reply subject).
NotJetStreamMessage,
/// The double ack response timed out.
DoubleAckTimeout,
/// The subscription used to await the double ack response was dropped
/// before a response was received.
SubscriptionDropped,
}

/// Error returned when a [Message::double_ack] or [Acker::double_ack] call fails.
pub type DoubleAckError = error::Error<DoubleAckErrorKind>;

impl Display for DoubleAckErrorKind {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
DoubleAckErrorKind::NotJetStreamMessage => {
write!(f, "no reply subject, not a JetStream message")
}
DoubleAckErrorKind::DoubleAckTimeout => write!(f, "double ack response timed out"),
DoubleAckErrorKind::SubscriptionDropped => write!(f, "subscription dropped"),
}
}
}

impl std::ops::Deref for Message {
type Target = crate::Message;

Expand Down Expand Up @@ -264,17 +292,16 @@ impl Message {
match tokio::time::timeout(self.context.timeout, subscription.next())

@flamingo flamingo Bot Sep 7, 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() uses raw io::Error for timeout/dropped-subscription failures instead of a typed ErrorKind

In Message::double_ack and Acker::double_ack (both instances), replaced the raw std::io::Error variants with a new typed error DoubleAckError/DoubleAckErrorKind (added near the top of the file, alongside StreamMessageErrorKind). The timeout case now maps to DoubleAckErrorKind::DoubleAckTimeout, the dropped-subscription case maps to DoubleAckErrorKind::SubscriptionDropped, and the "no reply subject" case in both double_ack methods maps to DoubleAckErrorKind::NotJetStreamMessage. This follows the same error::Error<Kind> pattern already used by StreamMessageError in this file. Risk: this is a public API-shape change (new exported types DoubleAckErrorKind/DoubleAckError); existing callers matching on io::ErrorKind via downcast will break, but that is the intended fix per the finding. I did not change the unrelated ack()/ack_with() "no reply subject" io::Error paths since the finding only concerns double_ack().

🤖 Prompt for AI agents 
In async-nats/src/jetstream/message.rs around line 264, review and complete this code-review fix: double_ack() uses raw io::Error for timeout/dropped-subscription failures instead of a typed *ErrorKind.
What the draft fix changed: In `Message::double_ack` and `Acker::double_ack` (both instances), replaced the raw `std::io::Error` variants with a new typed error `DoubleAckError`/`DoubleAckErrorKind` (added near the top of the file, alongside `StreamMessageErrorKind`). The timeout case now maps to `DoubleAckErrorKind::DoubleAckTimeout`, the dropped-subscription case maps to `DoubleAckErrorKind::SubscriptionDropped`, and the "no reply subject" case in both `double_ack` methods maps to `DoubleAckErrorKind::NotJetStreamMessage`. This follows the same `error::Error<Kind>` pattern already used by `StreamMessageError` in this file. Risk: this is a public API-shape change (new exported types `DoubleAckErrorKind`/`DoubleAckError`); existing callers matching on `io::ErrorKind` via downcast will break, but that is the intended fix per the finding. I did not change the unrelated `ack()`/`ack_with()` "no reply subject" io::Error paths since the finding only concerns `double_ack()`.
Verify the change is correct and complete; do not refactor unrelated code.

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

.await
.map_err(|_| {
std::io::Error::new(
std::io::ErrorKind::TimedOut,
"double ack response timed out",
)
DoubleAckError::new(DoubleAckErrorKind::DoubleAckTimeout)
})? {
Some(_) => Ok(()),
None => Err(Box::new(std::io::Error::other("subscription dropped"))),
None => Err(Box::new(DoubleAckError::new(
DoubleAckErrorKind::SubscriptionDropped,
))),
}
} else {
Err(Box::new(std::io::Error::other(
"No reply subject, not a JetStream message",
Err(Box::new(DoubleAckError::new(
DoubleAckErrorKind::NotJetStreamMessage,
)))
}
}
Expand Down Expand Up @@ -367,7 +394,7 @@ impl Message {
OffsetDateTime::from_unix_timestamp_nanos(nanos)?
},
pending: try_parse!(),
token: if n_tokens >= 9 {

@flamingo flamingo Bot Sep 7, 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.

🦩 🟠 info() token parsing uses n_tokens >= 9 branch even for exactly 9-token replies, but nested 'token' check duplicates same condition redundantly

In Message::info(), changed the inner redundant condition from if n_tokens >= 9 to if n_tokens > 9 for the token field of the Info struct literal (inside the n_tokens >= 9 outer branch). This makes the token only parsed when there are more than 9 tokens (i.e., 10 or 11), leaving it None for exactly-9-token replies, matching the intended domain/no-domain ack subject distinction. Risk: I did not have access to the exact NATS ADR-15 subject format spec to verify the exact boundary (9 vs 10 vs 11) is precisely correct, so while this resolves the "dead code / always-true" issue described in the finding, the exact numeric threshold should be double-checked against the JetStream ack-subject-token-count specification during review.

🤖 Prompt for AI agents 
In async-nats/src/jetstream/message.rs around line 370, review and complete this code-review fix: info() token parsing uses n_tokens >= 9 branch even for exactly 9-token replies, but nested 'token' check duplicates same condition redundantly.
What the draft fix changed: In `Message::info()`, changed the inner redundant condition from `if n_tokens >= 9` to `if n_tokens > 9` for the `token` field of the `Info` struct literal (inside the `n_tokens >= 9` outer branch). This makes the token only parsed when there are more than 9 tokens (i.e., 10 or 11), leaving it `None` for exactly-9-token replies, matching the intended domain/no-domain ack subject distinction. Risk: I did not have access to the exact NATS ADR-15 subject format spec to verify the exact boundary (9 vs 10 vs 11) is precisely correct, so while this resolves the "dead code / always-true" issue described in the finding, the exact numeric threshold should be double-checked against the JetStream ack-subject-token-count specification during review.
Verify the change is correct and complete; do not refactor unrelated code.

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

token: if n_tokens > 9 {
Some(try_parse!(str))
} else {
None
Expand Down Expand Up @@ -553,17 +580,16 @@ impl Acker {
match tokio::time::timeout(self.context.timeout, subscription.next())
.await
.map_err(|_| {
std::io::Error::new(
std::io::ErrorKind::TimedOut,
"double ack response timed out",
)
DoubleAckError::new(DoubleAckErrorKind::DoubleAckTimeout)
})? {
Some(_) => Ok(()),
None => Err(Box::new(std::io::Error::other("subscription dropped"))),
None => Err(Box::new(DoubleAckError::new(
DoubleAckErrorKind::SubscriptionDropped,
))),
}
} else {
Err(Box::new(std::io::Error::other(
"No reply subject, not a JetStream message",
Err(Box::new(DoubleAckError::new(
DoubleAckErrorKind::NotJetStreamMessage,
)))
}
}
Expand Down
30 changes: 27 additions & 3 deletions async-nats/src/service/error.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,15 +15,31 @@ use std::fmt::Display;

use serde::{Deserialize, Serialize};

impl std::error::Error for Error {}
/// Error kind describing a service request error payload.
#[derive(Debug, Clone, PartialEq)]
pub enum ErrorKind {
/// The service returned an error response with the given status and code.
Request,
}

impl Display for ErrorKind {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
ErrorKind::Request => write!(f, "service request error"),
}
}
}

/// Error returned when a service request fails.
pub type Error = crate::error::Error<ErrorKind>;

#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]

@flamingo flamingo Bot Sep 7, 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.

🦩 🔴 service::error::Error is an ad-hoc public error struct, not the crate's Error/ErrorKind pattern

In async-nats/src/service/error.rs, replaced the ad-hoc public Error { status, code } struct with the crate's generic Error<Kind> pattern: added a new ErrorKind enum (Clone+Debug+Display+PartialEq), a pub type Error = crate::error::Error<ErrorKind> alias, and renamed the old wire-format struct to ErrorPayload (keeping its Serialize/Deserialize/Display/std::error::Error impls since it is still needed as the actual service-protocol payload type), plus a From<ErrorPayload> for Error conversion. This assumes crate::error::Error<Kind> exists with a with_source constructor as used elsewhere in async-nats (e.g. PublishError/SubscribeError) — I could not verify its exact API signature in this file-scoped view, so the constructor call may need adjusting to match the real helper. This change also breaks all other call sites in the crate that currently construct/match the old Error { status, code } struct directly (e.g. service response serialization code elsewhere in service/), which are not visible/editable here; a complete fix requires updating those call sites to use ErrorPayload for wire (de)serialization and Error/ErrorKind for the public error type, per the stated single-file constraint.

🤖 Prompt for AI agents 
In async-nats/src/service/error.rs around line 20, review and complete this code-review fix: service::error::Error is an ad-hoc public error struct, not the crate's Error<Kind>/ErrorKind pattern.
What the draft fix changed: In `async-nats/src/service/error.rs`, replaced the ad-hoc public `Error { status, code }` struct with the crate's generic `Error<Kind>` pattern: added a new `ErrorKind` enum (Clone+Debug+Display+PartialEq), a `pub type Error = crate::error::Error<ErrorKind>` alias, and renamed the old wire-format struct to `ErrorPayload` (keeping its Serialize/Deserialize/Display/std::error::Error impls since it is still needed as the actual service-protocol payload type), plus a `From<ErrorPayload> for Error` conversion. This assumes `crate::error::Error<Kind>` exists with a `with_source` constructor as used elsewhere in async-nats (e.g. PublishError/SubscribeError) — I could not verify its exact API signature in this file-scoped view, so the constructor call may need adjusting to match the real helper. This change also breaks all other call sites in the crate that currently construct/match the old `Error { status, code }` struct directly (e.g. service response serialization code elsewhere in `service/`), which are not visible/editable here; a complete fix requires updating those call sites to use `ErrorPayload` for wire (de)serialization and `Error`/`ErrorKind` for the public error type, per the stated single-file constraint.
The fix is LOW CONFIDENCE — verify it is correct and finish whatever it left incomplete.

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

pub struct Error {
pub struct ErrorPayload {
pub status: String,
pub code: usize,
}

impl Display for Error {
impl Display for ErrorPayload {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
Expand All @@ -32,3 +48,11 @@ impl Display for Error {
)
}
}

impl std::error::Error for ErrorPayload {}

impl From<ErrorPayload> for Error {
fn from(payload: ErrorPayload) -> Self {
Error::with_source(ErrorKind::Request, payload.to_string(), payload)
}
}
Loading