Add subject validataion

Signed-off-by: Tomasz Pietrek <tomasz@nats.io>
Signed-off-by: Tomasz Pietrek <tomasz@synadia.com>

Author: Jarema

.config/nats.dic

@@ -202,3 +202,4 @@ NUID
 performant
 thread_rng
 NUID's
+whitespace

async-nats/benches/main.rs

@@ -3,5 +3,10 @@ use criterion::criterion_main;
 // Import the benchmark groups from both files
 mod core_nats;
 mod jetstream;
+mod subject_validation;
 
-criterion_main!(core_nats::core_nats, jetstream::jetstream);
+criterion_main!(
+    core_nats::core_nats,
+    jetstream::jetstream,
+    subject_validation::subject_validation
+);

async-nats/benches/subject_validation.rs

@@ -0,0 +1,103 @@
+use bytes::Bytes;
+use criterion::{criterion_group, criterion_main, Criterion};
+
+static PAYLOAD: &[u8] = &[22; 32];
+
+// Pre-defined subjects as static strs to avoid cloning overhead
+static SUBJECT_5: &str = "bench";
+static SUBJECT_16: &str = "bench.data.testx";
+static SUBJECT_32: &str = "events.data.bench.test.messages";
+static SUBJECT_128: &str = "segment.segment.segment.segment.segment.segment.segment.segment.segment.segment.segment.segment.segment.segment.segment.segment.x";
+
+async fn publish_with_static_str(
+    nc: async_nats::Client,
+    subject: &'static str,
+    payload: Bytes,
+    count: u64,
+) {
+    for _ in 0..count {
+        nc.publish(subject, payload.clone()).await.unwrap();
+    }
+}
+
+pub fn publish_validation_comparison(c: &mut Criterion) {
+    let messages_per_iter = 500_000;
+    let server = nats_server::run_basic_server();
+
+    let mut validation_group = c.benchmark_group("nats::publish_validation_comparison");
+    validation_group.sample_size(10);
+    validation_group.warm_up_time(std::time::Duration::from_secs(1));
+
+    // Test different subject lengths: 5, 16, 32, 128 characters
+    for (subject_len, subject_str) in [
+        (5, SUBJECT_5),
+        (16, SUBJECT_16),
+        (32, SUBJECT_32),
+        (128, SUBJECT_128),
+    ] {
+        validation_group.throughput(criterion::Throughput::Elements(messages_per_iter));
+
+        // Benchmark 1: With runtime validation (default)
+        validation_group.bench_with_input(
+            criterion::BenchmarkId::new("with_validation", subject_len),
+            &subject_len,
+            |b, _| {
+                let rt = tokio::runtime::Runtime::new().unwrap();
+                let nc = rt.block_on(async {
+                    let nc = async_nats::connect(server.client_url()).await.unwrap();
+                    nc.publish("data", "data".into()).await.unwrap();
+                    nc
+                });
+
+                b.to_async(rt).iter_with_large_drop(move || {
+                    let nc = nc.clone();
+                    async move {
+                        publish_with_static_str(
+                            nc,
+                            subject_str,
+                            Bytes::from_static(PAYLOAD),
+                            messages_per_iter,
+                        )
+                        .await
+                    }
+                });
+            },
+        );
+
+        // Benchmark 2: With validation disabled
+        validation_group.bench_with_input(
+            criterion::BenchmarkId::new("skip_validation", subject_len),
+            &subject_len,
+            |b, _| {
+                let rt = tokio::runtime::Runtime::new().unwrap();
+                let nc = rt.block_on(async {
+                    let nc = async_nats::ConnectOptions::new()
+                        .skip_subject_validation(true)
+                        .connect(server.client_url())
+                        .await
+                        .unwrap();
+                    nc.publish("data", "data".into()).await.unwrap();
+                    nc
+                });
+
+                b.to_async(rt).iter_with_large_drop(move || {
+                    let nc = nc.clone();
+                    async move {
+                        publish_with_static_str(
+                            nc,
+                            subject_str,
+                            Bytes::from_static(PAYLOAD),
+                            messages_per_iter,
+                        )
+                        .await
+                    }
+                });
+            },
+        );
+    }
+
+    validation_group.finish();
+}
+
+criterion_group!(subject_validation, publish_validation_comparison);
+criterion_main!(subject_validation);

async-nats/src/client.rs

@@ -35,7 +35,6 @@ use std::time::Duration;
 use thiserror::Error;
 use tokio::sync::{mpsc, oneshot};
 use tokio_util::sync::PollSender;
-use tracing::trace;
 
 static VERSION_RE: Lazy<Regex> =
     Lazy::new(|| Regex::new(r"\Av?([0-9]+)\.?([0-9]+)?\.?([0-9]+)?").unwrap());
