fix: prevent PR #403 failure paths from escaping verification

Bound queue admission and executor completion failures, then aligned CI gates and regression coverage so silent stalls and skipped targets block review before merge.

Constraint: Preserve public APIs and exclude the Base-only RPOPLPUSH atomicity defect.

Rejected: Remove the storage stats feature gate | it is required to keep fault injection out of production-only builds.

Confidence: high

Scope-risk: moderate

Tested: WSL fmt; workspace all-targets all-features Clippy; net, executor, runtime, RESP, Raft, storage, and fault-injection storage stats tests.

Not-tested: Fresh GitHub Actions on this commit.

Co-authored-by: OmX <omx@oh-my-codex.dev>

Author: AlexStocks

.github/workflows/ci.yml

@@ -136,9 +136,6 @@ jobs:
 
       - run: make lint
 
-      - name: Lint runtime baseline tool targets
-        run: cargo clippy -p runtime-baseline --all-targets -- -D warnings -D clippy::unwrap_used
-
   build-and-test:
     strategy:
       fail-fast: false
@@ -223,6 +220,10 @@ jobs:
           max_attempts: 2
           command: bash -o pipefail -c "make test 2>&1 | tee cargo-test.log"
 
+      - name: Test storage stats with fault injection
+        if: matrix.os == 'ubuntu-latest'
+        run: cargo test -p storage --features test-fault-injection --test storage_stats_test
+
       - name: Test runtime baseline tool
         run: cargo test -p runtime-baseline --all-targets
 

Makefile

@@ -51,7 +51,7 @@ fmt-check:
 	@cargo fmt --manifest-path ./Cargo.toml --all -- --check
 
 lint:
-	@cargo clippy --manifest-path ./Cargo.toml --all-features --workspace -- -D warnings -D clippy::unwrap_used
+	@cargo clippy --manifest-path ./Cargo.toml --all-features --workspace --all-targets -- -D warnings -D clippy::unwrap_used
 
 standalone:
 	@$(DEV) build

docs/quality/quality-gates.md

