redis: harden connection lifecycle (Handshake enum, promise-settlement fixes, -1251 LOC)

docs/runtime/redis.mdx

@@ -476,6 +476,7 @@ Common error codes:
 - `ERR_REDIS_CONNECTION_CLOSED` - Connection to the server was closed
 - `ERR_REDIS_AUTHENTICATION_FAILED` - Failed to authenticate with the server
 - `ERR_REDIS_INVALID_RESPONSE` - Received an invalid response from the server
+- `ERR_REDIS_SERVER_ERROR` - The Redis server returned an error reply
 
 ---
 

src/collections/linear_fifo.rs

@@ -430,6 +430,15 @@ impl<T, B: LinearFifoBuffer<T>> LinearFifo<T, B> {
         }
     }
 
+    /// Iterate every readable element in FIFO order across both halves of a
+    /// possibly-wrapped ring. `readable_slice(0)` alone yields only the first
+    /// contiguous segment.
+    pub fn iter(&self) -> impl Iterator<Item = &T> {
+        let first = self.readable_slice(0);
+        let second = self.readable_slice(first.len());
+        first.iter().chain(second.iter())
+    }
+
     /// Discard first `count` items in the fifo
     pub fn discard(&mut self, count: usize) {
         debug_assert!(count <= self.count);
@@ -1098,6 +1107,26 @@ mod tests {
         assert_eq!(expected, vec![200, 201, 202, 203, 204, 206, 207]);
     }
 
+    #[test]
+    fn iter_visits_both_halves_of_wrapped_ring() {
+        let mut fifo = WrapFifo::init();
+        for v in 0..12 {
+            fifo.write_item(v).unwrap();
+        }
+        for _ in 0..8 {
+            fifo.read_item().unwrap();
+        }
+        for v in 100..110 {
+            fifo.write_item(v).unwrap();
+        }
+        assert!(fifo.buf_len() - fifo.head < fifo.count, "must be wrapped");
+        assert!(fifo.readable_slice(0).len() < fifo.readable_length());
+
+        let got: Vec<i32> = fifo.iter().copied().collect();
+        assert_eq!(got, fifo_to_vec(&fifo));
+        assert_eq!(got.len(), fifo.readable_length());
+    }
+
     // Exhaustively remove every valid offset from a wrapped layout and compare
     // against a reference `Vec`. Uses a fresh FIFO per offset (remove mutates).
     #[test]

src/jsc/bindings/ErrorCode.ts

@@ -292,16 +292,14 @@ const errors: ErrorCodeMapping = [
   ["ERR_REDIS_CONNECTION_TIMEOUT", Error, "RedisError"],
   ["ERR_REDIS_IDLE_TIMEOUT", Error, "RedisError"],
   ["ERR_REDIS_INVALID_ARGUMENT", Error, "RedisError"],
-  ["ERR_REDIS_INVALID_ARRAY", Error, "RedisError"],
   ["ERR_REDIS_INVALID_BULK_STRING", Error, "RedisError"],
   ["ERR_REDIS_INVALID_COMMAND", Error, "RedisError"],
   ["ERR_REDIS_INVALID_DATABASE", Error, "RedisError"],
-  ["ERR_REDIS_INVALID_ERROR_STRING", Error, "RedisError"],
   ["ERR_REDIS_INVALID_INTEGER", Error, "RedisError"],
   ["ERR_REDIS_INVALID_PASSWORD", Error, "RedisError"],
   ["ERR_REDIS_INVALID_RESPONSE", Error, "RedisError"],
   ["ERR_REDIS_INVALID_RESPONSE_TYPE", Error, "RedisError"],
-  ["ERR_REDIS_INVALID_SIMPLE_STRING", Error, "RedisError"],
+  ["ERR_REDIS_SERVER_ERROR", Error, "RedisError"],
   ["ERR_REDIS_INVALID_STATE", Error, "RedisError"],
   ["ERR_REDIS_INVALID_USERNAME", Error, "RedisError"],
   ["ERR_REDIS_TLS_NOT_AVAILABLE", Error, "RedisError"],

src/runtime/api/BunObject.rs

@@ -113,7 +113,6 @@ use crate::api::{HashObject, JSON5Object, TOMLObject, UnsafeObject, YAMLObject};
 use crate::crypto as Crypto;
 use crate::node;
 use crate::test_runner::jest::Jest;
-use crate::valkey_jsc::js_valkey::SubscriptionCtx;
 use bun_core::zig_string::Slice as ZigStringSlice;
 use bun_jsc::ZigStringJsc as _; // to_error_instance / to_type_error_instance
 use bun_jsc::call_frame::ArgumentsSlice;
@@ -2025,21 +2024,7 @@ pub(crate) fn get_valkey_default_client(global_this: &JSGlobalObject, _: &JSObje
     };
 
     let as_js = JSValkeyClient::ptr_to_js(valkey, global_this);
-
-    // SAFETY: `valkey` is a fresh heap allocation owned by the JS wrapper; we
-    // hold the only reference for field init below.
-    let valkey_ref = unsafe { &*valkey };
-    valkey_ref.this_value.set(jsc::JsRef::init_weak(as_js));
-    match SubscriptionCtx::init(valkey_ref) {
-        Ok(ctx) => valkey_ref._subscription_ctx.set(ctx),
-        Err(jsc::JsError::Thrown) | Err(jsc::JsError::Terminated) => return JSValue::ZERO,
-        Err(err) => {
-            let _ =
-                global_this.throw_error(crate::Error::from(err), "Failed to create Redis client");
-            return JSValue::ZERO;
-        }
-    }
-
+    JSValkeyClient::bind_js(valkey, as_js);
     as_js
 }
 

src/runtime/valkey_jsc/ValkeyCommand.rs → src/runtime/valkey_jsc/command.rs

@@ -1,13 +1,14 @@
 use bun_collections::linear_fifo::{DynamicBuffer, LinearFifo};
-use bun_jsc::{self as jsc, JSGlobalObject, JSValue, JsResult};
+use bun_jsc::{self as jsc, JSGlobalObject, JSValue};
 use bun_valkey::valkey_protocol as protocol;
+use bun_valkey::valkey_protocol::RedisError;
 
 use super::protocol_jsc::{ToJSOptions, resp_value_to_js_with_options};
 
 type Slice = bun_core::ZigStringSlice;
 
 // Note: callers in `js_valkey_functions.rs` construct
-// `Vec<crate::node::types::BlobOrStringOrBuffer>` directly, so `Args::Args` must accept
+// `Vec<crate::node::types::BlobOrStringOrBuffer>` directly, so `Args::Blobs` must accept
 // that exact type. The upstream `bun_jsc::Node::BlobOrStringOrBuffer` re-export is a
 // stub; use the real in-crate definition (which already provides `slice()` /
 // `byte_length()`).
@@ -35,7 +36,7 @@ impl<'a> Default for Command<'a> {
 #[derive(Copy, Clone)]
 pub enum Args<'a> {
     Slices(&'a [Slice]),
-    Args(&'a [BlobOrStringOrBuffer]),
+    Blobs(&'a [BlobOrStringOrBuffer]),
     Raw(&'a [&'a [u8]]),
 }
 
@@ -49,45 +50,49 @@ impl<'a> Args<'a> {
     pub(crate) fn len(&self) -> usize {
         match self {
             Args::Slices(args) => args.len(),
-            Args::Args(args) => args.len(),
+            Args::Blobs(args) => args.len(),
             Args::Raw(args) => args.len(),
         }
     }
 }
 
 impl<'a> Command<'a> {
-    pub fn write(&self, writer: &mut impl bun_io::Write) -> Result<(), crate::Error> {
-        // Serialize as RESP array format directly
-        write!(writer, "*{}\r\n", 1 + self.args.len())?;
-        write!(writer, "${}\r\n", self.command.len())?;
-        writer.write_all(self.command)?;
-        writer.write_all(b"\r\n")?;
-
-        match &self.args {
-            Args::Slices(args) => {
-                for arg in args.iter() {
-                    let bytes = arg.slice();
-                    write!(writer, "${}\r\n", bytes.len())?;
-                    writer.write_all(bytes)?;
-                    writer.write_all(b"\r\n")?;
+    pub fn write(&self, writer: &mut impl bun_io::Write) -> Result<(), RedisError> {
+        // Serialize as RESP array format directly; `bun_io::Write` can only
+        // fail with an allocator error, so collapse to `OutOfMemory`.
+        (|| -> bun_io::Result<()> {
+            write!(writer, "*{}\r\n", 1 + self.args.len())?;
+            write!(writer, "${}\r\n", self.command.len())?;
+            writer.write_all(self.command)?;
+            writer.write_all(b"\r\n")?;
+
+            match &self.args {
+                Args::Slices(args) => {
+                    for arg in args.iter() {
+                        let bytes = arg.slice();
+                        write!(writer, "${}\r\n", bytes.len())?;
+                        writer.write_all(bytes)?;
+                        writer.write_all(b"\r\n")?;
+                    }
                 }
-            }
-            Args::Args(args) => {
-                for arg in args.iter() {
-                    write!(writer, "${}\r\n", arg.byte_length())?;
-                    writer.write_all(arg.slice())?;
-                    writer.write_all(b"\r\n")?;
+                Args::Blobs(args) => {
+                    for arg in args.iter() {
+                        write!(writer, "${}\r\n", arg.byte_length())?;
+                        writer.write_all(arg.slice())?;
+                        writer.write_all(b"\r\n")?;
+                    }
                 }
-            }
-            Args::Raw(args) => {
-                for arg in args.iter() {
-                    write!(writer, "${}\r\n", arg.len())?;
-                    writer.write_all(arg)?;
-                    writer.write_all(b"\r\n")?;
+                Args::Raw(args) => {
+                    for arg in args.iter() {
+                        write!(writer, "${}\r\n", arg.len())?;
+                        writer.write_all(arg)?;
+                        writer.write_all(b"\r\n")?;
+                    }
                 }
             }
-        }
-        Ok(())
+            Ok(())
+        })()
+        .map_err(|_| RedisError::OutOfMemory)
     }
 
     pub fn byte_length(&self) -> usize {
@@ -97,7 +102,7 @@ impl<'a> Command<'a> {
         counter.count
     }
 
-    pub fn serialize(&self) -> Result<Box<[u8]>, crate::Error> {
+    pub fn serialize(&self) -> Result<Box<[u8]>, RedisError> {
         let mut buf: Vec<u8> = Vec::with_capacity(self.byte_length());
         self.write(&mut buf)?;
         Ok(buf.into_boxed_slice())
@@ -107,24 +112,16 @@ impl<'a> Command<'a> {
 /// Command stored in offline queue when disconnected
 pub struct Entry {
     pub serialized_data: Box<[u8]>, // Pre-serialized RESP protocol bytes
-    pub meta: Meta,
     pub promise: Promise,
 }
 
-// Inherent associated
-// types are unstable on stable Rust, so expose as a sibling module alias instead.
-pub mod entry {
-    pub(crate) type Queue = super::LinearFifo<super::Entry, super::DynamicBuffer<super::Entry>>;
-}
+pub(crate) type EntryQueue = LinearFifo<Entry, DynamicBuffer<Entry>>;
 
 impl Entry {
     // Create an Offline by serializing the Valkey command directly
-    pub fn create(command: &Command<'_>, promise: Promise) -> Result<Entry, crate::Error> {
+    pub fn create(command: &Command<'_>, promise: Promise) -> Result<Entry, RedisError> {
         Ok(Entry {
             serialized_data: command.serialize()?,
-            // We should be calling .check against command here but due
-            // to a hack introduced to let SUBSCRIBE work, we are not doing that for now.
-            meta: command.meta,
             promise,
         })
     }
@@ -166,18 +163,49 @@ bun_core::comptime_string_set! {
         b"PIPELINE",
         b"SUBSCRIBE",
         b"PSUBSCRIBE",
+        b"SSUBSCRIBE",
         b"UNSUBSCRIBE",
-        b"UNPSUBSCRIBE",
+        b"PUNSUBSCRIBE",
+        b"SUNSUBSCRIBE",
+    };
+}
+
+bun_core::comptime_string_set! {
+    /// Commands whose reply is a subscribe/unsubscribe ack push rather than a
+    /// normal in-line reply; their promise must carry SUBSCRIPTION_REQUEST so
+    /// the push handler knows which in-flight entry to consume.
+    static SUBSCRIPTION_COMMANDS = {
+        b"SUBSCRIBE",
+        b"PSUBSCRIBE",
+        b"SSUBSCRIBE",
+        b"UNSUBSCRIBE",
+        b"PUNSUBSCRIBE",
+        b"SUNSUBSCRIBE",
     };
 }
 
 impl Meta {
-    pub fn check(self, command: &Command<'_>) -> Self {
+    pub fn check(self, command_name: &[u8]) -> Self {
         let mut new = self;
+        // Case-insensitive probe: all disallowed entries are ≤12 bytes, so any
+        // name longer than our 32-byte scratch cannot match and can skip the copy.
+        let mut upper = [0u8; 32];
+        let n = command_name.len().min(32);
+        for i in 0..n {
+            upper[i] = command_name[i].to_ascii_uppercase();
+        }
+        let probe: &[u8] = if command_name.len() <= 32 {
+            &upper[..n]
+        } else {
+            command_name
+        };
         new.set(
             Meta::SUPPORTS_AUTO_PIPELINING,
-            !AUTO_PIPELINE_DISALLOWED_COMMANDS.contains(command.command),
+            !AUTO_PIPELINE_DISALLOWED_COMMANDS.contains(probe),
         );
+        if SUBSCRIPTION_COMMANDS.contains(probe) {
+            new.insert(Meta::SUBSCRIPTION_REQUEST);
+        }
         new
     }
 }
@@ -188,6 +216,8 @@ pub struct Promise {
     pub promise: jsc::JSPromiseStrong,
 }
 
+pub(crate) type PromiseQueue = LinearFifo<Promise, DynamicBuffer<Promise>>;
+
 impl Promise {
     pub fn create(global_object: &JSGlobalObject, meta: Meta) -> Promise {
         let promise = jsc::JSPromiseStrong::init(global_object);
@@ -197,16 +227,21 @@ impl Promise {
     pub fn resolve(
         &mut self,
         global_object: &JSGlobalObject,
-        value: &mut protocol::RESPValue,
+        mut value: protocol::RESPValue,
     ) -> Result<(), jsc::JsTerminated> {
+        if self.meta.contains(Meta::RETURN_AS_BOOL) {
+            if let protocol::RESPValue::Integer(int_value) = value {
+                value = protocol::RESPValue::Boolean(int_value > 0);
+            }
+        }
         let options = ToJSOptions {
             return_as_buffer: self.meta.contains(Meta::RETURN_AS_BUFFER),
         };
 
         let js_value = match resp_value_to_js_with_options(value, global_object, options) {
             Ok(v) => v,
             Err(err) => {
-                self.reject(global_object, Ok(global_object.take_error(err)))?;
+                self.promise.reject(global_object, Err(err))?;
                 return Ok(());
             }
         };
@@ -217,32 +252,9 @@ impl Promise {
     pub fn reject(
         &mut self,
         global_object: &JSGlobalObject,
-        jsvalue: JsResult<JSValue>,
-    ) -> Result<(), jsc::JsTerminated> {
-        self.promise.reject(global_object, jsvalue)?;
-        Ok(())
-    }
-}
-
-// Command+Promise pair for tracking which command corresponds to which promise
-pub struct PromisePair {
-    pub meta: Meta,
-    pub promise: Promise,
-}
-
-// See `entry` note above.
-pub mod promise_pair {
-    pub(crate) type Queue =
-        super::LinearFifo<super::PromisePair, super::DynamicBuffer<super::PromisePair>>;
-}
-
-impl PromisePair {
-    pub fn reject_command(
-        &mut self,
-        global_object: &JSGlobalObject,
-        jsvalue: JSValue,
+        value: JSValue,
     ) -> Result<(), jsc::JsTerminated> {
-        self.promise.reject(global_object, Ok(jsvalue))?;
+        self.promise.reject(global_object, Ok(value))?;
         Ok(())
     }
 }