Add an injectable blob backend with an R2 store for the Cloudflare hosts

RhysSullivan · RhysSullivan · commit a66ce8ffec26 · 2026-06-11T16:08:08.000-07:00
The plugin blob seam (StorageDeps.blobs) was hardcoded to the FumaDB blob
table, so every blob write landed in the relational tier. createExecutor
now accepts a BlobStore, the host DB handle can carry one, and
makeScopedExecutor threads it through. makeR2BlobStore implements the
store over an R2 bucket via a structural R2BucketLike type, keeping the
SDK free of Cloudflare type dependencies.

cloud gains a BLOBS R2 binding (bucket must exist before deploy);
host-cloudflare reuses its existing BLOBS bucket directly for blob-seam
writes, which never enter D1 and so never need the oversized-value
offload wrap. Hosts without a bucket keep the FumaDB fallback. The prod
blob table is empty, so swapping the backend strands no data.
diff --git a/apps/cloud/src/db/fuma.ts b/apps/cloud/src/db/fuma.ts
@@ -1,3 +1,4 @@
+import { env } from "cloudflare:workers";
 import { Effect, Layer } from "effect";
 import { type FumaDB } from "@executor-js/fumadb";
 import { type DrizzleConfig } from "@executor-js/fumadb/adapters/drizzle";
@@ -9,7 +10,7 @@ import {
   type ExecutorDbHandle,
   type ExecutorDbProvider,
 } from "@executor-js/api/server";
-import type { FumaDb, FumaTables } from "@executor-js/sdk";
+import { makeR2BlobStore, type FumaDb, type FumaTables } from "@executor-js/sdk";
 
 import { DbService } from "./db";
 
@@ -66,6 +67,10 @@ export const cloudDbProviderLayer = (
         db: fuma.db,
         fuma: fuma.fuma,
         close: async () => {},
+        // Plugin blobs (multi-MB resolved specs) live in R2, not Postgres.
+        // Guarded because test workers / local dev may run without the
+        // binding — the executor then falls back to the FumaDB `blob` table.
+        blobs: env.BLOBS ? makeR2BlobStore(env.BLOBS) : undefined,
       };
     }),
   );
diff --git a/apps/cloud/src/env-augment.d.ts b/apps/cloud/src/env-augment.d.ts
@@ -20,6 +20,12 @@ declare global {
       DATABASE_URL?: string;
       EXECUTOR_DIRECT_DATABASE_URL?: string;
 
+      // Plugin blob seam backend (wrangler.jsonc `r2_buckets`). Declared here
+      // (optional) rather than regenerating worker-configuration.d.ts: test
+      // workers and older local setups run without the binding, and the db
+      // layer falls back to the Postgres `blob` table when absent.
+      BLOBS?: R2Bucket;
+
       // SSRF / private-network egress guard. Unset in production -> the guard is
       // ON; the test workers set "true" so fixtures can reach localhost.
       ALLOW_LOCAL_NETWORK?: string;
diff --git a/apps/cloud/wrangler.jsonc b/apps/cloud/wrangler.jsonc
@@ -43,6 +43,14 @@
       "localConnectionString": "postgresql://postgres:postgres@127.0.0.1:5433/postgres",
     },
   ],
+  // Plugin blob seam backend (multi-MB resolved specs). Bucket must exist
+  // before deploy: `wrangler r2 bucket create executor-cloud-blobs`.
+  "r2_buckets": [
+    {
+      "binding": "BLOBS",
+      "bucket_name": "executor-cloud-blobs",
+    },
+  ],
   "placement": {
     "region": "aws:us-east-1",
   },
diff --git a/apps/host-cloudflare/src/db/d1.ts b/apps/host-cloudflare/src/db/d1.ts
@@ -12,6 +12,7 @@ import {
   createExecutorFumaDb,
   type ExecutorDbHandle,
 } from "@executor-js/api/server";
+import { makeR2BlobStore } from "@executor-js/sdk";
 
 import { CLOUDFLARE_NAMESPACE, CLOUDFLARE_SCHEMA_VERSION } from "../config";
 
@@ -69,5 +70,9 @@ export const createD1ExecutorDb = async (
     fuma,
     // The D1 binding owns its own lifecycle; nothing to release.
     close: async () => {},
+    // Blob-seam writes go straight to R2 — they never enter D1, so they never
+    // need the offload wrap above (which remains only for legacy oversized
+    // values already inlined in D1 rows).
+    blobs: blobs ? makeR2BlobStore(blobs) : undefined,
   };
 };
