From 426f68b5db379849733a92b04ebbb6eca245818e Mon Sep 17 00:00:00 2001 From: Alistair Smith Date: Tue, 18 Aug 2026 11:40:20 -0700 Subject: [PATCH 01/11] valkey: use vecdeque for the command queues --- src/runtime/valkey_jsc/ValkeyCommand.rs | 6 +- src/runtime/valkey_jsc/js_valkey.rs | 14 +- src/runtime/valkey_jsc/valkey.rs | 130 ++++++++---------- .../reliability/connection-failures.test.ts | 77 +++++++++++ 4 files changed, 142 insertions(+), 85 deletions(-) diff --git a/src/runtime/valkey_jsc/ValkeyCommand.rs b/src/runtime/valkey_jsc/ValkeyCommand.rs index abcc35864f62..06430d9bf397 100644 --- a/src/runtime/valkey_jsc/ValkeyCommand.rs +++ b/src/runtime/valkey_jsc/ValkeyCommand.rs @@ -1,4 +1,3 @@ -use bun_collections::linear_fifo::{DynamicBuffer, LinearFifo}; use bun_jsc::{self as jsc, JSGlobalObject, JSValue, JsResult}; use bun_valkey::valkey_protocol as protocol; @@ -114,7 +113,7 @@ pub struct Entry { // 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>; + pub(crate) type Queue = std::collections::VecDeque; } impl Entry { @@ -248,8 +247,7 @@ pub struct PromisePair { // See `entry` note above. pub mod promise_pair { - pub(crate) type Queue = - super::LinearFifo>; + pub(crate) type Queue = std::collections::VecDeque; } impl PromisePair { diff --git a/src/runtime/valkey_jsc/js_valkey.rs b/src/runtime/valkey_jsc/js_valkey.rs index d5f4daa7ac0f..4493cf599c90 100644 --- a/src/runtime/valkey_jsc/js_valkey.rs +++ b/src/runtime/valkey_jsc/js_valkey.rs @@ -758,8 +758,8 @@ impl JSValkeyClient { protocol: uri, username, password, - in_flight: command::promise_pair::Queue::init(), - queue: command::entry::Queue::init(), + in_flight: command::promise_pair::Queue::new(), + queue: command::entry::Queue::new(), status: valkey::Status::NeverConnected, connection_strings, socket: Socket::SocketTcp(uws::SocketTCP { @@ -867,8 +867,8 @@ impl JSValkeyClient { protocol: client.protocol, username, password, - in_flight: command::promise_pair::Queue::init(), - queue: command::entry::Queue::init(), + in_flight: command::promise_pair::Queue::new(), + queue: command::entry::Queue::new(), status: valkey::Status::NeverConnected, connection_strings: connection_strings_copy, socket: Socket::SocketTcp(uws::SocketTCP { @@ -1610,13 +1610,13 @@ impl JSValkeyClient { memory_cost += client.read_buffer.byte_list.capacity() as usize; // Add queue sizes - memory_cost += client.in_flight.readable_length() + memory_cost += client.in_flight.len() * core::mem::size_of::(); - for command in client.queue.readable_slice(0) { + for command in client.queue.iter() { memory_cost += command.serialized_data.len(); } memory_cost += - client.queue.readable_length() * core::mem::size_of::(); + client.queue.len() * core::mem::size_of::(); memory_cost } diff --git a/src/runtime/valkey_jsc/valkey.rs b/src/runtime/valkey_jsc/valkey.rs index ff8d11287b11..e18c9fd20963 100644 --- a/src/runtime/valkey_jsc/valkey.rs +++ b/src/runtime/valkey_jsc/valkey.rs @@ -337,9 +337,8 @@ impl ValkeyClient { /// Clean up resources used by the Valkey client // Cannot be `Drop` — takes a JSGlobalObject param and has JS side effects. pub(crate) fn shutdown(&mut self, global_object_or_finalizing: Option<&JSGlobalObject>) { - let mut pending = - core::mem::replace(&mut self.in_flight, command::promise_pair::Queue::init()); - let mut commands = core::mem::replace(&mut self.queue, command::entry::Queue::init()); + let mut pending = core::mem::take(&mut self.in_flight); + let mut commands = core::mem::take(&mut self.queue); if let Some(global_this) = global_object_or_finalizing { let object = valkey_error_to_js( @@ -347,25 +346,25 @@ impl ValkeyClient { b"Connection closed", RedisError::ConnectionClosed, ); - while let Some(mut pair) = pending.read_item() { + while let Some(mut pair) = pending.pop_front() { // Any exception from the reject is swallowed so // every remaining pending command still gets rejected at shutdown. let _ = pair.reject_command(global_this, object); } - while let Some(mut offline_cmd) = commands.read_item() { + while let Some(mut offline_cmd) = commands.pop_front() { // Same as above: swallow reject exceptions so the whole queue drains. let _ = offline_cmd.promise.reject(global_this, Ok(object)); // Note: `offline_cmd.deinit()` — Entry/Box<[u8]> drops automatically. } } else { // finalizing. we can't call into JS. - while let Some(pair) = pending.read_item() { + while let Some(pair) = pending.pop_front() { // Note: `pair.promise.deinit()` — JSPromiseStrong drops automatically. drop(pair); } - while let Some(offline_cmd) = commands.read_item() { + while let Some(offline_cmd) = commands.pop_front() { // Note: `offline_cmd.promise.deinit()` / `offline_cmd.deinit()` — // JSPromiseStrong / Box<[u8]> drop automatically. drop(offline_cmd); @@ -408,11 +407,10 @@ impl ValkeyClient { // Start draining the command queue let mut total_bytelength: usize = 0; - // We compute the count first, then drain by `read_item`. + // We compute the count first, then drain by `pop_front`. let pipelineable_count: usize = { - let to_process = self.queue.readable_slice(0); let mut total: usize = 0; - for command in to_process { + for command in self.queue.iter() { if !command .meta .contains(command::Meta::SUPPORTS_AUTO_PIPELINING) @@ -429,13 +427,11 @@ impl ValkeyClient { .byte_list .ensure_unused_capacity(total_bytelength); for _ in 0..pipelineable_count { - let cmd = self.queue.read_item().expect("count was precomputed"); - self.in_flight - .write_item(command::PromisePair { - meta: cmd.meta, - promise: cmd.promise, - }) - .unwrap_or_oom(); + let cmd = self.queue.pop_front().expect("count was precomputed"); + self.in_flight.push_back(command::PromisePair { + meta: cmd.meta, + promise: cmd.promise, + }); self.write_buffer .write(&cmd.serialized_data) .unwrap_or_oom(); @@ -445,7 +441,7 @@ impl ValkeyClient { let _ = self.flush_data(); - let have_more = self.queue.readable_length() > 0; + let have_more = !self.queue.is_empty(); self.auto_flusher.registered.set(have_more); self.deref(); @@ -467,8 +463,8 @@ impl ValkeyClient { } pub(crate) fn has_any_pending_commands(&self) -> bool { - self.in_flight.readable_length() > 0 - || self.queue.readable_length() > 0 + !self.in_flight.is_empty() + || !self.queue.is_empty() || self.write_buffer.len() > 0 || self.read_buffer.len() > 0 } @@ -505,17 +501,17 @@ impl ValkeyClient { global_this: &JSGlobalObject, jsvalue: JSValue, ) -> JsResult<()> { - let mut pending = core::mem::replace(pending_ptr, command::promise_pair::Queue::init()); - let mut entries = core::mem::replace(entries_ptr, command::entry::Queue::init()); + let mut pending = core::mem::take(pending_ptr); + let mut entries = core::mem::take(entries_ptr); // Note: `defer pending.deinit()` / `defer entries.deinit()` — handled by Drop. // Reject commands in the command queue - while let Some(mut command_pair) = pending.read_item() { + while let Some(mut command_pair) = pending.pop_front() { command_pair.reject_command(global_this, jsvalue)?; } // Reject commands in the offline queue - while let Some(mut cmd) = entries.read_item() { + while let Some(mut cmd) = entries.pop_front() { // Note: `defer cmd.deinit(allocator)` — Entry should impl Drop. cmd.promise.reject(global_this, Ok(jsvalue))?; } @@ -523,7 +519,7 @@ impl ValkeyClient { } fn reject_in_flight_commands(&mut self, message: &[u8], err: RedisError) -> JsResult<()> { - if self.in_flight.readable_length() == 0 { + if self.in_flight.is_empty() { return Ok(()); } @@ -533,11 +529,8 @@ impl ValkeyClient { message: Box::<[u8]>::from(message), err, global_this: GlobalRef::from(vm.global()), - in_flight: core::mem::replace( - &mut self.in_flight, - command::promise_pair::Queue::init(), - ), - queue: command::entry::Queue::init(), + in_flight: core::mem::take(&mut self.in_flight), + queue: command::entry::Queue::new(), }); deferred_failure.enqueue(); return Ok(()); @@ -545,7 +538,7 @@ impl ValkeyClient { let global_this = self.global_object(); let jsvalue = valkey_error_to_js(&global_this, message, err); - let mut entries = command::entry::Queue::init(); + let mut entries = command::entry::Queue::new(); Self::reject_all_pending_commands(&mut self.in_flight, &mut entries, &global_this, jsvalue) } @@ -572,7 +565,7 @@ impl ValkeyClient { if self.flags.finalized { // We can't run promises inside finalizers. - if self.queue.readable_length() + self.in_flight.readable_length() > 0 { + if !self.queue.is_empty() || !self.in_flight.is_empty() { let vm = self.vm; let deferred_failure = Box::new(DeferredFailure { // This memory is not owned by us. @@ -580,11 +573,8 @@ impl ValkeyClient { err, global_this: GlobalRef::from(vm.global()), - in_flight: core::mem::replace( - &mut self.in_flight, - command::promise_pair::Queue::init(), - ), - queue: core::mem::replace(&mut self.queue, command::entry::Queue::init()), + in_flight: core::mem::take(&mut self.in_flight), + queue: core::mem::take(&mut self.queue), }); deferred_failure.enqueue(); } @@ -722,22 +712,21 @@ impl ValkeyClient { pub(crate) fn send_next_command(&mut self) { if self.write_buffer.remaining().is_empty() && self.connection_ready() { - if self.queue.readable_length() > 0 { + if let Some(head) = self.queue.front() { // Check the command at the head of the queue - let flags = self.queue.readable_slice(0)[0].meta; + let flags = head.meta; if !flags.contains(command::Meta::SUPPORTS_AUTO_PIPELINING) { // Head is non-pipelineable. Try to drain it serially if nothing is in-flight. - if self.in_flight.readable_length() == 0 { + if self.in_flight.is_empty() { let _ = self.drain(); // Send the single non-pipelineable command // After draining, check if the *new* head is pipelineable and schedule flush if needed. // This covers sequences like NON_PIPE -> PIPE -> PIPE ... - if self.queue.readable_length() > 0 - && self.queue.readable_slice(0)[0] - .meta + if self.queue.front().is_some_and(|head| { + head.meta .contains(command::Meta::SUPPORTS_AUTO_PIPELINING) - { + }) { self.register_auto_flusher(self.vm); } } else { @@ -747,7 +736,7 @@ impl ValkeyClient { // Head is pipelineable. Register the flusher to batch it with others. self.register_auto_flusher(self.vm); } - } else if self.in_flight.readable_length() == 0 { + } else if self.in_flight.is_empty() { // Without auto pipelining, wait for in-flight to empty before draining let _ = self.drain(); } @@ -1134,13 +1123,10 @@ impl ValkeyClient { | protocol::SubscriptionPushMessage::Unsubscribe, ) => { // Subscribe/unsubscribe pushes only need promise pairs if we have pending commands - if self.in_flight.readable_length() == 0 - || !self - .in_flight - .peek_item_mut(0) - .meta + if !self.in_flight.front().is_some_and(|pair| { + pair.meta .contains(command::Meta::SUBSCRIPTION_REQUEST) - { + }) { should_consume_promise_pair = false; } } @@ -1157,7 +1143,7 @@ impl ValkeyClient { // responses which indicate all the channels we have connected to. As a stop-gap, we currently ignore the // actual of content of the SUBSCRIBE responses and just resolve the first one with the count of channels. if should_consume_promise_pair { - pair_maybe = self.in_flight.read_item(); + pair_maybe = self.in_flight.pop_front(); } // We handle subscriptions specially because they are not regular commands and their failure will potentially @@ -1336,28 +1322,24 @@ impl ValkeyClient { pub(crate) fn drain(&mut self) -> bool { // If there's something in the in-flight queue and the next command // doesn't support pipelining, we should wait for in-flight commands to complete - if self.in_flight.readable_length() > 0 { - let queue_slice = self.queue.readable_slice(0); - if !queue_slice.is_empty() - && !queue_slice[0] - .meta - .contains(command::Meta::SUPPORTS_AUTO_PIPELINING) - { - return false; - } + if !self.in_flight.is_empty() + && let Some(head) = self.queue.front() + && !head + .meta + .contains(command::Meta::SUPPORTS_AUTO_PIPELINING) + { + return false; } - let Some(offline_cmd) = self.queue.read_item() else { + let Some(offline_cmd) = self.queue.pop_front() else { return false; }; // Add the promise to the command queue first - self.in_flight - .write_item(command::PromisePair { - meta: offline_cmd.meta, - promise: offline_cmd.promise, - }) - .unwrap_or_oom(); + self.in_flight.push_back(command::PromisePair { + meta: offline_cmd.meta, + promise: offline_cmd.promise, + }); let data = offline_cmd.serialized_data; if self.connection_ready() && self.write_buffer.remaining().is_empty() { @@ -1403,13 +1385,13 @@ impl ValkeyClient { let must_wait_for_queue = !command .meta .contains(command::Meta::SUPPORTS_AUTO_PIPELINING) - && self.queue.readable_length() > 0; + && !self.queue.is_empty(); if // If there are any pending commands, queue this one - self.queue.readable_length() > 0 + !self.queue.is_empty() // With auto pipelining, we can accept commands regardless of in_flight commands - || (!can_pipeline && self.in_flight.readable_length() > 0) + || (!can_pipeline && !self.in_flight.is_empty()) // We need authentication before processing commands || !self.connection_ready() // Commands that don't support pipelining must wait for the entire queue to drain @@ -1419,7 +1401,7 @@ impl ValkeyClient { { // We serialize the bytes in here, so we don't need to worry about the lifetime of the Command itself. let entry = command::Entry::create(command, promise)?; - self.queue.write_item(entry)?; + self.queue.push_back(entry); // If we're connected and using auto pipelining, schedule a flush if self.status == Status::Connected && can_pipeline { @@ -1446,7 +1428,7 @@ impl ValkeyClient { }; // Add to queue with command type - self.in_flight.write_item(cmd_pair)?; + self.in_flight.push_back(cmd_pair); let _ = self.flush_data(); Ok(()) @@ -1486,7 +1468,7 @@ impl ValkeyClient { .meta .contains(command::Meta::SUPPORTS_AUTO_PIPELINING) && self.status == Status::Connected - && self.queue.readable_length() > 0 + && !self.queue.is_empty() { self.register_auto_flusher(self.vm); } diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts index d2f3b147ce74..bbac50672568 100644 --- a/test/js/valkey/reliability/connection-failures.test.ts +++ b/test/js/valkey/reliability/connection-failures.test.ts @@ -1,4 +1,5 @@ import { RedisClient } from "bun"; +import { estimateShallowMemoryUsageOf } from "bun:jsc"; import { describe, expect, mock, test } from "bun:test"; import { once } from "events"; import { bunEnv, bunExe, isWindows, tempDir, tls as tlsCert } from "harness"; @@ -1324,3 +1325,79 @@ describe("Valkey: Recovering After fail()", () => { } }); }); + +describe("Valkey: Offline Queue", () => { + // Answers HELLO with `+OK` and every other command with `+PONG`, unless + // `answerHello` is false, in which case the connection never becomes ready + // and everything the client sends stays in its offline queue. + function stubServer({ answerHello = true } = {}) { + const sockets: net.Socket[] = []; + const server = net.createServer(socket => { + sockets.push(socket); + socket.on("data", chunk => { + if (!answerHello) return; + const text = chunk.toString("latin1"); + let commands = 0; + for (let i = text.indexOf("*"); i !== -1; i = text.indexOf("*", i + 1)) commands++; + const helloAt = text.indexOf("HELLO"); + socket.write((helloAt === -1 ? "" : "+OK\r\n") + "+PONG\r\n".repeat(commands - (helloAt === -1 ? 0 : 1))); + }); + socket.on("error", () => {}); + }); + return { + server, + listen: async () => { + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + return (server.address() as net.AddressInfo).port; + }, + close: () => { + for (const socket of sockets) socket.destroy(); + return new Promise(resolve => server.close(resolve)); + }, + }; + } + + test("estimated memory counts every queued command after the queue was drained and refilled", async () => { + const stub = stubServer(); + const port = await stub.listen(); + const client = new RedisClient(`redis://127.0.0.1:${port}`, { autoReconnect: false }); + try { + await client.connect(); + // Six commands go through the queue and are answered, so the queue is + // empty again but its read position is no longer at the start. + await Promise.all(Array.from({ length: 6 }, () => client.ping())); + const idleCost = estimateShallowMemoryUsageOf(client); + + // Five more commands are queued in one turn, before the pipeline + // flushes them. Their serialized bytes must all show up in the estimate. + const key = "k".repeat(1000); + const pending = Array.from({ length: 5 }, () => client.get(key)); + expect(estimateShallowMemoryUsageOf(client) - idleCost).toBeGreaterThanOrEqual(5 * key.length); + + expect(await Promise.all(pending)).toEqual(["PONG", "PONG", "PONG", "PONG", "PONG"]); + } finally { + client.close(); + await stub.close(); + } + }); + + test("close() rejects every command queued while the connection never became ready", async () => { + const stub = stubServer({ answerHello: false }); + const port = await stub.listen(); + const client = new RedisClient(`redis://127.0.0.1:${port}`, { autoReconnect: false }); + try { + const outcomes = Array.from({ length: 40 }, (_, i) => + client.get(`key-${i}`).then( + () => "fulfilled", + (err: Error & { code?: string }) => err.code, + ), + ); + client.close(); + expect(await Promise.all(outcomes)).toEqual(Array(40).fill("ERR_REDIS_CONNECTION_CLOSED")); + } finally { + client.close(); + await stub.close(); + } + }); +}); From 71586184c922cc903917ca40ca8c45a7d1c2722f Mon Sep 17 00:00:00 2001 From: Alistair Smith Date: Tue, 18 Aug 2026 11:40:20 -0700 Subject: [PATCH 02/11] linear fifo: reject item types with drop glue --- src/collections/linear_fifo.rs | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) diff --git a/src/collections/linear_fifo.rs b/src/collections/linear_fifo.rs index 304d5c105243..4e966bac0d0c 100644 --- a/src/collections/linear_fifo.rs +++ b/src/collections/linear_fifo.rs @@ -161,6 +161,12 @@ pub struct LinearFifo> { impl LinearFifo> { /// `init` for `.Static`. pub fn init() -> Self { + const { + assert!( + !mem::needs_drop::(), + "LinearFifo does not drop its items; use VecDeque for types with drop glue" + ) + }; Self { buf: StaticBuffer([const { MaybeUninit::uninit() }; N]), head: 0, @@ -173,6 +179,12 @@ impl LinearFifo> { impl LinearFifo> { /// `init` for `.Dynamic`. pub fn init() -> Self { + const { + assert!( + !mem::needs_drop::(), + "LinearFifo does not drop its items; use VecDeque for types with drop glue" + ) + }; Self { buf: DynamicBuffer(Box::new([])), head: 0, @@ -182,8 +194,11 @@ impl LinearFifo> { } } -// `pub fn deinit` → Drop. Dynamic frees `buf` via `Box` drop; Static/Slice are -// no-ops. Field drop glue covers it; no explicit impl needed. +// `pub fn deinit` → Drop. Dynamic frees `buf` via `Box` drop; Static is a +// no-op. Items are never dropped: the ring is for POD/pointer payloads only, +// and the `!needs_drop::()` const assert in each `init` enforces it, so +// e.g. `LinearFifo::, DynamicBuffer<_>>::init()` fails to compile at +// monomorphization. impl> LinearFifo { #[inline] From 17ac8ee91c678813f5f7f8fad15be979b4feffd3 Mon Sep 17 00:00:00 2001 From: "autofix-ci[bot]" <114827586+autofix-ci[bot]@users.noreply.github.com> Date: Tue, 18 Aug 2026 18:44:45 +0000 Subject: [PATCH 03/11] [autofix.ci] apply automated fixes --- src/runtime/valkey_jsc/js_valkey.rs | 7 +++---- src/runtime/valkey_jsc/valkey.rs | 16 +++++++--------- 2 files changed, 10 insertions(+), 13 deletions(-) diff --git a/src/runtime/valkey_jsc/js_valkey.rs b/src/runtime/valkey_jsc/js_valkey.rs index 4493cf599c90..ecc6964cff84 100644 --- a/src/runtime/valkey_jsc/js_valkey.rs +++ b/src/runtime/valkey_jsc/js_valkey.rs @@ -1610,13 +1610,12 @@ impl JSValkeyClient { memory_cost += client.read_buffer.byte_list.capacity() as usize; // Add queue sizes - memory_cost += client.in_flight.len() - * core::mem::size_of::(); + memory_cost += + client.in_flight.len() * core::mem::size_of::(); for command in client.queue.iter() { memory_cost += command.serialized_data.len(); } - memory_cost += - client.queue.len() * core::mem::size_of::(); + memory_cost += client.queue.len() * core::mem::size_of::(); memory_cost } diff --git a/src/runtime/valkey_jsc/valkey.rs b/src/runtime/valkey_jsc/valkey.rs index e18c9fd20963..d5f6152f4c45 100644 --- a/src/runtime/valkey_jsc/valkey.rs +++ b/src/runtime/valkey_jsc/valkey.rs @@ -724,8 +724,7 @@ impl ValkeyClient { // After draining, check if the *new* head is pipelineable and schedule flush if needed. // This covers sequences like NON_PIPE -> PIPE -> PIPE ... if self.queue.front().is_some_and(|head| { - head.meta - .contains(command::Meta::SUPPORTS_AUTO_PIPELINING) + head.meta.contains(command::Meta::SUPPORTS_AUTO_PIPELINING) }) { self.register_auto_flusher(self.vm); } @@ -1123,10 +1122,11 @@ impl ValkeyClient { | protocol::SubscriptionPushMessage::Unsubscribe, ) => { // Subscribe/unsubscribe pushes only need promise pairs if we have pending commands - if !self.in_flight.front().is_some_and(|pair| { - pair.meta - .contains(command::Meta::SUBSCRIPTION_REQUEST) - }) { + if !self + .in_flight + .front() + .is_some_and(|pair| pair.meta.contains(command::Meta::SUBSCRIPTION_REQUEST)) + { should_consume_promise_pair = false; } } @@ -1324,9 +1324,7 @@ impl ValkeyClient { // doesn't support pipelining, we should wait for in-flight commands to complete if !self.in_flight.is_empty() && let Some(head) = self.queue.front() - && !head - .meta - .contains(command::Meta::SUPPORTS_AUTO_PIPELINING) + && !head.meta.contains(command::Meta::SUPPORTS_AUTO_PIPELINING) { return false; } From e1ee35159ffd384ac821da06dc9a0669f4248fbd Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Tue, 18 Aug 2026 19:01:07 +0000 Subject: [PATCH 04/11] linear fifo: drop the PromisePair example from the element-type notes PromisePair now lives in a VecDeque and the needs_drop assert keeps it out of the ring, so the two notes that listed it as a stored element type are stale. --- src/collections/linear_fifo.rs | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/src/collections/linear_fifo.rs b/src/collections/linear_fifo.rs index 4e966bac0d0c..86114997d571 100644 --- a/src/collections/linear_fifo.rs +++ b/src/collections/linear_fifo.rs @@ -35,8 +35,7 @@ pub trait LinearFifoBuffer { /// layout to `T`; exposing uninitialized bytes as `T` is sound only when any /// bit pattern is a valid `T`. NOT every in-tree element type satisfies this: /// besides byte buffers and raw pointers, fifos today store `NonNull`-bearing -/// enums (`bun_test::RefDataValue`), `JSPromiseStrong`-bearing structs -/// (`ValkeyCommand::PromisePair`), and the `event_loop::Task` enum — see the +/// enums (`bun_test::RefDataValue`) and the `event_loop::Task` enum — see the /// `StaticBuffer` note below for the pending MaybeUninit accessor rework. /// Centralises the four per-buffer-kind casts behind one audited block. #[inline(always)] @@ -95,9 +94,9 @@ fn poison(slice: &mut [T], n: usize) { // (`writable_slice` hands out `&mut [T]` over not-yet-written slots) bakes in // the same exposure for every buffer kind. Sound only for `T` whose // any-bit-pattern is valid — and in-tree element types ALREADY violate that: -// `RefDataValue` (NonNull payload), `PromisePair` -// (JSPromiseStrong), and the `Task` enum are stored in fifos today, so -// materialising `&[T]` over uninitialized slots for those types is latent UB. +// `RefDataValue` (NonNull payload) and the `Task` enum are +// stored in fifos today, so materialising `&[T]` over uninitialized slots for +// those types is latent UB. // The fix is reworking the accessors to operate on `&[MaybeUninit]` and // only assume-init the logically-written subranges. That cannot be done by // touching this file alone — `writable_slice`-family callers in other crates From 9e8a8818e6ec9fc4d05874b5a494074aab990285 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Tue, 18 Aug 2026 19:01:07 +0000 Subject: [PATCH 05/11] valkey test: frame the stub server's commands and settle before asserting The stub counted '*' bytes per TCP chunk. It now buffers and answers each complete RESP command, reusing readCommands from the reconnect test. The memory test awaits its queued commands before it asserts, so a failed assertion no longer leaks five unhandled rejections into the next test. --- .../reliability/connection-failures.test.ts | 120 +++++++++--------- 1 file changed, 63 insertions(+), 57 deletions(-) diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts index bbac50672568..aa93e3954280 100644 --- a/test/js/valkey/reliability/connection-failures.test.ts +++ b/test/js/valkey/reliability/connection-failures.test.ts @@ -342,50 +342,52 @@ describe.skipIf(!isEnabled)("Valkey: Connection Failures", () => { }); }); -describe("Valkey: Auto-Reconnect In-Flight Commands", () => { - function readCommands(state: { buffer: Buffer }): string[][] { - const commands: string[][] = []; - while (true) { - const text = state.buffer.toString("latin1"); - if (text[0] !== "*") break; - const headerEnd = text.indexOf("\r\n"); - if (headerEnd === -1) break; - const argCount = parseInt(text.slice(1, headerEnd), 10); - if (!Number.isInteger(argCount) || argCount < 0) break; - let pos = headerEnd + 2; - const args: string[] = []; - let complete = true; - for (let i = 0; i < argCount; i++) { - if (text[pos] !== "$") { - complete = false; - break; - } - const lenEnd = text.indexOf("\r\n", pos); - if (lenEnd === -1) { - complete = false; - break; - } - const len = parseInt(text.slice(pos + 1, lenEnd), 10); - if (!Number.isInteger(len) || len < 0) { - complete = false; - break; - } - const dataStart = lenEnd + 2; - const dataEnd = dataStart + len; - if (text.length < dataEnd + 2) { - complete = false; - break; - } - args.push(text.slice(dataStart, dataEnd)); - pos = dataEnd + 2; +// Takes the complete RESP commands off the front of `state.buffer` and leaves +// any partial command in it for the next chunk. +function readCommands(state: { buffer: Buffer }): string[][] { + const commands: string[][] = []; + while (true) { + const text = state.buffer.toString("latin1"); + if (text[0] !== "*") break; + const headerEnd = text.indexOf("\r\n"); + if (headerEnd === -1) break; + const argCount = parseInt(text.slice(1, headerEnd), 10); + if (!Number.isInteger(argCount) || argCount < 0) break; + let pos = headerEnd + 2; + const args: string[] = []; + let complete = true; + for (let i = 0; i < argCount; i++) { + if (text[pos] !== "$") { + complete = false; + break; + } + const lenEnd = text.indexOf("\r\n", pos); + if (lenEnd === -1) { + complete = false; + break; + } + const len = parseInt(text.slice(pos + 1, lenEnd), 10); + if (!Number.isInteger(len) || len < 0) { + complete = false; + break; } - if (!complete) break; - commands.push(args); - state.buffer = state.buffer.subarray(pos); + const dataStart = lenEnd + 2; + const dataEnd = dataStart + len; + if (text.length < dataEnd + 2) { + complete = false; + break; + } + args.push(text.slice(dataStart, dataEnd)); + pos = dataEnd + 2; } - return commands; + if (!complete) break; + commands.push(args); + state.buffer = state.buffer.subarray(pos); } + return commands; +} +describe("Valkey: Auto-Reconnect In-Flight Commands", () => { test("rejects commands that were in flight when the connection dropped instead of pairing them with replies from the next connection", async () => { const sockets: net.Socket[] = []; let connections = 0; @@ -1327,20 +1329,22 @@ describe("Valkey: Recovering After fail()", () => { }); describe("Valkey: Offline Queue", () => { - // Answers HELLO with `+OK` and every other command with `+PONG`, unless - // `answerHello` is false, in which case the connection never becomes ready - // and everything the client sends stays in its offline queue. - function stubServer({ answerHello = true } = {}) { + // Answers each complete command in the order it arrives: HELLO with `+OK` + // and everything else with `+PONG`. With `answer: false` nothing is ever + // answered, so the connection never becomes ready and everything the client + // sends stays in its offline queue. + function stubServer({ answer = true } = {}) { const sockets: net.Socket[] = []; const server = net.createServer(socket => { sockets.push(socket); + const state = { buffer: Buffer.alloc(0) }; socket.on("data", chunk => { - if (!answerHello) return; - const text = chunk.toString("latin1"); - let commands = 0; - for (let i = text.indexOf("*"); i !== -1; i = text.indexOf("*", i + 1)) commands++; - const helloAt = text.indexOf("HELLO"); - socket.write((helloAt === -1 ? "" : "+OK\r\n") + "+PONG\r\n".repeat(commands - (helloAt === -1 ? 0 : 1))); + if (!answer) return; + state.buffer = Buffer.concat([state.buffer, chunk]); + const replies = readCommands(state).map(args => + (args[0] ?? "").toUpperCase() === "HELLO" ? "+OK\r\n" : "+PONG\r\n", + ); + if (replies.length > 0) socket.write(replies.join("")); }); socket.on("error", () => {}); }); @@ -1364,18 +1368,20 @@ describe("Valkey: Offline Queue", () => { const client = new RedisClient(`redis://127.0.0.1:${port}`, { autoReconnect: false }); try { await client.connect(); - // Six commands go through the queue and are answered, so the queue is - // empty again but its read position is no longer at the start. + // Six commands go through the queue and are answered. The queue is empty + // again, but its read position has moved to slot 6 of the 8 it grew to, + // so the next five commands wrap around to the start of its storage. await Promise.all(Array.from({ length: 6 }, () => client.ping())); const idleCost = estimateShallowMemoryUsageOf(client); // Five more commands are queued in one turn, before the pipeline // flushes them. Their serialized bytes must all show up in the estimate. - const key = "k".repeat(1000); - const pending = Array.from({ length: 5 }, () => client.get(key)); - expect(estimateShallowMemoryUsageOf(client) - idleCost).toBeGreaterThanOrEqual(5 * key.length); + const key = Buffer.alloc(1000, "k").toString(); + const pending = Promise.all(Array.from({ length: 5 }, () => client.get(key))); + const queuedCost = estimateShallowMemoryUsageOf(client); - expect(await Promise.all(pending)).toEqual(["PONG", "PONG", "PONG", "PONG", "PONG"]); + expect(await pending).toEqual(Array(5).fill("PONG")); + expect(queuedCost - idleCost).toBeGreaterThanOrEqual(5 * key.length); } finally { client.close(); await stub.close(); @@ -1383,7 +1389,7 @@ describe("Valkey: Offline Queue", () => { }); test("close() rejects every command queued while the connection never became ready", async () => { - const stub = stubServer({ answerHello: false }); + const stub = stubServer({ answer: false }); const port = await stub.listen(); const client = new RedisClient(`redis://127.0.0.1:${port}`, { autoReconnect: false }); try { From 853930d43d3018ab0af35b3c440c26c2964d865e Mon Sep 17 00:00:00 2001 From: Alistair Smith Date: Tue, 18 Aug 2026 12:14:52 -0700 Subject: [PATCH 06/11] linear fifo: require copy items instead of a const assert --- src/collections/linear_fifo.rs | 75 ++++++++--------------------- src/install/lockfile/Tree.rs | 1 + src/runtime/test_runner/bun_test.rs | 14 +++--- src/runtime/test_runner/jest.rs | 2 +- 4 files changed, 28 insertions(+), 64 deletions(-) diff --git a/src/collections/linear_fifo.rs b/src/collections/linear_fifo.rs index 86114997d571..bb3806b73c07 100644 --- a/src/collections/linear_fifo.rs +++ b/src/collections/linear_fifo.rs @@ -1,5 +1,6 @@ -// FIFO of fixed size items -// Usually used for e.g. byte buffers +// FIFO of fixed size `Copy` items, usually byte buffers, raw pointers or small +// plain structs. The ring never runs item destructors, so `T: Copy` is the +// contract; owning element types belong in `VecDeque`. use core::marker::PhantomData; use core::mem::{self, MaybeUninit}; @@ -53,22 +54,14 @@ fn assume_init_slice_mut(s: &mut [MaybeUninit]) -> &mut [T] { unsafe { &mut *(ptr::from_mut::<[MaybeUninit]>(s) as *mut [T]) } } -/// Shift `slice[1..]` down to `slice[0..len-1]` (memmove). Used by -/// `ordered_remove_item` for the four wrap/non-wrap segment shifts. Not -/// `slice::copy_within` because that requires `T: Copy`; this fifo permits -/// move-only `T` (the duplicated tail slot is logically discarded by the -/// subsequent `count -= 1`). +/// Shift `slice[1..]` down to `slice[0..len-1]`. Used by +/// `ordered_remove_item` for the four wrap/non-wrap segment shifts; the +/// duplicated tail slot is logically discarded by the subsequent `count -= 1`. #[inline(always)] -fn shift_down_one(slice: &mut [T]) { - let len = slice.len(); - if len <= 1 { - return; - } - let p = slice.as_mut_ptr(); - // SAFETY: src `[1..len)` and dst `[0..len-1)` are both in-bounds of - // `slice`; `ptr::copy` handles the overlap. Both pointers derive from one - // `as_mut_ptr()` so the src tag is not invalidated by a later Unique retag. - unsafe { ptr::copy(p.add(1), p, len - 1) }; +fn shift_down_one(slice: &mut [T]) { + if slice.len() > 1 { + slice.copy_within(1.., 0); + } } #[cfg(debug_assertions)] @@ -157,15 +150,9 @@ pub struct LinearFifo> { // re-exported as `bun_io::Write`), plus `std::io::Read`, `std::io::Write`, // and `core::fmt::Write` for std interop. -impl LinearFifo> { +impl LinearFifo> { /// `init` for `.Static`. pub fn init() -> Self { - const { - assert!( - !mem::needs_drop::(), - "LinearFifo does not drop its items; use VecDeque for types with drop glue" - ) - }; Self { buf: StaticBuffer([const { MaybeUninit::uninit() }; N]), head: 0, @@ -175,15 +162,9 @@ impl LinearFifo> { } } -impl LinearFifo> { +impl LinearFifo> { /// `init` for `.Dynamic`. pub fn init() -> Self { - const { - assert!( - !mem::needs_drop::(), - "LinearFifo does not drop its items; use VecDeque for types with drop glue" - ) - }; Self { buf: DynamicBuffer(Box::new([])), head: 0, @@ -194,12 +175,9 @@ impl LinearFifo> { } // `pub fn deinit` → Drop. Dynamic frees `buf` via `Box` drop; Static is a -// no-op. Items are never dropped: the ring is for POD/pointer payloads only, -// and the `!needs_drop::()` const assert in each `init` enforces it, so -// e.g. `LinearFifo::, DynamicBuffer<_>>::init()` fails to compile at -// monomorphization. +// no-op. Items are never dropped, which is why every impl requires `T: Copy`. -impl> LinearFifo { +impl> LinearFifo { #[inline] fn buf_len(&self) -> usize { self.buf.len() @@ -408,10 +386,7 @@ impl> LinearFifo { } /// Read data from the fifo into `dst`, returns number of items copied. - pub(crate) fn read(&mut self, dst: &mut [T]) -> usize - where - T: Copy, - { + pub(crate) fn read(&mut self, dst: &mut [T]) -> usize { let total = dst.len(); let mut dst_left = &mut dst[..]; @@ -480,10 +455,7 @@ impl> LinearFifo { /// Appends the data in `src` to the fifo. /// You must have ensured there is enough space. - pub(crate) fn write_assume_capacity(&mut self, src: &[T]) - where - T: Copy, - { + pub(crate) fn write_assume_capacity(&mut self, src: &[T]) { debug_assert!(self.writable_length() >= src.len()); let mut src_left = src; @@ -526,10 +498,7 @@ impl> LinearFifo { /// Appends the data in `src` to the fifo. /// Allocates more memory as necessary - pub fn write(&mut self, src: &[T]) -> Result<(), AllocError> - where - T: Copy, - { + pub fn write(&mut self, src: &[T]) -> Result<(), AllocError> { self.ensure_unused_capacity(src.len())?; self.write_assume_capacity(src); Ok(()) @@ -550,10 +519,7 @@ impl> LinearFifo { } /// Place data back into the read stream - pub fn unget(&mut self, src: &[T]) -> Result<(), AllocError> - where - T: Copy, - { + pub fn unget(&mut self, src: &[T]) -> Result<(), AllocError> { self.ensure_unused_capacity(src.len())?; self.rewind(src.len()); @@ -575,10 +541,7 @@ impl> LinearFifo { /// Returns the item at `offset`. /// Asserts offset is within bounds. - pub fn peek_item(&self, offset: usize) -> T - where - T: Copy, - { + pub fn peek_item(&self, offset: usize) -> T { debug_assert!(offset < self.count); let mut index = self.head + offset; diff --git a/src/install/lockfile/Tree.rs b/src/install/lockfile/Tree.rs index f6960ec29f18..6ed128eaa9d3 100644 --- a/src/install/lockfile/Tree.rs +++ b/src/install/lockfile/Tree.rs @@ -1073,6 +1073,7 @@ impl Tree { // FillItem / TreeFiller // ────────────────────────────────────────────────────────────────────────── +#[derive(Clone, Copy)] pub struct FillItem { pub(crate) tree_id: Id, pub(crate) dependency_id: DependencyID, diff --git a/src/runtime/test_runner/bun_test.rs b/src/runtime/test_runner/bun_test.rs index 40f339e47c30..30023ceab736 100644 --- a/src/runtime/test_runner/bun_test.rs +++ b/src/runtime/test_runner/bun_test.rs @@ -777,9 +777,9 @@ impl BunTest { return Ok(()); } - this.add_result(refdata.phase.clone()); + this.add_result(refdata.phase); // `this` borrow ends here (NLL); `run_next_tick` re-derives via `.get()`. - Self::run_next_tick(&refdata.buntest_weak, global_this, refdata.phase.clone()); + Self::run_next_tick(&refdata.buntest_weak, global_this, refdata.phase); Ok(()) } @@ -852,8 +852,8 @@ impl BunTest { }; // SAFETY: `&mut` derived via `UnsafeCell`; borrow ends before // `run_next_tick` re-derives. - strong.get().add_result(ref_in.phase.clone()); - Self::run_next_tick(&ref_in.buntest_weak, global_this, ref_in.phase.clone()); + strong.get().add_result(ref_in.phase); + Self::run_next_tick(&ref_in.buntest_weak, global_this, ref_in.phase); Ok(JSValue::UNDEFINED) } @@ -1201,7 +1201,7 @@ impl BunTest { if unsafe { (*dcb_data).called } { // done callback already called or the callback errored; add result immediately } else { - let r = Self::ref_(this_strong, cfg_data.clone()); + let r = Self::ref_(this_strong, cfg_data); let alias = NonNull::new(r.as_ptr()) .expect("ref_() returns a freshly-boxed RefData"); // SAFETY: see above. Move the sole +1 into the DoneCallback. @@ -1421,10 +1421,10 @@ pub struct EntryData { pub(crate) remaining_repeat_count: i64, } -// Clone: bitwise OK — `active_scope` is a non-owning borrow of a +// Clone/Copy: bitwise OK — `active_scope` is a non-owning borrow of a // `DescribeScope` whose lifetime spans the async boundary (see field note); // `EntryData.entry` likewise borrows. -#[derive(Clone)] +#[derive(Clone, Copy)] pub enum RefDataValue { Start, Collection { diff --git a/src/runtime/test_runner/jest.rs b/src/runtime/test_runner/jest.rs index 2889638ff424..932c0efa0c17 100644 --- a/src/runtime/test_runner/jest.rs +++ b/src/runtime/test_runner/jest.rs @@ -558,7 +558,7 @@ pub(crate) fn js_node_test_mark_result( // threaded JS VM, GC roots `done` (and its bound-this) for this frame. let (dcb_ref, dcb_called) = unsafe { ((*dcb).r#ref.as_deref(), (*dcb).called) }; let bound = match dcb_ref { - Some(refdata) => refdata.phase.clone(), + Some(refdata) => refdata.phase, // `r#ref` unset: `.then()` fired inside run_test_callback's microtask // drain before it stamps the DoneCallback. `get_current_state_data()` // can't name a sequence inside a concurrent group, but From cc25db75d63ef34390b5b60a4b9ac15d6bc19aaa Mon Sep 17 00:00:00 2001 From: Alistair Smith Date: Tue, 18 Aug 2026 12:14:53 -0700 Subject: [PATCH 07/11] valkey: drop the finalizing pop loops in shutdown --- src/runtime/valkey_jsc/valkey.rs | 19 +------------------ 1 file changed, 1 insertion(+), 18 deletions(-) diff --git a/src/runtime/valkey_jsc/valkey.rs b/src/runtime/valkey_jsc/valkey.rs index d5f6152f4c45..5e23976d16e2 100644 --- a/src/runtime/valkey_jsc/valkey.rs +++ b/src/runtime/valkey_jsc/valkey.rs @@ -340,6 +340,7 @@ impl ValkeyClient { let mut pending = core::mem::take(&mut self.in_flight); let mut commands = core::mem::take(&mut self.queue); + // When finalizing we cannot call into JS; the queues just drop. if let Some(global_this) = global_object_or_finalizing { let object = valkey_error_to_js( global_this, @@ -355,27 +356,9 @@ impl ValkeyClient { while let Some(mut offline_cmd) = commands.pop_front() { // Same as above: swallow reject exceptions so the whole queue drains. let _ = offline_cmd.promise.reject(global_this, Ok(object)); - // Note: `offline_cmd.deinit()` — Entry/Box<[u8]> drops automatically. - } - } else { - // finalizing. we can't call into JS. - while let Some(pair) = pending.pop_front() { - // Note: `pair.promise.deinit()` — JSPromiseStrong drops automatically. - drop(pair); - } - - while let Some(offline_cmd) = commands.pop_front() { - // Note: `offline_cmd.promise.deinit()` / `offline_cmd.deinit()` — - // JSPromiseStrong / Box<[u8]> drop automatically. - drop(offline_cmd); } } - // Note: `allocator.free(connection_strings)` and `write_buffer/read_buffer.deinit()` - // and `tls.deinit()` are handled by Drop on the owning fields. Only the side-effecting - // unregister remains explicit. - drop(pending); - drop(commands); self.unregister_auto_flusher(); } From 019f6ef50b09745dbeaed659e3c11dc818dc9519 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Tue, 18 Aug 2026 19:35:33 +0000 Subject: [PATCH 08/11] linear fifo: update the write_item safety note for the Copy bound The note said ptr::write was needed for non-Copy items. Every impl now requires T: Copy, so the reason is that the slot is uninitialized storage. --- src/collections/linear_fifo.rs | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/src/collections/linear_fifo.rs b/src/collections/linear_fifo.rs index bb3806b73c07..7349568a244c 100644 --- a/src/collections/linear_fifo.rs +++ b/src/collections/linear_fifo.rs @@ -489,9 +489,8 @@ impl> LinearFifo { tail %= self.buf_len(); } // SAFETY: `tail` is in-bounds (capacity reserved by caller). The slot is - // logically uninitialized — `ptr::write` does not drop the prior - // bit-pattern, which is required for non-`Copy` `T` whose backing - // storage is `MaybeUninit`. + // logically uninitialized `MaybeUninit` storage; `ptr::write` + // initializes it without reading the prior bit-pattern. unsafe { ptr::write(self.buf.as_mut_slice().as_mut_ptr().add(tail), item) }; self.update(1); } From 77475dd1aef27b8e76b76f2293fc9a75b3820813 Mon Sep 17 00:00:00 2001 From: Alistair Smith Date: Tue, 18 Aug 2026 13:15:33 -0700 Subject: [PATCH 09/11] valkey test: wrapped queue reaches the server in one write --- src/collections/linear_fifo.rs | 8 +- .../reliability/connection-failures.test.ts | 77 +++++++++++++++++++ 2 files changed, 81 insertions(+), 4 deletions(-) diff --git a/src/collections/linear_fifo.rs b/src/collections/linear_fifo.rs index 7349568a244c..3fa56abcc5ea 100644 --- a/src/collections/linear_fifo.rs +++ b/src/collections/linear_fifo.rs @@ -91,10 +91,10 @@ fn poison(slice: &mut [T], n: usize) { // stored in fifos today, so materialising `&[T]` over uninitialized slots for // those types is latent UB. // The fix is reworking the accessors to operate on `&[MaybeUninit]` and -// only assume-init the logically-written subranges. That cannot be done by -// touching this file alone — `writable_slice`-family callers in other crates -// see the signature change — so it is deferred to a dedicated change with -// Miri coverage for a NonNull-bearing element type. +// only assume-init the logically-written subranges; `writable_slice`-family +// callers in other crates see the signature change, so that lives in #31835. +// The `T: Copy` bound below only pins the no-destructor contract; it does not +// close this gap. pub struct StaticBuffer([MaybeUninit; N]); impl LinearFifoBuffer for StaticBuffer { diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts index aa93e3954280..dc0c2fb0cfee 100644 --- a/test/js/valkey/reliability/connection-failures.test.ts +++ b/test/js/valkey/reliability/connection-failures.test.ts @@ -1388,6 +1388,83 @@ describe("Valkey: Offline Queue", () => { } }); + test("commands queued after the queue wrapped reach the server in one write", async () => { + // The client runs in a child process with nothing else live and queues + // the GETs from a setImmediate callback, so the flush runs at the end of + // that tick and the loop then parks. A client that only flushes the part + // of the queue before the wrap point writes two GETs there and the other + // three on a later, unrelated wake. The stub records how many GETs the + // first read carrying a GET held, and holds the GET replies until it has + // all five so no reply can wake the child in between. The deadline only + // bounds the failure. + let getsSeen = 0; + let getsInFirstRead = 0; + const sockets: Bun.Socket<{ buffer: Buffer }>[] = []; + const allGetsSeen = Promise.withResolvers(); + const server = Bun.listen<{ buffer: Buffer }>({ + hostname: "127.0.0.1", + port: 0, + socket: { + open(socket) { + socket.data = { buffer: Buffer.alloc(0) }; + sockets.push(socket); + }, + error() {}, + close() {}, + data(socket, chunk) { + const state = socket.data; + state.buffer = Buffer.concat([state.buffer, chunk]); + let replies = ""; + let gets = 0; + for (const args of readCommands(state)) { + const name = (args[0] ?? "").toUpperCase(); + if (name === "GET") { + gets += 1; + } else { + replies += name === "HELLO" ? "+OK\r\n" : "+PONG\r\n"; + } + } + if (replies) socket.write(replies); + if (gets > 0 && getsSeen === 0) getsInFirstRead = gets; + getsSeen += gets; + if (getsSeen >= 5) allGetsSeen.resolve(); + }, + }, + }); + const releaseReplies = Promise.race([allGetsSeen.promise, Bun.sleep(2000)]).then(() => { + for (const socket of sockets) socket.write("$1\r\nv\r\n".repeat(getsSeen)); + }); + try { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + const client = new Bun.RedisClient("redis://127.0.0.1:${server.port}", { autoReconnect: false }); + await client.connect(); + await Promise.all(Array.from({ length: 6 }, () => client.ping())); + await new Promise(resolve => setImmediate(resolve)); + const values = await Promise.all(Array.from({ length: 5 }, () => client.get("k"))); + console.log(values.length, "replies"); + client.close(); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]); + await releaseReplies; + expect({ getsInFirstRead, stdout, exitCode }).toEqual({ + getsInFirstRead: 5, + stdout: "5 replies\n", + exitCode: 0, + }); + } finally { + server.stop(true); + } + }); + test("close() rejects every command queued while the connection never became ready", async () => { const stub = stubServer({ answer: false }); const port = await stub.listen(); From 78a748c4c83279bbf6a1ad021718e9830e482623 Mon Sep 17 00:00:00 2001 From: Alistair Smith Date: Tue, 18 Aug 2026 13:40:45 -0700 Subject: [PATCH 10/11] valkey test: drain and assert stderr in the wrapped queue test --- test/js/valkey/reliability/connection-failures.test.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts index dc0c2fb0cfee..14747c2a4ec1 100644 --- a/test/js/valkey/reliability/connection-failures.test.ts +++ b/test/js/valkey/reliability/connection-failures.test.ts @@ -1453,11 +1453,12 @@ describe("Valkey: Offline Queue", () => { stdout: "pipe", stderr: "pipe", }); - const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); await releaseReplies; - expect({ getsInFirstRead, stdout, exitCode }).toEqual({ + expect({ getsInFirstRead, stdout, stderr, exitCode }).toEqual({ getsInFirstRead: 5, stdout: "5 replies\n", + stderr: "", exitCode: 0, }); } finally { From 9186d6416ce053852c7ef142c537857521cf910e Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Tue, 18 Aug 2026 20:45:29 +0000 Subject: [PATCH 11/11] valkey test: answer the wrapped GETs from the stub instead of on a deadline The stub now writes the five GET replies when the fifth GET arrives. The 2s deadline started before the child was spawned, so a slow start could release zero replies and leave the child waiting until the test timeout. Without the fix the fifth GET still arrives in the second write, so the child still exits and the assertion still reports 2 of 5 in the first read. --- .../reliability/connection-failures.test.ts | 15 +++++---------- 1 file changed, 5 insertions(+), 10 deletions(-) diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts index 14747c2a4ec1..7c857b3c3625 100644 --- a/test/js/valkey/reliability/connection-failures.test.ts +++ b/test/js/valkey/reliability/connection-failures.test.ts @@ -1395,19 +1395,17 @@ describe("Valkey: Offline Queue", () => { // of the queue before the wrap point writes two GETs there and the other // three on a later, unrelated wake. The stub records how many GETs the // first read carrying a GET held, and holds the GET replies until it has - // all five so no reply can wake the child in between. The deadline only - // bounds the failure. + // all five so no reply can wake the child in between. It then answers all + // five, so the child exits and the count is asserted however the GETs + // arrived. let getsSeen = 0; let getsInFirstRead = 0; - const sockets: Bun.Socket<{ buffer: Buffer }>[] = []; - const allGetsSeen = Promise.withResolvers(); const server = Bun.listen<{ buffer: Buffer }>({ hostname: "127.0.0.1", port: 0, socket: { open(socket) { socket.data = { buffer: Buffer.alloc(0) }; - sockets.push(socket); }, error() {}, close() {}, @@ -1426,14 +1424,12 @@ describe("Valkey: Offline Queue", () => { } if (replies) socket.write(replies); if (gets > 0 && getsSeen === 0) getsInFirstRead = gets; + const getsSeenBefore = getsSeen; getsSeen += gets; - if (getsSeen >= 5) allGetsSeen.resolve(); + if (getsSeenBefore < 5 && getsSeen >= 5) socket.write("$1\r\nv\r\n".repeat(getsSeen)); }, }, }); - const releaseReplies = Promise.race([allGetsSeen.promise, Bun.sleep(2000)]).then(() => { - for (const socket of sockets) socket.write("$1\r\nv\r\n".repeat(getsSeen)); - }); try { await using proc = Bun.spawn({ cmd: [ @@ -1454,7 +1450,6 @@ describe("Valkey: Offline Queue", () => { stderr: "pipe", }); const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - await releaseReplies; expect({ getsInFirstRead, stdout, stderr, exitCode }).toEqual({ getsInFirstRead: 5, stdout: "5 replies\n",