Skip to content

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

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
+111 −58
Draft

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

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
169 changes: 111 additions & 58 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,66 @@ impl Display for StreamMessageErrorKind {
}
}

/// Error kind returned by [Message::ack], [Message::ack_with], [Message::double_ack],
/// [Acker::ack], [Acker::ack_with] and [Acker::double_ack].
#[derive(Debug, Clone, PartialEq)]
pub enum AckErrorKind {
/// The message does not have a reply subject, so it is not a JetStream message.
MissingReplySubject,
/// Timed out while waiting for double-ack confirmation from the server.
TimedOut,
/// The double-ack subscription was dropped before a response was received.
Dropped,
/// An error occurred while publishing or subscribing on the underlying client.
Other,
}

/// Error returned when acknowledging a message fails.
pub type AckError = error::Error<AckErrorKind>;

impl Display for AckErrorKind {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
AckErrorKind::MissingReplySubject => {
write!(f, "no reply subject, not a JetStream message")
}
AckErrorKind::TimedOut => write!(f, "double ack response timed out"),
AckErrorKind::Dropped => write!(f, "subscription dropped"),
AckErrorKind::Other => write!(f, "error acknowledging message"),
}
}
}

/// Error kind returned by [Message::info].
#[derive(Debug, Clone, PartialEq)]
pub enum InfoErrorKind {
/// The message does not have a reply subject.
MissingReplySubject,
/// The reply subject does not start with the expected JetStream ack prefix.
MissingPrefix,
/// The reply subject does not contain enough tokens to be parsed.
TooFewTokens,
/// The reply subject contains an unexpected number of tokens.
BadTokenNumber,
/// A token in the reply subject could not be parsed into the expected type.
ParseError,
}

/// Error returned when parsing [Info] out of a message's reply subject fails.
pub type InfoError = error::Error<InfoErrorKind>;

impl Display for InfoErrorKind {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
InfoErrorKind::MissingReplySubject => write!(f, "did not found reply subject"),
InfoErrorKind::MissingPrefix => write!(f, "did not found proper prefix"),
InfoErrorKind::TooFewTokens => write!(f, "too few tokens"),
InfoErrorKind::BadTokenNumber => write!(f, "bad token number"),
InfoErrorKind::ParseError => write!(f, "failed to parse token"),
}
}
}

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

