diff --git a/.changeset/lucky-donkeys-listen.md b/.changeset/lucky-donkeys-listen.md new file mode 100644 index 00000000000..ae8bb9f5cdd --- /dev/null +++ b/.changeset/lucky-donkeys-listen.md @@ -0,0 +1,20 @@ +--- +"@cloudflare/vitest-plugin": minor +--- + +Add an experimental `newConfig` option for loading the Worker's configuration from `cloudflare.config.ts` + +Projects that have migrated to the new TypeScript configuration format had no way to run their Vitest suite against their real bindings, since there was no Wrangler configuration file left to point `wrangler.configPath` at. This adds the missing option, modelled on `@cloudflare/vite-plugin`'s `experimental.newConfig`: + +```ts +import { cloudflareTest } from "@cloudflare/vitest-plugin"; +import { defineProject } from "vitest/config"; + +export default defineProject({ + plugins: [cloudflareTest({ experimental: { newConfig: true } })], +}); +``` + +`newConfig: true` loads `cloudflare.config.ts` from the project root; pass `{ configPath: "..." }` to load it from elsewhere. Config functions are called with `ctx.mode` set to Vite's mode, which defaults to `"test"` and can be overridden with `--mode`. `experimental.newConfig` cannot be combined with `wrangler`. + +This is experimental and may change without a major version bump. Wrangler environments, `wrangler.config.ts` tooling configuration, and type generation are not supported yet. diff --git a/.changeset/miniflare-sighup-orphan-workerd.md b/.changeset/miniflare-sighup-orphan-workerd.md new file mode 100644 index 00000000000..4c38d84f7c1 --- /dev/null +++ b/.changeset/miniflare-sighup-orphan-workerd.md @@ -0,0 +1,7 @@ +--- +"miniflare": patch +--- + +Shut down `workerd` when Miniflare is terminated with `SIGHUP` + +On `SIGHUP`, Miniflare now stops `workerd` and removes its temporary directory instead of leaving them behind. Previously only `SIGINT` and `SIGTERM` were handled, so tools that embed Miniflare, such as `@cloudflare/vitest-pool-workers` and `@cloudflare/vite-plugin`, could leave a stray process and directory behind on each run. diff --git a/.changeset/strong-ravens-share.md b/.changeset/strong-ravens-share.md new file mode 100644 index 00000000000..400dfb7694e --- /dev/null +++ b/.changeset/strong-ravens-share.md @@ -0,0 +1,26 @@ +--- +"miniflare": minor +--- + +Add experimental shared local storage, letting several Miniflare instances read and write one set of local resources + +Each instance previously kept its own copy of local state, so two dev sessions pointed at the same KV namespace or D1 database could not see each other's writes. Instances that opt in now elect a single storage owner through the dev registry and route storage through it, so resources with the same ID resolve to the same data. + +Opt in with `unsafeEnableSharedStorage`, which requires three paths to be set: + +```js +new Miniflare({ + unsafeEnableSharedStorage: true, + // Shared between instances: resources that participate in sharing live here + resourcePersistencePath: "/path/to/shared/state", + // Per project: resources that cannot be shared keep their own state here + isolatedResourcePersistencePath: "/path/to/project/state", + // Instances elect the storage owner through the dev registry + unsafeDevRegistryPath: "/path/to/registry", + // ... +}); +``` + +KV, D1, R2, Rate Limits, and Secrets Store participate in sharing. Cache, Durable Objects, Workflows, observability, and Hello World storage do not yet, and stay instance-local under `isolatedResourcePersistencePath`, keeping their state across restarts without concurrent access to the shared root. + +This is experimental and the `unsafe`-prefixed options may change without a major version bump. diff --git a/.changeset/tidy-spoons-smile.md b/.changeset/tidy-spoons-smile.md new file mode 100644 index 00000000000..1901766ed99 --- /dev/null +++ b/.changeset/tidy-spoons-smile.md @@ -0,0 +1,7 @@ +--- +"@cloudflare/config": patch +--- + +Fix declaration emit for values returned by `defineSettings` + +Projects can now export a `defineSettings()` result while generating TypeScript declarations without encountering TS4023. diff --git a/fixtures/vitest-plugin-examples/new-config/README.md b/fixtures/vitest-plugin-examples/new-config/README.md new file mode 100644 index 00000000000..5155af9c324 --- /dev/null +++ b/fixtures/vitest-plugin-examples/new-config/README.md @@ -0,0 +1,9 @@ +# ✅ new-config + +This Worker is configured with a `cloudflare.config.ts` file instead of a Wrangler configuration file, using the experimental `experimental.newConfig` pool option. Bindings, the compatibility date and the entrypoint all come from that file. + +Config functions receive `ctx.mode` set to Vite's mode, which defaults to `"test"` under Vitest and can be overridden with `--mode`. + +| Test | Overview | +| ----------------------------------- | -------------------------------------------------------------------- | +| [index.test.ts](test/index.test.ts) | Bindings, `SELF` dispatch and unit tests against a new-config Worker | diff --git a/fixtures/vitest-plugin-examples/new-config/cloudflare.config.ts b/fixtures/vitest-plugin-examples/new-config/cloudflare.config.ts new file mode 100644 index 00000000000..df995ece813 --- /dev/null +++ b/fixtures/vitest-plugin-examples/new-config/cloudflare.config.ts @@ -0,0 +1,13 @@ +import { bindings, defineWorker } from "wrangler/experimental-config"; +import * as entrypoint from "./src/index.ts" with { type: "cf-worker" }; + +export default defineWorker({ + name: "vitest-plugin-new-config", + entrypoint, + compatibilityDate: "2025-12-02", + compatibilityFlags: ["nodejs_compat"], + env: { + MY_TEXT: bindings.text("from cloudflare.config.ts"), + MY_KV: bindings.kv({ id: "vitest-plugin-new-config-kv" }), + }, +}); diff --git a/fixtures/vitest-plugin-examples/new-config/src/index.ts b/fixtures/vitest-plugin-examples/new-config/src/index.ts new file mode 100644 index 00000000000..9e943853028 --- /dev/null +++ b/fixtures/vitest-plugin-examples/new-config/src/index.ts @@ -0,0 +1,12 @@ +export default { + async fetch(request, env, ctx) { + const url = new URL(request.url); + + if (url.pathname === "/kv") { + await env.MY_KV.put("key", "value"); + return new Response(await env.MY_KV.get("key")); + } + + return new Response(env.MY_TEXT); + }, +} satisfies ExportedHandler; diff --git a/fixtures/vitest-plugin-examples/new-config/src/tsconfig.json b/fixtures/vitest-plugin-examples/new-config/src/tsconfig.json new file mode 100644 index 00000000000..b06fc49bf63 --- /dev/null +++ b/fixtures/vitest-plugin-examples/new-config/src/tsconfig.json @@ -0,0 +1,13 @@ +{ + "extends": "../../tsconfig.workerd.json", + "compilerOptions": { + // `cloudflare.config.ts` references the entrypoint by path, including its + // `.ts` extension + "allowImportingTsExtensions": true + }, + "include": [ + "./**/*.ts", + "../cloudflare.config.ts", + "../worker-configuration.d.ts" + ] +} diff --git a/fixtures/vitest-plugin-examples/new-config/test/index.test.ts b/fixtures/vitest-plugin-examples/new-config/test/index.test.ts new file mode 100644 index 00000000000..7ca00204bc2 --- /dev/null +++ b/fixtures/vitest-plugin-examples/new-config/test/index.test.ts @@ -0,0 +1,36 @@ +import { + createExecutionContext, + env, + SELF, + waitOnExecutionContext, +} from "cloudflare:test"; +import { it } from "vitest"; +import worker from "../src/index"; + +it("exposes bindings declared in cloudflare.config.ts", ({ expect }) => { + expect(env.MY_TEXT).toBe("from cloudflare.config.ts"); +}); + +it("dispatches to the entrypoint declared in cloudflare.config.ts", async ({ + expect, +}) => { + const response = await SELF.fetch("https://example.com"); + expect(await response.text()).toBe("from cloudflare.config.ts"); +}); + +it("reads and writes the KV namespace", async ({ expect }) => { + const response = await SELF.fetch("https://example.com/kv"); + expect(await response.text()).toBe("value"); + expect(await env.MY_KV.get("key")).toBe("value"); +}); + +it("can unit test the handler directly", async ({ expect }) => { + const ctx = createExecutionContext(); + const response = await worker.fetch( + new Request("https://example.com"), + env, + ctx + ); + await waitOnExecutionContext(ctx); + expect(await response.text()).toBe("from cloudflare.config.ts"); +}); diff --git a/fixtures/vitest-plugin-examples/new-config/test/tsconfig.json b/fixtures/vitest-plugin-examples/new-config/test/tsconfig.json new file mode 100644 index 00000000000..1f4ab5bc59e --- /dev/null +++ b/fixtures/vitest-plugin-examples/new-config/test/tsconfig.json @@ -0,0 +1,9 @@ +{ + "extends": "../../tsconfig.workerd-test.json", + "compilerOptions": { + // `worker-configuration.d.ts` infers `Env` from `cloudflare.config.ts`, + // which references the entrypoint by path, including its `.ts` extension + "allowImportingTsExtensions": true + }, + "include": ["./**/*.ts", "../src/index.ts", "../worker-configuration.d.ts"] +} diff --git a/fixtures/vitest-plugin-examples/new-config/tsconfig.json b/fixtures/vitest-plugin-examples/new-config/tsconfig.json new file mode 100644 index 00000000000..83fd73ef8cd --- /dev/null +++ b/fixtures/vitest-plugin-examples/new-config/tsconfig.json @@ -0,0 +1,4 @@ +{ + "extends": "../tsconfig.node.json", + "include": ["vitest.config.ts"] +} diff --git a/fixtures/vitest-plugin-examples/new-config/vitest.config.ts b/fixtures/vitest-plugin-examples/new-config/vitest.config.ts new file mode 100644 index 00000000000..aeba8489e05 --- /dev/null +++ b/fixtures/vitest-plugin-examples/new-config/vitest.config.ts @@ -0,0 +1,18 @@ +import { cloudflareTest } from "@cloudflare/vitest-plugin"; +import { defineProject, mergeConfig } from "vitest/config"; +import configShared from "../../../vitest.shared"; + +export default mergeConfig( + configShared, + defineProject({ + plugins: [ + cloudflareTest({ + experimental: { + // Load the Worker's configuration from `cloudflare.config.ts` + // instead of a Wrangler configuration file + newConfig: true, + }, + }), + ], + }) +); diff --git a/fixtures/vitest-plugin-examples/new-config/worker-configuration.d.ts b/fixtures/vitest-plugin-examples/new-config/worker-configuration.d.ts new file mode 100644 index 00000000000..66d51676d11 --- /dev/null +++ b/fixtures/vitest-plugin-examples/new-config/worker-configuration.d.ts @@ -0,0 +1,13 @@ +/* eslint-disable */ +// Generated by @cloudflare/config +type __WorkerConfig = import("wrangler/experimental-config").UnwrapConfig; +type __Env = import("wrangler/experimental-config").InferEnv<__WorkerConfig>; + +declare namespace Cloudflare { + interface GlobalProps { + mainModule: import("wrangler/experimental-config").InferMainModule<__WorkerConfig>; + durableNamespaces: import("wrangler/experimental-config").InferDurableNamespaces<__WorkerConfig>; + } + interface Env extends __Env {} +} +interface Env extends Cloudflare.Env {} diff --git a/packages/config/src/public.ts b/packages/config/src/public.ts index b022faed23e..14deb23d383 100644 --- a/packages/config/src/public.ts +++ b/packages/config/src/public.ts @@ -85,5 +85,8 @@ export type { WorkerConfigInput, } from "./worker-definition"; export { defineWorker } from "./worker-definition"; -export type { SettingsConfigInput } from "./settings-definition"; +export type { + SettingsConfigInput, + SettingsDefinition, +} from "./settings-definition"; export { defineSettings } from "./settings-definition"; diff --git a/packages/config/src/settings-definition.ts b/packages/config/src/settings-definition.ts index ed59c99ef3e..4078b1cb0d7 100644 --- a/packages/config/src/settings-definition.ts +++ b/packages/config/src/settings-definition.ts @@ -8,10 +8,22 @@ import type { SettingsConfig } from "./types"; */ export type SettingsConfigInput = Omit; +/** + * A settings definition created by {@link defineSettings}. + */ +export interface SettingsDefinition { + [DEFINITION]: { + config: ConfigInput; + type: "settings"; + }; +} + /** * Declare shared settings. * Authored as a named `settings` export. */ -export function defineSettings(config: ConfigInput) { +export function defineSettings( + config: ConfigInput +): SettingsDefinition { return { [DEFINITION]: { config, type: "settings" } }; } diff --git a/packages/miniflare/AGENTS.md b/packages/miniflare/AGENTS.md index 83eeca59ba3..9e8a9d8531c 100644 --- a/packages/miniflare/AGENTS.md +++ b/packages/miniflare/AGENTS.md @@ -12,6 +12,7 @@ Local dev simulator for Cloudflare Workers, powered by workerd runtime. Main cla - `src/workers/core/dev-registry-proxy.worker.ts` — Proxy worker for cross-process service bindings via debug port RPC - `src/workers/core/dev-registry-proxy-shared.worker.ts` — Shared proxy logic (registry Map, DO proxy class, tail serializers) - `src/shared/dev-registry.ts` — Filesystem-based worker registry (chokidar watch, heartbeat, stale cleanup) +- `src/shared/persist-root-lock.ts` — Token-safe startup serialisation for shared persistent storage - `src/shared/DEV_REGISTRY.md` — Full architecture doc for the dev registry - `src/runtime/config/generated/workerd.ts` — Generated workerd Cap'n Proto config types - `test/` — Tests (`.spec.ts` naming, NOT `.test.ts`) diff --git a/packages/miniflare/src/config/schema.ts b/packages/miniflare/src/config/schema.ts index bcabc861946..e3618d58e11 100644 --- a/packages/miniflare/src/config/schema.ts +++ b/packages/miniflare/src/config/schema.ts @@ -701,11 +701,28 @@ export const InstanceOptionsSchema = z.strictObject({ stripDisablePrettyError: z.boolean().default(true), // Persistence - /** Root directory for persisted local resource state; relative to cwd if not absolute. */ + /** + * Root directory for persisted local resource state; relative to cwd if not + * absolute. When `unsafeEnableSharedStorage` is set this is canonicalised to + * an absolute real path before use, so every instance sharing the directory + * derives the same ownership scope. + */ resourcePersistencePath: z.string().optional(), + /** + * Root for resources that cannot participate in shared storage. Belongs at + * the project level -- each project keeps its own copy of this state rather + * than partitioning it under the shared resource root. + * + * Required when `unsafeEnableSharedStorage` is set. Parsing resolves this to + * the effective isolated root, falling back to `resourcePersistencePath` + * when shared storage is off, so readers never need to decide themselves. + */ + isolatedResourcePersistencePath: z.string().optional(), /** Project temp directory for plugin files; relative to cwd if not absolute. */ resourceTmpPath: z.string().optional(), + unsafeEnableSharedStorage: z.boolean().optional(), + containerEngine: z .union([ z.string(), @@ -769,7 +786,47 @@ export type ParsedLegacyConfig = NonNullable; export const MiniflareOptionsSchema = InstanceOptionsSchema.extend({ workers: z.array(WorkerOptionsSchema), -}); +}) + .superRefine((options, ctx) => { + if (!options.unsafeEnableSharedStorage) { + return; + } + if (!options.resourcePersistencePath?.trim()) { + ctx.addIssue({ + code: "custom", + path: ["resourcePersistencePath"], + message: + "Shared storage requires `resourcePersistencePath` to be set to the directory instances should share.", + }); + } + if (!options.isolatedResourcePersistencePath?.trim()) { + ctx.addIssue({ + code: "custom", + path: ["isolatedResourcePersistencePath"], + message: + "Shared storage requires `isolatedResourcePersistencePath` to be set to a per-project directory, for resources that cannot be shared.", + }); + } + if (!options.unsafeDevRegistryPath?.trim()) { + ctx.addIssue({ + code: "custom", + path: ["unsafeDevRegistryPath"], + message: + "Shared storage requires `unsafeDevRegistryPath` to be set, as instances elect a storage owner through the dev registry.", + }); + } + }) + .transform((options) => ({ + ...options, + // Resolve the effective isolated root once, here, so that everything + // downstream reads a single field that is always the path to persist to. + // Without shared storage nothing is shared, so every resource is isolated + // and the configured resource root is the isolated root. Validation above + // has already required an explicit isolated root when sharing is enabled. + isolatedResourcePersistencePath: options.unsafeEnableSharedStorage + ? options.isolatedResourcePersistencePath + : options.resourcePersistencePath, + })); export type MiniflareOptions = z.input; diff --git a/packages/miniflare/src/config/v4-convert.ts b/packages/miniflare/src/config/v4-convert.ts index 70e47a642e8..a4a0c2dfab8 100644 --- a/packages/miniflare/src/config/v4-convert.ts +++ b/packages/miniflare/src/config/v4-convert.ts @@ -70,6 +70,7 @@ function convertSharedOptions(options: ParsedV4MiniflareOptions) { unsafeInspectDurableObjects: options.unsafeInspectDurableObjects, logRequests: options.logRequests, resourcePersistencePath: options.resourcePersistencePath, + isolatedResourcePersistencePath: options.isolatedResourcePersistencePath, resourceTmpPath: options.resourceTmpPath, stripDisablePrettyError: options.stripDisablePrettyError, telemetry: options.telemetry, diff --git a/packages/miniflare/src/config/v4-schema.ts b/packages/miniflare/src/config/v4-schema.ts index 05c558a017b..e6655602ca4 100644 --- a/packages/miniflare/src/config/v4-schema.ts +++ b/packages/miniflare/src/config/v4-schema.ts @@ -654,6 +654,8 @@ export const V4SharedOptionsSchema = z.object({ logRequests: z.boolean().default(true), /** Root directory for persisted local resource state; relative to cwd if not absolute. */ resourcePersistencePath: z.string().optional(), + /** Per-instance root for resources that cannot participate in shared storage. */ + isolatedResourcePersistencePath: z.string().optional(), /** Project temp directory for plugin files; relative to cwd if not absolute. */ resourceTmpPath: z.string().optional(), stripDisablePrettyError: z.boolean().default(true), @@ -946,6 +948,7 @@ export type V4SharedOptions = { unsafeInspectDurableObjects?: boolean; logRequests?: boolean; resourcePersistencePath?: string; + isolatedResourcePersistencePath?: string; resourceTmpPath?: string; stripDisablePrettyError?: boolean; telemetry?: { enabled?: boolean; deviceId?: string }; diff --git a/packages/miniflare/src/exit-hook.ts b/packages/miniflare/src/exit-hook.ts index a4f4a66e3a5..b1d3d0e479e 100644 --- a/packages/miniflare/src/exit-hook.ts +++ b/packages/miniflare/src/exit-hook.ts @@ -43,6 +43,12 @@ function onSignalTerm(): void { process.exit(128 + 15); } +function onSignalHup(): void { + runCallbacks(); + // eslint-disable-next-line unicorn/no-process-exit -- intentional: replicate default SIGHUP behavior + process.exit(128 + 1); +} + function onMessage(message: unknown): void { if (message === "shutdown") { runCallbacks(); @@ -57,6 +63,13 @@ function addListeners(): void { process.on("exit", onExit); process.on("SIGINT", onSignalInt); process.on("SIGTERM", onSignalTerm); + // Without this, `SIGHUP` exits without running any handler, so `dispose()` + // never reaches the `SIGKILL` that stops `workerd` and it is left reparented + // to init. Matters most when Miniflare is embedded rather than run under + // `wrangler dev`, which leaves `workerd` in the caller's process group where + // the signal reaches it anyway. + // See https://github.com/cloudflare/workers-sdk/issues/9193. + process.on("SIGHUP", onSignalHup); // Only listen for IPC "shutdown" messages (PM2 support) when the process // actually has an IPC channel. Even without this guard the listener is // harmless when there is no channel, but being explicit avoids any @@ -71,6 +84,7 @@ function removeListeners(): void { process.removeListener("exit", onExit); process.removeListener("SIGINT", onSignalInt); process.removeListener("SIGTERM", onSignalTerm); + process.removeListener("SIGHUP", onSignalHup); process.removeListener("message", onMessage); } diff --git a/packages/miniflare/src/index.ts b/packages/miniflare/src/index.ts index a745fca4542..f82a69eb41c 100644 --- a/packages/miniflare/src/index.ts +++ b/packages/miniflare/src/index.ts @@ -47,6 +47,7 @@ import { getGlobalServices, getPersistPath, getRemoteProxyConnectionString, + getStorageScope, getTriggersOfType, HELLO_WORLD_PLUGIN_NAME, HOST_CAPNP_CONNECT, @@ -60,6 +61,7 @@ import { QUEUES_PLUGIN_NAME, QueuesError, R2_PLUGIN_NAME, + RATELIMIT_PLUGIN_NAME, SECRET_STORE_PLUGIN_NAME, SERVICE_DEV_REGISTRY_PROXY, SERVICE_ENTRY, @@ -103,12 +105,20 @@ import { stripAnsi, } from "./shared"; import { createDurableObjectStorageHandle } from "./shared/dev-control"; -import { DevRegistry, getWorkerRegistry } from "./shared/dev-registry"; +import { + DevRegistry, + getStorageCandidateName, + getWorkerRegistry, +} from "./shared/dev-registry"; import { getOutboundDoProxyClassName, normaliseServiceDesignator, } from "./shared/external-service"; import { isCompressedByCloudflareFL } from "./shared/mime-types"; +import { + canonicalisePersistRoot, + withPersistRootStartupLock, +} from "./shared/persist-root-lock"; import { CacheHeaders, CoreBindings, @@ -902,7 +912,7 @@ export class Miniflare { this.#devRegistry = new DevRegistry( this.#sharedOpts.unsafeDevRegistryPath, (registry) => { - void this.#pushRegistryUpdate(); + void this.#queueRegistryUpdate(); this.#sharedOpts.unsafeHandleDevRegistryUpdate?.(registry); }, this.#log @@ -958,7 +968,20 @@ export class Miniflare { this.#disposeController = new AbortController(); this.#runtimeMutex = new Mutex(); this.#initPromise = this.#runtimeMutex - .runWith(() => this.#assembleAndUpdateConfig()) + .runWith(async () => { + if ( + this.#sharedOpts.unsafeEnableSharedStorage && + this.#sharedOpts.resourcePersistencePath !== undefined + ) { + this.#sharedOpts = { + ...this.#sharedOpts, + resourcePersistencePath: await canonicalisePersistRoot( + this.#sharedOpts.resourcePersistencePath + ), + }; + } + await this.#assembleAndUpdateConfig(); + }) .catch((e) => { // If initialisation failed, attempting to `dispose()` this instance // will too. Therefore, remove from the instance registry now, so we @@ -991,6 +1014,15 @@ export class Miniflare { */ #devRegistryDispatcher?: Dispatcher; #devRegistryPort?: number; + #registryPushPromise: Promise = Promise.resolve(); + + #queueRegistryUpdate(): Promise { + this.#registryPushPromise = this.#registryPushPromise.then( + () => this.#pushRegistryUpdate(), + () => this.#pushRegistryUpdate() + ); + return this.#registryPushPromise; + } async #pushRegistryUpdate(retries = 3): Promise { if (this.#disposeController.signal.aborted) return; @@ -1129,11 +1161,10 @@ export class Miniflare { ); assert(namespaceId, "Namespace ID is required"); - const coreSharedOpts = this.#sharedOpts; const doPersistPath = getPersistPath( DURABLE_OBJECTS_PLUGIN_NAME, this.#tmpPath, - coreSharedOpts.resourcePersistencePath + this.#sharedOpts.isolatedResourcePersistencePath ); const namespacePath = path.join(doPersistPath, namespaceId); @@ -1176,11 +1207,10 @@ export class Miniflare { ); assert(workflowName, "Workflow name is required"); - const coreSharedOpts = this.#sharedOpts; const workflowsPersistPath = getPersistPath( WORKFLOWS_PLUGIN_NAME, this.#tmpPath, - coreSharedOpts.resourcePersistencePath + this.#sharedOpts.isolatedResourcePersistencePath ); // Engine DOs are stored under: /miniflare-workflows-/.sqlite @@ -1387,11 +1417,10 @@ export class Miniflare { return new Response("Instance ID is required", { status: 400 }); } - const coreSharedOpts = this.#sharedOpts; const workflowsPersistPath = getPersistPath( WORKFLOWS_PLUGIN_NAME, this.#tmpPath, - coreSharedOpts.resourcePersistencePath + this.#sharedOpts.isolatedResourcePersistencePath ); const uniqueKey = `miniflare-workflows-${workflowName}`; @@ -1544,15 +1573,23 @@ export class Miniflare { const separator = dim("━".repeat(76)); this.#log.warn( `\n${separator}\n` + - `${bold(yellow("Cloudflare Access blocked a remote bindings request"))}\n` + + `${bold( + yellow("Cloudflare Access blocked a remote bindings request") + )}\n` + `${separator}\n` + `\n` + - `Remote binding "${bold(bindingName)}": request to ${proxyUrl} was blocked.\n` + + `Remote binding "${bold( + bindingName + )}": request to ${proxyUrl} was blocked.\n` + `\n` + `If your Cloudflare account protects workers.dev with Access, set the\n` + - `${bold("CLOUDFLARE_ACCESS_CLIENT_ID")} and ${bold("CLOUDFLARE_ACCESS_CLIENT_SECRET")}\n` + + `${bold("CLOUDFLARE_ACCESS_CLIENT_ID")} and ${bold( + "CLOUDFLARE_ACCESS_CLIENT_SECRET" + )}\n` + `environment variables (Service Token credentials), or run\n` + - ` ${bold("cloudflared access login ")}\n` + + ` ${bold( + "cloudflared access login " + )}\n` + `for interactive authentication.\n` + `\n` + `See https://developers.cloudflare.com/cloudflare-one/access-controls/service-credentials/service-tokens/\n` + @@ -1632,7 +1669,12 @@ export class Miniflare { // Used by the local explorer to aggregate resources across instances const registryPath = this.#devRegistry.getRegistryPath(); const registry = registryPath ? getWorkerRegistry(registryPath) : {}; - response = Response.json(registry); + response = Response.json(registry, { + headers: { + "X-Miniflare-Dev-Registry-Instance-Id": + this.#devRegistry.instanceId, + }, + }); } else if (url.pathname === "/core/public-url") { // Returns the public URL for this Miniflare instance. If a publicUrl // has been set (e.g. the Vite dev server URL), use that; otherwise @@ -1894,7 +1936,6 @@ export class Miniflare { async #assembleConfig( loopbackHost: string, loopbackPort: number, - devRegistryEnabled: boolean, reusePorts: boolean ): Promise { const allPreviousWorkerOpts = this.#previousWorkerOpts; @@ -1906,7 +1947,7 @@ export class Miniflare { sharedOpts.cf = await setupCf(this.#log, sharedOpts.cf); this.#cfObject = sharedOpts.cf; - const externalServices = devRegistryEnabled + const externalServices = this.#devRegistry.isEnabled() ? getExternalServiceEntrypoints(allWorkerOpts) : null; @@ -1992,6 +2033,7 @@ export class Miniflare { for (const [key, plugin] of this.#mergedPluginEntries) { const pluginBindings = await plugin.getBindings( workerOpts, + sharedOpts, workerIndex ); if (pluginBindings !== undefined) { @@ -2053,7 +2095,7 @@ export class Miniflare { const pluginServicesOptionsBase: Omit< PluginServicesOptions, - "options" | "sharedOptions" + "options" | "sharedOptions" | "devRegistryEnabled" > = { log: this.#log, workerBindings, @@ -2067,7 +2109,6 @@ export class Miniflare { unsafeEphemeralDurableObjects, queueProducers, queueConsumers, - devRegistryEnabled, containerPrivilegesCache: this.#containerPrivilegesCache, hyperdriveProxyController: this.#hyperdriveProxyController, }; @@ -2076,6 +2117,7 @@ export class Miniflare { ...pluginServicesOptionsBase, options: workerOpts, sharedOptions: sharedOpts, + devRegistryEnabled: this.#devRegistry.isEnabled(), }); if (pluginServicesExtensions !== undefined) { let pluginServices: Service[]; @@ -2179,10 +2221,17 @@ export class Miniflare { if ( this.#devRegistry.isEnabled() && - externalServices && - (externalServices.size > 0 || hasQueues) + externalServices !== null && + (externalServices.size > 0 || + hasQueues || + sharedOpts.unsafeEnableSharedStorage) ) { - await this.#devRegistry.watch(externalServices, hasQueues); + await this.#devRegistry.watch( + externalServices, + hasQueues, + sharedOpts.unsafeEnableSharedStorage === true && + sharedOpts.resourcePersistencePath !== undefined + ); const externalObjects = Array.from(externalServices).flatMap( ([scriptName, { classNames }]) => @@ -2241,6 +2290,10 @@ export class Miniflare { // workerdDebugPort bindings don't have any additional configuration workerdDebugPort: kVoid, }, + { + name: CoreBindings.DEV_REGISTRY_INSTANCE_ID, + text: this.#devRegistry.instanceId, + }, ], durableObjectStorage: { inMemory: kVoid }, // uniqueKey must match the target session's key for identical DO IDs. @@ -2322,11 +2375,16 @@ export class Miniflare { // unexplained dev server restart. Always say something: any crash is a // bug worth reporting, and the count distinguishes a one-off from a loop. this.#log.warn( - `The Workers runtime crashed unexpectedly and is being restarted (crash #${this.#workerdCrashCount}). ` + - "Any additional runtime output above may indicate the cause." + `The Workers runtime crashed unexpectedly and is being restarted (crash #${ + this.#workerdCrashCount + }). ` + "Any additional runtime output above may indicate the cause." ); // A crash destroys the proxy server heap just like a config update. this.#proxyClient?.poisonProxies(); + // The runtime behind this candidate is gone. Withdraw before restarting so + // peers can take ownership if recovery stalls; successful assembly registers + // this instance again with its new debug-port address. + this.#devRegistry.unregisterWorkers(); void this.#runtimeMutex .runWith(async () => { try { @@ -2381,6 +2439,7 @@ export class Miniflare { // This function must be run with `#runtimeMutex` held const initial = !this.#runtimeEntryURL; assert(this.#runtime !== undefined); + const runtime = this.#runtime; const configuredHost = this.#sharedOpts.host ?? DEFAULT_HOST; // For internal loopback communication with workerd, always use 127.0.0.1 // when localhost is configured. This prevents IPv6/IPv4 mismatch issues @@ -2394,7 +2453,6 @@ export class Miniflare { const config = await this.#assembleConfig( loopbackHost, loopbackPort, - this.#devRegistry.isEnabled(), reusePorts ); const configBuffer = serializeConfig(config); @@ -2464,11 +2522,17 @@ export class Miniflare { onWorkerdCrashRestart: () => this.#handleWorkerdCrash(), runtimeEnv: this.#sharedOpts.unsafeRuntimeEnv, }; - const maybeSocketPorts = await this.#runtime.updateConfig( - configBuffer, - runtimeOpts, - this.#workerOpts.map((w) => w.config.name), - this.#disposeController.signal + const maybeSocketPorts = await withPersistRootStartupLock( + this.#sharedOpts.unsafeEnableSharedStorage + ? this.#sharedOpts.resourcePersistencePath + : undefined, + () => + runtime.updateConfig( + configBuffer, + runtimeOpts, + this.#workerOpts.map((w) => w.config.name), + this.#disposeController.signal + ) ); if (this.#disposeController.signal.aborted) return; if (maybeSocketPorts === undefined) { @@ -2567,7 +2631,7 @@ export class Miniflare { // Catch any registry updates that occurred while workerd was booting. if (this.#devRegistry.isEnabled()) { - await this.#pushRegistryUpdate(); + await this.#queueRegistryUpdate(); } if (!this.#runtimeMutex.hasWaiting) { @@ -2672,6 +2736,20 @@ export class Miniflare { ); const entries: [string, WorkerDefinition][] = []; + const storageScope = this.#sharedOpts.unsafeEnableSharedStorage + ? getStorageScope(this.#sharedOpts.resourcePersistencePath) + : undefined; + if (storageScope !== undefined) { + entries.push([ + getStorageCandidateName(this.#devRegistry.instanceId), + { + debugPortAddress, + defaultEntrypointService: "", + userWorkerService: "", + storageScope, + }, + ]); + } for (const workerOpts of this.#workerOpts) { const workerName = workerOpts.config.name; if (!workerName || !workerOpts.dev?.unsafeRegisterWorker) { @@ -2798,7 +2876,19 @@ export class Miniflare { // This function must be run with `#runtimeMutex` held // Split and validate options - const [sharedOpts, workerOpts] = validateOptions(opts); + const [initialSharedOpts, workerOpts] = validateOptions(opts); + let sharedOpts = initialSharedOpts; + if ( + sharedOpts.unsafeEnableSharedStorage && + sharedOpts.resourcePersistencePath !== undefined + ) { + sharedOpts = { + ...sharedOpts, + resourcePersistencePath: await canonicalisePersistRoot( + sharedOpts.resourcePersistencePath + ), + }; + } this.#previousSharedOpts = this.#sharedOpts; this.#previousWorkerOpts = this.#workerOpts; this.#sharedOpts = sharedOpts; @@ -2810,7 +2900,7 @@ export class Miniflare { await this.#devRegistry.updateRegistryPath( sharedOpts.unsafeDevRegistryPath, (registry) => { - void this.#pushRegistryUpdate(); + void this.#queueRegistryUpdate(); newExternalOnUpdate?.(registry); } ); @@ -2997,7 +3087,9 @@ export class Miniflare { // corresponding route service binding. assert( fetcher !== undefined, - `Expected ${bindingName} service binding for worker ${JSON.stringify(workerName)}` + `Expected ${bindingName} service binding for worker ${JSON.stringify( + workerName + )}` ); return fetcher as ReplaceWorkersTypes; } @@ -3027,7 +3119,9 @@ export class Miniflare { ? `${bindingTypeDescription} binding` : "binding"; throw new TypeError( - `No ${bindingType} named ${JSON.stringify(bindingName)} found in ${friendlyWorkerName}.` + `No ${bindingType} named ${JSON.stringify( + bindingName + )} found in ${friendlyWorkerName}.` ); } return proxy as T; @@ -3144,7 +3238,9 @@ export class Miniflare { if (!durableObjectExists) { throw new TypeError( - `No Durable Object class named ${JSON.stringify(className)} found in ${JSON.stringify(scriptName)} worker.` + `No Durable Object class named ${JSON.stringify( + className + )} found in ${JSON.stringify(scriptName)} worker.` ); } @@ -3185,7 +3281,9 @@ export class Miniflare { ? `${JSON.stringify(resolvedWorkerName)} worker` : "the worker"; throw new TypeError( - `No Durable Object class or namespace binding named ${JSON.stringify(classNameOrBindingName)} found in ${friendlyWorkerName}.` + `No Durable Object class or namespace binding named ${JSON.stringify( + classNameOrBindingName + )} found in ${friendlyWorkerName}.` ); } @@ -3203,14 +3301,16 @@ export class Miniflare { if (namespaceKey === undefined) { throw new TypeError( - `Cannot list Durable Object ids for ${JSON.stringify(classNameOrBindingName)} because the namespace uses ephemeral local storage.` + `Cannot list Durable Object ids for ${JSON.stringify( + classNameOrBindingName + )} because the namespace uses ephemeral local storage.` ); } const durableObjectsPersistPath = getPersistPath( DURABLE_OBJECTS_PLUGIN_NAME, this.#tmpPath, - this.#sharedOpts.resourcePersistencePath + this.#sharedOpts.isolatedResourcePersistencePath ); try { @@ -3306,11 +3406,25 @@ export class Miniflare { } /** @internal */ - _getInternalDurableObjectNamespace( + async _getInternalDurableObjectNamespace( pluginName: string, serviceName: string, className: string ): Promise> { + if ( + this.#sharedOpts.unsafeEnableSharedStorage && + [ + D1_PLUGIN_NAME, + KV_PLUGIN_NAME, + R2_PLUGIN_NAME, + RATELIMIT_PLUGIN_NAME, + SECRET_STORE_PLUGIN_NAME, + ].includes(pluginName) + ) { + throw new TypeError( + "Direct internal storage access is unavailable while shared storage is enabled" + ); + } return this.#getProxy(`${pluginName}-internal`, className, serviceName); } @@ -3369,6 +3483,7 @@ export class Miniflare { } const runtimeCleanupOutcome = await runtimeDisposeOutcome; + this.#devRegistry.unregisterWorkers(); try { await Promise.all( [...this.#pendingWorkflowStorageDeletes.values()].map( diff --git a/packages/miniflare/src/plugins/cache/index.ts b/packages/miniflare/src/plugins/cache/index.ts index 8ca375e6913..aaf923ccee5 100644 --- a/packages/miniflare/src/plugins/cache/index.ts +++ b/packages/miniflare/src/plugins/cache/index.ts @@ -76,7 +76,7 @@ export const CACHE_PLUGIN: Plugin = { const persistPath = getPersistPath( CACHE_PLUGIN_NAME, tmpPath, - sharedOptions.resourcePersistencePath + sharedOptions.isolatedResourcePersistencePath ); await fs.mkdir(persistPath, { recursive: true }); const storageService: Service = { diff --git a/packages/miniflare/src/plugins/core/explorer.ts b/packages/miniflare/src/plugins/core/explorer.ts index fbff9d5e6f2..a9aa59c5896 100644 --- a/packages/miniflare/src/plugins/core/explorer.ts +++ b/packages/miniflare/src/plugins/core/explorer.ts @@ -215,17 +215,22 @@ export function constructExplorerBindingMap( binding.name?.startsWith( `${CoreBindings.DURABLE_OBJECT_NAMESPACE_PROXY}:kv:` ) && - "kvNamespace" in binding && - binding.kvNamespace?.name?.startsWith("kv:ns:") + "kvNamespace" in binding ) { - const namespaceId = - extractObjectEntryId(binding.kvNamespace.props?.json) ?? - binding.kvNamespace.name.replace(/^kv:ns:/, ""); + const namespaceId = extractObjectEntryId( + binding.kvNamespace?.props?.json + ); + const fallbackNamespaceId = binding.kvNamespace?.name?.startsWith( + "kv:ns:" + ) + ? binding.kvNamespace.name.replace(/^kv:ns:/, "") + : undefined; // Remote namespaces share one proxy service ("kv:ns:remote"). Remote // resources aren't surfaced in the explorer, so skip them — otherwise // they'd all collide under the literal id "remote". - if (namespaceId !== "remote") { - IDToBindingName.kv[namespaceId] = binding.name; + const id = namespaceId ?? fallbackNamespaceId; + if (id !== undefined && id !== "remote") { + IDToBindingName.kv[id] = binding.name; } } @@ -236,17 +241,20 @@ export function constructExplorerBindingMap( binding.name?.startsWith( `${CoreBindings.DURABLE_OBJECT_NAMESPACE_PROXY}:r2:` ) && - "r2Bucket" in binding && - binding.r2Bucket?.name?.startsWith("r2:bucket:") + "r2Bucket" in binding ) { - const bucketName = - extractObjectEntryId(binding.r2Bucket.props?.json) ?? - binding.r2Bucket.name.replace(/^r2:bucket:/, ""); + const bucketName = extractObjectEntryId(binding.r2Bucket?.props?.json); + const fallbackBucketName = binding.r2Bucket?.name?.startsWith( + "r2:bucket:" + ) + ? binding.r2Bucket.name.replace(/^r2:bucket:/, "") + : undefined; // Remote buckets share one proxy service ("r2:bucket:remote"). Remote // resources aren't surfaced in the explorer, so skip them — otherwise // they'd all collide under the literal id "remote". - if (bucketName !== "remote") { - IDToBindingName.r2[bucketName] = binding.name; + const name = bucketName ?? fallbackBucketName; + if (name !== undefined && name !== "remote") { + IDToBindingName.r2[name] = binding.name; } } } diff --git a/packages/miniflare/src/plugins/core/index.ts b/packages/miniflare/src/plugins/core/index.ts index 4dbd5727783..b1a3665cb23 100644 --- a/packages/miniflare/src/plugins/core/index.ts +++ b/packages/miniflare/src/plugins/core/index.ts @@ -360,7 +360,7 @@ function getServiceBindings( } export const CORE_PLUGIN: Plugin = { - getBindings(options, workerIndex) { + getBindings(options, _sharedOptions, workerIndex) { const { config, legacy, dev } = options; const bindings: Awaitable[] = []; @@ -820,14 +820,17 @@ export function getGlobalServices({ }, }); } - const r2PublicService = getR2PublicService(allWorkerOpts ?? []); + const r2PublicService = getR2PublicService( + allWorkerOpts ?? [], + sharedOptions + ); if (r2PublicService !== undefined) { serviceEntryBindings.push({ name: CoreBindings.SERVICE_R2_PUBLIC, service: { name: R2_PUBLIC_SERVICE_NAME }, }); } - const r2S3Service = getR2S3Service(allWorkerOpts ?? []); + const r2S3Service = getR2S3Service(allWorkerOpts ?? [], sharedOptions); if (r2S3Service !== undefined) { serviceEntryBindings.push({ name: CoreBindings.SERVICE_R2_S3, @@ -1003,7 +1006,7 @@ export function getGlobalServices({ services.push( ...getObservabilityServices( tmpPath, - sharedOptions.resourcePersistencePath + sharedOptions.isolatedResourcePersistencePath ) ); } diff --git a/packages/miniflare/src/plugins/core/observability.ts b/packages/miniflare/src/plugins/core/observability.ts index 52e9d8c972a..df5e4a0dcd8 100644 --- a/packages/miniflare/src/plugins/core/observability.ts +++ b/packages/miniflare/src/plugins/core/observability.ts @@ -24,7 +24,7 @@ const OBSERVABILITY_STORAGE_SERVICE_NAME = "obs:storage"; export function getObservabilityServices( tmpPath: string, - resourcePersistencePath: string | undefined + isolatedResourcePersistencePath: string | undefined ): Service[] { // The TraceStore DO is SQLite-backed, so it needs disk-backed storage (the // in-memory option doesn't support SQL). Persist under `.wrangler/state` when @@ -33,7 +33,7 @@ export function getObservabilityServices( const storagePath = getPersistPath( "observability", tmpPath, - resourcePersistencePath + isolatedResourcePersistencePath ); mkdirSync(storagePath, { recursive: true }); diff --git a/packages/miniflare/src/plugins/d1/index.ts b/packages/miniflare/src/plugins/d1/index.ts index 22a05c4f02b..4b03d71edf7 100644 --- a/packages/miniflare/src/plugins/d1/index.ts +++ b/packages/miniflare/src/plugins/d1/index.ts @@ -8,6 +8,7 @@ import { getMiniflareObjectBindings, getPersistPath, getRemoteProxyConnectionString, + getStorageService, objectEntryWorker, ProxyNodeBinding, remoteProxyClientWorker, @@ -36,7 +37,7 @@ const D1_DATABASE_OBJECT: Worker_Binding_DurableObjectNamespaceDesignator = { export const D1_PLUGIN: Plugin = { bindingTypeDescription: "D1 database", - getBindings(options) { + getBindings(options, sharedOptions) { return getEnvBindingsOfType(options.config, "d1").map( ([name, binding]) => { const id = binding.id; @@ -52,10 +53,11 @@ export const D1_PLUGIN: Plugin = { name: D1_REMOTE_SERVICE_NAME, props: buildRemoteProxyProps(remoteProxyConnectionString, name), } - : { - name: D1_LOCAL_ENTRY_SERVICE_NAME, - props: buildObjectEntryProps(id), - }; + : getStorageService( + D1_LOCAL_ENTRY_SERVICE_NAME, + buildObjectEntryProps(id), + sharedOptions + ); return { name, @@ -86,9 +88,11 @@ export const D1_PLUGIN: Plugin = { const services: Service[] = []; // One shared entry service for all local databases (id supplied via props). - const hasLocal = databases.some( - ([, db]) => getRemoteProxyConnectionString(db, options.dev) === undefined - ); + const hasLocal = + databases.some( + ([, db]) => + getRemoteProxyConnectionString(db, options.dev) === undefined + ) || sharedOptions.unsafeEnableSharedStorage; if (hasLocal) { services.push({ name: D1_LOCAL_ENTRY_SERVICE_NAME, @@ -155,7 +159,6 @@ export const D1_PLUGIN: Plugin = { }; services.push(storageService, objectService); } - return services; }, }; diff --git a/packages/miniflare/src/plugins/do/index.ts b/packages/miniflare/src/plugins/do/index.ts index aebf3ac4dad..29587873d07 100644 --- a/packages/miniflare/src/plugins/do/index.ts +++ b/packages/miniflare/src/plugins/do/index.ts @@ -68,7 +68,7 @@ export const DURABLE_OBJECTS_PLUGIN: Plugin = { const storagePath = getPersistPath( DURABLE_OBJECTS_PLUGIN_NAME, tmpPath, - sharedOptions.resourcePersistencePath + sharedOptions.isolatedResourcePersistencePath ); // `workerd` requires the `disk.path` to exist. Setting `recursive: true` // is like `mkdir -p`: it won't fail if the directory already exists, and it diff --git a/packages/miniflare/src/plugins/hello-world/index.ts b/packages/miniflare/src/plugins/hello-world/index.ts index cbcde15ff95..55dbba59689 100644 --- a/packages/miniflare/src/plugins/hello-world/index.ts +++ b/packages/miniflare/src/plugins/hello-world/index.ts @@ -46,7 +46,7 @@ export const HELLO_WORLD_PLUGIN: Plugin = { const persistPath = getPersistPath( HELLO_WORLD_PLUGIN_NAME, tmpPath, - sharedOptions.resourcePersistencePath + sharedOptions.isolatedResourcePersistencePath ); await fs.mkdir(persistPath, { recursive: true }); diff --git a/packages/miniflare/src/plugins/images/index.ts b/packages/miniflare/src/plugins/images/index.ts index 603ddf7fa8d..43c8299e3ea 100644 --- a/packages/miniflare/src/plugins/images/index.ts +++ b/packages/miniflare/src/plugins/images/index.ts @@ -88,7 +88,7 @@ export const IMAGES_PLUGIN: Plugin = { const persistPath = getPersistPath( IMAGES_PLUGIN_NAME, tmpPath, - sharedOptions.resourcePersistencePath + sharedOptions.isolatedResourcePersistencePath ); await fs.mkdir(persistPath, { recursive: true }); diff --git a/packages/miniflare/src/plugins/kv/index.ts b/packages/miniflare/src/plugins/kv/index.ts index f1a961ee929..c04cb984c9b 100644 --- a/packages/miniflare/src/plugins/kv/index.ts +++ b/packages/miniflare/src/plugins/kv/index.ts @@ -2,11 +2,13 @@ import fs from "node:fs/promises"; import SCRIPT_KV_NAMESPACE_OBJECT from "worker:kv/namespace"; import { SharedBindings } from "../../workers"; import { + buildObjectEntryProps, buildRemoteProxyProps, getEnvBindingsOfType, getMiniflareObjectBindings, getPersistPath, getRemoteProxyConnectionString, + getStorageService, objectEntryWorker, ProxyNodeBinding, remoteProxyClientWorker, @@ -47,7 +49,7 @@ function isWorkersSitesEnabled( export const KV_PLUGIN: Plugin = { bindingTypeDescription: "KV namespace", - async getBindings(options) { + async getBindings(options, sharedOptions) { const namespaces = getEnvBindingsOfType(options.config, "kv"); const bindings = namespaces.map(([name, binding]) => { const id = binding.id; @@ -70,14 +72,11 @@ export const KV_PLUGIN: Plugin = { // passed at runtime via props (read in object-entry.worker.ts). return { name, - kvNamespace: { - name: KV_LOCAL_ENTRY_SERVICE_NAME, - props: { - json: JSON.stringify({ - [SharedBindings.TEXT_NAMESPACE]: id, - }), - }, - }, + kvNamespace: getStorageService( + KV_LOCAL_ENTRY_SERVICE_NAME, + buildObjectEntryProps(id), + sharedOptions + ), }; }); @@ -112,9 +111,10 @@ export const KV_PLUGIN: Plugin = { const services: Service[] = []; // One shared entry service for all local namespaces (id supplied via props). - const hasLocalNamespace = namespaces.some( - ([, binding]) => !getRemoteProxyConnectionString(binding, options.dev) - ); + const hasLocalNamespace = + namespaces.some( + ([, binding]) => !getRemoteProxyConnectionString(binding, options.dev) + ) || sharedOptions.unsafeEnableSharedStorage; if (hasLocalNamespace) { services.push({ name: KV_LOCAL_ENTRY_SERVICE_NAME, @@ -177,7 +177,6 @@ export const KV_PLUGIN: Plugin = { }; services.push(storageService, objectService); } - if (isWorkersSitesEnabled(options)) { services.push(...getSitesServices(options.legacy, options.dev?.rootPath)); } diff --git a/packages/miniflare/src/plugins/r2/index.ts b/packages/miniflare/src/plugins/r2/index.ts index 663d373a99f..ca9e85496a8 100644 --- a/packages/miniflare/src/plugins/r2/index.ts +++ b/packages/miniflare/src/plugins/r2/index.ts @@ -12,6 +12,7 @@ import { getMiniflareObjectBindings, getPersistPath, getRemoteProxyConnectionString, + getStorageService, objectEntryWorker, ProxyNodeBinding, remoteProxyClientWorker, @@ -22,7 +23,12 @@ import type { Worker_Binding, Worker_Binding_DurableObjectNamespaceDesignator, } from "../../runtime"; -import type { MiniflareBinding, ParsedWorkerOptions, Plugin } from "../shared"; +import type { + MiniflareBinding, + ParsedInstanceOptions, + ParsedWorkerOptions, + Plugin, +} from "../shared"; /** Local-dev S3 credentials, derived from the parsed R2 binding. */ type R2S3Credentials = NonNullable< @@ -48,7 +54,11 @@ const R2_BUCKET_OBJECT: Worker_Binding_DurableObjectNamespaceDesignator = { }; export function getR2PublicService( - allWorkerOpts: ParsedWorkerOptions[] + allWorkerOpts: ParsedWorkerOptions[], + sharedOptions: Pick< + ParsedInstanceOptions, + "resourcePersistencePath" | "unsafeEnableSharedStorage" + > ): Service | undefined { const publicBucketIds = new Set(); for (const worker of allWorkerOpts) { @@ -64,10 +74,11 @@ export function getR2PublicService( } const bindings = Array.from(publicBucketIds).map((id) => ({ name: id, - r2Bucket: { - name: R2_LOCAL_ENTRY_SERVICE_NAME, - props: buildObjectEntryProps(id), - }, + r2Bucket: getStorageService( + R2_LOCAL_ENTRY_SERVICE_NAME, + buildObjectEntryProps(id), + sharedOptions + ), })); return { name: R2_PUBLIC_SERVICE_NAME, @@ -80,7 +91,11 @@ export function getR2PublicService( } export function getR2S3Service( - allWorkerOpts: ParsedWorkerOptions[] + allWorkerOpts: ParsedWorkerOptions[], + sharedOptions: Pick< + ParsedInstanceOptions, + "resourcePersistencePath" | "unsafeEnableSharedStorage" + > ): Service | undefined { const credentialsById: Record = {}; for (const worker of allWorkerOpts) { @@ -117,10 +132,11 @@ export function getR2S3Service( const bindings = bucketIds.map((id) => ({ name: `${R2S3Bindings.BUCKET_PREFIX}${id}`, - r2Bucket: { - name: R2_LOCAL_ENTRY_SERVICE_NAME, - props: buildObjectEntryProps(id), - }, + r2Bucket: getStorageService( + R2_LOCAL_ENTRY_SERVICE_NAME, + buildObjectEntryProps(id), + sharedOptions + ), })); bindings.push({ name: R2S3Bindings.JSON_CREDENTIALS, @@ -140,7 +156,7 @@ export function getR2S3Service( export const R2_PLUGIN: Plugin = { bindingTypeDescription: "R2 bucket", - getBindings(options) { + getBindings(options, sharedOptions) { return getEnvBindingsOfType(options.config, "r2").map( ([name, bucket]) => { const id = bucket.name; @@ -155,10 +171,11 @@ export const R2_PLUGIN: Plugin = { name: R2_REMOTE_SERVICE_NAME, props: buildRemoteProxyProps(remoteProxyConnectionString, name), } - : { - name: R2_LOCAL_ENTRY_SERVICE_NAME, - props: buildObjectEntryProps(id), - }, + : getStorageService( + R2_LOCAL_ENTRY_SERVICE_NAME, + buildObjectEntryProps(id), + sharedOptions + ), }; } ); @@ -177,9 +194,10 @@ export const R2_PLUGIN: Plugin = { const services: Service[] = []; // One shared entry service for all local buckets (id supplied via props). - const hasLocal = buckets.some( - ([, b]) => getRemoteProxyConnectionString(b, options.dev) === undefined - ); + const hasLocal = + buckets.some( + ([, b]) => getRemoteProxyConnectionString(b, options.dev) === undefined + ) || sharedOptions.unsafeEnableSharedStorage; if (hasLocal) { services.push({ name: R2_LOCAL_ENTRY_SERVICE_NAME, @@ -245,7 +263,6 @@ export const R2_PLUGIN: Plugin = { }; services.push(storageService, objectService); } - return services; }, }; diff --git a/packages/miniflare/src/plugins/ratelimit/index.ts b/packages/miniflare/src/plugins/ratelimit/index.ts index 7df83f26b6f..97eae86372a 100644 --- a/packages/miniflare/src/plugins/ratelimit/index.ts +++ b/packages/miniflare/src/plugins/ratelimit/index.ts @@ -7,6 +7,7 @@ import { getEnvBindingsOfType, getMiniflareObjectBindings, getPersistPath, + getStorageService, objectEntryWorker, ProxyNodeBinding, SERVICE_LOOPBACK, @@ -46,7 +47,7 @@ function buildJsonBindings(bindings: Record): Worker_Binding[] { export const RATELIMIT_PLUGIN: Plugin = { bindingTypeDescription: "Rate Limit", - getBindings(options) { + getBindings(options, sharedOptions) { return getEnvBindingsOfType( options.config, "rate-limit" @@ -57,10 +58,11 @@ export const RATELIMIT_PLUGIN: Plugin = { innerBindings: [ { name: "fetcher", - service: { - name: RATELIMIT_LOCAL_ENTRY_SERVICE_NAME, - props: buildObjectEntryProps(binding.namespace), - }, + service: getStorageService( + RATELIMIT_LOCAL_ENTRY_SERVICE_NAME, + buildObjectEntryProps(binding.namespace), + sharedOptions + ), }, ...buildJsonBindings({ limit: binding.simple.limit, @@ -80,10 +82,9 @@ export const RATELIMIT_PLUGIN: Plugin = { }, async getServices({ options, tmpPath, sharedOptions }) { const ratelimits = getEnvBindingsOfType(options.config, "rate-limit"); - if (ratelimits.length === 0) { + if (ratelimits.length === 0 && !sharedOptions.unsafeEnableSharedStorage) { return []; } - // Each namespace is supplied per-binding via props, so one service serves // every rate limiter while shared namespaces still use the same DO name. const services: Service[] = [ diff --git a/packages/miniflare/src/plugins/secret-store/index.ts b/packages/miniflare/src/plugins/secret-store/index.ts index 1391eb9205d..a999684770c 100644 --- a/packages/miniflare/src/plugins/secret-store/index.ts +++ b/packages/miniflare/src/plugins/secret-store/index.ts @@ -8,6 +8,7 @@ import { getEnvBindingsOfType, getMiniflareObjectBindings, getPersistPath, + getStorageService, getUserBindingServiceName, objectEntryWorker, ProxyNodeBinding, @@ -53,7 +54,7 @@ export const SECRET_STORE_PLUGIN: Plugin = { "secrets-store-secret" ).map(([, binding]) => binding); - if (configs.length === 0) { + if (configs.length === 0 && !sharedOptions.unsafeEnableSharedStorage) { return []; } @@ -126,10 +127,11 @@ export const SECRET_STORE_PLUGIN: Plugin = { bindings: [ { name: "store", - kvNamespace: { - name: SECRET_STORE_LOCAL_ENTRY_SERVICE_NAME, - props: buildObjectEntryProps(config.storeId), - }, + kvNamespace: getStorageService( + SECRET_STORE_LOCAL_ENTRY_SERVICE_NAME, + buildObjectEntryProps(config.storeId), + sharedOptions + ), }, { name: "secret_name", diff --git a/packages/miniflare/src/plugins/shared/constants.ts b/packages/miniflare/src/plugins/shared/constants.ts index 886c07634bb..66dbe5ebd8b 100644 --- a/packages/miniflare/src/plugins/shared/constants.ts +++ b/packages/miniflare/src/plugins/shared/constants.ts @@ -96,10 +96,8 @@ export function objectEntryWorker( // The resource id travels via props so a single entry service can route to any // number of resources; it is read back in `object-entry.worker.ts` via // `ctx.props` and used as the Durable Object name (`idFromName`). -export function buildObjectEntryProps(id: string): { json: string } { - return { - json: JSON.stringify({ [SharedBindings.TEXT_NAMESPACE]: id }), - }; +export function buildObjectEntryProps(id: string): Record { + return { [SharedBindings.TEXT_NAMESPACE]: id }; } // Inverse of `buildObjectEntryProps`: reads the resource id back out of a @@ -113,7 +111,12 @@ export function extractObjectEntryId( } try { const parsed = JSON.parse(propsJson) as Record; - const id = parsed[SharedBindings.TEXT_NAMESPACE]; + const userProps = parsed.userProps; + const props = + typeof userProps === "object" && userProps !== null + ? (userProps as Record) + : parsed; + const id = props[SharedBindings.TEXT_NAMESPACE]; return typeof id === "string" ? id : undefined; } catch { return undefined; diff --git a/packages/miniflare/src/plugins/shared/index.ts b/packages/miniflare/src/plugins/shared/index.ts index 0f0e1d0c089..f1298b3b9a9 100644 --- a/packages/miniflare/src/plugins/shared/index.ts +++ b/packages/miniflare/src/plugins/shared/index.ts @@ -2,6 +2,8 @@ import { createHash } from "node:crypto"; import path from "node:path"; import { pathToFileURL } from "node:url"; import { MiniflareCoreError } from "../../shared"; +import { getUserServiceName } from "../core"; +import { SERVICE_DEV_REGISTRY_PROXY, type UnsafeUniqueKey } from "./constants"; import type { ParsedInstanceOptions, ParsedWorkerOptions, @@ -9,6 +11,7 @@ import type { import type { Extension, Service, + ServiceDesignator, Worker_Binding, Worker_Module, } from "../../runtime"; @@ -21,7 +24,6 @@ import type { import type { ContainerPrivilegesCache } from "../core/container"; import type { DOContainerOptions } from "../do"; import type { HyperdriveProxyController } from "../hyperdrive/hyperdrive-proxy"; -import type { UnsafeUniqueKey } from "./constants"; import type { z } from "zod"; // Maps workflow binding names to their workflow options @@ -94,6 +96,7 @@ export interface Plugin { bindingTypeDescription?: string; getBindings( options: ParsedWorkerOptions, + sharedOptions: ParsedInstanceOptions, workerIndex: number ): Awaitable; getNodeBindings( @@ -270,3 +273,41 @@ export type { ParsedMiniflareWorkerConfig, ParsedWorkerOptions, } from "../../config/schema"; + +export function getStorageService( + localServiceName: string, + props: Record, + sharedOptions: Pick< + ParsedInstanceOptions, + "resourcePersistencePath" | "unsafeEnableSharedStorage" + > +): ServiceDesignator { + const storageScope = getStorageScope(sharedOptions.resourcePersistencePath); + return sharedOptions.unsafeEnableSharedStorage && storageScope !== undefined + ? { + name: getUserServiceName(SERVICE_DEV_REGISTRY_PROXY), + entrypoint: "ExternalServiceProxy", + props: { + json: JSON.stringify({ + service: localServiceName, + userProps: props, + storage: true, + storageScope, + }), + }, + } + : { + name: localServiceName, + props: { + json: JSON.stringify(props), + }, + }; +} + +export function getStorageScope( + resourcePersistencePath: string | undefined +): string | undefined { + return resourcePersistencePath === undefined + ? undefined + : path.resolve(resourcePersistencePath); +} diff --git a/packages/miniflare/src/plugins/stream/index.ts b/packages/miniflare/src/plugins/stream/index.ts index 02fa79f43d3..755dfc89c78 100644 --- a/packages/miniflare/src/plugins/stream/index.ts +++ b/packages/miniflare/src/plugins/stream/index.ts @@ -76,7 +76,7 @@ export const STREAM_PLUGIN: Plugin = { const persistPath = getPersistPath( STREAM_PLUGIN_NAME, tmpPath, - sharedOptions.resourcePersistencePath + sharedOptions.isolatedResourcePersistencePath ); await fs.mkdir(persistPath, { recursive: true }); diff --git a/packages/miniflare/src/plugins/workflows/index.ts b/packages/miniflare/src/plugins/workflows/index.ts index a9e25b9b392..070d4274fd1 100644 --- a/packages/miniflare/src/plugins/workflows/index.ts +++ b/packages/miniflare/src/plugins/workflows/index.ts @@ -79,7 +79,7 @@ export const WORKFLOWS_PLUGIN: Plugin = { const persistPath = getPersistPath( WORKFLOWS_PLUGIN_NAME, tmpPath, - sharedOptions.resourcePersistencePath + sharedOptions.isolatedResourcePersistencePath ); await fs.mkdir(persistPath, { recursive: true }); // each workflow should get its own storage service diff --git a/packages/miniflare/src/shared/DEV_REGISTRY.md b/packages/miniflare/src/shared/DEV_REGISTRY.md index bb153be5c6b..58616843be3 100644 --- a/packages/miniflare/src/shared/DEV_REGISTRY.md +++ b/packages/miniflare/src/shared/DEV_REGISTRY.md @@ -40,6 +40,8 @@ Each `workerd` process exposes a debug port (`--debug-port`) that provides nativ - **Durable Object access**: Full DO lifecycle including RPC methods - **Tail event forwarding**: Trace events forwarded via RPC +The binding exposes two ways to reach a debug port. `connect(address)` dials another process over TCP, while `current()` hands back the same interface for the calling process without touching the network. `openDebugPortClient()` chooses between them by comparing the registry entry's `instanceId` against the proxy worker's `DEV_REGISTRY_INSTANCE_ID` binding, so a target that turns out to be ourselves is reached in-process. + ## Components ### Filesystem Registry (`dev-registry.ts`) @@ -56,7 +58,7 @@ type WorkerDefinition = { - **Heartbeat**: Every 30s, the file's mtime is touched to signal that the Worker is still running. - **Registration**: Named workers are advertised by default. Workers with `unsafeRegisterWorker: false` are not advertised. -- **Stale cleanup**: On every read, files older than 5 minutes are deleted (5 minutes is much longer than 30s just to provide a safe buffer) +- **Stale cleanup**: Workers heartbeat every 10 seconds and entries older than 90 seconds are deleted. - **Change detection**: Chokidar watches the registry directory. When a file changes, `refresh()` compares the new state against the previous JSON snapshot and fires `onUpdate` only if a watched external service actually changed. ### Proxy Worker (`dev-registry-proxy.worker.ts`) @@ -78,6 +80,18 @@ When the filesystem watcher detects a change to an external service, Miniflare p The push always reads the latest registry state (not a captured snapshot) and retries up to 3 times with 500ms delays. +### Shared Storage Candidates + +When `unsafeEnableSharedStorage` is enabled, each Miniflare instance registers a storage candidate in addition to its user Workers. Candidates are scoped by the canonical physical persistence root, so one registry may safely coordinate unrelated projects. The oldest live candidate for a scope receives storage traffic; every candidate hosts the generic simulators required for handoff. + +Storage candidates heartbeat every 2 seconds and expire after 10 seconds. Registry watchers also refresh on a timer so a dead candidate expires without requiring another filesystem event. Graceful disposal stops the runtime before withdrawing its candidate, preventing overlap with the replacement owner. + +KV, D1, R2, Rate Limits, and Secrets Store route through the elected candidate. Cache, user Durable Objects, Workflows, observability, and Hello World storage remain instance-local while shared mode is active and use the configured `isolatedResourcePersistencePath`, allowing that state to persist across restarts without mounting the shared owner root concurrently. + +Every instance -- including the elected owner -- routes these bindings through the proxy worker, so the owner would otherwise connect back to its own debug port over TCP. Because the registry entry carries the owner's `instanceId`, the proxy recognises that case and uses the in-process debug port instead. + +Shared mode requires `resourcePersistencePath`, `isolatedResourcePersistencePath`, and `unsafeDevRegistryPath`. The shared persistence root is created and resolved through the filesystem before it is used as an ownership scope. + ## Request Flow ### Service Binding Fetch @@ -113,7 +127,7 @@ sequenceDiagram B-->>S: result S-->>A: result - Note over S: Fetcher cached per instance.
Invalidated when debugPortAddress changes. + Note over S: Fetcher cached per instance.
Invalidated when address or instance ID changes. ``` ### Scheduled Event diff --git a/packages/miniflare/src/shared/dev-registry-types.ts b/packages/miniflare/src/shared/dev-registry-types.ts index f25ed2410b3..71b8cbb0bc2 100644 --- a/packages/miniflare/src/shared/dev-registry-types.ts +++ b/packages/miniflare/src/shared/dev-registry-types.ts @@ -1,4 +1,11 @@ -export type WorkerRegistry = Record; +export type WorkerRegistry = Record< + string, + WorkerDefinition & { + // stat.birthtime + created: number; + instanceId?: string; + } +>; export type WorkerDefinition = { /** @@ -24,4 +31,6 @@ export type WorkerDefinition = { * this to route messages for these queues to this process's queue broker. */ queueConsumers?: string[]; + /** Canonical persistence root when this instance can own shared storage. */ + storageScope?: string; }; diff --git a/packages/miniflare/src/shared/dev-registry.ts b/packages/miniflare/src/shared/dev-registry.ts index 547984a8f3f..85911d8fa41 100644 --- a/packages/miniflare/src/shared/dev-registry.ts +++ b/packages/miniflare/src/shared/dev-registry.ts @@ -14,13 +14,31 @@ import { getGlobalConfigPath } from "@cloudflare/workers-utils"; import { watch } from "chokidar"; import type { WorkerDefinition, WorkerRegistry } from "./dev-registry-types"; export type { WorkerDefinition, WorkerRegistry }; +import { randomUUID } from "node:crypto"; import type { Log } from "./log"; import type { FSWatcher } from "chokidar"; +export const STORAGE_CANDIDATE_PREFIX = "__miniflare_storage_candidate__-"; +const STORAGE_CANDIDATE_HEARTBEAT_MS = 2_000; +const STORAGE_CANDIDATE_STALE_MS = 10_000; +const WORKER_HEARTBEAT_MS = 10_000; +const WORKER_STALE_MS = 90_000; + +export function getStorageCandidateName(instanceId: string): string { + return `${STORAGE_CANDIDATE_PREFIX}${instanceId}`; +} + +export function isStorageCandidateName(name: string): boolean { + return name.startsWith(STORAGE_CANDIDATE_PREFIX); +} + export class DevRegistry { private heartbeats = new Map(); + private registrationRetries = new Map(); private registeredWorkers: Set = new Set(); + private storageCandidateRefresh: NodeJS.Timeout | undefined; private watchQueueConsumers = false; + private watchStorageCandidates = false; private externalServices: Map< string, { @@ -30,11 +48,16 @@ export class DevRegistry { > = new Map(); private watcher: FSWatcher | undefined; + // UUID to let us tell whether a dev registry entry was added by _this_ Miniflare process + public instanceId: string; + constructor( private registryPath: string | undefined, private onUpdate: ((registry: WorkerRegistry) => void) | undefined, private log: Log - ) {} + ) { + this.instanceId = randomUUID(); + } /** * Watch files inside the registry directory for changes. @@ -47,14 +70,27 @@ export class DevRegistry { entrypoints: Set; } >, - watchQueueConsumers = false + watchQueueConsumers = false, + watchStorageCandidates = false ): void { - if ((services.size === 0 && !watchQueueConsumers) || !this.registryPath) { + if ( + (services.size === 0 && + !watchQueueConsumers && + !watchStorageCandidates) || + !this.registryPath + ) { return; } this.externalServices = new Map(services); this.watchQueueConsumers = watchQueueConsumers; + this.watchStorageCandidates = watchStorageCandidates; + if (watchStorageCandidates && this.storageCandidateRefresh === undefined) { + this.storageCandidateRefresh = setInterval( + () => this.refresh(), + STORAGE_CANDIDATE_HEARTBEAT_MS + ); + } mkdirSync(this.registryPath, { recursive: true }); @@ -82,6 +118,10 @@ export class DevRegistry { */ public dispose(): Promise | undefined { this.unregisterWorkers(); + if (this.storageCandidateRefresh !== undefined) { + clearInterval(this.storageCandidateRefresh); + this.storageCandidateRefresh = undefined; + } // Only this step is async and could be awaited return this.watcher?.close().finally(() => { @@ -93,6 +133,11 @@ export class DevRegistry { * Withdraw every entry this instance has registered. */ public unregisterWorkers() { + for (const retry of this.registrationRetries.values()) { + clearTimeout(retry); + } + this.registrationRetries.clear(); + for (const worker of this.registeredWorkers) { this.unregister(worker); } @@ -114,7 +159,11 @@ export class DevRegistry { } if (this.registryPath) { - unlinkSync(path.join(this.registryPath, name)); + const definitionPath = path.join(this.registryPath, name); + const definition = readDefinition(definitionPath); + if (definition?.instanceId === this.instanceId) { + unlinkSync(definitionPath); + } } } catch (e) { this.log?.debug(`Failed to unregister worker "${name}": ${e}`); @@ -154,6 +203,10 @@ export class DevRegistry { // Close the existing watcher if it exists. // It will watch the new path if there is any dependent services in a later step await this.watcher?.close(); + if (this.storageCandidateRefresh !== undefined) { + clearInterval(this.storageCandidateRefresh); + this.storageCandidateRefresh = undefined; + } this.watcher = undefined; this.registryPath = registryPath; @@ -167,6 +220,10 @@ export class DevRegistry { // Make sure the registry path exists mkdirSync(this.registryPath, { recursive: true }); + for (const retry of this.registrationRetries.values()) { + clearTimeout(retry); + } + this.registrationRetries.clear(); // Drop the entries for Workers this instance no longer has. Workers that // remain are overwritten in place below instead of being deleted and @@ -183,24 +240,84 @@ export class DevRegistry { } for (const [name, definition] of Object.entries(workers)) { - const definitionPath = path.join(this.registryPath, name); - const existingHeartbeat = this.heartbeats.get(name); - if (existingHeartbeat) { - clearInterval(existingHeartbeat); - } + this.registerWorker(name, definition); + } + this.refresh(); + } - writeFileSync(definitionPath, JSON.stringify(definition, null, 2)); - this.registeredWorkers.add(name); - this.heartbeats.set( + private registerWorker(name: string, definition: WorkerDefinition): void { + assert(this.registryPath); + const definitionPath = path.join(this.registryPath, name); + + const stats = statSync(definitionPath, { throwIfNoEntry: false }); + + const oldDefinition = stats ? readDefinition(definitionPath) : undefined; + const staleMs = isStorageCandidateName(name) + ? STORAGE_CANDIDATE_STALE_MS + : WORKER_STALE_MS; + if (stats && stats.mtime.getTime() < Date.now() - staleMs) { + try { + unlinkSync(definitionPath); + } catch {} + } else if (stats && oldDefinition?.instanceId !== this.instanceId) { + // Debug rather than warn: a non-graceful exit leaves an entry behind, so + // this fires routinely on restart. Registration is rescheduled for when + // that entry expires, and keeps retrying while another process holds the + // name. + this.log.debug( + `Skipping registration of Worker ${name} as a Worker with this name is already registered in the dev registry by another process` + ); + const retryDelay = Math.max( + 1, + stats.mtime.getTime() + staleMs - Date.now() + ); + this.registrationRetries.set( name, - setInterval(() => { - if (existsSync(definitionPath)) { - utimesSync(definitionPath, new Date(), new Date()); - } - }, 30_000) + setTimeout(() => { + this.registrationRetries.delete(name); + this.registerWorker(name, definition); + this.refresh(); + }, retryDelay) ); + return; } - this.refresh(); + + const retry = this.registrationRetries.get(name); + if (retry !== undefined) { + clearTimeout(retry); + this.registrationRetries.delete(name); + } + + const existingHeartbeat = this.heartbeats.get(name); + if (existingHeartbeat) { + clearInterval(existingHeartbeat); + } + + writeFileSync( + definitionPath, + JSON.stringify({ ...definition, instanceId: this.instanceId }, null, 2) + ); + this.registeredWorkers.add(name); + this.heartbeats.set( + name, + setInterval( + () => { + const currentDefinition = readDefinition(definitionPath); + if (currentDefinition?.instanceId === this.instanceId) { + utimesSync(definitionPath, new Date(), new Date()); + } else { + const heartbeat = this.heartbeats.get(name); + if (heartbeat !== undefined) { + clearInterval(heartbeat); + this.heartbeats.delete(name); + } + } + }, + isStorageCandidateName(name) + ? STORAGE_CANDIDATE_HEARTBEAT_MS + : WORKER_HEARTBEAT_MS + ) + ); } private previousJSON = "{}"; @@ -228,6 +345,14 @@ export class DevRegistry { this.onUpdate(registry); return; } + if ( + this.watchStorageCandidates && + getStorageCandidatesView(registry) !== + getStorageCandidatesView(previousRegistry) + ) { + this.onUpdate(registry); + return; + } for (const [service] of this.externalServices) { if ( JSON.stringify(registry[service]) !== @@ -240,6 +365,23 @@ export class DevRegistry { } } +function getStorageCandidatesView(registry: WorkerRegistry): string { + return JSON.stringify( + Object.entries(registry) + .filter(([name]) => isStorageCandidateName(name)) + .map(([name, definition]) => [ + name, + definition.instanceId, + definition.debugPortAddress, + definition.storageScope, + definition.created, + ]) + .sort(([previousName], [nextName]) => + String(previousName).localeCompare(String(nextName)) + ) + ); +} + /** * Serialise the parts of the registry that matter for routing cross-process * queue messages: which workers consume which queues, and the debug address @@ -266,7 +408,7 @@ function getQueueConsumersView(registry: WorkerRegistry): string { /** * Read the worker registry from the specified path. * - * Skips stale workers that haven't sent a heartbeat in over 5 minutes, + * Skips stale workers that haven't sent a heartbeat within their stale window, * and removes their files from disk. */ export function getWorkerRegistry(registryPath: string): WorkerRegistry { @@ -281,19 +423,27 @@ export function getWorkerRegistry(registryPath: string): WorkerRegistry { const definitionPath = path.join(registryPath, workerName); const stats = statSync(definitionPath, { throwIfNoEntry: false }); - // Cleanup old workers that have not sent a heartbeat in over 5 minutes - if (stats === undefined || stats.mtime.getTime() < Date.now() - 300_000) { + if (stats === undefined) { + continue; + } + const definition = readDefinition(definitionPath); + if (definition === undefined) { + continue; + } + const staleMs = isStorageCandidateName(workerName) + ? STORAGE_CANDIDATE_STALE_MS + : WORKER_STALE_MS; + if (stats.mtime.getTime() < Date.now() - staleMs) { try { unlinkSync(definitionPath); } catch {} continue; } - const file = readFileSync(definitionPath, { - encoding: "utf8", - flag: "r", - }); - registry[workerName] = JSON.parse(file); + registry[workerName] = { + ...definition, + created: stats.birthtimeMs, + }; } catch { // This can safely be ignored. It generally indicates the worker was too old and was removed by a parallel process } @@ -302,6 +452,20 @@ export function getWorkerRegistry(registryPath: string): WorkerRegistry { return registry; } +function readDefinition( + definitionPath: string +): (WorkerDefinition & { instanceId?: string }) | undefined { + try { + const value: unknown = JSON.parse( + readFileSync(definitionPath, { encoding: "utf8", flag: "r" }) + ); + if (typeof value === "object" && value !== null) { + return value as WorkerDefinition & { instanceId?: string }; + } + } catch {} + return undefined; +} + /** * Get the default path for the dev registry. * This is used by both Wrangler and the Vite plugin to ensure they use the same path. diff --git a/packages/miniflare/src/shared/persist-root-lock.ts b/packages/miniflare/src/shared/persist-root-lock.ts new file mode 100644 index 00000000000..c3006d39dd6 --- /dev/null +++ b/packages/miniflare/src/shared/persist-root-lock.ts @@ -0,0 +1,134 @@ +import crypto from "node:crypto"; +import fs from "node:fs/promises"; +import path from "node:path"; + +const LOCK_NAME = ".miniflare-startup.lock"; +// Startup is bounded, so an older lock belongs to a failed startup attempt. +const LOCK_STALE_MS = 30_000; +const LOCK_RETRY_MS = 50; +// Generous enough to outlast a stale lock being reclaimed, low enough that a +// lock we can never remove fails loudly instead of hanging startup forever. +const LOCK_MAX_ATTEMPTS = 2 * Math.ceil(LOCK_STALE_MS / LOCK_RETRY_MS); + +/** + * @param error - Value caught from a filesystem call. + * @param code - Errno code to test for, e.g. `"ENOENT"`. + * @returns Whether `error` is a Node filesystem error with that code. + */ +function isErrnoException(error: unknown, code: string): boolean { + return ( + error instanceof Error && + "code" in error && + (error as NodeJS.ErrnoException).code === code + ); +} + +/** + * @param lockPath - Path of the lock file. + * @returns The lock's token, or `undefined` if it doesn't exist. + */ +async function readLock(lockPath: string): Promise { + try { + return await fs.readFile(lockPath, "utf8"); + } catch (error) { + if (isErrnoException(error, "ENOENT")) { + return undefined; + } + throw error; + } +} + +/** + * Remove the lock only if we still hold it, so we never delete a lock another + * process acquired after ours was reclaimed as stale. + * + * @param lockPath - Path of the lock file. + * @param token - Token written when this process acquired the lock. + */ +async function removeOwnedLock(lockPath: string, token: string): Promise { + if ((await readLock(lockPath)) === token) { + await fs.rm(lockPath, { force: true }); + } +} + +/** + * Remove the lock if it has outlived {@link LOCK_STALE_MS}, which means the + * process that wrote it died before releasing it. + * + * @param lockPath - Path of the lock file. + */ +async function removeStaleLock(lockPath: string): Promise { + try { + const stats = await fs.stat(lockPath); + if (stats.mtimeMs < Date.now() - LOCK_STALE_MS) { + await fs.rm(lockPath, { force: true }); + } + } catch (error) { + if (!isErrnoException(error, "ENOENT")) { + throw error; + } + } +} + +/** + * Resolve a persistence root to a stable identity that every process agrees + * on, so instances sharing a directory compute the same ownership scope. + * + * @param persistRoot - Directory to canonicalise; created if missing. + * @returns The real path, lowercased on Windows for case-insensitive matching. + */ +export async function canonicalisePersistRoot( + persistRoot: string +): Promise { + await fs.mkdir(persistRoot, { recursive: true }); + const canonical = await fs.realpath(persistRoot); + return process.platform === "win32" ? canonical.toLowerCase() : canonical; +} + +/** + * Run `callback` while holding an exclusive on-disk lock for `persistRoot`, so + * concurrent instances sharing the directory start up one at a time. + * + * @param persistRoot - Directory to lock; no lock is taken when `undefined`. + * @param callback - Work to run under the lock. + * @returns The callback's result. + * @throws If the lock can't be acquired within {@link LOCK_MAX_ATTEMPTS}. + */ +export async function withPersistRootStartupLock( + persistRoot: string | undefined, + callback: () => Promise +): Promise { + if (persistRoot === undefined) { + return callback(); + } + + await fs.mkdir(persistRoot, { recursive: true }); + const lockPath = path.join(persistRoot, LOCK_NAME); + const token = crypto.randomUUID(); + let acquired = false; + + for (let attempt = 0; !acquired; attempt++) { + try { + await fs.writeFile(lockPath, token, { flag: "wx", mode: 0o600 }); + acquired = true; + } catch (error) { + if (!isErrnoException(error, "EEXIST")) { + throw error; + } + if (attempt >= LOCK_MAX_ATTEMPTS) { + throw new Error( + `Timed out waiting for the Miniflare startup lock at ${lockPath}. ` + + `Another process may still be holding it -- if none is running, delete the file and retry.` + ); + } + await removeStaleLock(lockPath); + await new Promise((resolve) => setTimeout(resolve, LOCK_RETRY_MS)); + } + } + + try { + return await callback(); + } finally { + await removeOwnedLock(lockPath, token); + } +} diff --git a/packages/miniflare/src/workers/core/constants.ts b/packages/miniflare/src/workers/core/constants.ts index b19aec075e8..5772e35bbf4 100644 --- a/packages/miniflare/src/workers/core/constants.ts +++ b/packages/miniflare/src/workers/core/constants.ts @@ -91,6 +91,7 @@ export const CoreBindings = { SERVICE_DEV_REGISTRY_PROXY: "MINIFLARE_DEV_REGISTRY_PROXY", JSON_TELEMETRY_CONFIG: "MINIFLARE_TELEMETRY_CONFIG", DEV_REGISTRY_DEBUG_PORT: "DEV_REGISTRY_DEBUG_PORT", + DEV_REGISTRY_INSTANCE_ID: "DEV_REGISTRY_INSTANCE_ID", SERVICE_STREAM: "MINIFLARE_STREAM", SERVICE_IMAGES_DELIVERY: "MINIFLARE_IMAGES_DELIVERY", SERVICE_R2_PUBLIC: "MINIFLARE_R2_PUBLIC", diff --git a/packages/miniflare/src/workers/core/dev-registry-proxy-shared.worker.ts b/packages/miniflare/src/workers/core/dev-registry-proxy-shared.worker.ts index c937192d522..afb069830b0 100644 --- a/packages/miniflare/src/workers/core/dev-registry-proxy-shared.worker.ts +++ b/packages/miniflare/src/workers/core/dev-registry-proxy-shared.worker.ts @@ -9,6 +9,13 @@ import { DurableObject } from "cloudflare:workers"; */ export interface WorkerdDebugPortConnector { connect(address: string): WorkerdDebugPortClient; + /** + * Access this workerd process's own debug port directly, without opening a + * TCP connection back to ourselves. + * + * @see https://github.com/cloudflare/workerd/pull/7005 + */ + current(): WorkerdDebugPortClient; } /** @@ -34,6 +41,9 @@ export interface RegistryEntry { userWorkerService: string; /** Queue names consumed by this worker, if any. */ queueConsumers?: string[]; + created: number; + instanceId?: string; + storageScope?: string; } let registry = new Map(); @@ -43,7 +53,8 @@ let registry = new Map(); * Called whenever the Node.js side pushes an updated registry snapshot. */ export function setRegistry(data: Record): void { - registry = new Map(Object.entries(data)); + const entries = Object.entries(data); + registry = new Map(entries); } /** @@ -57,6 +68,25 @@ export function resolveTarget(service: string): RegistryEntry | undefined { return entry; } +/** + * Find the instance that owns shared storage for a persistence root. The + * oldest live candidate wins, with the entry name breaking ties so every + * process independently elects the same owner. + * + * @param storageScope - Canonical persistence root to find the owner for. + * @returns The owning instance's registry entry, or `undefined` if none is live. + */ +export function resolveSharedStorageOwner( + storageScope: string +): RegistryEntry | undefined { + return Array.from(registry.entries()) + .filter(([, entry]) => entry.storageScope === storageScope) + .sort( + ([previousName, previous], [nextName, next]) => + previous.created - next.created || previousName.localeCompare(nextName) + )[0]?.[1]; +} + /** * Find the registry entry of a worker that consumes the given queue, if any * dev session advertises one. Each queue has at most one consumer, so the @@ -96,23 +126,57 @@ export function workerNotFoundMessage(service: string): string { } /** - * Connect to a Durable Object actor on a remote workerd instance via the - * debug port, returning a {@link Fetcher} that proxies requests to it. + * Open a debug port client for a registry entry. + * + * When the entry describes the current instance -- most commonly because this + * instance won the shared storage election -- take workerd's in-process fast + * path instead of dialling our own debug port over TCP. + * + * @param debugPort - The `workerdDebugPort` binding. + * @param target - Registry entry describing the worker to reach. + * @param selfInstanceId - This instance's dev registry ID, when the caller knows it. + * @returns A debug port client for the target process. + */ +export function openDebugPortClient( + debugPort: WorkerdDebugPortConnector, + target: RegistryEntry, + selfInstanceId?: string +): WorkerdDebugPortClient { + if ( + selfInstanceId !== undefined && + target.instanceId !== undefined && + target.instanceId === selfInstanceId + ) { + return debugPort.current(); + } + return debugPort.connect(target.debugPortAddress); +} + +/** + * Connect to a Durable Object actor on a workerd instance via the debug port, + * returning a {@link Fetcher} that proxies requests to it. */ export function connectToActor( debugPort: WorkerdDebugPortConnector, scriptName: string, className: string, - actorId: string + actorId: string, + selfInstanceId?: string ): Fetcher | null { const target = resolveTarget(scriptName); if (!target || !target.debugPortAddress) { return null; } - const client = debugPort.connect(target.debugPortAddress); + const client = openDebugPortClient(debugPort, target, selfInstanceId); return client.getActor(target.userWorkerService, className, actorId); } +interface ProxyDurableObjectEnv { + DEV_REGISTRY_DEBUG_PORT: WorkerdDebugPortConnector; + /** Absent for callers that don't bind it, e.g. the local explorer worker. */ + DEV_REGISTRY_INSTANCE_ID?: string; +} + /** * Create a {@link DurableObject} subclass that proxies all method calls * and fetch requests to a Durable Object running in a separate workerd @@ -126,41 +190,43 @@ export function createProxyDurableObjectClass({ scriptName: string; className: string; }): typeof DurableObject { - return class extends DurableObject<{ - DEV_REGISTRY_DEBUG_PORT: WorkerdDebugPortConnector; - }> { + return class extends DurableObject { _cachedFetcher: Fetcher | undefined; _cachedDebugPortAddress: string | undefined; + _cachedInstanceId: string | undefined; - // Lazily resolve and cache. Invalidates when debugPortAddress changes. + // Lazily resolve and cache. Invalidates when the target's debugPortAddress + // or instanceId changes -- the latter matters because it decides whether we + // reach the target in-process or over TCP. _resolve(): Fetcher | null { const target = resolveTarget(scriptName); if ( this._cachedFetcher && - target?.debugPortAddress === this._cachedDebugPortAddress + target?.debugPortAddress === this._cachedDebugPortAddress && + target?.instanceId === this._cachedInstanceId ) { return this._cachedFetcher; } this._cachedFetcher = undefined; this._cachedDebugPortAddress = undefined; + this._cachedInstanceId = undefined; const fetcher = connectToActor( this.env.DEV_REGISTRY_DEBUG_PORT, scriptName, className, - this.ctx.id.toString() + this.ctx.id.toString(), + this.env.DEV_REGISTRY_INSTANCE_ID ); if (fetcher && target) { this._cachedFetcher = fetcher; this._cachedDebugPortAddress = target.debugPortAddress; + this._cachedInstanceId = target.instanceId; } return fetcher; } - constructor( - ctx: DurableObjectState, - env: { DEV_REGISTRY_DEBUG_PORT: WorkerdDebugPortConnector } - ) { + constructor(ctx: DurableObjectState, env: ProxyDurableObjectEnv) { super(ctx, env); return new Proxy(this, { diff --git a/packages/miniflare/src/workers/core/dev-registry-proxy.worker.ts b/packages/miniflare/src/workers/core/dev-registry-proxy.worker.ts index 64e592d5199..52de0519df9 100644 --- a/packages/miniflare/src/workers/core/dev-registry-proxy.worker.ts +++ b/packages/miniflare/src/workers/core/dev-registry-proxy.worker.ts @@ -3,12 +3,17 @@ import { getQueueServiceName, HEADER_QUEUE_NAME } from "../queues/constants"; import { CorePaths } from "./constants"; import { findQueueConsumer, + openDebugPortClient, + resolveSharedStorageOwner, resolveTarget, tailEventsReplacer, tailEventsReviver, workerNotFoundMessage, } from "./dev-registry-proxy-shared.worker"; -import type { WorkerdDebugPortConnector } from "./dev-registry-proxy-shared.worker"; +import type { + RegistryEntry, + WorkerdDebugPortConnector, +} from "./dev-registry-proxy-shared.worker"; export { createProxyDurableObjectClass, @@ -30,6 +35,7 @@ const HANDLER_RESERVED_KEYS = new Set([ interface Env { DEV_REGISTRY_DEBUG_PORT: WorkerdDebugPortConnector; + DEV_REGISTRY_INSTANCE_ID: string; } interface Props { @@ -39,19 +45,35 @@ interface Props { // Forwarded to the remote entrypoint via the debug port so they are // available as `ctx.props` on the callee. userProps?: Record; + // Is this trying to access a "storage" miniflare service? + // If it is, the proxy will try to forward to the shared storage owner + // (first active worker in the dev registry) + storage?: boolean; + storageScope?: string; } -function resolve(props: Props, env: Env): Fetcher | null { - const { service, entrypoint, userProps } = props; - const target = resolveTarget(service); - if (!target || !target.debugPortAddress) { - return null; +function getTarget(props: Props): RegistryEntry | undefined { + if (props.storage) { + return props.storageScope === undefined + ? undefined + : resolveSharedStorageOwner(props.storageScope); } - const serviceName = - entrypoint === null || entrypoint === "default" + return resolveTarget(props.service); +} + +function resolve(props: Props, env: Env, target: RegistryEntry): Fetcher { + const { service, entrypoint, userProps, storage } = props; + + const serviceName = storage + ? service + : entrypoint === null || entrypoint === "default" ? target.defaultEntrypointService : target.userWorkerService; - const client = env.DEV_REGISTRY_DEBUG_PORT.connect(target.debugPortAddress); + const client = openDebugPortClient( + env.DEV_REGISTRY_DEBUG_PORT, + target, + env.DEV_REGISTRY_INSTANCE_ID + ); return client.getEntrypoint(serviceName, entrypoint ?? undefined, userProps); } @@ -80,8 +102,10 @@ export class ExternalQueueProxy extends WorkerEntrypoint { ); } - const client = this.env.DEV_REGISTRY_DEBUG_PORT.connect( - target.debugPortAddress + const client = openDebugPortClient( + this.env.DEV_REGISTRY_DEBUG_PORT, + target, + this.env.DEV_REGISTRY_INSTANCE_ID ); const broker = client.getEntrypoint(getQueueServiceName(queueName)); const headers = new Headers(request.headers); @@ -91,19 +115,22 @@ export class ExternalQueueProxy extends WorkerEntrypoint { } export class ExternalServiceProxy extends WorkerEntrypoint { - _fetcher: Fetcher | null = null; + _fetcher: Fetcher | undefined; + _targetAddress: string | undefined; + _targetInstanceId: string | undefined; _entryFetcher: Fetcher | null = null; constructor(ctx: ExecutionContext, env: Env) { super(ctx, env); - this._fetcher = resolve(ctx.props, env); // Separate connection for scheduled: the debug port's EventDispatcher // doesn't support runScheduled/runAlarm/queue, so we forward via HTTP. const target = resolveTarget(ctx.props.service); if (target && target.debugPortAddress) { - const client = env.DEV_REGISTRY_DEBUG_PORT.connect( - target.debugPortAddress + const client = openDebugPortClient( + env.DEV_REGISTRY_DEBUG_PORT, + target, + env.DEV_REGISTRY_INSTANCE_ID ); this._entryFetcher = client.getEntrypoint("core:entry"); } @@ -117,21 +144,45 @@ export class ExternalServiceProxy extends WorkerEntrypoint { return undefined; } - if (!target._fetcher) { + const fetcher = target._resolve(); + if (!fetcher) { throw new Error(workerNotFoundMessage(ctx.props.service)); } - return Reflect.get(target._fetcher, prop); + return Reflect.get(fetcher, prop); }, }); } + _resolve(): Fetcher | null { + const target = getTarget(this.ctx.props); + if (target === undefined || !target.debugPortAddress) { + this._fetcher = undefined; + this._targetAddress = undefined; + this._targetInstanceId = undefined; + return null; + } + if ( + this._fetcher !== undefined && + this._targetAddress === target.debugPortAddress && + this._targetInstanceId === target.instanceId + ) { + return this._fetcher; + } + + this._fetcher = resolve(this.ctx.props, this.env, target); + this._targetAddress = target.debugPortAddress; + this._targetInstanceId = target.instanceId; + return this._fetcher; + } + fetch(request: Request): Promise | Response { - if (!this._fetcher) { + const fetcher = this._resolve(); + if (!fetcher) { return new Response(workerNotFoundMessage(this.ctx.props.service), { status: 503, }); } - return this._fetcher.fetch(request); + return fetcher.fetch(request); } async scheduled(controller: ScheduledController) { @@ -162,7 +213,8 @@ export class ExternalServiceProxy extends WorkerEntrypoint { // Events with rpcMethod==="tail" are filtered out to prevent infinite // recursion (the remote tail() call would itself produce a tail event). async tail(events: TraceItem[]) { - if (!this._fetcher) { + const fetcher = this._resolve(); + if (fetcher === null) { return; } const filtered = events.filter( @@ -181,7 +233,7 @@ export class ExternalServiceProxy extends WorkerEntrypoint { // outside this `try`, so it escapes as an unhandled rejection instead of // being reported. // @ts-expect-error .tail is not in the `Fetcher` type but it's a valid RPC call - await this._fetcher.tail(serializedEvents); + await fetcher.tail(serializedEvents); } catch (e) { console.warn( `[dev-registry] Failed to forward tail events to "${ diff --git a/packages/miniflare/src/workers/local-explorer/aggregation.ts b/packages/miniflare/src/workers/local-explorer/aggregation.ts index 3e5ca07106b..d5288cad0bf 100644 --- a/packages/miniflare/src/workers/local-explorer/aggregation.ts +++ b/packages/miniflare/src/workers/local-explorer/aggregation.ts @@ -24,11 +24,15 @@ export const NO_AGGREGATE_HEADER = "X-Miniflare-Explorer-No-Aggregate"; */ function getPeerDebugPortAddresses( registry: WorkerRegistry, - selfWorkerNames: string[] + selfWorkerNames: string[], + selfInstanceId: string | null ): string[] { const selfSet = new Set(selfWorkerNames); const addresses = Object.entries(registry) - .filter(([name]) => !selfSet.has(name)) + .filter( + ([name, definition]) => + !selfSet.has(name) && definition.instanceId !== selfInstanceId + ) .map(([, def]) => def.debugPortAddress) .filter((addr): addr is string => typeof addr === "string"); // A single Miniflare process with multiple workers registers multiple @@ -47,7 +51,11 @@ export async function getPeerUrlsIfAggregating( const workerNames = c.env.LOCAL_EXPLORER_WORKER_NAMES; const response = await loopback.fetch("http://localhost/core/dev-registry"); const registry = (await response.json()) as WorkerRegistry; - return getPeerDebugPortAddresses(registry, workerNames); + return getPeerDebugPortAddresses( + registry, + workerNames, + response.headers.get("X-Miniflare-Dev-Registry-Instance-Id") + ); } /** diff --git a/packages/miniflare/test/config/schema.spec.ts b/packages/miniflare/test/config/schema.spec.ts index aaaab528476..2a3eb70b22b 100644 --- a/packages/miniflare/test/config/schema.spec.ts +++ b/packages/miniflare/test/config/schema.spec.ts @@ -1,5 +1,6 @@ import { describe, test, vi } from "vitest"; import { + MiniflareOptionsSchema, MiniflareWorkerConfigSchema, WorkerOptionsSchema, } from "../../src/config/schema"; @@ -268,3 +269,49 @@ describe("MiniflareWorkerConfigSchema", () => { }); }); }); + +describe("MiniflareOptionsSchema", () => { + const worker = { + config: { + type: "worker" as const, + name: "worker", + compatibilityDate: "2025-01-01", + }, + }; + + test("resolves the isolated root to the resource root without shared storage", ({ + expect, + }) => { + const parsed = MiniflareOptionsSchema.parse({ + resourcePersistencePath: "/state", + workers: [worker], + }); + + // Nothing is shared, so every resource is isolated and belongs under the + // configured resource root. Readers must not have to work this out. + expect(parsed.isolatedResourcePersistencePath).toBe("/state"); + }); + + test("keeps an explicit isolated root when shared storage is enabled", ({ + expect, + }) => { + const parsed = MiniflareOptionsSchema.parse({ + unsafeEnableSharedStorage: true, + resourcePersistencePath: "/shared", + isolatedResourcePersistencePath: "/isolated", + unsafeDevRegistryPath: "/registry", + workers: [worker], + }); + + expect(parsed.resourcePersistencePath).toBe("/shared"); + expect(parsed.isolatedResourcePersistencePath).toBe("/isolated"); + }); + + test("leaves the isolated root unset when nothing is persisted", ({ + expect, + }) => { + const parsed = MiniflareOptionsSchema.parse({ workers: [worker] }); + + expect(parsed.isolatedResourcePersistencePath).toBeUndefined(); + }); +}); diff --git a/packages/miniflare/test/dev-registry.spec.ts b/packages/miniflare/test/dev-registry.spec.ts index bb42242882a..41cbf9ca97b 100644 --- a/packages/miniflare/test/dev-registry.spec.ts +++ b/packages/miniflare/test/dev-registry.spec.ts @@ -1,11 +1,92 @@ +import fs from "node:fs/promises"; import path from "node:path"; import { watch } from "chokidar"; import { getWorkerRegistry, Miniflare } from "miniflare"; import { describe, onTestFinished, test, vi } from "vitest"; -import { singleModuleManifest, useDispose, useTmp } from "./test-shared"; -import type { MiniflareOptions, WorkerRegistry } from "miniflare"; +import { DevRegistry } from "../src/shared/dev-registry"; +import { + singleModuleManifest, + TestLog, + useDispose, + useTmp, +} from "./test-shared"; +import type { + MiniflareOptions, + WorkerDefinition, + WorkerRegistry, +} from "miniflare"; describe.sequential("DevRegistry", () => { + test("surfaces fresh legacy entries and removes them when stale", async ({ + expect, + }) => { + const unsafeDevRegistryPath = await useTmp(); + const definitionPath = path.join(unsafeDevRegistryPath, "legacy-worker"); + await fs.writeFile( + definitionPath, + JSON.stringify({ + debugPortAddress: "127.0.0.1:1234", + defaultEntrypointService: "core:user:legacy-worker", + userWorkerService: "core:user:legacy-worker", + }) + ); + + const legacyDefinition = getWorkerRegistry(unsafeDevRegistryPath)[ + "legacy-worker" + ]; + expect(legacyDefinition).toEqual( + expect.objectContaining({ + debugPortAddress: "127.0.0.1:1234", + }) + ); + expect(legacyDefinition.instanceId).toBeUndefined(); + + const stale = new Date(Date.now() - 91_000); + await fs.utimes(definitionPath, stale, stale); + expect(getWorkerRegistry(unsafeDevRegistryPath)).toEqual({}); + await expect(fs.stat(definitionPath)).rejects.toMatchObject({ + code: "ENOENT", + }); + }); + + test("registers after a conflicting entry becomes stale", async ({ + expect, + }) => { + const unsafeDevRegistryPath = await useTmp(); + const definitionPath = path.join(unsafeDevRegistryPath, "worker"); + const definition: WorkerDefinition = { + debugPortAddress: "127.0.0.1:1234", + defaultEntrypointService: "core:user:worker", + userWorkerService: "core:user:worker", + }; + await fs.writeFile( + definitionPath, + JSON.stringify({ ...definition, instanceId: "previous-instance" }) + ); + + const registry = new DevRegistry( + unsafeDevRegistryPath, + undefined, + new TestLog() + ); + vi.useFakeTimers(); + try { + registry.register({ worker: definition }); + expect( + JSON.parse(await fs.readFile(definitionPath, "utf8")).instanceId + ).toBe("previous-instance"); + + await vi.advanceTimersByTimeAsync(90_001); + + expect( + JSON.parse(await fs.readFile(definitionPath, "utf8")).instanceId + ).toBe(registry.instanceId); + } finally { + await registry.dispose(); + vi.useRealTimers(); + } + }); + test("registers workers by default unless opted out", async ({ expect }) => { const unsafeDevRegistryPath = await useTmp(); const worker = { diff --git a/packages/miniflare/test/exit-hook.spec.ts b/packages/miniflare/test/exit-hook.spec.ts new file mode 100644 index 00000000000..fe85a30512b --- /dev/null +++ b/packages/miniflare/test/exit-hook.spec.ts @@ -0,0 +1,85 @@ +import { afterEach, test, vi } from "vitest"; +import { exitHook } from "../src/exit-hook"; + +const SIGNALS = [ + ["SIGINT", 128 + 2], + ["SIGTERM", 128 + 15], + ["SIGHUP", 128 + 1], +] as const; + +const unregisters: (() => void)[] = []; + +function register(callback: () => void = () => {}) { + const unregister = exitHook(callback); + unregisters.push(unregister); + return unregister; +} + +afterEach(() => { + while (unregisters.length > 0) { + unregisters.pop()?.(); + } +}); + +test.for(SIGNALS)( + "exitHook: runs callbacks and exits on %s", + ([signal, exitCode], { expect }) => { + const listenersBefore = new Set(process.listeners(signal)); + const callback = vi.fn(); + register(callback); + + const signalHandler = process + .listeners(signal) + .find((listener) => !listenersBefore.has(listener)); + if (signalHandler === undefined) { + throw new Error(`Expected exitHook() to register a ${signal} listener`); + } + + const exit = vi.spyOn(process, "exit").mockImplementation((code) => { + throw new Error(`process.exit(${String(code)})`); + }); + try { + expect(() => signalHandler(signal)).toThrow(`process.exit(${exitCode})`); + expect(callback).toHaveBeenCalledOnce(); + expect(exit).toHaveBeenCalledWith(exitCode); + } finally { + exit.mockRestore(); + } + } +); + +test("exitHook: registers a listener for every termination signal", ({ + expect, +}) => { + const before = Object.fromEntries( + SIGNALS.map(([signal]) => [signal, process.listenerCount(signal)]) + ); + + register(); + + for (const [signal] of SIGNALS) { + expect(process.listenerCount(signal)).toBe(before[signal] + 1); + } +}); + +test("exitHook: removes every signal listener once the last callback unregisters", ({ + expect, +}) => { + const before = Object.fromEntries( + SIGNALS.map(([signal]) => [signal, process.listenerCount(signal)]) + ); + + const unregisterFirst = register(); + const unregisterSecond = register(); + + unregisterFirst(); + // Listeners stay while another callback is still registered. + for (const [signal] of SIGNALS) { + expect(process.listenerCount(signal)).toBe(before[signal] + 1); + } + + unregisterSecond(); + for (const [signal] of SIGNALS) { + expect(process.listenerCount(signal)).toBe(before[signal]); + } +}); diff --git a/packages/miniflare/test/plugins/local-explorer/binding-map.spec.ts b/packages/miniflare/test/plugins/local-explorer/binding-map.spec.ts index e111f511e67..3c102d04362 100644 --- a/packages/miniflare/test/plugins/local-explorer/binding-map.spec.ts +++ b/packages/miniflare/test/plugins/local-explorer/binding-map.spec.ts @@ -1,7 +1,12 @@ import { Miniflare } from "miniflare"; import { afterAll, beforeAll, describe, test } from "vitest"; import { CorePaths } from "../../../src/workers/core/constants"; -import { disposeWithRetry, singleModuleManifest } from "../../test-shared"; +import { + disposeWithRetry, + singleModuleManifest, + useDispose, + useTmp, +} from "../../test-shared"; import type { RemoteProxyConnectionString } from "miniflare"; const BASE_URL = `http://localhost${CorePaths.EXPLORER}/api`; @@ -93,3 +98,56 @@ describe("Local Explorer remote binding skipping", () => { expect(uuids).toEqual(["d1-local"]); }); }); + +test("lists shared resources once without aggregating the current instance", async ({ + expect, +}) => { + const mf = new Miniflare({ + inspectorPort: 0, + unsafeLocalExplorer: true, + unsafeEnableSharedStorage: true, + resourcePersistencePath: await useTmp(), + isolatedResourcePersistencePath: await useTmp(), + unsafeDevRegistryPath: await useTmp(), + workers: [ + { + config: { + type: "worker", + name: "worker", + compatibilityDate: "2025-01-01", + compatibilityFlags: ["experimental"], + manifest: singleModuleManifest( + `export default { fetch() { return new Response("user worker"); } }` + ), + env: { + KV: { type: "kv", id: "kv-id" }, + R2: { type: "r2", name: "r2-name" }, + D1: { type: "d1", id: "d1-id" }, + }, + }, + }, + ], + }); + useDispose(mf); + await mf.ready; + + const kvResponse = await mf.dispatchFetch( + `${BASE_URL}/storage/kv/namespaces` + ); + const kvBody = (await kvResponse.json()) as { + result: Array<{ id: string }>; + }; + expect(kvBody.result.map(({ id }) => id)).toEqual(["kv-id"]); + + const r2Response = await mf.dispatchFetch(`${BASE_URL}/r2/buckets`); + const r2Body = (await r2Response.json()) as { + result: { buckets: Array<{ name: string }> }; + }; + expect(r2Body.result.buckets.map(({ name }) => name)).toEqual(["r2-name"]); + + const d1Response = await mf.dispatchFetch(`${BASE_URL}/d1/database`); + const d1Body = (await d1Response.json()) as { + result: Array<{ uuid: string }>; + }; + expect(d1Body.result.map(({ uuid }) => uuid)).toEqual(["d1-id"]); +}); diff --git a/packages/miniflare/test/shared/persist-root-lock.spec.ts b/packages/miniflare/test/shared/persist-root-lock.spec.ts new file mode 100644 index 00000000000..0d853a9f2aa --- /dev/null +++ b/packages/miniflare/test/shared/persist-root-lock.spec.ts @@ -0,0 +1,122 @@ +import childProcess from "node:child_process"; +import { once } from "node:events"; +import fs from "node:fs/promises"; +import path from "node:path"; +import { describe, test } from "vitest"; +import { + canonicalisePersistRoot, + withPersistRootStartupLock, +} from "../../src/shared/persist-root-lock"; +import { useTmp } from "../test-shared"; + +const LOCK_CHILD_SCRIPT = String.raw` + const fs = require("node:fs/promises"); + const path = require("node:path"); + const { withPersistRootStartupLock } = require(process.env.LOCK_MODULE); + + (async () => { + const root = process.env.LOCK_ROOT; + const id = process.env.LOCK_CHILD_ID; + await fs.writeFile(path.join(root, "ready-" + id), ""); + while (true) { + try { + await fs.access(path.join(root, "go")); + break; + } catch { + await new Promise((resolve) => setTimeout(resolve, 5)); + } + } + await withPersistRootStartupLock(root, async () => { + const eventsPath = path.join(root, "events"); + await fs.appendFile(eventsPath, "start:" + id + "\n"); + await new Promise((resolve) => setTimeout(resolve, 75)); + await fs.appendFile(eventsPath, "end:" + id + "\n"); + }); + })().catch((error) => { + console.error(error); + process.exitCode = 1; + }); +`; + +describe("persist root startup lock", () => { + test("serialises callbacks across processes and removes its lock", async ({ + expect, + }) => { + const root = await useTmp(); + const childCount = 5; + const children = Array.from({ length: childCount }, (_, i) => + childProcess.spawn( + process.execPath, + ["-r", "esbuild-register", "-e", LOCK_CHILD_SCRIPT], + { + stdio: ["ignore", "ignore", "inherit"], + env: { + ...process.env, + LOCK_MODULE: path.resolve( + __dirname, + "../../src/shared/persist-root-lock.ts" + ), + LOCK_ROOT: root, + LOCK_CHILD_ID: String(i), + }, + } + ) + ); + const exits = children.map((child) => once(child, "exit")); + + while ( + (await fs.readdir(root)).filter((entry) => entry.startsWith("ready-")) + .length < childCount + ) { + await new Promise((resolve) => setTimeout(resolve, 5)); + } + await fs.writeFile(path.join(root, "go"), ""); + + for (const [code] of await Promise.all(exits)) { + expect(code).toBe(0); + } + const events = (await fs.readFile(path.join(root, "events"), "utf8")) + .trim() + .split("\n"); + expect(events).toHaveLength(childCount * 2); + for (let i = 0; i < events.length; i += 2) { + expect(events[i]).toMatch(/^start:/); + expect(events[i + 1]).toBe(events[i].replace("start:", "end:")); + } + await expect( + fs.stat(path.join(root, ".miniflare-startup.lock")) + ).rejects.toMatchObject({ + code: "ENOENT", + }); + }); + + test("recovers stale locks", async ({ expect }) => { + const root = await useTmp(); + const lockPath = path.join(root, ".miniflare-startup.lock"); + const stale = new Date(Date.now() - 60_000); + await fs.writeFile(lockPath, ""); + await fs.utimes(lockPath, stale, stale); + + let called = false; + await withPersistRootStartupLock(root, async () => { + called = true; + }); + + expect(called).toBe(true); + await expect(fs.stat(lockPath)).rejects.toMatchObject({ code: "ENOENT" }); + }); + + test.runIf(process.platform !== "win32")( + "canonicalises symlink aliases", + async ({ expect }) => { + const parent = await useTmp(); + const root = path.join(parent, "root"); + const alias = path.join(parent, "alias"); + await fs.mkdir(root); + await fs.symlink(root, alias); + expect(await canonicalisePersistRoot(alias)).toBe( + await canonicalisePersistRoot(root) + ); + } + ); +}); diff --git a/packages/miniflare/test/storage-owner.spec.ts b/packages/miniflare/test/storage-owner.spec.ts new file mode 100644 index 00000000000..d2423e55f40 --- /dev/null +++ b/packages/miniflare/test/storage-owner.spec.ts @@ -0,0 +1,820 @@ +import { Miniflare } from "miniflare"; +import { describe, it, vi } from "vitest"; +import { singleModuleManifest, useTmp } from "./test-shared"; +import type { MiniflareOptions } from "miniflare"; + +async function withIsolatedStorage( + options: MiniflareOptions +): Promise { + return { + ...options, + isolatedResourcePersistencePath: await useTmp(), + }; +} + +describe.sequential("owner presence integration", () => { + it("requires persistence and a dev registry", ({ expect }) => { + const worker = { + config: { + type: "worker" as const, + name: "worker", + compatibilityDate: "2025-01-01", + }, + }; + expect( + () => + new Miniflare({ + unsafeEnableSharedStorage: true, + unsafeDevRegistryPath: ".registry", + workers: [worker], + }) + ).toThrow( + "Shared storage requires `resourcePersistencePath` to be set to the directory instances should share." + ); + expect( + () => + new Miniflare({ + unsafeEnableSharedStorage: true, + resourcePersistencePath: ".state", + workers: [worker], + }) + ).toThrow( + "Shared storage requires `unsafeDevRegistryPath` to be set, as instances elect a storage owner through the dev registry." + ); + expect( + () => + new Miniflare({ + unsafeEnableSharedStorage: true, + resourcePersistencePath: ".state", + unsafeDevRegistryPath: ".registry", + workers: [worker], + }) + ).toThrow( + "Shared storage requires `isolatedResourcePersistencePath` to be set to a per-project directory, for resources that cannot be shared." + ); + }); + + it("routes a client's KV through the owner so storage is shared", async ({ + expect, + }) => { + const persistRoot = await useTmp(); + const registryPath = await useTmp(); + const KV_WORKER = `export default { + async fetch(request, env) { + const url = new URL(request.url); + const key = url.searchParams.get("key") ?? "k"; + if (request.method === "PUT") { + await env.NS.put(key, await request.text()); + return new Response("ok"); + } + const val = await env.NS.get(key); + return new Response(val ?? ""); + } + }`; + const common: MiniflareOptions = { + unsafeEnableSharedStorage: true, + resourcePersistencePath: persistRoot, + unsafeDevRegistryPath: registryPath, + workers: [ + { + config: { + type: "worker", + name: "worker", + compatibilityDate: "2025-01-01", + compatibilityFlags: ["experimental"], + manifest: singleModuleManifest(KV_WORKER), + env: { NS: { type: "kv", id: "NS" } }, + }, + }, + ], + }; + + const owner = new Miniflare(await withIsolatedStorage(common)); + await owner.ready; + const client = new Miniflare(await withIsolatedStorage(common)); + + try { + await client.ready; + + // Write through the client (which routes to the owner). + const putRes = await client.dispatchFetch("http://x/?key=greeting", { + method: "PUT", + body: "hello-from-client", + }); + expect(await putRes.text()).toBe("ok"); + + // The owner can read what the client wrote → storage is shared. + const ownerRes = await owner.dispatchFetch("http://x/?key=greeting"); + expect(await ownerRes.text()).toBe("hello-from-client"); + + // And the client can read it back through the proxy. + const clientRes = await client.dispatchFetch("http://x/?key=greeting"); + expect(await clientRes.text()).toBe("hello-from-client"); + } finally { + await client.dispose(); + await owner.dispose(); + } + }); + + it("an explicit owner hosts plugins absent from its user config", async ({ + expect, + }) => { + const persistRoot = await useTmp(); + const registryPath = await useTmp(); + const owner = new Miniflare( + await withIsolatedStorage({ + unsafeEnableSharedStorage: true, + resourcePersistencePath: persistRoot, + unsafeDevRegistryPath: registryPath, + workers: [ + { + config: { + type: "worker", + name: "owner", + compatibilityDate: "2025-01-01", + manifest: singleModuleManifest( + "export default { fetch() { return new Response('owner'); } }" + ), + env: { NS: { type: "kv", id: "NS" } }, + }, + }, + ], + }) + ); + await owner.ready; + const client = new Miniflare( + await withIsolatedStorage({ + unsafeEnableSharedStorage: true, + resourcePersistencePath: persistRoot, + unsafeDevRegistryPath: registryPath, + workers: [ + { + config: { + type: "worker", + name: "client", + compatibilityDate: "2025-01-01", + manifest: singleModuleManifest(`export default { + async fetch(request, env) { + await env.DB.prepare("CREATE TABLE IF NOT EXISTS t(v TEXT)").run(); + return new Response("ok"); + } + }`), + env: { DB: { type: "d1", id: "DB" } }, + }, + }, + ], + }) + ); + + try { + await client.ready; + expect(await (await client.dispatchFetch("http://x/")).text()).toBe("ok"); + } finally { + await client.dispose(); + await owner.dispose(); + } + }); + + it("routes a client's R2 and D1 through the owner so storage is shared", async ({ + expect, + }) => { + const persistRoot = await useTmp(); + const registryPath = await useTmp(); + const WORKER = `export default { + async fetch(request, env) { + const url = new URL(request.url); + const kind = url.searchParams.get("kind"); + if (kind === "r2") { + if (request.method === "PUT") { + await env.BUCKET.put("obj", await request.text()); + return new Response("ok"); + } + const o = await env.BUCKET.get("obj"); + return new Response(o ? await o.text() : ""); + } + // d1 + if (request.method === "PUT") { + await env.DB.prepare("CREATE TABLE IF NOT EXISTS t(v TEXT)").run(); + await env.DB.prepare("INSERT INTO t(v) VALUES (?)").bind(await request.text()).run(); + return new Response("ok"); + } + const { results } = await env.DB.prepare("SELECT v FROM t").all(); + return new Response(JSON.stringify(results.map((r) => r.v))); + } + }`; + const common: MiniflareOptions = { + unsafeEnableSharedStorage: true, + resourcePersistencePath: persistRoot, + unsafeDevRegistryPath: registryPath, + workers: [ + { + config: { + type: "worker", + name: "worker", + compatibilityDate: "2025-01-01", + compatibilityFlags: ["experimental"], + manifest: singleModuleManifest(WORKER), + env: { + BUCKET: { type: "r2", name: "BUCKET" }, + DB: { type: "d1", id: "DB" }, + }, + }, + }, + ], + }; + const owner = new Miniflare(await withIsolatedStorage(common)); + await owner.ready; + const client = new Miniflare(await withIsolatedStorage(common)); + + try { + await client.ready; + + // R2: client write → owner read. + expect( + await ( + await client.dispatchFetch("http://x/?kind=r2", { + method: "PUT", + body: "r2-from-client", + }) + ).text() + ).toBe("ok"); + expect( + await (await owner.dispatchFetch("http://x/?kind=r2")).text() + ).toBe("r2-from-client"); + expect( + await ( + await client.dispatchFetch( + "http://x/cdn-cgi/local/r2/public/BUCKET/obj" + ) + ).text() + ).toBe("r2-from-client"); + + // D1: client write → owner read. + expect( + await ( + await client.dispatchFetch("http://x/?kind=d1", { + method: "PUT", + body: "d1-from-client", + }) + ).text() + ).toBe("ok"); + expect( + await (await owner.dispatchFetch("http://x/?kind=d1")).text() + ).toBe(JSON.stringify(["d1-from-client"])); + } finally { + await client.dispose(); + await owner.dispose(); + } + }); + + it.todo("routes a client's Stream through the owner over RPC so storage is shared", async ({ + expect, + }) => { + const persistRoot = await useTmp(); + const registryPath = await useTmp(); + // Exercises the JSRPC path of the owner boundary (native RPC over the debug + // port), including nested RpcTargets (`videos.list()`). + const WORKER = `export default { + async fetch(request, env) { + try { + if (request.method === "PUT") { + const body = new Response( + new Uint8Array([0, 1, 2, 3, 4, 5, 6, 7]) + ).body; + const video = await env.STREAM.upload(body, {}); + return Response.json({ id: video.id }); + } + const videos = await env.STREAM.videos.list(); + return Response.json({ count: videos.length }); + } catch (e) { + return Response.json({ error: String(e && e.stack || e) }, { status: 500 }); + } + } + }`; + const common: MiniflareOptions = { + unsafeEnableSharedStorage: true, + resourcePersistencePath: persistRoot, + unsafeDevRegistryPath: registryPath, + workers: [ + { + config: { + type: "worker", + name: "worker", + compatibilityDate: "2025-01-01", + compatibilityFlags: ["experimental"], + manifest: singleModuleManifest(WORKER), + env: { STREAM: { type: "stream" } }, + }, + }, + ], + }; + const owner = new Miniflare(await withIsolatedStorage(common)); + await owner.ready; + const client = new Miniflare(await withIsolatedStorage(common)); + + try { + await client.ready; + + // Client uploads a video (RPC through the owner)... + const put = (await ( + await client.dispatchFetch("http://x/", { method: "PUT" }) + ).json()) as { id: string }; + expect(put.id).toBeTruthy(); + + // ...and the owner sees it (shared store), proving the RPC round-trip + // and the shared backing storage. + expect( + ( + (await (await owner.dispatchFetch("http://x/")).json()) as { + count: number; + } + ).count + ).toBe(1); + } finally { + await client.dispose(); + await owner.dispose(); + } + }); + + it("routes a client's Secrets Store secret through the owner over RPC", async ({ + expect, + }) => { + const persistRoot = await useTmp(); + const registryPath = await useTmp(); + const WORKER = `export default { + async fetch(request, env) { + try { + return new Response(await env.SECRET.get()); + } catch (e) { + return new Response(e.message, { status: 404 }); + } + } + }`; + const common: MiniflareOptions = { + unsafeEnableSharedStorage: true, + resourcePersistencePath: persistRoot, + unsafeDevRegistryPath: registryPath, + workers: [ + { + config: { + type: "worker", + name: "worker", + compatibilityDate: "2025-01-01", + compatibilityFlags: ["experimental"], + manifest: singleModuleManifest(WORKER), + env: { + SECRET: { + type: "secrets-store-secret", + storeId: "store_a", + secretName: "api_key", + }, + }, + }, + }, + ], + }; + const owner = new Miniflare(await withIsolatedStorage(common)); + await owner.ready; + const client = new Miniflare(await withIsolatedStorage(common)); + + try { + await client.ready; + + // Seed the secret value on the owner (which holds the local store)... + await ( + await owner.getSecretsStoreSecretAPI("SECRET") + )().create("super-secret"); + + // ...and the client reads it back over the routed RPC binding. + expect(await (await client.dispatchFetch("http://x/")).text()).toBe( + "super-secret" + ); + } finally { + await client.dispose(); + await owner.dispose(); + } + }); + + it.todo("routes a client's Images store to the owner without dangling services", async ({ + expect, + }) => { + const persistRoot = await useTmp(); + const registryPath = await useTmp(); + const common: MiniflareOptions = { + unsafeEnableSharedStorage: true, + resourcePersistencePath: persistRoot, + unsafeDevRegistryPath: registryPath, + workers: [ + { + config: { + type: "worker", + name: "worker", + compatibilityDate: "2025-01-01", + compatibilityFlags: ["experimental"], + manifest: singleModuleManifest( + "export default { async fetch(_request, env) { return new Response(typeof env.IMAGES.info); } }" + ), + env: { IMAGES: { type: "images" } }, + }, + }, + ], + }; + const owner = new Miniflare(await withIsolatedStorage(common)); + const client = new Miniflare(await withIsolatedStorage(common)); + try { + // Both reaching `ready` proves the routed client doesn't reference a + // local images storage service it no longer stands up (the owner does), + // and the transform worker + its routed `IMAGES_STORE` binding resolve. + await owner.ready; + await client.ready; + expect(await (await client.dispatchFetch("http://x/")).text()).toBe( + "function" + ); + } finally { + await client.dispose(); + await owner.dispose(); + } + }); + + it("hosts plugins and resources not used by the client that spawned it", async ({ + expect, + }) => { + const persistRoot = await useTmp(); + const registryPath = await useTmp(); + + const first = new Miniflare( + await withIsolatedStorage({ + unsafeEnableSharedStorage: true, + resourcePersistencePath: persistRoot, + unsafeDevRegistryPath: registryPath, + workers: [ + { + config: { + type: "worker", + name: "first", + compatibilityDate: "2025-01-01", + manifest: singleModuleManifest( + "export default { fetch() { return new Response('first'); } }" + ), + env: { NS: { type: "kv", id: "NS" } }, + }, + }, + ], + }) + ); + let second: Miniflare | undefined; + + try { + await first.ready; + + second = new Miniflare( + await withIsolatedStorage({ + unsafeEnableSharedStorage: true, + resourcePersistencePath: persistRoot, + unsafeDevRegistryPath: registryPath, + workers: [ + { + config: { + type: "worker", + name: "second", + compatibilityDate: "2025-01-01", + compatibilityFlags: ["experimental"], + manifest: singleModuleManifest(`export default { + async fetch(request, env) { + if (new URL(request.url).pathname === "/secret") { + return new Response(await env.SECRET.get()); + } + if (request.method === "PUT") { + await env.DB.prepare("CREATE TABLE IF NOT EXISTS t(v TEXT)").run(); + await env.DB.prepare("INSERT INTO t(v) VALUES (?)").bind(await request.text()).run(); + return new Response("ok"); + } + const row = await env.DB.prepare("SELECT v FROM t").first(); + return new Response(row?.v ?? ""); + } + }`), + env: { + DB: { type: "d1", id: "later-db" }, + SECRET: { + type: "secrets-store-secret", + storeId: "later-store", + secretName: "later-secret", + }, + }, + }, + }, + ], + }) + ); + await second.ready; + + expect( + await ( + await second.dispatchFetch("http://x/", { + method: "PUT", + body: "from-second", + }) + ).text() + ).toBe("ok"); + expect(await (await second.dispatchFetch("http://x/")).text()).toBe( + "from-second" + ); + + await ( + await second.getSecretsStoreSecretAPI("SECRET") + )().create("secret-from-second"); + expect(await (await second.dispatchFetch("http://x/secret")).text()).toBe( + "secret-from-second" + ); + } finally { + await second?.dispose().catch(() => {}); + await first.dispose().catch(() => {}); + } + }); + + it("lets many client instances write one D1 concurrently without contention", async ({ + expect, + }) => { + // This is the scenario that produces cross-process SQLITE_BUSY today: many + // processes opening the same SQLite file. With a shared owner, only the + // owner opens it, so concurrent writes from all clients succeed. + const persistRoot = await useTmp(); + const registryPath = await useTmp(); + + const N = 10; // client instances + const M = 10; // inserts per client + const WORKER = `export default { + async fetch(request, env) { + const url = new URL(request.url); + if (url.searchParams.get("init") === "1") { + await env.DB.prepare("CREATE TABLE IF NOT EXISTS t(v INTEGER)").run(); + return new Response("ok"); + } + if (request.method === "PUT") { + await env.DB.prepare("INSERT INTO t(v) VALUES (1)").run(); + return new Response("ok"); + } + const row = await env.DB.prepare("SELECT COUNT(*) AS c FROM t").first(); + return new Response(String(row.c)); + } + }`; + const make = async () => + new Miniflare( + await withIsolatedStorage({ + unsafeEnableSharedStorage: true, + resourcePersistencePath: persistRoot, + unsafeDevRegistryPath: registryPath, + workers: [ + { + config: { + type: "worker", + name: "worker", + compatibilityDate: "2025-01-01", + compatibilityFlags: ["experimental"], + manifest: singleModuleManifest(WORKER), + env: { DB: { type: "d1", id: "DB" } }, + }, + }, + ], + }) + ); + + const clients = await Promise.all(Array.from({ length: N }, make)); + try { + await Promise.all(clients.map((c) => c.ready)); + + // Create the table once, then hammer it concurrently from every client. + await clients[0].dispatchFetch("http://x/?init=1").then((r) => r.text()); + + const results = await Promise.all( + clients.flatMap((c) => + Array.from({ length: M }, async () => { + const r = await c.dispatchFetch("http://x/", { method: "PUT" }); + await r.text(); // consume body + return r.status; + }) + ) + ); + // No request failed (e.g. with a 500 from SQLITE_BUSY). + expect(results.every((s) => s === 200)).toBe(true); + + // All writes landed — no lost updates, no contention failures. + const count = await clients[0] + .dispatchFetch("http://x/") + .then((r) => r.text()); + expect(count).toBe(String(N * M)); + } finally { + await Promise.all(clients.map((c) => c.dispose().catch(() => {}))); + } + }, 30_000); + + it("hands storage ownership to another live instance", async ({ expect }) => { + const persistRoot = await useTmp(); + const registryPath = await useTmp(); + const options: MiniflareOptions = { + unsafeEnableSharedStorage: true, + resourcePersistencePath: persistRoot, + unsafeDevRegistryPath: registryPath, + workers: [ + { + config: { + type: "worker", + name: "worker", + compatibilityDate: "2025-01-01", + compatibilityFlags: ["experimental"], + manifest: singleModuleManifest(`export default { + async fetch(request, env) { + await env.DB.prepare("CREATE TABLE IF NOT EXISTS t(v TEXT)").run(); + if (request.method === "PUT") { + await env.DB.prepare("INSERT INTO t(v) VALUES (?)").bind(await request.text()).run(); + } + const row = await env.DB.prepare("SELECT COUNT(*) AS c FROM t").first(); + return new Response(String(row.c)); + } + }`), + env: { DB: { type: "d1", id: "DB" } }, + }, + }, + ], + }; + const first = new Miniflare(await withIsolatedStorage(options)); + let second: Miniflare | undefined; + + try { + await first.ready; + second = new Miniflare(await withIsolatedStorage(options)); + await second.ready; + + expect( + await ( + await second.dispatchFetch("http://x/", { + method: "PUT", + body: "before", + }) + ).text() + ).toBe("1"); + + await first.dispose(); + await vi.waitFor( + async () => { + const response = await second?.dispatchFetch("http://x/"); + const body = await response?.text(); + expect([response?.status, body]).toEqual([200, "1"]); + }, + { timeout: 10_000, interval: 100 } + ); + expect( + await ( + await second.dispatchFetch("http://x/", { + method: "PUT", + body: "after", + }) + ).text() + ).toBe("2"); + } finally { + await second?.dispose().catch(() => {}); + await first.dispose().catch(() => {}); + } + }); + + it("elects independent owners for different persistence roots", async ({ + expect, + }) => { + const registryPath = await useTmp(); + const WORKER = `export default { + async fetch(request, env) { + await env.DB.prepare("CREATE TABLE IF NOT EXISTS t(v TEXT)").run(); + if (request.method === "PUT") { + await env.DB.prepare("INSERT INTO t(v) VALUES (?)").bind(await request.text()).run(); + } + const row = await env.DB.prepare("SELECT COUNT(*) AS c FROM t").first(); + return new Response(String(row.c)); + } + }`; + const make = async (name: string) => + new Miniflare( + await withIsolatedStorage({ + unsafeEnableSharedStorage: true, + resourcePersistencePath: await useTmp(), + unsafeDevRegistryPath: registryPath, + workers: [ + { + config: { + type: "worker", + name, + compatibilityDate: "2025-01-01", + compatibilityFlags: ["experimental"], + manifest: singleModuleManifest(WORKER), + env: { DB: { type: "d1", id: "DB" } }, + }, + }, + ], + }) + ); + const first = await make("first"); + const second = await make("second"); + + try { + await first.ready; + await second.ready; + for (const instance of [first, second]) { + expect( + await ( + await instance.dispatchFetch("http://x/", { + method: "PUT", + body: "value", + }) + ).text() + ).toBe("1"); + } + } finally { + await Promise.all([first.dispose(), second.dispose()]); + } + }); + + it("persists isolated resources across restarts", async ({ expect }) => { + const persistRoot = await useTmp(); + const isolatedRoot = await useTmp(); + const registryPath = await useTmp(); + const options: MiniflareOptions = { + unsafeEnableSharedStorage: true, + resourcePersistencePath: persistRoot, + isolatedResourcePersistencePath: isolatedRoot, + unsafeDevRegistryPath: registryPath, + workers: [ + { + config: { + type: "worker", + name: "worker", + compatibilityDate: "2025-01-01", + manifest: singleModuleManifest(`export default { + async fetch(request) { + const key = new Request("http://cache/key"); + if (request.method === "PUT") { + await caches.default.put( + key, + new Response(await request.text(), { + headers: { "Cache-Control": "max-age=3600" }, + }) + ); + return new Response("ok"); + } + return (await caches.default.match(key)) ?? new Response("missing"); + } + }`), + }, + }, + ], + }; + + const first = new Miniflare(options); + await first.ready; + expect( + await ( + await first.dispatchFetch("http://x/", { + method: "PUT", + body: "persisted", + }) + ).text() + ).toBe("ok"); + expect(await (await first.dispatchFetch("http://x/")).text()).toBe( + "persisted" + ); + await first.dispose(); + + const restarted = new Miniflare(options); + try { + await restarted.ready; + expect(await (await restarted.dispatchFetch("http://x/")).text()).toBe( + "persisted" + ); + } finally { + await restarted.dispose(); + } + }); + + it("does nothing when the feature flag is off", async () => { + const persistRoot = await useTmp(); + const registryPath = await useTmp(); + const mf = new Miniflare({ + resourcePersistencePath: persistRoot, + unsafeDevRegistryPath: registryPath, + workers: [ + { + config: { + type: "worker", + name: "worker", + compatibilityDate: "2025-05-01", + compatibilityFlags: ["experimental"], + manifest: singleModuleManifest( + "export default { async fetch() { return new Response('plain'); } }" + ), + }, + }, + ], + }); + await mf.ready; + + await mf.dispose(); + }); +}); diff --git a/packages/vitest-plugin/AGENTS.md b/packages/vitest-plugin/AGENTS.md index dce95024d5c..6adddce0368 100644 --- a/packages/vitest-plugin/AGENTS.md +++ b/packages/vitest-plugin/AGENTS.md @@ -11,12 +11,13 @@ - Main export (`exports["."]`), built as ESM (`.mjs`) - Orchestrates test execution -### 2. Config (`src/config/index.ts`) +### 2. Config (`src/pool/config.ts`) -- Runs in Node.js -- Exported as `exports["./config"]`, built as CJS (`.cjs`) -- Provides `defineWorkersConfig()` / `defineWorkersProject()` helpers -- Injects Vite plugin for `cloudflare:test` resolution, sets resolve conditions (`workerd`, `worker`, `browser`) +- Runs in Node.js, part of the pool bundle (no separate package export) +- Validates pool options (`WorkersPoolOptionsSchema`) and resolves the project's + Worker configuration into Miniflare options +- `src/pool/plugin.ts` provides the `cloudflareTest()` Vite plugin, which injects + `cloudflare:test` resolution and sets resolve conditions (`workerd`, `worker`, `browser`) ### 3. Worker (`src/worker/index.ts`) @@ -25,6 +26,23 @@ - Contains HACK: monkeypatches VitestExecutor to access singleton - Has direct cross-package source import into `miniflare/src/workers/core/devalue` +## WORKER CONFIGURATION SOURCES + +`parseCustomPoolOptions()` in `src/pool/config.ts` resolves at most one configuration +file into a normalised `Config`, then shares every downstream step (remote proxy +session, `unstable_getMiniflareWorkerOptions()`, `main`, defines, module rules, tails): + +- `wrangler: { configPath, environment }` — a Wrangler configuration file, via + `wrangler.unstable_readConfig()` +- `experimental: { newConfig }` — a `cloudflare.config.ts`, via `src/pool/new-config.ts` + (`@cloudflare/config`'s `loadAndValidateConfig()` → `convertToWranglerConfig()` → + `normalizeAndValidateConfig()`). Mirrors `@cloudflare/vite-plugin`'s + `experimental.newConfig`. `ctx.mode` comes from `project.vite.config.mode`. + +The two are mutually exclusive. Whichever is used, the resolved path, config format and +Worker name are recorded on `options.resolvedConfig` for the pool to consume — never +re-read the config file from disk. + ## BUILD `tsdown.config.ts` defines 2 separate builds (all ESM): @@ -56,5 +74,5 @@ Resolved by custom Vite plugin (`@cloudflare/vitest-plugin:config`) that re-expo - Hook timeout: 60s, retry: 2 - Global setup starts mock npm registry, installs local package to temp dir - Test helper: custom `test` fixture with `tmpPath`, `seed()`, `vitestRun()`, `vitestDev()` -- Fixtures in `fixtures/vitest-plugin-examples/` (20+ sub-fixtures testing KV, R2, D1, DO, Queues, etc.) +- Fixtures in `fixtures/vitest-plugin-examples/` (20+ sub-fixtures testing KV, R2, D1, DO, Queues, `cloudflare.config.ts`, etc.) - Skipped on Windows CI due to flakiness diff --git a/packages/vitest-plugin/package.json b/packages/vitest-plugin/package.json index 92fdb532d9b..c218f0cc8ff 100644 --- a/packages/vitest-plugin/package.json +++ b/packages/vitest-plugin/package.json @@ -57,6 +57,7 @@ "zod": "catalog:default" }, "devDependencies": { + "@cloudflare/config": "workspace:*", "@cloudflare/mock-npm-registry": "workspace:*", "@cloudflare/remote-bindings": "workspace:*", "@cloudflare/workers-tsconfig": "workspace:*", diff --git a/packages/vitest-plugin/src/pool/cloudflare-pool-worker.ts b/packages/vitest-plugin/src/pool/cloudflare-pool-worker.ts index bca895ccaf3..6da37b98b66 100644 --- a/packages/vitest-plugin/src/pool/cloudflare-pool-worker.ts +++ b/packages/vitest-plugin/src/pool/cloudflare-pool-worker.ts @@ -120,9 +120,9 @@ export class CloudflarePoolWorker implements PoolWorker { }); this.mf = undefined; - if (this.parsedPoolOptions?.wrangler?.configPath) { + if (this.parsedPoolOptions?.resolvedConfig) { const session = remoteProxySessionsDataMap.get( - this.parsedPoolOptions.wrangler.configPath + this.parsedPoolOptions.resolvedConfig.path )?.session; await session?.dispose?.()?.catch((err) => { this.debug("remote proxy session dispose rejected: %O", err); diff --git a/packages/vitest-plugin/src/pool/compatibility-flag-assertions.ts b/packages/vitest-plugin/src/pool/compatibility-flag-assertions.ts index 48756704c95..12fa5d79bab 100644 --- a/packages/vitest-plugin/src/pool/compatibility-flag-assertions.ts +++ b/packages/vitest-plugin/src/pool/compatibility-flag-assertions.ts @@ -8,14 +8,16 @@ export class CompatibilityFlagAssertions { #compatibilityFlags: string[]; #optionsPath: string; #relativeProjectPath: string; - #relativeWranglerConfigPath?: string; + #relativeConfigPath?: string; + #camelCaseConfigFields: boolean; constructor(options: CommonOptions) { this.#compatibilityDate = options.compatibilityDate; this.#compatibilityFlags = options.compatibilityFlags; this.#optionsPath = options.optionsPath; this.#relativeProjectPath = options.relativeProjectPath; - this.#relativeWranglerConfigPath = options.relativeWranglerConfigPath; + this.#relativeConfigPath = options.relativeConfigPath; + this.#camelCaseConfigFields = options.camelCaseConfigFields ?? false; } /** @@ -36,8 +38,8 @@ export class CompatibilityFlagAssertions { */ #buildErrorMessageBase(): string { let message = `In project ${this.#relativeProjectPath}`; - if (this.#relativeWranglerConfigPath) { - message += `'s configuration file ${this.#relativeWranglerConfigPath}`; + if (this.#relativeConfigPath) { + message += `'s configuration file ${this.#relativeConfigPath}`; } return message; } @@ -46,14 +48,16 @@ export class CompatibilityFlagAssertions { * Constructs the configuration path part of the error message. */ #buildConfigPath(setting: string): string { - if (this.#relativeWranglerConfigPath) { - return `\`${setting}\``; - } - const camelCaseSetting = setting.replace(/_(\w)/g, (_, letter) => letter.toUpperCase() ); + if (this.#relativeConfigPath) { + // `cloudflare.config.ts` names its fields in camelCase, Wrangler + // configuration files use snake_case + return `\`${this.#camelCaseConfigFields ? camelCaseSetting : setting}\``; + } + return `\`${this.#optionsPath}.${camelCaseSetting}\``; } @@ -140,7 +144,12 @@ interface CommonOptions { compatibilityFlags: string[]; optionsPath: string; relativeProjectPath: string; - relativeWranglerConfigPath?: string; + relativeConfigPath?: string; + /** + * Whether the configuration file names its fields in camelCase, as + * `cloudflare.config.ts` does. Defaults to `false` (Wrangler's snake_case). + */ + camelCaseConfigFields?: boolean; } /** diff --git a/packages/vitest-plugin/src/pool/config.ts b/packages/vitest-plugin/src/pool/config.ts index 4ef2c7b6a9a..b078ad48a52 100644 --- a/packages/vitest-plugin/src/pool/config.ts +++ b/packages/vitest-plugin/src/pool/config.ts @@ -16,10 +16,12 @@ import { getRelativeProjectConfigPath, getRelativeProjectPath, } from "./helpers"; +import { loadNewConfig, NEW_CONFIG_FILENAME } from "./new-config"; import type { RemoteBindingsLogger, RemoteProxySessionData, } from "@cloudflare/remote-bindings"; +import type { Config } from "@cloudflare/workers-utils"; import type { LegacyWorkerOptions, V4ModuleRule } from "miniflare"; import type { TestProject } from "vitest/node"; import type { ZodError } from "zod"; @@ -28,6 +30,14 @@ export interface WorkersConfigPluginAPI { setMain(newMain?: string): void; } +const ExperimentalNewConfigSchema = z.object({ + /** + * Path to the `cloudflare.config.ts` file, resolved relative to the project + * root. Defaults to `cloudflare.config.ts` in the project root. + */ + configPath: z.string().optional(), +}); + const WorkersPoolOptionsSchema = z.object({ /** * Entrypoint to Worker run in the same isolate/context as tests. This is @@ -77,6 +87,22 @@ const WorkersPoolOptionsSchema = z.object({ environment: z.string().optional(), }) .optional(), + experimental: z + .object({ + /** + * Load the Worker's configuration from a `cloudflare.config.ts` file + * instead of a Wrangler configuration file. Cannot be combined with + * `wrangler`. + * + * Pass `true` to load `cloudflare.config.ts` from the project root, or + * an object to customise the behaviour. + * + * Config functions are called with `ctx.mode` set to Vite's mode, which + * defaults to `"test"` and can be overridden with `--mode`. + */ + newConfig: z.union([z.boolean(), ExperimentalNewConfigSchema]).optional(), + }) + .optional(), }); type CompatibleWorkerOptions = LegacyWorkerOptions & { @@ -102,6 +128,26 @@ export type WorkersPoolOptions = z.input & { export type WorkersPoolOptionsWithDefines = WorkersPoolOptions & { defines?: Record; moduleRules?: V4ModuleRule[]; + /** + * Details of the configuration file these options were resolved from. Set + * while parsing; not a user-facing option. Undefined when the project + * configures the Worker entirely through `miniflare` options. + */ + resolvedConfig?: { + /** Absolute path of the configuration file. */ + path: string; + /** + * Whether that file is a `cloudflare.config.ts` rather than a Wrangler + * configuration file. Determines how config fields are named in errors. + */ + newConfig: boolean; + /** + * The Worker name declared in the configuration file, before any + * environment name is appended. Self-referential service, tail and + * Workflow bindings are written against this name. + */ + workerName: string | undefined; + }; }; function normalizeMiniflareWorkerOptions(value: Record): void { @@ -243,9 +289,28 @@ export const remoteProxySessionsDataMap = new Map< RemoteProxySessionData | null >(); +/** + * Normalise the `experimental.newConfig` option into its resolved form. + * + * @param option The user-provided option value. + * @returns The resolved options, or `undefined` when new config is disabled. + */ +function normalizeNewConfigOption( + option: boolean | { configPath?: string } | undefined +): { configPath: string } | undefined { + if (option === undefined || option === false) { + return undefined; + } + if (option === true) { + return { configPath: NEW_CONFIG_FILENAME }; + } + return { configPath: option.configPath ?? NEW_CONFIG_FILENAME }; +} + async function parseCustomPoolOptions( rootPath: string, - value: unknown + value: unknown, + mode: string | undefined ): Promise { // Try to parse pool specific options const options = WorkersPoolOptionsSchema.parse( @@ -299,38 +364,64 @@ async function parseCustomPoolOptions( options.moduleRules = miniflareModuleRules; delete options.miniflare.modulesRules; - // Try to parse Wrangler config if any - if (options.wrangler?.configPath !== undefined) { - const configPath = path.resolve(rootPath, options.wrangler.configPath); + // Try to parse the project's configuration file, whichever format it uses + const newConfig = normalizeNewConfigOption(options.experimental?.newConfig); + + if (newConfig !== undefined && options.wrangler !== undefined) { + throw new TypeError( + "`wrangler` cannot be used together with `experimental.newConfig`. Configure the Worker via `cloudflare.config.ts` instead." + ); + } + + let configPath: string | undefined; + let config: Config | undefined; + // Wrangler environments have no `cloudflare.config.ts` equivalent yet + let environment: string | undefined; + + if (newConfig !== undefined) { + configPath = path.resolve(rootPath, newConfig.configPath); + config = await loadNewConfig(configPath, mode); + } else if (options.wrangler?.configPath !== undefined) { + configPath = path.resolve(rootPath, options.wrangler.configPath); // Make sure future accesses to `configPath` see a fully-resolved path // (e.g. for getting accurate relative paths in error messages) options.wrangler.configPath = configPath; + environment = options.wrangler.environment; - // Lazily import `wrangler` if and when we need it + // Lazily import `wrangler` if and when we need it. Parse the config once so + // we can pass the parsed config straight into + // `unstable_getMiniflareWorkerOptions` without re-parsing it. const wrangler = await import("wrangler"); - - // Parse the wrangler config once so we can pass the parsed config - // straight into `unstable_getMiniflareWorkerOptions` without - // re-parsing it. - const wranglerConfig = wrangler.unstable_readConfig({ + config = wrangler.unstable_readConfig({ config: configPath, - env: options.wrangler.environment, + env: environment, }); + } - const preExistingRemoteProxySessionData = options.wrangler?.configPath - ? remoteProxySessionsDataMap.get(options.wrangler.configPath) - : undefined; + if (configPath !== undefined && config !== undefined) { + options.resolvedConfig = { + path: configPath, + newConfig: newConfig !== undefined, + workerName: config.topLevelName, + }; + + // Already imported above for a Wrangler config; the module registry makes + // this a no-op when it was, and keeps it lazy when it wasn't + const wrangler = await import("wrangler"); + + const preExistingRemoteProxySessionData = + remoteProxySessionsDataMap.get(configPath); const remoteProxySessionData = options.remoteBindings ? await maybeStartOrUpdateRemoteProxySession( { - name: wranglerConfig.name ?? "worker", + name: config.name ?? "worker", bindings: wrangler.unstable_convertConfigBindingsToStartWorkerBindings( - wranglerConfig + config ) ?? {}, - complianceRegion: getCloudflareComplianceRegion(wranglerConfig), - account_id: wranglerConfig.account_id, + complianceRegion: getCloudflareComplianceRegion(config), + account_id: config.account_id, profileDir: path.dirname(configPath), }, preExistingRemoteProxySessionData ?? null, @@ -339,29 +430,22 @@ async function parseCustomPoolOptions( ) : null; - if (options.wrangler?.configPath && remoteProxySessionData) { - remoteProxySessionsDataMap.set( - options.wrangler.configPath, - remoteProxySessionData - ); + if (remoteProxySessionData) { + remoteProxySessionsDataMap.set(configPath, remoteProxySessionData); } const { workerOptions, externalWorkers, define, main } = - wrangler.unstable_getMiniflareWorkerOptions( - wranglerConfig, - options.wrangler.environment, - { - overrides: { - assets: options.miniflare.assets, - // doesn't work with containers yet so let's just disable it - enableContainers: false, - }, - remoteProxyConnectionString: - remoteProxySessionData?.session?.remoteProxyConnectionString, - } - ); - - // If `main` wasn't explicitly configured, fall back to Wrangler config's + wrangler.unstable_getMiniflareWorkerOptions(config, environment, { + overrides: { + assets: options.miniflare.assets, + // doesn't work with containers yet so let's just disable it + enableContainers: false, + }, + remoteProxyConnectionString: + remoteProxySessionData?.session?.remoteProxyConnectionString, + }); + + // If `main` wasn't explicitly configured, fall back to the config's entrypoint options.main ??= main; options.miniflare.workers = [ @@ -386,7 +470,7 @@ async function parseCustomPoolOptions( ) as SourcelessWorkerOptions; options.moduleRules = mergedModuleRules.modulesRules; - // Merge generated Miniflare options from Wrangler with specified overrides + // Merge generated Miniflare options from the config with specified overrides options.miniflare = mergeWorkerOptions( workerOptionsWithoutModuleRules, options.miniflare as SourcelessWorkerOptions @@ -400,7 +484,7 @@ async function parseCustomPoolOptions( ), }; - // Record any Wrangler `define`s + // Record any `define`s from the config options.defines = define; } @@ -444,7 +528,13 @@ export async function parseProjectOptions( const projectPath = getProjectPath(project); try { - return await parseCustomPoolOptions(projectPath, poolOptions); + return await parseCustomPoolOptions( + projectPath, + poolOptions, + // Vitest is Vite, so `cloudflare.config.ts` functions see the same mode + // Vite would give them. Defaults to `"test"`, overridable with `--mode`. + project.vite.config.mode + ); } catch (e) { if (!isZodErrorLike(e)) { throw e; diff --git a/packages/vitest-plugin/src/pool/index.ts b/packages/vitest-plugin/src/pool/index.ts index 5e7aae17c01..6a19924b76e 100644 --- a/packages/vitest-plugin/src/pool/index.ts +++ b/packages/vitest-plugin/src/pool/index.ts @@ -23,7 +23,6 @@ import { structuredSerializableRevivers, } from "miniflare"; import semverSatisfies from "semver/functions/satisfies.js"; -import { experimental_readRawConfig } from "wrangler"; import { CompatibilityFlagAssertions } from "./compatibility-flag-assertions"; import { guessWorkerExports } from "./guess-exports"; import { @@ -228,24 +227,12 @@ function getDurableObjectClasses(worker: SourcelessWorkerOptions): Set { return result; } -function getWranglerWorkerName( - relativeWranglerConfigPath?: string -): string | undefined { - if (!relativeWranglerConfigPath) { - return undefined; - } - const wranglerConfigObject = experimental_readRawConfig({ - config: relativeWranglerConfigPath, - }); - return wranglerConfigObject.rawConfig.name; -} - /** * Gets a set of class names for Workflows defined in the SELF Worker. */ function getWorkflowClasses( worker: SourcelessWorkerOptions, - relativeWranglerConfigPath: string | undefined + configWorkerName: string | undefined ): Set { // TODO(someday): may need to extend this to take into account other workers // if doing multi-worker tests across workspace projects @@ -260,11 +247,8 @@ function getWorkflowClasses( let workerName: string | undefined; // If the designator's scriptName matches its own Worker name, // use that as the worker name, otherwise use the vitest worker's name - const wranglerWorkerName = getWranglerWorkerName( - relativeWranglerConfigPath - ); - if (wranglerWorkerName && designator.scriptName === wranglerWorkerName) { - workerName = wranglerWorkerName; + if (configWorkerName && designator.scriptName === configWorkerName) { + workerName = configWorkerName; } else { workerName = worker.name; } @@ -290,18 +274,18 @@ const RUNNER_OBJECT_BINDING = "__VITEST_POOL_WORKERS_RUNNER_OBJECT"; function rewriteStreamingTailSelfReferences( worker: LegacyWorkerOptions, - wranglerWorkerName: string, + configWorkerName: string, runnerWorkerName: string ) { worker.streamingTails = worker.streamingTails?.map((tail) => { - if (tail === wranglerWorkerName) { + if (tail === configWorkerName) { return runnerWorkerName; } if ( typeof tail === "object" && tail !== null && "name" in tail && - tail.name === wranglerWorkerName + tail.name === configWorkerName ) { return { ...tail, name: runnerWorkerName }; } @@ -314,26 +298,26 @@ async function buildProjectWorkerOptions( customOptions: WorkersPoolOptionsWithDefines, main: string | undefined ): Promise { - const relativeWranglerConfigPath = maybeApply( + const relativeConfigPath = maybeApply( (v) => path.relative("", v), - customOptions.wrangler?.configPath + customOptions.resolvedConfig?.path ); const runnerWorker = customOptions.miniflare ?? {}; // `unstable_getMiniflareWorkerOptions` returns service bindings whose `name` // is the literal `config.name` for self-references (e.g. `{ name: "my-worker" }` - // when the wrangler config has `name: "my-worker"`). We rename the runner - // worker below, so rewrite those self-references to `kCurrentWorker` first. - // That symbol resolves at request time relative to the referer worker, so it - // survives the rename. - const wranglerWorkerName = getWranglerWorkerName(relativeWranglerConfigPath); - if (wranglerWorkerName && runnerWorker.serviceBindings) { + // when the config has `name: "my-worker"`). We rename the runner worker below, + // so rewrite those self-references to `kCurrentWorker` first. That symbol + // resolves at request time relative to the referer worker, so it survives the + // rename. + const configWorkerName = customOptions.resolvedConfig?.workerName; + if (configWorkerName && runnerWorker.serviceBindings) { for (const [key, sb] of Object.entries(runnerWorker.serviceBindings)) { if ( typeof sb === "object" && sb !== null && "name" in sb && - sb.name === wranglerWorkerName + sb.name === configWorkerName ) { runnerWorker.serviceBindings[key] = { ...sb, name: kCurrentWorker }; } @@ -374,7 +358,8 @@ async function buildProjectWorkerOptions( compatibilityFlags: runnerWorker.compatibilityFlags, optionsPath: `miniflare`, relativeProjectPath: getRelativeProjectPath(project), - relativeWranglerConfigPath, + relativeConfigPath, + camelCaseConfigFields: customOptions.resolvedConfig?.newConfig ?? false, }); const assertions = [ @@ -441,10 +426,7 @@ async function buildProjectWorkerOptions( runnerWorker.durableObjects ??= {}; const durableObjectClassNames = getDurableObjectClasses(runnerWorker); - const workflowClassNames = getWorkflowClasses( - runnerWorker, - relativeWranglerConfigPath - ); + const workflowClassNames = getWorkflowClasses(runnerWorker, configWorkerName); const selfWorkerExports: string[] = []; if ( @@ -616,10 +598,10 @@ async function buildProjectWorkerOptions( // Miniflare will validate these options const workerOptions = worker as LegacyWorkerOptions; - if (wranglerWorkerName) { + if (configWorkerName) { rewriteStreamingTailSelfReferences( workerOptions, - wranglerWorkerName, + configWorkerName, runnerWorker.name ); } diff --git a/packages/vitest-plugin/src/pool/new-config.ts b/packages/vitest-plugin/src/pool/new-config.ts new file mode 100644 index 00000000000..1bd882c5f92 --- /dev/null +++ b/packages/vitest-plugin/src/pool/new-config.ts @@ -0,0 +1,81 @@ +import { existsSync } from "node:fs"; +import { + convertToWranglerConfig, + loadAndValidateConfig, +} from "@cloudflare/config"; +import { normalizeAndValidateConfig } from "@cloudflare/workers-utils"; +import type { Config, RawConfig } from "@cloudflare/workers-utils"; + +export const NEW_CONFIG_FILENAME = "cloudflare.config.ts"; + +/** + * Load a `cloudflare.config.ts` and normalise it into the same `Config` shape + * that `wrangler.unstable_readConfig()` produces for a Wrangler configuration + * file, so that everything downstream (bindings, remote proxy sessions, + * `unstable_getMiniflareWorkerOptions()`) is oblivious to which config format + * the project uses. + * + * This mirrors the `@cloudflare/vite-plugin` implementation of + * `experimental.newConfig`: load and validate via `@cloudflare/config`, convert + * the result to a Wrangler `RawConfig`, then run it through the standard + * Wrangler normalisation/validation pipeline. + * + * @param configPath Absolute path to the `cloudflare.config.ts` file. + * @param mode The Vite mode, passed to config functions as `ctx.mode`. + * @returns The normalised config. + */ +export async function loadNewConfig( + configPath: string, + mode: string | undefined +): Promise { + if (!existsSync(configPath)) { + throw new TypeError( + `\`experimental.newConfig\` is enabled but no \`${NEW_CONFIG_FILENAME}\` was found at ${configPath}.` + ); + } + + const { result } = await loadAndValidateConfig(configPath, { mode }); + + if (!result.success) { + throw new TypeError( + `Invalid \`${NEW_CONFIG_FILENAME}\`:\n${result.error.message}` + ); + } + + const worker = + result.data.default?.type === "worker" ? result.data.default : undefined; + + if (worker === undefined) { + throw new TypeError( + `\`${NEW_CONFIG_FILENAME}\` must have a default worker export.` + ); + } + + const settings = + result.data.settings?.type === "settings" + ? result.data.settings + : undefined; + + const rawConfig: RawConfig = convertToWranglerConfig(worker, settings); + + // Passing `configPath` as both the config path and the user config path + // resolves `main` relative to the config file's directory, and lets + // `unstable_getMiniflareWorkerOptions()` derive the Worker's `rootPath` from + // that same directory — matching how a Wrangler configuration file behaves. + const { config, diagnostics } = normalizeAndValidateConfig( + rawConfig, + configPath, + configPath, + {} + ); + + if (diagnostics.hasWarnings()) { + console.warn(diagnostics.renderWarnings()); + } + + if (diagnostics.hasErrors()) { + throw new TypeError(diagnostics.renderErrors()); + } + + return config; +} diff --git a/packages/vitest-plugin/test/compatibility-flag-assertions.test.ts b/packages/vitest-plugin/test/compatibility-flag-assertions.test.ts index f744d9ee384..4e46e729c81 100644 --- a/packages/vitest-plugin/test/compatibility-flag-assertions.test.ts +++ b/packages/vitest-plugin/test/compatibility-flag-assertions.test.ts @@ -23,12 +23,14 @@ describe("FlagAssertions", () => { ); }); - it("includes relativeWranglerConfigPath in error message when provided", ({ + it("names fields in camelCase when the config is a cloudflare.config.ts", ({ expect, }) => { const options = { ...baseOptions, compatibilityFlags: ["disable-flag"], + relativeConfigPath: "cloudflare.config.ts", + camelCaseConfigFields: true, }; const flagAssertions = new CompatibilityFlagAssertions(options); const result = flagAssertions.assertIsEnabled({ @@ -37,7 +39,7 @@ describe("FlagAssertions", () => { }); expect(result.isValid).toBe(false); expect(result.errorMessage).toBe( - 'In project /path/to/project, `options.compatibilityFlags` must not contain "disable-flag".\nThis flag is incompatible with `@cloudflare/vitest-plugin`.' + 'In project /path/to/project\'s configuration file cloudflare.config.ts, `compatibilityFlags` must not contain "disable-flag".\nThis flag is incompatible with `@cloudflare/vitest-plugin`.' ); }); @@ -47,7 +49,7 @@ describe("FlagAssertions", () => { const options = { ...baseOptions, compatibilityFlags: ["disable-flag"], - relativeWranglerConfigPath: "wrangler.toml", + relativeConfigPath: "wrangler.toml", }; const flagAssertions = new CompatibilityFlagAssertions(options); const result = flagAssertions.assertIsEnabled({ @@ -210,14 +212,14 @@ describe("FlagAssertions", () => { ); }); - it("includes relativeWranglerConfigPath in error message when provided", ({ + it("includes relativeConfigPath in error message when provided", ({ expect, }) => { const options = { ...baseOptions, compatibilityDate: "2020-01-01", compatibilityFlags: [], - relativeWranglerConfigPath: "wrangler.toml", + relativeConfigPath: "wrangler.toml", }; const flagAssertions = new CompatibilityFlagAssertions(options); const result = flagAssertions.assertAtLeastOneFlagExists([ diff --git a/packages/vitest-plugin/test/new-config.test.ts b/packages/vitest-plugin/test/new-config.test.ts new file mode 100644 index 00000000000..ec4a5c634db --- /dev/null +++ b/packages/vitest-plugin/test/new-config.test.ts @@ -0,0 +1,232 @@ +import path from "node:path"; +import dedent from "ts-dedent"; +import { describe } from "vitest"; +import { test, vitestConfig } from "./helpers"; + +const worker = dedent` + export default { + async fetch(request, env, ctx) { + return new Response(env.MY_TEXT); + } + } +`; + +const workerTest = dedent` + import { env, SELF } from "cloudflare:test"; + import { it } from "vitest"; + + it("provides bindings from cloudflare.config.ts", ({ expect }) => { + expect(env.MY_TEXT).toBe("from the new config"); + }); + + it("dispatches to the entrypoint declared in cloudflare.config.ts", async ({ + expect, + }) => { + const response = await SELF.fetch("http://example.com"); + expect(await response.text()).toBe("from the new config"); + }); +`; + +test("loads cloudflare.config.ts from the project root", async ({ + expect, + seed, + vitestRun, +}) => { + await seed({ + "vitest.config.mts": vitestConfig({ + experimental: { newConfig: true }, + }), + "cloudflare.config.ts": dedent` + export default { + type: "worker", + name: "test-worker", + compatibilityDate: "2025-12-02", + entrypoint: "./index.ts", + env: { + MY_TEXT: { type: "text", value: "from the new config" }, + }, + }; + `, + "index.ts": worker, + "index.test.ts": workerTest, + }); + + const result = await vitestRun(); + + await expect(result.exitCode).resolves.toBe(0); +}); + +test("resolves a custom configPath and its entrypoint", async ({ + expect, + seed, + vitestRun, +}) => { + await seed({ + "vitest.config.mts": vitestConfig({ + experimental: { + newConfig: { configPath: "./config/cloudflare.config.ts" }, + }, + }), + // `entrypoint` is resolved relative to the config file, not the project root + "config/cloudflare.config.ts": dedent` + export default { + type: "worker", + name: "test-worker", + compatibilityDate: "2025-12-02", + entrypoint: "../index.ts", + env: { + MY_TEXT: { type: "text", value: "from the new config" }, + }, + }; + `, + "index.ts": worker, + "index.test.ts": workerTest, + }); + + const result = await vitestRun(); + + await expect(result.exitCode).resolves.toBe(0); +}); + +test("evaluates config functions with the Vite mode", async ({ + expect, + seed, + vitestRun, +}) => { + await seed({ + "vitest.config.mts": vitestConfig({ + experimental: { newConfig: true }, + }), + "cloudflare.config.ts": dedent` + export default (ctx) => ({ + type: "worker", + name: "test-worker", + compatibilityDate: "2025-12-02", + entrypoint: "./index.ts", + env: { + MY_TEXT: { type: "text", value: ctx.mode }, + }, + }); + `, + "index.ts": worker, + "index.test.ts": dedent` + import { env } from "cloudflare:test"; + import { it } from "vitest"; + + it("defaults the mode to \\"test\\"", ({ expect }) => { + expect(env.MY_TEXT).toBe("test"); + }); + `, + }); + + const result = await vitestRun(); + + await expect(result.exitCode).resolves.toBe(0); + + // ...and `--mode` overrides it, as it would in any other Vite project + await seed({ + "index.test.ts": dedent` + import { env } from "cloudflare:test"; + import { it } from "vitest"; + + it("uses the mode passed to --mode", ({ expect }) => { + expect(env.MY_TEXT).toBe("staging"); + }); + `, + }); + + const overridden = await vitestRun({ flags: ["--mode=staging"] }); + + await expect(overridden.exitCode).resolves.toBe(0); +}); + +describe("validation", () => { + test("rejects `wrangler` combined with `experimental.newConfig`", async ({ + expect, + seed, + vitestRun, + }) => { + await seed({ + "vitest.config.mts": vitestConfig({ + wrangler: { configPath: "./wrangler.jsonc" }, + experimental: { newConfig: true }, + }), + "index.test.ts": "", + }); + + const result = await vitestRun(); + + expect(await result.exitCode).toBe(1); + expect(result.stderr).toMatch( + "`wrangler` cannot be used together with `experimental.newConfig`. Configure the Worker via `cloudflare.config.ts` instead." + ); + }); + + test("reports a missing cloudflare.config.ts", async ({ + expect, + seed, + vitestRun, + tmpPath, + }) => { + await seed({ + "vitest.config.mts": vitestConfig({ + experimental: { newConfig: true }, + }), + "index.test.ts": "", + }); + + const result = await vitestRun(); + + expect(await result.exitCode).toBe(1); + expect(result.stderr).toMatch( + `\`experimental.newConfig\` is enabled but no \`cloudflare.config.ts\` was found at ${path.join(tmpPath, "cloudflare.config.ts")}` + ); + }); + + test("reports a config with no default worker export", async ({ + expect, + seed, + vitestRun, + }) => { + await seed({ + "vitest.config.mts": vitestConfig({ + experimental: { newConfig: true }, + }), + "cloudflare.config.ts": dedent` + export const settings = { type: "settings", accountId: "abc123" }; + `, + "index.test.ts": "", + }); + + const result = await vitestRun(); + + expect(await result.exitCode).toBe(1); + expect(result.stderr).toMatch( + "`cloudflare.config.ts` must have a default worker export." + ); + }); + + test("reports an invalid config", async ({ expect, seed, vitestRun }) => { + await seed({ + "vitest.config.mts": vitestConfig({ + experimental: { newConfig: true }, + }), + // `compatibilityDate` is required + "cloudflare.config.ts": dedent` + export default { + type: "worker", + name: "test-worker", + entrypoint: "./index.ts", + }; + `, + "index.ts": worker, + "index.test.ts": "", + }); + + const result = await vitestRun(); + + expect(await result.exitCode).toBe(1); + expect(result.stderr).toMatch("Invalid `cloudflare.config.ts`"); + expect(result.stderr).toMatch("compatibilityDate"); + }); +}); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 957f941b14b..e90fdcca376 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -4118,6 +4118,9 @@ importers: specifier: catalog:default version: 4.4.3 devDependencies: + '@cloudflare/config': + specifier: workspace:* + version: link:../config '@cloudflare/mock-npm-registry': specifier: workspace:* version: link:../mock-npm-registry