diff --git a/packages/deploy-core/src/artifact.ts b/packages/deploy-core/src/artifact.ts index df7f43137..ceff2c138 100644 --- a/packages/deploy-core/src/artifact.ts +++ b/packages/deploy-core/src/artifact.ts @@ -28,11 +28,43 @@ export type DeploySourceMetadata = { gitRef: string | null; }; -/** JSON payload accepted by the managed deployment API. */ +/** One Worker module uploaded to R2 before the deployment is finalized. */ +export type DeployWorkerModule = { + /** Module path relative to the Worker output, the same key `workerFiles` uses. */ + name: string; + /** Byte length of the module. */ + size: number; + /** SHA-256 of the module bytes as 64 lowercase hex characters. */ + sha256: string; +}; + +/** Request body of `POST /projects/:id/deploy/worker-upload-urls`. */ +export type DeployWorkerUploadRequest = { + modules: Array; +}; + +/** Upload URLs for each requested module, in request order. */ +export type DeployWorkerUploadResponse = { + id: string; + urls: Array<{ url: string; headers: Record }>; +}; + +/** Worker modules already uploaded through `worker-upload-urls`, named by its upload `id`. */ +export type DeployWorkerInput = { + id: string; + modules: Array; +}; + +/** + * JSON payload accepted by the managed deployment API. Worker modules travel + * either inline as `workerFiles` or uploaded beforehand as `workerInput`, + * never both. + */ export type DeployUploadPayload = { manifest: DeployManifest; assetManifest: Record; workerFiles?: Record; + workerInput?: DeployWorkerInput; } & DeploySourceMetadata & { preflight: 'true' | null; }; @@ -61,15 +93,31 @@ export function createDeployArtifact(input: { }; } -/** Convert the canonical artifact into the stable hosted-platform wire shape. */ +/** + * Convert the canonical artifact into the stable hosted-platform wire shape. + * Pass `workerInput` when the Worker modules were uploaded beforehand; the + * artifact must then carry no inline `workerFiles`. + */ export function createDeployUploadPayload( artifact: DeployArtifact, metadata: DeploySourceMetadata & { preflight: 'true' | null }, + workerInput?: DeployWorkerInput, ): DeployUploadPayload { + if (workerInput && artifact.workerFiles) { + throw new Error('A deployment sends its Worker modules inline or as uploads, not both.'); + } return { manifest: { ...artifact.manifest, bindings: { ...artifact.manifest.bindings } }, assetManifest: { ...artifact.assetManifest }, ...(artifact.workerFiles ? { workerFiles: { ...artifact.workerFiles } } : {}), + ...(workerInput + ? { + workerInput: { + id: workerInput.id, + modules: workerInput.modules.map((module) => ({ ...module })), + }, + } + : {}), ...metadata, }; } diff --git a/packages/deploy-core/src/index.ts b/packages/deploy-core/src/index.ts index 75704c1e3..16de95a65 100644 --- a/packages/deploy-core/src/index.ts +++ b/packages/deploy-core/src/index.ts @@ -338,6 +338,10 @@ export type { DeployAssetUploadRequest, DeploySourceMetadata, DeployUploadPayload, + DeployWorkerInput, + DeployWorkerModule, + DeployWorkerUploadRequest, + DeployWorkerUploadResponse, } from './artifact.ts'; export { DEPLOY_ARTIFACT_VERSION, diff --git a/packages/deploy-core/test/unit/artifact.test.ts b/packages/deploy-core/test/unit/artifact.test.ts index 271dd1d9c..c4a531741 100644 --- a/packages/deploy-core/test/unit/artifact.test.ts +++ b/packages/deploy-core/test/unit/artifact.test.ts @@ -57,4 +57,38 @@ describe('deployment artifacts', () => { }); expect(payload).not.toHaveProperty('artifactVersion'); }); + + it('names uploaded Worker modules by workerInput instead of carrying their bytes', () => { + const metadata = { + source: 'cli', + commit: null, + repositoryUrl: null, + commitUrl: null, + gitRef: null, + preflight: null, + }; + const workerInput = { + id: 'upload-1', + modules: [{ name: 'index.js', size: 18, sha256: 'a'.repeat(64) }], + }; + const payload = createDeployUploadPayload( + createDeployArtifact({ manifest, assetManifest: {} }), + metadata, + workerInput, + ); + + expect(payload.workerInput).toEqual(workerInput); + expect(payload).not.toHaveProperty('workerFiles'); + expect(() => + createDeployUploadPayload( + createDeployArtifact({ + manifest, + assetManifest: {}, + workerFiles: { 'index.js': 'ZXhwb3J0IGRlZmF1bHQge30=' }, + }), + metadata, + workerInput, + ), + ).toThrow('not both'); + }); }); diff --git a/packages/platform/src/discovery.ts b/packages/platform/src/discovery.ts index 2b9219b7e..7eb34b319 100644 --- a/packages/platform/src/discovery.ts +++ b/packages/platform/src/discovery.ts @@ -29,6 +29,11 @@ export type PlatformDiscoveryDocument = Readonly<{ capabilities: PlatformCapabilities; displayName: string; dashboardUrl?: string; + /** + * The API accepts Worker modules uploaded to R2 through + * `POST /projects/:id/deploy/worker-upload-urls`. Older platforms omit it. + */ + directWorkerUpload?: boolean; installationId: string | null; platformVersion: string; proxyUrl: string; @@ -54,11 +59,24 @@ function normalizeOrigin(value: string, name: 'apiUrl' | 'proxyUrl'): string { return url.origin; } +type PlatformDiscoveryInput = Omit< + PlatformDiscoveryDocument, + 'directWorkerUpload' | 'schemaVersion' | 'usageProtocolVersion' +> & { + usageProtocolVersion?: 0 | 1 | 2; +}; + +/** + * Builds the document a platform serves at `/.well-known/void-platform`. + * Every platform built from this source accepts direct Worker uploads. + */ export function createPlatformDiscoveryDocument( - input: Omit & { - usageProtocolVersion?: 0 | 1 | 2; - }, + input: PlatformDiscoveryInput, ): PlatformDiscoveryDocument { + return { ...normalizeDiscoveryDocument(input), directWorkerUpload: true }; +} + +function normalizeDiscoveryDocument(input: PlatformDiscoveryInput): PlatformDiscoveryDocument { if (!input.displayName.trim()) { throw new Error('displayName must not be empty.'); } @@ -195,7 +213,11 @@ export function parsePlatformDiscoveryDocument(value: unknown): PlatformDiscover }; const apiUrl = string('apiUrl'); const proxyUrl = string('proxyUrl'); - return createPlatformDiscoveryDocument({ + const directWorkerUpload = document.directWorkerUpload; + if (directWorkerUpload !== undefined && typeof directWorkerUpload !== 'boolean') { + throw new Error('directWorkerUpload must be a boolean.'); + } + const parsed = normalizeDiscoveryDocument({ apiUrl, applicationDomain: string('applicationDomain'), ...(document.applicationRouting !== undefined @@ -213,6 +235,8 @@ export function parsePlatformDiscoveryDocument(value: unknown): PlatformDiscover proxyUrl, usageProtocolVersion: parseUsageProtocolVersion(document.usageProtocolVersion), }); + // Report only what the platform said: an older platform omits the field. + return directWorkerUpload === undefined ? parsed : { ...parsed, directWorkerUpload }; } function parseAuthentication(value: unknown): PlatformAuthenticationDiscovery { diff --git a/packages/platform/test/unit/auth-discovery.test.ts b/packages/platform/test/unit/auth-discovery.test.ts index caa8dbece..f5cd5d862 100644 --- a/packages/platform/test/unit/auth-discovery.test.ts +++ b/packages/platform/test/unit/auth-discovery.test.ts @@ -110,3 +110,24 @@ describe('authentication discovery compatibility', () => { expect(JSON.stringify(parsed.authentication)).not.toContain('internal.example.com'); }); }); + +describe('direct Worker upload discovery', () => { + it('advertises direct Worker uploads on every document this platform serves', () => { + for (const input of [base, modern]) { + const discovery = createPlatformDiscoveryDocument(input); + expect(discovery.directWorkerUpload).toBe(true); + expect(parsePlatformDiscoveryDocument(discovery)).toEqual(discovery); + } + }); + + it('reports only what an older platform said', () => { + const { directWorkerUpload: _, ...older } = createPlatformDiscoveryDocument(base); + expect(parsePlatformDiscoveryDocument(older)).not.toHaveProperty('directWorkerUpload'); + expect( + parsePlatformDiscoveryDocument({ ...older, directWorkerUpload: false }).directWorkerUpload, + ).toBe(false); + expect(() => parsePlatformDiscoveryDocument({ ...older, directWorkerUpload: 'yes' })).toThrow( + 'directWorkerUpload must be a boolean.', + ); + }); +}); diff --git a/packages/void/src/cli/client.ts b/packages/void/src/cli/client.ts index 048f39ceb..36ef60ca2 100644 --- a/packages/void/src/cli/client.ts +++ b/packages/void/src/cli/client.ts @@ -13,7 +13,14 @@ import { type ProjectPermission, } from './platform-errors.ts'; import { normalizePlatformApiUrl } from './platform-registry.ts'; -import type { DeployAssetUploadRequest, DeployEvent, DeployUploadPayload } from './types.ts'; +import type { + DeployAssetUploadRequest, + DeployEvent, + DeployUploadPayload, + DeployWorkerModule, + DeployWorkerUploadRequest, + DeployWorkerUploadResponse, +} from './types.ts'; // Null omits `X-Void-CLI-Version`: the platform answers an unparseable version with 426. function readCliVersion(): string | null { @@ -34,21 +41,22 @@ export const MIN_PLATFORM_VERSION = '1.1.11'; const RELEASE_VERSION = /^(\d+)\.(\d+)\.(\d+)(-[0-9A-Za-z.-]+)?(\+[0-9A-Za-z.-]+)?$/; /** - * Version checks keyed by platform base URL. Every outcome (pass, inconclusive, - * or too old) is kept for the rest of the process, so polling commands probe - * discovery once. + * Discovery probes keyed by platform base URL. Every outcome (pass, + * inconclusive, or too old) is kept for the rest of the process, so polling + * commands probe discovery once. */ -const platformVersionChecks = new Map>(); +const platformVersionProbes = new Map>(); /** * What public discovery says about the platform version: - * - `version`: the reported `platformVersion` + * - `version`: the reported `platformVersion`, and whether the platform + * accepts Worker modules uploaded straight to R2 * - `missing`: discovery answered 404, so the platform predates discovery * - `unknown`: no usable answer (network error, timeout, disabled platform, * Access login page, or an invalid document) */ type PlatformVersionProbe = - | { kind: 'version'; version: string } + | { kind: 'version'; version: string; directWorkerUpload: boolean } | { kind: 'missing' } | { kind: 'unknown' }; @@ -661,19 +669,34 @@ export class PlatformClient { /** * Checks the platform version once per process and base URL. Concurrent - * first requests share one check, and its outcome is kept for later requests. + * first requests share one probe, and its outcome is kept for later requests. */ - private ensureSupportedPlatform(): Promise { - const apiUrl = this.baseUrl; - const cached = platformVersionChecks.get(apiUrl); + private async ensureSupportedPlatform(): Promise { + assertSupportedPlatform(this.baseUrl, await this.platformProbe()); + } + + private platformProbe(): Promise { + const cached = platformVersionProbes.get(this.baseUrl); if (cached) { return cached; } - const check = this.probePlatformVersion().then((probe) => - assertSupportedPlatform(apiUrl, probe), - ); - platformVersionChecks.set(apiUrl, check); - return check; + const probe = this.probePlatformVersion(); + platformVersionProbes.set(this.baseUrl, probe); + return probe; + } + + /** + * Whether discovery says the platform accepts Worker modules uploaded to R2. + * False when discovery omits the field or gave no usable answer. + */ + async supportsDirectWorkerUpload(): Promise { + let probe = await this.platformProbe(); + if (probe.kind === 'unknown') { + // One failed probe must not send a large Worker inline. Ask again, + // without replacing the outcome the version check keeps. + probe = await this.probePlatformVersion(); + } + return probe.kind === 'version' && probe.directWorkerUpload; } /** Reads the platform version from public discovery, without credentials. */ @@ -695,7 +718,11 @@ export class PlatformClient { } try { const discovery = await readPlatformDiscoveryResponse(this.baseUrl, response); - return { kind: 'version', version: discovery.platformVersion }; + return { + kind: 'version', + version: discovery.platformVersion, + directWorkerUpload: discovery.directWorkerUpload === true, + }; } catch { return { kind: 'unknown' }; } @@ -1117,6 +1144,53 @@ export class PlatformClient { return json.urls; } + /** + * Requests one presigned R2 PUT per Worker module, in `modules` order. The + * platform refuses a Worker over Cloudflare's size limit here (413 + * `worker_too_large`), before any bytes are uploaded. + */ + async requestWorkerUploadUrls( + projectId: string, + modules: Array, + ): Promise { + const url = `${this.baseUrl}/projects/${projectId}/deploy/worker-upload-urls`; + this.cliLog?.info('worker_upload_urls_post', { url, moduleCount: modules.length }); + + const request: DeployWorkerUploadRequest = { modules }; + const res = await this.request(url, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + ...this.authHeaders, + }, + body: JSON.stringify(request), + }); + this.cliLog?.info('worker_upload_urls_response', { + status: res.status, + cfRay: res.headers.get('cf-ray'), + }); + + if (!res.ok) { + await throwPlatformApiError('Failed to request worker upload URLs', res, 'deploy'); + } + + // Module i uploads to urls[i], so a short or malformed list must not start. + const json = (await res.json()) as Partial | null; + if ( + typeof json?.id !== 'string' || + !json.id || + !Array.isArray(json.urls) || + json.urls.length !== modules.length || + !json.urls.every( + (entry) => + typeof entry?.url === 'string' && typeof entry.headers === 'object' && !!entry.headers, + ) + ) { + throw new Error('Invalid worker upload URLs from platform.'); + } + return { id: json.id, urls: json.urls }; + } + async *finalizeDeploy( projectId: string, payload: ManifestOnlyDeployPayload, diff --git a/packages/void/src/cli/deploy-managed-events.ts b/packages/void/src/cli/deploy-managed-events.ts index 4507ab002..6a2372dd3 100644 --- a/packages/void/src/cli/deploy-managed-events.ts +++ b/packages/void/src/cli/deploy-managed-events.ts @@ -1,3 +1,4 @@ +import { createHash } from 'node:crypto'; import { createDeployArtifact, createDeployUploadPayload } from '@void/deploy-core'; import { getContentType } from '@void/edge'; import { type SpinnerResult } from '@voidzero-dev/vite-plus-prompts'; @@ -17,7 +18,7 @@ import { lintRoutingPhases, mergeRoutingRules } from './headers.ts'; import { type CliLogger } from './log-file.ts'; import { formatCount } from './output.ts'; import { type AssetManifestEntry } from './package.ts'; -import type { DeployEvent, DeployManifest } from './types.ts'; +import type { DeployEvent, DeployManifest, DeployWorkerInput } from './types.ts'; /** * Apply the visible side effects of a streamed deploy event (spinner @@ -210,19 +211,68 @@ export async function streamDeployEvents( throw new Error(formatConnectionLostMessage(deploymentId, cliLog?.path ?? null)); } +/** + * Upload Worker modules straight to R2 and return the `workerInput` that + * finalization names them by. The platform checks the Worker against + * Cloudflare's size limit when it issues the URLs, so a Worker that is too + * large fails before any bytes are sent. Each buffer leaves `workerFiles` + * once R2 acknowledges its PUT. + */ +async function uploadWorkerModules( + client: PlatformClient, + projectId: string, + workerFiles: Map, + s: SpinnerResult, + cliLog: CliLogger | undefined, +): Promise { + const modules = Array.from(workerFiles, ([name, body]) => ({ + name, + size: body.length, + sha256: createHash('sha256').update(body).digest('hex'), + })); + + s.message(`Requesting upload URLs for ${formatCount(modules.length, 'worker module')}...`); + cliLog?.info('worker_presign_request_start', { + moduleCount: modules.length, + totalBytes: modules.reduce((total, module) => total + module.size, 0), + }); + const presignStartedAt = Date.now(); + const { id, urls } = await client.requestWorkerUploadUrls(projectId, modules); + cliLog?.info('worker_presign_request_end', { + uploadId: id, + durationMs: Date.now() - presignStartedAt, + }); + + const items: Array = modules.map((module, index) => ({ + path: module.name, + hash: module.sha256, + body: workerFiles.get(module.name), + url: urls[index].url, + headers: urls[index].headers, + })); + s.message(`Uploading ${formatCount(items.length, 'worker module')} to R2...`); + await uploadAssetsToR2(items, cliLog, undefined, (item) => { + workerFiles.delete(item.path); + }); + return { id, modules }; +} + /** * Direct-to-R2 deploy orchestration. Issues presigned PUT URLs for each * needed asset, uploads bytes directly to R2, then calls the manifest-only - * deploy endpoint to finalize. + * deploy endpoint to finalize. When the platform supports it, Worker modules + * go to R2 the same way, before the assets. * * Key invariants: * 1. No needed assets → skip presign + R2 PUT phase entirely (saves a * round-trip when every asset was already present from a prior deploy). - * 2. Asset bytes come exclusively from the in-memory `assetFiles` cache - * computed by `collectAndHashAssets` — files are NEVER re-read from - * disk during this flow. + * 2. Asset and Worker bytes come exclusively from the in-memory + * `assetFiles` / `workerFiles` caches computed by `collectAndHashAssets` + * and `collectWorkerFiles` — files are NEVER re-read from disk during + * this flow. * 3. R2 PUT failures (4xx, 5xx, network) are surfaced with the asset - * path + hash so the user can correlate against the JSONL trace. + * path + hash (or Worker module name + SHA-256) so the user can + * correlate against the JSONL trace. */ export async function streamDirectR2Deploy({ client, @@ -250,6 +300,11 @@ export async function streamDirectR2Deploy({ spinner: SpinnerResult; cliLog?: CliLogger; }): Promise { + const workerInput = + workerFiles && workerFiles.size > 0 && (await client.supportsDirectWorkerUpload()) + ? await uploadWorkerModules(client, projectId, workerFiles, s, cliLog) + : undefined; + if (needed.length > 0) { // Drop buffers the server already has so peak memory tracks `needed`, // not the full bundle. @@ -349,11 +404,12 @@ export async function streamDirectR2Deploy({ cliLog?.info('presign_skipped', { reason: 'no_needed_assets' }); } - // Build the JSON-mode payload. Worker bytes go in as base64 — the - // manifest-only endpoint's design (0068) keeps worker bundles in the - // body since they are bounded at 32 MiB total. + // Build the JSON-mode payload. Uploaded Worker modules are named by + // `workerInput`. Otherwise their bytes go in as base64: a temporary + // fallback for platforms without direct Worker uploads. Delete it when + // MIN_PLATFORM_VERSION reaches the release that ships them. let encodedWorkerFiles: Record | undefined; - if (workerFiles && workerFiles.size > 0) { + if (!workerInput && workerFiles && workerFiles.size > 0) { encodedWorkerFiles = {}; for (const [relPath, buf] of workerFiles) { encodedWorkerFiles[relPath] = buf.toString('base64'); @@ -364,18 +420,23 @@ export async function streamDirectR2Deploy({ assetManifest, workerFiles: encodedWorkerFiles, }); - const payload: ManifestOnlyDeployPayload = createDeployUploadPayload(artifact, { - source: getTokenSource(), - commit: gitMetadata?.commit ?? null, - repositoryUrl: gitMetadata?.repositoryUrl ?? null, - commitUrl: gitMetadata?.commitUrl ?? null, - gitRef: gitMetadata?.gitRef ?? null, - preflight: preflightUsed ? 'true' : null, - }); + const payload: ManifestOnlyDeployPayload = createDeployUploadPayload( + artifact, + { + source: getTokenSource(), + commit: gitMetadata?.commit ?? null, + repositoryUrl: gitMetadata?.repositoryUrl ?? null, + commitUrl: gitMetadata?.commitUrl ?? null, + gitRef: gitMetadata?.gitRef ?? null, + preflight: preflightUsed ? 'true' : null, + }, + workerInput, + ); s.message('Finalizing deploy...'); cliLog?.info('finalize_start', { - workerFileCount: workerFiles?.size ?? 0, + workerFileCount: workerInput?.modules.length ?? workerFiles?.size ?? 0, + ...(workerInput && { workerUploadId: workerInput.id }), assetCount: Object.keys(assetManifest).length, }); const finalizeStartedAt = Date.now(); diff --git a/packages/void/src/cli/direct-r2-upload.ts b/packages/void/src/cli/direct-r2-upload.ts index 7d2f0c0e5..c82290e6a 100644 --- a/packages/void/src/cli/direct-r2-upload.ts +++ b/packages/void/src/cli/direct-r2-upload.ts @@ -42,28 +42,32 @@ export class DirectR2PutError extends Error { } export type DirectR2PutItem = { - /** Asset path relative to the build output (with leading `/`). */ + /** + * Asset path relative to the build output (with leading `/`), or a Worker + * module path relative to the Worker output. + */ path: string; - /** BLAKE3 storage hash — used for log correlation and R2 key. */ + /** Asset BLAKE3 storage hash or Worker module SHA-256, for log correlation. */ hash: string; /** - * Asset bytes. Cleared (set to `undefined`) by `uploadAssetsToR2` after a + * Bytes to upload. Cleared (set to `undefined`) by `uploadAssetsToR2` after a * successful PUT so the buffer becomes eligible for GC as soon as the * upload is acknowledged. Callers should not read `body` after the upload * completes. */ body: Buffer | undefined; - /** Presigned URL from `requestUploadUrls`. */ + /** Presigned URL from `requestUploadUrls` or `requestWorkerUploadUrls`. */ url: string; - /** Headers the platform required us to send (incl. Content-Md5, Content-Type). */ + /** Headers the platform required us to send (e.g. Content-Md5, Content-Type). */ headers: Record; }; /** - * Issue direct-to-R2 PUTs for every item in `items`, bounded to - * `R2_PUT_CONCURRENCY`. Surfaces the first failure (4xx immediately, 5xx - * after retry exhaustion) as a `DirectR2PutError` carrying the asset path - * and hash so the user can correlate against the JSONL log. + * Issue direct-to-R2 PUTs for every item in `items` (assets or Worker + * modules), bounded to `R2_PUT_CONCURRENCY`. Surfaces the first failure + * (4xx immediately, 5xx after retry exhaustion) as a `DirectR2PutError` + * carrying the item path and hash so the user can correlate against the + * JSONL log. * * Logs `r2_put_start` / `r2_put_end` per asset, plus `r2_put_retry` on * each retry attempt. Per-asset durations help identify slow uploads. diff --git a/packages/void/src/cli/types.ts b/packages/void/src/cli/types.ts index 9d6b6b7c9..218b0ec79 100644 --- a/packages/void/src/cli/types.ts +++ b/packages/void/src/cli/types.ts @@ -9,6 +9,10 @@ export type { DeployTarget, DeployTargetAdapter, DeployUploadPayload, + DeployWorkerInput, + DeployWorkerModule, + DeployWorkerUploadRequest, + DeployWorkerUploadResponse, HeaderRule, LiveManifestMeta, QueueMeta, diff --git a/packages/void/test/helpers/platform-discovery.ts b/packages/void/test/helpers/platform-discovery.ts index a2d5ceb05..15ade83e1 100644 --- a/packages/void/test/helpers/platform-discovery.ts +++ b/packages/void/test/helpers/platform-discovery.ts @@ -2,13 +2,18 @@ import { MIN_PLATFORM_VERSION } from '../../src/cli/client.ts'; const DISCOVERY_PATH = '/.well-known/void-platform'; +/** Optional discovery fields. A field left out is absent, as on an older platform. */ +export type PlatformDiscoveryFeatures = { directWorkerUpload?: unknown }; + /** A valid discovery response for `apiUrl` that advertises `platformVersion`. */ export function platformDiscoveryResponse( apiUrl: string, platformVersion: string = MIN_PLATFORM_VERSION, + features: PlatformDiscoveryFeatures = {}, ): Response { const origin = new URL(apiUrl).origin; return Response.json({ + ...features, apiUrl: origin, applicationDomain: 'void.app', authenticationProviders: ['github'], @@ -42,6 +47,7 @@ export function isPlatformDiscoveryRequest(input: unknown): boolean { export function withPlatformDiscovery( fetchMock: unknown, platformVersion: string = MIN_PLATFORM_VERSION, + features?: PlatformDiscoveryFeatures, ): typeof fetch { const next = fetchMock as (input: RequestInfo | URL, init?: RequestInit) => Promise; return async (input, init) => @@ -49,6 +55,7 @@ export function withPlatformDiscovery( ? platformDiscoveryResponse( input instanceof Request ? input.url : String(input), platformVersion, + features, ) : next(input, init); } diff --git a/packages/void/test/unit/client-platform-version.test.ts b/packages/void/test/unit/client-platform-version.test.ts index 1bf0f7623..e2a0dab96 100644 --- a/packages/void/test/unit/client-platform-version.test.ts +++ b/packages/void/test/unit/client-platform-version.test.ts @@ -3,6 +3,7 @@ import { PlatformClient } from '../../src/cli/client.ts'; import { isPlatformDiscoveryRequest, platformDiscoveryResponse, + type PlatformDiscoveryFeatures, } from '../helpers/platform-discovery.ts'; // Never read the keychain or run cloudflared. Without Access credentials, an @@ -160,3 +161,77 @@ describe('minimum platform version', () => { await expect(client().listProjectChoices()).resolves.toEqual([]); }); }); + +describe('direct Worker upload discovery', () => { + function serveFeatures(features: PlatformDiscoveryFeatures): void { + serve(() => platformDiscoveryResponse(origin, undefined, features)); + } + + it.each([ + ['true', { directWorkerUpload: true }, true], + ['false', { directWorkerUpload: false }, false], + ['absent, as on an older platform', {}, false], + ])('reports support only when discovery says so (%s)', async (_label, features, expected) => { + serveFeatures(features); + await expect(client().supportsDirectWorkerUpload()).resolves.toBe(expected); + }); + + it('reuses the version check probe', async () => { + serveFeatures({ directWorkerUpload: true }); + await client().listProjectChoices(); + await expect(client().supportsDirectWorkerUpload()).resolves.toBe(true); + expect(requestedPaths()).toEqual(['/.well-known/void-platform', '/projects/choices']); + }); + + it.each([ + ['a disabled platform (503)', () => Response.json({ error: 'disabled' }, { status: 503 })], + [ + 'an invalid field', + () => platformDiscoveryResponse(origin, undefined, { directWorkerUpload: 'yes' }), + ], + ])( + 'reports no support when discovery gives %s, and lets requests run', + async (_label, answer) => { + serve(answer); + await expect(client().supportsDirectWorkerUpload()).resolves.toBe(false); + await expect(client().listProjectChoices()).resolves.toEqual([]); + }, + ); + + it('asks discovery again when the cached answer was inconclusive', async () => { + let probes = 0; + serve(() => + ++probes === 1 + ? Response.json({ error: 'unavailable' }, { status: 503 }) + : platformDiscoveryResponse(origin, undefined, { directWorkerUpload: true }), + ); + await client().listProjectChoices(); + await expect(client().supportsDirectWorkerUpload()).resolves.toBe(true); + // The version check keeps its first outcome. + await client().listProjects(); + expect(requestedPaths()).toEqual([ + '/.well-known/void-platform', + '/projects/choices', + '/.well-known/void-platform', + '/projects', + ]); + }); + + it('reports no support when discovery stays inconclusive', async () => { + serve(() => Response.json({ error: 'unavailable' }, { status: 503 })); + await client().listProjectChoices(); + await expect(client().supportsDirectWorkerUpload()).resolves.toBe(false); + expect(requestedPaths()).toEqual([ + '/.well-known/void-platform', + '/projects/choices', + '/.well-known/void-platform', + ]); + }); + + it('does not ask again when discovery omits the field', async () => { + serveFeatures({}); + await client().listProjectChoices(); + await expect(client().supportsDirectWorkerUpload()).resolves.toBe(false); + expect(requestedPaths()).toEqual(['/.well-known/void-platform', '/projects/choices']); + }); +}); diff --git a/packages/void/test/unit/direct-r2-client.test.ts b/packages/void/test/unit/direct-r2-client.test.ts index affaa299d..bda60d9ce 100644 --- a/packages/void/test/unit/direct-r2-client.test.ts +++ b/packages/void/test/unit/direct-r2-client.test.ts @@ -120,6 +120,58 @@ describe('PlatformClient.requestUploadUrls', () => { }); }); +describe('PlatformClient.requestWorkerUploadUrls', () => { + const modules = [ + { name: 'index.js', size: 18, sha256: 'a'.repeat(64) }, + { name: 'chunks/b.js', size: 4, sha256: 'b'.repeat(64) }, + ]; + const urls = [ + { url: 'https://r2/worker/0', headers: { 'Content-Type': 'application/javascript' } }, + { url: 'https://r2/worker/1', headers: { 'Content-Type': 'application/javascript' } }, + ]; + + it('POSTs the modules in order and returns the upload id and URLs', async () => { + fetchMock.mockResolvedValueOnce(jsonResponse({ id: 'wup_1', urls })); + + await expect(makeClient().requestWorkerUploadUrls(projectId, modules)).resolves.toEqual({ + id: 'wup_1', + urls, + }); + + const [url, init] = fetchMock.mock.calls[0] as [string, RequestInit]; + expect(url).toBe(`${apiUrl}/projects/${projectId}/deploy/worker-upload-urls`); + expect(init.method).toBe('POST'); + const headers = init.headers as Record; + expect(headers['Content-Type']).toBe('application/json'); + expect(headers.Authorization).toBe(`Bearer ${token}`); + expect(JSON.parse(init.body as string)).toEqual({ modules }); + }); + + it("shows the platform's message when the Worker is over the size limit", async () => { + const message = 'This Worker is 70 MB without source maps. Cloudflare allows at most 64 MiB.'; + fetchMock.mockResolvedValueOnce(jsonResponse({ error: 'worker_too_large', message }, 413)); + + const request = makeClient().requestWorkerUploadUrls(projectId, modules); + await expect(request).rejects.toMatchObject({ + name: 'PlatformApiError', + status: 413, + code: 'worker_too_large', + }); + await expect(request).rejects.toThrow(message); + }); + + it.each([ + ['without an id', { urls }], + ['with fewer URLs than modules', { id: 'wup_1', urls: urls.slice(1) }], + ['with a malformed URL entry', { id: 'wup_1', urls: [urls[0], { url: 'https://r2/x' }] }], + ])('rejects a response %s', async (_label, body) => { + fetchMock.mockResolvedValueOnce(jsonResponse(body)); + await expect(makeClient().requestWorkerUploadUrls(projectId, modules)).rejects.toThrow( + 'Invalid worker upload URLs from platform.', + ); + }); +}); + describe('PlatformClient.getMigrationStatus', () => { it('passes the dialect query parameter when provided', async () => { fetchMock.mockResolvedValueOnce(jsonResponse({ applied: [] })); @@ -553,6 +605,39 @@ describe('PlatformClient.finalizeDeploy', () => { return events; } + it('re-sends the same workerInput and idempotency key on a transport retry', async () => { + vi.useFakeTimers(); + fetchMock + .mockRejectedValueOnce(new TypeError('fetch failed')) + .mockResolvedValueOnce(ndjsonStreamResponse([durableDone])); + const workerInput = { + id: 'wup_retry', + modules: [{ name: 'index.js', size: 18, sha256: 'c'.repeat(64) }], + }; + try { + const deploy = (async () => { + for await (const _event of makeClient().finalizeDeploy(projectId, { + ...makePayload(), + workerInput, + })) { + // Only the request bodies matter here. + } + })(); + await vi.runAllTimersAsync(); + await deploy; + + const [first, second] = deployPostCalls().map(([, init]) => init as RequestInit); + expect(second.body).toBe(first.body); + expect(JSON.parse(first.body as string)).toMatchObject({ workerInput }); + expect(JSON.parse(first.body as string)).not.toHaveProperty('workerFiles'); + const [firstKey, secondKey] = idempotencyKeys(); + expect(firstKey).toMatch(/^[a-f0-9-]{36}$/); + expect(secondKey).toBe(firstKey); + } finally { + vi.useRealTimers(); + } + }); + it('keeps one idempotency key across POST retries and chooses a new key for each invocation', async () => { vi.useFakeTimers(); fetchMock diff --git a/packages/void/test/unit/direct-r2-deploy.test.ts b/packages/void/test/unit/direct-r2-deploy.test.ts index 42bf8321f..0aa73a166 100644 --- a/packages/void/test/unit/direct-r2-deploy.test.ts +++ b/packages/void/test/unit/direct-r2-deploy.test.ts @@ -1,7 +1,10 @@ +import { createHash } from 'node:crypto'; import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; +import type { SpinnerResult } from '@voidzero-dev/vite-plus-prompts'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vite-plus/test'; +import type { DeployManifest } from '../../src/cli/types.ts'; import { withPlatformDiscovery } from '../helpers/platform-discovery.ts'; // Spy on `readFileSync` to count asset reads. Mocked before any importer of @@ -58,6 +61,8 @@ const fsModule = await import('node:fs'); const readFileSyncMock = fsModule.readFileSync as unknown as ReturnType; const { runStaticDeploy } = await import('../../src/cli/deploy-managed-static.ts'); +const { streamDirectR2Deploy } = await import('../../src/cli/deploy-managed-events.ts'); +const { collectAndHashAssets } = await import('../../src/cli/package.ts'); const { openDeployLog } = await import('../../src/cli/log-file.ts'); const { PlatformClient } = await import('../../src/cli/client.ts'); @@ -95,22 +100,36 @@ afterEach(() => { } }); +type WorkerModule = { name: string; size: number; sha256: string }; + type MockState = { preflightCalls: Array; uploadUrlsCalls: Array; + workerUploadUrlsCalls: Array<{ modules: Array }>; r2PutCalls: Array<{ url: string; init: RequestInit }>; finalizeCalls: Array<{ url: string; init: RequestInit }>; }; +/** Headers the mock platform signs into each Worker module's PUT. */ +function workerPutHeaders(module: WorkerModule): Record { + return { + 'Content-Type': 'application/javascript+module', + 'Content-Length': String(module.size), + 'x-amz-checksum-sha256': Buffer.from(module.sha256, 'hex').toString('base64'), + }; +} + function installPlatformMocks(opts?: { preflightSkipped?: number; preflightNeeded?: Array; r2PutBehavior?: (url: string) => Response | Promise; + workerUploadUrls?: () => Response; finalizeEvents?: Array; }): MockState { const state: MockState = { preflightCalls: [], uploadUrlsCalls: [], + workerUploadUrlsCalls: [], r2PutCalls: [], finalizeCalls: [], }; @@ -153,6 +172,21 @@ function installPlatformMocks(opts?: { }); } + if (url.endsWith('/deploy/worker-upload-urls') && method === 'POST') { + const body = JSON.parse(reqInit.body as string) as { modules: Array }; + state.workerUploadUrlsCalls.push(body); + if (opts?.workerUploadUrls) { + return opts.workerUploadUrls(); + } + return Response.json({ + id: 'wup_test', + urls: body.modules.map((module, index) => ({ + url: `https://r2-test/worker/${index}`, + headers: workerPutHeaders(module), + })), + }); + } + if (url.startsWith('https://r2-test/') && method === 'PUT') { state.r2PutCalls.push({ url, init: reqInit }); if (opts?.r2PutBehavior) { @@ -537,6 +571,127 @@ describe('direct-to-R2 deploy flow', () => { }); }); +describe('Worker module uploads', () => { + // Discovery is cached per origin, so each origin is only probed with one answer. + const directOrigin = 'https://api-direct.test'; + const workerManifest: DeployManifest = { + version: 3, + assetManifestVersion: 2, + bindings: {}, + type: 'void', + ssr: true, + workerMain: 'index.js', + }; + + function workerFiles(): Map { + return new Map([ + ['index.js', Buffer.from('export default { fetch() {} };\n')], + ['chunks/large.js', Buffer.alloc(150_000, 0x61)], + ]); + } + + function expectedModules(): Array { + return Array.from(workerFiles(), ([name, bytes]) => ({ + name, + size: bytes.length, + sha256: createHash('sha256').update(bytes).digest('hex'), + })); + } + + function advertiseDirectWorkerUpload(): void { + globalThis.fetch = withPlatformDiscovery(fetchMock, undefined, { directWorkerUpload: true }); + } + + async function deployWorker( + origin: string, + files: Map, + needed: Array = [], + ): Promise { + const { assetManifest, assetFiles } = await collectAndHashAssets(assetsDir); + await streamDirectR2Deploy({ + client: new PlatformClient('test-token', { apiUrl: origin }), + projectId: 'proj_t', + manifest: workerManifest, + assetManifest, + assetFiles, + needed, + preflightUsed: false, + workerFiles: files, + spinner: noopSpinner as unknown as SpinnerResult, + }); + } + + it('uploads modules to R2 before the assets and finalizes with workerInput', async () => { + advertiseDirectWorkerUpload(); + const state = installPlatformMocks(); + const files = workerFiles(); + + await deployWorker(directOrigin, files, ['/index.html']); + + const modules = expectedModules(); + expect(state.workerUploadUrlsCalls).toEqual([{ modules }]); + const workerPuts = state.r2PutCalls.filter(({ url }) => url.includes('/worker/')); + expect(workerPuts.map(({ url }) => url).sort()).toEqual([ + 'https://r2-test/worker/0', + 'https://r2-test/worker/1', + ]); + const bytes = [...workerFiles().values()]; + for (const { url, init } of workerPuts) { + const index = Number(url.split('/').pop()); + expect(init.headers).toEqual(workerPutHeaders(modules[index])); + expect(Buffer.from(init.body as Uint8Array)).toEqual(bytes[index]); + } + + // A Worker the platform refuses must fail before any asset upload. + const urls = fetchMock.mock.calls.map(([url]) => String(url)); + const assetPresign = urls.indexOf(`${directOrigin}/projects/proj_t/deploy/upload-urls`); + expect(urls.indexOf(`${directOrigin}/projects/proj_t/deploy/worker-upload-urls`)).toBe(0); + expect(assetPresign).toBeGreaterThan(0); + expect(urls.slice(assetPresign).filter((url) => url.includes('/worker/'))).toEqual([]); + expect(state.uploadUrlsCalls).toHaveLength(1); + + const body = JSON.parse(state.finalizeCalls[0].init.body as string); + expect(body.workerInput).toEqual({ id: 'wup_test', modules }); + expect(body).not.toHaveProperty('workerFiles'); + // Each buffer is released once R2 has it. + expect(files.size).toBe(0); + }); + + it('keeps sending modules inline when discovery does not advertise direct uploads', async () => { + const state = installPlatformMocks(); + + await deployWorker('https://api-inline.test', workerFiles()); + + expect(state.workerUploadUrlsCalls).toHaveLength(0); + expect(state.r2PutCalls).toHaveLength(0); + const body = JSON.parse(state.finalizeCalls[0].init.body as string); + expect(body).not.toHaveProperty('workerInput'); + expect(body.workerFiles).toEqual( + Object.fromEntries( + Array.from(workerFiles(), ([name, bytes]) => [name, bytes.toString('base64')]), + ), + ); + }); + + it('uploads nothing when the platform refuses the Worker size', async () => { + advertiseDirectWorkerUpload(); + const message = 'This Worker is 70 MB without source maps. Cloudflare allows at most 64 MiB.'; + const state = installPlatformMocks({ + workerUploadUrls: () => + Response.json({ error: 'worker_too_large', message }, { status: 413 }), + }); + + await expect( + deployWorker(directOrigin, workerFiles(), Object.keys(fixtureFiles)), + ).rejects.toThrow(message); + + expect(state.workerUploadUrlsCalls).toHaveLength(1); + expect(state.uploadUrlsCalls).toHaveLength(0); + expect(state.r2PutCalls).toHaveLength(0); + expect(state.finalizeCalls).toHaveLength(0); + }); +}); + function escapeRegex(s: string): string { return s.replace(/[.*+?^${}()|[\]\\]/g, '\\$&'); } diff --git a/platform/packages/api/src/deploy/execute.ts b/platform/packages/api/src/deploy/execute.ts index b3f56bb5d..1eae1e224 100644 --- a/platform/packages/api/src/deploy/execute.ts +++ b/platform/packages/api/src/deploy/execute.ts @@ -22,12 +22,18 @@ import { } from '../db.ts'; import { assertDeploymentExecutionAuthority, + deploymentInputModule, + deploymentInputPrefix, deploymentReceiptResponse, dispatchDeploymentOperation, findDeploymentOperation, getDeploymentOperation, installationDeploymentAdmissionClosed, prepareDeploymentInput, + recordUploadedDeploymentInput, + uploadedDeploymentInput, + uploadedWorkerInputProblem, + type DeploymentInput, } from '../deployment-execution.ts'; import { getSharedEmailRecipientReadiness, @@ -49,6 +55,7 @@ import { type Deployment, type Project, } from '../types.ts'; +import { blobWorkerModule, type WorkerModule } from '../worker-upload.ts'; import { zeroTrustNotReadyMessage } from '../zero-trust/errors.ts'; import { resolveProjectZeroTrustRouting } from '../zero-trust/routing.ts'; import { @@ -83,8 +90,13 @@ import { MAX_WORKER_FILE_BYTES, moduleContentType, workerFilesError, + workerInputProblem, + type WorkerInput, } from './shared.ts'; +const WORKER_UPLOAD_CLAIMED_MESSAGE = + 'this Worker upload belongs to another deployment; run the deploy again'; + /** * JSON-mode (manifest-only) deploy handler — direct-to-R2 asset upload flow. * @@ -94,8 +106,10 @@ import { * - `manifest` (DeployManifest, must declare `assetManifestVersion: 2`) * - `assetManifest` (path → { blake3, md5, size }) * - `source`, `commit`, `repositoryUrl`, `commitUrl`, `gitRef`, `preflight` - * - `workerFiles` (base64-encoded module bytes for WfP deploys; absent for - * static-only deploys) + * - for WfP deploys, either `workerInput` (the id and module list of a + * Worker upload the CLI made through `/deploy/worker-upload-urls`) or, + * from older CLIs and connected apps, `workerFiles` (base64-encoded + * module bytes). Static-only deploys send neither. * * The bytes for declared assets MUST already be in `STATIC_ASSETS_R2` under * `assets/${projectId}/${blake3}`; the deploy verifies them before it @@ -221,7 +235,7 @@ export async function handleManifestOnlyDeploy(context: Context): Promis if (overflowed) { return context.json( { - error: `request body exceeds ${MAX_JSON_DEPLOY_BODY_BYTES} bytes (asset bytes must be uploaded direct-to-R2, not in the request body)`, + error: `request body exceeds ${MAX_JSON_DEPLOY_BODY_BYTES} bytes: the Worker modules, including source maps, are too large to send inside the deploy request. Upgrade the void CLI; current versions upload Worker modules to storage directly.`, }, 413, ); @@ -265,6 +279,17 @@ export async function handleManifestOnlyDeploy(context: Context): Promis ); } if (context.get('oauthApproval') && !context.get('deploymentExecution')) { + // The approved artifact digest commits to inline `workerFiles` bytes, so + // an uploaded `workerInput` could not be checked against it. + if (body.workerInput !== undefined) { + return context.json( + { + event: 'error', + message: 'Connected app publishes must send Worker modules in workerFiles.', + }, + 403, + ); + } try { await verifyOAuthDeployment( context.get('db'), @@ -505,12 +530,51 @@ export async function handleManifestOnlyDeploy(context: Context): Promis const commitUrl = typeof body.commitUrl === 'string' ? body.commitUrl : null; const gitRef = typeof body.gitRef === 'string' ? body.gitRef : null; - // workerFiles is the only meaningful body content beyond the manifest - // tuples. Each entry is a base64-encoded module body keyed by relative - // path with a per-module cap of MAX_WORKER_FILE_BYTES (32 MiB). Absent or - // empty means a static-only deploy. + // Worker modules arrive one of two ways. Current CLIs upload them to R2 + // first and send `workerInput` (names, sizes, and digests). Older CLIs and + // connected apps send base64 `workerFiles` keyed by relative path, each + // capped at MAX_WORKER_FILE_BYTES (32 MiB). Neither means a static-only + // deploy. The Queue executor passes the modules of the admitted input. const execution = context.get('deploymentExecution'); - const workerFiles: Array<{ name: string; content: Blob }> = execution?.workerFiles ?? []; + const workerFiles: Array = execution?.workerFiles ?? []; + // Decoded inline modules, which durable admission stages in R2. + const inlineFiles: Array<{ name: string; content: Blob }> = []; + let workerInput: WorkerInput | null = null; + if (!execution && body.workerFiles !== undefined && body.workerInput !== undefined) { + return context.json( + { event: 'error', message: 'Send Worker modules in workerInput or workerFiles, not both.' }, + 400, + ); + } + if (!execution && body.workerInput !== undefined) { + const problem = workerInputProblem(body.workerInput); + if (problem) { + return context.json( + { + event: 'error', + ...(problem.error ? { error: problem.error } : {}), + message: problem.message, + }, + problem.status, + ); + } + workerInput = body.workerInput as WorkerInput; + } + // Keys derive from the authenticated project and the validated upload id, + // so a request can only name objects under its own project's prefix. + const uploadPrefix = workerInput ? deploymentInputPrefix(projectId, workerInput.id) : null; + const uploadedModules: DeploymentInput['modules'] = (workerInput?.modules ?? []).map( + (module, index) => ({ + name: module.name, + key: `${uploadPrefix}${index}`, + digest: module.sha256, + type: moduleContentType(module.name), + size: module.size, + }), + ); + for (const module of uploadedModules) { + workerFiles.push(deploymentInputModule(context.env.STATIC_ASSETS_R2, module, module.size!)); + } if (!execution && body.workerFiles !== undefined) { if ( body.workerFiles == null || @@ -567,12 +631,11 @@ export async function handleManifestOnlyDeploy(context: Context): Promis // Hand the bytes directly to a `Blob` so we never name the // `Uint8Array` past this scope — keeps per-iteration memory residency // bounded by one decoded module at a time. - workerFiles.push({ - name: filePath, - content: new Blob([decoded as unknown as BlobPart], { - type: moduleContentType(filePath), - }), + const content = new Blob([decoded as unknown as BlobPart], { + type: moduleContentType(filePath), }); + inlineFiles.push({ name: filePath, content }); + workerFiles.push(blobWorkerModule(filePath, content)); } } const hasWorkerFiles = workerFiles.length > 0; @@ -607,24 +670,69 @@ export async function handleManifestOnlyDeploy(context: Context): Promis if (workerFilesProblem) { return context.json({ event: 'error', message: workerFilesProblem }, 400); } + const durable = !execution && context.env.DEPLOYMENTS_QUEUE !== undefined; + const idempotencyKey = context.req.header('x-void-deploy-idempotency-key') ?? crypto.randomUUID(); + if (durable && !/^[A-Za-z0-9_-]{1,128}$/.test(idempotencyKey)) { + return context.json({ error: 'invalid deployment idempotency key' }, 400); + } + const inputOptions = { + appType: incomingManifest.type, + credentialId: context.get('deployCredential'), + oauthApproval: context.get('oauthApproval'), + }; + const uploadedInput = + durable && workerInput + ? await uploadedDeploymentInput(uploadPrefix!, body, uploadedModules, inputOptions) + : null; + if (uploadedInput) { + // A retry of an admitted deploy gets its receipt even after the + // operation's input cleanup deleted the upload. The precheck below + // rejects the same key with different input. + const existing = await findDeploymentOperation( + context.get('db'), + projectId, + actor.id, + idempotencyKey, + ); + if ( + existing && + existing.input_digest === uploadedInput.digest && + existing.build_id === (deployBuild ?? null) + ) { + return deploymentReceiptResponse(context.get('db'), existing.id); + } + } + if (workerInput) { + const uploadProblem = await uploadedWorkerInputProblem( + context.env.STATIC_ASSETS_R2, + uploadPrefix!, + workerInput.modules, + ); + if (uploadProblem) { + return context.json({ event: 'error', message: uploadProblem }, 400); + } + } // Pre-validation done — now run the atomic precheck-and-claim // transaction. FOR UPDATE on the project row, 10-minute stale-deploy // reaper, placeholder insert. Once the placeholder row exists, every // later failure must route through the catch / finally below to avoid // leaving it stranded in `pending`. - const durable = !execution && context.env.DEPLOYMENTS_QUEUE !== undefined; - const idempotencyKey = context.req.header('x-void-deploy-idempotency-key') ?? crypto.randomUUID(); - if (durable && !/^[A-Za-z0-9_-]{1,128}$/.test(idempotencyKey)) { - return context.json({ error: 'invalid deployment idempotency key' }, 400); - } - const stagedInput = durable - ? await prepareDeploymentInput(context.env.STATIC_ASSETS_R2, projectId, body, workerFiles, { - appType: incomingManifest.type, - credentialId: context.get('deployCredential'), - oauthApproval: context.get('oauthApproval'), - }) - : null; + const stagedInput = !durable + ? null + : uploadedInput + ? await recordUploadedDeploymentInput( + context.env.STATIC_ASSETS_R2, + uploadedInput, + incomingManifest.type, + ) + : await prepareDeploymentInput( + context.env.STATIC_ASSETS_R2, + projectId, + body, + inlineFiles, + inputOptions, + ); const cutoverOwnerId = execution?.operation.owner_id ?? crypto.randomUUID(); let precheckResult: | { @@ -691,6 +799,23 @@ export async function handleManifestOnlyDeploy(context: Context): Promis } return { receipt: existing.id } as const; } + if (workerInput) { + // One operation owns an uploaded prefix, and its input cleanup + // deletes it. A retry of the same deploy matched the key above. + const [claimed] = await tx + .select({ id: deploymentOperations.id }) + .from(deploymentOperations) + .where( + and( + eq(deploymentOperations.project_id, projectId), + eq(deploymentOperations.input_prefix, stagedInput.prefix), + ), + ) + .limit(1); + if (claimed) { + throw new Error(WORKER_UPLOAD_CLAIMED_MESSAGE); + } + } } const [unfinishedOperation] = await tx .select({ id: deploymentOperations.id }) @@ -1008,7 +1133,8 @@ export async function handleManifestOnlyDeploy(context: Context): Promis if ( err instanceof Error && (err.message === 'deployment admission closed for platform maintenance' || - err.message === 'deployment idempotency key was reused for different input') + err.message === 'deployment idempotency key was reused for different input' || + err.message === WORKER_UPLOAD_CLAIMED_MESSAGE) ) { return context.json({ error: err.message }, 409); } @@ -1017,7 +1143,10 @@ export async function handleManifestOnlyDeploy(context: Context): Promis } throw err; } - if (stagedInput && !('deployment' in precheckResult)) { + // Delete only input this request staged itself. A CLI upload's prefix can + // belong to the operation a retried request matched; its operation or the + // stale-input sweep removes it. + if (stagedInput && !workerInput && !('deployment' in precheckResult)) { await paginatedR2Delete(context.env.STATIC_ASSETS_R2, stagedInput.prefix).catch(() => {}); } if ('receipt' in precheckResult) { @@ -1091,5 +1220,10 @@ export async function handleManifestOnlyDeploy(context: Context): Promis gitRef, deployBuild, workerFiles, + // Without the Queue the deploy runs in this request. Delete the CLI's + // upload once it reaches a terminal state, as the Queue consumer does. + releaseWorkerInput: uploadPrefix + ? () => paginatedR2Delete(context.env.STATIC_ASSETS_R2, uploadPrefix).then(() => {}) + : undefined, }); } diff --git a/platform/packages/api/src/deploy/pipeline.ts b/platform/packages/api/src/deploy/pipeline.ts index 0f0da254c..a6cdbe6c1 100644 --- a/platform/packages/api/src/deploy/pipeline.ts +++ b/platform/packages/api/src/deploy/pipeline.ts @@ -57,6 +57,7 @@ import { type Project, type DeploymentStatus, } from '../types.ts'; +import type { WorkerModule } from '../worker-upload.ts'; import { resolveProjectZeroTrustRoutingForPublication, rewriteProjectRouteMetadata, @@ -104,7 +105,9 @@ export async function runDeploymentPipeline(input: { readonly commitUrl: string | null; readonly gitRef: string | null; readonly deployBuild: string | undefined; - readonly workerFiles: Array<{ name: string; content: Blob }>; + readonly workerFiles: Array; + /** Deletes the request's Worker upload once the pipeline settles. */ + readonly releaseWorkerInput?: () => Promise; }): Promise { let logger = input.logger; const { @@ -1193,6 +1196,14 @@ export async function runDeploymentPipeline(input: { err: err instanceof Error ? err.message : String(err), }); } + if (input.releaseWorkerInput) { + // Before the stream closes: closing the response can end the request. + await input.releaseWorkerInput().catch((err) => + logger.warn('worker_input', 'delete_failed', { + err: err instanceof Error ? err.message : String(err), + }), + ); + } await persistedEvents; streamOpen = false; try { diff --git a/platform/packages/api/src/deploy/publish-worker.ts b/platform/packages/api/src/deploy/publish-worker.ts index adef19d6a..f8ad01900 100644 --- a/platform/packages/api/src/deploy/publish-worker.ts +++ b/platform/packages/api/src/deploy/publish-worker.ts @@ -47,6 +47,7 @@ import { type Project, type DeploymentStatus, } from '../types.ts'; +import type { WorkerModule } from '../worker-upload.ts'; import { resolveProjectZeroTrustRouting } from '../zero-trust/routing.ts'; import { clearIsrCache, serializePrerenderHeaders, packIsrValue } from './cache.ts'; import { assertDeploymentCanProceed } from './guards.ts'; @@ -107,7 +108,7 @@ export async function publishWorkerDeployment(input: { knownRetainedHashes: ReadonlySet, ) => Promise; readonly projectDomains: Awaited>; - readonly workerFiles: Array<{ name: string; content: Blob }>; + readonly workerFiles: ReadonlyArray; readonly migrationsApplied: number; readonly finishOrphanSweep: () => Promise; }) { diff --git a/platform/packages/api/src/deploy/shared.ts b/platform/packages/api/src/deploy/shared.ts index cfbcf9ca7..1b2185931 100644 --- a/platform/packages/api/src/deploy/shared.ts +++ b/platform/packages/api/src/deploy/shared.ts @@ -12,11 +12,23 @@ export const MAX_UPLOAD_URLS_BATCH = 5000; export const DEFAULT_TAIL_SERVICE_NAME = 'void-tail'; -// CF's worker script size limit (free plan 1 MiB, paid plan 10 MiB compressed). -// Enforced per worker module so a single runaway worker file cannot OOM the -// request before the upload to CF ever happens. +// Per-module cap for Worker modules that older CLIs send inline as base64 +// `workerFiles`. It bounds the memory one decoded module holds in the request; +// it is not a Cloudflare limit. Modules uploaded to R2 (`workerInput`) are +// checked against CLOUDFLARE_WORKER_SIZE_LIMIT_BYTES instead. export const MAX_WORKER_FILE_BYTES = 32 * 1024 * 1024; +// Mirrors Cloudflare's Worker size limit: 64 MiB uncompressed on every plan, +// counting every module except source maps. Cloudflare limits source maps +// separately and enforces that limit itself. +// https://developers.cloudflare.com/workers/platform/limits/#worker-size +export const CLOUDFLARE_WORKER_SIZE_LIMIT_BYTES = 64 * 1024 * 1024; + +// A finalize accepts direct Worker uploads at most this old. The stale-input +// sweep (`dispatchPendingDeployments`) removes unclaimed inputs after 24 hours, +// so an accepted upload cannot be swept while its deploy is admitted or runs. +export const WORKER_UPLOAD_MAX_AGE_MS = 12 * 60 * 60 * 1000; + /** * Decode a base64 string to bytes with the lowest peak residency the * runtime allows. Prefers the (TC39 stage-4) `Uint8Array.fromBase64` helper @@ -95,16 +107,12 @@ export const ASSET_GC_MAX_HASHES_PER_DEPLOY = 200; // Per-blake3 / per-md5 validation: 32 lowercase hex characters. export const HEX_32_REGEX = /^[0-9a-f]{32}$/; -// Generous body cap for JSON-mode deploy POSTs. The body is dominated by -// `workerFiles` (base64-encoded user worker modules); MAX_WORKER_FILE_BYTES -// is 32 MiB per module and a worker can ship ~10 modules in practice. -// -// Base64 expansion is ~33%, so a single 32 MiB raw module weighs ~43 MiB -// on the wire. 64 MiB comfortably accommodates one full-size module (with -// safety margin) plus the manifest, assetManifest, vars, and source/commit -// JSON, while staying well under the 128 MiB Worker memory limit. Asset -// BYTES are NOT in this body — they go direct-to-R2 via presigned PUTs, -// which is the whole reason this endpoint exists. +// Body cap for JSON-mode deploy POSTs. Older CLIs send Worker modules inline +// as base64 `workerFiles`, which dominate the body; current CLIs upload them +// to R2 first (`workerInput`) and send only their names, sizes, and digests. +// The cap keeps the inline path's parsed body well under the 128 MiB Worker +// memory limit. Asset bytes never travel in this body — they go direct-to-R2 +// via presigned PUTs. export const MAX_JSON_DEPLOY_BODY_BYTES = 64 * 1024 * 1024; export type UploadUrlsRequestAsset = Readonly<{ @@ -176,6 +184,95 @@ export function workerFilesError( return null; } +const SHA256_HEX_REGEX = /^[0-9a-f]{64}$/; +const UUID_REGEX = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/; + +/** One Worker module the CLI uploads to R2 before it finalizes a deploy. */ +export type WorkerUploadModule = Readonly<{ name: string; size: number; sha256: string }>; + +/** A direct Worker upload the CLI finished before calling the deploy endpoint. */ +export type WorkerInput = Readonly<{ id: string; modules: ReadonlyArray }>; + +export type WorkerUploadProblem = Readonly<{ + status: 400 | 413; + error?: 'worker_too_large'; + message: string; +}>; + +function roundedMiB(bytes: number): string { + return String(Math.ceil((bytes / 1024 / 1024) * 10) / 10); +} + +/** + * Explain why a direct Worker upload's module list cannot be accepted, or + * return null. Names follow `workerModuleNameError`; the main module is + * checked once the manifest is known, at finalize. + */ +export function workerUploadModulesProblem( + value: unknown, + label: string, +): WorkerUploadProblem | null { + if (!Array.isArray(value) || value.length === 0 || value.length > MAX_UPLOAD_URLS_BATCH) { + return { + status: 400, + message: `${label} must be an array of 1 to ${MAX_UPLOAD_URLS_BATCH} modules`, + }; + } + const names = new Set(); + let codeBytes = 0; + for (let index = 0; index < value.length; index++) { + const entry: unknown = value[index]; + const at = `${label}[${index}]`; + if (entry == null || typeof entry !== 'object' || Array.isArray(entry)) { + return { status: 400, message: `${at} must be an object` }; + } + const { name, size, sha256 } = entry as Record; + if (typeof name !== 'string') { + return { status: 400, message: `${at}.name must be a string` }; + } + const nameError = workerModuleNameError(name); + if (nameError) { + return { status: 400, message: `${at}.name ${JSON.stringify(name)} ${nameError}` }; + } + if (names.has(name)) { + return { status: 400, message: `${at}.name ${JSON.stringify(name)} is listed twice` }; + } + names.add(name); + if (typeof size !== 'number' || !Number.isSafeInteger(size) || size < 0) { + return { status: 400, message: `${at}.size must be a non-negative integer` }; + } + if (typeof sha256 !== 'string' || !SHA256_HEX_REGEX.test(sha256)) { + return { status: 400, message: `${at}.sha256 must be 64 lowercase hex characters` }; + } + if (moduleContentType(name) !== 'application/source-map') { + codeBytes += size; + } + } + if (codeBytes > CLOUDFLARE_WORKER_SIZE_LIMIT_BYTES) { + return { + status: 413, + error: 'worker_too_large', + message: `The Worker is ${roundedMiB(codeBytes)} MiB without source maps, and Cloudflare allows ${roundedMiB(CLOUDFLARE_WORKER_SIZE_LIMIT_BYTES)} MiB.`, + }; + } + return null; +} + +/** Explain why a deploy body's `workerInput` cannot be accepted, or return null. */ +export function workerInputProblem(value: unknown): WorkerUploadProblem | null { + if (value == null || typeof value !== 'object' || Array.isArray(value)) { + return { status: 400, message: 'workerInput must be an object with id and modules' }; + } + const { id, modules } = value as Record; + if (typeof id !== 'string' || !UUID_REGEX.test(id)) { + return { + status: 400, + message: 'workerInput.id must be the id returned by /deploy/worker-upload-urls', + }; + } + return workerUploadModulesProblem(modules, 'workerInput.modules'); +} + export function moduleContentType(filename: string): string { if (filename.endsWith('.wasm')) { return 'application/wasm'; diff --git a/platform/packages/api/src/deploy/upload-worker.ts b/platform/packages/api/src/deploy/upload-worker.ts index 41950f89a..ec027cb8d 100644 --- a/platform/packages/api/src/deploy/upload-worker.ts +++ b/platform/packages/api/src/deploy/upload-worker.ts @@ -34,6 +34,7 @@ import { issueProjectProxyToken } from '../proxy-token-rotation.ts'; import { RESERVED_BINDING_NAMES } from '../reserved-bindings.ts'; import { deployments } from '../schema.ts'; import { type DeployManifest, type DeployEvent, type Deployment, type Project } from '../types.ts'; +import { withKnownLength, workerUploadBody, type WorkerModule } from '../worker-upload.ts'; import { buildWebSocketUploadBindings, buildSandboxUploadBindings, @@ -69,7 +70,7 @@ export async function uploadDeploymentWorker(input: { readonly assetsByHash: Map>; readonly incomingAssetManifest: Record; wfpMirroredManifestDeploymentId: string | null; - readonly workerFiles: Array<{ name: string; content: Blob }>; + readonly workerFiles: ReadonlyArray; sandboxControllerAttempted: boolean; readonly owner: ApiEnv['Variables']['user']; workerUploadUrl: string | undefined; @@ -237,6 +238,19 @@ export async function uploadDeploymentWorker(input: { session.buckets, assetsByHash, ); + // Cloudflare holds the requested asset bytes now. Keep only the + // deploy-local cache buster, which a Durable Object snapshot stores, so + // asset bytes are not held through the Worker upload. + const deployLocalHashes = new Set( + Object.entries(assetManifest) + .filter(([path]) => path.startsWith('/.void-deploy/')) + .map(([, entry]) => entry.hash), + ); + for (const hash of assetsByHash.keys()) { + if (!deployLocalHashes.has(hash)) { + assetsByHash.delete(hash); + } + } await assertDeploymentCanProceed(context.get('db'), deployment.id, projectId); logger.info('asset_upload', 'upload_complete', { hasCompletionJwt: completionJwt !== undefined, @@ -267,7 +281,6 @@ export async function uploadDeploymentWorker(input: { input.currentStep = 'worker_upload'; markPhase('worker_upload'); - const workerFormData = new FormData(); emit({ event: 'worker_upload', modules: workerFiles.length }); // Reserved-name set lives in `../reserved-bindings.ts` so the secrets @@ -403,26 +416,30 @@ export async function uploadDeploymentWorker(input: { assetsByHash, ); } - workerFormData.append( - 'metadata', - new Blob([JSON.stringify(metadata)], { type: 'application/json' }), - ); - for (const file of workerFiles) { - workerFormData.append(file.name, file.content, file.name); - } input.workerUploadUrl = `https://api.cloudflare.com/client/v4/accounts/${context.env.CF_ACCOUNT_ID}/workers/dispatch/namespaces/${dispatchNamespace}/scripts/${workerName}`; await retryWithBackoff( async (attempt) => { const attemptStart = Date.now(); - const cfResponse = await fetch(input.workerUploadUrl!, { - method: 'PUT', - headers: { - Authorization: `Bearer ${context.env.CF_API_TOKEN}`, - 'User-Agent': platformAppUploadUserAgent(context.env), - }, - body: workerFormData, - }); + // A streamed body is sent once, so each attempt reopens the modules. + const upload = workerUploadBody(metadata, workerFiles); + let cfResponse: Response; + try { + cfResponse = await fetch(input.workerUploadUrl!, { + method: 'PUT', + headers: { + Authorization: `Bearer ${context.env.CF_API_TOKEN}`, + 'User-Agent': platformAppUploadUserAgent(context.env), + 'Content-Type': upload.contentType, + }, + body: withKnownLength(upload.stream, upload.length), + }); + } catch (error) { + // A module that fails its size or digest check stops the body before + // its closing boundary. Report that, not the transport error it causes; + // thrown errors are not retried. + throw upload.failure() ?? error; + } if (!cfResponse.ok) { const error = await cfResponse.text(); diff --git a/platform/packages/api/src/deployment-execution.ts b/platform/packages/api/src/deployment-execution.ts index ea716e4f9..98993e785 100644 --- a/platform/packages/api/src/deployment-execution.ts +++ b/platform/packages/api/src/deployment-execution.ts @@ -3,11 +3,13 @@ import { paginatedR2Delete } from './assets.ts'; import { activeOAuthApproval } from './auth/oauth.ts'; import { R2_CONCURRENCY } from './concurrency.ts'; import { createDb, type Db, type DbOrTx } from './db.ts'; +import { WORKER_UPLOAD_MAX_AGE_MS, type WorkerUploadModule } from './deploy/shared.ts'; import type { DeploymentInput, DeploymentOperation } from './deployment-operation-types.ts'; import type { ApiEnv } from './env.ts'; import { deploymentOperations, projectDeployCredentials, projects, users } from './schema.ts'; import { isAdminSuspended } from './suspension.ts'; import type { DeployEvent, DeployManifest } from './types.ts'; +import { listR2Objects, r2WorkerModule, type WorkerModule } from './worker-upload.ts'; export type { DeploymentInput, DeploymentOperation, @@ -96,26 +98,79 @@ async function sha256Hex(bytes: BufferSource): Promise { ).join(''); } -export async function prepareDeploymentInput( +/** R2 prefix holding one deployment attempt's input. */ +export function deploymentInputPrefix(projectId: string, id: string): string { + return `deployment-inputs/${projectId}/${id}/`; +} + +/** Stream one input module, checked against its size and recorded digest. */ +export function deploymentInputModule( r2: R2Bucket, - projectId: string, + module: DeploymentInput['modules'][number], + size: number, +): WorkerModule { + return r2WorkerModule( + r2, + { name: module.name, key: module.key, type: module.type, size, sha256: module.digest }, + { + missing: 'Durable deployment input is missing', + changed: 'Durable deployment input digest mismatch', + }, + ); +} + +type DeploymentInputOptions = { + appType: DeployManifest['type']; + credentialId?: string; + oauthApproval?: string; +}; + +function deploymentInputRecord( body: Record, - files: Array<{ name: string; content: Blob }>, - options: { - appType: DeployManifest['type']; - credentialId?: string; - oauthApproval?: string; - }, -) { - const { appType, credentialId, oauthApproval } = options; - const prefix = `deployment-inputs/${projectId}/${crypto.randomUUID()}/`; + { credentialId, oauthApproval }: DeploymentInputOptions, +): DeploymentInput { const input: DeploymentInput = { body: { ...body }, modules: [], ...(credentialId ? { credentialId } : {}), ...(oauthApproval ? { oauthApproval } : {}), }; + // Module bytes live in R2; the executor reads them from `modules`. delete input.body.workerFiles; + delete input.body.workerInput; + return input; +} + +async function deploymentInputDigest(input: DeploymentInput): Promise { + const identity = JSON.stringify({ + body: input.body, + modules: input.modules.map(({ name, digest, type }) => ({ name, digest, type })), + }); + return sha256Hex(new TextEncoder().encode(identity)); +} + +async function writeDeploymentInputReceipt( + r2: R2Bucket, + prefix: string, + appType: DeployManifest['type'], + moduleCount: number, +): Promise { + // Attribute the existing staging write without extra requests. This marks + // prepared input, including no-op redeploys and later failures, not a live + // publication. Keep the parent prefix unchanged for cleanup and recovery. + const kind = appType === 'spa' ? 'spa' : moduleCount > 0 ? 'worker' : 'static'; + await r2.put(`${prefix}receipt-void-platform-${kind}`, JSON.stringify({ createdAt: Date.now() })); +} + +export async function prepareDeploymentInput( + r2: R2Bucket, + projectId: string, + body: Record, + files: Array<{ name: string; content: Blob }>, + options: DeploymentInputOptions, +) { + const prefix = deploymentInputPrefix(projectId, crypto.randomUUID()); + const input = deploymentInputRecord(body, options); try { // Hash one module at a time so at most one extra copy of module bytes is // alive beside the decoded Blobs; a near-cap request already sits close @@ -124,7 +179,13 @@ export async function prepareDeploymentInput( for (let index = 0; index < files.length; index++) { const { name, content } = files[index]!; const digest = await sha256Hex(await content.arrayBuffer()); - modules.push({ name, key: `${prefix}${index}`, digest, type: content.type }); + modules.push({ + name, + key: `${prefix}${index}`, + digest, + type: content.type, + size: content.size, + }); } // The request has not answered yet, so upload the Blobs themselves in // parallel. Workers let at most six R2 calls wait for a response at once @@ -154,19 +215,8 @@ export async function prepareDeploymentInput( throw rejected.reason; } input.modules = modules; - const identity = JSON.stringify({ - body: input.body, - modules: input.modules.map(({ name, digest, type }) => ({ name, digest, type })), - }); - const digest = await sha256Hex(new TextEncoder().encode(identity)); - // Attribute the existing staging write without extra requests. This marks - // prepared input, including no-op redeploys and later failures, not a live - // publication. Keep the parent prefix unchanged for cleanup and recovery. - const kind = appType === 'spa' ? 'spa' : files.length > 0 ? 'worker' : 'static'; - await r2.put( - `${prefix}receipt-void-platform-${kind}`, - JSON.stringify({ createdAt: Date.now() }), - ); + const digest = await deploymentInputDigest(input); + await writeDeploymentInputReceipt(r2, prefix, options.appType, files.length); return { input, prefix, digest }; } catch (error) { await paginatedR2Delete(r2, prefix).catch(() => {}); @@ -174,6 +224,67 @@ export async function prepareDeploymentInput( } } +/** + * Explain why a direct Worker upload under `prefix` cannot be deployed, or + * return null. One paginated listing covers every module: object `i` must + * exist with module `i`'s declared size and be recent enough that the + * stale-input sweep cannot remove it while the deploy runs. Bytes are checked + * against their digests when they stream to Cloudflare. + */ +export async function uploadedWorkerInputProblem( + r2: R2Bucket, + prefix: string, + modules: ReadonlyArray, +): Promise { + const stored = await listR2Objects(r2, prefix); + const oldest = Date.now() - WORKER_UPLOAD_MAX_AGE_MS; + for (const [index, module] of modules.entries()) { + const object = stored.get(`${prefix}${index}`); + const name = JSON.stringify(module.name); + if (!object) { + return `Worker module ${name} was not uploaded. Run the deploy again.`; + } + if (object.size !== module.size) { + return `Worker module ${name} has ${object.size} bytes in storage instead of ${module.size}. Run the deploy again.`; + } + if (object.uploaded.getTime() < oldest) { + return `Worker module ${name} was uploaded more than 12 hours ago. Run the deploy again.`; + } + } + return null; +} + +/** + * The durable input for a Worker upload the CLI already put under `prefix`, + * and its digest. Nothing is written, so a retry can match the operation that + * admitted this input after its cleanup deleted the prefix. + */ +export async function uploadedDeploymentInput( + prefix: string, + body: Record, + modules: DeploymentInput['modules'], + options: DeploymentInputOptions, +) { + const input = deploymentInputRecord(body, options); + input.modules = modules; + return { input, prefix, digest: await deploymentInputDigest(input) }; +} + +/** + * Record a Worker upload as durable input. The bytes stay where they were + * uploaded; only the receipt marker is written. A request never deletes this + * prefix itself: a retry of the same deploy reuses it, so the admitted + * operation's input cleanup or the stale-input sweep removes it. + */ +export async function recordUploadedDeploymentInput( + r2: R2Bucket, + uploaded: Awaited>, + appType: DeployManifest['type'], +) { + await writeDeploymentInputReceipt(r2, uploaded.prefix, appType, uploaded.input.modules.length); + return uploaded; +} + export async function assertDeploymentExecutionAuthority( db: DbOrTx, operation: DeploymentOperation, @@ -386,9 +497,17 @@ export async function dispatchPendingDeployments(env: ApiEnv['Bindings'], db: Db for (const operation of completed) { await deleteDeploymentInputs(env, db, operation); } - // Stage-before-admission avoids holding a project transaction across R2 - // uploads. A killed request can leave partial input; walk one bounded page - // per tick and delete only old prefixes with no acknowledged receipt. + await sweepStaleDeploymentInputs(env, db); +} + +/** + * Stage-before-admission avoids holding a project transaction across R2 + * uploads, and CLIs upload Worker modules before they finalize a deploy. A + * killed request or an abandoned upload can leave input behind; walk one + * bounded page per tick and delete only old prefixes with no acknowledged + * receipt. + */ +export async function sweepStaleDeploymentInputs(env: ApiEnv['Bindings'], db: Db): Promise { const sweepKey = 'deployment-input-sweep:v1'; const cursor = await env.ROUTING.get(sweepKey); const page = await env.STATIC_ASSETS_R2.list({ diff --git a/platform/packages/api/src/deployment-operation-types.ts b/platform/packages/api/src/deployment-operation-types.ts index 03fcc194f..01ee128d2 100644 --- a/platform/packages/api/src/deployment-operation-types.ts +++ b/platform/packages/api/src/deployment-operation-types.ts @@ -2,7 +2,8 @@ import type { DeployEvent } from './types.ts'; export type DeploymentInput = { body: Record; - modules: Array<{ name: string; key: string; digest: string; type: string }>; + /** `size` is absent on inputs staged before it was recorded. */ + modules: Array<{ name: string; key: string; digest: string; type: string; size?: number }>; credentialId?: string; oauthApproval?: string; }; diff --git a/platform/packages/api/src/env.ts b/platform/packages/api/src/env.ts index 47dac8d94..8e90259cf 100644 --- a/platform/packages/api/src/env.ts +++ b/platform/packages/api/src/env.ts @@ -231,7 +231,7 @@ export type ApiEnv = Env & { /** Only set by the Queue executor, never by request authentication. */ deploymentExecution?: { operation: import('./deployment-operation-types.ts').DeploymentOperation; - workerFiles: Array<{ name: string; content: Blob }>; + workerFiles: Array; }; }; }; diff --git a/platform/packages/api/src/index.ts b/platform/packages/api/src/index.ts index 84e8cdcfe..59e50a3d2 100644 --- a/platform/packages/api/src/index.ts +++ b/platform/packages/api/src/index.ts @@ -52,7 +52,11 @@ import { getUsersByIds, type Db, } from './db.ts'; -import { deploymentExecutionStatus, dispatchPendingDeployments } from './deployment-execution.ts'; +import { + deploymentExecutionStatus, + dispatchPendingDeployments, + sweepStaleDeploymentInputs, +} from './deployment-execution.ts'; import { pruneEmailActionHistory } from './email/admission.ts'; import { emailDomainRoutes, emailOperations } from './email/connection-schema.ts'; import { @@ -671,6 +675,14 @@ async function scheduled( } const db = deploymentDb ?? createDb(env.HYPERDRIVE.connectionString); const capabilities = getPlatformCapabilities(env); + // Without the deployment Queue the outbox above does not run, and deploys + // run inside their request. Worker uploads that never became a deployment + // are still removed here. + if (!deploymentDb) { + await sweepStaleDeploymentInputs(env, db).catch((error) => + console.error('[scheduled] deployment input sweep failed', error), + ); + } await reconcileProjectCleanupJobs(db, env).catch((error) => console.error('[scheduled] project cleanup retry failed:', error), diff --git a/platform/packages/api/src/middleware/auth.ts b/platform/packages/api/src/middleware/auth.ts index 7113b1166..cfc56788a 100644 --- a/platform/packages/api/src/middleware/auth.ts +++ b/platform/packages/api/src/middleware/auth.ts @@ -31,6 +31,7 @@ import { isAdminSuspended } from '../suspension.ts'; * - `GET /projects/:id/secrets` listSecrets (env.ts schema validation) * - `POST /projects/:id/deploy/preflight` asset preflight * - `POST /projects/:id/deploy/upload-urls` presigned R2 PUT URLs + * - `POST /projects/:id/deploy/worker-upload-urls` presigned Worker module PUTs * - `POST /projects/:id/deploy` manifest finalize * * `:id` is matched against `payload.project` by the caller — the entries @@ -49,6 +50,7 @@ const DEPLOY_TOKEN_ALLOWLIST: ReadonlyArray = [ { method: 'GET', segments: ['projects', null, 'secrets'] }, { method: 'POST', segments: ['projects', null, 'deploy', 'preflight'] }, { method: 'POST', segments: ['projects', null, 'deploy', 'upload-urls'] }, + { method: 'POST', segments: ['projects', null, 'deploy', 'worker-upload-urls'] }, { method: 'POST', segments: ['projects', null, 'deploy'] }, ]; diff --git a/platform/packages/api/src/provisioning/durable-objects.ts b/platform/packages/api/src/provisioning/durable-objects.ts index 38f7b9e62..1ef1c545e 100644 --- a/platform/packages/api/src/provisioning/durable-objects.ts +++ b/platform/packages/api/src/provisioning/durable-objects.ts @@ -14,12 +14,20 @@ import { generateId } from '../id.ts'; import { encryptPlatformCredential, decryptPlatformCredential } from '../project-secrets.ts'; import { projects, resources, deployments } from '../schema.ts'; import type { DeployManifest, Resource } from '../types.ts'; +import { + blobWorkerModule, + listR2Objects, + r2WorkerModule, + streamSha256, + withKnownLength, + workerUploadBody, + type WorkerModule, +} from '../worker-upload.ts'; import { CF_API_BASE, cfFetch, CloudflareApiError } from './cloudflare-api.ts'; -import { putDurableSnapshotObject } from './durable-snapshot.ts'; +import { putDurableSnapshotModule, putDurableSnapshotObject } from './durable-snapshot.ts'; type Env = ApiEnv['Bindings']; type Binding = Record & { type: string; name: string }; -type Module = { name: string; content: Blob }; type Snapshot = { metadata: Record & { bindings: Array }; files: Array<{ name: string; contentType: string; sha256: string }>; @@ -132,23 +140,27 @@ async function upload( env: Env, worker: string, metadata: Record, - modules: Array, + modules: ReadonlyArray, create = false, ): Promise { - const form = new FormData(); - form.append('metadata', new Blob([JSON.stringify(metadata)], { type: 'application/json' })); - for (const module of modules) { - form.append(module.name, module.content, module.name); + // A release replays the application's modules; stream them like the + // application upload so the whole bundle is never held in memory. + const body = workerUploadBody(metadata, modules); + let response: Response; + try { + response = await fetch(url(env, worker), { + method: 'PUT', + headers: { + Authorization: `Bearer ${env.CF_API_TOKEN}`, + 'Content-Type': body.contentType, + ...(create ? { 'If-None-Match': '*' } : {}), + }, + body: withKnownLength(body.stream, body.length), + signal: AbortSignal.timeout(60_000), + }); + } catch (error) { + throw body.failure() ?? error; } - const response = await fetch(url(env, worker), { - method: 'PUT', - headers: { - Authorization: `Bearer ${env.CF_API_TOKEN}`, - ...(create ? { 'If-None-Match': '*' } : {}), - }, - body: form, - signal: AbortSignal.timeout(60_000), - }); if (create && (response.status === 409 || response.status === 412)) { await response.body?.cancel(); return; @@ -167,7 +179,7 @@ function idle( projectId: string, className: string, create = false, -): { metadata: Record; modules: Array } { +): { metadata: Record; modules: Array } { return { metadata: { main_module: ENTRY, @@ -187,15 +199,15 @@ function idle( : {}), }, modules: [ - { - name: ENTRY, - content: new Blob( + blobWorkerModule( + ENTRY, + new Blob( [ `import {DurableObject} from 'cloudflare:workers';\nexport class ${className} extends DurableObject {fetch(){return new Response('Durable Object is inactive',{status:503});}}\nexport default {fetch(){return new Response(null,{status:404});}};`, ], { type: 'application/javascript+module' }, ), - }, + ), ], }; } @@ -317,6 +329,10 @@ function snapshotKey(projectId: string, deploymentId: string): string { return `worker-bundles/${projectId}/${deploymentId}/durable/snapshot.json`; } +function snapshotModulePrefix(projectId: string, deploymentId: string): string { + return `worker-bundles/${projectId}/${deploymentId}/durable/modules/`; +} + /** Store the exact code and encrypted bindings before any namespace code can change. */ export async function storeDurableObjectSnapshot( env: Env, @@ -324,7 +340,7 @@ export async function storeDurableObjectSnapshot( projectId: string, deploymentId: string, metadata: Snapshot['metadata'], - modules: Array, + modules: ReadonlyArray, assetsByHash: ReadonlyMap = new Map(), ): Promise { if (modules.some((module) => module.name === ENTRY)) { @@ -348,14 +364,15 @@ export async function storeDurableObjectSnapshot( ); } for (const module of modules) { - const bytes = await module.content.arrayBuffer(); - const file = { name: module.name, contentType: module.content.type, sha256: await hash(bytes) }; - files.push(file); - await putDurableSnapshotObject( + // Stored input modules carry their verified digest; inline modules are + // hashed from memory. The bytes then stream into the release copy. + const sha256 = module.sha256 ?? (await streamSha256(await module.open())); + files.push({ name: module.name, contentType: module.type, sha256 }); + await putDurableSnapshotModule( env.STATIC_ASSETS_R2, - `worker-bundles/${projectId}/${deploymentId}/durable/modules/${module.name}`, - bytes, - file.sha256, + `${snapshotModulePrefix(projectId, deploymentId)}${module.name}`, + module, + sha256, ); } const credential = await encryptPlatformCredential( @@ -379,7 +396,7 @@ async function loadSnapshot( env: Env, projectId: string, deploymentId: string, -): Promise<{ snapshot: Snapshot; modules: Array }> { +): Promise<{ snapshot: Snapshot; modules: Array }> { const object = await env.STATIC_ASSETS_R2.get(snapshotKey(projectId, deploymentId)); if (!object) { throw new Error('Custom Durable Object release snapshot is missing.'); @@ -388,20 +405,32 @@ async function loadSnapshot( const snapshot = JSON.parse( await decryptPlatformCredential(env, projectId, `durable:${deploymentId}`, credential), ) as Snapshot; - const modules: Array = []; - for (const file of snapshot.files) { - const module = await env.STATIC_ASSETS_R2.get( - `worker-bundles/${projectId}/${deploymentId}/durable/modules/${file.name}`, - ); + // One listing gives every module's size, which the streamed upload + // declares before it starts. Bytes are checked against the recorded + // digest as they stream. + const prefix = snapshotModulePrefix(projectId, deploymentId); + const stored = await listR2Objects(env.STATIC_ASSETS_R2, prefix); + const errors = { + missing: 'Custom Durable Object release module is missing.', + changed: 'Custom Durable Object release module changed.', + }; + const modules = snapshot.files.map((file) => { + const module = stored.get(`${prefix}${file.name}`); if (!module) { - throw new Error('Custom Durable Object release module is missing.'); - } - const bytes = await module.arrayBuffer(); - if ((await hash(bytes)) !== file.sha256) { - throw new Error('Custom Durable Object release module changed.'); + throw new Error(errors.missing); } - modules.push({ name: file.name, content: new Blob([bytes], { type: file.contentType }) }); - } + return r2WorkerModule( + env.STATIC_ASSETS_R2, + { + name: file.name, + key: `${prefix}${file.name}`, + type: file.contentType, + size: module.size, + sha256: file.sha256, + }, + errors, + ); + }); return { snapshot, modules }; } @@ -529,15 +558,15 @@ async function restoreObject( }, [ ...modules, - { - name: ENTRY, - content: new Blob( + blobWorkerModule( + ENTRY, + new Blob( [ `export {${resource.binding_name}} from ${JSON.stringify('./' + main)};\nexport default {fetch(){return new Response(null,{status:404});}};`, ], { type: 'application/javascript+module' }, ), - }, + ), ], ); } diff --git a/platform/packages/api/src/provisioning/durable-snapshot.ts b/platform/packages/api/src/provisioning/durable-snapshot.ts index a14ddcad5..d08f24878 100644 --- a/platform/packages/api/src/provisioning/durable-snapshot.ts +++ b/platform/packages/api/src/provisioning/durable-snapshot.ts @@ -1,3 +1,11 @@ +import { + streamSha256, + trackedModuleStream, + withKnownLength, + WorkerModuleIntegrityError, + type WorkerModule, +} from '../worker-upload.ts'; + export async function snapshotDigest(value: string | ArrayBuffer | Uint8Array): Promise { const bytes = new Uint8Array( typeof value === 'string' @@ -11,6 +19,11 @@ export async function snapshotDigest(value: string | ArrayBuffer | Uint8Array): ).join(''); } +type SnapshotBody = Readonly<{ + value: string | ArrayBuffer | Uint8Array | ReadableStream; + failure?: () => unknown; +}>; + /** Immutable release writes can be recovered after a lost R2 acknowledgement. */ export async function putDurableSnapshotObject( bucket: Pick, @@ -18,22 +31,52 @@ export async function putDurableSnapshotObject( value: string | ArrayBuffer | Uint8Array, digest?: string, ): Promise { - const expected = digest ?? (await snapshotDigest(value)); + await putVerifiedSnapshotObject(bucket, key, digest ?? (await snapshotDigest(value)), () => ({ + value, + })); +} + +/** Stream one Worker module into an immutable release copy. */ +export async function putDurableSnapshotModule( + bucket: Pick, + key: string, + module: WorkerModule, + digest: string, +): Promise { + await putVerifiedSnapshotObject(bucket, key, digest, () => { + // A stream is written once, so every attempt reopens the module. + const source = trackedModuleStream(module); + return { value: withKnownLength(source.stream, module.size), failure: source.failure }; + }); +} + +async function putVerifiedSnapshotObject( + bucket: Pick, + key: string, + expected: string, + body: () => SnapshotBody, +): Promise { let failure: unknown; for (let attempt = 0; attempt < 3; attempt++) { + const { value, failure: sourceFailure } = body(); try { const written = await bucket.put(key, value, { onlyIf: { etagDoesNotMatch: '*' } }); if (written) { return; } } catch (error) { + // Source bytes that fail their integrity check fail every retry too. + const integrity = sourceFailure?.(); + if (integrity instanceof WorkerModuleIntegrityError) { + throw integrity; + } failure = error; } // R2 reads are strongly consistent. Verify an acknowledged or ambiguous // write before accepting it; the condition also fences any late PUT. const stored = await bucket.get(key); if (stored) { - if ((await snapshotDigest(await stored.arrayBuffer())) === expected) { + if ((await streamSha256(stored.body)) === expected) { return; } throw new Error(`Custom Durable Object release snapshot changed: ${key}`); diff --git a/platform/packages/api/src/r2-presign.ts b/platform/packages/api/src/r2-presign.ts index 7b96c6f14..40c48106e 100644 --- a/platform/packages/api/src/r2-presign.ts +++ b/platform/packages/api/src/r2-presign.ts @@ -1,11 +1,12 @@ /** - * Presigned PUT URL generation for Cloudflare R2 with mandatory Content-MD5 - * upload-integrity verification. + * Presigned PUT URL generation for Cloudflare R2 with Content-MD5 + * upload-integrity verification for assets. * * Used by the direct-to-R2 asset upload flow (see * meta/design-docs/0068-direct-r2-asset-upload.md). The platform issues a * URL the CLI uses to PUT a single asset directly to R2, bypassing the API - * Worker's request body for large dist/ payloads. + * Worker's request body for large dist/ payloads. Worker module uploads use + * the same URLs without MD5 and with a signed Content-Length. * * Integrity: the URL signs `Content-Md5` as part of the canonical request, * so any tampering of the header value invalidates the signature. R2's S3 @@ -29,8 +30,12 @@ export type PresignAssetPutOptions = Readonly<{ bucket: string; /** Object key inside the bucket. May contain `/`; passed through encodeURI. */ key: string; - /** 32 lowercase hex characters — raw MD5 digest of the asset bytes. */ - md5: string; + /** + * 32 lowercase hex characters — raw MD5 digest of the asset bytes. Worker + * module uploads omit it: they sign `contentLength`, and the API checks + * their SHA-256 when it streams them to Cloudflare. + */ + md5?: string; contentType: string; /** Defaults to 600 (10 minutes). */ ttlSeconds?: number; @@ -93,12 +98,13 @@ function formatAmzDate(d: Date): string { /** * Generate a presigned PUT URL for an R2 object that enforces a Content-MD5 - * checksum at write time. + * checksum (when `md5` is given) and a Content-Length (when `contentLength` + * is given) at write time. * * The returned `headers` map is the verbatim set the client must send with - * the PUT. `Content-Md5` and `Content-Type` are included in - * `X-Amz-SignedHeaders` so they are covered by the signature, which means - * R2 rejects any PUT whose body MD5 doesn't match the declared header. + * the PUT. Every header in it is included in `X-Amz-SignedHeaders` so it is + * covered by the signature, which means R2 rejects any PUT whose body MD5 or + * length doesn't match the declared header. */ export async function presignAssetPut( opts: PresignAssetPutOptions, @@ -109,18 +115,19 @@ export async function presignAssetPut( throw new Error(`ttlSeconds must be a positive number, got ${ttl}`); } - const checksumB64 = hexToBase64(opts.md5); const url = r2Endpoint(opts.accountId, opts.bucket, opts.key); // The AwsV4Signer in `signQuery` mode produces a query-presigned URL. // `allHeaders: true` flips Content-Type and any custom headers from // the unsignable list into the signed-headers list, which is what we - // want — both Content-Type and Content-Md5 must be covered by the - // signature so R2 rejects tampered uploads. + // want — Content-Type, Content-Md5 and Content-Length must be covered by + // the signature so R2 rejects tampered uploads. const headers: Record = { 'Content-Type': opts.contentType, - 'Content-Md5': checksumB64, }; + if (opts.md5 !== undefined) { + headers['Content-Md5'] = hexToBase64(opts.md5); + } if (opts.ifNoneMatch) { headers['If-None-Match'] = opts.ifNoneMatch; } diff --git a/platform/packages/api/src/routes/deploy.ts b/platform/packages/api/src/routes/deploy.ts index cbe3ba1cd..dfd90c155 100644 --- a/platform/packages/api/src/routes/deploy.ts +++ b/platform/packages/api/src/routes/deploy.ts @@ -1,5 +1,5 @@ import { sql as drizzleSql } from 'drizzle-orm'; -import { Hono } from 'hono'; +import { Hono, type Context } from 'hono'; import { findMissingR2Assets, persistedR2AssetSizes } from '../asset-inventory.ts'; import { activeOAuthApproval, OAuthError } from '../auth/oauth.ts'; import { uploadPool } from '../concurrency.ts'; @@ -9,21 +9,27 @@ import { requireActiveDeployAccess } from '../deploy/recovery.ts'; import { HEX_32_REGEX, MAX_UPLOAD_URLS_BATCH, + moduleContentType, + workerUploadModulesProblem, type UploadUrlsRequestAsset, + type WorkerUploadModule, } from '../deploy/shared.ts'; import { appendDeploymentEvent, assertDeploymentExecutionAuthority, deploymentEventPage, + deploymentInputModule, + deploymentInputPrefix, getDeploymentOperation, type DeploymentOperation, } from '../deployment-execution.ts'; import type { ApiEnv } from '../env.ts'; import { checkCliVersionOrRespond } from '../middleware/cli-version.ts'; -import { createDeployLogger } from '../observability/logger.ts'; +import { createDeployLogger, type DeployLogger } from '../observability/logger.ts'; import { ProjectAccessDeniedError, projectAccessErrorBody } from '../project-access.ts'; import { ProjectRequiresNormalizationError, requireNormalizedProject } from '../project-state.ts'; import { presignAssetPut } from '../r2-presign.ts'; +import { listR2Objects, type WorkerModule } from '../worker-upload.ts'; import { requireProjectRouteCapability } from './project-access.ts'; export { @@ -34,13 +40,20 @@ export { const app = new Hono(); -// Body cap for `POST /:id/deploy/upload-urls`. The body is dominated by -// `assets[]` entries — each entry is ~150 bytes (blake3 + md5 + size + -// contentType). 5000 entries × ~150 bytes ≈ 750 KB, so 4 MiB carries a -// 4× safety margin. Asset BYTES never travel this endpoint — they go -// direct-to-R2 via the presigned PUTs this endpoint issues. +// Body cap for `POST /:id/deploy/upload-urls` and `/worker-upload-urls`. The +// body is dominated by `assets[]` entries — each entry is ~150 bytes (blake3 + +// md5 + size + contentType) — or `modules[]` entries of similar size (name + +// size + sha256). 5000 entries × ~150 bytes ≈ 750 KB, so 4 MiB carries a +// 4× safety margin. Asset and module BYTES never travel these endpoints — +// they go direct-to-R2 via the presigned PUTs the endpoints issue. const MAX_UPLOAD_URLS_BODY_BYTES = 4 * 1024 * 1024; +// 30 minutes. On a slow uplink a tail asset can wait in the upload pool for +// more than 10 minutes before its first PUT, and the CLI treats R2's 403 +// SignatureExpired as terminal. Longer deploys need the re-presign-on-expiry +// follow-up noted on `/deploy/upload-urls`. +const UPLOAD_URL_TTL_SECONDS = 1800; + // Bounded concurrency for the presign batch. Presigning is CPU-light; // 50 in-flight is generous and keeps tail latency bounded without // fanning out to 5000 simultaneous async tasks. @@ -104,7 +117,206 @@ const UPLOAD_URLS_PROJECT_DELETING = 'upload_urls:project_deleting'; * blake3s. Tracking issue: see meta/design-docs/0068 follow-up. */ app.post('/:id/deploy/upload-urls', async (context) => { - const projectId = context.req.param('id'); + const request = await beginUploadUrlsRequest(context); + if (request instanceof Response) { + return request; + } + const { projectId, logger, startedAt, body } = request; + // TODO(orphan-gc): authenticated users can upload bytes via upload-urls + // without ever finalizing a deploy, leaving durable orphans. Teardown + // cleans these up; for active projects, abandoned uploads accumulate + // until a future deploy's retention cleanup happens to walk over them. + // Track in a follow-up: add a periodic per-project R2 sweep that diffs + // `assets/${projectId}/*` against the union of retained manifests. + if (!Array.isArray(body.assets)) { + return context.json({ error: 'request body must include an `assets` array' }, 400); + } + if (body.assets.length > MAX_UPLOAD_URLS_BATCH) { + return context.json( + { + error: `assets[] has ${body.assets.length} entries, exceeds the ${MAX_UPLOAD_URLS_BATCH} per-request limit`, + }, + 400, + ); + } + for (let i = 0; i < body.assets.length; i++) { + const err = validateUploadUrlsAsset(body.assets[i], i); + if (err) { + return context.json({ error: err }, 400); + } + } + const assets = body.assets as ReadonlyArray; + + logger.info('upload_urls', 'request', { assetCount: assets.length }); + + const urls: Record }> = {}; + const failed = await presignUnderProjectLock( + context, + projectId, + logger, + { step: 'upload_urls', assetCount: assets.length }, + async (tx, signingKeyCache) => { + if (context.get('oauthApproval')) { + const { approval } = await activeOAuthApproval( + tx, + context.get('oauthApproval')!, + context.get('companyIdentity'), + true, + ); + const approved = Object.values(approval.data.artifact!.assetManifest); + if ( + approval.data.projectId !== projectId || + assets.some( + (asset) => + !approved.some( + (entry) => + entry.blake3 === asset.blake3 && + entry.md5 === asset.md5 && + entry.size === asset.size, + ), + ) + ) { + throw new OAuthError( + 'access_denied', + 'Upload only the assets approved for this publish.', + 403, + ); + } + } + await uploadPool(assets, PRESIGN_BATCH_CONCURRENCY, async (asset) => { + const result = await presignAssetPut( + { + accountId: context.env.CF_ACCOUNT_ID, + accessKeyId: context.env.R2_ACCESS_KEY_ID, + secretAccessKey: context.env.R2_SECRET_ACCESS_KEY, + bucket: context.env.R2_STATIC_ASSETS_BUCKET, + key: `assets/${projectId}/${asset.blake3}`, + md5: asset.md5, + contentType: asset.contentType, + ttlSeconds: context.get('oauthApproval') + ? Math.max( + 1, + Math.min( + 900, + context.get('deployTokenExpiresAt')! - Math.floor(Date.now() / 1000), + ), + ) + : UPLOAD_URL_TTL_SECONDS, + ...(context.get('oauthApproval') + ? { ifNoneMatch: '*' as const, contentLength: asset.size } + : {}), + }, + signingKeyCache, + ); + // Last-write-wins on duplicate blake3 — see docstring above. Map + // assignment is synchronous so concurrent pool tasks do not race + // beyond a benign last-writer. + urls[asset.blake3] = result; + }); + }, + ); + if (failed) { + return failed; + } + + logger.info('upload_urls', 'response', { + urlsIssued: Object.keys(urls).length, + durationMs: Date.now() - startedAt, + }); + + return context.json({ urls }); +}); + +/** + * Issue presigned R2 PUT URLs for a deploy's Worker modules. + * + * Worker modules, source maps included, can exceed what a deploy request + * body carries, so the CLI uploads them first and finalizes with + * `workerInput: { id, modules }`. Module `i` is stored at + * `deployment-inputs/${projectId}/${id}/${i}` — the durable deployment input + * prefix family, so operation input cleanup, the stale-input sweep, and + * project teardown all cover it. Each URL signs Content-Type and + * Content-Length, so R2 stores exactly the declared size; the SHA-256 is + * checked when the bytes stream to Cloudflare. URLs expire in 30 minutes. + * + * Connected-app credentials cannot reach this route (see `authMiddleware`): + * they send modules inline, where their approval digest covers them. + * + * Returns `{ id, urls: [{ url, headers }] }`, with `urls` in `modules` order. + */ +app.post('/:id/deploy/worker-upload-urls', async (context) => { + const request = await beginUploadUrlsRequest(context); + if (request instanceof Response) { + return request; + } + const { projectId, logger, startedAt, body } = request; + const problem = workerUploadModulesProblem(body.modules, 'modules'); + if (problem) { + return context.json( + problem.error + ? { error: problem.error, message: problem.message } + : { error: problem.message }, + problem.status, + ); + } + const modules = body.modules as ReadonlyArray; + const id = crypto.randomUUID(); + const prefix = deploymentInputPrefix(projectId, id); + + logger.info('worker_upload_urls', 'request', { moduleCount: modules.length }); + + const urls: Array<{ url: string; headers: Record }> = []; + const failed = await presignUnderProjectLock( + context, + projectId, + logger, + { step: 'worker_upload_urls', moduleCount: modules.length }, + async (_tx, signingKeyCache) => { + await uploadPool([...modules.keys()], PRESIGN_BATCH_CONCURRENCY, async (index) => { + const module = modules[index]!; + urls[index] = await presignAssetPut( + { + accountId: context.env.CF_ACCOUNT_ID, + accessKeyId: context.env.R2_ACCESS_KEY_ID, + secretAccessKey: context.env.R2_SECRET_ACCESS_KEY, + bucket: context.env.R2_STATIC_ASSETS_BUCKET, + key: `${prefix}${index}`, + contentType: moduleContentType(module.name), + contentLength: module.size, + ttlSeconds: UPLOAD_URL_TTL_SECONDS, + }, + signingKeyCache, + ); + }); + }, + ); + if (failed) { + return failed; + } + + logger.info('worker_upload_urls', 'response', { + urlsIssued: urls.length, + durationMs: Date.now() - startedAt, + }); + + return context.json({ id, urls }); +}); + +/** + * The shared start of the direct-upload URL routes: deploy capability, the + * deleting/version/normalization gates, and a size-capped JSON object body. + * Returns the error response when a gate rejects the request. + */ +async function beginUploadUrlsRequest(context: Context): Promise< + | Response + | Readonly<{ + projectId: string; + logger: DeployLogger; + startedAt: number; + body: Record; + }> +> { + const projectId = context.req.param('id') as string; const access = await requireProjectRouteCapability(context, projectId, 'deploy'); if (access instanceof Response) { return access; @@ -143,12 +355,6 @@ app.post('/:id/deploy/upload-urls', async (context) => { } throw error; } - // TODO(orphan-gc): authenticated users can upload bytes via upload-urls - // without ever finalizing a deploy, leaving durable orphans. Teardown - // cleans these up; for active projects, abandoned uploads accumulate - // until a future deploy's retention cleanup happens to walk over them. - // Track in a follow-up: add a periodic per-project R2 sweep that diffs - // `assets/${projectId}/*` against the union of retained manifests. const startedAt = Date.now(); const logger = createDeployLogger({ @@ -205,28 +411,22 @@ app.post('/:id/deploy/upload-urls', async (context) => { return context.json({ error: 'request body must be a JSON object' }, 400); } const body = parsed as Record; - if (!Array.isArray(body.assets)) { - return context.json({ error: 'request body must include an `assets` array' }, 400); - } - if (body.assets.length > MAX_UPLOAD_URLS_BATCH) { - return context.json( - { - error: `assets[] has ${body.assets.length} entries, exceeds the ${MAX_UPLOAD_URLS_BATCH} per-request limit`, - }, - 400, - ); - } - for (let i = 0; i < body.assets.length; i++) { - const err = validateUploadUrlsAsset(body.assets[i], i); - if (err) { - return context.json({ error: err }, 400); - } - } - const assets = body.assets as ReadonlyArray; - - logger.info('upload_urls', 'request', { assetCount: assets.length }); + return { projectId, logger, startedAt, body }; +} - const urls: Record }> = {}; +/** + * Run `presign` under the project's share lock and map the failures both + * direct-upload URL routes report. Returns an error response, or null once + * every URL is issued. + */ +async function presignUnderProjectLock( + context: Context, + projectId: string, + logger: DeployLogger, + log: Readonly<{ step: string } & Record>, + presign: (tx: Db, signingKeyCache: Map) => Promise, +): Promise { + const { step, ...logFields } = log; try { // Serialize against project teardown. `lockAndMarkDeleting` takes // FOR UPDATE on the project row in `provisioning/cleanup.ts`; we take @@ -235,8 +435,8 @@ app.post('/:id/deploy/upload-urls', async (context) => { // queueing as Postgres allows) but blocks teardown until this // transaction commits. Once the lock is acquired we re-read the // project row inside the transaction; if status flipped to `deleting` - // between the unlocked early-out above and the lock acquisition, - // throw the sentinel to surface a 410 to the CLI. + // between the unlocked early-out and the lock acquisition, throw the + // sentinel to surface a 410 to the CLI. // // Lock window: presigning is pure HMAC (no R2 / DB / network I/O // beyond the initial SELECT FOR SHARE). A 5000-entry batch fans out @@ -253,67 +453,7 @@ app.post('/:id/deploy/upload-urls', async (context) => { const signingKeyCache = new Map(); await requireActiveDeployAccess(tx, projectId, context.get('user').id); await requireNormalizedProject(tx, lockedProject); - if (context.get('oauthApproval')) { - const { approval } = await activeOAuthApproval( - tx, - context.get('oauthApproval')!, - context.get('companyIdentity'), - true, - ); - const approved = Object.values(approval.data.artifact!.assetManifest); - if ( - approval.data.projectId !== projectId || - assets.some( - (asset) => - !approved.some( - (entry) => - entry.blake3 === asset.blake3 && - entry.md5 === asset.md5 && - entry.size === asset.size, - ), - ) - ) { - throw new OAuthError( - 'access_denied', - 'Upload only the assets approved for this publish.', - 403, - ); - } - } - await uploadPool(assets, PRESIGN_BATCH_CONCURRENCY, async (asset) => { - const result = await presignAssetPut( - { - accountId: context.env.CF_ACCOUNT_ID, - accessKeyId: context.env.R2_ACCESS_KEY_ID, - secretAccessKey: context.env.R2_SECRET_ACCESS_KEY, - bucket: context.env.R2_STATIC_ASSETS_BUCKET, - key: `assets/${projectId}/${asset.blake3}`, - md5: asset.md5, - contentType: asset.contentType, - // 30 minutes. On a slow uplink a tail asset can wait in the upload - // pool for more than 10 minutes before its first PUT, and the CLI - // treats R2's 403 SignatureExpired as terminal. Longer deploys need - // the re-presign-on-expiry follow-up noted in the docstring above. - ttlSeconds: context.get('oauthApproval') - ? Math.max( - 1, - Math.min( - 900, - context.get('deployTokenExpiresAt')! - Math.floor(Date.now() / 1000), - ), - ) - : 1800, - ...(context.get('oauthApproval') - ? { ifNoneMatch: '*' as const, contentLength: asset.size } - : {}), - }, - signingKeyCache, - ); - // Last-write-wins on duplicate blake3 — see docstring above. Map - // assignment is synchronous so concurrent pool tasks do not race - // beyond a benign last-writer. - urls[asset.blake3] = result; - }); + await presign(tx, signingKeyCache); }); } catch (err) { if (err instanceof OAuthError) { @@ -328,7 +468,7 @@ app.post('/:id/deploy/upload-urls', async (context) => { if (err instanceof Error && err.message === UPLOAD_URLS_PROJECT_DELETING) { return context.json({ error: 'project is being deleted' }, 410); } - logger.error('upload_urls', 'presign_error', err, { assetCount: assets.length }); + logger.error(step, 'presign_error', err, logFields); return context.json( { error: `failed to presign upload URLs: ${err instanceof Error ? err.message : String(err)}`, @@ -336,14 +476,8 @@ app.post('/:id/deploy/upload-urls', async (context) => { 500, ); } - - logger.info('upload_urls', 'response', { - urlsIssued: Object.keys(urls).length, - durationMs: Date.now() - startedAt, - }); - - return context.json({ urls }); -}); + return null; +} // Deploy to a project app.post('/:id/deploy', handleManifestOnlyDeploy); @@ -396,21 +530,23 @@ export async function executeQueuedDeployment( }); return; } - const workerFiles: Array<{ name: string; content: Blob }> = []; + // One listing confirms every staged module and its size before execution + // touches provider state, and supplies sizes for inputs staged before they + // were recorded. Bytes stream from R2 during the Cloudflare upload and are + // checked against their digests there. + const workerFiles: Array = []; + const staged = operation.payload.modules.length + ? await listR2Objects(env.STATIC_ASSETS_R2, operation.input_prefix) + : new Map(); for (const module of operation.payload.modules) { - const object = await env.STATIC_ASSETS_R2.get(module.key); + const object = staged.get(module.key); if (!object) { throw new Error('Durable deployment input is missing'); } - const bytes = await object.arrayBuffer(); - const digest = Array.from( - new Uint8Array(await crypto.subtle.digest('SHA-256', bytes)), - (byte) => byte.toString(16).padStart(2, '0'), - ).join(''); - if (digest !== module.digest) { + if (module.size !== undefined && module.size !== object.size) { throw new Error('Durable deployment input digest mismatch'); } - workerFiles.push({ name: module.name, content: new Blob([bytes], { type: module.type }) }); + workerFiles.push(deploymentInputModule(env.STATIC_ASSETS_R2, module, object.size)); } const executor = new Hono() .onError((error) => { diff --git a/platform/packages/api/src/worker-upload.ts b/platform/packages/api/src/worker-upload.ts new file mode 100644 index 000000000..7326118ba --- /dev/null +++ b/platform/packages/api/src/worker-upload.ts @@ -0,0 +1,255 @@ +/** + * Streaming access to user Worker modules. + * + * A Worker can carry 64 MiB of code plus source maps of any size, while one + * API isolate has 128 MiB for every request it serves. Module bytes are + * therefore never held whole: a module opens only when its turn comes and its + * bytes pass through in chunks. + */ +import { createHash } from 'node:crypto'; + +/** One Worker module, opened only when its bytes are needed. */ +export type WorkerModule = Readonly<{ + name: string; + /** Multipart Content-Type Cloudflare uses to classify the module. */ + type: string; + size: number; + /** Lowercase hex SHA-256, when known before the bytes are read. */ + sha256?: string; + open(): Promise>; +}>; + +/** Stored module bytes are missing or differ from their recorded size and digest. */ +export class WorkerModuleIntegrityError extends Error {} + +/** A stream that remembers the first failure of the sources it reads. */ +export type TrackedStream = Readonly<{ + stream: ReadableStream; + failure(): unknown; +}>; + +export function blobWorkerModule(name: string, content: Blob): WorkerModule { + return { + name, + type: content.type, + size: content.size, + open: async () => content.stream() as ReadableStream, + }; +} + +/** A module stored in R2, checked against its size and digest as it streams. */ +export function r2WorkerModule( + bucket: Pick, + module: Readonly<{ name: string; key: string; type: string; size: number; sha256: string }>, + errors: Readonly<{ missing: string; changed: string }>, +): WorkerModule { + return { + name: module.name, + type: module.type, + size: module.size, + sha256: module.sha256, + async open() { + const object = await bucket.get(module.key); + if (!object) { + throw new WorkerModuleIntegrityError(errors.missing); + } + return verifiedStream(object.body, module.size, module.sha256, errors.changed); + }, + }; +} + +/** + * Pass `body` through while counting and hashing it. The stream errors as + * soon as it runs past `size`, and before it ends when the size or digest + * differs, so a reader never sees a complete copy of changed bytes. + */ +function verifiedStream( + body: ReadableStream, + size: number, + sha256: string, + message: string, +): ReadableStream { + // node:crypto hashes incrementally in both workerd (nodejs_compat) and the + // Node integration tests; crypto.DigestStream exists only in workerd. + const hash = createHash('sha256'); + let received = 0; + return body.pipeThrough( + new TransformStream({ + transform(chunk, controller) { + received += chunk.byteLength; + if (received > size) { + throw new WorkerModuleIntegrityError(message); + } + hash.update(chunk); + controller.enqueue(chunk); + }, + flush() { + if (received !== size || hash.digest('hex') !== sha256) { + throw new WorkerModuleIntegrityError(message); + } + }, + }), + ); +} + +/** Lowercase hex SHA-256 of a stream, read in chunks. */ +export async function streamSha256(stream: ReadableStream): Promise { + const hash = createHash('sha256'); + const reader = stream.getReader(); + for (;;) { + const { done, value } = await reader.read(); + if (done) { + return hash.digest('hex'); + } + hash.update(value); + } +} + +/** + * Read byte pieces and lazily opened sources one after another. Pull-based: + * a source opens only after the previous one ends, and one chunk is in + * flight at a time. + */ +function sequence( + pieces: ReadonlyArray Promise>)>, +): TrackedStream { + let failure: unknown; + let canceled = false; + let index = 0; + let reader: ReadableStreamDefaultReader | undefined; + return { + failure: () => failure, + stream: new ReadableStream({ + async pull(controller) { + try { + for (;;) { + if (reader) { + const { done, value } = await reader.read(); + if (!done) { + controller.enqueue(value); + return; + } + reader = undefined; + } else if (index === pieces.length) { + controller.close(); + return; + } else { + const piece = pieces[index++]!; + if (piece instanceof Uint8Array) { + controller.enqueue(piece); + return; + } + reader = (await piece()).getReader(); + } + } + } catch (error) { + // A canceled reader's own errors are not source failures. + if (!canceled) { + failure ??= error; + } + throw error; + } + }, + async cancel(reason) { + canceled = true; + await reader?.cancel(reason).catch(() => {}); + }, + }), + }; +} + +/** Open one module for a single write, remembering a source failure. */ +export function trackedModuleStream(module: WorkerModule): TrackedStream { + return sequence([() => module.open()]); +} + +/** HTML multipart/form-data encoding of a part name or filename. */ +function quoted(value: string): string { + return value.replaceAll('"', '%22').replaceAll('\r', '%0D').replaceAll('\n', '%0A'); +} + +/** + * The Workers script-upload body as a stream of known length: a `metadata` + * part, then one part per module, with the part headers FormData writes (a + * Blob without a filename is named "blob"). Each module opens only when the + * body reaches it. A module failure stops the body before the closing + * boundary, so the receiver never gets a complete form. + */ +export function workerUploadBody( + metadata: unknown, + modules: ReadonlyArray, +): TrackedStream & Readonly<{ contentType: string; length: number }> { + const encoder = new TextEncoder(); + const boundary = `----VoidWorkerUpload${crypto.randomUUID().replaceAll('-', '')}`; + const metadataBytes = encoder.encode(JSON.stringify(metadata)); + const parts = [ + { + name: 'metadata', + filename: 'blob', + type: 'application/json', + size: metadataBytes.byteLength, + open: async () => new Blob([metadataBytes]).stream() as ReadableStream, + }, + ...modules.map((module) => ({ + name: module.name, + filename: module.name, + type: module.type, + size: module.size, + open: () => module.open(), + })), + ]; + const pieces: Array Promise>)> = []; + let length = 0; + for (const part of parts) { + const header = encoder.encode( + `--${boundary}\r\nContent-Disposition: form-data; name="${quoted(part.name)}"; filename="${quoted(part.filename)}"\r\nContent-Type: ${part.type}\r\n\r\n`, + ); + pieces.push(header, part.open, new Uint8Array([13, 10])); + length += header.byteLength + part.size + 2; + } + const closing = encoder.encode(`--${boundary}--\r\n`); + pieces.push(closing); + length += closing.byteLength; + return { + ...sequence(pieces), + contentType: `multipart/form-data; boundary=${boundary}`, + length, + }; +} + +/** + * Give a stream the exact length it will produce. Workers send a stream + * without one chunked, and R2 refuses to store one. A FixedLengthStream sends + * Content-Length and fails if the stream ends short or runs long. Node, where + * the integration tests run, has no FixedLengthStream and takes the stream as is. + */ +export function withKnownLength( + stream: ReadableStream, + length: number, +): ReadableStream { + if (typeof FixedLengthStream !== 'function') { + return stream; + } + const fixed = new FixedLengthStream(length); + stream.pipeTo(fixed.writable).catch(() => { + // The readable side reports the failed pipe to whoever reads it. + }); + return fixed.readable as ReadableStream; +} + +/** Every object under `prefix`, keyed by its full key, in paginated listings. */ +export async function listR2Objects( + bucket: Pick, + prefix: string, +): Promise>> { + const objects = new Map>(); + let cursor: string | undefined; + do { + const page = await bucket.list({ prefix, ...(cursor ? { cursor } : {}) }); + for (const object of page.objects) { + objects.set(object.key, { size: object.size, uploaded: object.uploaded }); + } + cursor = page.truncated ? page.cursor : undefined; + } while (cursor !== undefined); + return objects; +} diff --git a/platform/packages/api/test/integration/deploy-direct-r2.test.ts b/platform/packages/api/test/integration/deploy-direct-r2.test.ts index c5d3f7a1d..324a3f0ff 100644 --- a/platform/packages/api/test/integration/deploy-direct-r2.test.ts +++ b/platform/packages/api/test/integration/deploy-direct-r2.test.ts @@ -1855,3 +1855,383 @@ describe('CF asset-upload-session body never carries v2 manifest extras', () => } }); }); + +describe('direct Worker module upload', () => { + const MiB = 1024 * 1024; + let user: Awaited>; + let originalDispatch: DispatchNamespace; + + beforeEach(async () => { + user = await createTestUser(env.db); + originalDispatch = env.DISPATCH; + env.DISPATCH = { + get: () => ({ fetch: async () => Response.json({ ready: true }) }), + } as unknown as DispatchNamespace; + }); + afterEach(() => { + env.DISPATCH = originalDispatch; + }); + + async function sha256(bytes: Uint8Array): Promise { + return Array.from(new Uint8Array(await crypto.subtle.digest('SHA-256', bytes)), (byte) => + byte.toString(16).padStart(2, '0'), + ).join(''); + } + + function requestUrls(projectId: string, body: unknown) { + return app.fetch( + new Request(`http://localhost/projects/${projectId}/deploy/worker-upload-urls`, { + method: 'POST', + headers: { Authorization: `Bearer ${user.jwt}`, 'Content-Type': 'application/json' }, + body: JSON.stringify(body), + }), + env, + ); + } + + /** What the CLI does: request URLs, then PUT each module to its URL's key. */ + async function upload(projectId: string, files: Record>) { + const modules = await Promise.all( + Object.entries(files).map(async ([name, bytes]) => ({ + name, + size: bytes.byteLength, + sha256: await sha256(bytes), + })), + ); + const res = await requestUrls(projectId, { modules }); + expect(res.status).toBe(200); + const { id, urls } = (await res.json()) as { + id: string; + urls: Array<{ url: string; headers: Record }>; + }; + for (const [index, module] of modules.entries()) { + const key = decodeURI(new URL(urls[index]!.url).pathname).slice( + `/${env.R2_STATIC_ASSETS_BUCKET}/`.length, + ); + await env.STATIC_ASSETS_R2.put(key, files[module.name]!); + } + return { id, modules }; + } + + function finalize(projectId: string, extra: Record) { + return app.fetch( + new Request(`http://localhost/projects/${projectId}/deploy`, { + method: 'POST', + headers: { Authorization: `Bearer ${user.jwt}`, 'Content-Type': 'application/json' }, + body: JSON.stringify({ + manifest: { version: 3, assetManifestVersion: 2, bindings: {}, type: 'void' }, + assetManifest: {}, + source: null, + commit: null, + preflight: null, + ...extra, + }), + }), + env, + ); + } + + function workerUploadCalls(cf: ReturnType, slug: string) { + return cf.mock.calls.filter( + ([url, init]) => + String(url).includes(`/scripts/user-${slug}-`) && + (init as RequestInit | undefined)?.method === 'PUT', + ); + } + + it('signs one size-bound PUT per module under the deployment input prefix', async () => { + const project = await createTestProject(env.db, user.id, 'wu-sign'); + const modules = [ + { name: 'index.js', size: 10, sha256: 'a'.repeat(64) }, + { name: 'index.js.map', size: 20, sha256: 'b'.repeat(64) }, + { name: 'lib/add.wasm', size: 0, sha256: 'c'.repeat(64) }, + ]; + const res = await requestUrls(project.id, { modules }); + expect(res.status).toBe(200); + const { id, urls } = (await res.json()) as { + id: string; + urls: Array<{ url: string; headers: Record }>; + }; + expect(id).toMatch(/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/); + expect(urls).toHaveLength(3); + const types = ['application/javascript+module', 'application/source-map', 'application/wasm']; + for (const [index, { url, headers }] of urls.entries()) { + const parsed = new URL(url); + expect(parsed.pathname).toBe( + `/${env.R2_STATIC_ASSETS_BUCKET}/deployment-inputs/${project.id}/${id}/${index}`, + ); + expect(headers).toEqual({ + 'Content-Type': types[index], + 'Content-Length': String(modules[index]!.size), + }); + expect(parsed.searchParams.get('X-Amz-SignedHeaders')).toContain('content-length'); + expect(parsed.searchParams.get('X-Amz-SignedHeaders')).toContain('content-type'); + expect(parsed.searchParams.get('X-Amz-Expires')).toBe('1800'); + } + }); + + it.each([ + ['no modules field', {}], + ['an empty list', { modules: [] }], + [ + 'a path outside the bundle', + { modules: [{ name: '../a.js', size: 1, sha256: 'a'.repeat(64) }] }, + ], + ['an unsupported extension', { modules: [{ name: 'a.css', size: 1, sha256: 'a'.repeat(64) }] }], + [ + 'a repeated name', + { + modules: [ + { name: 'a.js', size: 1, sha256: 'a'.repeat(64) }, + { name: 'a.js', size: 2, sha256: 'b'.repeat(64) }, + ], + }, + ], + ['a negative size', { modules: [{ name: 'a.js', size: -1, sha256: 'a'.repeat(64) }] }], + ['a fractional size', { modules: [{ name: 'a.js', size: 1.5, sha256: 'a'.repeat(64) }] }], + ['an uppercase digest', { modules: [{ name: 'a.js', size: 1, sha256: 'A'.repeat(64) }] }], + [ + 'more than 5000 modules', + { + modules: Array.from({ length: 5001 }, (_, index) => ({ + name: `m${index}.js`, + size: 1, + sha256: 'a'.repeat(64), + })), + }, + ], + ])('rejects a module list with %s', async (_case, body) => { + const project = await createTestProject(env.db, user.id, 'wu-invalid'); + const res = await requestUrls(project.id, body); + expect(res.status).toBe(400); + expect(await res.json()).toMatchObject({ error: expect.any(String) }); + }); + + it('applies the 64 MiB Worker limit to code but not to source maps', async () => { + const project = await createTestProject(env.db, user.id, 'wu-limit'); + const tooLarge = await requestUrls(project.id, { + modules: [ + { name: 'index.js', size: 40 * MiB, sha256: 'a'.repeat(64) }, + { name: 'chunk.wasm', size: 24 * MiB + 1, sha256: 'b'.repeat(64) }, + { name: 'index.js.map', size: 1, sha256: 'c'.repeat(64) }, + ], + }); + expect(tooLarge.status).toBe(413); + expect(await tooLarge.json()).toEqual({ + error: 'worker_too_large', + message: 'The Worker is 64.1 MiB without source maps, and Cloudflare allows 64 MiB.', + }); + const withMaps = await requestUrls(project.id, { + modules: [ + { name: 'index.js', size: 64 * MiB, sha256: 'a'.repeat(64) }, + { name: 'index.js.map', size: 500 * MiB, sha256: 'b'.repeat(64) }, + ], + }); + expect(withMaps.status).toBe(200); + }); + + it('streams uploaded modules byte-for-byte to Cloudflare and deletes the upload afterwards', async () => { + const project = await createTestProject(env.db, user.id, 'wu-deploy'); + const files = { + 'index.js': new TextEncoder().encode( + "import './lib/util.js'; export default { fetch() { return new Response('ok'); } }", + ), + 'lib/util.js': new TextEncoder().encode('export const util = 1;'), + 'index.js.map': new TextEncoder().encode('{"version":3,"sources":[],"mappings":""}'), + 'add.wasm': Uint8Array.from({ length: 256 }, (_, index) => index), + }; + const { id, modules } = await upload(project.id, files); + const cf = mockCfApi(); + const res = await finalize(project.id, { workerInput: { id, modules } }); + expect(res.status).toBe(200); + const events = await readAllEvents(res); + expect(events.find((e) => isEvent(e, 'done'))).toMatchObject({ workers: 4 }); + + const calls = workerUploadCalls(cf, project.slug); + expect(calls).toHaveLength(1); + const form = (calls[0]![1] as RequestInit).body as FormData; + const metadata = JSON.parse(await (form.get('metadata') as Blob).text()) as { + main_module: string; + }; + expect(metadata.main_module).toBe('index.js'); + const types: Record = { + 'index.js': 'application/javascript+module', + 'lib/util.js': 'application/javascript+module', + 'index.js.map': 'application/source-map', + 'add.wasm': 'application/wasm', + }; + for (const [name, bytes] of Object.entries(files)) { + const file = form.get(name) as File; + expect(file.type).toBe(types[name]); + expect(new Uint8Array(await file.arrayBuffer())).toEqual(bytes); + } + const [deployment] = await env.db + .select({ status: deployments.status }) + .from(deployments) + .where(eq(deployments.project_id, project.id)); + expect(deployment?.status).toBe('live'); + const left = await env.STATIC_ASSETS_R2.list({ + prefix: `deployment-inputs/${project.id}/${id}/`, + }); + expect(left.objects).toEqual([]); + }); + + it('rejects an upload that is missing, resized, stale, oversized, from another project, or sent twice', async () => { + const project = await createTestProject(env.db, user.id, 'wu-reject'); + const other = await createTestProject(env.db, user.id, 'wu-other'); + const code = new TextEncoder().encode('export default {};'); + const { id, modules } = await upload(project.id, { 'index.js': code }); + const foreign = await upload(other.id, { 'index.js': code }); + const cf = mockCfApi(); + const reject = async (extra: Record, status: number, message: string) => { + const res = await finalize(project.id, extra); + expect(res.status).toBe(status); + expect(await res.json()).toMatchObject({ message: expect.stringContaining(message) }); + }; + + await reject( + { + workerInput: { + id, + modules: [...modules, { name: 'more.js', size: 3, sha256: 'a'.repeat(64) }], + }, + }, + 400, + '"more.js" was not uploaded', + ); + await reject( + { workerInput: { id, modules: [{ ...modules[0]!, size: code.byteLength + 1 }] } }, + 400, + `has ${code.byteLength} bytes in storage instead of ${code.byteLength + 1}`, + ); + // Keys derive from the authenticated project, so another project's id + // names nothing here, and its upload stays untouched. + await reject({ workerInput: foreign }, 400, '"index.js" was not uploaded'); + expect( + await env.STATIC_ASSETS_R2.head(`deployment-inputs/${other.id}/${foreign.id}/0`), + ).not.toBeNull(); + await reject( + { workerInput: { id, modules }, workerFiles: { 'index.js': btoa('export default {};') } }, + 400, + 'not both', + ); + await reject( + { workerInput: { id: 'not-an-id', modules } }, + 400, + 'workerInput.id must be the id returned', + ); + await reject( + { + workerInput: { + id, + modules: [{ ...modules[0]!, size: 64 * MiB + 1 }], + }, + }, + 413, + 'Cloudflare allows 64 MiB', + ); + ( + env.STATIC_ASSETS_R2 as unknown as { setUploadedAt(key: string, uploaded: Date): void } + ).setUploadedAt( + `deployment-inputs/${project.id}/${id}/0`, + new Date(Date.now() - 13 * 60 * 60 * 1000), + ); + await reject({ workerInput: { id, modules } }, 400, 'more than 12 hours ago'); + expect(workerUploadCalls(cf, project.slug)).toEqual([]); + expect(await env.db.select().from(deployments)).toEqual([]); + }); + + it('fails the deploy without a complete Cloudflare upload when a stored module changes', async () => { + const project = await createTestProject(env.db, user.id, 'wu-corrupt'); + const code = new TextEncoder().encode('export default {};'); + const { id, modules } = await upload(project.id, { 'index.js': code }); + const uploaded = vi.fn(async () => cfSuccess({})); + mockCfApi({ uploadWorker: uploaded }); + // Same size, so the finalize listing accepts it; only the digest differs. + await env.STATIC_ASSETS_R2.put( + `deployment-inputs/${project.id}/${id}/0`, + new TextEncoder().encode('export default [];'), + ); + const res = await finalize(project.id, { workerInput: { id, modules } }); + const events = await readAllEvents(res); + expect(events).toContainEqual( + expect.objectContaining({ + event: 'error', + message: 'Durable deployment input digest mismatch', + step: 'worker_upload', + }), + ); + expect(uploaded).not.toHaveBeenCalled(); + const [deployment] = await env.db + .select({ status: deployments.status }) + .from(deployments) + .where(eq(deployments.project_id, project.id)); + expect(deployment?.status).toBe('failed'); + // A failed deploy is terminal too, so its upload is gone. + const left = await env.STATIC_ASSETS_R2.list({ + prefix: `deployment-inputs/${project.id}/${id}/`, + }); + expect(left.objects).toEqual([]); + }); + + it('reopens every module when Cloudflare asks for the upload again', async () => { + const project = await createTestProject(env.db, user.id, 'wu-retry'); + const files = { + 'index.js': new TextEncoder().encode('export default {};'), + 'index.js.map': new TextEncoder().encode('{"version":3}'), + }; + const { id, modules } = await upload(project.id, files); + const received: Array> = []; + mockCfApi({ + uploadWorker: async (_url, init) => { + const form = init!.body as FormData; + received.push({ + 'index.js': await (form.get('index.js') as Blob).text(), + 'index.js.map': await (form.get('index.js.map') as Blob).text(), + }); + return received.length === 1 + ? new Response('Assets have already been uploaded', { status: 200 }) + : cfSuccess({}); + }, + }); + const events = await readAllEvents( + await finalize(project.id, { workerInput: { id, modules } }), + ); + expect(events.find((e) => isEvent(e, 'done'))).toBeDefined(); + const expected = { + 'index.js': 'export default {};', + 'index.js.map': '{"version":3}', + }; + expect(received).toEqual([expected, expected]); + }); + + it('points an oversized inline Worker body at a CLI upgrade', async () => { + const project = await createTestProject(env.db, user.id, 'wu-inline-cap'); + const chunk = new Uint8Array(MiB).fill(0x61); + let sent = 0; + const body = new ReadableStream({ + pull(controller) { + if (sent > 64) { + controller.close(); + return; + } + sent++; + controller.enqueue(chunk); + }, + }); + const res = await app.fetch( + new Request(`http://localhost/projects/${project.id}/deploy`, { + method: 'POST', + headers: { Authorization: `Bearer ${user.jwt}`, 'Content-Type': 'application/json' }, + body, + duplex: 'half', + } as RequestInit), + env, + ); + expect(res.status).toBe(413); + const { error } = (await res.json()) as { error: string }; + expect(error).toContain('source maps'); + expect(error).toContain('Upgrade the void CLI'); + expect(error).not.toContain('asset bytes'); + }); +}); diff --git a/platform/packages/api/test/integration/deploy-token-scope.test.ts b/platform/packages/api/test/integration/deploy-token-scope.test.ts index 0cb87088a..302a21347 100644 --- a/platform/packages/api/test/integration/deploy-token-scope.test.ts +++ b/platform/packages/api/test/integration/deploy-token-scope.test.ts @@ -99,6 +99,26 @@ describe('deploy token scope enforcement', () => { expect(res.status).not.toBe(403); }); + it('lets a deploy token request Worker module upload URLs', async () => { + const user = await createTestUser(env.db); + const project = await createTestProject(env.db, user.id, 'deploy-worker-upload'); + const token = await deployToken(project.id, user.id); + mockCfApi(); + + const res = await app.fetch( + new Request(`http://localhost/projects/${project.id}/deploy/worker-upload-urls`, { + method: 'POST', + headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json' }, + body: JSON.stringify({ + modules: [{ name: 'index.js', size: 18, sha256: 'a'.repeat(64) }], + }), + }), + env, + ); + expect(res.status).toBe(200); + expect(await res.json()).toMatchObject({ id: expect.any(String), urls: [expect.any(Object)] }); + }); + it('lets the deploy sibling endpoints through (preflight, upload-urls)', async () => { const user = await createTestUser(env.db); const project = await createTestProject(env.db, user.id, 'deploy-siblings'); @@ -495,6 +515,7 @@ describe('deploy token scope enforcement', () => { { name: 'POST /projects/:id/database', path: 'database' }, { name: 'POST /projects/:id/deploy/preflight', path: 'deploy/preflight' }, { name: 'POST /projects/:id/deploy/upload-urls', path: 'deploy/upload-urls' }, + { name: 'POST /projects/:id/deploy/worker-upload-urls', path: 'deploy/worker-upload-urls' }, ]; for (const status of ['cancelled', 'failed'] as const) { diff --git a/platform/packages/api/test/integration/deployment-execution.test.ts b/platform/packages/api/test/integration/deployment-execution.test.ts index e9704e514..dc03110dc 100644 --- a/platform/packages/api/test/integration/deployment-execution.test.ts +++ b/platform/packages/api/test/integration/deployment-execution.test.ts @@ -1140,3 +1140,219 @@ describe('durable execution claim and revocation', () => { }); }); }); + +describe('durable execution of directly uploaded Worker modules', () => { + beforeEach(() => { + queuedEnv.DISPATCH = { + get: () => ({ fetch: async () => Response.json({ ready: true }) }), + } as unknown as DispatchNamespace; + }); + + /** What the CLI does: request upload URLs, then PUT each module to its key. */ + async function uploadWorker(files: Record) { + const modules = await Promise.all( + Object.entries(files).map(async ([name, text]) => { + const bytes = new TextEncoder().encode(text); + const digest = new Uint8Array(await crypto.subtle.digest('SHA-256', bytes)); + return { + name, + size: bytes.byteLength, + sha256: Array.from(digest, (byte) => byte.toString(16).padStart(2, '0')).join(''), + }; + }), + ); + const response = await app.fetch( + new Request(`https://api.test/projects/${project.id}/deploy/worker-upload-urls`, { + method: 'POST', + headers: { Authorization: `Bearer ${user.jwt}`, 'content-type': 'application/json' }, + body: JSON.stringify({ modules }), + }), + queuedEnv, + ); + expect(response.status).toBe(200); + const { id } = (await response.json()) as { id: string }; + const prefix = `deployment-inputs/${project.id}/${id}/`; + for (const [index, text] of Object.values(files).entries()) { + await env.STATIC_ASSETS_R2.put(`${prefix}${index}`, text); + } + return { workerInput: { id, modules }, prefix }; + } + + function workerPayload(workerInput: unknown) { + return { + ...payload(), + manifest: { version: 3, assetManifestVersion: 2, bindings: {}, type: 'void' }, + workerInput, + }; + } + + async function keys(prefix: string) { + return (await env.STATIC_ASSETS_R2.list({ prefix })).objects.map(({ key }) => key).sort(); + } + + it('records the upload in place as durable input and executes it after disconnect', async () => { + const worker = 'export default { fetch() { return new Response("uploaded"); } }'; + const uploadedModules: Array = []; + mockCfApi({ + uploadWorker: async (_url, init) => { + const module = (init!.body as FormData).get('index.js') as File; + uploadedModules.push(await module.text()); + return cfSuccess({}); + }, + }); + const { workerInput, prefix } = await uploadWorker({ 'index.js': worker }); + const response = await post(crypto.randomUUID(), workerPayload(workerInput)); + expect(response.status).toBe(200); + const id = response.headers.get('x-void-deployment-id')!; + const operation = await getDeploymentOperation(env.db, id); + expect(operation?.input_prefix).toBe(prefix); + expect(operation?.payload.body).not.toHaveProperty('workerInput'); + expect(operation?.payload.modules).toEqual([ + { + name: 'index.js', + key: `${prefix}0`, + digest: workerInput.modules[0]!.sha256, + type: 'application/javascript+module', + size: workerInput.modules[0]!.size, + }, + ]); + // Nothing was copied: the prefix holds the upload and the receipt marker. + expect(await keys(prefix)).toEqual([`${prefix}0`, `${prefix}receipt-void-platform-worker`]); + + const message = await run(id); + expect(message.ack).toHaveBeenCalledOnce(); + expect(await env.db.select().from(deployments)).toMatchObject([{ id, status: 'live' }]); + expect(uploadedModules).toEqual([worker]); + expect(await keys(prefix)).toEqual([]); + }); + + it('keeps a retried upload for the operation that owns it', async () => { + const { workerInput, prefix } = await uploadWorker({ 'index.js': 'export default {};' }); + const key = crypto.randomUUID(); + const first = await post(key, workerPayload(workerInput)); + const id = first.headers.get('x-void-deployment-id')!; + // A transport retry repeats the request: the same key and the same upload. + const retry = await post(key, workerPayload(workerInput)); + expect(retry.status).toBe(200); + expect(retry.headers.get('x-void-deployment-id')).toBe(id); + expect(await keys(prefix)).toContain(`${prefix}0`); + + // A different deploy cannot claim an upload another operation owns. + const other = await app.fetch( + new Request(`https://api.test/projects/${project.id}/deploy`, { + method: 'POST', + headers: { + Authorization: `Bearer ${user.jwt}`, + 'content-type': 'application/json', + 'x-void-deploy-idempotency-key': crypto.randomUUID(), + }, + body: JSON.stringify(workerPayload(workerInput)), + }), + queuedEnv, + ); + expect(other.status).toBe(409); + expect(await other.json()).toEqual({ + error: 'this Worker upload belongs to another deployment; run the deploy again', + }); + expect(await keys(prefix)).toContain(`${prefix}0`); + expect(await env.db.select().from(deploymentOperations)).toHaveLength(1); + + expect((await run(id)).ack).toHaveBeenCalledOnce(); + expect(await env.db.select().from(deployments)).toMatchObject([{ id, status: 'live' }]); + expect(await keys(prefix)).toEqual([]); + }); + + it('returns the receipt for a retry after the operation deleted its upload', async () => { + const { workerInput, prefix } = await uploadWorker({ 'index.js': 'export default {};' }); + const key = crypto.randomUUID(); + const first = await post(key, workerPayload(workerInput)); + const id = first.headers.get('x-void-deployment-id')!; + expect((await run(id)).ack).toHaveBeenCalledOnce(); + expect(await getDeploymentOperation(env.db, id)).toMatchObject({ + status: 'terminal', + inputs_deleted: true, + }); + expect(await keys(prefix)).toEqual([]); + + // A transport retry can arrive after execution. It gets the original + // receipt and writes nothing under the deleted upload. + const retry = await post(key, workerPayload(workerInput)); + expect(retry.status).toBe(200); + expect(retry.headers.get('x-void-deployment-id')).toBe(id); + expect(await keys(prefix)).toEqual([]); + expect(await env.db.select().from(deploymentOperations)).toHaveLength(1); + + // Different input under the same key still needs its own upload. + const changed = await post(key, { ...workerPayload(workerInput), commit: 'different-commit' }); + expect(changed.status).toBe(400); + expect(await env.db.select().from(deploymentOperations)).toHaveLength(1); + }); + + it('fails a queued Worker whose stored module changes after admission', async () => { + const uploaded = vi.fn(async () => cfSuccess({})); + mockCfApi({ uploadWorker: uploaded }); + const { workerInput, prefix } = await uploadWorker({ 'index.js': 'export default {};' }); + const response = await post(crypto.randomUUID(), workerPayload(workerInput)); + const id = response.headers.get('x-void-deployment-id')!; + await env.STATIC_ASSETS_R2.put(`${prefix}0`, 'export default [];'); + + const message = await run(id); + expect(message.ack).toHaveBeenCalledOnce(); + expect(uploaded).not.toHaveBeenCalled(); + expect(await env.db.select().from(deployments)).toMatchObject([{ id, status: 'failed' }]); + expect((await getDeploymentOperation(env.db, id))?.events.at(-1)?.event).toMatchObject({ + event: 'error', + message: 'Durable deployment input digest mismatch', + }); + }); + + it('sweeps abandoned uploads on installations without the deployment Queue', async () => { + const stale = `deployment-inputs/${project.id}/${crypto.randomUUID()}/0`; + const fresh = `deployment-inputs/${project.id}/${crypto.randomUUID()}/0`; + await env.STATIC_ASSETS_R2.put(stale, 'export default {};'); + await env.STATIC_ASSETS_R2.put(fresh, 'export default {};'); + ( + env.STATIC_ASSETS_R2 as unknown as { setUploadedAt(key: string, uploaded: Date): void } + ).setUploadedAt(stale, new Date(Date.now() - 25 * 60 * 60 * 1000)); + await app.scheduled( + { cron: FAST_CRON, scheduledTime: Date.now(), noRetry() {} } as ScheduledController, + env as unknown as ApiEnv['Bindings'], + {} as ExecutionContext, + ); + expect(await env.STATIC_ASSETS_R2.head(stale)).toBeNull(); + expect(await env.STATIC_ASSETS_R2.head(fresh)).not.toBeNull(); + }); + + it('executes inline input staged before module sizes were recorded', async () => { + const worker = 'export default { fetch() { return new Response("legacy"); } }'; + const uploadedModules: Array = []; + mockCfApi({ + uploadWorker: async (_url, init) => { + const module = (init!.body as FormData).get('index.js') as File; + uploadedModules.push(await module.text()); + return cfSuccess({}); + }, + }); + const response = await post(crypto.randomUUID(), { + ...payload(), + manifest: { version: 3, assetManifestVersion: 2, bindings: {}, type: 'void' }, + workerFiles: { 'index.js': btoa(worker) }, + }); + const id = response.headers.get('x-void-deployment-id')!; + const operation = await getDeploymentOperation(env.db, id); + expect(operation?.payload.modules[0]?.size).toBe(worker.length); + await env.db + .update(deploymentOperations) + .set({ + payload: { + ...operation!.payload, + modules: operation!.payload.modules.map(({ size: _size, ...module }) => module), + }, + }) + .where(eq(deploymentOperations.id, id)); + + expect((await run(id)).ack).toHaveBeenCalledOnce(); + expect(await env.db.select().from(deployments)).toMatchObject([{ id, status: 'live' }]); + expect(uploadedModules).toEqual([worker]); + }); +}); diff --git a/platform/packages/api/test/integration/helpers.ts b/platform/packages/api/test/integration/helpers.ts index 0c4b62514..76ea4f970 100644 --- a/platform/packages/api/test/integration/helpers.ts +++ b/platform/packages/api/test/integration/helpers.ts @@ -904,6 +904,24 @@ function createSandboxMockState(): SandboxMockState { }; } +/** + * Worker uploads stream their multipart body. Parse the bytes the platform + * actually sends and put the resulting FormData on `init.body`, so handlers + * and later `mock.calls` assertions read the parts as they would read a + * FormData body. A body stream that errors (a module failing its integrity + * check) rejects here, as `fetch` would; the upload never reaches a handler. + */ +async function readStreamedMultipartBody(init: RequestInit): Promise { + const contentType = new Headers(init.headers).get('content-type') ?? ''; + if (!contentType.startsWith('multipart/form-data')) { + return; + } + const form = await new Response(init.body, { + headers: { 'content-type': contentType }, + }).formData(); + init.body = form; +} + /** * Best-effort multipart-body parser tailored to the platform's CF Workers * Upload API usage. Returns the parsed `metadata` JSON and the raw bindings @@ -1617,6 +1635,9 @@ export function mockCfApi(overrides: Partial = {}): MockCfApiHandle const url = typeof input === 'string' ? input : input instanceof URL ? input.toString() : input.url; const method = init?.method ?? 'GET'; + if (init?.body instanceof ReadableStream) { + await readStreamedMultipartBody(init); + } // Record every call for ordering assertions. Non-string bodies are // summarized as `[FormData]` or `[binary]`. diff --git a/platform/packages/api/test/integration/oauth.test.ts b/platform/packages/api/test/integration/oauth.test.ts index 20b9ca6c0..42953f605 100644 --- a/platform/packages/api/test/integration/oauth.test.ts +++ b/platform/packages/api/test/integration/oauth.test.ts @@ -642,6 +642,37 @@ describe('registered OAuth applications', () => { expect(project.active_deployment_id).toBeNull(); }, ); + it('keeps connected-app Worker modules inside the approved artifact', async () => { + const { token } = await connect(); + const project = await createTestProject(env.db, user.id, 'static-approved-app'); + // A static artifact: its approval digest commits to no Worker modules. + const prepared = createDeployArtifact({ manifest: artifact.manifest, assetManifest: {} }); + const flow = await publishApproval(token.access_token, project.slug, prepared); + const consent = await approve(flow.id, flow.cookie); + const { access_token: access } = (await ( + await exchange(new URL(consent.headers.get('Location')!).searchParams.get('code')!) + ).json()) as { access_token: string }; + const headers = { Authorization: 'Bearer ' + access, 'Content-Type': 'application/json' }; + const module = { name: 'index.js', size: 18, sha256: 'a'.repeat(64) }; + const urls = await request('/projects/' + project.id + '/deploy/worker-upload-urls', { + method: 'POST', + headers, + body: JSON.stringify({ modules: [module] }), + }); + expect(urls.status).toBe(403); + const cf = mockCfApi(); + const deploy = await request('/projects/' + project.id + '/deploy', { + method: 'POST', + headers: { ...headers, 'x-void-deploy-idempotency-key': flow.id }, + body: JSON.stringify({ + ...prepared, + workerInput: { id: crypto.randomUUID(), modules: [module] }, + }), + }); + expect(deploy.status).toBe(403); + expect(await deploy.json()).toMatchObject({ message: expect.stringContaining('workerFiles') }); + expect(cf).not.toHaveBeenCalled(); + }); it('registers and disables clients through the administrator UI with an audit trail', async () => { const admin = await createTestUser(env.db, 'administrator', 'github', 'admin', 'free', 'admin'); const input = new URLSearchParams({ diff --git a/platform/packages/api/test/unit/durable-snapshot.test.ts b/platform/packages/api/test/unit/durable-snapshot.test.ts index cc8b797b8..d76e77f17 100644 --- a/platform/packages/api/test/unit/durable-snapshot.test.ts +++ b/platform/packages/api/test/unit/durable-snapshot.test.ts @@ -1,8 +1,10 @@ import { afterEach, expect, it, vi } from 'vite-plus/test'; import { + putDurableSnapshotModule, putDurableSnapshotObject, snapshotDigest, } from '../../src/provisioning/durable-snapshot.ts'; +import { r2WorkerModule } from '../../src/worker-upload.ts'; const value = new TextEncoder().encode('immutable release'); function fixture() { let stored: Uint8Array | undefined; @@ -14,7 +16,7 @@ function fixture() { return {}; }); const get = vi.fn(async () => - stored ? { arrayBuffer: async () => new Uint8Array(stored!).buffer } : null, + stored ? { body: new Blob([new Uint8Array(stored)]).stream() } : null, ); return { bucket: { put, get } as unknown as Pick, @@ -88,6 +90,39 @@ it('bounds persistent failures after three inspected attempts', async () => { expect(f.get).toHaveBeenCalledTimes(3); }); +it('stops without retrying when a streamed module fails its integrity check', async () => { + const digest = await snapshotDigest(value); + const source = { + get: async () => ({ body: new Blob(['changed']).stream() }), + } as unknown as Pick; + const module = r2WorkerModule( + source, + { + name: 'index.js', + key: 'input/0', + type: 'application/javascript+module', + size: value.byteLength, + sha256: digest, + }, + { missing: 'module missing', changed: 'module changed' }, + ); + const put = vi.fn(async (_key: string, body: ReadableStream) => { + await new Response(body).arrayBuffer(); + return {}; + }); + const get = vi.fn(); + await expect( + putDurableSnapshotModule( + { put, get } as unknown as Pick, + 'release/index.js', + module, + digest, + ), + ).rejects.toThrow('module changed'); + expect(put).toHaveBeenCalledOnce(); + expect(get).not.toHaveBeenCalled(); +}); + it('accepts an identical existing object after its conditional write is rejected', async () => { const f = fixture(); f.set(value); diff --git a/platform/packages/api/test/unit/worker-upload.test.ts b/platform/packages/api/test/unit/worker-upload.test.ts new file mode 100644 index 000000000..2ea58bd52 --- /dev/null +++ b/platform/packages/api/test/unit/worker-upload.test.ts @@ -0,0 +1,200 @@ +import { createHash } from 'node:crypto'; +import { afterEach, describe, expect, it, vi } from 'vite-plus/test'; +import { + blobWorkerModule, + r2WorkerModule, + withKnownLength, + workerUploadBody, + WorkerModuleIntegrityError, +} from '../../src/worker-upload.ts'; + +const encoder = new TextEncoder(); + +function sha256(bytes: Uint8Array): string { + return createHash('sha256').update(bytes).digest('hex'); +} + +/** Serves fixed objects in small chunks, like an R2 body. */ +function bucket(objects: Record) { + const opened: Array = []; + return { + opened, + get: async (key: string) => { + const bytes = objects[key]; + if (!bytes) { + return null; + } + opened.push(key); + let offset = 0; + return { + body: new ReadableStream({ + pull(controller) { + if (offset >= bytes.byteLength) { + controller.close(); + return; + } + controller.enqueue(bytes.slice(offset, offset + 3)); + offset += 3; + }, + }), + }; + }, + } as unknown as Pick & { opened: Array }; +} + +const errors = { missing: 'module missing', changed: 'module changed' }; + +async function readAll(stream: ReadableStream): Promise> { + return new Uint8Array(await new Response(stream).arrayBuffer()); +} + +afterEach(() => { + vi.unstubAllGlobals(); +}); + +describe('workerUploadBody', () => { + it('declares the exact length it streams and keeps every part byte-identical', async () => { + const code = encoder.encode('export default {};'); + const map = encoder.encode('{"version":3,"mappings":""}'); + const wasm = Uint8Array.from({ length: 256 }, (_, index) => index); + const objects = { 'in/0': code, 'in/1': map }; + const r2 = bucket(objects); + const modules = [ + r2WorkerModule( + r2, + { + name: 'index.js', + key: 'in/0', + type: 'application/javascript+module', + size: code.byteLength, + sha256: sha256(code), + }, + errors, + ), + // Names may carry quotes and non-ASCII characters. + r2WorkerModule( + r2, + { + name: 'chunks/ü"x.js.map', + key: 'in/1', + type: 'application/source-map', + size: map.byteLength, + sha256: sha256(map), + }, + errors, + ), + blobWorkerModule('add.wasm', new Blob([wasm], { type: 'application/wasm' })), + ]; + const upload = workerUploadBody({ main_module: 'index.js' }, modules); + // No module opens before the body is read. + expect(r2.opened).toEqual([]); + const bytes = await readAll(upload.stream); + expect(bytes.byteLength).toBe(upload.length); + expect(upload.failure()).toBeUndefined(); + + const form = await new Response(bytes, { + headers: { 'content-type': upload.contentType }, + }).formData(); + const metadata = form.get('metadata') as File; + expect(metadata.type).toBe('application/json'); + expect(JSON.parse(await metadata.text())).toEqual({ main_module: 'index.js' }); + const parts = [...form.keys()]; + expect(parts).toEqual(['metadata', 'index.js', 'chunks/ü"x.js.map', 'add.wasm']); + for (const [name, expected, type] of [ + ['index.js', code, 'application/javascript+module'], + ['chunks/ü"x.js.map', map, 'application/source-map'], + ['add.wasm', wasm, 'application/wasm'], + ] as const) { + const file = form.get(name) as File; + expect(file.type).toBe(type); + expect(new Uint8Array(await file.arrayBuffer())).toEqual(expected); + } + }); + + it.each([ + ['same-size different bytes', encoder.encode('export default {!};')], + ['longer bytes', encoder.encode('export default {}; // more')], + ['shorter bytes', encoder.encode('export {};')], + ])('stops before the closing boundary when a stored module has %s', async (_case, stored) => { + const expected = encoder.encode('export default {};'); + const r2 = bucket({ 'in/0': stored }); + const upload = workerUploadBody({}, [ + r2WorkerModule( + r2, + { + name: 'index.js', + key: 'in/0', + type: 'application/javascript+module', + size: expected.byteLength, + sha256: sha256(expected), + }, + errors, + ), + ]); + const reader = upload.stream.getReader(); + const received: Array = []; + let failure: unknown; + try { + for (;;) { + const { done, value } = await reader.read(); + if (done) { + break; + } + received.push(value); + } + } catch (error) { + failure = error; + } + expect(failure).toBeInstanceOf(WorkerModuleIntegrityError); + expect((failure as Error).message).toBe('module changed'); + expect(upload.failure()).toBe(failure); + const decoder = new TextDecoder(); + const text = + received.map((chunk) => decoder.decode(chunk, { stream: true })).join('') + decoder.decode(); + const boundary = upload.contentType.split('boundary=')[1]!; + expect(text).not.toContain(`--${boundary}--`); + // A longer object never passes more bytes than the declared size. + expect(text).not.toContain('more'); + }); + + it('reports a module missing from storage', async () => { + const upload = workerUploadBody({}, [ + r2WorkerModule( + bucket({}), + { + name: 'index.js', + key: 'in/0', + type: 'application/javascript+module', + size: 1, + sha256: 'a'.repeat(64), + }, + errors, + ), + ]); + await expect(readAll(upload.stream)).rejects.toThrow('module missing'); + expect(upload.failure()).toBeInstanceOf(WorkerModuleIntegrityError); + }); +}); + +describe('withKnownLength', () => { + it('declares the length to a FixedLengthStream where the runtime has one', async () => { + const declared: Array = []; + vi.stubGlobal( + 'FixedLengthStream', + class { + readonly readable: ReadableStream; + readonly writable: WritableStream; + constructor(length: number) { + declared.push(length); + ({ readable: this.readable, writable: this.writable } = new TransformStream()); + } + }, + ); + const upload = workerUploadBody({}, [ + blobWorkerModule('index.js', new Blob(['export default {};'])), + ]); + const bytes = await readAll(withKnownLength(upload.stream, upload.length)); + expect(declared).toEqual([upload.length]); + expect(bytes.byteLength).toBe(upload.length); + }); +});