fix: resolve code-quality issues #396-#401

Eliminate panic!/unreachable!/unimplemented! from production paths and
harden runtime behavior across the workspace.

- #396: replace panic/unreachable in production paths with Result
  propagation or graceful degradation:
  - executor.rs: degrade to error reply instead of panicking on send
  - baseline.rs: from_u8 returns Option, state() falls back to Abandoned
  - redis_lists.rs: return Err via OptionNoneSnafu instead of unreachable
  - checkpoint.rs: return io::Error instead of unreachable
  - encode.rs: fall back to "txt" for invalid VerbatimString format
  - network.rs: log + fallback endpoint instead of panicking on bad addr
  - storage_murmur3.rs: add message to unreachable
- #397: SlotIndexer::reshard_slots returns Err instead of unimplemented!
- #398: net/pipeline switches unbounded mpsc to bounded channel with
  backpressure (config.command_queue_size)
- #399: optimized_handler captures client write errors and cleans up the
  connection instead of silently dropping
- #400: tag ownerless TODO/FIXME with #400 tracking id
- #401: add #![allow(clippy::unwrap_used)] to raft/storage test modules,
  eliminating 135 clippy errors so `cargo clippy --all-targets` passes

Also cleans up pre-existing workspace clippy warnings (complex type
alias, doc indentation, useless_vec, items_after_test_module,
field_reassign_with_default) and a missing required-features gate on
storage_stats_test, so `cargo clippy --workspace --all-targets` reaches
zero errors.

Verified: cargo check --workspace and cargo clippy --workspace
--all-targets -- -D warnings -D clippy::unwrap_used both pass with 0
errors; resp/client/kstd lib tests pass. RocksDB-dependent tests cannot
run on this Windows host (pre-existing DLL entrypoint issue), to be
confirmed in CI.

Closes #396
Closes #397
Closes #398
Closes #399
Closes #400
Closes #401

Author: AlexStocks

src/cmd/src/set.rs