diff --git a/packages/core/api/src/server/scoped-executor.ts b/packages/core/api/src/server/scoped-executor.ts
@@ -178,7 +178,7 @@ export const makeScopedExecutor = <
   _organizationName: string,
 ): Effect.Effect<Executor<TPlugins>, StorageFailure, DbProvider | PluginsProvider | HostConfig> =>
   Effect.gen(function* () {
-    const { db } = yield* DbProvider;
+    const { db, blobs } = yield* DbProvider;
     const { plugins: pluginsFactory } = yield* PluginsProvider;
     const config = yield* HostConfig;
     // Explicit config wins; otherwise fall back to the request origin if a host
@@ -219,6 +219,7 @@ export const makeScopedExecutor = <
       tenant: Tenant.make(organizationId),
       subject: Subject.make(accountId),
       db,
+      blobs,
       plugins,
       httpClientLayer,
       onElicitation: "accept-all",
diff --git a/packages/core/sdk/src/blob.test.ts b/packages/core/sdk/src/blob.test.ts
@@ -3,7 +3,13 @@ import { Effect } from "effect";
 
 import { StorageError } from "./fuma-runtime";
 
-import { makeInMemoryBlobStore, pluginBlobStore, type OwnerPartitions } from "./blob";
+import {
+  makeInMemoryBlobStore,
+  makeR2BlobStore,
+  pluginBlobStore,
+  type OwnerPartitions,
+  type R2BucketLike,
+} from "./blob";
 
 // v2: owner partitions instead of a scope stack. Reads fall through
 // [user, org] (user = innermost); writes/deletes name an explicit owner.
@@ -100,6 +106,106 @@ describe("pluginBlobStore", () => {
   );
 });
 
+// Minimal in-memory R2 bucket double — same observable surface as the
+// `R2BucketLike` slice of a real binding.
+const makeFakeBucket = (): { bucket: R2BucketLike; objects: Map<string, string> } => {
+  const objects = new Map<string, string>();
+  return {
+    objects,
+    bucket: {
+      get: async (key) => {
+        const value = objects.get(key);
+        return value === undefined ? null : { text: async () => value };
+      },
+      put: async (key, value) => {
+        objects.set(key, value);
+      },
+      delete: async (key) => {
+        objects.delete(key);
+      },
+      head: async (key) => (objects.has(key) ? {} : null),
+    },
+  };
+};
+
+describe("makeR2BlobStore", () => {
+  it.effect("round-trips a value and names objects `namespace/key`", () =>
+    Effect.gen(function* () {
+      const { bucket, objects } = makeFakeBucket();
+      const store = makeR2BlobStore(bucket);
+
+      yield* store.put("o:t1/openapi", "spec/abc123", "spec-text");
+      expect(yield* store.get("o:t1/openapi", "spec/abc123")).toBe("spec-text");
+      expect(objects.has("o:t1/openapi/spec/abc123")).toBe(true);
+    }),
+  );
+
+  it.effect("get returns null for a missing object", () =>
+    Effect.gen(function* () {
+      const { bucket } = makeFakeBucket();
+      const store = makeR2BlobStore(bucket);
+      expect(yield* store.get("o:t1/openapi", "missing")).toBeNull();
+    }),
+  );
+
+  it.effect("getMany returns hits keyed by namespace", () =>
+    Effect.gen(function* () {
+      const { bucket } = makeFakeBucket();
+      const store = makeR2BlobStore(bucket);
+      yield* store.put("u:t1:s1/openapi", "k", "user-value");
+      yield* store.put("o:t1/openapi", "k", "org-value");
+
+      const hits = yield* store.getMany(["u:t1:s1/openapi", "o:t1/openapi", "o:t2/openapi"], "k");
+      expect(hits.size).toBe(2);
+      expect(hits.get("u:t1:s1/openapi")).toBe("user-value");
+      expect(hits.get("o:t1/openapi")).toBe("org-value");
+    }),
+  );
+
+  it.effect("delete and has agree", () =>
+    Effect.gen(function* () {
+      const { bucket } = makeFakeBucket();
+      const store = makeR2BlobStore(bucket);
+      yield* store.put("ns/p", "k", "v");
+      expect(yield* store.has("ns/p", "k")).toBe(true);
+      yield* store.delete("ns/p", "k");
+      expect(yield* store.has("ns/p", "k")).toBe(false);
+      expect(yield* store.get("ns/p", "k")).toBeNull();
+    }),
+  );
+
+  it.effect("bucket failures surface as StorageError", () =>
+    Effect.gen(function* () {
+      const failing: R2BucketLike = {
+        get: async () => {
+          // oxlint-disable-next-line executor/no-try-catch-or-throw -- test double simulating a bucket outage
+          throw { name: "R2Error", message: "bucket unavailable" };
+        },
+        put: async () => {},
+        delete: async () => {},
+        head: async () => null,
+      };
+      const store = makeR2BlobStore(failing);
+      const err = yield* store.get("ns", "k").pipe(Effect.flip);
+      expect(err).toBeInstanceOf(StorageError);
+    }),
+  );
+
+  it.effect("works behind pluginBlobStore owner partitioning", () =>
+    Effect.gen(function* () {
+      const { bucket } = makeFakeBucket();
+      const store = makeR2BlobStore(bucket);
+      const plugin = pluginBlobStore(store, partitions("o:t1", "u:t1:s1"), "openapi");
+
+      yield* plugin.put("spec/h1", "org-spec", { owner: "org" });
+      expect(yield* plugin.get("spec/h1")).toBe("org-spec");
+
+      yield* plugin.put("spec/h1", "user-spec", { owner: "user" });
+      expect(yield* plugin.get("spec/h1")).toBe("user-spec");
+    }),
+  );
+});
+
 describe("BlobStore.getMany", () => {
   it.effect("returns hits keyed by namespace", () =>
     Effect.gen(function* () {
diff --git a/packages/core/sdk/src/blob.ts b/packages/core/sdk/src/blob.ts
@@ -157,6 +157,71 @@ export const makeInMemoryBlobStore = (): BlobStore => {
   };
 };
 
+// ---------------------------------------------------------------------------
+// R2-backed BlobStore.
+//
+// Structural view of a Cloudflare R2 bucket binding — only the four calls the
+// store needs, so the SDK takes no dependency on `@cloudflare/workers-types`.
+// A real `R2Bucket` satisfies this shape as-is.
+// ---------------------------------------------------------------------------
+
+export interface R2BucketLike {
+  get(key: string): Promise<{ text(): Promise<string> } | null>;
+  put(key: string, value: string): Promise<unknown>;
+  delete(key: string): Promise<void>;
+  head(key: string): Promise<unknown | null>;
+}
+
+// Object name: `${namespace}/${key}`. Unambiguous because a namespace is
+// always `partition/pluginId` (exactly one slash; partitions use `:`
+// separators, plugin ids contain no slash), so the first two segments always
+// recover the namespace and the rest is the key.
+const r2ObjectName = (namespace: string, key: string): string => `${namespace}/${key}`;
+
+/**
+ * BlobStore over a Cloudflare R2 bucket. The right backend for multi-MB
+ * values (resolved OpenAPI specs, introspection documents) that have no
+ * business living in OLTP rows.
+ *
+ * Unlike `makeFumaBlobStore`, writes do NOT participate in FumaDB
+ * transactions — a rolled-back transaction leaves the blob behind. Callers
+ * should use idempotent (content-derived) keys so orphaned writes are
+ * harmless and re-puts are no-ops in effect.
+ */
+export const makeR2BlobStore = (bucket: R2BucketLike): BlobStore => {
+  const tryR2 = <A>(op: string, run: () => Promise<A>): Effect.Effect<A, StorageError> =>
+    Effect.tryPromise({
+      try: run,
+      catch: (cause) => new StorageError({ message: `R2 blob ${op} failed`, cause }),
+    });
+
+  return {
+    get: (namespace, key) =>
+      tryR2("get", async () => {
+        const object = await bucket.get(r2ObjectName(namespace, key));
+        return object == null ? null : await object.text();
+      }),
+    getMany: (namespaces, key) =>
+      tryR2("getMany", async () => {
+        const hits = new Map<string, string>();
+        await Promise.all(
+          namespaces.map(async (ns) => {
+            const object = await bucket.get(r2ObjectName(ns, key));
+            if (object != null) hits.set(ns, await object.text());
+          }),
+        );
+        return hits;
+      }),
+    put: (namespace, key, value) =>
+      tryR2("put", async () => {
+        await bucket.put(r2ObjectName(namespace, key), value);
+      }),
+    delete: (namespace, key) => tryR2("delete", () => bucket.delete(r2ObjectName(namespace, key))),
+    has: (namespace, key) =>
+      tryR2("head", async () => (await bucket.head(r2ObjectName(namespace, key))) != null),
+  };
+};
+
 const blobId = (namespace: string, key: string): string => JSON.stringify([namespace, key]);
 
 type BlobRow = {
diff --git a/packages/core/sdk/src/executor-fuma-db.ts b/packages/core/sdk/src/executor-fuma-db.ts
@@ -23,6 +23,7 @@ import { type DrizzleRuntimeProvider } from "@executor-js/fumadb/adapters/drizzl
 import { drizzleAdapter } from "@executor-js/fumadb/adapters/drizzle";
 import { schema as fumaSchema, type RelationsMap } from "@executor-js/fumadb/schema";
 
+import type { BlobStore } from "./blob";
 import type { FumaDb, FumaTables } from "./fuma-runtime";
 
 // The FumaDB provider both the runtime-schema generator and the drizzle adapter
@@ -106,4 +107,11 @@ export interface ExecutorDbHandle<
   TTables extends FumaTables = FumaTables,
 > extends ExecutorFumaDb<TTables> {
   readonly close: () => Promise<void>;
+  /**
+   * Optional blob backend assembled alongside the driver (an R2 bucket on the
+   * Cloudflare hosts). `makeScopedExecutor` threads it into
+   * `createExecutor({ blobs })`; absent, the executor defaults to the FumaDB
+   * `blob` table over `db`.
+   */
+  readonly blobs?: BlobStore;
 }
diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts
@@ -16,7 +16,7 @@ import {
   type FumaTables,
   type StorageFailure,
 } from "./fuma-runtime";
-import { makeFumaBlobStore, pluginBlobStore, type OwnerPartitions } from "./blob";
+import { makeFumaBlobStore, pluginBlobStore, type BlobStore, type OwnerPartitions } from "./blob";
 import { coreToolsPlugin } from "./core-tools";
 import type {
   Connection,
@@ -328,6 +328,12 @@ export interface ExecutorConfig<TPlugins extends readonly AnyPlugin[] = readonly
   /** The acting member, or omit for a pure-org executor (no `owner:"user"`). */
   readonly subject?: Subject;
   readonly db?: ExecutorDbInput | ExecutorDbFactory;
+  /**
+   * Backend for the plugin blob seam (`StorageDeps.blobs`). Defaults to the
+   * FumaDB `blob` table over `db`. Hosts with an object store hand one in
+   * (`makeR2BlobStore`) so multi-MB values stay out of the relational tier.
+   */
+  readonly blobs?: BlobStore;
   readonly plugins?: TPlugins;
   /** Config-level credential providers, merged with every
    *  `plugin.credentialProviders`. Config providers register first, so the
@@ -1217,7 +1223,7 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
     const rootDb = withQueryContext(rootDbUntyped, ownerContext);
     const fuma = makeFumaClient(rootDb);
     const core = makeCoreDb(fuma);
-    const blobs = makeFumaBlobStore(fuma);
+    const blobs = config.blobs ?? makeFumaBlobStore(fuma);
     const transaction = <A, E>(effect: Effect.Effect<A, E>) => fuma.transaction(effect);
 
     // Populated once, never mutated after startup.
diff --git a/packages/core/sdk/src/index.ts b/packages/core/sdk/src/index.ts
@@ -166,15 +166,18 @@ export {
   type InvokeOptions,
 } from "./elicitation";
 
-// Blob store — the plugin-facing CONTRACT only. The concrete makers
-// (`makeFumaBlobStore`/`makeInMemoryBlobStore`) are SDK-internal.
+// Blob store — the plugin-facing contract (`BlobStore`/`PluginBlobStore`)
+// plus the concrete backends hosts compose (`makeFumaBlobStore` default,
+// `makeR2BlobStore` for object-store hosts, `makeInMemoryBlobStore` for tests).
 export {
   pluginBlobStore,
   makeInMemoryBlobStore,
   makeFumaBlobStore,
+  makeR2BlobStore,
   type BlobStore,
   type PluginBlobStore,
   type OwnerPartitions,
+  type R2BucketLike,
 } from "./blob";
 
 // Plugin storage.
