diff --git a/src/rust/jsg/BUILD.bazel b/src/rust/jsg/BUILD.bazel index 150feab7506..37e832d3f1f 100644 --- a/src/rust/jsg/BUILD.bazel +++ b/src/rust/jsg/BUILD.bazel @@ -7,7 +7,10 @@ load("//:build/wd_rust_crate.bzl", "wd_rust_crate") CPP_SHIMS = glob([ "*.c++", "*.h", -]) +]) + [ + "//src/workerd/io/io-context:ffi.c++", + "//src/workerd/io/io-context:ffi.h", +] # Fails when a shim adds a use of C++ JSG; see cpp-jsg-allowlist-test.py. py_test( diff --git a/src/rust/jsg/jsg.h b/src/rust/jsg/jsg.h index 8b9afe6e5d7..0c462418fb7 100644 --- a/src/rust/jsg/jsg.h +++ b/src/rust/jsg/jsg.h @@ -18,8 +18,19 @@ namespace workerd::rust::jsg { -// Calls a Rust module callback. An empty result means it panicked and threw. -inline v8::Local callModuleCallback(ModuleCallback callback, ::workerd::jsg::Lock& js) { +// The adapters below construct a `CallScope` around each call of a Rust module callback. A +// module's exports must not depend on the request that happens to import it first, but either +// registry can create them during a request: the legacy registry when a request first resolves +// the module (for example, by reading the `process` global), and the new one when a request +// resolves it directly. Callers pass `SuppressIoContextScope`, so that the callback sees no +// current IoContext, as ES module evaluation in the new registry already does. jsg cannot depend +// on the IoContext, so the scope is a template parameter, with no default so that every caller +// has to choose it. + +// Calls a Rust module callback in a `CallScope`. An empty result means it panicked and threw. +template +v8::Local callModuleCallback(ModuleCallback callback, ::workerd::jsg::Lock& js) { + CallScope scope; v8::Local value; if (!maybe_local_from_ffi(callback(js.v8Isolate)).ToLocal(&value)) { throw ::workerd::jsg::JsExceptionThrown(); @@ -29,7 +40,7 @@ inline v8::Local callModuleCallback(ModuleCallback callback, ::worker // Adapter for registering Rust-implemented modules into the OLD module registry // (jsg::ModuleRegistryImpl). Translates Rust ModuleCallback into ModuleInfo. -template +template struct RustModuleRegistry: public ::workerd::rust::jsg::ModuleRegistry { virtual ~RustModuleRegistry() = default; RustModuleRegistry(Registry& registry): registry(registry) {} @@ -56,7 +67,7 @@ struct RustModuleRegistry: public ::workerd::rust::jsg::ModuleRegistry { ::workerd::jsg::Lock& js, ::workerd::jsg::ModuleRegistry::ResolveMethod, kj::Maybe&) mutable -> kj::Maybe<::workerd::jsg::ModuleRegistry::ModuleInfo> { - auto value = callModuleCallback(callback, js); + v8::Local value = callModuleCallback(callback, js); KJ_DASSERT(value->IsObject()); using ModuleInfo = ::workerd::jsg::ModuleRegistry::ModuleInfo; @@ -77,8 +88,9 @@ struct RustModuleRegistry: public ::workerd::rust::jsg::ModuleRegistry { // callModuleCallback() unwraps before the synthetic module is registered. // // Usage: -// RustBuiltinModuleAdapter adapter(builder); +// RustBuiltinModuleAdapter adapter(builder); // ::workerd::rust::api::register_nodejs_modules(adapter); +template struct RustBuiltinModuleAdapter: public ::workerd::rust::jsg::ModuleRegistry { ::workerd::jsg::modules::ModuleBundle::BuiltinBuilder& builder; @@ -112,7 +124,7 @@ struct RustBuiltinModuleAdapter: public ::workerd::rust::jsg::ModuleRegistry { [callback = kj::mv(moduleCallback)](::workerd::jsg::Lock& js, const ::workerd::jsg::Url&, const ::workerd::jsg::modules::Module::ModuleNamespace& ns, const ::workerd::jsg::CompilationObserver&) mutable { - auto value = callModuleCallback(callback, js); + v8::Local value = callModuleCallback(callback, js); KJ_DASSERT(value->IsObject()); ns.setDefault(js, ::workerd::jsg::JsValue(value)); return true; diff --git a/src/rust/jsg/modules.rs b/src/rust/jsg/modules.rs index bc085ddb5a3..e6ca6701a29 100644 --- a/src/rust/jsg/modules.rs +++ b/src/rust/jsg/modules.rs @@ -20,9 +20,10 @@ pub use crate::v8::ffi::ModuleType; /// /// `callback` produces the module's exports when JavaScript imports it. It runs in the module /// registry's current scope, so it receives a [`Lock`] for that scope and returns a handle -/// branded with the same scope; an error it returns is thrown to the importer. It must capture -/// nothing: C++ calls it through a plain function pointer, so there is nowhere to keep captured -/// state (see [`UnitType`]). +/// branded with the same scope; an error it returns is thrown to the importer. It sees no +/// request, even when a request imports the module: the C++ registration runs it with no current +/// `IoContext`. It must capture nothing: C++ calls it through a plain function pointer, so there +/// is nowhere to keep captured state (see [`UnitType`]). pub fn add_builtin( registry: Pin<&mut ffi::ModuleRegistry>, specifier: &str, diff --git a/src/workerd/api/node/node.h b/src/workerd/api/node/node.h index fab99022512..43c3d55265b 100644 --- a/src/workerd/api/node/node.h +++ b/src/workerd/api/node/node.h @@ -13,6 +13,7 @@ #include #include #include +#include #include #include #include @@ -234,7 +235,7 @@ void registerNodeJsCompatModules( } } - ::workerd::rust::jsg::RustModuleRegistry r(registry); + ::workerd::rust::jsg::RustModuleRegistry r(registry); ::workerd::rust::api::register_nodejs_modules(r); } @@ -267,7 +268,7 @@ kj::Own getInternalNodeJsCompatModuleBundle( // by the same registration functions run in // getExternalNodeJsCompatModuleBundle(). { - ::workerd::rust::jsg::RustBuiltinModuleAdapter adapter(builder); + ::workerd::rust::jsg::RustBuiltinModuleAdapter adapter(builder); ::workerd::rust::api::register_nodejs_modules(adapter); } @@ -386,7 +387,7 @@ kj::Own getExternalNodeJsCompatModuleBundle( // legacy path registers Rust modules unconditionally with their declared // type). { - ::workerd::rust::jsg::RustBuiltinModuleAdapter adapter(builder); + ::workerd::rust::jsg::RustBuiltinModuleAdapter adapter(builder); ::workerd::rust::api::register_nodejs_modules(adapter); } return builder.finish(); diff --git a/src/workerd/io/AGENTS.md b/src/workerd/io/AGENTS.md index 495f6103f94..733ecc49dd8 100644 --- a/src/workerd/io/AGENTS.md +++ b/src/workerd/io/AGENTS.md @@ -27,6 +27,7 @@ I/O lifecycle, per-request context, worker/isolate management, actor storage, co | ------------------------------- | --------------------------------------------------------------------------------- | | Promise bridging KJ↔JS | `io-context.h` — `awaitIo()`, `awaitJs()` | | Request lifecycle / subrequests | `io-context.{h,c++}`, `worker-entrypoint.{h,c++}` | +| `IoContext` from Rust | `io-context/` — the `io-context` crate; identifies an `IoContext` by its `Id` | | Actor storage ops | `actor-cache.h` (`ActorCacheOps`), `actor-sqlite.h`, `actor-storage.capnp` | | DO gate semantics | `io-gate.{h,c++}` — `InputGate::CriticalSection`, `OutputGate::lockWhile()` | | Worker/isolate creation | `worker.{h,c++}`, `worker-modules.{h,c++}` | diff --git a/src/workerd/io/io-context.c++ b/src/workerd/io/io-context.c++ index 4c93dcb7a6a..b2fe3a2244a 100644 --- a/src/workerd/io/io-context.c++ +++ b/src/workerd/io/io-context.c++ @@ -1840,18 +1840,11 @@ void IoContext::runImpl(Runnable& runnable, }); } -static constexpr auto kAsyncIoErrorMessage = - "Disallowed operation called within global scope. Asynchronous I/O " - "(ex: fetch() or connect()), setting a timeout, and generating random " - "values are not allowed within global scope. To fix this error, perform this " - "operation within a handler. " - "https://developers.cloudflare.com/workers/runtime-apis/handlers/"; - IoContext& IoContext::current() { if (!hasCurrent()) { v8::Isolate* isolate = v8::Isolate::TryGetCurrent(); KJ_REQUIRE(isolate != nullptr, "there is no current request on this thread"); - isolate->ThrowError(jsg::v8StrIntern(isolate, kAsyncIoErrorMessage)); + isolate->ThrowError(jsg::v8StrIntern(isolate, NO_REQUEST_MESSAGE)); throw jsg::JsExceptionThrown(); } else { return *threadLocalRequest; diff --git a/src/workerd/io/io-context.h b/src/workerd/io/io-context.h index 136e61c6aac..0262b40b10c 100644 --- a/src/workerd/io/io-context.h +++ b/src/workerd/io/io-context.h @@ -554,6 +554,14 @@ class IoContext final: public kj::Refcounted, private kj::TaskSet::ErrorHandler // Throws an exception if there is no current context (see hasCurrent() below). static IoContext& current(); + // The message of the error current() throws when there is no current context. + static constexpr kj::StringPtr NO_REQUEST_MESSAGE = + "Disallowed operation called within global scope. Asynchronous I/O " + "(ex: fetch() or connect()), setting a timeout, and generating random " + "values are not allowed within global scope. To fix this error, perform this " + "operation within a handler. " + "https://developers.cloudflare.com/workers/runtime-apis/handlers/"_kj; + // Like current(), but returns kj::none if there is no current context. static kj::Maybe tryCurrent(); diff --git a/src/workerd/io/io-context/BUILD.bazel b/src/workerd/io/io-context/BUILD.bazel new file mode 100644 index 00000000000..5658d9fdd1d --- /dev/null +++ b/src/workerd/io/io-context/BUILD.bazel @@ -0,0 +1,61 @@ +load("//:build/kj_test.bzl", "kj_test") +load("//:build/wd_cc_library.bzl", "wd_cc_library") +load("//:build/wd_rust_crate.bzl", "wd_rust_crate") + +# The IoContext of the current request, for APIs implemented in Rust. A package of its own +# because //src/workerd/io depends on jsg. +wd_rust_crate( + name = "io-context", + srcs = [ + "ffi.rs", + "lib.rs", + "lib-test.rs", + ], + cxx_bridge_deps = ["//src/workerd/io"], + cxx_bridge_hdrs = ["ffi.h"], + cxx_bridge_src = "ffi.rs", + link_deps = [":ffi"], + test_deps = ["//src/rust/jsg-test"], + visibility = ["//visibility:public"], + deps = ["//src/rust/jsg"], +) + +wd_cc_library( + name = "ffi", + srcs = ["ffi.c++"], + hdrs = ["ffi.h"], + deps = ["//src/workerd/io"], +) + +# Checked by //src/rust/jsg:cpp-jsg-allowlist-test. +exports_files([ + "ffi.c++", + "ffi.h", +]) + +# The Rust half of io-context-test. +wd_rust_crate( + name = "io-context-test-bridge", + testonly = True, + srcs = ["test_bridge.rs"], + crate_root = "test_bridge.rs", + cxx_bridge_deps = ["//src/rust/jsg"], + cxx_bridge_hdrs = [], + cxx_bridge_src = "test_bridge.rs", + deps = [ + ":io-context", + "//src/rust/jsg", + ], +) + +# io-context inside a request of a real worker and in a Durable Object, and the IoContext that +# Rust builtin modules see when a request resolves them. +kj_test( + src = "io-context-test.c++", + deps = [ + ":ffi", + ":io-context-test-bridge", + "//src/rust/jsg:bridge", + "//src/workerd/tests:test-fixture", + ], +) diff --git a/src/workerd/io/io-context/ffi.c++ b/src/workerd/io/io-context/ffi.c++ new file mode 100644 index 00000000000..efbbe71e23e --- /dev/null +++ b/src/workerd/io/io-context/ffi.c++ @@ -0,0 +1,81 @@ +// 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 + +#include "ffi.h" + +#include +#include + +#include +#include + +namespace workerd::io_context { + +namespace { + +// IoContext::Id holds only the process-unique uint64_t that Rust stores. +static_assert(sizeof(IoContext::Id) == sizeof(uint64_t)); +static_assert(std::is_trivially_copyable_v); + +// The current IoContext, which must be the one `id` identifies. +IoContext& currentWithId(uint64_t id) { + auto& context = KJ_REQUIRE_NONNULL(IoContext::tryCurrent(), "there is no current request"); + KJ_REQUIRE(context.getId() == std::bit_cast(id), + "the IoContext is not the one of the current request"); + return context; +} + +} // namespace + +::rust::Str io_context_no_request_message() { + return IoContext::NO_REQUEST_MESSAGE.cStr(); +} + +kj::Maybe io_context_try_current_id() { + return IoContext::tryGetCurrentId().map( + [](IoContext::Id id) { return std::bit_cast(id); }); +} + +int64_t io_context_now(uint64_t id) { + return (currentWithId(id).now() - kj::UNIX_EPOCH) / kj::NANOSECONDS; +} + +void io_context_write_logfwdr(uint64_t id, uint32_t channel, ::rust::Slice message) { + auto& context = currentWithId(id); + + // Rust does not align the bytes to words, so they are copied. + auto bytes = kj::arrayPtr(message.data(), message.size()); + KJ_REQUIRE(bytes.size() > 0, "the serialized message is empty"); + KJ_REQUIRE(bytes.size() % sizeof(capnp::word) == 0, + "the serialized message is not a whole number of words"); + auto words = kj::heapArray(bytes.size() / sizeof(capnp::word)); + words.asBytes().copyFrom(bytes); + capnp::FlatArrayMessageReader reader(words); + KJ_REQUIRE(reader.getEnd() == words.end(), "bytes follow the serialized message"); + + KJ_IF_SOME(outputLocks, context.waitForOutputLocksIfNecessary()) { + // In a Durable Object, the message must not leave before the storage writes that precede it + // are confirmed, as in AnalyticsEngine::writeDataPoint(). The message is copied now, so that + // an invalid one fails this call rather than the write. + auto copy = kj::heap(); + copy->getRoot().set(reader.getRoot()); + context.addWaitUntil( + outputLocks.then([weakRef = context.getWeakRef(), channel, copy = kj::mv(copy)]() mutable { + // The IoContext owns its waitUntil tasks, so it is alive while one runs. + auto& context = KJ_ASSERT_NONNULL(weakRef->tryGet()); + // IoChannelFactory::writeLogfwdr() needs the current IoContext, so the write runs in it. + return context.run([channel, copy = kj::mv(copy)](Worker::Lock&, IoContext& context) mutable { + context.writeLogfwdr(channel, [&](capnp::AnyPointer::Builder builder) { + builder.set(copy->getRoot().asReader()); + }); + }); + })); + } else { + context.writeLogfwdr(channel, [&](capnp::AnyPointer::Builder builder) { + builder.set(reader.getRoot()); + }); + } +} + +} // namespace workerd::io_context diff --git a/src/workerd/io/io-context/ffi.h b/src/workerd/io/io-context/ffi.h new file mode 100644 index 00000000000..67240317000 --- /dev/null +++ b/src/workerd/io/io-context/ffi.h @@ -0,0 +1,32 @@ +// 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 + +#pragma once + +#include + +#include + +#include + +// The IoContext calls `ffi.rs` makes. Rust identifies an IoContext by its IoContext::Id, as a +// uint64_t, and holds no reference to it. Each call below that takes an `id` uses the current +// IoContext, and fails unless `id` identifies it: ids are never reused, so a stale id cannot +// reach another request's IoContext, or one that has been destroyed. +namespace workerd::io_context { + +// The message of the error IoContext::current() throws outside a request. +::rust::Str io_context_no_request_message(); + +// IoContext::tryGetCurrentId(), as a uint64_t. +kj::Maybe io_context_try_current_id(); + +// IoContext::now(), in nanoseconds since the Unix epoch. +int64_t io_context_now(uint64_t id); + +// IoContext::writeLogfwdr(), for a serialized Cap'n Proto message. In a Durable Object, the +// write waits for the output gate, as AnalyticsEngine::writeDataPoint() does. +void io_context_write_logfwdr(uint64_t id, uint32_t channel, ::rust::Slice message); + +} // namespace workerd::io_context diff --git a/src/workerd/io/io-context/ffi.rs b/src/workerd/io/io-context/ffi.rs new file mode 100644 index 00000000000..7c15117dc42 --- /dev/null +++ b/src/workerd/io/io-context/ffi.rs @@ -0,0 +1,19 @@ +// 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 to the C++ `IoContext` calls in `ffi.h`. + +#![expect(unsafe_code, reason = "the cxx bridge expands to unsafe FFI glue")] + +#[cxx::bridge(namespace = "workerd::io_context")] +pub mod bridge { + unsafe extern "C++" { + include!("workerd/io/io-context/ffi.h"); + + fn io_context_no_request_message() -> &'static str; + fn io_context_try_current_id() -> KjMaybe; + fn io_context_now(id: u64) -> Result; + fn io_context_write_logfwdr(id: u64, channel: u32, message: &[u8]) -> Result<()>; + } +} diff --git a/src/workerd/io/io-context/io-context-test.c++ b/src/workerd/io/io-context/io-context-test.c++ new file mode 100644 index 00000000000..73f68d822b3 --- /dev/null +++ b/src/workerd/io/io-context/io-context-test.c++ @@ -0,0 +1,219 @@ +// 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 + +// Tests the io-context crate inside and outside a request of a real worker. The Rust half is +// test_bridge.rs. + +#include +#include +#include +#include +#include +#include + +#include +#include +#include + +#include + +namespace workerd::io_context::test { +namespace { + +uintptr_t isolateOf(const TestFixture::V8Environment& env) { + return reinterpret_cast(env.isolate); +} + +uint64_t idOf(const IoContext& context) { + return std::bit_cast(context.getId()); +} + +int64_t nanosNow(IoContext& context) { + return (context.now() - kj::UNIX_EPOCH) / kj::NANOSECONDS; +} + +::rust::Slice slice(kj::ArrayPtr bytes) { + return ::rust::Slice(bytes.begin(), bytes.size()); +} + +// A serialized message whose root is `text`. +kj::Array serialize(kj::StringPtr text) { + capnp::MallocMessageBuilder message; + message.initRoot().setAs(text); + auto words = capnp::messageToFlatArray(message); + return kj::heapArray(words.asBytes()); +} + +// What the worker wrote to logfwdr, as ": " for each message. +struct LogfwdrLog { + kj::Vector writes; +}; + +struct RecordingIoChannelFactory final: public TestFixture::DummyIoChannelFactory { + RecordingIoChannelFactory(TimerChannel& timer, LogfwdrLog& log) + : DummyIoChannelFactory(timer), + log(log) {} + + kj::Promise writeLogfwdr( + uint channel, kj::FunctionParam buildMessage) override { + capnp::MallocMessageBuilder message; + auto root = message.initRoot(); + buildMessage(root); + log.writes.add(kj::str(channel, ": ", root.getAs())); + return kj::READY_NOW; + } + + LogfwdrLog& log; +}; + +TestFixture::SetupParams recordingParams( + LogfwdrLog& log, kj::Maybe actorId = kj::none) { + return { + .actorId = kj::mv(actorId), + .useRealTimers = false, + .ioChannelFactory = kj::Function(TimerChannel&)>( + [&log](TimerChannel& timer) -> kj::Rc { + return kj::rc(timer, log); + }), + }; +} + +KJ_TEST("inside a request, current() finds it, and now() is its clock") { + TestFixture fixture; + fixture.runInIoContext([&](const TestFixture::Environment& env) { + KJ_EXPECT(has_current(isolateOf(env))); + // The clock can advance between reads. + auto before = nanosNow(env.context); + auto now = now_nanos(isolateOf(env)); + KJ_EXPECT(before <= now && now <= nanosNow(env.context), before, now); + }); +} + +KJ_TEST("write_logfwdr() writes the message to the channel") { + LogfwdrLog log; + TestFixture fixture(recordingParams(log)); + fixture.runInIoContext([&](const TestFixture::Environment& env) { + auto message = serialize("hello"); + write_logfwdr(isolateOf(env), 7, slice(message)); + }); + KJ_ASSERT(log.writes.size() == 1); + KJ_EXPECT(log.writes[0] == "7: hello"); +} + +KJ_TEST("write_logfwdr() rejects anything but exactly one message") { + LogfwdrLog log; + TestFixture fixture(recordingParams(log)); + fixture.runInIoContext([&](const TestFixture::Environment& env) { + auto message = serialize("hello"); + KJ_EXPECT_THROW_MESSAGE( + "the serialized message is empty", write_logfwdr(isolateOf(env), 7, slice(nullptr))); + KJ_EXPECT_THROW_MESSAGE("not a whole number of words", + write_logfwdr(isolateOf(env), 7, slice(message.first(message.size() - 1)))); + auto trailing = kj::heapArray(message.size() + sizeof(capnp::word)); + trailing.first(message.size()).copyFrom(message); + trailing.slice(message.size()).fill(0); + KJ_EXPECT_THROW_MESSAGE( + "bytes follow the serialized message", write_logfwdr(isolateOf(env), 7, slice(trailing))); + }); + KJ_EXPECT(log.writes.size() == 0); +} + +KJ_TEST("in a Durable Object, write_logfwdr() waits for the output gate") { + LogfwdrLog log; + TestFixture fixture(recordingParams(log, Worker::Actor::Id(kj::str("logfwdr-actor")))); + auto gate = kj::newPromiseAndFulfiller(); + auto blocker = fixture.getActor().getOutputGate().lockWhile(kj::mv(gate.promise), nullptr); + + auto context = fixture.newIoContext(); + auto request = fixture.newIncomingRequest(*context); + fixture.enterContext(*request, [&](const TestFixture::Environment& env) { + auto message = serialize("after the gate"); + write_logfwdr(isolateOf(env), 3, slice(message)); + }); + fixture.pollEventLoop(); + KJ_EXPECT(log.writes.size() == 0); + + gate.fulfiller->fulfill(); + blocker.wait(fixture.getWaitScope()); + fixture.pollEventLoop(); + KJ_ASSERT(log.writes.size() == 1); + KJ_EXPECT(log.writes[0] == "3: after the gate"); + fixture.drainAndDestroy(kj::mv(request)); +} + +KJ_TEST("an IoContext id works only while its IoContext is current") { + TestFixture fixture; + auto first = fixture.newIncomingRequest(); + uint64_t firstId = 0; + fixture.enterContext(*first, [&](const TestFixture::Environment& env) { + firstId = idOf(env.context); + auto before = nanosNow(env.context); + auto now = io_context_now(firstId); + KJ_EXPECT(before <= now && now <= nanosNow(env.context), before, now); + }); + + auto second = fixture.newIncomingRequest(); + fixture.enterContext(*second, [&](const TestFixture::Environment& env) { + KJ_EXPECT(idOf(env.context) != firstId); + KJ_EXPECT_THROW_MESSAGE("not the one of the current request", io_context_now(firstId)); + }); + + fixture.enterWorkerLockSynchronously([&](Worker::Lock&) { + KJ_EXPECT_THROW_MESSAGE("there is no current request", io_context_now(firstId)); + }); + + fixture.drainAndDestroy(kj::mv(second)); + fixture.drainAndDestroy(kj::mv(first)); +} + +constexpr kj::StringPtr PROBE_SPECIFIER = "test:io-context-probe"_kj; + +KJ_TEST("the legacy registry creates a Rust module's exports outside the request resolving it") { + // The type wrapper is used only to register C++ resources, which this test doesn't. + struct UnusedTypeWrapper {}; + jsg::CompilationObserver observer; + jsg::ModuleRegistryImpl registry(observer); + ::workerd::rust::jsg::RustModuleRegistry adapter( + registry); + register_probe_module(adapter); + + TestFixture fixture; + fixture.runInIoContext([&](const TestFixture::Environment& env) { + KJ_EXPECT(registry.resolve(env.js, kj::Path::parse(PROBE_SPECIFIER), kj::none, + jsg::ModuleRegistry::ResolveOption::INTERNAL_ONLY) != kj::none); + KJ_EXPECT(IoContext::hasCurrent()); + }); + KJ_EXPECT(take_probe_observation() == ProbeObservation::NoRequest); +} + +KJ_TEST("the new registry creates a Rust module's exports outside the request resolving it") { + constexpr auto mainModule = "export default {};"_kjc; + const auto base = "file:///"_url; + jsg::modules::ModuleRegistry::Builder registryBuilder(base); + jsg::modules::ModuleBundle::BundleBuilder bundleBuilder(base); + bundleBuilder.addEsmModule("main"_kj, jsg::copyToArc(mainModule)); + jsg::modules::ModuleBundle::BuiltinBuilder builtinBuilder( + jsg::modules::ModuleBundle::BuiltinBuilder::Type::BUILTIN_ONLY); + ::workerd::rust::jsg::RustBuiltinModuleAdapter adapter(builtinBuilder); + register_probe_module(adapter); + registryBuilder.add(bundleBuilder.finish()).add(builtinBuilder.finish()); + + capnp::MallocMessageBuilder flagsMessage; + auto flags = flagsMessage.initRoot(); + flags.setNewModuleRegistry(true); + TestFixture fixture({ + .featureFlags = flags.asReader(), + .mainModuleSource = mainModule, + .useRealTimers = false, + .newModuleRegistry = registryBuilder.finish(), + }); + fixture.runInIoContext([&](const TestFixture::Environment& env) { + KJ_EXPECT(env.js.resolveInternalModule(PROBE_SPECIFIER) != kj::none); + KJ_EXPECT(IoContext::hasCurrent()); + }); + KJ_EXPECT(take_probe_observation() == ProbeObservation::NoRequest); +} + +} // namespace +} // namespace workerd::io_context::test diff --git a/src/workerd/io/io-context/lib-test.rs b/src/workerd/io/io-context/lib-test.rs new file mode 100644 index 00000000000..1a0b88e7dd9 --- /dev/null +++ b/src/workerd/io/io-context/lib-test.rs @@ -0,0 +1,23 @@ +// 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 + +use jsg_test::Harness; + +use super::*; + +/// Outside a request, `current()` fails with the error that C++ `IoContext::current()` throws +/// there, and `try_current()` returns `None`. +#[test] +fn outside_a_request_there_is_no_io_context() { + let harness = Harness::new(); + harness.run_in_context(|lock, _ctx| { + assert!(IoContext::try_current(lock).is_none()); + let Err(err) = IoContext::current(lock) else { + panic!("there should be no IoContext outside a request"); + }; + assert_eq!(err.name, jsg::ExceptionType::Error); + assert_eq!(err.message, bridge::io_context_no_request_message()); + Ok(()) + }); +} diff --git a/src/workerd/io/io-context/lib.rs b/src/workerd/io/io-context/lib.rs new file mode 100644 index 00000000000..92c6d058c9d --- /dev/null +++ b/src/workerd/io/io-context/lib.rs @@ -0,0 +1,88 @@ +// 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 `IoContext` of the current request, for APIs implemented in Rust. +//! +//! `IoContext` (`src/workerd/io/io-context.h`) holds the state of the request that JavaScript is +//! running on behalf of. This crate is a package of its own because `//src/workerd/io` depends on +//! `jsg`. + +mod ffi; + +use std::marker::PhantomData; +use std::time::Duration; +use std::time::SystemTime; +use std::time::UNIX_EPOCH; + +use ffi::bridge; + +/// The `IoContext` of the current request. +/// +/// It holds the process-unique id of the C++ `IoContext` (`IoContext::getId()`), not a reference +/// to it. Each method uses the current `IoContext`, and fails unless it is the one this handle +/// identifies. Ids are never reused, so a handle can never reach another request's `IoContext`, +/// or one that has been destroyed. A handle can be used only with the [`jsg::Lock`] it was +/// obtained with, and is neither `Send` nor `Sync`. +pub struct IoContext<'s> { + id: u64, + _lock: PhantomData<(&'s (), *const ())>, +} + +impl<'s> IoContext<'s> { + /// Returns the `IoContext` of the current request. + /// + /// The closure that `jsg::modules::add_builtin` runs to create a builtin module's exports + /// never sees a request, even when a request triggers it, so this fails there. + /// + /// # Errors + /// Outside a request, such as in a Worker's global scope, fails with the error that C++ + /// `IoContext::current()` throws, which explains that the operation needs a request. + pub fn current(lock: &mut jsg::Lock<'s>) -> jsg::Result { + Self::try_current(lock) + .ok_or_else(|| jsg::Error::new_error(bridge::io_context_no_request_message())) + } + + /// Returns the `IoContext` of the current request, or `None` outside a request. + #[must_use] + pub fn try_current(_lock: &mut jsg::Lock<'s>) -> Option { + Option::from(bridge::io_context_try_current_id()).map(|id| Self { + id, + _lock: PhantomData, + }) + } + + /// Returns the current time, as the request observes it, like `Date.now()` does in + /// JavaScript but to the nanosecond. Mitigating Spectre, the clock advances only when + /// JavaScript is waiting for I/O, unless the `precise_timers` compatibility flag is on. + /// + /// # Errors + /// Fails with an internal error if this `IoContext` is no longer the current one, or is + /// handling no request, which only happens while it is not running JavaScript. + pub fn now(&self) -> jsg::Result { + let nanos = bridge::io_context_now(self.id)?; + let offset = Duration::from_nanos(nanos.unsigned_abs()); + Ok(if nanos < 0 { + UNIX_EPOCH - offset + } else { + UNIX_EPOCH + offset + }) + } + + /// Sends `message`, one serialized Cap'n Proto message (as `capnp::serialize::write_message()` + /// writes it), to logfwdr channel `channel`. The request waits for the write, as it does for + /// `ctx.waitUntil()`, and the write makes no delivery guarantees. In a Durable Object, the + /// message is sent only once the output gate opens, that is, once the storage writes made + /// before it are confirmed. + /// + /// # Errors + /// Fails if `message` is not exactly one valid serialized message, and with an internal error + /// if this `IoContext` is no longer the current one. + pub fn write_logfwdr(&self, channel: u32, message: &[u8]) -> jsg::Result<()> { + Ok(bridge::io_context_write_logfwdr(self.id, channel, message)?) + } +} + +#[cfg(test)] +#[path = "lib-test.rs"] +mod tests; diff --git a/src/workerd/io/io-context/test_bridge.rs b/src/workerd/io/io-context/test_bridge.rs new file mode 100644 index 00000000000..0b2b0609b5d --- /dev/null +++ b/src/workerd/io/io-context/test_bridge.rs @@ -0,0 +1,111 @@ +// 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 Rust half of `io-context-test.c++`, which calls these functions inside and outside a +//! request of a real worker. +//! +//! Those that take an `isolate` take the `v8::Isolate*` of the caller: locked by the calling +//! thread, with a context entered and a `v8::HandleScope` open. + +use std::cell::Cell; +use std::pin::Pin; +use std::time::UNIX_EPOCH; + +use io_context::IoContext; +use jsg::ToJS; +use jsg::v8; + +/// C++ receives an error as an exception whose description is the `jsg::Error`'s name and +/// message. +type Result = std::io::Result; + +fn to_io(err: &jsg::Error) -> std::io::Error { + std::io::Error::other(err.to_string()) +} + +#[cxx::bridge(namespace = "workerd::io_context::test")] +mod ffi { + /// What the closure of the module `register_probe_module()` registers saw when it last ran. + #[derive(Clone, Copy, Debug, PartialEq, Eq)] + enum ProbeObservation { + NotRun, + NoRequest, + Request, + } + + #[namespace = "workerd::rust::jsg"] + unsafe extern "C++" { + include!("workerd/rust/jsg/ffi.h"); + + type ModuleRegistry = jsg::v8::ffi::ModuleRegistry; + } + + extern "Rust" { + fn register_probe_module(registry: Pin<&mut ModuleRegistry>); + fn take_probe_observation() -> ProbeObservation; + + unsafe fn has_current(isolate: usize) -> bool; + unsafe fn now_nanos(isolate: usize) -> Result; + unsafe fn write_logfwdr(isolate: usize, channel: u32, message: &[u8]) -> Result<()>; + } +} + +/// Whether `IoContext::try_current()` finds a request. +unsafe fn has_current(isolate: usize) -> bool { + // SAFETY: see the module documentation. + jsg::callback_scope!(unsafe let lock, isolate as *mut v8::ffi::Isolate); + IoContext::try_current(lock).is_some() +} + +/// `IoContext::current()?.now()`, in nanoseconds since the Unix epoch. +unsafe fn now_nanos(isolate: usize) -> Result { + // SAFETY: see the module documentation. + jsg::callback_scope!(unsafe let lock, isolate as *mut v8::ffi::Isolate); + let now = IoContext::current(lock) + .and_then(|context| context.now()) + .map_err(|err| to_io(&err))?; + let nanos = now + .duration_since(UNIX_EPOCH) + .map_err(std::io::Error::other)? + .as_nanos(); + i64::try_from(nanos).map_err(std::io::Error::other) +} + +/// `IoContext::current()?.write_logfwdr(channel, message)`. +unsafe fn write_logfwdr(isolate: usize, channel: u32, message: &[u8]) -> Result<()> { + // SAFETY: see the module documentation. + jsg::callback_scope!(unsafe let lock, isolate as *mut v8::ffi::Isolate); + IoContext::current(lock) + .and_then(|context| context.write_logfwdr(channel, message)) + .map_err(|err| to_io(&err)) +} + +thread_local! { + static PROBE_OBSERVATION: Cell = + const { Cell::new(ffi::ProbeObservation::NotRun) }; +} + +/// Registers the internal module `test:io-context-probe`, whose closure records whether it sees a +/// request. +fn register_probe_module(registry: Pin<&mut ffi::ModuleRegistry>) { + jsg::modules::add_builtin( + registry, + "test:io-context-probe", + |lock| { + let observation = if IoContext::try_current(lock).is_some() { + ffi::ProbeObservation::Request + } else { + ffi::ProbeObservation::NoRequest + }; + PROBE_OBSERVATION.with(|cell| cell.set(observation)); + Vec::::new().to_js(lock) + }, + jsg::modules::ModuleType::Internal, + ); +} + +/// What the probe module's closure saw when it last ran, which is then forgotten. +fn take_probe_observation() -> ffi::ProbeObservation { + PROBE_OBSERVATION.with(|cell| cell.replace(ffi::ProbeObservation::NotRun)) +}