Cache API

src/workerd/io/io-context.c++

@@ -135,6 +135,7 @@ ThreadContext::HeaderIdBundle::HeaderIdBundle(kj::HttpHeaderTable::Builder& buil
       contentEncoding(builder.add("Content-Encoding")),
       cfCacheStatus(builder.add("CF-Cache-Status")),
       cacheControl(builder.add("Cache-Control")),
+      cfCacheNamespace(builder.add("CF-Cache-Namespace")),
       cfKvMetadata(builder.add("CF-KV-Metadata")),
       cfR2ErrorHeader(builder.add("CF-R2-Error")),
       cfBlobMetadataSize(builder.add("CF-R2-Metadata-Size")),

src/workerd/io/io-context.h

@@ -55,6 +55,7 @@ class ThreadContext {
     const kj::HttpHeaderId contentEncoding;
     const kj::HttpHeaderId cfCacheStatus;         // used by cache API implementation
     const kj::HttpHeaderId cacheControl;
+    const kj::HttpHeaderId cfCacheNamespace;       // used by Cache binding implementation
     const kj::HttpHeaderId cfKvMetadata;          // used by KV binding implementation
     const kj::HttpHeaderId cfR2ErrorHeader;       // used by R2 binding implementation
     const kj::HttpHeaderId cfBlobMetadataSize;    // used by R2 binding implementation

src/workerd/server/server-test.c++

@@ -2103,6 +2103,155 @@ KJ_TEST("Server: disk service allow dotfiles") {
   KJ_EXPECT(test.root->openFile(kj::Path({"secret"}))->readAllText() == "this is super-secret");
 }
 
+
+KJ_TEST("Cache: If no cache service is defined, access to the cache API should error") {
+  TestServer test(singleWorker(R"((
+    compatibilityDate = "2022-08-17",
+    modules = [
+      ( name = "test.js",
+        esModule =
+          `export default {
+          `  async fetch(request) {
+          `    try {
+          `      return new Response(await caches.default.match(request))
+          `    } catch (e) {return new Response(e.message)}
+          `
+          `  }
+          `}
+      )
+    ]
+  ))"_kj));
+
+
+  test.start();
+  auto conn = test.connect("test-addr");
+  conn.httpGet200("/",
+      "No Cache was configured");
+
+}
+
+
+KJ_TEST("Cache: cached response") {
+  TestServer test(R"((
+    services = [
+      ( name = "hello",
+        worker = (
+          cacheApiOutbound = "cache-outbound",
+          compatibilityDate = "2022-08-17",
+          modules = [
+            ( name = "main.js",
+              esModule =
+                `export default {
+                `  async fetch(request, env, ctx) {
+                `    const cache = caches.default;
+                `    let response = await cache.match(request);
+                `    return response ?? new Response('not cached');
+                `  }
+                `}
+            )
+          ]
+        )
+      ),
+      ( name = "cache-outbound", external = "cache-host" ),
+    ],
+    sockets = [
+      ( name = "main",
+        address = "test-addr",
+        service = "hello"
+      )
+    ]
+  ))"_kj);
+
+  test.start();
+  auto conn = test.connect("test-addr");
+  conn.sendHttpGet("/");
+
+  {
+    auto subreq = test.receiveSubrequest("cache-host");
+    subreq.recv(R"(
+      GET / HTTP/1.1
+      Host: foo
+      Cache-Control: only-if-cached
+
+    )"_blockquote);
+    subreq.send(R"(
+      HTTP/1.1 200 OK
+      CF-Cache-Status: HIT
+      Content-Length: 6
+
+      cached)"_blockquote);
+  }
+
+  conn.recv(R"(
+    HTTP/1.1 200 OK
+    Content-Length: 6
+    CF-Cache-Status: HIT
+
+    cached)"_blockquote);
+
+}
+
+
+KJ_TEST("Cache: cache name is passed through to service") {
+  TestServer test(R"((
+    services = [
+      ( name = "hello",
+        worker = (
+          cacheApiOutbound = "cache-outbound",
+          compatibilityDate = "2022-08-17",
+          modules = [
+            ( name = "main.js",
+              esModule =
+                `export default {
+                `  async fetch(request, env, ctx) {
+                `    const cache = await caches.open('test-cache');
+                `    let response = await cache.match(request);
+                `    return response ?? new Response('not cached');
+                `  }
+                `}
+            )
+          ]
+        )
+      ),
+      ( name = "cache-outbound", external = "cache-host" ),
+    ],
+    sockets = [
+      ( name = "main",
+        address = "test-addr",
+        service = "hello"
+      )
+    ]
+  ))"_kj);
+
+  test.start();
+  auto conn = test.connect("test-addr");
+  conn.sendHttpGet("/");
+
+  {
+    auto subreq = test.receiveSubrequest("cache-host");
+    subreq.recv(R"(
+      GET / HTTP/1.1
+      Host: foo
+      Cache-Control: only-if-cached
+      CF-Cache-Namespace: test-cache
+
+    )"_blockquote);
+    subreq.send(R"(
+      HTTP/1.1 200 OK
+      CF-Cache-Status: HIT
+      Content-Length: 6
+
+      cached)"_blockquote);
+  }
+
+  conn.recv(R"(
+    HTTP/1.1 200 OK
+    Content-Length: 6
+    CF-Cache-Status: HIT
+
+    cached)"_blockquote);
+
+}
 // =======================================================================================
 
 // TODO(beta): Test TLS (send and receive)

src/workerd/server/server.c++

@@ -1093,6 +1093,7 @@ public:
     // I/O channels, delivered when link() is called.
     kj::Array<Service*> subrequest;
     kj::Array<kj::Maybe<ActorNamespace&>> actor;  // null = configuration error
+    kj::Maybe<Service&> cache;
   };
   using LinkCallback = kj::Function<LinkedIoChannels(WorkerService&)>;
 
