Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 13 additions & 1 deletion docs/guide/queues.md
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ Use queues to process work asynchronously, such as sending emails or handling up

Create a consumer in `queues/`. TypeScript and JavaScript files are supported (`.ts`, `.mts`, `.js`, `.mjs`). Its path determines the queue name: `queues/emails.ts` creates `emails`, and `queues/order/notifications.ts` creates `order-notifications`. Each queue must have a unique name.

The resulting name must follow Cloudflare's queue naming rules: 1–63 characters, only letters, digits, and `-`, beginning and ending with a letter or digit.
The resulting name must follow Cloudflare's queue naming rules: 1–63 characters, only letters, digits, and `-`, beginning and ending with a letter or digit. On `void deploy --platform cloudflare`, the Cloudflare queue also carries the Worker name, so the limit applies to that full name; see [Cloudflare queue names](#cloudflare-queue-names).

Export a default handler wrapped with [`defineQueue<T>`](../reference/api/handlers.md#definequeue-t-handler). `T` defines the message body for both the consumer and calls to `send()`:

Expand Down Expand Up @@ -116,6 +116,18 @@ deletes messages that exhaust their retries without one; see

Void provisions queues and connects their producers and consumers when you deploy.

### Cloudflare queue names

Queue names are unique per Cloudflare account. So `void deploy --platform cloudflare` names each queue after the Worker: in a Worker named `app-staging`, `queues/jobs.ts` uses the Cloudflare queue `app-staging-jobs` (a `_` in the Worker name becomes `-`). Two Workers in one account, such as `app-staging` and `app-production`, get separate queues. Your code keeps using the name from `queues/` (`queues.jobs.send()`), and `vp dev` uses that name too.

The deploy stops before creating anything when:

- a new queue's full name is longer than 63 characters. Shorten the Worker name or the file name.
- one queue's full name is the name of another file in `queues/`, such as `queues/jobs.ts` and `queues/app-jobs.ts` in Worker `app`. Rename one of the files.
- another Worker already consumes the Cloudflare queue the deploy would use. A Cloudflare queue has one consumer.

A queue that the Worker already consumes keeps its name, so an app deployed before Worker-named queues keeps its queue (for example `jobs`) and the messages in it. The deploy prints a note when it keeps a name. To move to the Worker-named queue, remove the Worker's consumer from the old queue in the Cloudflare dashboard, then deploy again. Messages that reach the old queue before that deploy finishes stay there, so do this when the app is not sending messages, and check that the old queue is empty before you delete it.

## Local development

In native Void apps, sending a message through a queue binding automatically invokes the consumer in Miniflare. Local delivery uses the same handler as production.
Expand Down
15 changes: 14 additions & 1 deletion packages/deploy-cloudflare/test/operation-fixture.ts
Original file line number Diff line number Diff line change
Expand Up @@ -99,7 +99,20 @@ export function operationFixtureResult(result: CloudflareCommandResult): Cloudfl
}
const queueName = result.stdout.match(/^Queue Name:\s*(\S+)/m)?.[1];
if (queueName) {
data = { name: queueName };
// `Consumers: a, b` lists the consumer Workers; `-` is a non-Worker consumer.
const consumers =
result.stdout
.split('\n')
.find((line) => line.startsWith('Consumers:'))
?.slice('Consumers:'.length) ?? '';
data = {
name: queueName,
consumers: consumers
.split(',')
.map((name) => name.trim())
.filter(Boolean)
.map((name) => (name === '-' ? null : name)),
};
}
if (result.stdout.includes('No migrations to apply!')) {
data = { pending: [] };
Expand Down
1 change: 1 addition & 0 deletions packages/void/AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,7 @@ generated entries must preserve the module-boundary rules below.

- **Repeatable framework wrappers**: `generateWrapper` recognizes current and legacy wrapper entries and preserves the original application across `--skip-build` runs. Compile handlers before replacing the entry, stage the wrapper for an atomic rename, and copy a new original only after an actual new build. A missing or recursively wrapped original requires a rebuild; readiness alone cannot validate the application entry.

- **Cloudflare queue names**: Only `void deploy --platform cloudflare` renames queues. Preflight (`cli/deploy-cloudflare-queues.ts`) reads the account and picks `<worker>-<name>` (the Worker part in resource-name form, as for R2 and KV, untrimmed), or keeps the plain name that this Worker already consumes (Cloudflare's consumer list is the record: `void.lock.json` is shared by every Worker that deploys from the checkout). It stores the result in `desired.queueNames`, and the build gets it through `VOID_CLOUDFLARE_QUEUE_NAMES`. Bindings, the route table and the framework wrapper read the same map, and the resolver then matches names exactly, because a `-name` suffix can pick the wrong queue. Dev, previews, plain builds and the hosted platform keep plain names and the suffix fallback.
- **Framework queue delivery**: Native queue handlers receive the original `MessageBatch`; only managed HTTP delivery serializes it. In deployed Workers the generated `/__void/queue` endpoint requires `__VOID_PROXY_TOKEN` (dev accepts only `x-void-dev-trigger`); it reconstructs `Date` timestamps and first-decision-wins ack/retry methods, and returns per-message decisions even if the consumer throws. Queue-only wrappers must also serve readiness.

- **Shared cron expressions**: Each scheduled invocation runs every matching job once and awaits all results before surfacing failures. Native trigger configuration and managed scheduler delivery deduplicate expressions; keep this coordinated with worker dispatch to avoid multiplying jobs on managed delivery.
Expand Down
43 changes: 42 additions & 1 deletion packages/void/src/cli/cloudflare-operation-process.ts
Original file line number Diff line number Diff line change
Expand Up @@ -618,6 +618,42 @@ function listAllResourcePages(command: Array<string>, extraFlags: Array<string>
throw new Error(`Cloudflare ${command.join(' ')} exceeded the supported page limit.`);
}

/**
* Each consumer's Worker script name, or null for one that is not a Worker (HTTP pull).
* Undefined when the list is incomplete or unreadable; only queue naming needs it.
*/
function queueConsumerWorkers(queue: Json): Array<string | null> | undefined {
const consumers = queue.consumers;
if (
!Array.isArray(consumers) ||
(queue.consumers_total_count !== undefined && queue.consumers_total_count !== consumers.length)
) {
return undefined;
}
const workers: Array<string | null> = [];
for (const entry of consumers) {
const consumer = object(entry);
const { script_name: scriptName, script } = consumer;
// The same identity rules as consumer removal: any doubt leaves the list unread.
if (
(scriptName !== undefined && (typeof scriptName !== 'string' || !scriptName)) ||
(script !== undefined && (typeof script !== 'string' || !script)) ||
(scriptName !== undefined && script !== undefined && scriptName !== script)
) {
return undefined;
}
const name = scriptName ?? script;
if (typeof name === 'string' && (consumer.type === undefined || consumer.type === 'worker')) {
workers.push(name);
} else if (name === undefined && consumer.type === 'http_pull') {
workers.push(null);
} else {
return undefined;
}
}
return workers;
}

async function readAllQueues(): Promise<Array<Json>> {
// cf queues list has no page option, so read the API's paginated response directly.
const queues: Array<Json> = [];
Expand Down Expand Up @@ -813,8 +849,13 @@ async function resources(): Promise<Result> {
return runCf(['queues', 'create', '--queue-name', operation.name]);
case 'queues.info': {
const queue = (await readAllQueues()).find((item) => item.queue_name === operation.name);
const consumers = queue && queueConsumerWorkers(queue);
return queue
? result(0, '', '', { id: queue.queue_id, name: queue.queue_name })
? result(0, '', '', {
id: queue.queue_id,
name: queue.queue_name,
...(consumers ? { consumers } : {}),
})
: { ...result(1, '', 'Cloudflare Queue does not exist.'), failure: 'not-found' };
}
default:
Expand Down
7 changes: 6 additions & 1 deletion packages/void/src/cli/deploy-cloudflare-build.ts
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ import { type VoidConfig } from '../config.ts';
import { collectMigrations } from '../migrations/collect.ts';
import { SqliteMigrationError } from '../migrations/sqlite-validation.ts';
import { detectFramework } from '../plugin-inference.ts';
import { CLOUDFLARE_DEPLOY_BUILD_ENV } from '../shared/env.ts';
import { CLOUDFLARE_DEPLOY_BUILD_ENV, CLOUDFLARE_QUEUE_NAMES_BUILD_ENV } from '../shared/env.ts';
import { CLOUDFLARE_PRESERVE_BINDINGS_BUILD_ENV } from '../shared/version-metadata.ts';
import { prepareCfBuildOutput, prepareCfFrameworkBuildOutput } from './cf-build-output.ts';
import {
Expand Down Expand Up @@ -81,6 +81,10 @@ export async function buildCloudflareWorker(input: {
// but local files cannot be serialized into the artifact this command uploads.
deleteEnvVarCaseAware(buildChildEnv, CLOUDFLARE_DEPLOY_BUILD_ENV);
buildChildEnv[CLOUDFLARE_DEPLOY_BUILD_ENV] = '1';
deleteEnvVarCaseAware(buildChildEnv, CLOUDFLARE_QUEUE_NAMES_BUILD_ENV);
if (desired.queueNames) {
buildChildEnv[CLOUDFLARE_QUEUE_NAMES_BUILD_ENV] = JSON.stringify(desired.queueNames);
}
deleteEnvVarCaseAware(buildChildEnv, CLOUDFLARE_PRESERVE_BINDINGS_BUILD_ENV);
if (existingWorker) {
buildChildEnv[CLOUDFLARE_PRESERVE_BINDINGS_BUILD_ENV] = '1';
Expand Down Expand Up @@ -119,6 +123,7 @@ export async function buildCloudflareWorker(input: {
requiredSecretNames: schemaFacts.requiredKeys,
ai: needsAI ? { binding: aiBindingName } : undefined,
preserveBindings: existingWorker !== null,
cloudflareQueueNames: desired.queueNames,
});
} catch (error) {
log.error(
Expand Down
75 changes: 31 additions & 44 deletions packages/void/src/cli/deploy-cloudflare-preflight.ts
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ import { analyzeApplicationSync, type ApplicationAnalysis } from '../application
import { MATERIALIZED_CLOUDFLARE_FILE } from '../cloudflare-config.ts';
import { resolveBindingNames, type VoidConfig } from '../config.ts';
import { detectFramework } from '../plugin-inference.ts';
import { cloudflareQueueName } from '../queues/cloudflare-names.ts';
import { assertSandboxNamespaceConfig, resolveSandboxConfig } from '../sandbox/config.ts';
import { classifyCloudflareDeployApp } from './cloudflare-deploy/classify.ts';
import { approvedCloudflareIsrBinding, chooseCloudflareIsr } from './cloudflare-isr.ts';
Expand All @@ -37,6 +38,7 @@ import {
connectCloudflareDeployAccount,
discoverRootWranglerConfig,
} from './deploy-cloudflare-helpers.ts';
import { chooseCloudflareQueueNames } from './deploy-cloudflare-queues.ts';
import { log } from './deploy-helpers.ts';
import {
resolveWranglerMigrationPattern,
Expand All @@ -61,9 +63,7 @@ import { selfHostSchemaFailureMessage, type SelfHostSchemaFacts } from './env-va
import { type CliLogger } from './log-file.ts';
import { getErrorMessage } from './output.ts';
import {
assertValidResourceName,
checkCloudflareProvisionDrift,
isQueueConfirmedAbsent,
provisionCloudflareResources,
type ProvisionDriftEntry,
} from './provision.ts';
Expand Down Expand Up @@ -580,51 +580,38 @@ export async function preflightCloudflareWorker(input: {
}
};

// Queues have no id in the config, so the drift check cannot see a missing one. Naming reads
// each queue by name; only a confirmed not-found triggers provisioning.
let queueNames: Awaited<ReturnType<typeof chooseCloudflareQueueNames>>;
try {
queueNames = await chooseCloudflareQueueNames({
desired,
workerName,
configPath,
runWrangler,
info: (message) => log.info(message),
});
} catch (error) {
log.error(getErrorMessage(error));
return exit(1);
}
if (!queueNames.ok) {
log.error(queueNames.message);
return exit(1);
}
const absentQueues = queueNames.absent;

const { missing: initiallyMissing } = checkCloudflareProvisionDrift(configPath, desired);
// A handoff applies no migrations, so preserve the existing local migration settings.
// Reconciling migrations_dir must not send an already-bound database to the provisioner.
const d1NeedsConfigReconciliation = !existingWorker && d1MigrationsDirMissing();
if (options.provision || initiallyMissing.length > 0 || d1NeedsConfigReconciliation) {
if (
options.provision ||
initiallyMissing.length > 0 ||
d1NeedsConfigReconciliation ||
absentQueues.length > 0
) {
await provisionAndVerify();
} else {
// Queue names come from unsanitized file paths. Report every invalid one before probing.
const invalidQueues = desired.queues.filter((name) => {
try {
assertValidResourceName('queue', name);
return false;
} catch {
return true;
}
});
if (invalidQueues.length > 0) {
log.error(
[
'Deploy blocked — these queue names are not valid Cloudflare queue names:',
...invalidQueues.map((name) => ` • ${name} (queue)`),
'',
'A queue takes its name from its path in queues/; nested segments are joined with "-".',
'A queue name must be 1–63 characters, contain only letters, digits and "-", and',
'begin and end with a letter or digit.',
'',
'Rename the offending file(s) in queues/ and retry.',
].join('\n'),
);
return exit(1);
}
// Queues have no id in the config, so the drift check cannot see a missing one. Probe each
// by name; only a confirmed not-found triggers provisioning, never an inconclusive probe.
const absentQueues: Array<string> = [];
for (const name of desired.queues) {
const info = await Promise.resolve(
runWrangler({ type: 'queues.info', name, configPath }, { capture: true }),
);
if (isQueueConfirmedAbsent(info)) {
absentQueues.push(name);
}
}
if (absentQueues.length > 0) {
await provisionAndVerify();
}
}

// Persist the Worker name and account before the build. The CF Vite plugin copies
Expand All @@ -640,11 +627,11 @@ export async function preflightCloudflareWorker(input: {
? {
queues: {
producers: frameworkQueues.map((queue) => ({
queue: queue.name,
queue: cloudflareQueueName(queue.name, desired.queueNames),
binding: queue.bindingName,
})),
consumers: frameworkQueues.map((queue) => ({
queue: queue.name,
queue: cloudflareQueueName(queue.name, desired.queueNames),
...(queue.maxBatchSize != null && { max_batch_size: queue.maxBatchSize }),
...(queue.maxBatchTimeout != null && {
max_batch_timeout: queue.maxBatchTimeout,
Expand Down
Loading
Loading