Skip to content

fix(NATSRS-002-2): CU-86akbh48z 2 review findings in kv.rs #113

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
+72 −45
Draft

fix(NATSRS-002-2): CU-86akbh48z 2 review findings in kv.rs #113

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
117 changes: 72 additions & 45 deletions nats/src/kv.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@

//! Support for Key Value Store.

use std::fmt;
use std::io;
use std::time::Duration;

Expand All @@ -25,6 +26,43 @@ use crate::message::Message;
use lazy_static::lazy_static;
use regex::Regex;

/// The kind of error that can occur when interacting with a key-value store.
#[derive(Debug, Eq, PartialEq, Clone, Copy)]
pub enum KvErrorKind {
/// The connected server does not support key-value stores.
UnsupportedServerVersion,
/// The provided bucket name is invalid.
InvalidBucketName,
/// The provided key is invalid.
InvalidKey,
/// The requested bucket is not a valid key-value store.
InvalidBucket,
/// The requested history value exceeds the maximum allowed.
HistoryTooLarge,
/// Any other error.
Other,
}

impl fmt::Display for KvErrorKind {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
KvErrorKind::UnsupportedServerVersion => {
write!(f, "key-value requires at least server version 2.6.2")
}
KvErrorKind::InvalidBucketName => write!(f, "invalid bucket name"),
KvErrorKind::InvalidKey => write!(f, "invalid key"),
KvErrorKind::InvalidBucket => write!(f, "bucket not valid key-value store"),
KvErrorKind::HistoryTooLarge => write!(f, "history limited to a max of 64"),
KvErrorKind::Other => write!(f, "key-value error"),
}
}
}

/// Converts a [`KvErrorKind`] and message into an [`io::Error`].
fn kv_error(kind: KvErrorKind, message: impl Into<String>) -> io::Error {
io::Error::new(io::ErrorKind::Other, format!("{kind}: {}", message.into()))
}

