fix: eliminate production panics in client decoder and server startup

crates/ember-client/src/commands.rs

@@ -364,8 +364,13 @@ fn scan_page(frame: Frame) -> Result<ScanPage, ClientError> {
     }
 
     let mut iter = elems.into_iter();
-    let cursor_frame = iter.next().unwrap();
-    let keys_frame = iter.next().unwrap();
+    // Safety: length was validated to be exactly 2 above.
+    let cursor_frame = iter.next().ok_or_else(|| {
+        ClientError::Protocol("SCAN response missing cursor element".into())
+    })?;
+    let keys_frame = iter.next().ok_or_else(|| {
+        ClientError::Protocol("SCAN response missing keys element".into())
+    })?;
 
     let cursor = match cursor_frame {
         Frame::Bulk(b) => {
@@ -443,7 +448,9 @@ fn slowlog_entries(frame: Frame) -> Result<Vec<SlowlogEntry>, ClientError> {
                     )))
                 }
             };
-            let command = match entry.into_iter().nth(3).unwrap() {
+            let command = match entry.into_iter().nth(3).ok_or_else(|| {
+                ClientError::Protocol("slowlog entry missing command field".into())
+            })? {
                 Frame::Array(parts) => parts
                     .into_iter()
                     .map(|p| match p {

crates/ember-server/src/connection/execute.rs

@@ -28,14 +28,20 @@ fn set_expire_to_duration(expire: SetExpire) -> Duration {
         SetExpire::ExAt(ts) => {
             let now = SystemTime::now()
                 .duration_since(UNIX_EPOCH)
-                .unwrap_or_default()
+                .unwrap_or_else(|_| {
+                    tracing::warn!("system clock is before UNIX epoch; EXAT TTL calculations may be incorrect");
+                    Duration::ZERO
+                })
                 .as_secs();
             Duration::from_secs(ts.saturating_sub(now))
         }
         SetExpire::PxAt(ts_ms) => {
             let now_ms = SystemTime::now()
                 .duration_since(UNIX_EPOCH)
-                .unwrap_or_default()
+                .unwrap_or_else(|_| {
+                    tracing::warn!("system clock is before UNIX epoch; PXAT TTL calculations may be incorrect");
+                    Duration::ZERO
+                })
                 .as_millis() as u64;
             Duration::from_millis(ts_ms.saturating_sub(now_ms))
         }
@@ -2101,7 +2107,9 @@ pub(super) async fn execute(
 
             // Start with the smallest set to minimise comparisons.
             sets.sort_unstable_by_key(|s| s.len());
-            let (first, rest) = sets.split_first().expect("non-empty");
+            let Some((first, rest)) = sets.split_first() else {
+                return Frame::Integer(0);
+            };
             let mut count = 0usize;
             'outer: for member in first {
                 for other in rest {

crates/ember-server/src/server.rs

@@ -793,10 +793,16 @@ pub async fn run_threaded(
                 .spawn(move || {
                     pin_to_core(id);
 
-                    let rt = tokio::runtime::Builder::new_current_thread()
+                    let rt = match tokio::runtime::Builder::new_current_thread()
                         .enable_all()
                         .build()
-                        .expect("failed to build worker runtime");
+                    {
+                        Ok(rt) => rt,
+                        Err(e) => {
+                            error!("worker {id}: failed to build tokio runtime: {e}");
+                            return;
+                        }
+                    };
 
                     rt.block_on(worker_main(
                         id, prepared, addr, engine, ctx, slow_log, pubsub, semaphore, tls, shutdown,
@@ -805,9 +811,9 @@ pub async fn run_threaded(
                     // give in-flight connection handlers time to finish
                     rt.shutdown_timeout(Duration::from_secs(30));
                 })
-                .expect("failed to spawn worker thread")
+                .map_err(|e| std::io::Error::other(format!("failed to spawn worker thread {id}: {e}")))
         })
-        .collect();
+        .collect::<Result<Vec<_>, std::io::Error>>()?;
 
     info!(
         "listening on {addr} with {shard_count} shards, thread-per-core (max {max_conn} connections)"