Skip to content
rachel/workerd
File

Blob: src/workerd/server/fallback-service.c++

12.1 KB
1#include "fallback-service.h"
2 
3#include <capnp/compat/json.h>
4#include <capnp/message.h>
5#include <kj/async-io.h>
6#include <kj/compat/http.h>
7#include <kj/compat/url.h>
8#include <kj/debug.h>
9#include <kj/one-of.h>
10#include <kj/string.h>
11#include <kj/thread.h>
12 
13namespace workerd::fallback {
14namespace {
15 
16constexpr kj::StringPtr getMethodFromType(ImportType type) {
17  switch (type) {
18    case ImportType::IMPORT:
19      return "import"_kjc;
20    case ImportType::REQUIRE:
21      return "require"_kjc;
22    case ImportType::INTERNAL:
23      return "internal"_kjc;
24  }
25  KJ_UNREACHABLE;
26}
27 
28ModuleOrRedirect handleReturnPayload(
29    kj::Maybe<kj::String> jsonPayload, bool redirect, kj::StringPtr specifier) {
30  KJ_IF_SOME(payload, jsonPayload) {
31    // If the payload is empty then the fallback service failed to fetch the module.
32    if (payload.size() == 0) return kj::none;
33 
34    // If redirect is true then the fallback service returned a 301 redirect. The
35    // payload is the specifier of the new target module.
36    if (redirect) {
37      return kj::Maybe(kj::mv(payload));
38    }
39 
40    // The response from the fallback service must be a valid JSON serialization
41    // of the workerd module configuration. If it is not, or if there is any other
42    // error when processing here, we'll log the exception and return nothing.
43    KJ_TRY {
44      capnp::MallocMessageBuilder moduleMessage;
45      capnp::JsonCodec json;
46      json.handleByAnnotation<server::config::Worker::Module>();
47      auto moduleBuilder = moduleMessage.initRoot<server::config::Worker::Module>();
48      json.decode(payload, moduleBuilder);
49 
50      // If the module fallback service returns a name in the module then it has to
51      // match the specifier we passed in. This is an optional sanity check.
52      if (moduleBuilder.hasName()) {
53        if (moduleBuilder.getName() != specifier) {
54          KJ_LOG(ERROR,
55              "Fallback service failed to fetch module: returned module "
56              "name does not match specifier",
57              moduleBuilder.getName(), specifier);
58          return kj::none;
59        }
60      } else {
61        moduleBuilder.setName(kj::str(specifier));
62      }
63 
64      kj::Own<server::config::Worker::Module::Reader> ret = capnp::clone(moduleBuilder.asReader());
65      return ModuleOrRedirect(kj::mv(ret));
66    }
67    KJ_CATCH(exception) {
68      KJ_LOG(ERROR, "Fallback service failed to fetch module", exception, specifier);
69      return kj::none;
70    }
71  }
72 
73  // If we got here, no jsonPayload was received and we return nothing.
74  return kj::none;
75}
76 
77}  // namespace
78 
79// ---- FallbackServiceClient implementation ----
80 
81FallbackServiceClient::FallbackServiceClient(kj::String address)
82    : ownedAddress(kj::mv(address)),
83      thread([this]() { threadMain(); }) {}
84 
85FallbackServiceClient::~FallbackServiceClient() noexcept(false) {
86  // Signal the background thread to exit. The kj::Thread destructor
87  // (which runs after this body) will join the thread.
88  auto lock = state.lockExclusive();
89  lock->shutdown = true;
90}
91 
92ModuleOrRedirect FallbackServiceClient::tryResolve(Version version,
93    ImportType type,
94    kj::StringPtr specifier,
95    kj::StringPtr rawSpecifier,
96    kj::StringPtr referrer,
97    const kj::HashMap<kj::StringPtr, kj::StringPtr>& attributes) {
98  // Submit request to background thread.
99  {
100    auto lock = state.lockExclusive();
101    KJ_REQUIRE(!lock->shutdown, "FallbackServiceClient has been shut down");
102    KJ_ASSERT(!lock->hasRequest, "FallbackServiceClient does not support concurrent requests");
103    lock->version = version;
104    lock->type = type;
105    lock->specifier = specifier;
106    lock->rawSpecifier = rawSpecifier;
107    lock->referrer = referrer;
108    lock->attributes = &attributes;
109    lock->hasRequest = true;
110  }
111 
112  // Block until the background thread has processed our request.
113  return state.when([](const SharedState& s) { return s.responseReady || s.shutdown; },
114      [](SharedState& s) -> ModuleOrRedirect {
115    if (!s.responseReady) {
116      // Background thread shut down without producing a response.
117      return kj::none;
118    }
119    auto result = kj::mv(s.response);
120    s.responseReady = false;
121    return result;
122  });
123}
124 
125void FallbackServiceClient::threadMain() {
126  KJ_TRY {
127    // Set up the async I/O context, DNS resolution, and HTTP client once.
128    // These are reused for all subsequent requests.
129    kj::AsyncIoContext io = kj::setupAsyncIo();
130    kj::HttpHeaderTable::Builder builder;
131    kj::HttpHeaderId kMethod = builder.add("x-resolve-method");
132    auto headerTable = builder.build();
133 
134    auto addr = io.provider->getNetwork().parseAddress(ownedAddress, 80).wait(io.waitScope);
135    auto client = kj::newHttpClient(io.provider->getTimer(), *headerTable, *addr, {});
136 
137    while (true) {
138      // Wait for a request or shutdown signal.
139      // We copy to owned kj::String locals so that lifetimes are robust -- the caller
140      // blocks on responseReady (not hasRequest), but owning the strings makes this
141      // safe against future refactors that might change that invariant.
142      Version version;
143      ImportType type;
144      kj::String specifier;
145      kj::Maybe<kj::String> rawSpecifier;
146      kj::String referrer;
147      const kj::HashMap<kj::StringPtr, kj::StringPtr>* attributes;
148      bool shouldExit = state.when([](const SharedState& s) { return s.hasRequest || s.shutdown; },
149          [&](SharedState& s) -> bool {
150        if (s.shutdown) return true;
151        version = s.version;
152        type = s.type;
153        specifier = kj::str(s.specifier);
154        if (s.rawSpecifier != nullptr) {
155          rawSpecifier = kj::str(s.rawSpecifier);
156        }
157        referrer = kj::str(s.referrer);
158        attributes = s.attributes;
159        s.hasRequest = false;
160        return false;
161      });
162      if (shouldExit) return;
163 
164      // Process the request using the shared HTTP client.
165      ModuleOrRedirect result = kj::none;
166 
167      if (version == Version::V1) {
168        // === V1: GET request with query parameters ===
169        kj::Maybe<kj::String> jsonPayload;
170        bool redirect = false;
171        bool prefixed = false;
172        kj::Url url;
173        kj::StringPtr actualSpecifier = nullptr;
174 
175        KJ_IF_SOME(pos, specifier.findLast('/')) {
176          auto segment = specifier.slice(pos + 1);
177          if (segment.startsWith("node:") || segment.startsWith("cloudflare:") ||
178              segment.startsWith("workerd:")) {
179            actualSpecifier = segment;
180            url.query.add(
181                kj::Url::QueryParam{.name = kj::str("specifier"), .value = kj::str(segment)});
182            prefixed = true;
183          }
184        }
185        if (!prefixed) {
186          actualSpecifier = specifier;
187          if (actualSpecifier.startsWith("/")) {
188            actualSpecifier = specifier.slice(1);
189          }
190          url.query.add(kj::Url::QueryParam{kj::str("specifier"), kj::str(specifier)});
191        }
192        url.query.add(kj::Url::QueryParam{kj::str("referrer"), kj::str(referrer)});
193        // V1 always includes rawSpecifier in the query, defaulting to empty if absent.
194        KJ_IF_SOME(rs, rawSpecifier) {
195          url.query.add(kj::Url::QueryParam{kj::str("rawSpecifier"), kj::str(rs)});
196        } else {
197          url.query.add(kj::Url::QueryParam{kj::str("rawSpecifier"), kj::str("")});
198        }
199 
200        auto spec = url.toString(kj::Url::HTTP_REQUEST);
201 
202        // Retry once on disconnect (stale pooled connection).
203        for (int attempt = 0; attempt < 2; attempt++) {
204          KJ_TRY {
205            kj::HttpHeaders headers(*headerTable);
206            headers.setPtr(kMethod, getMethodFromType(type));
207            headers.setPtr(kj::HttpHeaderId::HOST, "localhost"_kj);
208 
209            auto request = client->request(kj::HttpMethod::GET, spec, headers, kj::none);
210            kj::HttpClient::Response resp = request.response.wait(io.waitScope);
211 
212            if (resp.statusCode == 301) {
213              KJ_IF_SOME(loc, resp.headers->get(kj::HttpHeaderId::LOCATION)) {
214                redirect = true;
215                jsonPayload = kj::str(loc);
216              } else {
217                KJ_LOG(ERROR, "Fallback service returned a redirect with no location", spec);
218              }
219              // Drain the response body to allow HTTP/1.1 connection reuse.
220              resp.body->readAllBytes().wait(io.waitScope);
221            } else if (resp.statusCode != 200) {
222              auto payload = resp.body->readAllText().wait(io.waitScope);
223              KJ_LOG(ERROR, "Fallback service failed to fetch module", payload, spec);
224            } else {
225              jsonPayload = resp.body->readAllText().wait(io.waitScope);
226            }
227            break;  // Success, no retry needed.
228          }
229          KJ_CATCH(exception) {
230            if (attempt == 0 && exception.getType() == kj::Exception::Type::DISCONNECTED) {
231              // Stale pooled connection; retry with a fresh one.
232              continue;
233            }
234            KJ_LOG(ERROR, "Fallback service failed to fetch module", exception, spec);
235          }
236        }
237 
238        result = handleReturnPayload(kj::mv(jsonPayload), redirect, actualSpecifier);
239 
240      } else {
241        // === V2: POST request with JSON body ===
242        capnp::JsonCodec json;
243        capnp::MallocMessageBuilder moduleMessage;
244        auto requestMsg = moduleMessage.initRoot<server::config::FallbackServiceRequest>();
245        requestMsg.setType(getMethodFromType(type));
246        requestMsg.setSpecifier(specifier);
247        requestMsg.setReferrer(referrer);
248 
249        KJ_IF_SOME(rs, rawSpecifier) {
250          requestMsg.setRawSpecifier(rs);
251        }
252 
253        KJ_ASSERT(attributes != nullptr);
254        if (attributes->size() > 0) {
255          auto attrs = requestMsg.initAttributes(attributes->size());
256          size_t n = 0;
257          for (auto& attr: *attributes) {
258            attrs[n].setName(attr.key);
259            attrs[n].setValue(attr.value);
260            n++;
261          }
262        }
263 
264        auto payload = json.encode(requestMsg);
265 
266        kj::Maybe<kj::String> jsonPayload;
267        bool redirect = false;
268 
269        // Retry once on disconnect (stale pooled connection).
270        for (int attempt = 0; attempt < 2; attempt++) {
271          KJ_TRY {
272            kj::HttpHeaders headers(*headerTable);
273            headers.setPtr(kj::HttpHeaderId::HOST, "localhost");
274 
275            auto request = client->request(kj::HttpMethod::POST, "/", headers, payload.size());
276            request.body->write(payload.asPtr().asBytes()).wait(io.waitScope);
277 
278            kj::HttpClient::Response resp = request.response.wait(io.waitScope);
279 
280            if (resp.statusCode == 301) {
281              KJ_IF_SOME(loc, resp.headers->get(kj::HttpHeaderId::LOCATION)) {
282                redirect = true;
283                jsonPayload = kj::str(loc);
284              } else {
285                KJ_LOG(ERROR, "Fallback service returned a redirect with no location", specifier);
286              }
287              // Drain the response body to allow HTTP/1.1 connection reuse.
288              resp.body->readAllBytes().wait(io.waitScope);
289            } else if (resp.statusCode != 200) {
290              auto body = resp.body->readAllText().wait(io.waitScope);
291              KJ_LOG(ERROR, "Fallback service failed to fetch module", body, specifier);
292            } else {
293              jsonPayload = resp.body->readAllText().wait(io.waitScope);
294            }
295            break;  // Success, no retry needed.
296          }
297          KJ_CATCH(exception) {
298            if (attempt == 0 && exception.getType() == kj::Exception::Type::DISCONNECTED) {
299              // Stale pooled connection; retry with a fresh one.
300              continue;
301            }
302            KJ_LOG(ERROR, "Fallback service failed to fetch module", exception);
303          }
304        }
305 
306        result = handleReturnPayload(kj::mv(jsonPayload), redirect, specifier);
307      }
308 
309      // Deliver the result to the calling thread.
310      {
311        auto lock = state.lockExclusive();
312        lock->response = kj::mv(result);
313        lock->responseReady = true;
314      }
315    }
316  }
317  KJ_CATCH(exception) {
318    KJ_LOG(ERROR, "Fallback service thread exiting; module resolution disabled", exception);
319    // Signal any waiting caller and prevent future requests.
320    auto lock = state.lockExclusive();
321    lock->response = kj::none;
322    lock->responseReady = true;
323    lock->shutdown = true;
324  }
325}
326 
327}  // namespace workerd::fallback