/// Configuration values for key value stores.
#[derive(Debug, Default)]
pub struct Config {
Expand Down Expand Up @@ -127,17 +165,14 @@ impl JetStream {
/// ```
pub fn key_value(&self, bucket: &str) -> io::Result<Store> {

@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.

🦩 🔴 kv.rs Store uses io::Error/io::Result instead of the required Error pattern

Added a KvErrorKind enum (with Display impl) and a kv_error(kind, message) helper near the top of nats/src/kv.rs, and replaced all raw io::Error::new(io::ErrorKind::Other/InvalidInput, "...") construction sites in JetStream::key_value, JetStream::create_key_value, JetStream::delete_key_value, Store::entry, Store::put, Store::update, Store::delete, Store::purge with calls to kv_error(KvErrorKind::..., "..."). This is a partial fix: the function signatures still return io::Result<T> rather than the crate's generic Error<Kind> type used by async-nats, since that would require a much larger architectural change (a dedicated error type threaded through the whole sync nats crate, which this legacy crate does not use anywhere). The change only makes the kind of error explicit/typed at construction time via KvErrorKind, it does not change the public return types, so it does not fully satisfy NATSRS-001/NATSRS-001-2 as literally stated. A complete fix would require introducing an Error<KvErrorKind> type and updating every public signature in this file (and likely callers elsewhere), which is out of scope/risk for a minimal fix in one file.

🤖 Prompt for AI agents 
In nats/src/kv.rs around line 128, review and complete this code-review fix: kv.rs Store uses io::Error/io::Result instead of the required Error<Kind> pattern.
What the draft fix changed: Added a `KvErrorKind` enum (with `Display` impl) and a `kv_error(kind, message)` helper near the top of `nats/src/kv.rs`, and replaced all raw `io::Error::new(io::ErrorKind::Other/InvalidInput, "...")` construction sites in `JetStream::key_value`, `JetStream::create_key_value`, `JetStream::delete_key_value`, `Store::entry`, `Store::put`, `Store::update`, `Store::delete`, `Store::purge` with calls to `kv_error(KvErrorKind::..., "...")`. This is a partial fix: the function signatures still return `io::Result<T>` rather than the crate's generic `Error<Kind>` type used by async-nats, since that would require a much larger architectural change (a dedicated error type threaded through the whole sync `nats` crate, which this legacy crate does not use anywhere). The change only makes the *kind* of error explicit/typed at construction time via `KvErrorKind`, it does not change the public return types, so it does not fully satisfy NATSRS-001/NATSRS-001-2 as literally stated. A complete fix would require introducing an `Error<KvErrorKind>` type and updating every public signature in this file (and likely callers elsewhere), which is out of scope/risk for a minimal fix in one file.
The fix is LOW CONFIDENCE — verify it is correct and finish whatever it left incomplete.

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

if !self.connection.is_server_compatible_version(2, 6, 2) {
return Err(io::Error::new(
io::ErrorKind::Other,
return Err(kv_error(
KvErrorKind::UnsupportedServerVersion,
"key-value requires at least server version 2.6.2",
));
}

if !is_valid_bucket_name(bucket) {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
"invalid bucket name",
));
return Err(kv_error(KvErrorKind::InvalidBucketName, "invalid bucket name"));
}

let stream_name = format!("KV_{bucket}");
Expand All @@ -146,8 +181,8 @@ impl JetStream {
// Do some quick sanity checks that this is a correctly formed stream for KV.
// Max msgs per subject should be > 0.
if stream_info.config.max_msgs_per_subject < 1 {
return Err(io::Error::new(
io::ErrorKind::Other,
return Err(kv_error(
KvErrorKind::InvalidBucket,
"bucket not valid key-value store",
));
}
Expand Down Expand Up @@ -185,8 +220,8 @@ impl JetStream {
/// ```
pub fn create_key_value(&self, config: &Config) -> io::Result<Store> {
if !self.connection.is_server_compatible_version(2, 6, 2) {
return Err(io::Error::new(
io::ErrorKind::Other,
return Err(kv_error(
KvErrorKind::UnsupportedServerVersion,
"key-value requires at least server version 2.6.2",
));
}
Expand All @@ -200,16 +235,16 @@ impl JetStream {
};

if !is_valid_bucket_name(&config.bucket) {
return Err(io::Error::new(io::ErrorKind::Other, "invalid bucket name"));
return Err(kv_error(KvErrorKind::InvalidBucketName, "invalid bucket name"));
}

self.account_info()?;

// Default to 1 for history. Max is 64 for now.
let history = if config.history > 0 {
if config.history > MAX_HISTORY {
return Err(io::Error::new(
io::ErrorKind::Other,
return Err(kv_error(
KvErrorKind::HistoryTooLarge,
"history limited to a max of 64",
));
}
Expand Down Expand Up @@ -275,14 +310,14 @@ impl JetStream {
/// ```
pub fn delete_key_value(&self, bucket: &str) -> io::Result<()> {
if !self.connection.is_server_compatible_version(2, 6, 2) {
return Err(io::Error::new(
io::ErrorKind::Other,
return Err(kv_error(
KvErrorKind::UnsupportedServerVersion,
"key-value requires at least server version 2.6.2",
));
}

if !is_valid_bucket_name(bucket) {
return Err(io::Error::new(io::ErrorKind::Other, "invalid bucket name"));
return Err(kv_error(KvErrorKind::InvalidBucketName, "invalid bucket name"));
}

let stream_name = format!("KV_{bucket}");
Expand Down Expand Up @@ -332,6 +367,17 @@ impl Store {
})
}

/// Builds the full subject for a given key, honoring the configured domain prefix.
fn key_subject(&self, key: &str) -> String {
let mut subject = String::new();
if let Some(api_prefix) = self.domain_prefix.as_ref() {
subject.push_str(api_prefix);
}
subject.push_str(&self.prefix);
subject.push_str(key);
subject
}

/// Returns the latest entry for the key, if any.
///
/// # Examples
Expand All @@ -358,12 +404,10 @@ impl Store {
/// ```
pub fn entry(&self, key: &str) -> io::Result<Option<Entry>> {
if !is_valid_key(key) {
return Err(io::Error::new(io::ErrorKind::InvalidInput, "invalid key"));
return Err(kv_error(KvErrorKind::InvalidKey, "invalid key"));
}

let mut subject = String::new();
subject.push_str(&self.prefix);
subject.push_str(key);
let subject = self.key_subject(key);

match self.context.get_last_message(&self.stream_name, &subject) {
Ok(message) => {
Expand Down Expand Up @@ -453,15 +497,10 @@ impl Store {
/// ```
pub fn put(&self, key: &str, value: impl AsRef<[u8]>) -> io::Result<u64> {
if !is_valid_key(key) {
return Err(io::Error::new(io::ErrorKind::InvalidInput, "invalid key"));
return Err(kv_error(KvErrorKind::InvalidKey, "invalid key"));
}

let mut subject = String::new();

@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.

🦩 🟠 put() rebuilds subject prefix inconsistently with domain_prefix vs info()/entry() which never apply it

Added a private helper Store::key_subject(&self, key: &str) -> String that builds the subject consistently by applying self.domain_prefix (if set) followed by self.prefix and the key. Updated Store::entry (used by get), Store::put, Store::update, Store::delete, and Store::purge to all use this helper instead of ad-hoc subject-building, so every read/write path now consistently includes domain_prefix when present. Previously entry, keys, history, purge omitted the domain prefix while put/update/delete included it — now all key-addressed operations (entry, put, update, delete, purge) use the same logic. keys/history/watch (which subscribe using self.prefix with wildcards for iteration, not point lookups via get_last_message) were left unchanged since domain-prefixing subscription subjects is a separate, riskier behavioral change not explicitly required by the finding and could affect wildcard subscription semantics differently than direct publish/lookup by exact subject.

🤖 Prompt for AI agents 
In nats/src/kv.rs around line 459, review and complete this code-review fix: put() rebuilds subject prefix inconsistently with domain_prefix vs info()/entry() which never apply it.
What the draft fix changed: Added a private helper `Store::key_subject(&self, key: &str) -> String` that builds the subject consistently by applying `self.domain_prefix` (if set) followed by `self.prefix` and the key. Updated `Store::entry` (used by `get`), `Store::put`, `Store::update`, `Store::delete`, and `Store::purge` to all use this helper instead of ad-hoc subject-building, so every read/write path now consistently includes `domain_prefix` when present. Previously `entry`, `keys`, `history`, `purge` omitted the domain prefix while `put`/`update`/`delete` included it — now all key-addressed operations (`entry`, `put`, `update`, `delete`, `purge`) use the same logic. `keys`/`history`/`watch` (which subscribe using `self.prefix` with wildcards for iteration, not point lookups via `get_last_message`) were left unchanged since domain-prefixing subscription subjects is a separate, riskier behavioral change not explicitly required by the finding and could affect wildcard subscription semantics differently than direct publish/lookup by exact subject.
Verify the change is correct and complete; do not refactor unrelated code.

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

if let Some(api_prefix) = self.domain_prefix.as_ref() {
subject.push_str(api_prefix);
}
subject.push_str(&self.prefix);
subject.push_str(key);
let subject = self.key_subject(key);

let publish_ack = self.context.publish(&subject, value)?;

Expand Down Expand Up @@ -530,15 +569,10 @@ impl Store {
/// ```
pub fn update(&self, key: &str, value: impl AsRef<[u8]>, revision: u64) -> io::Result<u64> {
if !is_valid_key(key) {
return Err(io::Error::new(io::ErrorKind::InvalidInput, "invalid key"));
return Err(kv_error(KvErrorKind::InvalidKey, "invalid key"));
}

let mut subject = String::new();
if let Some(api_prefix) = self.domain_prefix.as_ref() {
subject.push_str(api_prefix);
}
subject.push_str(&self.prefix);
subject.push_str(key);
let subject = self.key_subject(key);

let mut headers = HeaderMap::default();
headers.insert(
Expand Down Expand Up @@ -577,15 +611,10 @@ impl Store {
/// ```
pub fn delete(&self, key: &str) -> io::Result<()> {
if !is_valid_key(key) {
return Err(io::Error::new(io::ErrorKind::InvalidInput, "invalid key"));
return Err(kv_error(KvErrorKind::InvalidKey, "invalid key"));
}

let mut subject = String::new();
if let Some(api_prefix) = self.domain_prefix.as_ref() {
subject.push_str(api_prefix);
}
subject.push_str(&self.prefix);
subject.push_str(key);
let subject = self.key_subject(key);

let mut headers = HeaderMap::default();
headers.insert(KV_OPERATION, KV_OPERATION_DELETE.to_string());
Expand Down Expand Up @@ -620,12 +649,10 @@ impl Store {
/// ```
pub fn purge(&self, key: &str) -> io::Result<()> {
if !is_valid_key(key) {
return Err(io::Error::new(io::ErrorKind::InvalidInput, "invalid key"));
return Err(kv_error(KvErrorKind::InvalidKey, "invalid key"));
}

let mut subject = String::new();
subject.push_str(&self.prefix);
subject.push_str(key);
let subject = self.key_subject(key);

let mut headers = HeaderMap::default();
headers.insert(KV_OPERATION, KV_OPERATION_PURGE.to_string());
Expand Down
Loading