History

Add an injectable blob backend with an R2 store for the Cloudflare hosts (#958)

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. The SDK keeps only the platform-neutral contract and backends; the R2 implementation lives with the Cloudflare host package (@executor-js/cloudflare/blob-store), typed against the real R2Bucket. 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.

Rhys Sullivan committed 5 months ago

Commit 4400c541 parent e3c49ef12 files · +231 −5Browse files

Unified · Split

12 files
  1. apps/cloud/src/db/fuma.ts +6 −0
  2. apps/cloud/src/env-augment.d.ts +8 −0
  3. apps/cloud/wrangler.jsonc +8 −0
  4. apps/host-cloudflare/src/db/d1.ts +5 −0
  5. bun.lock +1 −0
  6. packages/core/api/src/server/scoped-executor.ts +2 −1
  7. packages/core/sdk/src/executor-fuma-db.ts +8 −0
  8. packages/core/sdk/src/executor.ts +9 −2
  9. packages/core/sdk/src/index.ts +4 −2
  10. packages/hosts/cloudflare/package.json +5 −0
  11. packages/hosts/cloudflare/src/blob-store.test.ts +105 −0
  12. packages/hosts/cloudflare/src/blob-store.ts +70 −0
apps/cloud/src/db/fuma.ts+6−0
@@ -1,3 +1,4 @@
1import { env } from "cloudflare:workers";
1import { Effect, Layer } from "effect";2import { Effect, Layer } from "effect";
2import { type FumaDB } from "@executor-js/fumadb";3import { type FumaDB } from "@executor-js/fumadb";
3import { type DrizzleConfig } from "@executor-js/fumadb/adapters/drizzle";4import { type DrizzleConfig } from "@executor-js/fumadb/adapters/drizzle";
@@ -9,6 +10,7 @@ import { · 5 unmodified lines
9 type ExecutorDbHandle,10 type ExecutorDbHandle,
10 type ExecutorDbProvider,11 type ExecutorDbProvider,
11} from "@executor-js/api/server";12} from "@executor-js/api/server";
13import { makeR2BlobStore } from "@executor-js/cloudflare/blob-store";
12import type { FumaDb, FumaTables } from "@executor-js/sdk";14import type { FumaDb, FumaTables } from "@executor-js/sdk";
13 15
14import { DbService } from "./db";16import { DbService } from "./db";
@@ -66,6 +68,10 @@ export const cloudDbProviderLayer = ( · 51 unmodified lines
66 db: fuma.db,68 db: fuma.db,
67 fuma: fuma.fuma,69 fuma: fuma.fuma,
68 close: async () => {},70 close: async () => {},
71 // Plugin blobs (multi-MB resolved specs) live in R2, not Postgres.
72 // Guarded because test workers / local dev may run without the
73 // binding — the executor then falls back to the FumaDB `blob` table.
74 blobs: env.BLOBS ? makeR2BlobStore(env.BLOBS) : undefined,
69 };75 };
70 }),76 }),
71 );77 );
apps/cloud/src/env-augment.d.ts+8−0
@@ -20,6 +20,14 @@ declare global { · 19 unmodified lines
20 DATABASE_URL?: string;20 DATABASE_URL?: string;
21 EXECUTOR_DIRECT_DATABASE_URL?: string;21 EXECUTOR_DIRECT_DATABASE_URL?: string;
22 22
23 // Plugin blob seam backend (wrangler.jsonc `r2_buckets`). Declared here
24 // (optional) rather than regenerating worker-configuration.d.ts: test
25 // workers and older local setups run without the binding, and the db
26 // layer falls back to the Postgres `blob` table when absent. Typed via
27 // @cloudflare/workers-types (not the wrangler-generated global) to match
28 // what `@executor-js/cloudflare/blob-store` accepts.
29 BLOBS?: import("@cloudflare/workers-types").R2Bucket;
30
23 // SSRF / private-network egress guard. Unset in production -> the guard is31 // SSRF / private-network egress guard. Unset in production -> the guard is
24 // ON; the test workers set "true" so fixtures can reach localhost.32 // ON; the test workers set "true" so fixtures can reach localhost.
25 ALLOW_LOCAL_NETWORK?: string;33 ALLOW_LOCAL_NETWORK?: string;
apps/cloud/wrangler.jsonc+8−0
@@ -43,6 +43,14 @@ · 42 unmodified lines
43 "localConnectionString": "postgresql://postgres:postgres@127.0.0.1:5433/postgres",43 "localConnectionString": "postgresql://postgres:postgres@127.0.0.1:5433/postgres",
44 },44 },
45 ],45 ],
46 // Plugin blob seam backend (multi-MB resolved specs). Bucket must exist
47 // before deploy: `wrangler r2 bucket create executor-cloud-blobs`.
48 "r2_buckets": [
49 {
50 "binding": "BLOBS",
51 "bucket_name": "executor-cloud-blobs",
52 },
53 ],
46 "placement": {54 "placement": {
47 "region": "aws:us-east-1",55 "region": "aws:us-east-1",
48 },56 },
apps/host-cloudflare/src/db/d1.ts+5−0
@@ -12,6 +12,7 @@ import { · 11 unmodified lines
12 createExecutorFumaDb,12 createExecutorFumaDb,
13 type ExecutorDbHandle,13 type ExecutorDbHandle,
14} from "@executor-js/api/server";14} from "@executor-js/api/server";
15import { makeR2BlobStore } from "@executor-js/cloudflare/blob-store";
15 16
16import { CLOUDFLARE_NAMESPACE, CLOUDFLARE_SCHEMA_VERSION } from "../config";17import { CLOUDFLARE_NAMESPACE, CLOUDFLARE_SCHEMA_VERSION } from "../config";
17 18
@@ -69,5 +70,9 @@ export const createD1ExecutorDb = async ( · 51 unmodified lines
69 fuma,70 fuma,
70 // The D1 binding owns its own lifecycle; nothing to release.71 // The D1 binding owns its own lifecycle; nothing to release.
71 close: async () => {},72 close: async () => {},
73 // Blob-seam writes go straight to R2 — they never enter D1, so they never
74 // need the offload wrap above (which remains only for legacy oversized
75 // values already inlined in D1 rows).
76 blobs: blobs ? makeR2BlobStore(blobs) : undefined,
72 };77 };
73};78};
bun.lock+1−0
@@ -628,6 +628,7 @@ · 627 unmodified lines
628 "@executor-js/api": "workspace:*",628 "@executor-js/api": "workspace:*",
629 "@executor-js/execution": "workspace:*",629 "@executor-js/execution": "workspace:*",
630 "@executor-js/host-mcp": "workspace:*",630 "@executor-js/host-mcp": "workspace:*",
631 "@executor-js/sdk": "workspace:*",
631 "@modelcontextprotocol/sdk": "^1.29.0",632 "@modelcontextprotocol/sdk": "^1.29.0",
632 "agents": "^0.10.0",633 "agents": "^0.10.0",
633 "effect": "catalog:",634 "effect": "catalog:",
packages/core/api/src/server/scoped-executor.ts+2−1
@@ -178,7 +178,7 @@ export const makeScopedExecutor = < · 177 unmodified lines
178 _organizationName: string,178 _organizationName: string,
179): Effect.Effect<Executor<TPlugins>, StorageFailure, DbProvider | PluginsProvider | HostConfig> =>179): Effect.Effect<Executor<TPlugins>, StorageFailure, DbProvider | PluginsProvider | HostConfig> =>
180 Effect.gen(function* () {180 Effect.gen(function* () {
181 const { db } = yield* DbProvider;181 const { db, blobs } = yield* DbProvider;
182 const { plugins: pluginsFactory } = yield* PluginsProvider;182 const { plugins: pluginsFactory } = yield* PluginsProvider;
183 const config = yield* HostConfig;183 const config = yield* HostConfig;
184 // Explicit config wins; otherwise fall back to the request origin if a host184 // Explicit config wins; otherwise fall back to the request origin if a host
@@ -219,6 +219,7 @@ export const makeScopedExecutor = < · 34 unmodified lines
219 tenant: Tenant.make(organizationId),219 tenant: Tenant.make(organizationId),
220 subject: Subject.make(accountId),220 subject: Subject.make(accountId),
221 db,221 db,
222 blobs,
222 plugins,223 plugins,
223 httpClientLayer,224 httpClientLayer,
224 onElicitation: "accept-all",225 onElicitation: "accept-all",
packages/core/sdk/src/executor-fuma-db.ts+8−0
@@ -23,6 +23,7 @@ import { type DrizzleRuntimeProvider } from "@executor-js/fumadb/adapters/drizzl · 22 unmodified lines
23import { drizzleAdapter } from "@executor-js/fumadb/adapters/drizzle";23import { drizzleAdapter } from "@executor-js/fumadb/adapters/drizzle";
24import { schema as fumaSchema, type RelationsMap } from "@executor-js/fumadb/schema";24import { schema as fumaSchema, type RelationsMap } from "@executor-js/fumadb/schema";
25 25
26import type { BlobStore } from "./blob";
26import type { FumaDb, FumaTables } from "./fuma-runtime";27import type { FumaDb, FumaTables } from "./fuma-runtime";
27 28
28// The FumaDB provider both the runtime-schema generator and the drizzle adapter29// The FumaDB provider both the runtime-schema generator and the drizzle adapter
@@ -106,4 +107,11 @@ export interface ExecutorDbHandle< · 77 unmodified lines
106 TTables extends FumaTables = FumaTables,107 TTables extends FumaTables = FumaTables,
107> extends ExecutorFumaDb<TTables> {108> extends ExecutorFumaDb<TTables> {
108 readonly close: () => Promise<void>;109 readonly close: () => Promise<void>;
110 /**
111 * Optional blob backend assembled alongside the driver (an R2 bucket on the
112 * Cloudflare hosts). `makeScopedExecutor` threads it into
113 * `createExecutor({ blobs })`; absent, the executor defaults to the FumaDB
114 * `blob` table over `db`.
115 */
116 readonly blobs?: BlobStore;
109}117}
packages/core/sdk/src/executor.ts+9−2
@@ -16,7 +16,7 @@ import { · 15 unmodified lines
16 type FumaTables,16 type FumaTables,
17 type StorageFailure,17 type StorageFailure,
18} from "./fuma-runtime";18} from "./fuma-runtime";
19import { makeFumaBlobStore, pluginBlobStore, type OwnerPartitions } from "./blob";19import { makeFumaBlobStore, pluginBlobStore, type BlobStore, type OwnerPartitions } from "./blob";
20import { coreToolsPlugin } from "./core-tools";20import { coreToolsPlugin } from "./core-tools";
21import type {21import type {
22 Connection,22 Connection,
@@ -328,6 +328,13 @@ export interface ExecutorConfig<TPlugins extends readonly AnyPlugin[] = readonly · 305 unmodified lines
328 /** The acting member, or omit for a pure-org executor (no `owner:"user"`). */328 /** The acting member, or omit for a pure-org executor (no `owner:"user"`). */
329 readonly subject?: Subject;329 readonly subject?: Subject;
330 readonly db?: ExecutorDbInput | ExecutorDbFactory;330 readonly db?: ExecutorDbInput | ExecutorDbFactory;
331 /**
332 * Backend for the plugin blob seam (`StorageDeps.blobs`). Defaults to the
333 * FumaDB `blob` table over `db`. Hosts with an object store hand one in
334 * (e.g. the R2 store in `@executor-js/cloudflare/blob-store`) so multi-MB
335 * values stay out of the relational tier.
336 */
337 readonly blobs?: BlobStore;
331 readonly plugins?: TPlugins;338 readonly plugins?: TPlugins;
332 /** Config-level credential providers, merged with every339 /** Config-level credential providers, merged with every
333 * `plugin.credentialProviders`. Config providers register first, so the340 * `plugin.credentialProviders`. Config providers register first, so the
@@ -1217,7 +1224,7 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 883 unmodified lines
1217 const rootDb = withQueryContext(rootDbUntyped, ownerContext);1224 const rootDb = withQueryContext(rootDbUntyped, ownerContext);
1218 const fuma = makeFumaClient(rootDb);1225 const fuma = makeFumaClient(rootDb);
1219 const core = makeCoreDb(fuma);1226 const core = makeCoreDb(fuma);
1220 const blobs = makeFumaBlobStore(fuma);1227 const blobs = config.blobs ?? makeFumaBlobStore(fuma);
1221 const transaction = <A, E>(effect: Effect.Effect<A, E>) => fuma.transaction(effect);1228 const transaction = <A, E>(effect: Effect.Effect<A, E>) => fuma.transaction(effect);
1222 1229
1223 // Populated once, never mutated after startup.1230 // Populated once, never mutated after startup.
packages/core/sdk/src/index.ts+4−2
@@ -166,8 +166,10 @@ export { · 165 unmodified lines
166 type InvokeOptions,166 type InvokeOptions,
167} from "./elicitation";167} from "./elicitation";
168 168
169// Blob store — the plugin-facing CONTRACT only. The concrete makers169// Blob store — the plugin-facing contract (`BlobStore`/`PluginBlobStore`)
170// (`makeFumaBlobStore`/`makeInMemoryBlobStore`) are SDK-internal.170// plus the platform-neutral backends (`makeFumaBlobStore` default,
171// `makeInMemoryBlobStore` for tests). Platform-specific backends live with
172// their host (R2: `@executor-js/cloudflare/blob-store`).
171export {173export {
172 pluginBlobStore,174 pluginBlobStore,
173 makeInMemoryBlobStore,175 makeInMemoryBlobStore,
packages/hosts/cloudflare/package.json+5−0
@@ -4,6 +4,10 @@ · 3 unmodified lines
4 "private": true,4 "private": true,
5 "type": "module",5 "type": "module",
6 "exports": {6 "exports": {
7 "./blob-store": {
8 "types": "./src/blob-store.ts",
9 "default": "./src/blob-store.ts"
10 },
7 "./mcp/worker-transport": {11 "./mcp/worker-transport": {
8 "types": "./src/mcp/worker-transport.ts",12 "types": "./src/mcp/worker-transport.ts",
9 "default": "./src/mcp/worker-transport.ts"13 "default": "./src/mcp/worker-transport.ts"
@@ -38,6 +42,7 @@ · 28 unmodified lines
38 "@executor-js/api": "workspace:*",42 "@executor-js/api": "workspace:*",
39 "@executor-js/execution": "workspace:*",43 "@executor-js/execution": "workspace:*",
40 "@executor-js/host-mcp": "workspace:*",44 "@executor-js/host-mcp": "workspace:*",
45 "@executor-js/sdk": "workspace:*",
41 "@modelcontextprotocol/sdk": "^1.29.0",46 "@modelcontextprotocol/sdk": "^1.29.0",
42 "agents": "^0.10.0",47 "agents": "^0.10.0",
43 "effect": "catalog:"48 "effect": "catalog:"
packages/hosts/cloudflare/src/blob-store.test.ts+105−0
@@ -0,0 +1,105 @@
1import { describe, expect, it } from "@effect/vitest";
2import { Effect } from "effect";
3import type { R2Bucket } from "@cloudflare/workers-types";
4
5import { StorageError, pluginBlobStore } from "@executor-js/sdk";
6
7import { makeR2BlobStore } from "./blob-store";
8
9// Minimal in-memory R2 bucket double — covers the get/put/delete/head slice
10// the store uses. Typed via the boundary cast because a full R2Bucket carries
11// a dozen unrelated methods.
12const makeFakeBucket = (): { bucket: R2Bucket; objects: Map<string, string> } => {
13 const objects = new Map<string, string>();
14 // oxlint-disable-next-line executor/no-double-cast -- test double: only the slice the store calls is implemented
15 const bucket = {
16 get: async (key: string) => {
17 const value = objects.get(key);
18 return value === undefined ? null : { text: async () => value };
19 },
20 put: async (key: string, value: string) => {
21 objects.set(key, value);
22 },
23 delete: async (key: string) => {
24 objects.delete(key);
25 },
26 head: async (key: string) => (objects.has(key) ? {} : null),
27 } as unknown as R2Bucket;
28 return { bucket, objects };
29};
30
31describe("makeR2BlobStore", () => {
32 it.effect("round-trips a value and names objects `namespace/key`", () =>
33 Effect.gen(function* () {
34 const { bucket, objects } = makeFakeBucket();
35 const store = makeR2BlobStore(bucket);
36
37 yield* store.put("o:t1/openapi", "spec/abc123", "spec-text");
38 expect(yield* store.get("o:t1/openapi", "spec/abc123")).toBe("spec-text");
39 expect(objects.has("o:t1/openapi/spec/abc123")).toBe(true);
40 }),
41 );
42
43 it.effect("get returns null for a missing object", () =>
44 Effect.gen(function* () {
45 const { bucket } = makeFakeBucket();
46 const store = makeR2BlobStore(bucket);
47 expect(yield* store.get("o:t1/openapi", "missing")).toBeNull();
48 }),
49 );
50
51 it.effect("getMany returns hits keyed by namespace", () =>
52 Effect.gen(function* () {
53 const { bucket } = makeFakeBucket();
54 const store = makeR2BlobStore(bucket);
55 yield* store.put("u:t1:s1/openapi", "k", "user-value");
56 yield* store.put("o:t1/openapi", "k", "org-value");
57
58 const hits = yield* store.getMany(["u:t1:s1/openapi", "o:t1/openapi", "o:t2/openapi"], "k");
59 expect(hits.size).toBe(2);
60 expect(hits.get("u:t1:s1/openapi")).toBe("user-value");
61 expect(hits.get("o:t1/openapi")).toBe("org-value");
62 }),
63 );
64
65 it.effect("delete and has agree", () =>
66 Effect.gen(function* () {
67 const { bucket } = makeFakeBucket();
68 const store = makeR2BlobStore(bucket);
69 yield* store.put("ns/p", "k", "v");
70 expect(yield* store.has("ns/p", "k")).toBe(true);
71 yield* store.delete("ns/p", "k");
72 expect(yield* store.has("ns/p", "k")).toBe(false);
73 expect(yield* store.get("ns/p", "k")).toBeNull();
74 }),
75 );
76
77 it.effect("bucket failures surface as StorageError", () =>
78 Effect.gen(function* () {
79 // oxlint-disable-next-line executor/no-double-cast -- test double: only the slice the store calls is implemented
80 const failing = {
81 get: async () => {
82 // oxlint-disable-next-line executor/no-try-catch-or-throw -- test double simulating a bucket outage
83 throw { name: "R2Error", message: "bucket unavailable" };
84 },
85 } as unknown as R2Bucket;
86 const store = makeR2BlobStore(failing);
87 const err = yield* store.get("ns", "k").pipe(Effect.flip);
88 expect(err).toBeInstanceOf(StorageError);
89 }),
90 );
91
92 it.effect("works behind pluginBlobStore owner partitioning", () =>
93 Effect.gen(function* () {
94 const { bucket } = makeFakeBucket();
95 const store = makeR2BlobStore(bucket);
96 const plugin = pluginBlobStore(store, { org: "o:t1", user: "u:t1:s1" }, "openapi");
97
98 yield* plugin.put("spec/h1", "org-spec", { owner: "org" });
99 expect(yield* plugin.get("spec/h1")).toBe("org-spec");
100
101 yield* plugin.put("spec/h1", "user-spec", { owner: "user" });
102 expect(yield* plugin.get("spec/h1")).toBe("user-spec");
103 }),
104 );
105});
packages/hosts/cloudflare/src/blob-store.ts+70−0
@@ -0,0 +1,70 @@
1// ---------------------------------------------------------------------------
2// BlobStore over a Cloudflare R2 bucket — the object-store backend for the
3// SDK's blob seam on the Cloudflare hosts. The SDK defines only the
4// `BlobStore` contract (plus its FumaDB/in-memory defaults); this R2 binding
5// lives here so the SDK stays platform-agnostic.
6//
7// Object name: `${namespace}/${key}`. Unambiguous because a namespace is
8// always `partition/pluginId` (exactly one slash; partitions use `:`
9// separators, plugin ids contain no slash), so the first two segments always
10// recover the namespace and the rest is the key.
11//
12// Unlike `makeFumaBlobStore`, writes do NOT participate in FumaDB
13// transactions — a rolled-back transaction leaves the blob behind. Callers
14// should use idempotent (content-derived) keys so orphaned writes are
15// harmless and re-puts are no-ops in effect.
16// ---------------------------------------------------------------------------
17
18import { Effect } from "effect";
19import type { R2Bucket } from "@cloudflare/workers-types";
20
21import { StorageError, type BlobStore } from "@executor-js/sdk";
22
23const objectName = (namespace: string, key: string): string => `${namespace}/${key}`;
24
25const storeError = (op: string) => (cause: unknown) =>
26 new StorageError({ message: `R2 blob ${op} failed`, cause });
27
28export const makeR2BlobStore = (bucket: R2Bucket): BlobStore => ({
29 get: (namespace, key) =>
30 Effect.tryPromise({
31 try: async () => {
32 const object = await bucket.get(objectName(namespace, key));
33 return object == null ? null : await object.text();
34 },
35 catch: storeError("get"),
36 }),
37 // R2 has no multi-get; fetch the (at most two — user + org partition)
38 // namespaces concurrently.
39 getMany: (namespaces, key) =>
40 Effect.tryPromise({
41 try: async () => {
42 const hits = new Map<string, string>();
43 await Promise.all(
44 namespaces.map(async (namespace) => {
45 const object = await bucket.get(objectName(namespace, key));
46 if (object != null) hits.set(namespace, await object.text());
47 }),
48 );
49 return hits;
50 },
51 catch: storeError("getMany"),
52 }),
53 put: (namespace, key, value) =>
54 Effect.tryPromise({
55 try: async () => {
56 await bucket.put(objectName(namespace, key), value);
57 },
58 catch: storeError("put"),
59 }),
60 delete: (namespace, key) =>
61 Effect.tryPromise({
62 try: () => bucket.delete(objectName(namespace, key)),
63 catch: storeError("delete"),
64 }),
65 has: (namespace, key) =>
66 Effect.tryPromise({
67 try: async () => (await bucket.head(objectName(namespace, key))) != null,
68 catch: storeError("has"),
69 }),
70});