@@ -48,7 +48,7 @@ impl Cmd for SetCmd {
 
     /// SET key value
     fn do_initial(&self, client: &Client) -> bool {
-        // TODO: support xx, nx, ex, px
+        // TODO(#400): support xx, nx, ex, px
         let argv = client.argv();
 
         let key = argv[1].clone();

src/common/runtime/additional_unit_tests.rs

@@ -372,7 +372,7 @@ mod serialization_tests {
 
     #[test]
     fn test_noop_storage_stats_collector() {
-        let collector = NoopStorageStatsCollector::default();
+        let collector = NoopStorageStatsCollector;
 
         collector.record_read(3, 5);
         collector.record_write(7, 11);

src/common/runtime/baseline.rs

@@ -80,17 +80,17 @@ pub enum AttemptState {
 }
 
 impl AttemptState {
-    fn from_u8(value: u8) -> Self {
+    fn from_u8(value: u8) -> Option<Self> {
         match value {
-            0 => Self::Offered,
-            1 => Self::ChannelQueued,
-            2 => Self::BatchQueued,
-            3 => Self::WaitingGate,
-            4 => Self::Running,
-            5 => Self::ExecutionFinished,
-            6 => Self::ShutdownRejectedAfterAccept,
-            7 => Self::Abandoned,
-            _ => unreachable!("baseline attempt state is only written from AttemptState"),
+            0 => Some(Self::Offered),
+            1 => Some(Self::ChannelQueued),
+            2 => Some(Self::BatchQueued),
+            3 => Some(Self::WaitingGate),
+            4 => Some(Self::Running),
+            5 => Some(Self::ExecutionFinished),
+            6 => Some(Self::ShutdownRejectedAfterAccept),
+            7 => Some(Self::Abandoned),
+            _ => None,
         }
     }
 
@@ -322,6 +322,7 @@ impl BaselineAttempt {
 
     pub fn state(&self) -> AttemptState {
         AttemptState::from_u8(self.inner.state.load(Ordering::Acquire))
+            .unwrap_or(AttemptState::Abandoned)
     }
 
     /// Irrevocably reject a physical attempt before the request channel accepts it.
@@ -1912,7 +1913,7 @@ mod tests {
         source.transition(AttemptState::ChannelQueued).unwrap();
 
         assert_eq!(
-            AttemptState::from_u8(target_inner.state.load(Ordering::Acquire)),
+            AttemptState::from_u8(target_inner.state.load(Ordering::Acquire)).unwrap(),
             AttemptState::Abandoned
         );
         assert!(target_handle.failed());

src/common/runtime/storage_server.rs

@@ -1096,7 +1096,7 @@ impl BackgroundTaskManager {
 
                         // Record compaction started
                         if let Some(ref _tracker) = metrics_tracker {
-                            // TODO: Fix type annotation issue
+                            // TODO(#400): re-enable after type annotation fixed
                             // tracker.record_compaction_started().await;
                         }
 
@@ -1112,7 +1112,7 @@ impl BackgroundTaskManager {
 
                             // Record compaction completed
                             if let Some(ref _tracker) = metrics_tracker {
-                                // TODO: Fix type annotation issue
+                                // TODO(#400): re-enable after type annotation fixed
                                 // tracker.record_compaction_completed(0, compaction_duration).await; // TODO: Get actual bytes compacted
                             }
                         }
@@ -1170,7 +1170,7 @@ impl BackgroundTaskManager {
 
                             // Record flush started
                             if let Some(ref _tracker) = metrics_tracker {
-                                // TODO: Fix type annotation issue
+                                // TODO(#400): re-enable after type annotation fixed
                                 // tracker.record_flush_started().await;
                             }
 
@@ -1180,7 +1180,7 @@ impl BackgroundTaskManager {
 
                             // Record flush completed (simulated)
                             if let Some(ref _tracker) = metrics_tracker {
-                                // TODO: Fix type annotation issue
+                                // TODO(#400): re-enable after type annotation fixed
                                 // tracker.record_flush_completed(0, Duration::from_millis(10)).await; // TODO: Get actual flush metrics
                             }
                         }
@@ -1236,7 +1236,7 @@ impl BackgroundTaskManager {
 
                     // Update metrics tracker if available
                     if let Some(ref _tracker) = metrics_tracker {
-                        // TODO: Fix type annotation issue
+                        // TODO(#400): re-enable after type annotation fixed
                         // tracker.update_rocksdb_metrics(
                         //     rocksdb_stats.total_keys,
                         //     rocksdb_stats.total_size_bytes,

src/executor/src/executor.rs

@@ -100,15 +100,18 @@ impl CmdExecutor {
         // send the work to the worker pool
         match self.work_tx.send(work).await {
             Ok(_) => {}
-            Err(async_channel::SendError(_)) => {
+            Err(async_channel::SendError(work)) => {
                 // this should not happen, because the only case when all the workers
                 // has been closed is when the executor is closed. and we've already
-                // checked the cancellation_token.
-                panic!("Failed to send work to worker; executor likely closed");
+                // checked the cancellation_token. Degrade gracefully instead of panicking.
+                error!("failed to send work to worker; executor likely closed");
+                work.exec
+                    .client
+                    .set_reply(RespData::Error("ERR executor unavailable".into()));
             }
         }
 
-        // TODO: add a timeout for waiting
+        // TODO(#400): add a timeout for waiting
         let _ = done_rx.await;
     }
 

src/kstd/src/env.rs

@@ -29,7 +29,7 @@ pub fn is_dir<P: AsRef<Path>>(path: P) -> io::Result<bool> {
 
 /// Creates a directory and all its parent directories with the specified mode.
 /// This corresponds to the 'mkpath' functionality.
-/// TODO: remove allow dead code
+/// TODO(#400): remove allow dead code
 #[allow(dead_code)]
 pub fn mkdir_with_path<P: AsRef<Path>>(path: P, _mode: u32) -> io::Result<()> {
     // Use the fs::create_dir_all method to create the directory path.
@@ -46,7 +46,7 @@ pub fn mkdir_with_path<P: AsRef<Path>>(path: P, _mode: u32) -> io::Result<()> {
     Ok(())
 }
 
-/// TODO: remove allow dead code
+/// TODO(#400): remove allow dead code
 #[allow(dead_code)]
 pub fn delete_dir<P: AsRef<Path>>(dirname: P) -> io::Result<()> {
     let path = dirname.as_ref();

src/kstd/src/status.rs

@@ -23,7 +23,7 @@ pub struct Status {
     message: String,
 }
 
-/// TODO: remove allow dead code.
+/// TODO(#400): remove allow dead code.
 #[allow(dead_code)]
 #[derive(Debug, PartialEq)]
 pub enum Code {
@@ -32,7 +32,7 @@ pub enum Code {
     Busy,
 }
 
-/// TODO: remove allow dead code.
+/// TODO(#400): remove allow dead code.
 #[allow(dead_code)]
 impl Status {
     // Create a success status.

src/net/src/lib.rs

@@ -28,7 +28,7 @@ pub mod pool;
 pub mod storage_client;
 pub mod tcp;
 
-// TODO: delete this module
+// TODO(#400): delete this module
 pub mod error;
 pub mod unix;
 

src/net/src/optimized_handler.rs

@@ -201,7 +201,13 @@ impl OptimizedConnectionHandler {
                                             let error_response = RespData::Error(format!("ERR {}", e).into());
                                             let mut encoder = RespEncoder::new(RespVersion::RESP2);
                                             encoder.encode_resp_data(&error_response);
-                                            let _ = client.write(encoder.get_response().as_ref()).await;
+                                            if let Err(write_err) = client.write(encoder.get_response().as_ref()).await {
+                                                warn!("failed to write error response to client: {write_err}; closing connection");
+                                                if self.config.enable_buffer_pooling {
+                                                    self.buffer_manager.return_buffer(read_buffer).await;
+                                                }
+                                                return Ok(());
+                                            }
                                         }
                                     }
                                 }

src/net/src/pipeline.rs

@@ -106,7 +106,7 @@ pub struct CommandPipeline {
     storage: Arc<Storage>,
     cmd_table: Arc<CmdTable>,
     executor: Arc<CmdExecutor>,
-    command_tx: mpsc::UnboundedSender<PipelineCommand>,
+    command_tx: mpsc::Sender<PipelineCommand>,
     semaphore: Arc<Semaphore>,
     batch_counter: Arc<std::sync::atomic::AtomicU64>,
 }
@@ -118,9 +118,10 @@ impl CommandPipeline {
         cmd_table: Arc<CmdTable>,
         executor: Arc<CmdExecutor>,
     ) -> Self {
-        // TODO: Consider using bounded channel instead of unbounded to properly enforce queue_capacity
-        // Currently using unbounded_channel means the command_queue_size config is not actually enforced
-        let (command_tx, command_rx) = mpsc::unbounded_channel();
+        // Bounded channel enforces backpressure: when the queue is full,
+        // senders await instead of growing the queue unboundedly.
+        // Guard against 0 (rendezvous) to avoid unintentional stalls.
+        let (command_tx, command_rx) = mpsc::channel(config.command_queue_size.max(1));
         let semaphore = Arc::new(Semaphore::new(config.max_concurrent_pipelines));
 
         let pipeline = Self {
@@ -154,9 +155,11 @@ impl CommandPipeline {
             response_tx,
         };
 
-        // Send command to pipeline
+        // Send command to pipeline. On a full bounded channel this awaits
+        // (backpressure) rather than growing the queue indefinitely.
         self.command_tx
             .send(command)
+            .await
             .map_err(|_| PipelineError::ChannelClosed)?;
 
         // Wait for response with timeout
@@ -167,7 +170,7 @@ impl CommandPipeline {
     }
 
     /// Start the background batch processor
-    fn start_batch_processor(&self, mut command_rx: mpsc::UnboundedReceiver<PipelineCommand>) {
+    fn start_batch_processor(&self, mut command_rx: mpsc::Receiver<PipelineCommand>) {
         let config = self.config.clone();
         let storage = self.storage.clone();
         let cmd_table = self.cmd_table.clone();
@@ -368,11 +371,7 @@ impl CommandPipeline {
         PipelineStats {
             available_permits: self.semaphore.available_permits(),
             max_concurrent_pipelines: self.config.max_concurrent_pipelines,
-            // Note: queue_capacity stat is currently misleading because the channel at line 121
-            // is unbounded (mpsc::unbounded_channel). This returns command_queue_size from config,
-            // but the actual channel has no capacity limit.
-            // TODO: Consider using bounded channel (mpsc::channel) and passing config.command_queue_size
-            // as the capacity to make this stat accurate.
+            // The channel is bounded by config.command_queue_size.
             queue_capacity: self.config.command_queue_size,
         }
     }