@@ -88,6 +87,7 @@ pub struct Client {
     request_timeout: Option<Duration>,
     max_payload: Arc<AtomicUsize>,
     connection_stats: Arc<Statistics>,
+    skip_subject_validation: bool,
 }
 
 pub mod traits {
@@ -100,30 +100,37 @@ pub mod traits {
     use super::{PublishError, Request, RequestError, SubscribeError};
 
     pub trait Publisher {
-        fn publish_with_reply<S: ToSubject, R: ToSubject>(
+        fn publish_with_reply<S, R>(
             &self,
             subject: S,
             reply: R,
             payload: Bytes,
-        ) -> impl Future<Output = Result<(), PublishError>>;
+        ) -> impl Future<Output = Result<(), PublishError>>
+        where
+            S: ToSubject,
+            R: ToSubject;
 
         fn publish_message(
             &self,
             msg: message::OutboundMessage,
         ) -> impl Future<Output = Result<(), PublishError>>;
     }
     pub trait Subscriber {
-        fn subscribe<S: ToSubject>(
+        fn subscribe<S>(
             &self,
             subject: S,
-        ) -> impl Future<Output = Result<crate::Subscriber, SubscribeError>>;
+        ) -> impl Future<Output = Result<crate::Subscriber, SubscribeError>>
+        where
+            S: ToSubject;
     }
     pub trait Requester {
-        fn send_request<S: ToSubject>(
+        fn send_request<S>(
             &self,
             subject: S,
             request: Request,
-        ) -> impl Future<Output = Result<Message, RequestError>>;
+        ) -> impl Future<Output = Result<Message, RequestError>>
+        where
+            S: ToSubject;
     }
     pub trait TimeoutProvider {
         fn timeout(&self) -> Option<Duration>;
@@ -147,12 +154,16 @@ impl traits::TimeoutProvider for Client {
 }
 
 impl traits::Publisher for Client {
-    fn publish_with_reply<S: ToSubject, R: ToSubject>(
+    fn publish_with_reply<S, R>(
         &self,
         subject: S,
         reply: R,
         payload: Bytes,
-    ) -> impl Future<Output = Result<(), PublishError>> {
+    ) -> impl Future<Output = Result<(), PublishError>>
+    where
+        S: ToSubject,
+        R: ToSubject,
+    {
         self.publish_with_reply(subject, reply, payload)
     }
 
@@ -165,10 +176,10 @@ impl traits::Publisher for Client {
 }
 
 impl traits::Subscriber for Client {
-    fn subscribe<S: ToSubject>(
-        &self,
-        subject: S,
-    ) -> impl Future<Output = Result<Subscriber, SubscribeError>> {
+    fn subscribe<S>(&self, subject: S) -> impl Future<Output = Result<Subscriber, SubscribeError>>
+    where
+        S: ToSubject,
+    {
         self.subscribe(subject)
     }
 }
@@ -206,6 +217,7 @@ impl Client {
         request_timeout: Option<Duration>,
         max_payload: Arc<AtomicUsize>,
         statistics: Arc<Statistics>,
+        skip_subject_validation: bool,
     ) -> Client {
         let poll_sender = PollSender::new(sender.clone());
         Client {
@@ -219,7 +231,25 @@ impl Client {
             request_timeout,
             max_payload,
             connection_stats: statistics,
+            skip_subject_validation,
+        }
+    }
+
+    /// Converts a subject to a [`Subject`], optionally validating it.
+    ///
+    /// If subject validation is enabled (the default), the subject is validated
+    /// before conversion. If validation is disabled via
+    /// [`ConnectOptions::skip_subject_validation`], the subject is converted
+    /// without validation.
+    fn maybe_validate_subject<S: ToSubject>(
+        &self,
+        subject: S,
+    ) -> Result<crate::Subject, crate::subject::SubjectError> {
+        let subject = subject.to_subject();
+        if !self.skip_subject_validation && !subject.is_valid() {
+            return Err(crate::subject::SubjectError::InvalidFormat);
         }
+        Ok(subject)
     }
 
     /// Returns the default timeout for requests set when creating the client.
@@ -318,7 +348,10 @@ impl Client {
         subject: S,
         payload: Bytes,
     ) -> Result<(), PublishError> {
-        let subject = subject.to_subject();
+        let subject = self
+            .maybe_validate_subject(subject)
+            .map_err(|e| PublishError::with_source(PublishErrorKind::BadSubject, e))?;
+
         let max_payload = self.max_payload.load(Ordering::Relaxed);
         if payload.len() > max_payload {
             return Err(PublishError::with_source(
@@ -367,7 +400,9 @@ impl Client {
         headers: HeaderMap,
         payload: Bytes,
     ) -> Result<(), PublishError> {
-        let subject = subject.to_subject();
+        let subject = self
+            .maybe_validate_subject(subject)
+            .map_err(|e| PublishError::with_source(PublishErrorKind::BadSubject, e))?;
 
         self.sender
             .send(Command::Publish(OutboundMessage {
@@ -402,8 +437,12 @@ impl Client {
         reply: R,
         payload: Bytes,
     ) -> Result<(), PublishError> {
-        let subject = subject.to_subject();
-        let reply = reply.to_subject();
+        let subject = self
+            .maybe_validate_subject(subject)
+            .map_err(|e| PublishError::with_source(PublishErrorKind::BadSubject, e))?;
+        let reply = self
+            .maybe_validate_subject(reply)
+            .map_err(|e| PublishError::with_source(PublishErrorKind::BadSubject, e))?;
 
         self.sender
             .send(Command::Publish(OutboundMessage {
@@ -441,8 +480,12 @@ impl Client {
         headers: HeaderMap,
         payload: Bytes,
     ) -> Result<(), PublishError> {
-        let subject = subject.to_subject();
-        let reply = reply.to_subject();
+        let subject = self
+            .maybe_validate_subject(subject)
+            .map_err(|e| PublishError::with_source(PublishErrorKind::BadSubject, e))?;
+        let reply = self
+            .maybe_validate_subject(reply)
+            .map_err(|e| PublishError::with_source(PublishErrorKind::BadSubject, e))?;
 
         self.sender
             .send(Command::Publish(OutboundMessage {
@@ -471,13 +514,6 @@ impl Client {
         subject: S,
         payload: Bytes,
     ) -> Result<Message, RequestError> {
-        let subject = subject.to_subject();
-
-        trace!(
-            "request sent to subject: {} ({})",
-            subject.as_ref(),
-            payload.len()
-        );
         let request = Request::new().payload(payload);
         self.send_request(subject, request).await
     }
@@ -503,8 +539,6 @@ impl Client {
         headers: HeaderMap,
         payload: Bytes,
     ) -> Result<Message, RequestError> {
-        let subject = subject.to_subject();
-
         let request = Request::new().headers(headers).payload(payload);
         self.send_request(subject, request).await
     }
@@ -527,7 +561,9 @@ impl Client {
         subject: S,
         request: Request,
     ) -> Result<Message, RequestError> {
-        let subject = subject.to_subject();
+        let subject = self
+            .maybe_validate_subject(subject)
+            .map_err(|e| RequestError::with_source(RequestErrorKind::Other, e))?;
 
         if let Some(inbox) = request.inbox {
             let timeout = request.timeout.unwrap_or(self.request_timeout);
@@ -640,7 +676,13 @@ impl Client {
     /// # }
     /// ```
     pub async fn subscribe<S: ToSubject>(&self, subject: S) -> Result<Subscriber, SubscribeError> {
-        let subject = subject.to_subject();
+        let subject = self.maybe_validate_subject(subject).map_err(|e| {
+            SubscribeError(Box::new(std::io::Error::new(
+                std::io::ErrorKind::InvalidInput,
+                e,
+            )))
+        })?;
+
         let sid = self.next_subscription_id.fetch_add(1, Ordering::Relaxed);
         let (sender, receiver) = mpsc::channel(self.subscription_capacity);
 
@@ -677,7 +719,12 @@ impl Client {
         subject: S,
         queue_group: String,
     ) -> Result<Subscriber, SubscribeError> {
-        let subject = subject.to_subject();
+        let subject = self.maybe_validate_subject(subject).map_err(|e| {
+            SubscribeError(Box::new(std::io::Error::new(
+                std::io::ErrorKind::InvalidInput,
+                e,
+            )))
+        })?;
 
         let sid = self.next_subscription_id.fetch_add(1, Ordering::Relaxed);
         let (sender, receiver) = mpsc::channel(self.subscription_capacity);

async-nats/src/jetstream/consumer/mod.rs

@@ -58,6 +58,9 @@ impl<T: IntoConsumerConfig> Consumer<T> {
     /// Retrieves `info` about [Consumer] from the server, updates the cached `info` inside
     /// [Consumer] and returns it.
     ///
+    /// When possible, use [`Message::info()`][crate::jetstream::Message::info] instead —
+    /// message metadata often already contains the needed information and does not require a server call.
+    ///
     /// # Examples
     ///
     /// ```no_run
@@ -85,6 +88,9 @@ impl<T: IntoConsumerConfig> Consumer<T> {
 
     /// Retrieves `info` about [Consumer] from the server. Does not update the cache.
     ///
+    /// When possible, use [`Message::info()`][crate::jetstream::Message::info] instead —
+    /// message metadata often already contains the needed information and does not require a server call.
+    ///
     /// # Examples
     ///
     /// ```no_run