Expand Down Expand Up @@ -168,17 +228,15 @@ impl Message {
/// # Ok(())
/// # }
/// ```
pub async fn ack(&self) -> Result<(), Error> {
pub async fn ack(&self) -> Result<(), AckError> {
if let Some(ref reply) = self.reply {
self.context
.client
.publish(reply.clone(), "".into())
.map_err(Error::from)
.map_err(|err| AckError::with_source(AckErrorKind::Other, err))
.await
} else {
Err(Box::new(std::io::Error::other(
"No reply subject, not a JetStream message",
)))
Err(AckError::new(AckErrorKind::MissingReplySubject))
}
}

Comment on lines 228 to 242

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

🦩 🔴 *Message::ack/ack_with/double_ack/info return plain std::io::Error boxed as Error instead of a typed ErrorKind

Introduced AckErrorKind/AckError (pub type AckError = error::Error<AckErrorKind>) with variants MissingReplySubject, TimedOut, Dropped, Other, implementing Display. Changed Message::ack, Message::ack_with, and Message::double_ack to return Result<(), AckError> instead of Result<(), Error>, replacing the ad-hoc Box::new(std::io::Error::other(...))/Box::new(std::io::Error::new(...)) constructions with AckError::new(...)/AckError::with_source(...). Also updated the parallel Acker::ack, Acker::ack_with, Acker::double_ack methods (same struct, same pattern) for consistency, since they share the identical ad-hoc error construction. Note: this is a public API signature change (return type changed from Error alias to AckError), which is a breaking change for callers matching on the old boxed error type, though Result<(), AckError> still coerces via ? into Result<(), Error> at call sites since AckError: std::error::Error + Send + Sync + 'static. Reviewer should confirm this breaking change is acceptable per crate versioning policy.

🤖 Prompt for AI agents 
In async-nats/src/jetstream/message.rs around line 191, review and complete this code-review fix: Message::ack/ack_with/double_ack/info return plain std::io::Error boxed as Error instead of a typed *ErrorKind.
What the draft fix changed: Introduced `AckErrorKind`/`AckError` (`pub type AckError = error::Error<AckErrorKind>`) with variants `MissingReplySubject`, `TimedOut`, `Dropped`, `Other`, implementing `Display`. Changed `Message::ack`, `Message::ack_with`, and `Message::double_ack` to return `Result<(), AckError>` instead of `Result<(), Error>`, replacing the ad-hoc `Box::new(std::io::Error::other(...))`/`Box::new(std::io::Error::new(...))` constructions with `AckError::new(...)`/`AckError::with_source(...)`. Also updated the parallel `Acker::ack`, `Acker::ack_with`, `Acker::double_ack` methods (same struct, same pattern) for consistency, since they share the identical ad-hoc error construction. Note: this is a public API signature change (return type changed from `Error` alias to `AckError`), which is a breaking change for callers matching on the old boxed error type, though `Result<(), AckError>` still coerces via `?` into `Result<(), Error>` at call sites since `AckError: std::error::Error + Send + Sync + 'static`. Reviewer should confirm this breaking change is acceptable per crate versioning policy.
Verify the change is correct and complete; do not refactor unrelated code.

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

Expand Down Expand Up @@ -209,17 +267,15 @@ impl Message {
/// # Ok(())
/// # }
/// ```
pub async fn ack_with(&self, kind: AckKind) -> Result<(), Error> {
pub async fn ack_with(&self, kind: AckKind) -> Result<(), AckError> {
if let Some(ref reply) = self.reply {
self.context
.client
.publish(reply.to_owned(), kind.into())
.map_err(Error::from)
.map_err(|err| AckError::with_source(AckErrorKind::Other, err))
.await
} else {
Err(Box::new(std::io::Error::other(
"No reply subject, not a JetStream message",
)))
Err(AckError::new(AckErrorKind::MissingReplySubject))
}
}

Expand Down Expand Up @@ -253,47 +309,46 @@ impl Message {
/// # Ok(())
/// # }
/// ```
pub async fn double_ack(&self) -> Result<(), Error> {
pub async fn double_ack(&self) -> Result<(), AckError> {
if let Some(ref reply) = self.reply {
let inbox = self.context.client.new_inbox();
let mut subscription = self.context.client.subscribe(inbox.clone()).await?;
let mut subscription = self
.context
.client
.subscribe(inbox.clone())
.await
.map_err(|err| AckError::with_source(AckErrorKind::Other, err))?;
self.context
.client
.publish_with_reply(reply.clone(), inbox, AckKind::Ack.into())
.await?;
.await
.map_err(|err| AckError::with_source(AckErrorKind::Other, err))?;
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",
)
})? {

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

🦩 🔴 Message::info() returns generic boxed std::io::Error variants instead of a typed InfoError/InfoErrorKind

Introduced InfoErrorKind/InfoError (pub type InfoError = error::Error<InfoErrorKind>) with variants MissingReplySubject, MissingPrefix, TooFewTokens, BadTokenNumber, ParseError, implementing Display. Changed Message::info to return Result<Info<'_>, InfoError> instead of Result<Info<'_>, Error>, replacing all Box<std::io::Error> constructions (missing reply subject, missing prefix, too few tokens, bad token number) with the corresponding typed variant, and changed the try_parse! macro's parse-failure and OffsetDateTime::from_unix_timestamp_nanos error paths to wrap the underlying parse errors via InfoError::with_source(InfoErrorKind::ParseError, ...) instead of Box::new(e)/using ? with From<time::error::ComponentRange>. This is also a breaking public API signature change; verify no other file in the crate matches on Message::info()'s old boxed-error type (only visible within this file, so cross-file impact could not be checked).

🤖 Prompt for AI agents 
In async-nats/src/jetstream/message.rs around line 271, review and complete this code-review fix: Message::info() returns generic boxed std::io::Error variants instead of a typed InfoError/InfoErrorKind.
What the draft fix changed: Introduced `InfoErrorKind`/`InfoError` (`pub type InfoError = error::Error<InfoErrorKind>`) with variants `MissingReplySubject`, `MissingPrefix`, `TooFewTokens`, `BadTokenNumber`, `ParseError`, implementing `Display`. Changed `Message::info` to return `Result<Info<'_>, InfoError>` instead of `Result<Info<'_>, Error>`, replacing all `Box<std::io::Error>` constructions (missing reply subject, missing prefix, too few tokens, bad token number) with the corresponding typed variant, and changed the `try_parse!` macro's parse-failure and `OffsetDateTime::from_unix_timestamp_nanos` error paths to wrap the underlying parse errors via `InfoError::with_source(InfoErrorKind::ParseError, ...)` instead of `Box::new(e)`/using `?` with `From<time::error::ComponentRange>`. This is also a breaking public API signature change; verify no other file in the crate matches on `Message::info()`'s old boxed-error type (only visible within this file, so cross-file impact could not be checked).
Verify the change is correct and complete; do not refactor unrelated code.

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

.map_err(|_| AckError::new(AckErrorKind::TimedOut))?
{
Some(_) => Ok(()),
None => Err(Box::new(std::io::Error::other("subscription dropped"))),
None => Err(AckError::new(AckErrorKind::Dropped)),
}
} else {
Err(Box::new(std::io::Error::other(
"No reply subject, not a JetStream message",
)))
Err(AckError::new(AckErrorKind::MissingReplySubject))
}
}

/// Returns the `JetStream` message ID
/// if this is a `JetStream` message.
#[allow(clippy::mixed_read_write_in_expression)]
pub fn info(&self) -> Result<Info<'_>, Error> {
pub fn info(&self) -> Result<Info<'_>, InfoError> {
const PREFIX: &str = "$JS.ACK.";
const SKIP: usize = PREFIX.len();

let mut reply: &str = self.reply.as_ref().ok_or_else(|| {
std::io::Error::new(std::io::ErrorKind::NotFound, "did not found reply subject")
})?;
let mut reply: &str = self
.reply
.as_ref()
.ok_or_else(|| InfoError::new(InfoErrorKind::MissingReplySubject))?;

if !reply.starts_with(PREFIX) {
return Err(Box::new(std::io::Error::other(
"did not found proper prefix",
)));
return Err(InfoError::new(InfoErrorKind::MissingPrefix));
}

reply = &reply[SKIP..];
Expand All @@ -319,7 +374,7 @@ impl Message {
match str::parse(try_parse!(str)) {
Ok(parsed) => parsed,
Err(e) => {
return Err(Box::new(e));
return Err(InfoError::with_source(InfoErrorKind::ParseError, e));
}
}
};
Expand All @@ -333,7 +388,7 @@ impl Message {
}
next
} else {
return Err(Box::new(std::io::Error::other("too few tokens")));
return Err(InfoError::new(InfoErrorKind::TooFewTokens));
}
};
}
Expand Down Expand Up @@ -364,7 +419,8 @@ impl Message {
consumer_sequence: try_parse!(),
published: {
let nanos: i128 = try_parse!();
OffsetDateTime::from_unix_timestamp_nanos(nanos)?
OffsetDateTime::from_unix_timestamp_nanos(nanos)
.map_err(|err| InfoError::with_source(InfoErrorKind::ParseError, err))?
},
pending: try_parse!(),
token: if n_tokens >= 9 {
Expand All @@ -386,13 +442,14 @@ impl Message {
consumer_sequence: try_parse!(),
published: {
let nanos: i128 = try_parse!();
OffsetDateTime::from_unix_timestamp_nanos(nanos)?
OffsetDateTime::from_unix_timestamp_nanos(nanos)
.map_err(|err| InfoError::with_source(InfoErrorKind::ParseError, err))?
},
pending: try_parse!(),
token: None,
})
} else {
Err(Box::new(std::io::Error::other("bad token number")))
Err(InfoError::new(InfoErrorKind::BadTokenNumber))
}
}
}
Expand Down Expand Up @@ -445,17 +502,15 @@ impl Acker {
/// # Ok(())
/// # }
/// ```
pub async fn ack(&self) -> Result<(), Error> {
pub async fn ack(&self) -> Result<(), AckError> {
if let Some(ref reply) = self.reply {
self.context
.client
.publish(reply.to_owned(), "".into())
.map_err(Error::from)
.map_err(|err| AckError::with_source(AckErrorKind::Other, err))
.await
} else {
Err(Box::new(std::io::Error::other(
"No reply subject, not a JetStream message",
)))
Err(AckError::new(AckErrorKind::MissingReplySubject))
}
}

Expand Down Expand Up @@ -492,17 +547,15 @@ impl Acker {
/// # Ok(())
/// # }
/// ```
pub async fn ack_with(&self, kind: AckKind) -> Result<(), Error> {
pub async fn ack_with(&self, kind: AckKind) -> Result<(), AckError> {
if let Some(ref reply) = self.reply {
self.context
.client
.publish(reply.to_owned(), kind.into())
.map_err(Error::from)
.map_err(|err| AckError::with_source(AckErrorKind::Other, err))
.await
} else {
Err(Box::new(std::io::Error::other(
"No reply subject, not a JetStream message",
)))
Err(AckError::new(AckErrorKind::MissingReplySubject))
}
}

Expand Down Expand Up @@ -542,29 +595,29 @@ impl Acker {
/// # Ok(())
/// # }
/// ```
pub async fn double_ack(&self) -> Result<(), Error> {
pub async fn double_ack(&self) -> Result<(), AckError> {
if let Some(ref reply) = self.reply {
let inbox = self.context.client.new_inbox();
let mut subscription = self.context.client.subscribe(inbox.to_owned()).await?;
let mut subscription = self
.context
.client
.subscribe(inbox.to_owned())
.await
.map_err(|err| AckError::with_source(AckErrorKind::Other, err))?;
self.context
.client
.publish_with_reply(reply.to_owned(), inbox, AckKind::Ack.into())
.await?;
.await
.map_err(|err| AckError::with_source(AckErrorKind::Other, err))?;
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",
)
})? {
.map_err(|_| AckError::new(AckErrorKind::TimedOut))?
{
Some(_) => Ok(()),
None => Err(Box::new(std::io::Error::other("subscription dropped"))),
None => Err(AckError::new(AckErrorKind::Dropped)),
}
} else {
Err(Box::new(std::io::Error::other(
"No reply subject, not a JetStream message",
)))
Err(AckError::new(AckErrorKind::MissingReplySubject))
}
}
}
Expand Down
Loading