diff --git a/src/rust/jsg-test/tests/local_value.rs b/src/rust/jsg-test/tests/local_value.rs index 2db3fb617fc..bc9d0e2305b 100644 --- a/src/rust/jsg-test/tests/local_value.rs +++ b/src/rust/jsg-test/tests/local_value.rs @@ -144,3 +144,24 @@ fn stored_value_round_trips_through_traced_global() { Ok(()) }); } + +#[test] +fn to_detail_string_describes_a_value_without_running_javascript() { + crate::Harness::new().run_in_context(|lock, ctx| { + let cases = [ + ("new TypeError('bad')", "TypeError: bad"), + ("'text'", "text"), + ("42", "42"), + ("undefined", "undefined"), + ( + "({ toString() { throw new Error('ran'); } })", + "[object Object]", + ), + ]; + for (code, expected) in cases { + let value = ctx.eval_raw(lock, code).unwrap(); + assert_eq!(value.to_detail_string(lock)?, expected, "{code}"); + } + Ok(()) + }); +} diff --git a/src/rust/jsg-test/tests/promise.rs b/src/rust/jsg-test/tests/promise.rs index ff462aa9cac..c3ba23cfb4b 100644 --- a/src/rust/jsg-test/tests/promise.rs +++ b/src/rust/jsg-test/tests/promise.rs @@ -354,6 +354,8 @@ impl jsg::Traced for DropCounter {} static SETTLED_DATA_DROPS: AtomicUsize = AtomicUsize::new(0); +/// As in C++ JSG, reaction data is dropped once a reaction has run, without waiting for the +/// garbage collector, while the promise stays reachable. #[test] fn then_data_is_dropped_once_the_promise_settles() { let harness = crate::Harness::new(); @@ -375,6 +377,7 @@ fn then_data_is_dropped_once_the_promise_settles() { let value = lock.undefined(); resolver.resolve(lock, value)?; crate::Harness::run_microtasks(lock); + assert_eq!(SETTLED_DATA_DROPS.load(Ordering::SeqCst), 1); Ok(()) }); harness.run_in_context(|lock, _ctx| { @@ -384,6 +387,39 @@ fn then_data_is_dropped_once_the_promise_settles() { }); } +static PASSED_THROUGH_DROPS: AtomicUsize = AtomicUsize::new(0); + +/// The data of `then` is dropped when a rejection passes through, that of `catch` when a +/// fulfillment does, and that of a callback that fails once it has returned. +#[test] +fn reaction_data_is_dropped_whichever_way_the_promise_settles() { + crate::Harness::new().run_in_context(|lock, ctx| { + let reason = lock.undefined(); + let rejected = v8::Promise::rejected(lock, reason)?; + let passed = rejected.then(lock, DropCounter(&PASSED_THROUGH_DROPS), |lock, _, _| { + Ok(lock.undefined()) + })?; + passed.mark_as_handled(); + let value = lock.undefined(); + let fulfilled = v8::Promise::resolved(lock, value)?; + fulfilled.catch(lock, DropCounter(&PASSED_THROUGH_DROPS), |lock, _, _| { + Ok(lock.undefined()) + })?; + let value = lock.undefined(); + let fulfilled = v8::Promise::resolved(lock, value)?; + let failed = fulfilled.then(lock, DropCounter(&PASSED_THROUGH_DROPS), |_, _, _| { + Err(jsg::Error::new_type_error("failed")) + })?; + failed.mark_as_handled(); + // All three stay reachable, so only their reactions can have dropped the data. + ctx.set_global(lock, "passed", passed); + ctx.set_global(lock, "failed", failed); + crate::Harness::run_microtasks(lock); + assert_eq!(PASSED_THROUGH_DROPS.load(Ordering::SeqCst), 3); + Ok(()) + }); +} + static SHARED_LEAF_DROPS: AtomicUsize = AtomicUsize::new(0); #[jsg_resource] diff --git a/src/rust/jsg/ffi.c++ b/src/rust/jsg/ffi.c++ index 9917be07649..95a4ce9a285 100644 --- a/src/rust/jsg/ffi.c++ +++ b/src/rust/jsg/ffi.c++ @@ -302,6 +302,11 @@ bool local_is_shared_array_buffer(const Local& val) { return ::rust::String(*utf8, utf8.length()); } +MaybeLocal local_to_detail_string(Isolate* isolate, const Local& val) { + return maybe_local_to_ffi( + local_as_ref_from_ffi(val)->ToDetailString(isolate->GetCurrentContext())); +} + // Utf8Value Utf8Value utf8_value_new(Isolate* isolate, Local value) { auto* v = new v8::String::Utf8Value(isolate, local_from_ffi(kj::mv(value))); @@ -534,18 +539,6 @@ bool exceptionPending(Isolate* isolate) { } } // namespace -MaybeLocal local_promise_then(Isolate* isolate, const Local& promise, const Local& on_fulfilled) { - if (exceptionPending(isolate)) return maybe_local_to_ffi(v8::MaybeLocal()); - return maybe_local_to_ffi(local_as_ref_from_ffi(promise)->Then( - isolate->GetCurrentContext(), local_as_ref_from_ffi(on_fulfilled))); -} - -MaybeLocal local_promise_catch(Isolate* isolate, const Local& promise, const Local& on_rejected) { - if (exceptionPending(isolate)) return maybe_local_to_ffi(v8::MaybeLocal()); - return maybe_local_to_ffi(local_as_ref_from_ffi(promise)->Catch( - isolate->GetCurrentContext(), local_as_ref_from_ffi(on_rejected))); -} - MaybeLocal local_promise_then_or_catch( Isolate* isolate, const Local& promise, const Local& on_fulfilled, const Local& on_rejected) { if (exceptionPending(isolate)) return maybe_local_to_ffi(v8::MaybeLocal()); diff --git a/src/rust/jsg/ffi.h b/src/rust/jsg/ffi.h index ed1429709aa..f1ca3b91cd0 100644 --- a/src/rust/jsg/ffi.h +++ b/src/rust/jsg/ffi.h @@ -164,6 +164,7 @@ bool local_is_name(const Local& val); bool local_is_big_int(const Local& val); bool local_is_date(const Local& val); ::rust::String local_type_of(Isolate* isolate, const Local& val); +MaybeLocal local_to_detail_string(Isolate* isolate, const Local& val); // Local Local local_string_empty(Isolate* isolate); @@ -248,18 +249,15 @@ PromiseState local_promise_state(const Local& promise); // if it is. Local local_promise_result(const Local& promise); void local_promise_mark_as_handled(const Local& promise); -// `promise.then(on_fulfilled)`, `promise.catch(on_rejected)` and // `promise.then(on_fulfilled, on_rejected)`, without looking up `then` or the species // constructor. An empty result means an exception is pending (one that a promise hook threw, or a -// termination), as local_function_call() leaves it. While one is pending these, and the +// termination), as local_function_call() leaves it. While one is pending this, and the // PromiseResolver shims, call nothing in V8, which would clear it. -MaybeLocal local_promise_then(Isolate* isolate, const Local& promise, const Local& on_fulfilled); -MaybeLocal local_promise_catch(Isolate* isolate, const Local& promise, const Local& on_rejected); MaybeLocal local_promise_then_or_catch( Isolate* isolate, const Local& promise, const Local& on_fulfilled, const Local& on_rejected); // Local -// An empty result, or false, means an exception is pending, as for local_promise_then(). +// An empty result, or false, means an exception is pending, as for local_promise_then_or_catch(). MaybeLocal promise_resolver_new(Isolate* isolate); Local local_promise_resolver_promise(const Local& resolver); // Resolving with a thenable reads its `then` property, which can run JavaScript; an exception it diff --git a/src/rust/jsg/promise.rs b/src/rust/jsg/promise.rs index a1028f4f3b1..40cc98f2c82 100644 --- a/src/rust/jsg/promise.rs +++ b/src/rust/jsg/promise.rs @@ -10,6 +10,8 @@ reason = "settles and reacts to promises through the V8 FFI" )] +use std::cell::RefCell; + use crate::Error; use crate::FromJS; use crate::Lock; @@ -171,12 +173,27 @@ impl Promise { } } +/// The data of a promise's reactions, which the first of them to run drops once it returns, as +/// C++ JSG drops a `ThenCatchPair`: a promise settles once, so the other can no longer run. Until +/// then, the garbage collector traces it through the reactions. +struct ReactionData(RefCell>); + +impl Traced for ReactionData { + fn holds_gc_references() -> bool { + T::holds_gc_references() + } + + fn trace(&self, visitor: &mut crate::GcVisitor) { + self.0.trace(visitor); + } +} + /// The callback of a reaction that [`Local::then`], [`Local::catch`] or /// [`Local::then_or_catch`] registers: calls `F` with the reaction's data and its argument, the -/// fulfillment value or rejection reason. +/// fulfillment value or rejection reason, then drops the data. fn react<'a, T, F>( lock: &mut Lock<'a>, - data: &T, + data: &ReactionData, args: &mut FunctionCallbackInfo<'a>, ) -> Result, Error> where @@ -187,10 +204,20 @@ where { // SAFETY: `then`, `catch` and `then_or_catch` were handed a value of `F`. let f = unsafe { crate::scope::unit_value::() }; - f(lock, data, args.get(0)) + // The data stays in the cell, where the garbage collector traces it, while `f` runs. + let result = match data.0.borrow().as_ref() { + Some(data) => f(lock, data, args.get(0)), + None => Err(Error::new_internal_error( + "a promise reaction ran after another reaction to the promise", + )), + }; + // Dropped after the borrow ends, so that what its drop does cannot find the cell borrowed. + let data = data.0.borrow_mut().take(); + drop(data); + result } -/// The callback of a function created with data `T` that calls the reaction `F`. +/// The callback of a function created with reaction data `T` that calls the reaction `F`. fn reaction() -> crate::RawFunctionCallback where T: Traced + 'static, @@ -198,7 +225,18 @@ where + 'static + for<'b> Fn(&mut Lock<'b>, &T, Local<'b, Value>) -> Result, Error>, { - Function::callback::(react::) + Function::callback::, _>(react::) +} + +/// The reaction [`Local::then`] registers for a rejection: passes it through. +fn pass_rejection<'a, T>( + lock: &mut Lock<'a>, + _data: &T, + reason: Local<'a, Value>, +) -> Result, Error> { + // SAFETY: `lock` guarantees the isolate is locked; `reason` is a handle in its scope. + unsafe { ffi::isolate_throw_exception(lock.isolate().as_ffi(), reason.into_ffi()) }; + Err(Error::pending()) } impl<'s> Local<'s, Promise> { @@ -237,12 +275,13 @@ impl<'s> Local<'s, Promise> { /// no JavaScript runs until the promise settles. /// /// As with [`Function::new`], the callback must capture nothing, which a compile-time - /// assertion checks, and whatever it needs comes from `data`. The garbage collector traces - /// `data` until the promise settles and drops it once the callback can no longer run, so it - /// may hold `TracedGlobal`s and `jsg::Rc`s, including ones that lead back to the promise. It - /// must be their only owner that the garbage collector traces: to share a resource's state, - /// hold a `jsg::Rc` or `jsg::Weak` to the resource rather than an `std::rc::Rc` of the - /// state. + /// assertion checks, and whatever it needs comes from `data`. As in C++ JSG, `data` is + /// dropped as soon as the promise settles: after the callback returns, or when a rejection + /// passes through. If the promise never settles, the garbage collector drops it once the + /// promise is unreachable. Until then it traces `data`, so it may hold `TracedGlobal`s and + /// `jsg::Rc`s, including ones that lead back to the promise. It must be their only owner that + /// the garbage collector traces: to share a resource's state, hold a `jsg::Rc` or + /// `jsg::Weak` to the resource rather than an `std::rc::Rc` of the state. /// /// # Errors /// @@ -254,28 +293,19 @@ impl<'s> Local<'s, Promise> { + 'static + for<'a> Fn(&mut Lock<'a>, &T, Local<'a, Value>) -> Result, Error>, { - self.debug_assert_isolate(lock); - let _ = on_fulfilled; - let [on_fulfilled] = Function::new_sharing_data(lock, data, [reaction::()])?; - // SAFETY: lock guarantees the isolate is locked and a HandleScope is active; both are - // valid handles, and a non-empty result is a promise in the lock's scope. - unsafe { - Local::from_ffi_maybe(ffi::local_promise_then( - lock.isolate().as_ffi(), - self.raw(), - on_fulfilled.raw(), - )) - } - .ok_or_else(Error::pending) + // As C++ JSG does, a rejection passes through a reaction of its own, so that `data` is + // dropped then too. + self.then_or_catch(lock, data, on_fulfilled, pass_rejection::) } /// Registers `on_rejected` to run when the promise is rejected, like /// `promise.catch(onRejected)`, and returns the promise for its result. A fulfillment passes /// through to the returned promise unchanged. /// - /// Like [`Local::then`] otherwise, with the rejection reason as the callback's argument. - /// Unlike `promise.catch()`, this does not call `promise.then`. Registering `on_rejected` - /// marks the promise as handled, as in JavaScript. + /// Like [`Local::then`] otherwise, with the rejection reason as the callback's argument, and + /// `data` dropped when a fulfillment passes through. Unlike `promise.catch()`, this does not + /// call `promise.then`. Registering `on_rejected` marks the promise as handled, as in + /// JavaScript. /// /// # Errors /// @@ -287,19 +317,9 @@ impl<'s> Local<'s, Promise> { + 'static + for<'a> Fn(&mut Lock<'a>, &T, Local<'a, Value>) -> Result, Error>, { - self.debug_assert_isolate(lock); - let _ = on_rejected; - let [on_rejected] = Function::new_sharing_data(lock, data, [reaction::()])?; - // SAFETY: lock guarantees the isolate is locked and a HandleScope is active; both are - // valid handles, and a non-empty result is a promise in the lock's scope. - unsafe { - Local::from_ffi_maybe(ffi::local_promise_catch( - lock.isolate().as_ffi(), - self.raw(), - on_rejected.raw(), - )) - } - .ok_or_else(Error::pending) + // As C++ JSG does, a fulfillment passes through a reaction of its own, so that `data` is + // dropped then too. + self.then_or_catch(lock, data, |_, _, value| Ok(value), on_rejected) } /// Registers `on_fulfilled` and `on_rejected` to run when the promise settles, like @@ -330,6 +350,7 @@ impl<'s> Local<'s, Promise> { { self.debug_assert_isolate(lock); let _ = (on_fulfilled, on_rejected); + let data = ReactionData(RefCell::new(Some(data))); let [on_fulfilled, on_rejected] = Function::new_sharing_data(lock, data, [reaction::(), reaction::()])?; // SAFETY: lock guarantees the isolate is locked and a HandleScope is active; all three diff --git a/src/rust/jsg/v8.rs b/src/rust/jsg/v8.rs index 27474385e6d..a5ac5c35ef6 100644 --- a/src/rust/jsg/v8.rs +++ b/src/rust/jsg/v8.rs @@ -313,6 +313,7 @@ pub mod ffi { pub unsafe fn local_is_big_int(value: &Local) -> bool; pub unsafe fn local_is_date(value: &Local) -> bool; pub unsafe fn local_type_of(isolate: *mut Isolate, value: &Local) -> String; + pub unsafe fn local_to_detail_string(isolate: *mut Isolate, value: &Local) -> MaybeLocal; // Local pub unsafe fn local_string_length(value: &Local) -> i32; @@ -428,16 +429,6 @@ pub mod ffi { pub unsafe fn local_promise_state(promise: &Local) -> PromiseState; pub unsafe fn local_promise_result(promise: &Local) -> Local; pub unsafe fn local_promise_mark_as_handled(promise: &Local); - pub unsafe fn local_promise_then( - isolate: *mut Isolate, - promise: &Local, - on_fulfilled: &Local, - ) -> MaybeLocal; - pub unsafe fn local_promise_catch( - isolate: *mut Isolate, - promise: &Local, - on_rejected: &Local, - ) -> MaybeLocal; pub unsafe fn local_promise_then_or_catch( isolate: *mut Isolate, promise: &Local, @@ -1608,6 +1599,31 @@ impl<'s, T> Local<'s, T> { // HandleScope. unsafe { ffi::local_type_of(lock.isolate().as_ffi(), self.raw()) } } + + /// Describes the value for a log or a trace, as C++ `kj::str()` of a `v8::Local` does: an + /// error as `"TypeError: message"`, a string as itself, and a plain object as + /// `"[object Object]"`. Unlike `String(value)`, it runs no JavaScript, such as a `toString()` + /// method. + /// + /// Corresponds to `v8::Value::ToDetailString()`. + /// + /// # Errors + /// + /// If V8 fails, which only happens if execution is terminating, the error is one for which + /// [`Error::is_pending`] is true. + pub fn to_detail_string(&self, lock: &mut Lock<'s>) -> crate::Result { + let detail: Local<'s, String> = call_into_js(lock, |lock| { + // SAFETY: `lock` proves the isolate is locked, with a context entered and a + // HandleScope for `'s` open; a non-empty result is a string V8 created in that scope. + Ok(unsafe { + Local::from_ffi_maybe(ffi::local_to_detail_string( + lock.isolate().as_ffi(), + self.raw(), + )) + }) + })?; + Ok(detail.to_string(lock)) + } } impl PartialEq for Local<'_, Value> { diff --git a/src/rust/memory-cache/ffi/BUILD.bazel b/src/rust/memory-cache/ffi/BUILD.bazel index 259d5dcfb09..f7f863c0a47 100644 --- a/src/rust/memory-cache/ffi/BUILD.bazel +++ b/src/rust/memory-cache/ffi/BUILD.bazel @@ -6,6 +6,7 @@ wd_rust_crate( "@capnp-cpp//src/kj:kj-async-io", ], cxx_bridge_src = "lib.rs", + cxx_bridge_visibility = ["//visibility:public"], visibility = ["//visibility:public"], deps = [ "//src/rust/memory-cache", diff --git a/src/rust/memory-cache/ffi/lib.rs b/src/rust/memory-cache/ffi/lib.rs index 3d61179d4a3..b65c35ec851 100644 --- a/src/rust/memory-cache/ffi/lib.rs +++ b/src/rust/memory-cache/ffi/lib.rs @@ -139,10 +139,16 @@ struct Namespace { inner: memory_cache::Namespace, } -struct Binding { +pub struct Binding { inner: memory_cache::Binding, } +impl Binding { + pub fn into_inner(self) -> memory_cache::Binding { + self.inner + } +} + struct ReadDecision { inner: memory_cache::ReadDecision, } diff --git a/src/workerd/api/BUILD.bazel b/src/workerd/api/BUILD.bazel index 84205ee0fa5..801e06df602 100644 --- a/src/workerd/api/BUILD.bazel +++ b/src/workerd/api/BUILD.bazel @@ -366,6 +366,7 @@ wd_cc_library( ], visibility = ["//visibility:public"], deps = [ + "//src/rust/cxx", "//src/workerd/io:compatibility-date_capnp", "//src/workerd/jsg", ], @@ -669,6 +670,9 @@ kj_test( src = "memory-cache-test.c++", deps = [ ":memory-cache", + "//src/rust/jsg:ffi", + "//src/workerd/api/memory-cache-api", + "//src/workerd/api/memory-cache-api:ffi.rs@cxx", "//src/workerd/io:trace", "//src/workerd/tests:test-fixture", "//src/workerd/util:thread-scopes", diff --git a/src/workerd/api/memory-cache-api/BUILD.bazel b/src/workerd/api/memory-cache-api/BUILD.bazel new file mode 100644 index 00000000000..279612ecb69 --- /dev/null +++ b/src/workerd/api/memory-cache-api/BUILD.bazel @@ -0,0 +1,24 @@ +load("//:build/wd_rust_crate.bzl", "wd_rust_crate") + +# MemoryCache, the JavaScript API of the memory cache, in Rust. C++ creates it for a memory cache +# binding when the RUST_MEMORY_CACHE autogate is on; memory-cache.c++ otherwise. +wd_rust_crate( + name = "memory-cache-api", + cxx_bridge_deps = [ + "//src/rust/jsg", + "//src/rust/memory-cache/ffi:lib.rs@cxx", + "//src/workerd/util:thread-scopes", + ], + cxx_bridge_hdrs = [], + cxx_bridge_src = "ffi.rs", + cxx_bridge_visibility = ["//visibility:public"], + proc_macro_deps = ["//src/rust/jsg-macros"], + visibility = ["//visibility:public"], + deps = [ + "//src/rust/jsg", + "//src/rust/memory-cache", + "//src/rust/memory-cache/ffi:memory-cache-ffi", + "//src/workerd/io/io-context", + "@crates_vendor//:tracing", + ], +) diff --git a/src/workerd/api/memory-cache-api/fallback.rs b/src/workerd/api/memory-cache-api/fallback.rs new file mode 100644 index 00000000000..90b0c535043 --- /dev/null +++ b/src/workerd/api/memory-cache-api/fallback.rs @@ -0,0 +1,341 @@ +// Copyright (c) 2026 Cloudflare, Inc. +// Licensed under the Apache 2.0 license found in the LICENSE file or at: +// https://opensource.org/licenses/Apache-2.0 + +//! A read that waits for the cache: as the leader, until it runs its fallback, or as a waiter, +//! until the read it waits for finishes. + +use std::cell::RefCell; + +use io_context::IoContext; +use io_context::IoOwned; +use io_context::SpanBuilder; +use jsg::FromJS; +use jsg::v8; +use memory_cache::CacheError; +use memory_cache::FallbackPermit; + +use crate::cache_error; +use crate::now_ms; +use crate::trace; +use crate::trace::ReadSpans; + +/// What a read needs to finish once the cache lets it. `IoContext::await_io()` holds it outside +/// the garbage collector's view, so its handles are strong. +pub struct PendingRead { + resolver: v8::Global, + fallback: v8::Global, + /// The key as C++ holds it: UTF-8, with lone surrogates encoded as such. + key: Vec, + spans: ReadSpans, +} + +impl PendingRead { + /// A pending read, and the promise that it settles. + pub fn new<'s>( + lock: &mut jsg::Lock<'s>, + fallback: v8::Local<'s, v8::Function>, + key: Vec, + spans: ReadSpans, + ) -> jsg::Result<(Self, v8::Local<'s, v8::Promise>)> { + let resolver = v8::PromiseResolver::new(lock)?; + let promise = resolver.promise(); + let pending = Self { + resolver: resolver.to_global(lock), + fallback: fallback.to_global(lock), + key, + spans, + }; + Ok((pending, promise)) + } + + /// Resolves the read with the value another read cached. + pub fn resolve_with( + mut self, + lock: &mut jsg::Lock<'_>, + value: Result, + ) { + let resolver = self.resolver.as_local(lock); + let value = match value { + Ok(value) => value, + Err(err) => return reject(lock, resolver, &cache_error(err)), + }; + self.spans.read.set_tag(c"fallback_cache_hit", true); + self.spans + .read + .set_tag(c"entry_size", trace::count(value.bytes().len())); + match lock.try_catch(|lock| jsg::ser::deserialize(lock, value.bytes())) { + Ok(Ok(value)) => { + let _ = resolver.resolve(lock, value); + } + Ok(Err(exception)) => { + let _ = resolver.reject(lock, exception); + } + // Execution is terminating. + Err(_) => {} + } + } + + /// Rejects the read with `err`. + pub fn reject(self, lock: &mut jsg::Lock<'_>, err: &jsg::Error) { + let resolver = self.resolver.as_local(lock); + reject(lock, resolver, err); + } + + /// Runs the read's fallback, holding `permit` until it produces a value or fails, or the + /// request ends. + pub fn run_fallback(self, lock: &mut jsg::Lock<'_>, permit: FallbackPermit) { + let Self { + resolver, + fallback, + key, + // The user span ends when the fallback starts, as memory-cache.c++'s does. + spans: ReadSpans { read, user: _user }, + } = self; + let resolver = resolver.as_local(lock); + // As memory-cache.c++ passes it, decoded from the bytes it holds, so that a lone + // surrogate reaches the fallback as U+FFFD. + let Some(key_js) = v8::String::new_from_utf8(lock, &key) else { + let err = jsg::Error::new_internal_error("the memory cache key is too long"); + return reject(lock, resolver, &err); + }; + let fill = match Fill::new(lock, permit, &key, read) { + Ok(fill) => fill, + Err(err) => return reject(lock, resolver, &err), + }; + let fallback = fallback.as_local(lock); + match lock + .try_catch(|lock| fallback.call::>(lock, None, &[key_js.into()])) + { + Ok(Ok(produced)) => fill.attach(lock, resolver, produced), + // A fallback that throws rejects the read, as one whose promise rejects does. + Ok(Err(exception)) => fill.reject_now(lock, resolver, exception), + // Execution is terminating. + Err(_) => {} + } + } +} + +fn reject<'s>( + lock: &mut jsg::Lock<'s>, + resolver: v8::Local<'s, v8::PromiseResolver>, + err: &jsg::Error, +) { + // A pending error is a termination, which leaves the read pending. + if !err.is_pending() { + let _ = resolver.reject_with_error(lock, err); + } +} + +/// What the reactions to a fallback's promise hold. +struct Fill { + permit: IoOwned, + /// The key, for the span of the write. + key: String, + spans: RefCell, +} + +/// The spans that end once the fallback settles, when the reactions to its promise drop the +/// [`Fill`] that holds them, as in memory-cache.c++. +struct FillSpans { + _read: SpanBuilder, + fallback: SpanBuilder, +} + +impl jsg::Traced for Fill {} + +impl Fill { + /// Gives the request `permit`, so that it is dropped with the request: a fallback that never + /// settles then does not keep the reads waiting for it until the garbage collector drops its + /// reactions. Starts the span of the fallback. + fn new( + lock: &mut jsg::Lock<'_>, + permit: FallbackPermit, + key: &[u8], + mut read_span: SpanBuilder, + ) -> jsg::Result { + let key = String::from_utf8_lossy(key).into_owned(); + let mut fallback_span = read_span.new_child(c"memory_cache_fallback"); + fallback_span.set_tag(c"key", &key); + Ok(Self { + permit: IoContext::current(lock)?.add_object(permit)?, + key, + spans: RefCell::new(FillSpans { + _read: read_span, + fallback: fallback_span, + }), + }) + } + + /// Settles the read as the promise of what the fallback produced settles. + fn attach<'s>( + self, + lock: &mut jsg::Lock<'s>, + resolver: v8::Local<'s, v8::PromiseResolver>, + produced: v8::Local<'s, v8::Value>, + ) { + let filled = v8::Promise::resolved(lock, produced) + .and_then(|produced| produced.then_or_catch(lock, self, fulfilled, rejected)); + match filled { + Ok(filled) => { + let _ = resolver.resolve(lock, filled.into()); + } + Err(err) => reject(lock, resolver, &err), + } + } + + /// Fails as [`Fill::fail`] does, for a fallback that threw, and rejects the read with what + /// it threw. + fn reject_now<'s>( + self, + lock: &mut jsg::Lock<'s>, + resolver: v8::Local<'s, v8::PromiseResolver>, + exception: v8::Local<'s, v8::Value>, + ) { + if self.fail(lock, exception).is_ok() { + let _ = resolver.reject(lock, exception); + } + } + + fn tag_fallback<'v>( + &self, + key: &'static std::ffi::CStr, + value: impl Into>, + ) { + self.spans.borrow_mut().fallback.set_tag(key, value); + } + + /// Abandons the fallback, so that a read waiting for it runs its own, and records why. + /// Fails only if execution is terminating. + fn fail<'a>( + &self, + lock: &mut jsg::Lock<'a>, + exception: v8::Local<'a, v8::Value>, + ) -> jsg::Result<()> { + drop(self.permit.take()); + self.tag_fallback(c"fallback_success", false); + self.tag_fallback(c"fallback_error", &exception.to_detail_string(lock)?); + Ok(()) + } + + /// Fails as [`Fill::fail`] does, and rejects with `exception`. + fn reject<'a>( + &self, + lock: &mut jsg::Lock<'a>, + exception: v8::Local<'a, v8::Value>, + ) -> jsg::Result> { + self.fail(lock, exception)?; + Ok(v8::Promise::rejected(lock, exception)?.into()) + } + + /// Caches what the fallback produced, and records the write. Like memory-cache.c++, which + /// only logs it, a failure to do either still passes the value to the reads waiting for it. + fn succeed(&self, lock: &mut jsg::Lock<'_>, bytes: Vec, expiration: Option) { + let Some(permit) = self.permit.take() else { + return; + }; + let result = IoContext::current(lock).and_then(|context| { + let now = now_ms(&context)?; + let trace = permit + .succeed(bytes, expiration, now) + .map_err(cache_error)?; + trace::record_write(&context, &self.key, &trace)?; + self.tag_fallback(c"waiters_notified", trace::count(trace.waiters_notified)); + Ok(()) + }); + if let Err(err) = result { + tracing::error!(error = %err, "memory cache fallback completion failed"); + } + } +} + +/// Caches what the fallback produced, and fulfills the read with its value. If the value cannot +/// be cached, abandons the fallback and rejects the read. Tags the fallback's span as +/// memory-cache.c++ does, so that a failure leaves the same tags. +fn fulfilled<'a>( + lock: &mut jsg::Lock<'a>, + fill: &Fill, + result: v8::Local<'a, v8::Value>, +) -> jsg::Result> { + let (value, expiration) = match lock.try_catch(|lock| read_result(lock, result))? { + Ok(fields) => fields, + Err(exception) => return fill.reject(lock, exception), + }; + let bytes = match serialize(lock, value)? { + Ok(bytes) => bytes, + Err(exception) => return fill.reject(lock, exception), + }; + fill.tag_fallback(c"fallback_result_size", trace::count(bytes.len())); + if expiration.is_some_and(f64::is_nan) { + let err = jsg::Error::new_type_error("Expiration time must not be NaN."); + let exception = lock.exception(&err)?; + return fill.reject(lock, exception); + } + fill.tag_fallback(c"has_expiration", expiration.is_some()); + // Only now, so that a failure above does not overwrite it. + fill.tag_fallback(c"fallback_success", true); + fill.succeed(lock, bytes, expiration); + Ok(value) +} + +/// Abandons the fallback, and rejects the read with what the fallback's promise rejected with. +fn rejected<'a>( + lock: &mut jsg::Lock<'a>, + fill: &Fill, + reason: v8::Local<'a, v8::Value>, +) -> jsg::Result> { + fill.reject(lock, reason) +} + +/// Serializes the value a fallback produced. Fails only if execution is terminating; otherwise +/// an exception is what the read rejects with. +fn serialize<'a>( + lock: &mut jsg::Lock<'a>, + value: v8::Local<'a, v8::Value>, +) -> jsg::Result, v8::Local<'a, v8::Value>>> { + let exception = match lock.try_catch(|lock| jsg::ser::serialize(lock, value))? { + Ok(bytes) => return Ok(Ok(bytes)), + Err(exception) => exception, + }; + if !is_data_clone_error(lock, exception)? { + return Ok(Err(exception)); + } + // As memory-cache.c++, whose DataCloneError says only what type failed. + let err = + jsg::Error::new_data_clone_error(format!("failed to serialize {}", value.type_of(lock))); + Ok(Err(lock.exception(&err)?)) +} + +/// The `value` and `expiration` of what a fallback produced. +fn read_result<'a>( + lock: &mut jsg::Lock<'a>, + result: v8::Local<'a, v8::Value>, +) -> jsg::Result<(v8::Local<'a, v8::Value>, Option)> { + let result = v8::Local::::try_from(result).map_err(|_| { + jsg::Error::new_type_error("The fallback must produce an object with a value property.") + })?; + let value = result.get(lock, "value")?; + let expiration = result.get(lock, "expiration")?; + let expiration = if expiration.is_undefined() { + None + } else { + Some(jsg::Number::from_js(lock, expiration)?.value()) + }; + Ok((value, expiration)) +} + +/// Whether the serializer threw a `DataCloneError`, rather than passed on what a getter threw. +fn is_data_clone_error<'a>( + lock: &mut jsg::Lock<'a>, + exception: v8::Local<'a, v8::Value>, +) -> jsg::Result { + let Ok(exception) = v8::Local::::try_from(exception) else { + return Ok(false); + }; + // What a getter threw may run JavaScript when read, and throw. + let Ok(name) = lock.try_catch(|lock| exception.get(lock, "name"))? else { + return Ok(false); + }; + Ok(v8::Local::::try_from(name) + .is_ok_and(|name| name.to_string(lock) == "DataCloneError")) +} diff --git a/src/workerd/api/memory-cache-api/ffi.rs b/src/workerd/api/memory-cache-api/ffi.rs new file mode 100644 index 00000000000..a46df4017a7 --- /dev/null +++ b/src/workerd/api/memory-cache-api/ffi.rs @@ -0,0 +1,67 @@ +// Copyright (c) 2026 Cloudflare, Inc. +// Licensed under the Apache 2.0 license found in the LICENSE file or at: +// https://opensource.org/licenses/Apache-2.0 + +//! The bridge through which C++ creates a [`MemoryCache`](crate::MemoryCache), and the C++ it +//! calls. + +#![allow( + unsafe_code, + reason = "the cxx bridge expands to unsafe FFI glue, and memory_cache_new() takes C++'s raw \ + isolate" +)] + +pub use bridge::is_predictable_mode_for_test; +use jsg::ToJS; +use jsg::v8; + +#[cxx::bridge(namespace = "workerd::api")] +mod bridge { + #[namespace = "workerd::rust::jsg"] + unsafe extern "C++" { + include!("workerd/rust/jsg/ffi.h"); + include!("workerd/rust/jsg/v8.rs.h"); + + type Isolate = jsg::v8::ffi::Isolate; + type MaybeLocal = jsg::v8::ffi::MaybeLocal; + } + + #[namespace = "workerd::rust::memory_cache"] + unsafe extern "C++" { + include!("workerd/rust/memory-cache/ffi/lib.rs.h"); + } + + #[namespace = "workerd"] + unsafe extern "C++" { + include!("workerd/util/thread-scopes.h"); + + /// Whether the process runs in predictable mode, for tests. + #[cxx_name = "isPredictableModeForTest"] + fn is_predictable_mode_for_test() -> bool; + } + + #[namespace = "workerd::rust::memory_cache"] + extern "Rust" { + type Binding = memory_cache_ffi::Binding; + } + + extern "Rust" { + /// A new `MemoryCache` for `binding`. + unsafe fn memory_cache_new(isolate: *mut Isolate, binding: Box) -> MaybeLocal; + } +} + +/// # Safety +/// C++ calls this with the isolate locked, a context entered and a `HandleScope` open. +unsafe fn memory_cache_new( + isolate: *mut bridge::Isolate, + binding: Box, +) -> bridge::MaybeLocal { + // SAFETY: forwarded from the caller. + unsafe { + v8::run_ffi_callback(isolate, |lock| { + let cache = jsg::Rc::new(crate::MemoryCache::new(binding.into_inner())); + cache.to_js(lock) + }) + } +} diff --git a/src/workerd/api/memory-cache-api/lib.rs b/src/workerd/api/memory-cache-api/lib.rs new file mode 100644 index 00000000000..a5e2b329cef --- /dev/null +++ b/src/workerd/api/memory-cache-api/lib.rs @@ -0,0 +1,264 @@ +// Copyright (c) 2026 Cloudflare, Inc. +// Licensed under the Apache 2.0 license found in the LICENSE file or at: +// https://opensource.org/licenses/Apache-2.0 + +//! `MemoryCache`, the JavaScript API of a memory cache binding, over the cache in +//! `src/rust/memory-cache`. It does what `src/workerd/api/memory-cache.c++` does, and records +//! the same trace spans. + +mod fallback; +mod ffi; +mod trace; + +use std::time::SystemTime; +use std::time::UNIX_EPOCH; + +use io_context::IoContext; +use jsg::Member; +use jsg::Resource; +use jsg::feature_flags::compatibility_flags; +use jsg::v8; +use jsg_macros::jsg_method; +use jsg_macros::jsg_resource; +use memory_cache::CacheError; +use memory_cache::ReadDecision; +use memory_cache::ReadKind; +use memory_cache::ReadMode; +use memory_cache::WaitKind; + +use crate::fallback::PendingRead; +use crate::trace::ReadSpans; + +/// The longest key, in UTF-8 bytes, as in memory-cache.c++. +const MAX_KEY_SIZE: usize = 2 * 1024; + +#[jsg_resource(name = "MemoryCache")] +pub struct MemoryCache { + binding: CacheBinding, +} + +impl MemoryCache { + fn new(binding: memory_cache::Binding) -> Self { + Self { + binding: CacheBinding(binding), + } + } +} + +#[jsg_resource(custom_members)] +impl MemoryCache { + /// `read(key, fallback?)`: a promise for the value cached under `key`. On a miss, without + /// `fallback` it is `undefined`. With it, one read of all those that miss calls + /// `fallback(key)`, caches what it produces and passes it to the others, which wait for it. + /// If it fails, that read rejects with what it threw, and one of the others calls its own + /// fallback. + #[jsg_method] + fn read<'s>( + &self, + lock: &mut jsg::Lock<'s>, + key: v8::Local<'s, v8::Value>, + fallback: Option>, + ) -> jsg::Result> { + // As a C++ method that returns a promise, it rejects rather than throws. + match self.read_promise(lock, key, fallback) { + Err(err) if !err.is_pending() => v8::Promise::rejected_with_error(lock, &err), + result => result, + } + } + + /// `delete(key)`: removes the value cached under `key`, if any. + #[jsg_method(name = "delete")] + fn remove<'s>( + &self, + lock: &mut jsg::Lock<'s>, + key: v8::Local<'s, v8::Value>, + ) -> jsg::Result<()> { + let key = key_bytes(lock, key)?; + // Keys that are too large are never cached. + if key.len() > MAX_KEY_SIZE { + return Err(jsg::Error::new_range_error("Key too large.")); + } + self.binding.0.delete(&key); + Ok(()) + } +} + +impl Resource for MemoryCache { + fn members(flags: compatibility_flags::Reader<'_>) -> Vec { + let mut members = vec![Self::read_member()]; + if flags.get_memory_cache_delete() { + members.push(Self::remove_member()); + } + members + } +} + +impl MemoryCache { + fn read_promise<'s>( + &self, + lock: &mut jsg::Lock<'s>, + key: v8::Local<'s, v8::Value>, + fallback: Option>, + ) -> jsg::Result> { + let key = key_bytes(lock, key)?; + // JSG checks the parameters of C++'s read() before it runs, and starts any span. + let fallback = fallback + .map(|fallback| { + v8::Local::::try_from(fallback).map_err(|_| { + jsg::Error::new_type_error( + "Failed to execute 'read' on 'MemoryCache': parameter 2 is not of type \ + 'Function'.", + ) + }) + }) + .transpose()?; + if key.len() > MAX_KEY_SIZE { + return Err(jsg::Error::new_range_error("Key too large.")); + } + let io_context = IoContext::current(lock)?; + let mut spans = ReadSpans::start(&io_context)?; + let now = now_ms(&io_context)?; + + let Some(fallback) = fallback else { + let decision = self + .binding + .0 + .read(&key, now, ReadMode::CacheOnly) + .map_err(cache_error)?; + trace::tag_read(&mut spans.read, &decision.trace()); + return read_cached(lock, decision); + }; + let decision = self + .binding + .0 + .read(&key, now, ReadMode::WithFallback) + .map_err(cache_error)?; + read_with_fallback(lock, &io_context, decision, fallback, key, spans) + } +} + +/// A read without a fallback: the cached value, or `undefined`. +fn read_cached<'s>( + lock: &mut jsg::Lock<'s>, + mut decision: ReadDecision, +) -> jsg::Result> { + let value = match decision.kind() { + ReadKind::Value => { + let value = decision.take_value().map_err(cache_error)?; + jsg::ser::deserialize(lock, value.bytes())? + } + _ => lock.undefined(), + }; + v8::Promise::resolved(lock, value) +} + +/// A read with a fallback: the cached value, or the value the leader of the reads that miss +/// produces with its fallback. +fn read_with_fallback<'s>( + lock: &mut jsg::Lock<'s>, + io_context: &IoContext<'s>, + mut decision: ReadDecision, + fallback: v8::Local<'s, v8::Function>, + key: Vec, + mut spans: ReadSpans, +) -> jsg::Result> { + let read_trace = decision.trace(); + match decision.kind() { + // Served without releasing the isolate lock. + ReadKind::Value => { + trace::tag_read(&mut spans.read, &read_trace); + let value = decision.take_value().map_err(cache_error)?; + let value = jsg::ser::deserialize(lock, value.bytes())?; + v8::Promise::resolved(lock, value) + } + ReadKind::Leader => { + trace::tag_leader(&mut spans.read, &read_trace); + let permit = decision.take_permit().map_err(cache_error)?; + let (pending, promise) = PendingRead::new(lock, fallback, key, spans)?; + // As in C++, the fallback runs in a later turn, not while `read` is running. + io_context.await_io(async move { permit }, move |lock, permit| { + pending.run_fallback(lock, permit); + })?; + Ok(promise) + } + ReadKind::Waiter => { + let wait_span = + trace::tag_waiter(&mut spans.read, &read_trace, &String::from_utf8_lossy(&key)); + let waiter = decision.take_waiter().map_err(cache_error)?; + let (pending, promise) = PendingRead::new(lock, fallback, key, spans)?; + io_context.await_io(waiter.wait(), move |lock, mut outcome| { + // The span of the wait ends with it. It cannot go in the future, which is `Send`, + // so it ends when the callback runs, just after the wait completes. + drop(wait_span); + match outcome.kind() { + // The read it waited for cached a value, or failed to but produced one. + WaitKind::Value => pending.resolve_with(lock, outcome.take_value()), + // The read it waited for failed, and this one is next. + WaitKind::Leader => match outcome.take_permit() { + Ok(permit) => pending.run_fallback(lock, permit), + Err(err) => pending.reject(lock, &cache_error(err)), + }, + } + })?; + Ok(promise) + } + ReadKind::Miss => Err(jsg::Error::new_internal_error( + "the memory cache missed without a fallback read", + )), + } +} + +/// The cache binding. Its empty `Traced` impl is correct: the cache holds serialized bytes, +/// never JavaScript values, so the binding refers to nothing in the V8 heap. +struct CacheBinding(memory_cache::Binding); + +impl jsg::Traced for CacheBinding {} + +impl Drop for CacheBinding { + fn drop(&mut self) { + // memory-cache.c++ uses the request's clock if a request is current, and the system + // clock otherwise. A drop has no lock to find the request with, so this always uses the + // system clock. The clock matters only if releasing the binding shrinks the cache, to + // tell which entries have expired, and a request's clock trails the system clock only + // by the time since the request last waited for I/O. + self.0.release(system_now_ms()); + } +} + +/// `key` as UTF-8, with lone surrogates encoded as such, as C++ `kj::String` holds them, so that +/// keys differing only in them are distinct. +fn key_bytes<'s>(lock: &mut jsg::Lock<'s>, key: v8::Local<'s, v8::Value>) -> jsg::Result> { + let key = v8::Local::::try_from(key).map_err(|_| { + jsg::Error::new_type_error( + "Failed to execute 'read' on 'MemoryCache': parameter 1 is not of type 'string'.", + ) + })?; + let mut bytes = vec![0; key.utf8_length(lock)]; + let written = key.write_utf8(lock, &mut bytes, v8::WriteFlags::None); + bytes.truncate(written); + Ok(bytes) +} + +/// The request's clock in whole milliseconds since the Unix epoch, as C++ `dateNow()`. +fn now_ms(io_context: &IoContext<'_>) -> jsg::Result { + Ok(millis_since_epoch(io_context.now()?)) +} + +/// The system clock, as memory-cache.c++ uses outside a request. +fn system_now_ms() -> f64 { + millis_since_epoch(SystemTime::now()) +} + +#[expect( + clippy::cast_precision_loss, + reason = "milliseconds since the epoch fit in an f64 exactly" +)] +fn millis_since_epoch(time: SystemTime) -> f64 { + time.duration_since(UNIX_EPOCH) + .map_or(0.0, |since| since.as_millis() as f64) +} + +/// A cache error, which only a bug causes, so JavaScript sees it as an internal error. +fn cache_error(err: CacheError) -> jsg::Error { + jsg::Error::new_internal_error(format!("memory cache error: {err}")) +} diff --git a/src/workerd/api/memory-cache-api/trace.rs b/src/workerd/api/memory-cache-api/trace.rs new file mode 100644 index 00000000000..8ab085db86d --- /dev/null +++ b/src/workerd/api/memory-cache-api/trace.rs @@ -0,0 +1,144 @@ +// Copyright (c) 2026 Cloudflare, Inc. +// Licensed under the Apache 2.0 license found in the LICENSE file or at: +// https://opensource.org/licenses/Apache-2.0 + +//! The trace spans of a read, with the names and tags memory-cache.c++ records. + +use io_context::IoContext; +use io_context::SpanBuilder; +use io_context::TraceContext; +use memory_cache::EvictionReason; +use memory_cache::ReadTrace; +use memory_cache::WriteOutcome; +use memory_cache::WriteTrace; + +use crate::ffi::is_predictable_mode_for_test; + +/// The spans of one read, which end when it no longer needs them. +pub struct ReadSpans { + /// The span of internal tracing, which the read's tags go on. + pub read: SpanBuilder, + /// The span of user tracing, which has no tags. + pub user: TraceContext, +} + +impl ReadSpans { + pub fn start(io_context: &IoContext<'_>) -> jsg::Result { + Ok(Self { + read: io_context.make_trace_span(c"memory_cache_read")?, + user: io_context.make_user_trace_span(c"memory_cache_read")?, + }) + } +} + +/// Tags a read that the cache answered, with a value or a miss. +pub fn tag_read(span: &mut SpanBuilder, trace: &ReadTrace) { + span.set_tag( + c"memory_cache_lock_wait_time_ns", + lock_wait_ns(trace.lock_wait_ns), + ); + span.set_tag(c"cache_hit", trace.cache_hit); + if trace.cache_hit { + span.set_tag(c"entry_size", count(trace.entry_size)); + } + tag_cache_size(span, trace); +} + +/// Tags a read that missed and runs its fallback. +pub fn tag_leader(span: &mut SpanBuilder, trace: &ReadTrace) { + span.set_tag( + c"memory_cache_lock_wait_time_ns", + lock_wait_ns(trace.lock_wait_ns), + ); + span.set_tag(c"cache_hit", false); + span.set_tag(c"coalesced_request", false); + span.set_tag(c"initiating_fallback", true); + tag_cache_size(span, trace); +} + +/// Tags a read that missed and waits for another's fallback, and starts the span of the wait. +pub fn tag_waiter(span: &mut SpanBuilder, trace: &ReadTrace, key: &str) -> SpanBuilder { + span.set_tag( + c"memory_cache_lock_wait_time_ns", + lock_wait_ns(trace.lock_wait_ns), + ); + span.set_tag(c"cache_hit", false); + span.set_tag(c"coalesced_request", true); + span.set_tag(c"waiting_on_inflight", true); + span.set_tag( + c"inflight_waiters_count", + count(trace.waiters_ahead.saturating_add(1)), + ); + let mut wait = span.new_child(c"memory_cache_coalesce_wait"); + wait.set_tag(c"key", key); + wait.set_tag(c"waiters_ahead", count(trace.waiters_ahead)); + wait +} + +fn tag_cache_size(span: &mut SpanBuilder, trace: &ReadTrace) { + span.set_tag(c"cache_total_size", count(trace.total_value_size)); + span.set_tag(c"cache_entry_count", count(trace.entry_count)); +} + +/// Records the write of what a fallback produced, and the evictions it caused, as spans of +/// their own. +pub fn record_write(io_context: &IoContext<'_>, key: &str, trace: &WriteTrace) -> jsg::Result<()> { + let mut span = io_context.make_trace_span(c"memory_cache_write")?; + span.set_tag(c"key", key); + span.set_tag(c"value_size", count(trace.value_size)); + span.set_tag(c"has_expiration", trace.has_expiration); + match trace.outcome { + WriteOutcome::Success { + is_update, + total_after, + entries_after, + } => { + span.set_tag(c"write_success", true); + span.set_tag(c"is_update", is_update); + span.set_tag(c"evictions_triggered", count(trace.evictions.len())); + span.set_tag(c"cache_total_size_after", count(total_after)); + span.set_tag(c"cache_entry_count_after", count(entries_after)); + } + WriteOutcome::ValueTooLarge { max_value_size } => { + span.set_tag(c"write_rejected", true); + span.set_tag(c"rejection_reason", c"value_too_large"); + span.set_tag(c"max_value_size", count(max_value_size)); + } + WriteOutcome::AlreadyExpired => { + span.set_tag(c"write_rejected", true); + span.set_tag(c"rejection_reason", c"already_expired"); + } + } + + for eviction in &trace.evictions { + let mut span = io_context.make_trace_span(c"memory_cache_eviction")?; + let reason = match eviction.reason { + EvictionReason::Expiration => c"expiration", + EvictionReason::Lru => c"lru", + }; + span.set_tag(c"eviction_reason", reason); + span.set_tag(c"evicted_key", &*String::from_utf8_lossy(&eviction.key)); + span.set_tag(c"evicted_size", count(eviction.value_size)); + span.set_tag(c"cache_size_before", count(eviction.total_before)); + span.set_tag(c"cache_entries_before", count(eviction.entries_before)); + } + Ok(()) +} + +/// The time a read waited for the cache's lock, which is 0 in tests, so that traces are +/// deterministic. +fn lock_wait_ns(ns: u64) -> i64 { + if is_predictable_mode_for_test() { + return 0; + } + i64::try_from(ns).unwrap_or(i64::MAX) +} + +/// A size or count as a tag, which memory-cache.c++ records as a float. +#[expect( + clippy::cast_precision_loss, + reason = "as in memory-cache.c++; sizes and counts of a memory cache are far below 2^53" +)] +pub fn count(n: usize) -> f64 { + n as f64 +} diff --git a/src/workerd/api/memory-cache-test.c++ b/src/workerd/api/memory-cache-test.c++ index 4c160f167b8..3a6891c5fe9 100644 --- a/src/workerd/api/memory-cache-test.c++ +++ b/src/workerd/api/memory-cache-test.c++ @@ -1,10 +1,13 @@ // Tests for the C++ side of the memory cache: binding lifetime, singleflight -// leader/waiter handoff, and the FallbackDoneCallback contract. The storage and -// coordination logic itself is covered by the Rust crate's own tests. +// leader/waiter handoff, the FallbackDoneCallback contract, and the trace spans of +// MemoryCache.read(), which the Rust MemoryCache must record as the C++ one does. The +// storage and coordination logic itself is covered by the Rust crate's own tests. #include "memory-cache.h" +#include #include +#include #include #include @@ -212,8 +215,9 @@ KJ_TEST("fallback callback stores the value and survives binding destruction") { }); } -// Records the spans of a request as they end, each as "name <- parent: key=type:value ...", with -// its tags in key order. +// Records the spans of a request in the order they open, each as "name <- parent: key=type:value +// ...", with its tags, which a span delivers when it ends, in key order. The order spans end in +// depends on when each implementation of MemoryCache releases them, so it is not recorded. class SpanRecorder final: public SpanObserver { public: SpanRecorder(kj::Vector& spans, kj::StringPtr parent) @@ -225,6 +229,8 @@ class SpanRecorder final: public SpanObserver { } void onOpen(kj::ConstString operationName, kj::Date) override { name = kj::str(operationName); + index = spans.size(); + spans.add(kj::str(name, " <- ", parent, ": (open)")); } void onClose(kj::Date, Span::TagMap&& tags, kj::Vector&&) override { auto entries = KJ_MAP(entry, tags) { @@ -245,13 +251,14 @@ class SpanRecorder final: public SpanObserver { KJ_UNREACHABLE; }; std::sort(entries.begin(), entries.end()); - spans.add(kj::str(name, " <- ", parent, ": ", kj::strArray(entries, " "))); + spans[index] = kj::str(name, " <- ", parent, ": ", kj::strArray(entries, " ")); } private: kj::Vector& spans; kj::String parent; kj::String name; + size_t index = 0; }; class SpanRequestObserver final: public RequestObserver { @@ -267,37 +274,60 @@ class SpanRequestObserver final: public RequestObserver { SpanBuilder span; }; -KJ_TEST("a fallback whose result is rejected is tagged as failed, without overwriting a tag") { - // In predictable mode, overwriting a span's tag logs an error, which fails this test. It logs - // only the first overwrite in the process, so no other test here may overwrite one. +// Makes the MemoryCache that a test's script reads as `globalThis.CACHE`. +using MakeCache = kj::Function(const TestFixture::Environment&)>; + +// Runs a request of `script`, which reads `globalThis.CACHE`, and returns the spans it records. +// Runs in predictable mode, in which overwriting a span's tag logs an error, which fails the +// test. It logs only the first overwrite in the process. +kj::Array recordSpans(kj::StringPtr script, MakeCache makeCache) { setPredictableModeForTest(); - MemoryCacheProvider provider; kj::Vector spans; TestFixture fixture({ - .mainModuleSource = R"SCRIPT( - export default { - async fetch() { - const settle = (promise) => promise.then(() => {}, () => {}); - await settle(globalThis.CACHE.read("nan", () => ({ value: 1, expiration: NaN }))); - await settle(globalThis.CACHE.read("function", () => ({ value: () => {} }))); - return new Response("ok"); - } - }; - )SCRIPT"_kj, + .mainModuleSource = script, .requestObserverFactory = kj::Function()>([&spans]() -> kj::Own { return kj::refcounted(spans); }), }); fixture.runInIoContext([&](const TestFixture::Environment& env) { - auto& handler = KJ_REQUIRE_NONNULL(env.js.tryGetTypeHandler>()); - auto cache = - handler.wrap(env.js, env.js.alloc(provider.getUse(kj::none, testLimits()))); - jsg::check( - env.js.v8Context()->Global()->Set(env.js.v8Context(), env.js.str("CACHE"_kj), cache)); + jsg::check(env.js.v8Context()->Global()->Set( + env.js.v8Context(), env.js.str("CACHE"_kj), makeCache(env))); }); auto response = fixture.runRequest(kj::HttpMethod::GET, "http://worker/"_kj, ""_kj); KJ_EXPECT(response.statusCode == 200, response.body); + return spans.releaseAsArray(); +} + +// A C++ MemoryCache over `provider`. +MakeCache cppCache(MemoryCacheProvider& provider, MemoryCacheLimits limits) { + return [&provider, limits](const TestFixture::Environment& env) -> v8::Local { + auto& handler = KJ_REQUIRE_NONNULL(env.js.tryGetTypeHandler>()); + return handler.wrap(env.js, env.js.alloc(provider.getUse(kj::none, limits))); + }; +} + +// A Rust MemoryCache over `provider`. +MakeCache rustCache(MemoryCacheProvider& provider, MemoryCacheLimits limits) { + return [&provider, limits](const TestFixture::Environment& env) -> v8::Local { + return jsg::check(::workerd::rust::jsg::maybe_local_from_ffi( + memory_cache_new(env.isolate, provider.bindForRust(kj::none, limits)))); + }; +} + +KJ_TEST("a fallback whose result is rejected is tagged as failed, without overwriting a tag") { + MemoryCacheProvider provider; + auto spans = recordSpans(R"SCRIPT( + export default { + async fetch() { + const settle = (promise) => promise.then(() => {}, () => {}); + await settle(globalThis.CACHE.read("nan", () => ({ value: 1, expiration: NaN }))); + await settle(globalThis.CACHE.read("function", () => ({ value: () => {} }))); + return new Response("ok"); + } + }; + )SCRIPT"_kj, + cppCache(provider, testLimits())); kj::Vector fallbacks; for (auto& span: spans) { @@ -316,5 +346,85 @@ KJ_TEST("a fallback whose result is rejected is tagged as failed, without overwr fallbacks[1]); } +// Reads that take every path of read(), with and without a fallback, and fill the cache past +// its limits. The fallbacks of a key's reads are coalesced: the second read of "b" waits for the +// first, and the second of "c" runs its own once the first's fails. +// TODO(#7951): Add a value that cannot be serialized, once Rust JSG throws DOMExceptions. Rust +// throws a plain Error, so its fallback_error tag differs from that of C++'s DataCloneError. +constexpr auto PARITY_SCENARIO = R"SCRIPT( + export default { + async fetch() { + const cache = globalThis.CACHE; + const settle = (promise) => promise.then(() => {}, () => {}); + + await cache.read("missing"); + await cache.read("a", async () => ({ value: "x".repeat(10) })); + await cache.read("a"); + await cache.read("a", () => { throw new Error("unused"); }); + + const leader = cache.read("b", async () => ({ value: 1, expiration: Date.now() + 1e6 })); + const waiter = cache.read("b", () => ({ value: 2 })); + await Promise.all([leader, waiter]); + + const failing = cache.read("c", async () => { throw new Error("boom"); }); + const promoted = cache.read("c", () => ({ value: "second" })); + await settle(failing); + await promoted; + + await settle(cache.read("d", () => { throw new TypeError("sync"); })); + await settle(cache.read("e", () => ({ value: 1, expiration: NaN }))); + await settle(cache.read("f", () => Promise.reject("not an error"))); + await settle(cache.read("g", () => ({ value: "y".repeat(100) }))); + await settle(cache.read("h", () => ({ value: 1, expiration: 0 }))); + await cache.read("i", () => ({ value: "evicts a" })); + return new Response("ok"); + } + }; +)SCRIPT"_kj; + +constexpr MemoryCacheLimits PARITY_LIMITS{ + .maxKeys = 3, + .maxValueSize = 64, + .maxTotalValueSize = 1024, +}; + +// The memory_cache_* spans of PARITY_SCENARIO. +kj::Array parityScenarioSpans( + MakeCache (*makeCache)(MemoryCacheProvider&, MemoryCacheLimits)) { + MemoryCacheProvider provider; + kj::Vector result; + for (auto& span: recordSpans(PARITY_SCENARIO, makeCache(provider, PARITY_LIMITS))) { + if (span.startsWith("memory_cache")) result.add(kj::mv(span)); + } + return result.releaseAsArray(); +} + +KJ_TEST("the Rust MemoryCache records the spans the C++ one does") { + auto cpp = parityScenarioSpans(cppCache); + auto rust = parityScenarioSpans(rustCache); + auto describe = [&]() { + return kj::str("C++:\n", kj::strArray(cpp, "\n"), "\nRust:\n", kj::strArray(rust, "\n")); + }; + KJ_EXPECT(cpp.size() == rust.size(), describe()); + for (auto i: kj::zeroTo(kj::min(cpp.size(), rust.size()))) { + KJ_EXPECT(cpp[i] == rust[i], i, describe()); + } + + // The comparison means something only if C++ records what it should. + auto contains = [&](kj::StringPtr expected) { + for (auto& span: cpp) { + if (span == expected) return true; + } + return false; + }; + KJ_EXPECT(contains("memory_cache_read <- request: cache_entry_count=float:0 " + "cache_hit=bool:false cache_total_size=float:0 " + "memory_cache_lock_wait_time_ns=int:0"_kj), + describe()); + KJ_EXPECT(contains("memory_cache_fallback <- memory_cache_read: fallback_error=str:Error: boom " + "fallback_success=bool:false key=str:c"_kj), + describe()); +} + } // namespace } // namespace workerd::api diff --git a/src/workerd/api/memory-cache.c++ b/src/workerd/api/memory-cache.c++ index c46ad433157..38898110e66 100644 --- a/src/workerd/api/memory-cache.c++ +++ b/src/workerd/api/memory-cache.c++ @@ -207,13 +207,17 @@ MemoryCacheProvider::~MemoryCacheProvider() noexcept(false) = default; kj::Own MemoryCacheProvider::getUse( kj::Maybe cacheId, MemoryCacheLimits limits) const { + return kj::heap(bindForRust(cacheId, limits)); +} + +::rust::Box MemoryCacheProvider::bindForRust( + kj::Maybe cacheId, MemoryCacheLimits limits) const { ::rust::Str name; bool isPrivate = cacheId == kj::none; KJ_IF_SOME(id, cacheId) { name = asRustStr(id); } - auto binding = impl->cacheNamespace->bind(name, isPrivate, toRustLimits(limits)); - return kj::heap(kj::mv(binding)); + return impl->cacheNamespace->bind(name, isPrivate, toRustLimits(limits)); } kj::Maybe> MemoryCacheUseImpl::getWithoutFallback( diff --git a/src/workerd/api/memory-cache.h b/src/workerd/api/memory-cache.h index d2e9ac5d600..b88645321d5 100644 --- a/src/workerd/api/memory-cache.h +++ b/src/workerd/api/memory-cache.h @@ -3,10 +3,16 @@ #include #include +#include + namespace workerd { class SpanBuilder; } +namespace workerd::rust::memory_cache { +struct Binding; +} // namespace workerd::rust::memory_cache + namespace workerd::api { // The MemoryCache mechanism is an in-process, memory-resident data cache that @@ -166,6 +172,9 @@ class MemoryCacheProvider { kj::Own getUse(kj::Maybe cacheId, MemoryCacheLimits limits) const; + ::rust::Box<::workerd::rust::memory_cache::Binding> bindForRust( + kj::Maybe cacheId, MemoryCacheLimits limits) const; + private: struct Impl; kj::Own impl; diff --git a/src/workerd/api/tests/memory-cache-test.js b/src/workerd/api/tests/memory-cache-test.js index dc8aa8fea36..891dade2634 100644 --- a/src/workerd/api/tests/memory-cache-test.js +++ b/src/workerd/api/tests/memory-cache-test.js @@ -1,8 +1,10 @@ // Copyright (c) 2024 Cloudflare, Inc. // Licensed under the Apache 2.0 license found in the LICENSE file or at: // https://opensource.org/licenses/Apache-2.0 -import { strictEqual } from 'node:assert'; +import { rejects, strictEqual } from 'node:assert'; +import { DurableObject } from 'cloudflare:workers'; import unsafe from 'workerd:unsafe'; +import { AsyncLocalStorage } from 'node:async_hooks'; export const basic = { async test(ctrl, env) { @@ -332,3 +334,96 @@ export const fallbackNotLocked = { strictEqual(raced, 'bbb'); }, }; + +export default { + // Runs a fallback that never settles, and ends the request once it has started. Its promise + // stays reachable, so that the garbage collector does not drop the reactions to it. + async fetch(request, env) { + const { promise: started, resolve } = Promise.withResolvers(); + env.CACHE.read('abandoned', () => { + resolve(); + globalThis.neverSettles = new Promise(() => {}); + return globalThis.neverSettles; + }); + await started; + return new Response('started'); + }, +}; + +export const fallbackAbandonedWhenRequestEnds = { + async test(ctrl, env) { + const response = await env.SELF.fetch('http://example.com/'); + strictEqual(await response.text(), 'started'); + // The fallback of the request that ended is abandoned with it, rather than when the garbage + // collector drops its promise, so this read runs its own fallback. + const value = await Promise.race([ + env.CACHE.read('abandoned', () => ({ value: 'taken over' })), + scheduler.wait(1000).then(() => 'still waiting'), + ]); + strictEqual(value, 'taken over'); + }, +}; + +export const fallbackRunsInAsyncContextOfRead = { + async test(ctrl, env) { + const storage = new AsyncLocalStorage(); + const value = await storage.run('of the read', () => + env.CACHE.read('asyncContext', () => ({ + value: storage.getStore() ?? 'none', + })) + ); + strictEqual(value, 'of the read'); + }, +}; + +export const fallbackGetsKeyAsCacheHoldsIt = { + async test(ctrl, env) { + // The cache holds keys as UTF-8, with a lone surrogate encoded as such, and passes the key to + // the fallback decoded from it: each of the surrogate's three bytes becomes U+FFFD. + let fallbackKey; + await env.CACHE.read('lone \udbff', (key) => { + fallbackKey = key; + return { value: 'unused' }; + }); + strictEqual(fallbackKey, 'lone \ufffd\ufffd\ufffd'); + env.CACHE.delete('lone \udbff'); + }, +}; + +export const nanExpirationRejected = { + async test(ctrl, env) { + await rejects( + env.CACHE.read('nanExpiration', () => ({ value: 1, expiration: NaN })), + { name: 'TypeError', message: 'Expiration time must not be NaN.' } + ); + // The value is not cached, and the next read runs its own fallback. + strictEqual( + await env.CACHE.read('nanExpiration', () => ({ value: 'next' })), + 'next' + ); + env.CACHE.delete('nanExpiration'); + }, +}; + +export class CacheReader extends DurableObject { + // Reads with a fallback inside blockConcurrencyWhile(). The read resumes inside the critical + // section; if it waited for the input gate instead, which the critical section holds, it would + // never resume. + async fetch() { + const value = await this.ctx.blockConcurrencyWhile(() => + this.env.CACHE.read('criticalSection', () => ({ value: 'in section' })) + ); + return new Response(value); + } +} + +export const fallbackInDurableObjectCriticalSection = { + async test(ctrl, env) { + const reader = env.READER.get(env.READER.idFromName('reader')); + const value = await Promise.race([ + reader.fetch('http://reader/').then((response) => response.text()), + scheduler.wait(5000).then(() => 'never resumed'), + ]); + strictEqual(value, 'in section'); + }, +}; diff --git a/src/workerd/api/tests/memory-cache-test.wd-test b/src/workerd/api/tests/memory-cache-test.wd-test index 70e058e7b66..11900484481 100644 --- a/src/workerd/api/tests/memory-cache-test.wd-test +++ b/src/workerd/api/tests/memory-cache-test.wd-test @@ -8,6 +8,10 @@ const unitTests :Workerd.Config = ( (name = "worker", esModule = embed "memory-cache-test.js") ], compatibilityFlags = ["nodejs_compat", "unsafe_module", "memory_cache_delete"], + durableObjectNamespaces = [ + (className = "CacheReader", uniqueKey = "memory-cache-test-cache-reader"), + ], + durableObjectStorage = (inMemory = void), bindings = [ (name = "CACHE", memoryCache = ( limits = ( @@ -29,7 +33,9 @@ const unitTests :Workerd.Config = ( maxValueSize = 500, maxTotalValueSize = 600, ), - )) + )), + (name = "SELF", service = "memory-cache-test"), + (name = "READER", durableObjectNamespace = "CacheReader"), ] ) ), diff --git a/src/workerd/server/BUILD.bazel b/src/workerd/server/BUILD.bazel index 9f555dac3dc..1e7fc1a0767 100644 --- a/src/workerd/server/BUILD.bazel +++ b/src/workerd/server/BUILD.bazel @@ -213,6 +213,7 @@ wd_cc_library( "workerd-api.h", ], implementation_deps = [ + "//src/rust/jsg:ffi", "//src/workerd/api:analytics-engine", "//src/workerd/api:capnp", "//src/workerd/api:kv", @@ -220,6 +221,8 @@ wd_cc_library( "//src/workerd/api:urlpattern", "//src/workerd/api:urlpattern-standard", "//src/workerd/api:worker-loader", + "//src/workerd/api/memory-cache-api", + "//src/workerd/api/memory-cache-api:ffi.rs@cxx", "//src/workerd/io:promise-wrapper", ], local_defines = select({ diff --git a/src/workerd/server/workerd-api.c++ b/src/workerd/server/workerd-api.c++ index 6736d48e755..d86582d9644 100644 --- a/src/workerd/server/workerd-api.c++ +++ b/src/workerd/server/workerd-api.c++ @@ -22,6 +22,7 @@ #include #include #include +#include #include #include #include @@ -56,6 +57,7 @@ #include #include #include +#include #ifdef WORKERD_USE_TRANSPILER #include #endif // defined(WORKERD_USE_TRANSPILER) @@ -651,13 +653,18 @@ static v8::Local createBindingValue(JsgWorkerdIsolate::Lock& lock, // A cache without an id is not shared. kj::Maybe cacheId; if (cache.getCacheId().size() > 0) cacheId = cache.getCacheId(); - return lock.wrap(context, - lock.alloc(memoryCacheProvider.getUse(cacheId, - { - .maxKeys = cache.getMaxKeys(), - .maxValueSize = cache.getMaxValueSize(), - .maxTotalValueSize = cache.getMaxTotalValueSize(), - }))); + api::MemoryCacheLimits limits{ + .maxKeys = cache.getMaxKeys(), + .maxValueSize = cache.getMaxValueSize(), + .maxTotalValueSize = cache.getMaxTotalValueSize(), + }; + if (util::Autogate::isEnabled(util::AutogateKey::RUST_MEMORY_CACHE)) { + return jsg::check( + ::workerd::rust::jsg::maybe_local_from_ffi(api::memory_cache_new( + lock.v8Isolate, memoryCacheProvider.bindForRust(cacheId, limits)))); + } + return lock.wrap( + context, lock.alloc(memoryCacheProvider.getUse(cacheId, limits))); } case Global::EPHEMERAL_ACTOR_NAMESPACE: diff --git a/src/workerd/util/autogate.h b/src/workerd/util/autogate.h index 7e60f9a8c3b..b758005a7fa 100644 --- a/src/workerd/util/autogate.h +++ b/src/workerd/util/autogate.h @@ -130,7 +130,10 @@ namespace workerd::util { i18n::transcode) is provided by the Rust implementation (src/rust/i18n) \ instead of the C++ implementation. The C++ implementation is retained \ for rollback. */ \ - V(NODEJS_I18N_RUST) + V(NODEJS_I18N_RUST) \ + /* When enabled, memory cache bindings are the Rust MemoryCache \ + (src/workerd/api/memory-cache-api) instead of the C++ one (api/memory-cache.c++). */ \ + V(RUST_MEMORY_CACHE) // clang-format on // --------------------------------------------------------------------------------------