redis: re-issue SUBSCRIBE after an auto-reconnect

docs/runtime/redis.mdx

@@ -417,6 +417,9 @@ When a connection is lost, the client automatically attempts to reconnect with e
 4. Commands executed during disconnection are:
    - Queued if `enableOfflineQueue` is true (default)
    - Rejected immediately if `enableOfflineQueue` is false
+5. Channels registered with `.subscribe()` are re-subscribed as soon as the new
+   connection finishes its handshake, so listeners keep receiving messages.
+   Patterns sent with `psubscribe` are not replayed
 
 ---
 

src/jsc/JSMap.rs

@@ -58,6 +58,12 @@ impl JSMap {
         crate::cpp::JSC__JSMap__size(self, global)
     }
 
+    /// Retrieve this JS Map object's keys as a JS `Array`, in insertion order.
+    #[track_caller]
+    pub fn keys(&mut self, global: &JSGlobalObject) -> JsResult<JSValue> {
+        crate::cpp::JSC__JSMap__keys(self, global)
+    }
+
     /// Attempt to convert a `JSValue` to a `*JSMap`.
     ///
     /// Returns `None` if the value is not a Map.

src/jsc/bindings/bindings.cpp

@@ -5943,6 +5943,28 @@ CPP_DECL [[ZIG_EXPORT(check_slow)]] uint32_t JSC__JSMap__size(JSC::JSMap* map, J
     return map->size();
 }
 
+// Returns the map's keys as a JSArray, in insertion order.
+CPP_DECL [[ZIG_EXPORT(zero_is_throw)]] JSC::EncodedJSValue JSC__JSMap__keys(JSC::JSMap* map, JSC::JSGlobalObject* globalObject)
+{
+    auto& vm = JSC::getVM(globalObject);
+    auto scope = DECLARE_THROW_SCOPE(vm);
+
+    JSC::JSArray* keys = JSC::constructEmptyArray(globalObject, nullptr, map->size());
+    RETURN_IF_EXCEPTION(scope, {});
+
+    auto* iterator = JSC::JSMapIterator::create(vm, globalObject->mapIteratorStructure(), map, JSC::IterationKind::Keys);
+    RETURN_IF_EXCEPTION(scope, {});
+
+    unsigned index = 0;
+    JSC::JSValue key;
+    while (iterator->next(globalObject, key)) {
+        keys->putDirectIndex(globalObject, index++, key);
+        RETURN_IF_EXCEPTION(scope, {});
+    }
+
+    return JSC::JSValue::encode(keys);
+}
+
 CPP_DECL void JSC__VM__setControlFlowProfiler(JSC::VM* vm, bool isEnabled)
 {
     if (isEnabled) {

src/runtime/valkey_jsc/js_valkey.rs

@@ -142,6 +142,11 @@ impl SubscriptionCtx {
         Ok(count)
     }
 
+    /// The channels this context has listeners for, as a JS array of strings.
+    pub fn subscribed_channels(&self, global_object: &JSGlobalObject) -> JsResult<JSValue> {
+        self.subscription_callback_map().keys(global_object)
+    }
+
     /// Test whether this context has any subscriptions. It is mandatory to
     /// guard deinit with this function.
     pub fn has_subscriptions(&self, global_object: &JSGlobalObject) -> JsResult<bool> {
@@ -722,6 +727,7 @@ impl JSValkeyClient {
                 read_buffer: Default::default(),
                 reply_scanner: Default::default(),
                 retry_attempts: 0,
+                resubscribe_pending: 0,
                 auto_flusher: Default::default(),
             }),
             global_object,
@@ -846,6 +852,7 @@ impl JSValkeyClient {
                 read_buffer: Default::default(),
                 reply_scanner: Default::default(),
                 retry_attempts: 0,
+                resubscribe_pending: 0,
                 auto_flusher: Default::default(),
             }),
             global_object,
@@ -1301,6 +1308,58 @@ impl JSValkeyClient {
                 self.client_mut().flags.connection_promise_returns_client = false;
             }
         }
