[Miniflare 3] Re-implement R2 gateway and add support for multipart uploads using new storage system

packages/tre/src/index.ts

@@ -33,6 +33,7 @@ import {
   HEADER_CF_BLOB,
   HEADER_PROBE,
   PLUGIN_ENTRIES,
+  Persistence,
   Plugins,
   SERVICE_ENTRY,
   SOCKET_ENTRY,
@@ -63,17 +64,18 @@ import {
   serializeConfig,
 } from "./runtime";
 import {
-  Clock,
   HttpError,
   Log,
   MiniflareCoreError,
   Mutex,
   NoOpLog,
   OptionalZodTypeOf,
-  defaultClock,
+  Timers,
+  defaultTimers,
   formatResponse,
 } from "./shared";
 import { anyAbortSignal } from "./shared/signal";
+import { NewStorage } from "./storage2";
 import { waitForRequest } from "./wait";
 
 // ===== `Miniflare` User Options =====
@@ -262,7 +264,7 @@ export class Miniflare {
   #sharedOpts: PluginSharedOptions;
   #workerOpts: PluginWorkerOptions[];
   #log: Log;
-  readonly #clock: Clock;
+  readonly #timers: Timers;
 
   #runtime?: Runtime;
   #removeRuntimeExitHook?: () => void;
@@ -314,7 +316,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.#timers = this.#sharedOpts.core.timers ?? defaultTimers;
     this.#initPlugins();
 
     this.#liveReloadServer = new WebSocketServer({ noServer: true });
@@ -360,7 +362,7 @@ export class Miniflare {
       if (plugin.gateway !== undefined && plugin.router !== undefined) {
         const gatewayFactory = new GatewayFactory<any>(
           this.#log,
-          this.#clock,
+          this.#timers,
           this.#sharedOpts.core.cloudflareFetch,
           key,
           plugin.gateway,
@@ -852,6 +854,17 @@ export class Miniflare {
     return response;
   }
 
+  /** @internal */
+  _getPluginStorage(
+    plugin: keyof Plugins,
+    namespace: string,
+    persist?: Persistence
+  ): NewStorage {
+    const factory = this.#gatewayFactories[plugin];
+    assert(factory !== undefined);
+    return factory.getStorage(namespace, persist).getNewStorage();
+  }
+
   async dispose(): Promise<void> {
     this.#disposeController.abort();
     try {

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

@@ -4,7 +4,7 @@ import http from "http";
 import { ReadableStream, TransformStream } from "stream/web";
 import CachePolicy from "http-cache-semantics";
 import { Headers, HeadersInit, Request, Response, fetch } from "../../http";
-import { Clock, DeferredPromise, Log } from "../../shared";
+import { DeferredPromise, Log, Timers } from "../../shared";
 import { Storage } from "../../storage";
 import {
   InclusiveRange,
@@ -26,7 +26,7 @@ interface CacheMetadata {
   size: number;
 }
 
-function getExpiration(clock: Clock, req: Request, res: Response) {
+function getExpiration(timers: Timers, req: Request, res: Response) {
   // Cloudflare ignores request Cache-Control
   const reqHeaders = normaliseHeaders(req.headers);
   delete reqHeaders["cache-control"];
@@ -59,7 +59,7 @@ function getExpiration(clock: Clock, req: Request, res: Response) {
   // @ts-expect-error `now` isn't included in CachePolicy's type definitions
   const originalNow = CachePolicy.prototype.now;
   // @ts-expect-error `now` isn't included in CachePolicy's type definitions
-  CachePolicy.prototype.now = clock;
+  CachePolicy.prototype.now = timers.now;
   try {
     const policy = new CachePolicy(cacheReq, cacheRes, { shared: true });
 
@@ -230,9 +230,13 @@ class SizingStream extends TransformStream<Uint8Array, Uint8Array> {
 export class CacheGateway {
   private readonly storage: KeyValueStorage<CacheMetadata>;
 
-  constructor(log: Log, legacyStorage: Storage, private readonly clock: Clock) {
+  constructor(
+    private readonly log: Log,
+    legacyStorage: Storage,
+    private readonly timers: Timers
+  ) {
     const storage = legacyStorage.getNewStorage();
-    this.storage = new KeyValueStorage(storage, clock);
+    this.storage = new KeyValueStorage(storage, timers);
   }
 
   async match(request: Request, cacheKey?: string): Promise<Response> {
@@ -291,7 +295,7 @@ export class CacheGateway {
     assert(body !== null);
 
     const { storable, expiration, headers } = getExpiration(
-      this.clock,
+      this.timers,
       request,
       response
     );
@@ -321,7 +325,7 @@ export class CacheGateway {
     await this.storage.put({
       key: cacheKey,
       value: body,
-      expiration: this.clock() + expiration,
+      expiration: this.timers.now() + expiration,
       metadata,
     });
     return new Response(null, { status: 204 });

packages/tre/src/plugins/core/index.ts

@@ -16,6 +16,7 @@ import {
   Log,
   LogLevel,
   MiniflareCoreError,
+  Timers,
 } from "../../shared";
 import { getCacheServiceName } from "../cache";
 import { DURABLE_OBJECTS_STORAGE_SERVICE_NAME } from "../do";
@@ -68,7 +69,7 @@ export const CoreSharedOptionsSchema = z.object({
   verbose: z.boolean().optional(),
 
   log: z.instanceof(Log).optional(),
-  clock: z.function().returns(z.number()).optional(),
+  timers: z.custom<Timers>().optional(),
   cloudflareFetch: CloudflareFetchSchema.optional(),
 
   // TODO: add back validation of cf object

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

@@ -1,11 +1,11 @@
-import { Clock, Log } from "../../shared";
+import { Log, Timers } from "../../shared";
 import { Storage } from "../../storage";
 
 export class DurableObjectsStorageGateway {
   constructor(
     private readonly log: Log,
     private readonly storage: Storage,
-    private readonly clock: Clock
+    private readonly timers: Timers
   ) {}
 
   async get(_key: string) {

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

@@ -1,8 +1,8 @@
 import { ReadableStream, TransformStream } from "stream/web";
 import {
-  Clock,
   HttpError,
   Log,
+  Timers,
   maybeApply,
   millisToSeconds,
   secondsToMillis,
@@ -159,10 +159,10 @@ export class KVGateway {
   constructor(
     private readonly log: Log,
     legacyStorage: Storage,
-    private readonly clock: Clock
+    private readonly timers: Timers
   ) {
     const storage = legacyStorage.getNewStorage();
-    this.storage = new KeyValueStorage(storage, clock);
+    this.storage = new KeyValueStorage(storage, timers);
   }
 
   async get<Metadata = unknown>(
@@ -187,7 +187,7 @@ export class KVGateway {
     validateKey(key);
 
     // Normalise and validate expiration
-    const now = millisToSeconds(this.clock());
+    const now = millisToSeconds(this.timers.now());
     let expiration = normaliseInt(options.expiration);
     const expirationTtl = normaliseInt(options.expirationTtl);
     if (expirationTtl !== undefined) {

packages/tre/src/plugins/kv/remote.ts

@@ -2,7 +2,7 @@ import assert from "assert";
 import { Blob } from "buffer";
 import { z } from "zod";
 import { BodyInit, FormData, Response } from "../../http";
-import { Clock, millisToSeconds } from "../../shared";
+import { Timers, millisToSeconds } from "../../shared";
 import {
   RemoteStorage,
   StorageListOptions,
@@ -78,16 +78,17 @@ const DEFAULT_CACHE_TTL = 60;
 // Returns seconds since UNIX epoch key should expire, using the specified
 // expiration only if it is sooner than the cache TTL
 function getCacheExpiration(
-  clock: Clock,
+  timers: Timers,
   expiration?: number,
   cacheTtl = DEFAULT_CACHE_TTL
 ): number {
   // Return minimum expiration
-  const cacheExpiration = millisToSeconds(clock()) + cacheTtl;
+  const cacheExpiration = millisToSeconds(timers.now()) + cacheTtl;
   if (expiration === undefined || isNaN(expiration)) return cacheExpiration;
   else return Math.min(cacheExpiration, expiration);
 }
 
+// TODO(soon): remove all this
 export class KVRemoteStorage extends RemoteStorage {
   async get(
     key: string,
@@ -100,7 +101,7 @@ export class KVRemoteStorage extends RemoteStorage {
       // cacheTtl may have changed between the original get call that cached
       // this value and now, so check the cache is still fresh with the new TTL
       const newExpiration = cachedValue.metadata.storedAt + cacheTtl;
-      if (newExpiration >= millisToSeconds(this.clock())) {
+      if (newExpiration >= millisToSeconds(this.timers.now())) {
         // If the cache is still fresh, update the expiration and return
         await this.cache.put<RemoteCacheMetadata>(key, {
           value: cachedValue.value,
@@ -157,9 +158,9 @@ export class KVRemoteStorage extends RemoteStorage {
     const result: StoredValueMeta = { value, expiration, metadata };
     await this.cache.put<RemoteCacheMetadata>(key, {
       value: result.value,
-      expiration: getCacheExpiration(this.clock, expiration, cacheTtl),
+      expiration: getCacheExpiration(this.timers, expiration, cacheTtl),
       metadata: {
-        storedAt: millisToSeconds(this.clock()),
+        storedAt: millisToSeconds(this.timers.now()),
         actualExpiration: result.expiration,
         actualMetadata: result.metadata,
       },
@@ -176,7 +177,7 @@ export class KVRemoteStorage extends RemoteStorage {
     if (value.expiration !== undefined) {
       // Send expiration as TTL to avoid "expiration times must be at least 60s
       // in the future" issues from clock skew when setting `expirationTtl: 60`.
-      const desiredTtl = value.expiration - millisToSeconds(this.clock());
+      const desiredTtl = value.expiration - millisToSeconds(this.timers.now());
       const ttl = Math.max(desiredTtl, 60);
       searchParams.set("expiration_ttl", ttl.toString());
     }
@@ -197,9 +198,9 @@ export class KVRemoteStorage extends RemoteStorage {
     // Store this value in the cache
     await this.cache.put<RemoteCacheMetadata>(key, {
       value: value.value,
-      expiration: getCacheExpiration(this.clock, value.expiration),
+      expiration: getCacheExpiration(this.timers, value.expiration),
       metadata: {
-        storedAt: millisToSeconds(this.clock()),
+        storedAt: millisToSeconds(this.timers.now()),
         actualExpiration: value.expiration,
         actualMetadata: value.metadata,
       },
@@ -219,8 +220,11 @@ export class KVRemoteStorage extends RemoteStorage {
     // "Store" delete in cache as tombstone
     await this.cache.put<RemoteCacheMetadata>(key, {
       value: new Uint8Array(),
-      expiration: getCacheExpiration(this.clock),
-      metadata: { storedAt: millisToSeconds(this.clock()), tombstone: true },
+      expiration: getCacheExpiration(this.timers),
+      metadata: {
+        storedAt: millisToSeconds(this.timers.now()),
+        tombstone: true,
+      },
     });
 
     // Technically, it's incorrect to always say we deleted the key by returning

packages/tre/src/plugins/r2/errors.ts

@@ -14,13 +14,17 @@ enum CfCode {
   InternalError = 10001,
   NoSuchObjectKey = 10007,
   EntityTooLarge = 100100,
+  EntityTooSmall = 10011,
   MetadataTooLarge = 10012,
   InvalidObjectName = 10020,
   InvalidMaxKeys = 10022,
+  NoSuchUpload = 10024,
+  InvalidPart = 10025,
   InvalidArgument = 10029,
   PreconditionFailed = 10031,
   BadDigest = 10037,
   InvalidRange = 10039,
+  BadUpload = 10048,
 }
 
 export class R2Error extends HttpError {
@@ -110,6 +114,16 @@ export class EntityTooLarge extends R2Error {
   }
 }
 
+export class EntityTooSmall extends R2Error {
+  constructor() {
+    super(
+      Status.BadRequest,
+      "Your proposed upload is smaller than the minimum allowed object size.",
+      CfCode.EntityTooSmall
+    );
+  }
+}
+
 export class MetadataTooLarge extends R2Error {
   constructor() {
     super(
@@ -160,6 +174,26 @@ export class InvalidMaxKeys extends R2Error {
   }
 }
 
+export class NoSuchUpload extends R2Error {
+  constructor() {
+    super(
+      Status.BadRequest,
+      "The specified multipart upload does not exist.",
+      CfCode.NoSuchUpload
+    );
+  }
+}
+
+export class InvalidPart extends R2Error {
+  constructor() {
+    super(
+      Status.BadRequest,
+      "One or more of the specified parts could not be found.",
+      CfCode.InvalidPart
+    );
+  }
+}
+
 export class PreconditionFailed extends R2Error {
   constructor() {
     super(
@@ -179,3 +213,13 @@ export class InvalidRange extends R2Error {
     );
   }
 }
+
+export class BadUpload extends R2Error {
+  constructor() {
+    super(
+      Status.RangeNotSatisfiable,
+      "There was a problem with the multipart upload.",
+      CfCode.BadUpload
+    );
+  }
+}