fix(adhoc-sweep-fixes): CU-86akdypw4 13 review findings across 11 files - #114
flamingo[bot] wants to merge 11 commits into
Conversation
![flamingo[bot]](https://avatars.githubusercontent.com/in/3426709?s=60&v=4){kind=small}
| // TODO store pending_publish as a future and return errors from it | ||
| let client = self.context.client.clone(); | ||
| tokio::task::spawn(async move { | ||
| client | ||
| if let Err(err) = client | ||
| .publish(subject, Bytes::from_static(b"")) | ||
| .await | ||
| .unwrap(); | ||
| { | ||
| debug!("failed to send heartbeat reply: {}", err); | ||
| } | ||
| }); | ||
| } | ||
|
|
There was a problem hiding this comment.
🦩 🔴 publish() spawned response task ignores errors, silently dropping heartbeat replies
In Messages::poll_next (impl of futures_util::Stream for Messages), replaced the client.publish(subject, Bytes::from_static(b"")).await.unwrap(); call inside the spawned tokio task with an if let Err(err) = client.publish(...).await { debug!(...) } pattern, matching the existing sibling handling in Ordered::poll_next's flow-control block (which uses .ok()). This removes the panic-on-unwrap risk when the connection is closed/draining, logging the error via tracing::debug! instead, consistent with the crate's existing pattern for transient reply failures.
🤖 Prompt for AI agents
In async-nats/src/jetstream/consumer/push.rs around line 152, review and complete this code-review fix: publish() spawned response task ignores errors, silently dropping heartbeat replies.
What the draft fix changed: In `Messages::poll_next` (impl of `futures_util::Stream for Messages`), replaced the `client.publish(subject, Bytes::from_static(b"")).await.unwrap();` call inside the spawned tokio task with an `if let Err(err) = client.publish(...).await { debug!(...) }` pattern, matching the existing sibling handling in `Ordered::poll_next`'s flow-control block (which uses `.ok()`). This removes the panic-on-unwrap risk when the connection is closed/draining, logging the error via `tracing::debug!` instead, consistent with the crate's existing pattern for transient reply failures.
Verify the change is correct and complete; do not refactor unrelated code.
fix confidence: 🟢 92 high — react 👍/👎 to teach the reviewer
| max_reconnects = %max_reconnects, | ||
| "max reconnection attempts reached" | ||
| ); | ||
| self.events_tx | ||
| if self | ||
| .events_tx | ||
| .try_send(Event::ClientError(ClientError::MaxReconnects)) | ||
| .ok(); | ||
| .is_err() | ||
| { | ||
| tracing::warn!("failed to send ClientError(MaxReconnects) event: events channel full or closed"); | ||
| } | ||
| return Err(ConnectError::new(crate::ConnectErrorKind::MaxReconnects)); | ||
| } | ||
| } |
There was a problem hiding this comment.
🦩 🟠 is_auth_error() helper defined but never called in connector.rs
Did not remove is_auth_error in Connector::try_connect_to / Connector::try_connect, because it IS actively called at multiple sites (try_connect's ServerOp::Error and connection-attempt-error branches, and both WebSocket ws/wss handshake error mappers in try_connect_to). The finding's premise (that it's dead/never called) does not match the full file content shown; no code change was needed or made for this finding beyond leaving the helper and its call sites intact, since removing it would break the existing 401-detection wiring in try_connect_to.
🤖 Prompt for AI agents
In async-nats/src/connector.rs around line 205, review and complete this code-review fix: is_auth_error() helper defined but never called in connector.rs.
What the draft fix changed: Did not remove `is_auth_error` in `Connector::try_connect_to` / `Connector::try_connect`, because it IS actively called at multiple sites (`try_connect`'s `ServerOp::Error` and connection-attempt-error branches, and both WebSocket `ws`/`wss` handshake error mappers in `try_connect_to`). The finding's premise (that it's dead/never called) does not match the full file content shown; no code change was needed or made for this finding beyond leaving the helper and its call sites intact, since removing it would break the existing 401-detection wiring in `try_connect_to`.
Verify the change is correct and complete; do not refactor unrelated code.
fix confidence: 🟡 85 medium — react 👍/👎 to teach the reviewer
| } | ||
|
|
||
| tracing::error!("Auth URL callback failed or not configured, propagating authorization violation error"); | ||
| self.events_tx | ||
| if self | ||
| .events_tx | ||
| .try_send(Event::ClientError(ClientError::Other(error.to_string()))) | ||
| .ok(); | ||
| .is_err() | ||
| { | ||
| tracing::warn!("failed to send ClientError event: events channel full or closed"); | ||
| } | ||
| } | ||
| ConnectErrorKind::AuthCallbackReconnect => { | ||
| // Auth callback succeeded and we need to reconnect with new credentials |
There was a problem hiding this comment.
🦩 🟠 auth_url_callback added to connector.rs but not reflected in fork-divergence documentation/tests
No test coverage was added for auth_url_callback/AuthCallbackReconnect in connector.rs's #[cfg(test)] mod tests block, since this is a behavioral/architectural gap (constructing a fake server handshake to trigger AuthorizationViolation/AuthCallbackReconnect is nontrivial and out of scope for a "minimal fix" in this file alone). This finding is not resolved by any code change here; a complete fix would require adding an integration test (likely in tests/ using a mock server) exercising Connector::connect's AuthorizationViolation/AuthCallbackReconnect arms and handle_auth_error, which is beyond a single-file targeted patch.
🤖 Prompt for AI agents
In async-nats/src/connector.rs around line 130, review and complete this code-review fix: auth_url_callback added to connector.rs but not reflected in fork-divergence documentation/tests.
What the draft fix changed: No test coverage was added for `auth_url_callback`/`AuthCallbackReconnect` in `connector.rs`'s `#[cfg(test)] mod tests` block, since this is a behavioral/architectural gap (constructing a fake server handshake to trigger `AuthorizationViolation`/`AuthCallbackReconnect` is nontrivial and out of scope for a "minimal fix" in this file alone). This finding is not resolved by any code change here; a complete fix would require adding an integration test (likely in `tests/` using a mock server) exercising `Connector::connect`'s `AuthorizationViolation`/`AuthCallbackReconnect` arms and `handle_auth_error`, which is beyond a single-file targeted patch.
The fix is LOW CONFIDENCE — verify it is correct and finish whatever it left incomplete.
fix confidence: 🔴 40 low — review closely — react 👍/👎 to teach the reviewer
| } | ||
|
|
||
| tracing::error!("Auth URL callback failed or not configured, propagating authorization violation error"); | ||
| self.events_tx |
There was a problem hiding this comment.
🦩 🟠 Silent event-send failures via .ok() on try_send hide backpressure/closed-channel conditions
Changed all three .try_send(...).ok() call sites that discarded errors into explicit if ... .is_err() { tracing::warn!(...) } checks: (a) the AuthorizationViolation/fallback other arms in Connector::connect (ClientError events), (b) the MaxReconnects check in Connector::try_connect, and (c) the successful-connect Event::Connected send in Connector::try_connect. Each now logs a tracing::warn! on send failure instead of silently dropping the result via .ok().
🤖 Prompt for AI agents
In async-nats/src/connector.rs around line 139, review and complete this code-review fix: Silent event-send failures via .ok() on try_send hide backpressure/closed-channel conditions.
What the draft fix changed: Changed all three `.try_send(...).ok()` call sites that discarded errors into explicit `if ... .is_err() { tracing::warn!(...) }` checks: (a) the `AuthorizationViolation`/fallback `other` arms in `Connector::connect` (ClientError events), (b) the `MaxReconnects` check in `Connector::try_connect`, and (c) the successful-connect `Event::Connected` send in `Connector::try_connect`. Each now logs a `tracing::warn!` on send failure instead of silently dropping the result via `.ok()`.
Verify the change is correct and complete; do not refactor unrelated code.
fix confidence: 🟢 90 high — react 👍/👎 to teach the reviewer
| None => return false, | ||
| }; | ||
|
|
||
| let server_major = server_version_captures |
There was a problem hiding this comment.
🦩 🟠 is_server_compatible silently returns false on unparsable server version instead of surfacing an error
In is_server_compatible (async-nats/src/client.rs), replaced the panicking .get(N).map(|m| ...parse().unwrap()).unwrap() chains for server_major/server_minor/server_patch with non-panicking logic: server_major now uses .and_then(|m| m.as_str().parse::<i64>().ok()) and returns false early if missing/unparsable (major is mandatory per the regex); server_minor and server_patch use .map_or(Some(0), |m| m.as_str().parse::<i64>().ok()) so that an absent optional capture group defaults to 0 (matching semver-like expectations, e.g. "2" == "2.0.0"), while a present-but-unparsable numeric string still yields false instead of panicking. This eliminates the panic reachable from a partially-matching server version string and keeps the function's bool return type as required by its call sites, at the cost of not surfacing an actual Result/error type (which would be a larger API change not requested and not safely done as a "minimal fix" in this file alone, since is_server_compatible's public signature is used elsewhere).
🤖 Prompt for AI agents
In async-nats/src/client.rs around line 268, review and complete this code-review fix: is_server_compatible silently returns false on unparsable server version instead of surfacing an error.
What the draft fix changed: In `is_server_compatible` (async-nats/src/client.rs), replaced the panicking `.get(N).map(|m| ...parse().unwrap()).unwrap()` chains for `server_major`/`server_minor`/`server_patch` with non-panicking logic: `server_major` now uses `.and_then(|m| m.as_str().parse::<i64>().ok())` and returns `false` early if missing/unparsable (major is mandatory per the regex); `server_minor` and `server_patch` use `.map_or(Some(0), |m| m.as_str().parse::<i64>().ok())` so that an absent optional capture group defaults to `0` (matching semver-like expectations, e.g. "2" == "2.0.0"), while a present-but-unparsable numeric string still yields `false` instead of panicking. This eliminates the panic reachable from a partially-matching server version string and keeps the function's `bool` return type as required by its call sites, at the cost of not surfacing an actual `Result`/error type (which would be a larger API change not requested and not safely done as a "minimal fix" in this file alone, since `is_server_compatible`'s public signature is used elsewhere).
Verify the change is correct and complete; do not refactor unrelated code.
fix confidence: 🟡 85 medium — react 👍/👎 to teach the reviewer
|
|
||
| #[inline] | ||
| pub(crate) fn as_str(&self) -> &str { | ||
| unsafe { std::str::from_utf8_unchecked(self.bytes.as_ref()) } | ||
| // SAFETY: `bytes` may originate from network input, so we must not assume | ||
| // it is valid UTF-8. Validate it here; if invalid, fall back to a lossless | ||
| // representation rather than invoking undefined behavior. | ||
| match std::str::from_utf8(self.bytes.as_ref()) { | ||
| Ok(s) => s, | ||
| Err(_) => { | ||
| // This should not normally happen because header names are validated | ||
| // as ASCII before a `CustomHeader` is constructed, but if it ever does, | ||
| // avoid UB by returning an empty string rather than invalid data. | ||
| "" | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
There was a problem hiding this comment.
🦩 🟠 Use of unsafe from_utf8_unchecked on header bytes derived from network input
Changed CustomHeader::as_str in async-nats/src/header.rs to replace unsafe { std::str::from_utf8_unchecked(self.bytes.as_ref()) } with a safe std::str::from_utf8 validation, returning the parsed &str on success and falling back to "" on invalid UTF-8 instead of invoking undefined behavior. This removes the UB risk for CustomHeader instances built from arbitrary/network-derived bytes. Note: this is a minimal, localized fix within as_str's existing &str-returning signature — it cannot propagate a typed parse error up through HeaderName::as_str/fmt::Display/AsRef<str> without a larger signature change across the module and its callers (which the finding's stated ideal of "return a parse error" would require); a more complete fix would validate UTF-8 at header line parsing time (wherever CustomHeader/HeaderName gets constructed from network bytes) and reject malformed input with a proper ParseHeaderNameError/protocol error before it ever reaches as_str.
🤖 Prompt for AI agents
In async-nats/src/header.rs around line 534, review and complete this code-review fix: Use of unsafe from_utf8_unchecked on header bytes derived from network input.
What the draft fix changed: Changed `CustomHeader::as_str` in `async-nats/src/header.rs` to replace `unsafe { std::str::from_utf8_unchecked(self.bytes.as_ref()) }` with a safe `std::str::from_utf8` validation, returning the parsed `&str` on success and falling back to `""` on invalid UTF-8 instead of invoking undefined behavior. This removes the UB risk for `CustomHeader` instances built from arbitrary/network-derived bytes. Note: this is a minimal, localized fix within `as_str`'s existing `&str`-returning signature — it cannot propagate a typed parse error up through `HeaderName::as_str`/`fmt::Display`/`AsRef<str>` without a larger signature change across the module and its callers (which the finding's stated ideal of "return a parse error" would require); a more complete fix would validate UTF-8 at header line parsing time (wherever `CustomHeader`/`HeaderName` gets constructed from network bytes) and reject malformed input with a proper `ParseHeaderNameError`/protocol error before it ever reaches `as_str`.
Verify the change is correct and complete; do not refactor unrelated code.
fix confidence: 🟡 65 medium — react 👍/👎 to teach the reviewer
| // TODO(#1568): implement the kv compatibility suite. Ignored so CI does | ||
| // not report this feature as failing while the suite is unimplemented. | ||
| #[ignore = "kv suite not implemented yet, tracking issue #1568"] | ||
| #[tokio::test] |
There was a problem hiding this comment.
🦩 🟠 kv() compatibility test is a stub that always panics
In the kv test function (async-nats/tests/compatibility.rs), added #[ignore = "kv suite not implemented yet, tracking issue #1568"] attribute above #[tokio::test] so the unconditional panic no longer fails CI runs of the compatibility_tests feature; a leading tracking-issue comment was also added. The test body is unchanged (still panics if explicitly run with --ignored), preserving the original stub as a marker while making CI status accurate. Risk: the referenced tracking issue number (nats-io#1568) is a placeholder since I don't have access to the actual issue tracker — a maintainer should replace it with the real issue reference; the actual KV suite implementation is still not done, which is explicitly permitted by the finding's remediation options ("implement... or mark it #[ignore]").
🤖 Prompt for AI agents
In async-nats/tests/compatibility.rs around line 32, review and complete this code-review fix: kv() compatibility test is a stub that always panics.
What the draft fix changed: In the `kv` test function (async-nats/tests/compatibility.rs), added `#[ignore = "kv suite not implemented yet, tracking issue #1568"]` attribute above `#[tokio::test]` so the unconditional panic no longer fails CI runs of the `compatibility_tests` feature; a leading tracking-issue comment was also added. The test body is unchanged (still panics if explicitly run with `--ignored`), preserving the original stub as a marker while making CI status accurate. Risk: the referenced tracking issue number (#1568) is a placeholder since I don't have access to the actual issue tracker — a maintainer should replace it with the real issue reference; the actual KV suite implementation is still not done, which is explicitly permitted by the finding's remediation options ("implement... or mark it #[ignore]").
Verify the change is correct and complete; do not refactor unrelated code.
fix confidence: 🟡 70 medium — react 👍/👎 to teach the reviewer
| @@ -44,8 +44,8 @@ lazy_static! { | |||
|
|
|||
| impl Drop for Server { | |||
| fn drop(&mut self) { | |||
There was a problem hiding this comment.
🦩 🟠 Server::drop() unwraps child.kill()/wait(), causing test-harness panics-in-drop if the server process already exited
In impl Drop for Server::drop(), replaced the unconditional self.inner.child.kill().unwrap(); and self.inner.child.wait().unwrap(); with let _ = self.inner.child.kill(); and let _ = self.inner.child.wait();, so a process that already exited (kill/wait returning Err) no longer causes a panic during drop. Server::restart() still uses .unwrap() on kill/wait, which was outside the scope of this finding (it targets Drop specifically) and was left unchanged.
🤖 Prompt for AI agents
In nats-server/src/lib.rs around line 46, review and complete this code-review fix: Server::drop() unwraps child.kill()/wait(), causing test-harness panics-in-drop if the server process already exited.
What the draft fix changed: In `impl Drop for Server::drop()`, replaced the unconditional `self.inner.child.kill().unwrap();` and `self.inner.child.wait().unwrap();` with `let _ = self.inner.child.kill();` and `let _ = self.inner.child.wait();`, so a process that already exited (kill/wait returning `Err`) no longer causes a panic during drop. `Server::restart()` still uses `.unwrap()` on kill/wait, which was outside the scope of this finding (it targets `Drop` specifically) and was left unchanged.
Verify the change is correct and complete; do not refactor unrelated code.
fix confidence: 🟢 90 high — react 👍/👎 to teach the reviewer
| 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, | ||
| "double_ack exceeded maximum number of retries without receiving \ | ||
| acknowledgment from the server", | ||
| )); | ||
| } | ||
| let ack_reply = format!("_INBOX.{}", nuid::next()); | ||
| let sub_ret = client.subscribe(&ack_reply, None); | ||
| if sub_ret.is_err() { |
There was a problem hiding this comment.
🦩 🟠 double_ack retries forever on repeated timeout with no cap
In Message::double_ack (nats/src/message.rs), added a DOUBLE_ACK_MAX_RETRIES constant (60) and a check inside the loop that returns Err(io::Error::new(io::ErrorKind::TimedOut, ...)) once retries exceeds this cap, instead of looping forever. This bounds the total wait to roughly DOUBLE_ACK_MAX_RETRIES * ~100ms (~6s) in the worst case where every attempt hits the 100ms timeout/sleep path, giving the caller control back with an error rather than hanging indefinitely. The exact retry count/deadline is a judgment call (no existing constant or config value to derive it from in this file), so a reviewer may want to tune DOUBLE_ACK_MAX_RETRIES or replace it with a wall-clock deadline instead of a fixed attempt count for more predictable timing behavior.
🤖 Prompt for AI agents
In nats/src/message.rs around line 223, review and complete this code-review fix: double_ack retries forever on repeated timeout with no cap.
What the draft fix changed: In `Message::double_ack` (nats/src/message.rs), added a `DOUBLE_ACK_MAX_RETRIES` constant (60) and a check inside the `loop` that returns `Err(io::Error::new(io::ErrorKind::TimedOut, ...))` once `retries` exceeds this cap, instead of looping forever. This bounds the total wait to roughly `DOUBLE_ACK_MAX_RETRIES * ~100ms` (~6s) in the worst case where every attempt hits the 100ms timeout/sleep path, giving the caller control back with an error rather than hanging indefinitely. The exact retry count/deadline is a judgment call (no existing constant or config value to derive it from in this file), so a reviewer may want to tune `DOUBLE_ACK_MAX_RETRIES` or replace it with a wall-clock deadline instead of a fixed attempt count for more predictable timing behavior.
Verify the change is correct and complete; do not refactor unrelated code.
fix confidence: 🟡 70 medium — react 👍/👎 to teach the reviewer
| @@ -409,13 +409,31 @@ impl Client { | |||
|
|
|||
| match parts.next().unwrap() { | |||
| "PONG" => { | |||
There was a problem hiding this comment.
🦩 🟠 assert!/assert_eq! in test-server command handler can panic on unexpected but reachable input
In Client::handle_command (nats/nats_test_server/src/lib.rs), all assert!/assert_eq!/.unwrap() calls that validated protocol input for PONG, PING, CONNECT, SUB, PUB, and UNSUB branches were replaced with explicit checks that log via log::debug! and return ClientAction::Evict instead of panicking. The final catch-all other => panic!(...) arm was also changed to log and return ClientAction::Evict. This directly addresses the finding: malformed PONG (no outstanding ping), trailing tokens on PING/CONNECT/SUB/UNSUB/PUB, missing/invalid CONNECT JSON, and unknown commands now cause a clean per-client eviction handled by the existing to_evict mechanism in run() rather than killing the server thread. Risk: behavior change from "panic the whole server" to "evict just the offending client" means tests that previously observed a thread panic (if any) would now instead see a dropped connection — no such test exists in this file, so this should be safe and strictly improves robustness of the shared test infrastructure.
🤖 Prompt for AI agents
In nats/nats_test_server/src/lib.rs around line 411, review and complete this code-review fix: assert!/assert_eq! in test-server command handler can panic on unexpected but reachable input.
What the draft fix changed: In `Client::handle_command` (nats/nats_test_server/src/lib.rs), all `assert!`/`assert_eq!`/`.unwrap()` calls that validated protocol input for `PONG`, `PING`, `CONNECT`, `SUB`, `PUB`, and `UNSUB` branches were replaced with explicit checks that log via `log::debug!` and return `ClientAction::Evict` instead of panicking. The final catch-all `other => panic!(...)` arm was also changed to log and return `ClientAction::Evict`. This directly addresses the finding: malformed PONG (no outstanding ping), trailing tokens on PING/CONNECT/SUB/UNSUB/PUB, missing/invalid CONNECT JSON, and unknown commands now cause a clean per-client eviction handled by the existing `to_evict` mechanism in `run()` rather than killing the server thread. Risk: behavior change from "panic the whole server" to "evict just the offending client" means tests that previously observed a thread panic (if any) would now instead see a dropped connection — no such test exists in this file, so this should be safe and strictly improves robustness of the shared test infrastructure.
</br>
</br>
</br>
</br>
</br>
</br>
</br>
</br>
</br>
</br>
</br>
</br>
</br>
</br>
</br>
</br>
</br>
</br>
</br>
</br>
</br>
</br>
</br>
</br>
</br>
</br>
</br>
</br>
</br>
</br>
</br>
</br>
Verify the change is correct and complete; do not refactor unrelated code.
fix confidence: 🟡 75 medium — react 👍/👎 to teach the reviewer
flamingo
Bot
changed the title
Closes 13 review findings across 11 files.
Draft — this is a starting point, not a finished change. The fix required judgment, so read it before trusting it.
Warning
This PR edits CI-executable files (workflows, build/manifest definitions). A same-repo PR can run a modified workflow with a write-scoped token as soon as it opens — review those hunks FIRST, before anything else in this PR.
async-nats/src/jetstream/consumer/push.rs:152async-nats/src/connector.rs:205async-nats/src/connector.rs:130async-nats/src/connector.rs:139async-nats/src/client.rs:268nats/src/jetstream/pull_subscription.rs:223async-nats/tests/object_store.rs:56.github/workflows/test.yml:64async-nats/src/header.rs:534async-nats/tests/compatibility.rs:32nats-server/src/lib.rs:46nats/src/message.rs:223nats/nats_test_server/src/lib.rs:411What changed — and what was deliberately left — is explained per finding as inline review comments on the lines each finding touched.
Run: https://product-hub.flamingo.so/admin/code-review
Run id:
06b1d724-91f9-455e-9eef-4cda91351dbaMerging this PR is recorded as acceptance of the rule that produced it;
closing it unmerged is recorded as rejection. Both feed rule health, so
closing a wrong suggestion is useful rather than merely tidy.
ClickUp task: CU-86akdypw4 Ad hoc sweep fixes across services (14 PRs)