refactor!: restrict public items for misuse safety (#227)

Author: quettabit

Cargo.lock

@@ -37,8 +37,8 @@ reqwest = { version = "0.13.1", default-features = false, features = [
     "zstd",
     "rustls",
 ] }
-s2-api = { git = "https://github.com/s2-streamstore/s2", rev = "b69e012f07290c9fb6ada77552362a93634287c7", package = "s2-api" }
-s2-common = { git = "https://github.com/s2-streamstore/s2", rev = "b69e012f07290c9fb6ada77552362a93634287c7", package = "s2-common" }
+s2-api = { git = "https://github.com/s2-streamstore/s2", rev = "6f50d0a1a133b8b4ee0fef900e586a0f10d75637", package = "s2-api" }
+s2-common = { git = "https://github.com/s2-streamstore/s2", rev = "6f50d0a1a133b8b4ee0fef900e586a0f10d75637", package = "s2-common" }
 secrecy = "0.10.3"
 serde = { version = "1.0.228" }
 serde_json = "1.0.148"
@@ -53,6 +53,9 @@ url = "2.5.7"
 urlencoding = "2.1.3"
 uuid = { version = "1.19.0", features = ["v4", "fast-rng"] }
 
+[features]
+_hidden = []
+
 [dev-dependencies]
 assert_matches = "1.5.0"
 rstest = "0.26.1"

Cargo.toml

@@ -1,7 +1,7 @@
 use s2::{
     S2,
     types::{
-        AccessTokenScope, BasinMatcher, BasinName, IssueAccessTokenInput, Operation,
+        AccessTokenScopeInput, BasinMatcher, BasinName, IssueAccessTokenInput, Operation,
         OperationGroupPermissions, ReadWritePermissions, S2Config, StreamMatcher,
     },
 };
@@ -19,7 +19,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     let input = IssueAccessTokenInput::new(
         "ro-token".parse()?,
-        AccessTokenScope::from_op_group_perms(
+        AccessTokenScopeInput::from_op_group_perms(
             OperationGroupPermissions::new().with_account(ReadWritePermissions::read_only()),
         )
         .with_ops([Operation::CreateStream])

examples/issue_access_token.rs

@@ -13,21 +13,11 @@ use tokio::time::Instant;
 use crate::types::{AppendInput, AppendRecord, AppendRecordBatch, FencingToken, MeteredBytes};
 
 #[derive(Debug, Clone)]
-#[non_exhaustive]
-/// Configuration for batching [AppendRecord]s.
+/// Configuration for batching [`AppendRecord`]s.
 pub struct BatchingConfig {
-    /// How long to wait for more records before flushing a batch.
-    ///
-    /// Defaults to `5ms`.
-    pub linger: Duration,
-    /// Maximum bytes per batch.
-    ///
-    /// Defaults to `1MiB`.
-    pub max_batch_bytes: usize,
-    /// Maximum number of records per batch.
-    ///
-    /// Defaults to `1000`.
-    pub max_batch_records: usize,
+    linger: Duration,
+    max_batch_bytes: usize,
+    max_batch_records: usize,
 }
 
 impl Default for BatchingConfig {
@@ -41,21 +31,27 @@ impl Default for BatchingConfig {
 }
 
 impl BatchingConfig {
-    /// Create a new [BatchingConfig] with default settings.
+    /// Create a new [`BatchingConfig`] with default settings.
     pub fn new() -> Self {
         Self::default()
     }
 
-    /// Set the linger duration.
+    /// Set the duration for how long to wait for more records before flushing a batch.
+    ///
+    /// Defaults to `5ms`.
     pub fn with_linger(self, linger: Duration) -> Self {
         Self { linger, ..self }
     }
 
-    /// Set the maximum bytes per batch.
+    /// Set the maximum metered bytes per batch.
+    ///
+    /// **Note:** It must not exceed `1MiB`.
+    ///
+    /// Defaults to `1MiB`.
     pub fn with_max_batch_bytes(self, max_batch_bytes: usize) -> Result<Self, ValidationError> {
         if max_batch_bytes > RECORD_BATCH_MAX.bytes {
             return Err(ValidationError(format!(
-                "max_batch_bytes: {max_batch_bytes} exceeds {}",
+                "max_batch_bytes ({max_batch_bytes}) exceeds {}",
                 RECORD_BATCH_MAX.bytes
             )));
         }
@@ -66,10 +62,14 @@ impl BatchingConfig {
     }
 
     /// Set the maximum number of records per batch.
+    ///
+    /// **Note:** It must not exceed `1000`.
+    ///
+    /// Defaults to `1000`.
     pub fn with_max_batch_records(self, max_batch_records: usize) -> Result<Self, ValidationError> {
         if max_batch_records > RECORD_BATCH_MAX.count {
             return Err(ValidationError(format!(
-                "max_batch_records: {max_batch_records} exceeds {}",
+                "max_batch_records ({max_batch_records}) exceeds {}",
                 RECORD_BATCH_MAX.count
             )));
         }
@@ -80,16 +80,15 @@ impl BatchingConfig {
     }
 }
 
-#[non_exhaustive]
-/// A [Stream] that batches [AppendRecord]s into [AppendInput]s.
+/// A [`Stream`] that batches [`AppendRecord`]s into [`AppendInput`]s.
 pub struct AppendInputs {
     pub(crate) batches: AppendRecordBatches,
     pub(crate) fencing_token: Option<FencingToken>,
     pub(crate) match_seq_num: Option<u64>,
 }
 
 impl AppendInputs {
-    /// Create a new [AppendInputs] with the given records and config.
+    /// Create a new [`AppendInputs`] with the given records and config.
     pub fn new(
         records: impl Stream<Item = impl Into<AppendRecord> + Send> + Send + Unpin + 'static,
         config: BatchingConfig,
@@ -101,15 +100,15 @@ impl AppendInputs {
         }
     }
 
-    /// Set the fencing token for all [AppendInput]s.
+    /// Set the fencing token for all [`AppendInput`]s.
     pub fn with_fencing_token(self, fencing_token: FencingToken) -> Self {
         Self {
             fencing_token: Some(fencing_token),
             ..self
         }
     }
 
-    /// Set the match sequence number for the initial [AppendInput]. It will be auto-incremented
+    /// Set the match sequence number for the initial [`AppendInput`]. It will be auto-incremented
     /// for the subsequent ones.
     pub fn with_match_seq_num(self, seq_num: u64) -> Self {
         Self {
@@ -142,13 +141,13 @@ impl Stream for AppendInputs {
     }
 }
 
-/// A [Stream] that batches [AppendRecord]s into [AppendRecordBatch]es.
+/// A [`Stream`] that batches [`AppendRecord`]s into [`AppendRecordBatch`]es.
 pub struct AppendRecordBatches {
     inner: Pin<Box<dyn Stream<Item = Result<AppendRecordBatch, ValidationError>> + Send>>,
 }
 
 impl AppendRecordBatches {
-    /// Create a new [AppendRecordBatches] with the given records and config.
+    /// Create a new [`AppendRecordBatches`] with the given records and config.
     pub fn new(
         records: impl Stream<Item = impl Into<AppendRecord> + Send> + Send + Unpin + 'static,
         config: BatchingConfig,
@@ -203,7 +202,7 @@ fn append_record_batches(
             let record_bytes = first_record.metered_bytes();
             if record_bytes > config.max_batch_bytes {
                 yield Err(ValidationError(format!(
-                    "record size ({record_bytes} bytes) exceeds max_batch_bytes ({})",
+                    "record size in metered bytes ({record_bytes}) exceeds max_batch_bytes ({})",
                     config.max_batch_bytes
                 )));
                 break;
@@ -310,7 +309,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn batching_should_fail_when_it_sees_oversized_record() -> Result<(), ValidationError> {
+    async fn batching_should_error_when_it_sees_oversized_record() -> Result<(), ValidationError> {
         let record = AppendRecord::new("hello")?;
         let record_bytes = record.metered_bytes();
         let max_batch_bytes = 1;
@@ -324,7 +323,7 @@ mod tests {
         assert_matches!(&results[0], Err(err) => {
             assert_eq!(
                 err.to_string(),
-                format!("record size ({record_bytes} bytes) exceeds max_batch_bytes ({max_batch_bytes})")
+                format!("record size in metered bytes ({record_bytes}) exceeds max_batch_bytes ({max_batch_bytes})")
             );
         });
 

src/batching.rs

@@ -93,7 +93,7 @@ pub mod types;
 pub use ops::{S2, S2Basin, S2Stream};
 /// Append session for pipelining multiple appends with backpressure control.
 ///
-/// See [AppendSession](append_session::AppendSession).
+/// See [`AppendSession`](append_session::AppendSession).
 pub mod append_session {
     pub use crate::session::append::{AppendSession, AppendSessionConfig, BatchSubmitTicket};
 }

src/lib.rs

@@ -22,14 +22,14 @@ pub struct S2 {
 }
 
 impl S2 {
-    /// Create a new [S2].
+    /// Create a new [`S2`].
     pub fn new(config: S2Config) -> Result<Self, S2Error> {
         Ok(Self {
             client: AccountClient::init(config)?,
         })
     }
 
-    /// Get an [S2Basin].
+    /// Get an [`S2Basin`].
     pub fn basin(&self, name: BasinName) -> S2Basin {
         S2Basin {
             client: self.client.basin_client(name),
@@ -148,13 +148,13 @@ impl S2 {
 #[derive(Debug, Clone)]
 /// A basin in an S2 account.
 ///
-/// See [S2::basin].
+/// See [`S2::basin`].
 pub struct S2Basin {
     client: BasinClient,
 }
 
 impl S2Basin {
-    /// Get an [S2Stream].
+    /// Get an [`S2Stream`].
     pub fn stream(&self, name: StreamName) -> S2Stream {
         S2Stream {
             client: self.client.clone(),
@@ -215,7 +215,7 @@ impl S2Basin {
 #[derive(Debug, Clone)]
 /// A stream in an S2 basin.
 ///
-/// See [S2Basin::stream].
+/// See [`S2Basin::stream`].
 pub struct S2Stream {
     client: BasinClient,
     name: StreamName,

src/ops.rs

@@ -21,7 +21,7 @@ use crate::{
     types::{AppendAck, AppendInput, AppendRecord, FencingToken, MeteredBytes, S2Error},
 };
 
-/// A [Future] that resolves to an acknowledgement once the record is appended.
+/// A [`Future`] that resolves to an acknowledgement once the record is appended.
 pub struct RecordSubmitTicket {
     rx: oneshot::Receiver<Result<IndexedAppendAck, S2Error>>,
 }
@@ -40,6 +40,7 @@ impl Future for RecordSubmitTicket {
 
 /// Acknowledgement for an appended record.
 #[derive(Debug, Clone)]
+#[non_exhaustive]
 pub struct IndexedAppendAck {
     /// Sequence number assigned to the record.
     pub seq_num: u64,
@@ -49,41 +50,41 @@ pub struct IndexedAppendAck {
 
 #[derive(Debug, Clone, Default)]
 #[non_exhaustive]
-/// Configuration for [Producer].
+/// Configuration for [`Producer`].
 pub struct ProducerConfig {
-    /// Configuration for batching records into [AppendInput]s before appending.
+    /// Configuration for batching records into [`AppendInput`]s before appending.
     pub batching: BatchingConfig,
-    /// Fencing token for all [AppendInput]s.
+    /// Fencing token for all [`AppendInput`]s.
     ///
     /// Defaults to `None`.
     pub fencing_token: Option<FencingToken>,
-    /// Match sequence number for the initial [AppendInput]. It will be auto-incremented for the
+    /// Match sequence number for the initial [`AppendInput`]. It will be auto-incremented for the
     /// subsequent ones.
     ///
     /// Defaults to `None`.
     pub match_seq_num: Option<u64>,
 }
 
 impl ProducerConfig {
-    /// Create a new [ProducerConfig] with default settings.
+    /// Create a new [`ProducerConfig`] with default settings.
     pub fn new() -> Self {
         Self::default()
     }
 
-    /// Set the configuration for batching records into [AppendInput]s before appending.
+    /// Set the configuration for batching records into [`AppendInput`]s before appending.
     pub fn with_batching(self, batching: BatchingConfig) -> Self {
         Self { batching, ..self }
     }
 
-    /// Set the fencing token for all [AppendInput]s.
+    /// Set the fencing token for all [`AppendInput`]s.
     pub fn with_fencing_token(self, fencing_token: FencingToken) -> Self {
         Self {
             fencing_token: Some(fencing_token),
             ..self
         }
     }
 
-    /// Set the match sequence number for the initial [AppendInput]. It will be auto-incremented
+    /// Set the match sequence number for the initial [`AppendInput`]. It will be auto-incremented
     /// for the subsequent ones.
     pub fn with_match_seq_num(self, match_seq_num: u64) -> Self {
         Self {
@@ -93,17 +94,17 @@ impl ProducerConfig {
     }
 }
 
-/// High-level interface for submitting individual [AppendRecord]s.
+/// High-level interface for submitting individual [`AppendRecord`]s.
 ///
-/// Wraps an [AppendSession] and handles batching records into [AppendInput]s automatically based on
-/// the provided [configuration](ProducerConfig).
+/// Wraps an [`AppendSession`] and handles batching records into [`AppendInput`]s automatically
+/// based on the provided [`configuration`](ProducerConfig).
 pub struct Producer {
     cmd_tx: mpsc::Sender<Command>,
     _handle: AbortOnDropHandle<()>,
 }
 
 impl Producer {
-    /// Create a new [Producer].
+    /// Create a new [`Producer`].
     pub fn new(session: AppendSession, config: ProducerConfig) -> Self {
         let (cmd_tx, cmd_rx) = mpsc::channel::<Command>(RECORD_BATCH_MAX.count);
         let _handle = AbortOnDropHandle::new(tokio::spawn(Self::run(session, config, cmd_rx)));
@@ -112,8 +113,7 @@ impl Producer {
 
     /// Submit a record for appending.
     ///
-    /// **Note**: You must call [Producer::close] if want to ensure all submitted records are
-    /// appended.
+    /// **Note**: You must call [`Producer::close`] to ensure all submitted records are appended.
     pub async fn submit(&self, record: AppendRecord) -> Result<RecordSubmitTicket, S2Error> {
         let (ack_tx, ack_rx) = oneshot::channel();
         self.cmd_tx