diff --git a/src/collections/linear_fifo.rs b/src/collections/linear_fifo.rs index 304d5c105243..3fa56abcc5ea 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}; @@ -35,8 +36,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)] @@ -54,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)] @@ -95,14 +87,14 @@ 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 -// 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 { @@ -158,7 +150,7 @@ 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 { Self { @@ -170,7 +162,7 @@ impl LinearFifo> { } } -impl LinearFifo> { +impl LinearFifo> { /// `init` for `.Dynamic`. pub fn init() -> Self { Self { @@ -182,10 +174,10 @@ 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, which is why every impl requires `T: Copy`. -impl> LinearFifo { +impl> LinearFifo { #[inline] fn buf_len(&self) -> usize { self.buf.len() @@ -394,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[..]; @@ -466,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; @@ -503,19 +489,15 @@ 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); } /// 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(()) @@ -536,10 +518,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()); @@ -561,10 +540,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 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..ecc6964cff84 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,12 @@ impl JSValkeyClient { memory_cost += client.read_buffer.byte_list.capacity() as usize; // Add queue sizes - memory_cost += client.in_flight.readable_length() - * core::mem::size_of::(); - for command in client.queue.readable_slice(0) { + 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.readable_length() * 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 ff8d11287b11..5e23976d16e2 100644 --- a/src/runtime/valkey_jsc/valkey.rs +++ b/src/runtime/valkey_jsc/valkey.rs @@ -337,46 +337,28 @@ 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); + // 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, 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() { - // Note: `pair.promise.deinit()` — JSPromiseStrong drops automatically. - drop(pair); - } - - while let Some(offline_cmd) = commands.read_item() { - // 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(); } @@ -408,11 +390,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 +410,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 +424,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 +446,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 +484,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 +502,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 +512,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 +521,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 +548,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 +556,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 +695,20 @@ 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 - .contains(command::Meta::SUPPORTS_AUTO_PIPELINING) - { + 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 +718,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,12 +1105,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 - .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; } @@ -1157,7 +1126,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 +1305,22 @@ 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 +1366,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 +1382,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 +1409,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 +1449,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..7c857b3c3625 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"; @@ -341,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; } - if (!complete) break; - commands.push(args); - state.buffer = state.buffer.subarray(pos); + 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; } - 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; @@ -1324,3 +1327,156 @@ describe("Valkey: Recovering After fail()", () => { } }); }); + +describe("Valkey: Offline Queue", () => { + // 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 (!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", () => {}); + }); + 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. 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 = Buffer.alloc(1000, "k").toString(); + const pending = Promise.all(Array.from({ length: 5 }, () => client.get(key))); + const queuedCost = estimateShallowMemoryUsageOf(client); + + expect(await pending).toEqual(Array(5).fill("PONG")); + expect(queuedCost - idleCost).toBeGreaterThanOrEqual(5 * key.length); + } finally { + client.close(); + await stub.close(); + } + }); + + 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. 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 server = Bun.listen<{ buffer: Buffer }>({ + hostname: "127.0.0.1", + port: 0, + socket: { + open(socket) { + socket.data = { buffer: Buffer.alloc(0) }; + }, + 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; + const getsSeenBefore = getsSeen; + getsSeen += gets; + if (getsSeenBefore < 5 && getsSeen >= 5) 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, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ getsInFirstRead, stdout, stderr, exitCode }).toEqual({ + getsInFirstRead: 5, + stdout: "5 replies\n", + stderr: "", + 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(); + 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(); + } + }); +});