redis: give SubscriptionCtx an explicit BackRef to fix the subscribe() refcount over-release

src/bun_core/lib.rs

@@ -723,7 +723,7 @@ macro_rules! from_field_ptr {
 /// bun_core::impl_field_parent! { Assets => DevServer.assets; pub fn owner; fn owner_mut; }
 ///
 /// // (2) ref-only                   (&self -> &P)
-/// bun_core::impl_field_parent! { SubscriptionCtx => JSValkeyClient._subscription_ctx; fn parent; }
+/// bun_core::impl_field_parent! { ValkeyClient => JSValkeyClient.client; fn parent; }
 ///
 /// // (3) mut-only                   (&mut self -> *mut P)
 /// bun_core::impl_field_parent! { DirectoryWatchStore => DevServer.directory_watchers; fn mut owner; }
@@ -745,6 +745,10 @@ macro_rules! from_field_ptr {
 /// `$Parent.$field` for its entire lifetime. If `$Child` can exist
 /// standalone, the generated accessors are unsound; keep a hand-rolled
 /// `pub unsafe fn` instead.
+///
+/// The `&self` forms carry `&$Child`'s provenance (`noalias readonly` when
+/// `$Child: Freeze`), so treat the returned `&$Parent` as read-only; store a
+/// `BackRef<$Parent>` on the child if it must write.
 #[macro_export]
 macro_rules! impl_field_parent {
     // ref + raw-mut pair

src/ptr/ref_count.rs

@@ -278,7 +278,17 @@ impl<T: RefCounted> RefCount<T> {
             dump_stack_hook(None, return_address());
         }
         count.assert_single_threaded();
-        count.raw_count.set(count.raw_count.get() + 1);
+        let next = count.raw_count.get() + 1;
+        count.raw_count.set(next);
+        // Volatile re-read catches an O2+ dead-store when `self_` was derived
+        // from a readonly-provenance `&T` (container_of on a Freeze child).
+        #[cfg(debug_assertions)]
+        debug_assert_eq!(
+            // SAFETY: `count` is live per the caller contract.
+            unsafe { core::ptr::read_volatile(count.raw_count.as_ptr()) },
+            next,
+            "RefCount::ref_ store elided (pointer derived from readonly provenance?)",
+        );
     }
 
     /// # Safety

src/runtime/api/BunObject.rs

@@ -1958,7 +1958,9 @@ pub(crate) fn get_valkey_default_client(global_this: &JSGlobalObject, _: &JSObje
     // 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) {
+    // SAFETY: `valkey` is the fresh `heap::into_raw` pointer from
+    // `create_no_js_no_pubsub` above.
+    match unsafe { SubscriptionCtx::init(valkey) } {
         Ok(ctx) => valkey_ref._subscription_ctx.set(ctx),
         Err(jsc::JsError::Thrown) | Err(jsc::JsError::Terminated) => return JSValue::ZERO,
         Err(err) => {

src/runtime/valkey_jsc/js_valkey.rs

@@ -66,6 +66,10 @@ type Socket = uws::AnySocket;
 
 #[derive(Default)]
 pub struct SubscriptionCtx {
+    /// Stored (not `container_of`-recovered) so writes to the parent carry the
+    /// Box's provenance rather than `&self`'s `noalias readonly`. `None` only
+    /// until [`SubscriptionCtx::init`].
+    parent: Option<BackRef<JSValkeyClient>>,
     pub is_subscriber: bool,
     pub original_enable_offline_queue: bool,
     pub original_enable_auto_pipelining: bool,
@@ -76,36 +80,40 @@ pub struct SubscriptionCtx {
 /// free-fns plus `to_js`/`from_js`. Re-exported here as `Js`.
 pub use crate::generated_classes::js_RedisClient as Js;
 
-// SAFETY: `SubscriptionCtx` lives at `JSValkeyClient._subscription_ctx`
-// (intrusive backref). `JsCell<SubscriptionCtx>` is `#[repr(transparent)]`.
-bun_core::impl_field_parent! { SubscriptionCtx => JSValkeyClient._subscription_ctx; fn parent; }
-
 impl SubscriptionCtx {
-    pub fn init(valkey_parent: &JSValkeyClient) -> JsResult<Self> {
-        let callback_map = JSMap::create(&valkey_parent.global_object);
-        let parent_this = valkey_parent
-            .this_value
-            .get()
-            .try_get()
-            .expect("unreachable");
+    /// # Safety
+    /// `valkey_parent` must be the live `heap::into_raw` pointer for the
+    /// owning `JSValkeyClient` (full-allocation provenance), not a pointer
+    /// derived from a narrower borrow, and must outlive the returned context.
+    pub unsafe fn init(valkey_parent: *mut JSValkeyClient) -> JsResult<Self> {
+        // SAFETY: caller contract.
+        let parent = unsafe { BackRef::from_raw(valkey_parent) };
+        let callback_map = JSMap::create(&parent.global_object);
+        let parent_this = parent.this_value.get().try_get().expect("unreachable");
 
-        Js::subscription_callback_map_set_cached(
-            parent_this,
-            &valkey_parent.global_object,
-            callback_map,
-        );
+        Js::subscription_callback_map_set_cached(parent_this, &parent.global_object, callback_map);
 
         Ok(SubscriptionCtx {
-            original_enable_offline_queue: valkey_parent.client.get().flags.enable_offline_queue,
-            original_enable_auto_pipelining: valkey_parent
-                .client
-                .get()
-                .flags
-                .enable_auto_pipelining,
+            parent: Some(parent),
+            original_enable_offline_queue: parent.client.get().flags.enable_offline_queue,
+            original_enable_auto_pipelining: parent.client.get().flags.enable_auto_pipelining,
             is_subscriber: false,
         })
     }
 
+    #[inline]
+    fn parent(&self) -> &JSValkeyClient {
+        self.parent
+            .as_ref()
+            .expect("SubscriptionCtx used before init()")
+            .get()
+    }
+
+    #[inline]
+    fn parent_backref(&self) -> BackRef<JSValkeyClient> {
+        self.parent.expect("SubscriptionCtx used before init()")
+    }
+
     fn subscription_callback_map(&self) -> &mut JSMap {
         let parent_this = self
             .parent()
@@ -202,10 +210,7 @@ impl SubscriptionCtx {
         channel_name: JSValue,
         callback: JSValue,
     ) -> JsResult<()> {
-        // `BackRef` (Copy + Deref) detaches the borrow so the guard closure is
-        // safe even though intervening JS may re-enter `&self`.
-        let parent_br = BackRef::new(self.parent());
-        let _guard = scopeguard::guard(parent_br, |p| {
+        let _guard = scopeguard::guard(self.parent_backref(), |p| {
             p.on_new_subscription_callback_insert();
         });
         let map = self.subscription_callback_map();
@@ -283,10 +288,7 @@ impl SubscriptionCtx {
 
         // After we go through every single callback, we will have to update the poll ref.
         // The user may, for example, unsubscribe in the callbacks, or even stop the client.
-        // `BackRef` (Copy + Deref) detaches the borrow so the guard closure is
-        // safe even though intervening JS may re-enter `&self`.
-        let parent_br = BackRef::new(self.parent());
-        let _update = scopeguard::guard(parent_br, |p| p.update_poll_ref());
+        let _update = scopeguard::guard(self.parent_backref(), |p| p.update_poll_ref());
 
         // If callbacks is an array, iterate and call each one
         let mut iter = callbacks.array_iterator(global_object)?;
@@ -833,9 +835,11 @@ impl JSValkeyClient {
         new_client.this_value.set(JsRef::init_weak(js_this));
 
         // Need to associate the subscription context, after the JS ref has been populated.
+        // SAFETY: `new_client_ptr` is the fresh `heap::into_raw` pointer from
+        // `create_no_js_no_pubsub` above.
         new_client
             ._subscription_ctx
-            .set(SubscriptionCtx::init(new_client)?);
+            .set(unsafe { SubscriptionCtx::init(new_client_ptr) }?);
 
         Ok(new_client_ptr)
     }
@@ -1437,9 +1441,8 @@ impl JSValkeyClient {
         self.ref_();
         // socket close can potentially call JS so we need to enqueue the deinit
         struct Holder {
-            // BACKREF — JSValkeyClient is intrusively ref-counted (RefCount + @fieldParentPtr
-            // recovery in SubscriptionCtx::parent). The `self.ref_()` above / `(*ctx).deref()`
-            // in run() keep it alive across the task hop.
+            // The `self.ref_()` above / `(*ctx).deref()` in run() keep the
+            // intrusively ref-counted client alive across the task hop.
             ctx: *const JSValkeyClient,
             task: jsc::AnyTask::AnyTask,
         }
@@ -1490,11 +1493,11 @@ impl JSValkeyClient {
         this.this_value.with_mut(|t| t.finalize());
         this.client_mut().flags.finalized = true;
         this.close_socket_next_tick();
-        // `_subscription_ctx` is three inline bools (no allocation, no GC
-        // ref); `is_subscriber` can legitimately still be set here if the
-        // server never confirmed UNSUBSCRIBE before disconnect, since
-        // `update_poll_ref()` gates on the JS handler map, not this flag.
-        // Nothing to release.
+        // `_subscription_ctx` owns no allocation and no GC ref (the backref is
+        // non-owning, the rest are inline bools); `is_subscriber` can
+        // legitimately still be set here if the server never confirmed
+        // UNSUBSCRIBE before disconnect, since `update_poll_ref()` gates on
+        // the JS handler map, not this flag. Nothing to release.
     }
 
     pub fn stop_timers(&self) {

src/runtime/valkey_jsc/js_valkey_functions.rs

@@ -1910,9 +1910,11 @@ impl JSValkeyClient {
 
         let new_client_js = JSValkeyClient::ptr_to_js(new_client_ptr, global);
         new_client.this_value.set(JsRef::init_weak(new_client_js));
+        // SAFETY: `new_client_ptr` is the fresh `heap::into_raw` pointer from
+        // `clone_without_connecting` above.
         new_client
             ._subscription_ctx
-            .set(SubscriptionCtx::init(new_client)?);
+            .set(unsafe { SubscriptionCtx::init(new_client_ptr) }?);
         // If the original client is already connected and not manually closed, start connecting the new client.
         if this.client.get().status == valkey::Status::Connected
             && !this.client.get().flags.is_manually_closed

src/runtime/webcore/ByteBlobLoader.rs

@@ -166,7 +166,10 @@ impl ByteBlobLoader {
             blob.content_type.set(ct);
         }
 
-        self.parent_const().is_closed.set(true);
+        // SAFETY: `impl_field_parent!` contract; `is_closed` is disjoint from
+        // `context`. Reached via the `&mut self` accessor (not `parent_const`)
+        // so the store is not routed through a Freeze-`&self` readonly pointer.
+        unsafe { (*self.parent()).is_closed.set(true) };
         Some(blob::Any::Blob(blob))
     }
 

src/runtime/webcore/ReadableStream.rs

@@ -665,8 +665,9 @@ pub struct NewSource<C: SourceContext> {
     /// `Finalized` so [`Self::on_js_close`] reads `None` instead of a
     /// dead-but-unswept cell.
     pub this_jsvalue: jsc::JsRef,
-    /// R-2: written by `&self` context methods (`ByteStream::to_any_blob`,
-    /// `ByteBlobLoader::to_any_blob`) via `parent_const()`, so interior-mutable.
+    /// R-2: written by context methods (`ByteStream::to_any_blob`,
+    /// `ByteBlobLoader::to_any_blob`) via `impl_field_parent!` recovery, so
+    /// interior-mutable.
     pub is_closed: Cell<bool>,
 }
 

test/js/valkey/valkey-gc.test.ts

@@ -127,6 +127,93 @@ test.concurrent(
 // on_writable/update_poll_ref before send() takes its own ref, so a
 // connect/close fault path inside could free the client under the live
 // `&self`. This variant races a server-side RST against subscribe()+close().
+// subscribe() used to release one more intrusive ref than it took:
+// SubscriptionCtx.parent() recovered &JSValkeyClient via container_of on
+// &SubscriptionCtx (three plain bools, a Freeze type). The shared-ref argument
+// is lowered as noalias readonly, so at O2+ LLVM may drop the refcount .set()
+// reached through it while the paired ScopedRef drop (an opaque call)
+// survives. After close() releases the socket ref, only the connection-timeout
+// timer holds the allocation; when it fires the count reaches zero and deinit
+// frees the Box under the still-live JS wrapper. The next GC finalize (or any
+// property read) is a heap-use-after-free.
+//
+// Debug cargo builds are opt-level=0 and do not perform this elision, so this
+// test exercises the fault only on optimized ASAN builds. It still asserts the
+// healthy-server subscribe/close/GC path stays balanced everywhere.
+test.concurrent(
+  "RedisClient survives subscribe() + close() against a healthy server across connection-timeout + GC",
+  async () => {
+    const src = `
+    const CRLF = "\\r\\n";
+    const blk = s => "$" + s.length + CRLF + s + CRLF;
+    const server = Bun.listen({
+      hostname: "127.0.0.1",
+      port: 0,
+      socket: {
+        open(s) { s.data = { buf: "" }; },
+        data(s, d) {
+          s.data.buf += d.toString("latin1");
+          if (s.data.buf.includes("HELLO")) {
+            s.write("%1" + CRLF + blk("proto") + ":3" + CRLF);
+            s.data.buf = "";
+          }
+        },
+        close() {},
+      },
+    });
+    const url = "redis://127.0.0.1:" + server.port;
+    const timeoutMs = 1000;
+    const clients = [];
+    for (let i = 0; i < 40; i++) {
+      const c = new Bun.RedisClient(url, {
+        connectionTimeout: timeoutMs,
+        idleTimeout: 0,
+        autoReconnect: false,
+      });
+      c.onconnect = () => {}; c.onclose = () => {};
+      await c.connect();
+      // upsert_receive_handler runs synchronously here; the server never
+      // replies to SUBSCRIBE so the returned promise is discarded.
+      try { c.subscribe("ch" + i, () => {}).catch(() => {}); } catch {}
+      c.close();
+      clients.push(c);
+    }
+    // Let every client's connection-timeout timer (armed at connect()) fire.
+    // This is the condition under test, not an arbitrary wait.
+    const { promise, resolve } = Promise.withResolvers();
+    setTimeout(resolve, timeoutMs + 200);
+    await promise;
+    Bun.gc(true);
+    for (const c of clients) {
+      // Any native read on the wrapper is a heap-UAF once the backing Box is
+      // freed under it.
+      if (typeof c.bufferedAmount !== "number") throw new Error("bufferedAmount");
+      if (c.connected !== false) throw new Error("connected");
+    }
+    clients.length = 0;
+    server.stop(true);
+    Bun.gc(true);
+    await 1;
+    Bun.gc(true);
+    console.log("OK");
+    process.exit(0);
+  `;
+
+    await using proc = Bun.spawn({
+      cmd: [bunExe(), "-e", src],
+      env: bunEnv,
+      stdout: "pipe",
+      stderr: "inherit",
+    });
+
+    const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]);
+
+    expect(stdout.trim()).toBe("OK");
+    expect(proc.signalCode).toBeNull();
+    expect(exitCode).toBe(0);
+  },
+);
+
 test.concurrent("RedisClient survives subscribe() + close() against a server that resets the connection", async () => {
   const src = `
     const CRLF = "\\r\\n";