feat: allow append concurrency control on fence and trim too

Cargo.toml

@@ -6,6 +6,7 @@ license = "Apache-2.0"
 
 [dependencies]
 async-stream = "0.3.6"
+base16ct = { version = "0.2.0", features = ["alloc"] }
 clap = { version = "4.5.20", features = ["derive"] }
 color-print = "0.3.6"
 colored = "2.1.0"
@@ -26,4 +27,4 @@ tracing-subscriber = { version = "0.3.18", features = ["env-filter"] }
 
 [dependencies.streamstore]
 git = "https://github.com/s2-streamstore/s2-sdk-rust.git"
-rev = "b1547bcd55a4f3b5b76104a6ffd374014671d617"
+rev = "8f6995159a79f0d2334d071115dfecf9965fda73"

src/error.rs

@@ -7,7 +7,7 @@ use crate::config::S2ConfigError;
 const HELP: &str = color_print::cstr!(
     "\n<cyan><bold>Notice something wrong?</bold></cyan>\n\n\
      <green> > Open an issue:</green>\n\
-     <bold>https://github.com/s2-cli/issues</bold>\n\n\
+     <bold>https://github.com/s2-streamstore/s2-cli/issues</bold>\n\n\
      <green> > Reach out to us:</green>\n\
      <bold>hi@s2.dev</bold>"
 );

src/main.rs

@@ -13,8 +13,8 @@ use stream::{RecordStream, StreamService};
 use streamstore::{
     client::{BasinClient, Client, ClientConfig, S2Endpoints, StreamClient},
     types::{
-        BasinInfo, BasinName, CommandRecord, FencingToken, MeteredBytes as _, ReadOutput,
-        StreamInfo,
+        BasinInfo, BasinName, CommandRecord, ConvertError, FencingToken, MeteredBytes as _,
+        ReadOutput, StreamInfo,
     },
     HeaderValue,
 };
@@ -194,26 +194,46 @@ enum Commands {
         /// Name of the stream.
         stream: String,
 
-        /// Trim point.
-        /// This sequence number is only allowed to advance, and any regression
-        /// will be ignored.
+        /// Earliest sequence number that should be retained.
+        /// This sequence number is only allowed to advance,
+        /// and any regression will be ignored.
         trim_point: u64,
+
+        /// Enforce fencing token specified in hex.
+        #[arg(short = 'f', long, value_parser = parse_fencing_token)]
+        fencing_token: Option<FencingToken>,
+
+        /// Enforce that the sequence number issued to the first record matches.
+        #[arg(short = 'm', long)]
+        match_seq_num: Option<u64>,
     },
 
-    /// Set the fencing token for the stream.
+    /// Set a fencing token for the stream.
     ///
     /// Fencing is strongly consistent, and subsequent appends that specify a
-    /// fencing token will be rejected if it does not match.
+    /// token will be rejected if it does not match.
+    ///
+    /// Note that fencing is a cooperative mechanism,
+    /// and it is only enforced when a token is provided.
     Fence {
         /// Name of the basin.
         basin: BasinName,
 
         /// Name of the stream.
         stream: String,
 
-        /// Payload upto 16 bytes in hex to set as the fencing token.
-        /// An empty payload clears the token.
+        /// New fencing token specified in hex.
+        /// It may be upto 16 bytes, and can be empty.
+        #[arg(value_parser = parse_fencing_token)]
+        new_fencing_token: FencingToken,
+
+        /// Enforce existing fencing token, specified in hex.
+        #[arg(short = 'f', long, value_parser = parse_fencing_token)]
         fencing_token: Option<FencingToken>,
+
+        /// Enforce that the sequence number issued to this command matches.
+        #[arg(short = 'm', long)]
+        match_seq_num: Option<u64>,
     },
 
     /// Append records to a stream.