@@ -1303,10 +1304,70 @@ private:
   capnp::Capability::Client getCapability(uint channel) override {
     KJ_FAIL_REQUIRE("no capability channels");
   }
+  class CacheClientImpl final: public CacheClient {
+  public:
+    CacheClientImpl(Service& cacheService, kj::HttpHeaderId cacheNamespaceHeader)
+        : cacheService(cacheService), cacheNamespaceHeader(cacheNamespaceHeader) {}
+
+    kj::Own<kj::HttpClient> getDefault(kj::Maybe<kj::String> cfBlobJson,
+                                       kj::Maybe<Tracer::Span&> parentSpan) override {
+
+      return kj::heap<CacheHttpClientImpl>(
+          cacheService, cacheNamespaceHeader, nullptr, kj::mv(cfBlobJson), kj::mv(parentSpan));
+    }
+
+    kj::Own<kj::HttpClient> getNamespace(kj::StringPtr cacheName,
+                                         kj::Maybe<kj::String> cfBlobJson,
+                                         kj::Maybe<Tracer::Span&> parentSpan) override {
+      auto encodedName = kj::encodeUriComponent(cacheName);
+      return kj::heap<CacheHttpClientImpl>(
+          cacheService, cacheNamespaceHeader, kj::mv(encodedName), kj::mv(cfBlobJson), parentSpan);
+    }
+
+  private:
+    Service& cacheService;
+    kj::HttpHeaderId cacheNamespaceHeader;
+
+  };
+
+  class CacheHttpClientImpl final: public kj::HttpClient {
+  public:
+    CacheHttpClientImpl(Service& parent, kj::HttpHeaderId cacheNamespaceHeader,
+                        kj::Maybe<kj::String> cacheName, kj::Maybe<kj::String> cfBlobJson,
+                        kj::Maybe<Tracer::Span&> parentSpan)
+        : client(asHttpClient(parent.startRequest({kj::mv(cfBlobJson), parentSpan}))),
+          cacheName(kj::mv(cacheName)),
+          cacheNamespaceHeader(cacheNamespaceHeader) {}
+
+    Request request(kj::HttpMethod method, kj::StringPtr url,
+                    const kj::HttpHeaders &headers,
+                    kj::Maybe<uint64_t> expectedBodySize = nullptr) override {
+
+      return client->request(method, url, addCacheNameHeader(headers, cacheName),
+                             expectedBodySize);
+    }
+
+  private:
+    kj::Own<kj::HttpClient> client;
+    kj::Maybe<kj::String> cacheName;
+    kj::HttpHeaderId cacheNamespaceHeader;
+
+    kj::HttpHeaders addCacheNameHeader(const kj::HttpHeaders& headers,
+                                       kj::Maybe<kj::StringPtr> cacheName) {
+      auto headersCopy = headers.cloneShallow();
+      KJ_IF_MAYBE (name, cacheName) {
+        headersCopy.set(cacheNamespaceHeader, *name);
+      }
+
+      return headersCopy;
+    }
+  };
 
   kj::Own<CacheClient> getCache() override {
-    // TODO(beta): Cache API emulation.
-    JSG_FAIL_REQUIRE(Error, "The cache API is not yet implemented.");
+    auto& channels = KJ_REQUIRE_NONNULL(ioChannels.tryGet<LinkedIoChannels>(),
+                                        "link() has not been called");
+    auto& cache = JSG_REQUIRE_NONNULL(channels.cache, Error, "No Cache was configured");
+    return kj::heap<CacheClientImpl>(cache, threadContext.getHeaderIds().cfCacheNamespace);
   }
 
   TimerChannel& getTimer() override {
@@ -1791,10 +1852,22 @@ kj::Own<Server::Service> Server::makeWorker(kj::StringPtr name, config::Worker::
       return targetService->getActorNamespace(channel.designator.getClassName());
     };
 
-    return WorkerService::LinkedIoChannels {
-      .subrequest = services.finish(),
-      .actor = kj::mv(actors)
-    };
+    if (conf.hasCacheApiOutbound()) {
+      Service& cacheApi = lookupService(conf.getCacheApiOutbound(),
+                                        kj::str("Worker \"", name, "\"'s cacheApiOutbound"));
+
+      return WorkerService::LinkedIoChannels{
+          .subrequest = services.finish(),
+          .actor = kj::mv(actors),
+          .cache = &cacheApi
+      };
+    } else {
+      return WorkerService::LinkedIoChannels{
+          .subrequest = services.finish(),
+          .actor = kj::mv(actors)
+      };
+    }
+
   };
 
   return kj::heap<WorkerService>(globalContext->threadContext, kj::mv(worker),

src/workerd/server/workerd.capnp

@@ -424,6 +424,9 @@ struct Worker {
   # Where should the global "fetch" go to? The default is the service called "internet", which
   # should usually be configured to talk to the public internet.
 
+  cacheApiOutbound @11 :ServiceDesignator;
+  # Where should cache API (i.e. caches.default and caches.open(...)) requests go?
+
   durableObjectNamespaces @7 :List(DurableObjectNamespace);
   # List of durable object namespaces in this Worker.