@@ -16,7 +16,7 @@
 ```text
 license header / third-party license check
 cargo fmt --check
-cargo clippy --all-features --workspace -- -D warnings -D clippy::unwrap_used
+cargo clippy --all-features --workspace --all-targets -- -D warnings -D clippy::unwrap_used
 targeted unit tests
 affected compatibility manifest tests
 Cache OFF compatibility and recovery evidence

src/common/runtime/baseline.rs

@@ -1015,6 +1015,45 @@ mod tests {
         )
     }
 
+    #[test]
+    fn attempt_state_decodes_every_valid_discriminant() {
+        let states = [
+            AttemptState::Offered,
+            AttemptState::ChannelQueued,
+            AttemptState::BatchQueued,
+            AttemptState::WaitingGate,
+            AttemptState::Running,
+            AttemptState::ExecutionFinished,
+            AttemptState::ShutdownRejectedAfterAccept,
+            AttemptState::Abandoned,
+        ];
+
+        for state in states {
+            assert_eq!(AttemptState::from_u8(state as u8), Some(state));
+        }
+    }
+
+    #[test]
+    fn attempt_state_rejects_unknown_discriminant() {
+        assert_eq!(AttemptState::from_u8(8), None);
+        assert_eq!(AttemptState::from_u8(u8::MAX), None);
+    }
+
+    #[test]
+    fn attempt_state_accessor_falls_back_to_abandoned_for_unknown_raw_state() {
+        let observer = Arc::new(RecordingObserver::default());
+        let attempt = attempt(observer);
+
+        attempt.inner.state.store(u8::MAX, Ordering::Release);
+        assert_eq!(attempt.state(), AttemptState::Abandoned);
+
+        // Restore a valid pre-accept state so Drop follows the ordinary path.
+        attempt
+            .inner
+            .state
+            .store(AttemptState::Offered as u8, Ordering::Release);
+    }
+
     #[test]
     fn legal_transitions_record_previous_and_next_state() {
         let observer = Arc::new(RecordingObserver::default());

src/executor/src/executor.rs

@@ -83,6 +83,7 @@ impl CmdExecutor {
     }
 
     pub async fn execute(&self, exec: CmdExecution) {
+        let client = Arc::clone(&exec.client);
         let (done_tx, done_rx) = oneshot::channel();
         let work = CmdExecutionWork {
             exec,
@@ -108,11 +109,15 @@ impl CmdExecutor {
                 work.exec
                     .client
                     .set_reply(RespData::Error("ERR executor unavailable".into()));
+                return;
             }
         }
 
         // TODO(#400): add a timeout for waiting
-        let _ = done_rx.await;
+        if let Err(err) = done_rx.await {
+            error!("executor worker exited before completing work: {err}");
+            client.set_reply(RespData::Error("ERR executor unavailable".into()));
+        }
     }
 
     pub async fn close(&mut self) {
@@ -366,6 +371,66 @@ mod tests {
         executor.close().await;
     }
 
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn accepted_work_reports_unavailable_when_worker_exits() {
+        let storage = Arc::new(Storage::new(1, 0));
+        let exclusive = storage.acquire_exclusive_command_access().await;
+        let attempts = Arc::new(AtomicUsize::new(0));
+        let attempt_notify = Arc::new(tokio::sync::Notify::new());
+        let executed = Arc::new(AtomicUsize::new(0));
+        let command = Arc::new(WaitingSharedCmd::new(
+            Arc::clone(&attempts),
+            Arc::clone(&attempt_notify),
+            Arc::clone(&executed),
+        ));
+        let client = Arc::new(Client::new(Box::new(TestStream::new())));
+        client.set_cmd_name(b"waiting-shared");
+        client.set_argv(&[b"waiting-shared".to_vec()]);
+        let executor = Arc::new(CmdExecutor::new(1, 1));
+
+        let executor_for_task = Arc::clone(&executor);
+        let client_for_task = Arc::clone(&client);
+        let storage_for_task = Arc::clone(&storage);
+        let execute_task = tokio::spawn(async move {
+            executor_for_task
+                .execute(CmdExecution {
+                    cmd: command,
+                    client: client_for_task,
+                    storage: storage_for_task,
+                })
+                .await;
+        });
+
+        tokio::time::timeout(Duration::from_secs(1), attempt_notify.notified())
+            .await
+            .expect("worker must dequeue accepted work");
+        assert_eq!(attempts.load(Ordering::SeqCst), 1);
+
+        executor.workers[0].abort();
+        tokio::time::timeout(Duration::from_secs(1), async {
+            while !executor.workers[0].is_finished() {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("aborted worker must finish");
+        tokio::time::timeout(Duration::from_secs(1), execute_task)
+            .await
+            .expect("execute must observe the cancelled completion channel")
+            .expect("execute task must not panic");
+
+        assert_eq!(executed.load(Ordering::SeqCst), 0);
+        assert_eq!(
+            client.take_reply(),
+            RespData::Error("ERR executor unavailable".into())
+        );
+
+        drop(exclusive);
+        let mut executor = Arc::try_unwrap(executor)
+            .unwrap_or_else(|_| panic!("test must release all executor references"));
+        executor.close().await;
+    }
+
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     async fn executor_routes_commands_through_shared_storage_access() {
         let storage = Arc::new(Storage::new(1, 0));

src/net/Cargo.toml

@@ -28,6 +28,7 @@ raft = { path = "../raft" }
 
 [dev-dependencies]
 env_logger = "0.11"
+tokio = { workspace = true, features = ["test-util"] }
 
 [[bench]]
 name = "network_benchmark"

src/net/src/pipeline.rs

@@ -27,6 +27,8 @@ use storage::storage::Storage;
 use tokio::sync::{Semaphore, mpsc, oneshot};
 use tokio::time::timeout;
 
+const COMMAND_TIMEOUT: Duration = Duration::from_secs(30);
+
 /// Configuration for pipeline processing
 #[derive(Debug, Clone)]
 pub struct PipelineConfig {
@@ -113,15 +115,16 @@ pub struct CommandPipeline {
 
 impl CommandPipeline {
     pub fn new(
-        config: PipelineConfig,
+        mut config: PipelineConfig,
         storage: Arc<Storage>,
         cmd_table: Arc<CmdTable>,
         executor: Arc<CmdExecutor>,
     ) -> Self {
         // 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));
+        config.command_queue_size = config.command_queue_size.max(1);
+        let (command_tx, command_rx) = mpsc::channel(config.command_queue_size);
         let semaphore = Arc::new(Semaphore::new(config.max_concurrent_pipelines));
 
         let pipeline = Self {
@@ -155,18 +158,7 @@ impl CommandPipeline {
             response_tx,
         };
 
-        // 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
-        timeout(Duration::from_secs(30), response_rx)
-            .await
-            .map_err(|_| PipelineError::Timeout)?
-            .map_err(|_| PipelineError::ResponseChannelClosed)
+        send_and_receive(&self.command_tx, command, response_rx, COMMAND_TIMEOUT).await
     }
 
     /// Start the background batch processor
@@ -396,12 +388,122 @@ pub enum PipelineError {
     Timeout,
 }
 
+async fn send_and_receive(
+    command_tx: &mpsc::Sender<PipelineCommand>,
+    command: PipelineCommand,
+    response_rx: oneshot::Receiver<RespData>,
+    request_timeout: Duration,
+) -> Result<RespData, PipelineError> {
+    timeout(request_timeout, async {
+        command_tx
+            .send(command)
+            .await
+            .map_err(|_| PipelineError::ChannelClosed)?;
+        response_rx
+            .await
+            .map_err(|_| PipelineError::ResponseChannelClosed)
+    })
+    .await
+    .map_err(|_| PipelineError::Timeout)?
+}
+
 #[allow(clippy::unwrap_used)]
 #[cfg(test)]
 mod tests {
     use super::*;
     use std::time::Duration;
-    use tokio::time::sleep;
+    use tokio::time::{advance, sleep};
+
+    struct TestStream;
+
+    #[async_trait::async_trait]
+    impl client::StreamTrait for TestStream {
+        async fn read(&mut self, _buf: &mut [u8]) -> Result<usize, std::io::Error> {
+            Ok(0)
+        }
+
+        async fn write(&mut self, _data: &[u8]) -> Result<usize, std::io::Error> {
+            Ok(0)
+        }
+    }
+
+    fn pipeline_command(response_tx: oneshot::Sender<RespData>) -> PipelineCommand {
+        PipelineCommand {
+            data: RespData::Array(Some(Vec::new())),
+            client: Arc::new(Client::new(Box::new(TestStream))),
+            received_at: Instant::now(),
+            response_tx,
+        }
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn queue_wait_and_response_share_single_deadline() {
+        let (command_tx, mut command_rx) = mpsc::channel(1);
+        let (filler_response_tx, _filler_response_rx) = oneshot::channel();
+        command_tx
+            .send(pipeline_command(filler_response_tx))
+            .await
+            .expect("filler command must fit in the queue");
+
+        let (target_response_tx, target_response_rx) = oneshot::channel();
+        let target_command = pipeline_command(target_response_tx);
+        let submit_task = tokio::spawn(async move {
+            send_and_receive(
+                &command_tx,
+                target_command,
+                target_response_rx,
+                Duration::from_secs(30),
+            )
+            .await
+        });
+        tokio::task::yield_now().await;
+
+        advance(Duration::from_secs(20)).await;
+        let filler_command = command_rx
+            .recv()
+            .await
+            .expect("filler command must remain queued");
+        drop(filler_command);
+        tokio::task::yield_now().await;
+
+        advance(Duration::from_secs(11)).await;
+        tokio::task::yield_now().await;
+        assert!(
+            submit_task.is_finished(),
+            "queue wait and response wait must share one deadline"
+        );
+        let result = submit_task.await.expect("submission task must not panic");
+        assert!(matches!(result, Err(PipelineError::Timeout)));
+
+        let target_command = command_rx
+            .recv()
+            .await
+            .expect("target command must have entered the queue");
+        assert!(
+            target_command
+                .response_tx
+                .send(RespData::SimpleString("late".into()))
+                .is_err(),
+            "timing out must close the target response receiver"
+        );
+    }
+
+    #[tokio::test]
+    async fn zero_command_queue_size_reports_effective_capacity() {
+        let config = PipelineConfig {
+            command_queue_size: 0,
+            ..Default::default()
+        };
+        let pipeline = CommandPipeline::new(
+            config,
+            Arc::new(Storage::new(1, 0)),
+            Arc::new(CmdTable::new()),
+            Arc::new(CmdExecutor::new(1, 1)),
+        );
+
+        assert_eq!(pipeline.command_tx.max_capacity(), 1);
+        assert_eq!(pipeline.stats().queue_capacity, 1);
+    }
 
     #[tokio::test]
     async fn test_pipeline_basic() {

src/raft/src/network.rs

@@ -331,3 +331,39 @@ impl RaftNetwork<KiwiTypeConfig> for KiwiNetwork {
         Err(rpc_failed_network_error(&self.target, last_error))
     }
 }
+
+#[allow(clippy::unwrap_used)]
+#[cfg(test)]
+mod tests {
+    use openraft::Vote;
+
+    use super::*;
+
+    #[tokio::test]
+    async fn invalid_raft_endpoint_returns_network_error_without_panicking() {
+        let invalid_target = "not a valid raft address";
+        let mut factory = KiwiNetworkFactory::with_config(ConnectionConfig {
+            max_retries: 0,
+            ..Default::default()
+        });
+        let node = KiwiNode {
+            raft_addr: invalid_target.to_string(),
+            resp_addr: String::new(),
+        };
+
+        let mut network = factory.new_client(2, &node).await;
+        let error = network
+            .vote(
+                VoteRequest::new(Vote::new(1, 1), None),
+                RPCOption::new(Duration::from_millis(50)),
+            )
+            .await
+            .expect_err("invalid endpoint must fail as a network error");
+
+        assert!(matches!(error, RPCErr::Network(_)));
+        assert!(
+            error.to_string().contains(invalid_target),
+            "network error must identify the configured target: {error}"
+        );
+    }
+}