@@ -226,9 +246,8 @@ enum Commands {
         /// Name of the stream.
         stream: String,
 
-        /// Enforce a fencing token specified in hex,
-        /// which must have been previously set by a `fence` command.
-        #[arg(short = 'f', long)]
+        /// Enforce fencing token specified in hex.
+        #[arg(short = 'f', long, value_parser = parse_fencing_token)]
         fencing_token: Option<FencingToken>,
 
         /// Enforce that the sequence number issued to the first record matches.
@@ -340,6 +359,12 @@ fn parse_records_output_source(s: &str) -> Result<RecordsOut, std::io::Error> {
     }
 }
 
+fn parse_fencing_token(s: &str) -> Result<FencingToken, ConvertError> {
+    base16ct::mixed::decode_vec(s)
+        .map_err(|_| "invalid hex")?
+        .try_into()
+}
+
 fn client_config(auth_token: String) -> Result<ClientConfig, S2CliError> {
     let endpoints = S2Endpoints::from_env().map_err(S2CliError::EndpointsFromEnv)?;
     let client_config = ClientConfig::new(auth_token.to_string())
@@ -576,28 +601,50 @@ async fn run() -> Result<(), S2CliError> {
             basin,
             stream,
             trim_point,
+            fencing_token,
+            match_seq_num,
         } => {
             let cfg = config::load_config(&config_path)?;
             let client_config = client_config(cfg.auth_token)?;
             let stream_client = StreamClient::new(client_config, basin, stream);
-            StreamService::new(stream_client)
-                .append_command_record(CommandRecord::trim(trim_point))
+            let out = StreamService::new(stream_client)
+                .append_command_record(
+                    CommandRecord::trim(trim_point),
+                    fencing_token,
+                    match_seq_num,
+                )
                 .await?;
-            eprintln!("{}", "✓ Trim requested".green().bold());
+            eprintln!(
+                "{}",
+                format!("✓ Trim requested at seq_num={}", out.start_seq_num)
+                    .green()
+                    .bold()
+            );
         }
 
         Commands::Fence {
             basin,
             stream,
+            new_fencing_token,
             fencing_token,
+            match_seq_num,
         } => {
             let cfg = config::load_config(&config_path)?;
             let client_config = client_config(cfg.auth_token)?;
             let stream_client = StreamClient::new(client_config, basin, stream);
-            StreamService::new(stream_client)
-                .append_command_record(CommandRecord::fence(fencing_token))
+            let out = StreamService::new(stream_client)
+                .append_command_record(
+                    CommandRecord::fence(new_fencing_token),
+                    fencing_token,
+                    match_seq_num,
+                )
                 .await?;
-            eprintln!("{}", "✓ Fencing token set".green().bold());
+            eprintln!(
+                "{}",
+                format!("✓ Fencing token set at seq_num: {}", out.start_seq_num)
+                    .green()
+                    .bold()
+            );
         }
 
         Commands::Append {
@@ -704,7 +751,7 @@ async fn run() -> Result<(), S2CliError> {
                                                 let (cmd, description) = match command_record {
                                                     CommandRecord::Fence { fencing_token } => (
                                                         "fence",
-                                                        format!("{fencing_token:?}"),
+                                                        format!("FencingToken({})", base16ct::lower::encode_string(fencing_token.as_ref())),
                                                     ),
                                                     CommandRecord::Trim { seq_num } => (
                                                         "trim",

src/stream.rs

@@ -74,14 +74,21 @@ impl StreamService {
     pub async fn append_command_record(
         &self,
         cmd: CommandRecord,
+        fencing_token: Option<FencingToken>,
+        match_seq_num: Option<u64>,
     ) -> Result<AppendOutput, ServiceError> {
         let context = match &cmd {
             CommandRecord::Fence { .. } => ServiceErrorContext::Fence,
             CommandRecord::Trim { .. } => ServiceErrorContext::Trim,
         };
-        let batch = AppendRecordBatch::try_from_iter([cmd]).expect("single valid append record");
+        let records = AppendRecordBatch::try_from_iter([cmd]).expect("single valid append record");
+        let append_input = AppendInput {
+            records,
+            fencing_token,
+            match_seq_num,
+        };
         self.client
-            .append(AppendInput::new(batch))
+            .append(append_input)
             .await
             .map_err(|e| ServiceError::new(context, e))
     }