diff --git a/src/workerd/api/container.c++ b/src/workerd/api/container.c++ index 394fdc6cfe0..b47206e390a 100644 --- a/src/workerd/api/container.c++ +++ b/src/workerd/api/container.c++ @@ -760,7 +760,9 @@ class Container::TcpPortOutgoingFactory final: public Fetcher::OutgoingFactory { kj::Own newSingleUseClient(kj::Maybe cfStr) override { // At present we have no use for `cfStr`. - return kj::heap(byteStreamFactory, entropySource, headerTable, port); + return IoContext::current().getSubrequestNoChecks([&](auto& tracing, auto& channelFactory) { + return kj::heap(byteStreamFactory, entropySource, headerTable, port); + }, {.inHouse = false, .wrapMetrics = false}); } private: diff --git a/src/workerd/api/http.c++ b/src/workerd/api/http.c++ index ffb9ac613bd..71e1a4dfe3e 100644 --- a/src/workerd/api/http.c++ +++ b/src/workerd/api/http.c++ @@ -2434,12 +2434,15 @@ Fetcher::ClientWithTracing Fetcher::getClientWithTracing( return ClientWithTracing{kj::mv(client), kj::mv(traceContext)}; } KJ_CASE_ONEOF(outgoingFactory, IoOwn) { - // For outgoing factories, no trace context needed + // Outgoing factories are responsible for routing through getSubrequestNoChecks() (or + // getSubrequest()) internally if they create HTTP connections, to ensure external memory + // adjustment and other subrequest accounting are applied. auto client = outgoingFactory->newSingleUseClient(kj::mv(cfStr)); return ClientWithTracing{kj::mv(client), kj::none}; } KJ_CASE_ONEOF(outgoingFactory, kj::Own) { - // For cross-context outgoing factories, no trace context needed + // Same as OutgoingFactory above -- the factory is responsible for routing through + // getSubrequestNoChecks() internally. auto client = outgoingFactory->newSingleUseClient(ioContext, kj::mv(cfStr)); return ClientWithTracing{kj::mv(client), kj::none}; } diff --git a/src/workerd/api/http.h b/src/workerd/api/http.h index efbdb670349..df9f1a4c4f9 100644 --- a/src/workerd/api/http.h +++ b/src/workerd/api/http.h @@ -280,6 +280,10 @@ class Fetcher: public JsRpcClientProvider { // Used by Fetchers that use ad-hoc, single-use WorkerInterface instances, such as ones // created for Actors. // + // Implementations that create HTTP connections should route through + // IoContext::getSubrequestNoChecks() (or getSubrequest()) internally, to ensure external memory + // adjustment and other subrequest accounting are applied. + // // TODO(cleanup): Consider removing this in favor of `IoChannelFactory::SubrequestChannel`, which // is almost the same thing. class OutgoingFactory { @@ -297,6 +301,9 @@ class Fetcher: public JsRpcClientProvider { // Used by Fetchers that obtain their HttpClient in a custom way, but which aren't tied // to a specific I/O context. The factory object moves with the isolate across threads and // contexts, and must work from any context. + // + // Same as OutgoingFactory: implementations that create HTTP connections should route through + // IoContext::getSubrequestNoChecks() internally. class CrossContextOutgoingFactory { public: virtual kj::Own newSingleUseClient( diff --git a/src/workerd/api/sockets.c++ b/src/workerd/api/sockets.c++ index ebfcd6b4ccb..985fd6cc794 100644 --- a/src/workerd/api/sockets.c++ +++ b/src/workerd/api/sockets.c++ @@ -8,6 +8,7 @@ #include "streams/standard.h" #include "system-streams.h" +#include #include #include #include @@ -636,8 +637,11 @@ class StreamWorkerInterface final: public WorkerInterface { kj::Own StreamOutgoingFactory::newSingleUseClient(kj::Maybe cfStr) { JSG_ASSERT(stream.get() != nullptr, Error, "Fetcher created from internalNewHttpClient can only be used once"); - // Create a WorkerInterface that wraps the stream - return kj::heap(kj::addRef(*this)); + // Create a WorkerInterface that wraps the stream, routing through getSubrequestNoChecks to apply + // external memory adjustment for GC pressure. + return IoContext::current().getSubrequestNoChecks([&](auto& tracing, auto& channelFactory) { + return kj::heap(kj::addRef(*this)); + }, {.inHouse = false, .wrapMetrics = false}); } jsg::Promise> SocketsModule::internalNewHttpClient( diff --git a/src/workerd/io/io-context.h b/src/workerd/io/io-context.h index 25a18596c38..38756f59d30 100644 --- a/src/workerd/io/io-context.h +++ b/src/workerd/io/io-context.h @@ -758,6 +758,10 @@ class IoContext final: public kj::Refcounted, private kj::TaskSet::ErrorHandler kj::Maybe existingTraceContext; }; + // Wraps a WorkerInterface factory with subrequest accounting: tracing, optional metrics wrapping, + // and an external memory adjustment to pressure V8's GC. All code paths that create HTTP + // connections (including those built from capnp capabilities via getHttpOverCapnpFactory()) + // should route through this function or getSubrequest(). kj::Own getSubrequestNoChecks( kj::FunctionParam(TraceContext&, IoChannelFactory&)> func, SubrequestOptions options); @@ -834,6 +838,11 @@ class IoContext final: public kj::Refcounted, private kj::TaskSet::ErrorHandler // TODO(cleanup): Make it the caller's job to call asHttpClient() on the result of // getSubrequest*(). + // Get a raw Cap'n Proto capability for the given channel. This is appropriate for pure RPC use + // cases (e.g. actor operations, email dispatch) that don't create HTTP connections. If you're + // converting the capability to an HTTP service via getHttpOverCapnpFactory(), use + // getSubrequestNoChecks() instead and call channelFactory.getCapability() from the callback, + // so that the external memory adjustment and other subrequest accounting are applied. capnp::Capability::Client getCapnpChannel(uint channel) { return getIoChannelFactory().getCapability(channel); }