+
+        // Last, so a failure here cannot strand `.connect()`. Still ahead of the
+        // offline queue: nothing drains it until this scope's microtask drain and
+        // `on_writable` run, both of which happen after this returns.
+        self.resubscribe()
+    }
+
+    /// Replay the subscription set onto a freshly (re)connected socket.
+    ///
+    /// The server forgets every subscription when the connection drops, but the
+    /// JS-side callback map survives it, so without this a reconnected
+    /// subscriber reports `connected` while silently receiving nothing.
+    ///
+    /// The command goes straight into the write buffer (ahead of anything the
+    /// offline queue still holds) and carries no promise, so the confirmations
+    /// it produces are counted off by `resubscribe_pending` rather than
+    /// consuming another command's promise.
+    fn resubscribe(&self) -> JsTerminatedResult<()> {
+        if !self.is_subscriber() || self.this_value.get().try_get().is_none() {
+            return Ok(());
+        }
+
+        let global_object = self.global_object;
+        let channels = self
+            ._subscription_ctx
+            .get()
+            .subscribed_channels(&global_object)?;
+        let mut iter = channels.array_iterator(&global_object)?;
+        let mut args: Vec<bun_core::ZigStringSlice> = Vec::with_capacity(iter.len as usize);
+        while let Some(channel) = iter.next()? {
+            let channel_str = channel.to_bun_string(&global_object)?;
+            args.push(channel_str.to_utf8());
+            channel_str.deref();
+        }
+        if args.is_empty() {
+            return Ok(());
+        }
+
+        debug!("Replaying SUBSCRIBE for {} channel(s)", args.len());
+        let subscribe_cmd = Command {
+            command: b"SUBSCRIBE",
+            args: command::Args::Slices(&args),
+            meta: command::Meta::default(),
+        };
+        let client = self.client_mut();
+        if subscribe_cmd.write(client.writer()).is_err() {
+            return narrow_terminated(client.fail(
+                b"Failed to write SUBSCRIBE command",
+                protocol::RedisError::OutOfMemory,
+            ));
+        }
+        client.resubscribe_pending = u32::try_from(args.len()).expect("int cast");
         Ok(())
     }
 

src/runtime/valkey_jsc/valkey.rs

@@ -264,6 +264,10 @@ pub struct ValkeyClient {
     /// Commands that are waiting to be sent to the server. When pipelining is implemented, this usually will be empty.
     pub queue: command::entry::Queue,
 
+    /// Subscribe confirmations still owed by the promise-less SUBSCRIBE that
+    /// `JSValkeyClient::resubscribe` replays after a reconnect. One per channel.
+    pub resubscribe_pending: u32,
+
     // Connection parameters
     // `connection_strings` is retained because `js_valkey.rs` still slices it
     // when constructing/duplicating clients.
@@ -1099,13 +1103,30 @@ impl ValkeyClient {
         let mut should_consume_promise_pair = true;
         let mut pair_maybe: Option<command::PromisePair> = None;
 
+        if self.resubscribe_pending > 0 && matches!(value, RESPValue::Error(_)) {
+            // The replayed SUBSCRIBE was refused (an ACL change during the outage,
+            // say). It owns no promise pair, so this error must not eat one, and no
+            // confirmation is coming for any of its channels.
+            self.resubscribe_pending = 0;
+            should_consume_promise_pair = false;
+        }
+
         // For subscription clients, check if this is a push message that doesn't need a promise pair
         if let RESPValue::Push(push) = value {
             match protocol::SubscriptionPushMessage::from_bytes(&push.kind) {
                 Some(protocol::SubscriptionPushMessage::Message) => {
                     // Message pushes never need promise pairs
                     should_consume_promise_pair = false;
                 }
+                Some(protocol::SubscriptionPushMessage::Subscribe)
+                    if self.resubscribe_pending > 0 =>
+                {
+                    // Confirmation of the SUBSCRIBE replayed on reconnect. It was
+                    // written without a promise, and its confirmations are the first
+                    // replies on the new connection, so no pair belongs to it.
+                    self.resubscribe_pending -= 1;
+                    should_consume_promise_pair = false;
+                }
                 Some(
                     protocol::SubscriptionPushMessage::Subscribe
                     | protocol::SubscriptionPushMessage::Unsubscribe,
@@ -1282,6 +1303,7 @@ impl ValkeyClient {
         self.flags.failed = false;
         self.flags.is_authenticated = false;
         self.flags.is_selecting_db_internal = false;
+        self.resubscribe_pending = 0;
         if matches!(self.socket, AnySocket::SocketTcp(_)) {
             // if is tcp, we need to start the connection process
             // if is tls, we need to wait for the handshake to complete