Cache API tests (#447)

* Respect `cf.cacheKey` in Cache API

* Fix caching of `ReadableStream`-bodied `Response`s

* Propagate `clock` through `GatewayFactory` for tests

* Catch loopback handler errors and return 500 response

* Add tests for `_getRangeResponse` and `_parseRanges` helpers

* Add `Miniflare` integration test helper

* Bump `undici` to `5.13.0` and add `@cloudflare/workers-types`

* Add Cache API tests

Author: mrbbot

package-lock.json

@@ -37,13 +37,14 @@
     "kleur": "^4.1.5",
     "source-map-support": "0.5.21",
     "stoppable": "^1.1.0",
-    "undici": "^5.12.0",
+    "undici": "^5.13.0",
     "workerd": "^1.20221111.5",
     "ws": "^8.11.0",
     "youch": "^3.2.2",
     "zod": "^3.18.0"
   },
   "devDependencies": {
+    "@cloudflare/workers-types": "^4.20221111.1",
     "@types/better-sqlite3": "^7.6.2",
     "@types/debug": "^4.1.7",
     "@types/estree": "^1.0.0",

packages/tre/package.json

@@ -46,6 +46,7 @@ import {
   serializeConfig,
 } from "./runtime";
 import {
+  Clock,
   HttpError,
   Log,
   MiniflareCoreError,
@@ -54,6 +55,7 @@ import {
   OptionalZodTypeOf,
   UnionToIntersection,
   ValueOf,
+  defaultClock,
 } from "./shared";
 import { anyAbortSignal } from "./shared/signal";
 import { waitForRequest } from "./wait";
@@ -157,6 +159,7 @@ export class Miniflare {
   #sharedOpts: PluginSharedOptions;
   #workerOpts: PluginWorkerOptions[];
   #log: Log;
+  readonly #clock: Clock;
 
   readonly #runtimeConstructor: RuntimeConstructor;
   #runtime?: Runtime;
@@ -201,6 +204,7 @@ export class Miniflare {
     this.#sharedOpts = sharedOpts;
     this.#workerOpts = workerOpts;
     this.#log = this.#sharedOpts.core.log ?? new NoOpLog();
+    this.#clock = this.#sharedOpts.core.clock ?? defaultClock;
     this.#initPlugins();
 
     // Get supported shell for executing runtime binary
@@ -220,6 +224,7 @@ export class Miniflare {
       if (plugin.gateway !== undefined && plugin.router !== undefined) {
         const gatewayFactory = new GatewayFactory<any>(
           this.#log,
+          this.#clock,
           this.#sharedOpts.core.cloudflareFetch,
           key,
           plugin.gateway,
@@ -350,24 +355,30 @@ export class Miniflare {
     });
 
     let response: Response | undefined;
-    const customService = request.headers.get(HEADER_CUSTOM_SERVICE);
-    if (customService !== null) {
-      response = await this.#handleLoopbackCustomService(
-        request,
-        customService
-      );
-    } else if (url.pathname === "/core/error") {
-      const workerSrcOpts = this.#workerOpts.map<SourceOptions>(
-        ({ core }) => core
-      );
-      response = await handlePrettyErrorRequest(
-        this.#log,
-        workerSrcOpts,
-        request
-      );
-    } else {
-      // TODO: check for proxying/outbound fetch header first (with plans for fetch mocking)
-      response = await this.#handleLoopbackPlugins(request, url);
+    try {
+      const customService = request.headers.get(HEADER_CUSTOM_SERVICE);
+      if (customService !== null) {
+        response = await this.#handleLoopbackCustomService(
+          request,
+          customService
+        );
+      } else if (url.pathname === "/core/error") {
+        const workerSrcOpts = this.#workerOpts.map<SourceOptions>(
+          ({ core }) => core
+        );
+        response = await handlePrettyErrorRequest(
+          this.#log,
+          workerSrcOpts,
+          request
+        );
+      } else {
+        // TODO: check for proxying/outbound fetch header first (with plans for fetch mocking)
+        response = await this.#handleLoopbackPlugins(request, url);
+      }
+    } catch (e: any) {
+      this.#log.error(e);
+      res.writeHead(500);
+      return res.end(e?.stack ?? String(e));
     }
 
     if (response === undefined) {

packages/tre/src/index.ts

@@ -1,3 +1,4 @@
+import assert from "assert";
 import crypto from "crypto";
 import http from "http";
 import { AddressInfo } from "net";
@@ -171,12 +172,13 @@ class HttpParser {
     });
   }
   private listen(request: http.IncomingMessage, response: http.ServerResponse) {
-    if (request.url) {
-      response?.socket?.write(
-        this.responses.get(request.url) ?? new Uint8Array()
-      );
-    }
-    response.end();
+    assert(request.url !== undefined);
+    assert(response.socket !== null);
+    const array = this.responses.get(request.url);
+    assert(array !== undefined);
+    // Write response to parse directly to underlying socket
+    response.socket.write(array);
+    response.socket.end();
   }
   public async parse(response: Uint8Array): Promise<ParsedHttpResponse> {
     await this.connected;
@@ -207,11 +209,12 @@ export class CacheGateway {
     private readonly clock: Clock
   ) {}
 
-  async match(request: Request): Promise<Response> {
+  async match(request: Request, cacheKey?: string): Promise<Response> {
     // Never cache Workers Sites requests, so we always return on-disk files
     if (isSitesRequest(request)) throw new CacheMiss();
 
-    const cached = await this.storage.get<CacheMetadata>(request.url);
+    cacheKey ??= request.url;
+    const cached = await this.storage.get<CacheMetadata>(cacheKey);
     if (cached?.metadata === undefined) throw new CacheMiss();
 
     const response = new CacheResponse(
@@ -228,11 +231,15 @@ export class CacheGateway {
     );
   }
 
-  async put(request: Request, value: ArrayBuffer): Promise<Response> {
+  async put(
+    request: Request,
+    value: Uint8Array,
+    cacheKey?: string
+  ): Promise<Response> {
     // Never cache Workers Sites requests, so we always return on-disk files
     if (isSitesRequest(request)) return new Response(null, { status: 204 });
 
-    const response = await HttpParser.get().parse(new Uint8Array(value));
+    const response = await HttpParser.get().parse(value);
 
     const { storable, expiration, headers } = getExpiration(
       this.clock,
@@ -246,7 +253,8 @@ export class CacheGateway {
       throw new StorageFailure();
     }
 
-    await this.storage.put<CacheMetadata>(request.url, {
+    cacheKey ??= request.url;
+    await this.storage.put<CacheMetadata>(cacheKey, {
       value: response.body,
       expiration: millisToSeconds(this.clock() + expiration),
       metadata: {
@@ -257,8 +265,9 @@ export class CacheGateway {
     return new Response(null, { status: 204 });
   }
 
-  async delete(request: Request): Promise<Response> {
-    const deleted = await this.storage.delete(request.url);
+  async delete(request: Request, cacheKey?: string): Promise<Response> {
+    cacheKey ??= request.url;
+    const deleted = await this.storage.delete(cacheKey);
     // This is an extremely vague error, but it fits with what the cache API in workerd expects
     if (!deleted) throw new PurgeFailure();
     return new Response(null);

packages/tre/src/plugins/cache/gateway.ts

@@ -97,3 +97,4 @@ export const CACHE_PLUGIN: Plugin<
 };
 
 export * from "./gateway";
+export * from "./range";