From 992df8b2031341a647152414d06f946523b46f83 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miguel=20=C3=81ngel?= Date: Mon, 19 Oct 2026 19:23:51 +0000 Subject: [PATCH] perf(cloud): move project bytes once through shared content storage (#5772) * perf(producer): distributed chunks download only the videos they show A project video under compiled/ that chunks extract their own frames from now goes only to the chunks that show it, like video-sources/ files. A chunk skips the readiness wait for a deferred video it does not show, so it no longer logs media_load_failed for a file it was never given. * fix(producer): cloud site uploads pack the files their id hashes, symlinks included * perf(aws-lambda): chunks read only the byte ranges of a video they need A chunk that extracts its own frames from a deferred project video now reads it from S3 by byte range through a short-lived pre-signed URL, instead of downloading and hashing the whole object first. S3 attests the object's stored SHA-256 before the URL is signed; every other artifact is still downloaded and verified as before. * perf(aws-lambda): upload project files once to a shared content-addressed area * chore(producer): keep planV2 comment share at the base level * fix(engine): verify the TLS certificate of https video inputs and never surface their signed query * fix(aws-lambda): renders re-upload a site's expired files instead of failing * perf(aws-lambda): the plan step probes project videos in place instead of downloading them * fix(engine): https video reads reconnect and time out, and redaction covers wrapped causes Signed-URL reads now pass -rw_timeout, -reconnect and -reconnect_delay_max (both Lambda builds accept them) and ffmpeg also -reconnect_on_network_error, so a transient S3 error or a stalled read fails fast or recovers instead of waiting out the process timeout. Query redaction now descends into error.cause. * fix(aws-lambda): share entry bundles and ratchet package budget * fix(aws-lambda): site ids uploaded before files.json still render A site id from an earlier sites create has only sites//project.tar.gz. renderToLambda now falls back to that tarball, which the plan step still reads, and when neither key exists it fails before starting with both keys named and the fix (run sites create again). * fix(aws-lambda): renders fail before upload on a stack too old to read sites A stack deployed before files.json sites untars the site file in its plan step and fails after four retries. renderToLambda now reads the state machine definition first and stops with STACK_TOO_OLD and the redeploy command when the plan task does not list SITE_MANIFEST_INVALID, which new SAM and CDK stacks now treat as terminal. The README's upgrade section drops the planProtocol v1 fallback, which hits the same failure. * docs(aws-lambda): document shared content storage and the upgrade path The deploy guide still described the tarball HeadObject short-circuit. It now covers the cas/sha256/ store and its 30-day lifecycle rule, the STACK_TOO_OLD, SITE_FILES_MISSING and SITE_NOT_FOUND errors, that old site ids keep working, and that every project uploads once more after the upgrade. A test keeps a troubleshooting entry for each pre-submit error. * fix(producer): chunks log each video they read by URL, without its query A chunk reading a video straight from the bucket left no trace in its log. It now logs one line per remote source with the path and the URL minus its query string, so signatures never reach the log. * fix(engine): ffmpeg errors never keep part of a signed URL's query A process's stderr tail was trimmed before its URLs were redacted, so a cut inside a signed URL left a query fragment the URL pattern no longer matched, and the fragment reached the error. The stderr tail is now redacted before every trim, and the redactor also strips a signing parameter cut off from its URL. * fix(aws-lambda): comments name the site manifest instead of a file path * fix(engine): stderr redaction survives chunk splits, and every ffmpeg spawn uses it The stderr tail was redacted on each chunk and then cut, so once a cut severed a long URL's scheme, the rest of its signature in the next chunk matched neither pattern. The tail is now kept raw with 2x slack and redacted only when the text is produced, then cut to size. The unused producer audioExtractor (its own spawn, unredacted stderr) is deleted and the parity harness's PCM decode goes through ManagedChildProcess; a test fails on any ffmpeg/ffprobe spawn in engine, producer or aws-lambda that does not. * chore(producer): record the audioExtractor deletion as intended * fix(engine): drop a signature fragment the stderr tail cut from its URL The raw tail is kept at twice the cap, but that cut can itself land inside a signature, and when redaction shrinks the rest below the cap the fragment survived. After a cut, the severed token is dropped before redaction. * refactor(engine): name the severed-token drop instead of commenting it * test(engine): a cut-off URL leaves nothing behind; aws-lambda size budget to measured * fix(producer): chunk URL-read tests use main's frame list; producer size budget to measured * chore(producer): size budget at the measured size, no headroom * chore(producer): size budget back to measured x1.05, the check's own rule --- bun.lock | 3 + docs/deploy/aws-lambda.mdx | 26 +- docs/deploy/gcp-cloud-run.mdx | 3 +- .../migrating-to-hyperframes-lambda.mdx | 2 +- docs/packages/cli.mdx | 5 +- examples/aws-lambda/template.yaml | 7 + packages/aws-lambda/README.md | 13 +- packages/aws-lambda/build.mjs | 26 +- packages/aws-lambda/package.json | 1 + .../HyperframesRenderStack.contract.test.ts | 9 +- .../HyperframesRenderStack.snapshot.test.ts | 37 ++ .../src/cdk/HyperframesRenderStack.ts | 8 + packages/aws-lambda/src/contentStore.ts | 15 + packages/aws-lambda/src/handler.test.ts | 273 ++++++++++++++- packages/aws-lambda/src/handler.ts | 114 +++--- .../aws-lambda/src/s3PlanV2Publisher.test.ts | 51 ++- packages/aws-lambda/src/s3PlanV2Publisher.ts | 31 +- packages/aws-lambda/src/s3Transport.test.ts | 102 +++++- packages/aws-lambda/src/s3Transport.ts | 180 +++++++++- .../aws-lambda/src/sdk/__fixtures__/fakeS3.ts | 70 ++-- .../aws-lambda/src/sdk/deploySite.test.ts | 86 ++++- packages/aws-lambda/src/sdk/deploySite.ts | 218 +++++++----- packages/aws-lambda/src/sdk/index.ts | 7 +- .../aws-lambda/src/sdk/renderToLambda.test.ts | 126 ++++++- packages/aws-lambda/src/sdk/renderToLambda.ts | 46 ++- packages/aws-lambda/src/siteManifest.ts | 106 ++++++ .../cli/src/commands/lambda/render-batch.ts | 24 +- packages/cli/src/commands/lambda/render.ts | 20 +- packages/cli/src/commands/lambda/sites.ts | 12 +- packages/core/src/index.ts | 6 +- packages/core/src/telemetryRedaction.test.ts | 8 + packages/core/src/telemetryRedaction.ts | 7 +- packages/engine/src/services/audioMixer.ts | 2 + .../src/services/videoFrameExtractor.ts | 23 +- packages/engine/src/utils/ffprobe.test.ts | 32 ++ packages/engine/src/utils/ffprobe.ts | 17 +- .../src/utils/managedChildProcess.test.ts | 51 +++ .../engine/src/utils/managedChildProcess.ts | 41 ++- packages/engine/src/utils/urlInput.test.ts | 62 ++++ packages/engine/src/utils/urlInput.ts | 42 +++ packages/gcp-cloud-run/src/gcsTransport.ts | 12 +- .../gcp-cloud-run/src/sdk/deploySite.test.ts | 23 +- packages/gcp-cloud-run/src/sdk/deploySite.ts | 7 +- .../producer/scripts/test-classification.mjs | 1 + packages/producer/src/distributed.ts | 6 +- packages/producer/src/plan-parity-analysis.ts | 92 +++-- .../src/regression-harness-lambda-local.ts | 3 +- .../producer/src/services/audioExtractor.ts | 324 ------------------ .../src/services/distributed/chunkFrames.ts | 26 +- .../producer/src/services/distributed/plan.ts | 67 +++- .../src/services/distributed/planV2.test.ts | 52 ++- .../src/services/distributed/planV2.ts | 123 ++++++- .../services/distributed/planV2Publisher.ts | 1 + .../distributed/planV2RemoteFiles.test.ts | 103 ++++++ .../services/distributed/projectHash.test.ts | 74 ++++ .../src/services/distributed/projectHash.ts | 81 +++-- .../services/distributed/renderChunk.test.ts | 261 +++++++++++++- .../src/services/distributed/renderChunk.ts | 49 ++- .../producer/src/services/htmlCompiler.ts | 22 +- .../src/services/render/stages/audioStage.ts | 4 +- .../services/render/stages/compileStage.ts | 3 + .../render/stages/extractVideosStage.ts | 3 + .../src/services/render/stages/freezePlan.ts | 20 ++ .../src/utils/ffprobeArgvContract.test.ts | 23 ++ scripts/check-no-main-deletions.mjs | 4 + scripts/package-size-budgets.json | 4 +- scripts/test-reachability.json | 2 +- 67 files changed, 2534 insertions(+), 768 deletions(-) create mode 100644 packages/aws-lambda/src/contentStore.ts create mode 100644 packages/aws-lambda/src/siteManifest.ts create mode 100644 packages/engine/src/utils/urlInput.test.ts create mode 100644 packages/engine/src/utils/urlInput.ts delete mode 100644 packages/producer/src/services/audioExtractor.ts create mode 100644 packages/producer/src/services/distributed/planV2RemoteFiles.test.ts create mode 100644 packages/producer/src/services/distributed/projectHash.test.ts diff --git a/bun.lock b/bun.lock index 1317e1e49f..45be98330f 100644 --- a/bun.lock +++ b/bun.lock @@ -32,6 +32,7 @@ "@aws-sdk/client-lambda": "^3.1146.0", "@aws-sdk/client-s3": "^3.1146.0", "@aws-sdk/client-sfn": "^3.1146.0", + "@aws-sdk/s3-request-presigner": "3.1146.0", "@hyperframes/producer": "workspace:^", "@sparticuz/chromium": "148.0.0", "ffmpeg-static": "^5.2.0", @@ -467,6 +468,8 @@ "@aws-sdk/nested-clients": ["@aws-sdk/nested-clients@3.997.46", "", { "dependencies": { "@aws-sdk/core": "^3.978.1", "@aws-sdk/signature-v4-multi-region": "^3.996.47", "@aws-sdk/types": "^3.974.6", "@smithy/core": "^3.35.0", "@smithy/fetch-http-handler": "^5.8.0", "@smithy/node-http-handler": "^4.12.1", "@smithy/types": "^4.19.0", "tslib": "^2.6.2" } }, "sha512-oRxtBcka/JGHGs9l9p9IVajGoTP8vTPmoAzdHGy4Qcy9P5vPnDf6nhIeM/COQNY9k/OahImTRaLkHftoXvfcmQ=="], + "@aws-sdk/s3-request-presigner": ["@aws-sdk/s3-request-presigner@3.1146.0", "", { "dependencies": { "@aws-sdk/core": "^3.978.1", "@aws-sdk/signature-v4-multi-region": "^3.996.47", "@aws-sdk/types": "^3.974.6", "@smithy/core": "^3.35.0", "@smithy/types": "^4.19.0", "tslib": "^2.6.2" } }, "sha512-5BEfVr2OYX2bwB45NSWAXPoTN5M2NPLOHohgKV+XfMNcL5TF7UQN4hAOVF4S6DwrA4BrI3MsAry7vqdZ2aJEjg=="], + "@aws-sdk/signature-v4-multi-region": ["@aws-sdk/signature-v4-multi-region@3.996.47", "", { "dependencies": { "@aws-sdk/types": "^3.974.6", "@smithy/signature-v4": "^5.7.3", "@smithy/types": "^4.19.0", "tslib": "^2.6.2" } }, "sha512-Zk08macMvQTHzQJCLJVkOlviVoqwYMrpXv4lmLN7b7sAbiMoOK7Go0NYdR5UeF+MW8LIbRmwrNy9u/5VvX1U5g=="], "@aws-sdk/token-providers": ["@aws-sdk/token-providers@3.1138.0", "", { "dependencies": { "@aws-sdk/core": "^3.978.1", "@aws-sdk/nested-clients": "^3.997.46", "@aws-sdk/types": "^3.974.6", "@smithy/core": "^3.35.0", "@smithy/types": "^4.19.0", "tslib": "^2.6.2" } }, "sha512-GpyAr0DD63YOEmYFM6Df+gJuIgC92MMTiBK4FTKfxii5MJ9ge20epR7LyroulscYlG89J+ZB2ivFDPjvfQhzdw=="], diff --git a/docs/deploy/aws-lambda.mdx b/docs/deploy/aws-lambda.mdx index dd4a0e00c2..d6f769431c 100644 --- a/docs/deploy/aws-lambda.mdx +++ b/docs/deploy/aws-lambda.mdx @@ -87,7 +87,7 @@ See the [CLI reference](/packages/cli#hyperframes-lambda) for full flag document #### Pre-staging a project with `sites create` -Re-rendering the same project tree on every `lambda render` call re-tars and re-uploads it each time. For tight inner loops (CI smoke jobs, prompt iteration in a demo flow), pre-stage the project once and reuse the upload: +Every `lambda render` call without `--site-id` hashes the project tree and checks each file against the bucket. For tight inner loops (CI smoke jobs, prompt iteration in a demo flow), pre-stage the project once and reuse the upload: ```bash hyperframes lambda sites create ./my-project @@ -97,7 +97,15 @@ hyperframes lambda render ./my-project --site-id=a1b2c3d4e5f6g7h8 \ --width 1920 --height 1080 --wait ``` -The `siteId` is content-addressed via a SHA-256 of the project tree; re-running `sites create` on an unchanged tree skips the upload via a `HeadObject` short-circuit. Pass the same `--site-id` to as many `lambda render` calls as you like — they all reuse the one S3 PUT. +The `siteId` is content-addressed via a SHA-256 of the project tree. A site is a small `sites//files.json` that lists each file's path and SHA-256. The files themselves live once per bucket under `cas/sha256/`, shared by every site and render, so a file another project already uploaded is never uploaded again. Pass the same `--site-id` to as many `lambda render` calls as you like. + +The stack's bucket expires objects under `cas/sha256/` 30 days after they were last written (the `ExpireSharedContent` lifecycle rule). Each render or `sites create` that reuses a file at least 7 days old rewrites it in place, so files in use never expire. If a site's files did expire, `lambda render --site-id` re-uploads them from the project directory. `renderToLambda` given a `siteHandle` but no `projectDir` stops with `SITE_FILES_MISSING` before starting; run `sites create` again to restore them. + +#### Upgrading from a version before shared content storage + +- **Redeploy the stack before upgrading the CLI or SDK that starts renders.** A stack from an earlier version cannot read `files.json` sites. The CLI and `renderToLambda` check the state machine first and stop with `STACK_TOO_OLD` before uploading anything; run `hyperframes lambda deploy` (or redeploy your SAM template or CDK construct from the same version) and render again. +- **Existing site ids keep working.** When `sites//files.json` is missing, the render uses that site's `project.tar.gz` from the earlier version. If neither exists the render stops with `SITE_NOT_FOUND`, naming both keys; run `sites create` again. +- **Every project uploads once more.** The site id scheme changed (it now also covers symlinked files), so `sites create` or the first `lambda render` of each project after the upgrade returns a new site id and uploads its files once. ### Path 2 — Direct SAM deploy @@ -166,6 +174,8 @@ hyperframes lambda policies role hyperframes lambda policies validate ./infra/iam/hyperframes-deploy.json ``` +Code that calls the SDK's `renderToLambda` with its own role needs `states:DescribeStateMachine` on the state machine besides `states:StartExecution` and S3 read/write on the bucket: each render reads the definition to confirm the stack is new enough. + The generated documents grant `Resource: "*"` for the CLI's required action set. After your first successful deploy you can narrow `Resource` to the deployed ARNs — predictable per the CloudFormation outputs above. Adopters running the CLI in CI typically check the policy doc into source control and run `policies validate` as a pre-deploy step to catch drift. ## Cost shape @@ -220,6 +230,14 @@ If progress doesn't advance for >10 minutes, check the Step Functions execution The render bucket is created with CloudFormation `Retain` on delete — `hyperframes lambda destroy` (or `sam delete`) tears the function + state machine down but the bucket survives. This is intentional: it protects final-rendered MP4s from being lost when you re-deploy. To fully reclaim storage, empty + delete the bucket via the AWS console / `aws s3 rb`. +### `STACK_TOO_OLD` + +The stack was deployed by an earlier version that cannot read this version's sites. Redeploy it (`hyperframes lambda deploy`), then render again. See [Upgrading](#upgrading-from-a-version-before-shared-content-storage). + +### `SITE_FILES_MISSING` or `SITE_NOT_FOUND` + +`SITE_FILES_MISSING`: some of the site's files expired from `cas/sha256/` after 30 days unused. Pass the project directory to `lambda render`, or run `sites create` again. `SITE_NOT_FOUND`: the bucket has no site under that id in either format; run `sites create` again and use the id it prints. + ## Current limits - There is no completion webhook. Poll with `hyperframes lambda progress` or @@ -231,8 +249,8 @@ The render bucket is created with CloudFormation `Retain` on delete — `hyperfr - The distributed AWS path is SDR-only. - Each render chunk extracts only the video frames it shows, so the plan step no longer grows with video length. On a 10,240 MB function, a 26-minute - 576×1024 24 fps H.264 clip plans in about 20 seconds. Each chunk downloads - the whole source file of every video it shows. Variable-frame-rate and HDR + 576×1024 24 fps H.264 clip plans in about 20 seconds. Each chunk reads + only the byte ranges of each video it shows, straight from the bucket. Variable-frame-rate and HDR sources are the exception: the plan step still extracts every frame of those, inside Lambda's 15-minute limit, in 2-minute segments, so `FFMPEG_PROCESS_TIMEOUT_MS` applies to each segment, not to the whole clip. diff --git a/docs/deploy/gcp-cloud-run.mdx b/docs/deploy/gcp-cloud-run.mdx index 3baa640219..72f655aa59 100644 --- a/docs/deploy/gcp-cloud-run.mdx +++ b/docs/deploy/gcp-cloud-run.mdx @@ -72,7 +72,8 @@ use H.264 or H.265. Distributed rendering is currently SDR-only. ## Reuse an upload Projects are content-addressed. Upload an unchanged project once, then reuse -its site ID for later renders: +its site ID for later renders. Site IDs from earlier versions keep working, but +the ID scheme changed, so each project uploads once more under a new ID: ```bash hyperframes cloudrun sites create ./my-project diff --git a/docs/deploy/migrating-to-hyperframes-lambda.mdx b/docs/deploy/migrating-to-hyperframes-lambda.mdx index 7301df8033..cba939ee59 100644 --- a/docs/deploy/migrating-to-hyperframes-lambda.mdx +++ b/docs/deploy/migrating-to-hyperframes-lambda.mdx @@ -10,7 +10,7 @@ If you're already running a different framework that deploys a serverless video | In your current framework you call... | In HyperFrames you call... | Notes | | ------------------------------------- | ---------------------------------------------------------------- | -------------------------------------------------------------------------------------------- | | One-shot deploy command | `hyperframes lambda deploy` | Builds `packages/aws-lambda/dist/handler.zip` and runs `sam deploy`. Idempotent. | -| One-shot site upload | `hyperframes lambda sites create ./project` | Content-addressed S3 key — re-uploads of an unchanged tree are skipped via a HeadObject 200. | +| One-shot site upload | `hyperframes lambda sites create ./project` | Content-addressed file list; files already in the bucket's shared store are skipped. | | Trigger a render | `hyperframes lambda render ./project --width 1920 --height 1080` | Returns immediately with a `renderId`; add `--wait` to stream per-chunk progress. | | Poll render progress | `hyperframes lambda progress ` | Includes accrued cost in the same response. | | Tear down | `hyperframes lambda destroy` | The S3 bucket is `Retain`'d — documented in the deploy guide. | diff --git a/docs/packages/cli.mdx b/docs/packages/cli.mdx index 2c4b74a486..684a0a8515 100644 --- a/docs/packages/cli.mdx +++ b/docs/packages/cli.mdx @@ -1983,8 +1983,9 @@ hyperframes lambda deploy \ ### `lambda sites create ` -Tars and uploads the project to S3 under a content-addressed key, and returns a -`siteId` you can reuse. A re-render of the same tree then skips the upload. +Uploads the project's files to the bucket's shared content store, skipping any +file already there, and returns a `siteId` you can reuse. A re-render of the +same tree then skips the upload. ```bash hyperframes lambda sites create ./my-project diff --git a/examples/aws-lambda/template.yaml b/examples/aws-lambda/template.yaml index d44d00d437..bc8e38cd3f 100644 --- a/examples/aws-lambda/template.yaml +++ b/examples/aws-lambda/template.yaml @@ -130,6 +130,11 @@ Resources: # Users keep the final mp4 (different key prefix); the rest # can age out after a week to keep storage costs flat. ExpirationInDays: 7 + - Id: ExpireSharedContent + Status: Enabled + Prefix: cas/sha256/ + # Project files and plan artifacts by SHA-256, shared across renders. + ExpirationInDays: 30 Tags: - Key: Project Value: !Ref ProjectName @@ -283,6 +288,7 @@ Resources: - PLAN_ARTIFACT_DIGEST_MISMATCH - FORMAT_NOT_SUPPORTED_IN_DISTRIBUTED - CHUNK_EXCEEDS_WORKER_TIMEOUT + - SITE_MANIFEST_INVALID MaxAttempts: 0 # Another render on the same function can still take the reserved slots; wait minutes. - ErrorEquals: [Lambda.TooManyRequestsException] @@ -340,6 +346,7 @@ Resources: - FORMAT_NOT_SUPPORTED_IN_DISTRIBUTED - CHUNK_EXCEEDS_WORKER_TIMEOUT - ChromeBinaryUnavailableError + - SITE_MANIFEST_INVALID MaxAttempts: 0 - ErrorEquals: [Lambda.TooManyRequestsException] IntervalSeconds: 5 diff --git a/packages/aws-lambda/README.md b/packages/aws-lambda/README.md index 0ef9a34cb6..0ae0cf54e8 100644 --- a/packages/aws-lambda/README.md +++ b/packages/aws-lambda/README.md @@ -77,10 +77,15 @@ infrastructure upgrade. Before upgrading an application that calls same new package version. 3. Resume renders, then upgrade the application/SDK dependency. -Older state machines can default missing protocol fields to v1 or lack v2 -branches, while the new SDK sends explicit v2. If infrastructure cannot be -redeployed first, keep the application on its previous package version or -pass `planProtocol: "v1"` explicitly until the redeploy is complete. +A site is a `sites//files.json` file list that an older stack cannot +read, under either plan protocol. Before it uploads anything, +`renderToLambda` reads the state machine definition +(`states:DescribeStateMachine`) and fails with `STACK_TOO_OLD` when the +stack predates that format. If infrastructure cannot be redeployed first, +keep the application on its previous package version until the redeploy is +complete. Site ids from earlier versions keep working after the redeploy: +when `sites//files.json` is missing, the render uses that site's +`project.tar.gz`. ## Chrome runtime diff --git a/packages/aws-lambda/build.mjs b/packages/aws-lambda/build.mjs index d639aca5c5..5472558b20 100644 --- a/packages/aws-lambda/build.mjs +++ b/packages/aws-lambda/build.mjs @@ -2,10 +2,11 @@ /** * Build script for @hyperframes/aws-lambda (public OSS package). * - * Bundles each subpath barrel via esbuild → dist/, then emits .d.ts via tsc. + * Bundles the subpath barrels in one esbuild pass → dist/, then emits .d.ts via tsc. * * Subpaths (each gets its own dist entry so adopters that import one path - * don't load the others' transitive graphs at module-load time): + * don't load the others' transitive graphs at module-load time; code two + * entries share is emitted once, in dist/_chunks/): * * . (the umbrella barrel: handler + sdk + cdk types re-exported) * ./handler (the Lambda runtime entry — what `scripts/build-zip.ts` ZIPs) @@ -33,6 +34,7 @@ const sharedOpts = { "@aws-sdk/client-lambda", "@aws-sdk/client-s3", "@aws-sdk/client-sfn", + "@aws-sdk/s3-request-presigner", "@hyperframes/producer", "@hyperframes/producer/distributed", "@sparticuz/chromium", @@ -45,12 +47,18 @@ const sharedOpts = { ], }; -await Promise.all([ - build({ ...sharedOpts, entryPoints: ["src/index.ts"], outfile: "dist/index.js" }), - build({ ...sharedOpts, entryPoints: ["src/handler.ts"], outfile: "dist/handler.js" }), - build({ ...sharedOpts, entryPoints: ["src/sdk/index.ts"], outfile: "dist/sdk/index.js" }), - build({ ...sharedOpts, entryPoints: ["src/cdk/index.ts"], outfile: "dist/cdk/index.js" }), -]); +await build({ + ...sharedOpts, + entryPoints: { + index: "src/index.ts", + handler: "src/handler.ts", + "sdk/index": "src/sdk/index.ts", + "cdk/index": "src/cdk/index.ts", + }, + outdir: "dist", + splitting: true, + chunkNames: "_chunks/[name]-[hash]", +}); // esbuild doesn't emit .d.ts. tsc does, with a build-only tsconfig that // drops the workspace `paths` overrides so `@hyperframes/producer` resolves @@ -62,4 +70,4 @@ execSync( { stdio: "inherit" }, ); -console.log("[Build] Complete: dist/{index,handler,sdk/index,cdk/index}.js + .d.ts"); +console.log("[Build] Complete: dist/{index,handler,sdk/index,cdk/index}.js + dist/_chunks + .d.ts"); diff --git a/packages/aws-lambda/package.json b/packages/aws-lambda/package.json index a42cde95a2..a45dd3ca4e 100644 --- a/packages/aws-lambda/package.json +++ b/packages/aws-lambda/package.json @@ -72,6 +72,7 @@ "@aws-sdk/client-lambda": "^3.1146.0", "@aws-sdk/client-s3": "^3.1146.0", "@aws-sdk/client-sfn": "^3.1146.0", + "@aws-sdk/s3-request-presigner": "3.1146.0", "@hyperframes/producer": "workspace:^", "@sparticuz/chromium": "148.0.0", "ffmpeg-static": "^5.2.0", diff --git a/packages/aws-lambda/src/cdk/HyperframesRenderStack.contract.test.ts b/packages/aws-lambda/src/cdk/HyperframesRenderStack.contract.test.ts index 9db00548df..80313afb66 100644 --- a/packages/aws-lambda/src/cdk/HyperframesRenderStack.contract.test.ts +++ b/packages/aws-lambda/src/cdk/HyperframesRenderStack.contract.test.ts @@ -15,6 +15,7 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import { App, Stack } from "aws-cdk-lib"; import { Template } from "aws-cdk-lib/assertions"; +import { CONTENT_STORE_PREFIX, CONTENT_STORE_RETENTION_DAYS } from "../contentStore.js"; import { HyperframesRenderStack } from "./HyperframesRenderStack.js"; // CDK synth is slow on cold start and reached 32s under full-workspace CI @@ -61,7 +62,7 @@ describe("HyperframesRenderStack — contract", () => { }); }); - it("provisions exactly one S3 bucket with PublicAccessBlockConfiguration and a 7-day intermediates lifecycle", () => { + it("provisions exactly one S3 bucket with PublicAccessBlockConfiguration and a 7-day intermediates and 30-day shared content lifecycle", () => { const t = DEFAULT_TEMPLATE; t.resourceCountIs("AWS::S3::Bucket", 1); t.hasResourceProperties("AWS::S3::Bucket", { @@ -79,6 +80,12 @@ describe("HyperframesRenderStack — contract", () => { Prefix: "renders/", ExpirationInDays: 7, }, + { + Id: "ExpireSharedContent", + Status: "Enabled", + Prefix: `${CONTENT_STORE_PREFIX}/`, + ExpirationInDays: CONTENT_STORE_RETENTION_DAYS, + }, ], }, }); diff --git a/packages/aws-lambda/src/cdk/HyperframesRenderStack.snapshot.test.ts b/packages/aws-lambda/src/cdk/HyperframesRenderStack.snapshot.test.ts index e1493ae7b4..8179f9c0cf 100644 --- a/packages/aws-lambda/src/cdk/HyperframesRenderStack.snapshot.test.ts +++ b/packages/aws-lambda/src/cdk/HyperframesRenderStack.snapshot.test.ts @@ -26,6 +26,12 @@ import { join } from "node:path"; import { App, Stack } from "aws-cdk-lib"; import { Template } from "aws-cdk-lib/assertions"; import { parse as parseYaml } from "yaml"; +import { + CONTENT_STORE_PREFIX, + CONTENT_STORE_REFRESH_AFTER_DAYS, + CONTENT_STORE_RETENTION_DAYS, +} from "../contentStore.js"; +import { SITE_MANIFEST_INVALID } from "../siteManifest.js"; import { HyperframesRenderStack } from "./HyperframesRenderStack.js"; // CDK synth + Template.fromStack is slow on cold start in CI (~5-8s on @@ -84,6 +90,7 @@ const EXPECTED_NON_RETRYABLE_ERRORS = new Set([ "PLAN_ARTIFACT_DIGEST_MISMATCH", "FORMAT_NOT_SUPPORTED_IN_DISTRIBUTED", "ChromeBinaryUnavailableError", + "SITE_MANIFEST_INVALID", ]); function doSynth(reservedConcurrency?: number): { @@ -213,6 +220,16 @@ describe("HyperframesRenderStack — snapshot", () => { } }); + it("lists the error the SDK checks for before rendering in every SAM and CDK plan task", () => { + for (const definition of [SYNTHED.definition, readSamDefinition()]) { + for (const plan of [getV1TaskStates(definition).Plan, getV2TaskStates(definition).PlanV2]) { + const errors = new Set(); + collectNonRetryableErrors(plan, errors); + expect(errors.has(SITE_MANIFEST_INVALID)).toBe(true); + } + } + }); + it("keeps SAM and CDK terminal classifiers identical for every v2 Lambda task", () => { const cdkTasks = getV2TaskStates(SYNTHED.definition); const samTasks = getV2TaskStates(readSamDefinition()); @@ -267,6 +284,26 @@ describe("HyperframesRenderStack — snapshot", () => { expect(v2).not.toContain("PlanS3Uri"); }); + it("keeps the SAM bucket lifecycle identical to CDK's, shared content on the transport's terms", () => { + const cdkBucket = Object.values(SYNTHED.template.findResources("AWS::S3::Bucket"))[0] as { + Properties: { LifecycleConfiguration: unknown }; + }; + const samBucket = requireRecordProperty(readSamResources(), "RenderBucket", "SAM RenderBucket"); + const samProperties = requireRecordProperty(samBucket, "Properties", "SAM bucket properties"); + expect(samProperties.LifecycleConfiguration).toEqual( + cdkBucket.Properties.LifecycleConfiguration, + ); + expect(CONTENT_STORE_REFRESH_AFTER_DAYS * 2).toBeLessThan(CONTENT_STORE_RETENTION_DAYS); + expect(cdkBucket.Properties.LifecycleConfiguration).toMatchObject({ + Rules: expect.arrayContaining([ + expect.objectContaining({ + Prefix: `${CONTENT_STORE_PREFIX}/`, + ExpirationInDays: CONTENT_STORE_RETENTION_DAYS, + }), + ]), + }); + }); + it("sends the same payload from every Lambda task as the SAM template", () => { // CDK drops null-valued payload fields when it renders the definition (#5259), so compare // what each deployed task sends rather than the construct source. diff --git a/packages/aws-lambda/src/cdk/HyperframesRenderStack.ts b/packages/aws-lambda/src/cdk/HyperframesRenderStack.ts index 5e6031a0c1..08038d1432 100644 --- a/packages/aws-lambda/src/cdk/HyperframesRenderStack.ts +++ b/packages/aws-lambda/src/cdk/HyperframesRenderStack.ts @@ -30,6 +30,7 @@ import * as s3 from "aws-cdk-lib/aws-s3"; import * as sfn from "aws-cdk-lib/aws-stepfunctions"; import * as tasks from "aws-cdk-lib/aws-stepfunctions-tasks"; import { Construct } from "constructs"; +import { CONTENT_STORE_PREFIX, CONTENT_STORE_RETENTION_DAYS } from "../contentStore.js"; /** Construction-time props for {@link HyperframesRenderStack}. */ export interface HyperframesRenderStackProps { @@ -93,6 +94,12 @@ export class HyperframesRenderStack extends Construct { prefix: "renders/", expiration: Duration.days(7), }, + { + id: "ExpireSharedContent", + enabled: true, + prefix: `${CONTENT_STORE_PREFIX}/`, + expiration: Duration.days(CONTENT_STORE_RETENTION_DAYS), + }, ], }); @@ -218,6 +225,7 @@ export class HyperframesRenderStack extends Construct { "FORMAT_NOT_SUPPORTED_IN_DISTRIBUTED", "CHUNK_EXCEEDS_WORKER_TIMEOUT", "ChromeBinaryUnavailableError", + "SITE_MANIFEST_INVALID", ]; const NON_RETRYABLE_CHUNK = [ "FFMPEG_VERSION_MISMATCH", diff --git a/packages/aws-lambda/src/contentStore.ts b/packages/aws-lambda/src/contentStore.ts new file mode 100644 index 0000000000..bcccd5f8a5 --- /dev/null +++ b/packages/aws-lambda/src/contentStore.ts @@ -0,0 +1,15 @@ +// Shared by every render: project files and plan-v2 artifacts by SHA-256. Dependency-free for CDK. + +export const CONTENT_STORE_PREFIX = "cas/sha256"; +export const CONTENT_STORE_RETENTION_DAYS = 30; +/** Reusing an object at least this old rewrites it, so a plan's objects outlive it by 23+ days. */ +export const CONTENT_STORE_REFRESH_AFTER_DAYS = 7; + +export function contentStoreUri(bucket: string, sha256: string): string { + if (!/^[a-f0-9]{64}$/.test(sha256)) { + throw new Error( + `[contentStore] expected lowercase SHA-256 digest, got ${JSON.stringify(sha256)}`, + ); + } + return `s3://${bucket}/${CONTENT_STORE_PREFIX}/${sha256.slice(0, 2)}/${sha256}`; +} diff --git a/packages/aws-lambda/src/handler.test.ts b/packages/aws-lambda/src/handler.test.ts index 2e4e22a19d..4731384d0c 100644 --- a/packages/aws-lambda/src/handler.test.ts +++ b/packages/aws-lambda/src/handler.test.ts @@ -25,6 +25,7 @@ import { LambdaClient } from "@aws-sdk/client-lambda"; import { S3Client } from "@aws-sdk/client-s3"; import { CURRENT_PLAN_PROTOCOL, + PLAN_V2_MATERIALIZATION_MARKER, PlanVideosMetadataError, type AssembleResult, type ChunkResult, @@ -32,12 +33,14 @@ import { type PlanResult, type PlanV2ArtifactPublisher, type PlanV2Manifest, + type PlanV2WithPublisherOptions, publishPlanV2FromExecutionPlan, } from "@hyperframes/producer/distributed"; import { resolveMaxParallelChunks } from "../../producer/src/services/distributed/plan.js"; import { recomputePlanHashFromPlanDir } from "../../producer/src/services/render/stages/freezePlan.js"; import type { AssembleEvent, LambdaEvent, PlanEvent, RenderChunkEvent } from "./events.js"; import { handler as handleEvent, unwrapEvent } from "./handler.js"; +import { S3PlanV2ArtifactPublisher } from "./s3PlanV2Publisher.js"; interface FakeS3Op { kind: "download" | "upload" | "copy"; @@ -55,6 +58,7 @@ class FakeS3Client { // Map S3 URIs → byte buffers the fake serves. objects = new Map(); metadata = new Map>(); + checksums = new Map(); // Methods called by the real S3 transport — minimal surface so the // handler's call sites don't need rewriting under test. @@ -95,6 +99,7 @@ class FakeS3Client { return { ContentLength: bytes.length, Metadata: this.metadata.get(uri), + ChecksumSHA256: this.checksums.get(uri), }; } if (cmdName === "PutObjectCommand") { @@ -114,6 +119,8 @@ class FakeS3Client { this.objects.set(uri, bytes); const metadata = (command as { input: { Metadata?: Record } }).input.Metadata; if (metadata) this.metadata.set(uri, metadata); + const checksum = (command as { input: { ChecksumSHA256?: string } }).input.ChecksumSHA256; + if (checksum) this.checksums.set(uri, checksum); return {}; } if (cmdName === "CreateMultipartUploadCommand") return { UploadId: "upload-1" }; @@ -157,6 +164,15 @@ const accountUnreadable = { } as unknown as LambdaClient; const handler: typeof handleEvent = (event, deps) => handleEvent(event, { lambda: accountUnreadable, ...deps }); +/** A real client (so it can pre-sign offline) whose requests go to the fake. */ +function presigningClient(fake: FakeS3Client): S3Client { + const client = new S3Client({ + region: "us-east-2", + credentials: { accessKeyId: "AKIDTEST", secretAccessKey: "test-secret" }, + }); + client.send = fake.send.bind(fake) as S3Client["send"]; + return client; +} const tmpDirs: string[] = []; @@ -866,7 +882,7 @@ describe("handler dispatch", () => { expect(planned).toMatchObject({ PlanProtocol: "v2", PlanV2ManifestS3Uri: "s3://bucket/renders/v2/v2/manifest.json", - PlanV2ArtifactS3Prefix: "s3://bucket/renders/v2/v2/artifacts/sha256", + PlanV2ArtifactS3Prefix: "s3://bucket/cas/sha256", }); expect("PlanS3Uri" in planned).toBe(false); if (!("PlanProtocol" in planned) || planned.PlanProtocol !== "v2") { @@ -917,6 +933,85 @@ describe("handler dispatch", () => { ).toBe(true); }); + it("hands a chunk its deferred video as a pre-signed URL and downloads the rest", async () => { + const tmpRoot = makeTmpRoot(); + const fake = new FakeS3Client(); + const s3 = presigningClient(fake); + const v1Dir = join(tmpRoot, "v1"); + makeMinimalV1PlanDir(v1Dir, true, { withDeferredVideo: true }); + const publisher = new S3PlanV2ArtifactPublisher({ + s3, + planOutputS3Prefix: "s3://bucket/renders/deferred/", + temporaryRoot: tmpRoot, + }); + const manifest = await publishPlanV2FromExecutionPlan(v1Dir, publisher); + const video = manifest.artifacts.find((artifact) => artifact.path === "compiled/clip.mp4")!; + expect(video.chunks).toEqual([0]); + const blobUri = (digest: string) => + `${publisher.artifactPrefix}/${digest.slice(0, 2)}/${digest}`; + + let marker: { remoteSources?: Record } = {}; + const renderChunkMock = mock( + async (planDir: string, _chunkIndex: number, outputPath: string): Promise => { + expect(existsSync(join(planDir, "compiled", "clip.mp4"))).toBe(false); + marker = JSON.parse(readFileSync(join(planDir, PLAN_V2_MATERIALIZATION_MARKER), "utf-8")); + writeFileSync(outputPath, "CHUNK"); + return { + outputPath, + outputKind: "file", + framesEncoded: 30, + sha256: "c".repeat(64), + durationMs: 1, + planHashMs: 0, + sessionBootMs: 0, + captureStageMs: 1, + encodeStageMs: 0, + workers: 1, + captureMode: "beginframe", + perfPath: `${outputPath}.perf.json`, + }; + }, + ); + const beforeChunk = fake.ops.length; + await handler( + { + Action: "renderChunk", + PlanV2ManifestS3Uri: publisher.manifestUri, + PlanV2ArtifactS3Prefix: publisher.artifactPrefix, + PlanHash: manifest.planHash, + ChunkIndex: 0, + ChunkOutputS3Prefix: "s3://bucket/renders/deferred/", + Format: "mp4", + }, + { + s3, + primitives: { + plan: mock() as unknown as typeof import("@hyperframes/producer/distributed").plan, + renderChunk: + renderChunkMock as unknown as typeof import("@hyperframes/producer/distributed").renderChunk, + assemble: + mock() as unknown as typeof import("@hyperframes/producer/distributed").assemble, + }, + tmpRoot, + skipChromeResolution: true, + }, + ); + + const downloads = fake.ops + .slice(beforeChunk) + .filter((operation) => operation.kind === "download") + .map((operation) => operation.uri); + const expectedDownloads = manifest.artifacts + .filter((artifact) => artifact.chunks === "all") + .map((artifact) => blobUri(artifact.sha256)); + expect(new Set(downloads)).toEqual(new Set([publisher.manifestUri, ...expectedDownloads])); + const url = new URL(marker.remoteSources?.["compiled/clip.mp4"] ?? ""); + expect(url.protocol).toBe("https:"); + expect(url.pathname).toEndWith(`/${video.sha256}`); + expect(url.searchParams.get("X-Amz-Expires")).toBe("3600"); + expect(url.searchParams.get("X-Amz-Signature")).toMatch(/^[a-f0-9]{64}$/); + }); + async function planStitched(chunkCount: number) { const tmpRoot = makeTmpRoot(); const s3 = new FakeS3Client(); @@ -1063,6 +1158,132 @@ describe("handler dispatch", () => { }); }); +/** A real client (so it can pre-sign offline) whose requests go to the fake. */ +function presigningClient(fake: FakeS3Client): import("@aws-sdk/client-s3").S3Client { + const client = new S3Client({ + region: "us-east-2", + credentials: { accessKeyId: "AKIDTEST", secretAccessKey: "test-secret" }, + }); + client.send = fake.send.bind(fake) as S3Client["send"]; + return client; +} + +describe("handler — plan from a deployed site", () => { + function storeSiteFile(s3: FakeS3Client, bytes: string): string { + const sha256 = createHash("sha256").update(bytes).digest("hex"); + const uri = `s3://bucket/cas/sha256/${sha256.slice(0, 2)}/${sha256}`; + s3.objects.set(uri, Buffer.from(bytes)); + s3.metadata.set(uri, { sha256 }); + return sha256; + } + + function planEvent(): PlanEvent { + return { + Action: "plan", + ProjectS3Uri: "s3://bucket/sites/site1/files.json", + PlanOutputS3Prefix: "s3://bucket/renders/r1/", + Config: { fps: 30, width: 640, height: 360, format: "mp4" }, + }; + } + + it("plans a site's video in place: never downloads, never uploads its bytes", async () => { + const s3 = new FakeS3Client(); + const html = storeSiteFile(s3, ""); + const media = storeSiteFile(s3, "VIDEO-BYTES"); + s3.objects.set( + "s3://bucket/sites/site1/files.json", + Buffer.from( + JSON.stringify({ + files: [ + { path: "assets/bg.mp4", sha256: media, sizeBytes: 11 }, + { path: "index.html", sha256: html, sizeBytes: 13 }, + ], + }), + ), + ); + const planV2WithPublisher = mock( + async ( + projectDir: string, + _config: unknown, + publisher: PlanV2ArtifactPublisher, + options: PlanV2WithPublisherOptions, + ) => { + expect(readFileSync(join(projectDir, "index.html"), "utf-8")).toBe(""); + expect(existsSync(join(projectDir, "assets", "bg.mp4"))).toBe(false); + const remote = options.remoteFiles?.["assets/bg.mp4"]; + expect(remote).toMatchObject({ sha256: media, sizeBytes: 11 }); + expect(remote?.url).toContain(`/cas/sha256/${media.slice(0, 2)}/${media}?`); + expect(remote?.url).toContain("X-Amz-Signature="); + await publisher.referenceStoredBlob!({ sha256: media, sizeBytes: 11 }); + const manifest = { + planHash: "c".repeat(64), + chunkCount: 1, + totalFrames: 30, + fps: 30, + width: 640, + height: 360, + format: "mp4", + ffmpegVersion: "test", + producerVersion: "test", + artifacts: [{ path: "compiled/assets/bg.mp4", sha256: media, sizeBytes: 11 }], + }; + await publisher.commitManifest(JSON.stringify(manifest)); + return manifest as unknown as PlanV2Manifest; + }, + ); + + const result = await handler(planEvent(), { + s3: presigningClient(s3), + primitives: { + plan: mock() as unknown as typeof import("@hyperframes/producer/distributed").plan, + planV2WithPublisher: + planV2WithPublisher as unknown as typeof import("@hyperframes/producer/distributed").planV2WithPublisher, + renderChunk: + mock() as unknown as typeof import("@hyperframes/producer/distributed").renderChunk, + assemble: mock() as unknown as typeof import("@hyperframes/producer/distributed").assemble, + }, + tmpRoot: makeTmpRoot(), + skipChromeResolution: true, + }); + + expect(result).toMatchObject({ PlanV2ArtifactS3Prefix: "s3://bucket/cas/sha256" }); + expect(planV2WithPublisher).toHaveBeenCalledTimes(1); + expect(s3.ops.filter((op) => op.kind === "upload").map((op) => op.uri)).toEqual([ + "s3://bucket/renders/r1/v2/manifest.json", + ]); + expect(s3.ops.some((op) => op.uri.endsWith(".tar.gz"))).toBe(false); + const mediaKey = `s3://bucket/cas/sha256/${media.slice(0, 2)}/${media}`; + expect(s3.ops.filter((op) => op.uri === mediaKey && op.kind === "download")).toEqual([]); + }); + + it("refuses a site manifest path that escapes the project", async () => { + const s3 = new FakeS3Client(); + const sha256 = storeSiteFile(s3, "x"); + s3.objects.set( + "s3://bucket/sites/site1/files.json", + Buffer.from(JSON.stringify({ files: [{ path: "../escape", sha256, sizeBytes: 1 }] })), + ); + const planV2WithPublisher = mock(); + await expect( + handler(planEvent(), { + s3: s3 as unknown as import("@aws-sdk/client-s3").S3Client, + primitives: { + plan: mock() as unknown as typeof import("@hyperframes/producer/distributed").plan, + planV2WithPublisher: + planV2WithPublisher as unknown as typeof import("@hyperframes/producer/distributed").planV2WithPublisher, + renderChunk: + mock() as unknown as typeof import("@hyperframes/producer/distributed").renderChunk, + assemble: + mock() as unknown as typeof import("@hyperframes/producer/distributed").assemble, + }, + tmpRoot: makeTmpRoot(), + skipChromeResolution: true, + }), + ).rejects.toMatchObject({ name: "SITE_MANIFEST_INVALID" }); + expect(planV2WithPublisher).not.toHaveBeenCalled(); + }); +}); + describe("handler — S3 URI allowlist (security: F-004)", () => { let prevBucket: string | undefined; @@ -1177,7 +1398,7 @@ async function makeMinimalPlanTar(): Promise { function makeMinimalV1PlanDir( dir: string, withAudio: boolean, - options: { chunkCount?: number; chunksCarryAudio?: boolean } = {}, + options: { chunkCount?: number; chunksCarryAudio?: boolean; withDeferredVideo?: boolean } = {}, ): void { const chunkCount = options.chunkCount ?? 1; mkdirSync(join(dir, "meta"), { recursive: true }); @@ -1208,6 +1429,54 @@ function makeMinimalV1PlanDir( options.chunksCarryAudio ? JSON.stringify({ chunksCarryAudio: true }) : "{}", ); if (withAudio) writeFileSync(join(dir, "audio.m4a"), "AAC"); + if (options.withDeferredVideo) writeDeferredVideo(dir); planJson.planHash = recomputePlanHashFromPlanDir(dir); writeFileSync(join(dir, "plan.json"), JSON.stringify(planJson)); } + +/** One clip the chunk extracts from `compiled/clip.mp4` itself. */ +function writeDeferredVideo(dir: string): void { + writeFileSync(join(dir, "compiled", "clip.mp4"), "VIDEO-BYTES"); + const metadata = { + durationSeconds: 1, + videoStreamDurationSeconds: 1, + width: 640, + height: 360, + fps: 30, + videoCodec: "h264", + hasAudio: false, + isVFR: false, + hasAlpha: false, + colorSpace: null, + }; + const videos = { + videos: [ + { + id: "clip", + src: "clip.mp4", + start: 0, + end: 1, + mediaStart: 0, + loop: false, + hasAudio: false, + }, + ], + extracted: [ + { + videoId: "clip", + srcPath: "/planner/clip.mp4", + framePattern: "frame_%05d.jpg", + fps: 30, + totalFrames: 30, + metadata, + deferredRange: { + sourcePath: "compiled/clip.mp4", + startTime: 0, + durationSeconds: 1, + format: "jpg", + }, + }, + ], + }; + writeFileSync(join(dir, "meta", "videos.json"), JSON.stringify(videos)); +} diff --git a/packages/aws-lambda/src/handler.ts b/packages/aws-lambda/src/handler.ts index f0fa4831fe..5e2bbc06be 100644 --- a/packages/aws-lambda/src/handler.ts +++ b/packages/aws-lambda/src/handler.ts @@ -32,10 +32,12 @@ import { resolvePlanAudioPath, planV2WithPublisher, type PlanResult, + isPlanV2DeferredSourceArtifact, type PlanV2Artifact, type PlanV2Manifest, type PlanV2MaterializationTarget, readPlanV2Manifest, + type RemoteProjectFile, renderChunk, } from "@hyperframes/producer/distributed"; import { resolveChromeExecutablePath } from "./chromium.js"; @@ -58,12 +60,15 @@ import type { import { downloadS3ObjectToFile, downloadS3ObjectToFileVerified, + mapConcurrent, parseS3Uri, + presignVerifiedS3GetUrl, tarDirectory, untarDirectory, uploadFileToS3, } from "./s3Transport.js"; import { S3PlanV2ArtifactPublisher } from "./s3PlanV2Publisher.js"; +import { downloadSiteFiles, isSiteManifestUri } from "./siteManifest.js"; /** * Lazily-constructed S3 client. Cached at module scope so warm Lambda @@ -401,17 +406,10 @@ async function handlePlan(event: PlanEvent, deps?: HandlerDeps): Promise }> { + const projectDir = join(work, "project"); + if (isSiteManifestUri(projectS3Uri)) { + const remoteFiles = await downloadSiteFiles(s3, projectS3Uri, projectDir, leaveVideosRemote); + return { projectDir, remoteFiles }; + } + const projectArchive = join(work, "project.tar.gz"); + let stageStarted = Date.now(); + await downloadS3ObjectToFile(s3, projectS3Uri, projectArchive); + logEvent({ event: "plan_stage", stage: "download", ms: Date.now() - stageStarted }); + stageStarted = Date.now(); + await untarDirectory(projectArchive, projectDir); + logEvent({ event: "plan_stage", stage: "untar", ms: Date.now() - stageStarted }); + return { projectDir, remoteFiles: {} }; +} + // Plan v2 orchestration is kept as one transactional boundary: stage, CAS upload, // and manifest-last publication must stay ordered and fail together. // fallow-ignore-next-line complexity @@ -476,15 +499,8 @@ async function handlePlanV2( } const work = mkdtempSync(join(deps?.tmpRoot ?? tmpdir(), "hf-lambda-plan-v2-")); - const projectArchive = join(work, "project.tar.gz"); - const projectDir = join(work, "project"); try { - let stageStarted = Date.now(); - await downloadS3ObjectToFile(s3, event.ProjectS3Uri, projectArchive); - logEvent({ event: "plan_stage", stage: "download", ms: Date.now() - stageStarted }); - stageStarted = Date.now(); - await untarDirectory(projectArchive, projectDir); - logEvent({ event: "plan_stage", stage: "untar", ms: Date.now() - stageStarted }); + const { projectDir, remoteFiles } = await fetchProject(s3, event.ProjectS3Uri, work, true); const publisher = new S3PlanV2ArtifactPublisher({ s3, planOutputS3Prefix: event.PlanOutputS3Prefix, @@ -492,6 +508,7 @@ async function handlePlanV2( }); const manifest: PlanV2Manifest = await primitive(projectDir, config, publisher, { stagingParentDir: work, + remoteFiles, chunksCarryAudio: true, }); @@ -828,17 +845,54 @@ async function downloadAndMaterializePlanV2( throwPlanHashMismatch(event.PlanHash, manifest.planHash); } const artifacts = listPlanV2ArtifactsForTarget(manifest, target); + const remoteSources = await presignDeferredSources(s3, event.PlanV2ArtifactS3Prefix, artifacts); const uniqueArtifacts = [ - ...new Map(artifacts.map((artifact) => [artifact.sha256, artifact])).values(), + ...new Map( + artifacts + .filter((artifact) => !Object.hasOwn(remoteSources, artifact.path)) + .map((artifact) => [artifact.sha256, artifact]), + ).values(), ]; await mapConcurrent(uniqueArtifacts, 16, async (artifact) => { await downloadPlanV2Artifact(s3, event.PlanV2ArtifactS3Prefix, transportDir, artifact); }); const planDir = join(work, "plan"); - materializePlanV2Target(transportDir, target, planDir); + materializePlanV2Target(transportDir, target, planDir, { remoteSources }); return { planDir, manifest }; } +/** Outlives the longest chunk (Lambda caps an invocation at 900 s) with margin. */ +const DEFERRED_SOURCE_URL_TTL_SECONDS = 3600; + +/** + * A chunk's ffmpeg reads a deferred video by byte range from a pre-signed URL + * instead of downloading it whole. The key is the content hash, written once. + */ +async function presignDeferredSources( + s3: S3Client, + artifactPrefix: string, + artifacts: readonly Readonly[], +): Promise> { + const entries = await Promise.all( + artifacts + .filter(isPlanV2DeferredSourceArtifact) + .map( + async (artifact) => + [ + artifact.path, + await presignVerifiedS3GetUrl( + s3, + planV2BlobUri(artifactPrefix, artifact.sha256), + artifact.sha256, + artifact.sizeBytes, + DEFERRED_SOURCE_URL_TTL_SECONDS, + ), + ] as const, + ), + ); + return Object.fromEntries(entries); +} + async function downloadPlanV2Artifact( s3: S3Client, artifactPrefix: string, @@ -869,30 +923,6 @@ function throwPlanHashMismatch(expected: string, actual: string): never { throw error; } -async function mapConcurrent( - values: readonly T[], - concurrency: number, - fn: (value: T) => Promise, -): Promise { - let cursor = 0; - async function worker(): Promise { - while (cursor < values.length) { - const index = cursor++; - await fn(values[index]!); - } - } - const results = await Promise.allSettled( - Array.from({ length: Math.min(concurrency, values.length) }, () => worker()), - ); - const failure = results.find( - (result): result is PromiseRejectedResult => result.status === "rejected", - ); - // Do not reject while sibling workers may still be writing into invocation - // scratch. The caller removes that directory in `finally`; draining the pool - // first prevents late S3 streams from racing cleanup after another GET fails. - if (failure) throw failure.reason; -} - async function downloadChunkObjects( s3: S3Client, uris: string[], diff --git a/packages/aws-lambda/src/s3PlanV2Publisher.test.ts b/packages/aws-lambda/src/s3PlanV2Publisher.test.ts index 9b437f9cf3..23ecde2279 100644 --- a/packages/aws-lambda/src/s3PlanV2Publisher.test.ts +++ b/packages/aws-lambda/src/s3PlanV2Publisher.test.ts @@ -112,14 +112,61 @@ describe("S3PlanV2ArtifactPublisher", () => { planOutputS3Prefix: `s3://bucket/render${"/".repeat(10_000)}`, }); - expect(publisher.artifactPrefix).toBe("s3://bucket/render/v2/artifacts/sha256"); + expect(publisher.artifactPrefix).toBe("s3://bucket/cas/sha256"); expect(publisher.manifestUri).toBe("s3://bucket/render/v2/manifest.json"); }); + it("references a remote project file already in the store, and refuses one that is not", async () => { + const digest = createHash("sha256").update("remote video").digest("hex"); + const s3 = new FakeS3(); + s3.objects.set(`s3://bucket/cas/sha256/${digest.slice(0, 2)}/${digest}`, { + bytes: Buffer.from("remote video"), + sha256: digest, + }); + const publisher = new S3PlanV2ArtifactPublisher({ + s3: s3.asClient(), + planOutputS3Prefix: "s3://bucket/renders/remote", + temporaryRoot: makeSource("x").root, + }); + + await publisher.referenceStoredBlob({ sha256: digest, sizeBytes: 12 }); + await expect( + publisher.referenceStoredBlob({ sha256: "e".repeat(64), sizeBytes: 12 }), + ).rejects.toMatchObject({ name: "PLAN_ARTIFACT_MISSING" }); + await publisher.commitManifest(manifestFor(digest)); + + expect(s3.puts.map((operation) => operation.uri)).toEqual([ + "s3://bucket/renders/remote/v2/manifest.json", + ]); + }); + + it("reuses a blob the SDK or another render already stored instead of uploading it", async () => { + const source = makeSource("shared media"); + const s3 = new FakeS3(); + const blobUri = `s3://bucket/cas/sha256/${source.digest.slice(0, 2)}/${source.digest}`; + s3.objects.set(blobUri, { bytes: Buffer.from("shared media"), sha256: source.digest }); + const publisher = new S3PlanV2ArtifactPublisher({ + s3: s3.asClient(), + planOutputS3Prefix: "s3://bucket/renders/second", + temporaryRoot: source.root, + }); + + await publisher.putBlob({ + sourcePath: source.path, + sha256: source.digest, + sizeBytes: source.sizeBytes, + }); + await publisher.commitManifest(manifestFor(source.digest)); + + expect(s3.puts.map((operation) => operation.uri)).toEqual([ + "s3://bucket/renders/second/v2/manifest.json", + ]); + }); + it("publishes immutable blobs before the fixed-key manifest", async () => { const source = makeSource("hello"); const s3 = new FakeS3(); - const artifactPrefix = "s3://bucket/render/v2/artifacts/sha256"; + const artifactPrefix = "s3://bucket/cas/sha256"; const manifestUri = "s3://bucket/render/v2/manifest.json"; const publisher = new S3PlanV2ArtifactPublisher({ s3: s3.asClient(), diff --git a/packages/aws-lambda/src/s3PlanV2Publisher.ts b/packages/aws-lambda/src/s3PlanV2Publisher.ts index 8eee0d8681..b3f1d6814e 100644 --- a/packages/aws-lambda/src/s3PlanV2Publisher.ts +++ b/packages/aws-lambda/src/s3PlanV2Publisher.ts @@ -8,7 +8,14 @@ import { type PlanV2ArtifactPublisher, type PlanV2PublishBlob, } from "@hyperframes/producer/distributed"; -import { parseS3Uri, uploadContentAddressedFileToS3 } from "./s3Transport.js"; +import { CONTENT_STORE_PREFIX } from "./contentStore.js"; +import { + ensureContentStoreObject, + formatS3Uri, + parseS3Uri, + putContentStoreFile, + uploadContentAddressedFileToS3, +} from "./s3Transport.js"; export interface S3PlanV2ArtifactPublisherOptions { readonly s3: S3Client; @@ -66,15 +73,18 @@ export class S3PlanV2ArtifactPublisher implements PlanV2ArtifactPublisher { readonly artifactPrefix: string; readonly manifestUri: string; readonly #s3: S3Client; + readonly #bucket: string; readonly #temporaryRoot: string; readonly #publishedDigests = new Set(); #state: "open" | "committed" | "aborted" = "open"; constructor(options: Readonly) { const outputPrefix = `${trimTrailingSlash(options.planOutputS3Prefix)}/v2`; - parseS3Uri(outputPrefix); + const { bucket } = parseS3Uri(outputPrefix); this.#s3 = options.s3; - this.artifactPrefix = `${outputPrefix}/artifacts/sha256`; + this.#bucket = bucket; + // Shared by every render, so a blob the SDK or an earlier render uploaded is reused, not re-sent. + this.artifactPrefix = formatS3Uri({ bucket, key: CONTENT_STORE_PREFIX }); this.manifestUri = `${outputPrefix}/manifest.json`; this.#temporaryRoot = options.temporaryRoot ?? tmpdir(); mkdirSync(this.#temporaryRoot, { recursive: true }); @@ -90,8 +100,14 @@ export class S3PlanV2ArtifactPublisher implements PlanV2ArtifactPublisher { `S3 published blob size changed for ${digest}: expected ${blob.sizeBytes}, got ${sourceSize}`, ); } - const uri = `${this.artifactPrefix}/${digest.slice(0, 2)}/${digest}`; - await uploadContentAddressedFileToS3(this.#s3, blob.sourcePath, uri, digest); + await putContentStoreFile(this.#s3, blob.sourcePath, this.#bucket, digest); + this.#publishedDigests.add(digest); + } + + async referenceStoredBlob(blob: Readonly>): Promise { + this.#assertOpen("reference a stored blob"); + const digest = assertSha256(blob.sha256, "S3 referenced blob sha256"); + await ensureContentStoreObject(this.#s3, this.#bucket, digest, blob.sizeBytes); this.#publishedDigests.add(digest); } @@ -125,9 +141,8 @@ export class S3PlanV2ArtifactPublisher implements PlanV2ArtifactPublisher { async abort(): Promise { if (this.#state === "open") this.#state = "aborted"; - // Remote CAS blobs are immutable and may already be reused by a retry. - // Without a committed manifest they are unreachable and expire under the - // render bucket's intermediate-object lifecycle policy. + // Remote CAS blobs are immutable and may already be reused by a retry or + // another render. They expire under the shared content store's lifecycle. } #assertOpen(operation: string): void { diff --git a/packages/aws-lambda/src/s3Transport.test.ts b/packages/aws-lambda/src/s3Transport.test.ts index 6e5ea6dd58..888bb48017 100644 --- a/packages/aws-lambda/src/s3Transport.test.ts +++ b/packages/aws-lambda/src/s3Transport.test.ts @@ -10,13 +10,17 @@ import { createHash } from "node:crypto"; import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import { CONTENT_STORE_REFRESH_AFTER_DAYS, contentStoreUri } from "./contentStore.js"; +import { HeadObjectCommand, S3Client } from "@aws-sdk/client-s3"; import { downloadS3ObjectToFileVerified, formatS3Uri, parseS3Uri, + presignVerifiedS3GetUrl, sha256File, tarDirectory, untarDirectory, + putContentStoreFile, uploadContentAddressedFileToS3, } from "./s3Transport.js"; @@ -176,6 +180,37 @@ describe("content-addressed v2 artifacts", () => { expect(s3.putCount).toBe(0); }); + it("rewrites a reused object at the refresh age so its lifecycle expiry restarts", async () => { + const source = join(scratchRoot, "artifact-aging.bin"); + writeFileSync(source, "aging bytes"); + const digest = await sha256File(source); + const uri = contentStoreUri("bucket", digest); + const s3 = new ContentAddressedFakeS3(); + const days = (n: number) => new Date(Date.now() - n * 86_400_000); + s3.objects.set(uri, { + bytes: Buffer.from("aging bytes"), + sha256: digest, + lastModified: days(CONTENT_STORE_REFRESH_AFTER_DAYS - 1), + }); + + expect(await putContentStoreFile(s3.asClient(), source, "bucket", digest)).toBe("reused"); + expect(s3.copies).toEqual([]); + + s3.objects.get(uri)!.lastModified = days(CONTENT_STORE_REFRESH_AFTER_DAYS); + expect(await putContentStoreFile(s3.asClient(), source, "bucket", digest)).toBe("reused"); + expect(s3.copies).toEqual([ + { + uri, + CopySource: `bucket/cas/sha256/${digest.slice(0, 2)}/${digest}`, + MetadataDirective: "REPLACE", + Metadata: { sha256: digest }, + ChecksumAlgorithm: "SHA256", + }, + ]); + expect(s3.objects.get(uri)).toMatchObject({ sha256: digest }); + expect(s3.putCount).toBe(0); + }); + it("deletes a downloaded artifact when digest verification fails", async () => { const expectedSource = join(scratchRoot, "artifact-expected.bin"); const destination = join(scratchRoot, "artifact-download.bin"); @@ -193,8 +228,57 @@ describe("content-addressed v2 artifacts", () => { }); }); +describe("presignVerifiedS3GetUrl", () => { + const digest = "ab".repeat(32); + const uri = `s3://bucket/v2/artifacts/sha256/ab/${digest}`; + function clientHolding(head: { ChecksumSHA256?: string; ContentLength: number }): S3Client { + const client = new S3Client({ + region: "us-east-2", + credentials: { accessKeyId: "AKIDTEST", secretAccessKey: "test-secret" }, + }); + client.send = (async (command: unknown) => { + expect(command).toBeInstanceOf(HeadObjectCommand); + expect((command as HeadObjectCommand).input.ChecksumMode).toBe("ENABLED"); + return head; + }) as S3Client["send"]; + return client; + } + + it.each([ + { uri, key: `/v2/artifacts/sha256/ab/${digest}` }, + { uri: contentStoreUri("bucket", digest), key: `/cas/sha256/ab/${digest}` }, + ])("signs an attested legacy or shared-content source at $key", async (source) => { + const checksum = Buffer.from(digest, "hex").toString("base64"); + const url = new URL( + await presignVerifiedS3GetUrl( + clientHolding({ ChecksumSHA256: checksum, ContentLength: 42 }), + source.uri, + digest, + 42, + 600, + ), + ); + expect(url.pathname).toBe(source.key); + expect(url.searchParams.get("X-Amz-Expires")).toBe("600"); + }); + + it("refuses an object whose stored checksum is not the pinned digest", async () => { + const other = Buffer.from("cd".repeat(32), "hex").toString("base64"); + await expect( + presignVerifiedS3GetUrl( + clientHolding({ ChecksumSHA256: other, ContentLength: 42 }), + uri, + digest, + 42, + 600, + ), + ).rejects.toMatchObject({ name: "PLAN_ARTIFACT_DIGEST_MISMATCH" }); + }); +}); + class ContentAddressedFakeS3 { - readonly objects = new Map(); + readonly objects = new Map(); + readonly copies: Record[] = []; putCount = 0; raceOnNextPut: { bytes: Buffer; sha256: string } | undefined; @@ -213,10 +297,25 @@ class ContentAddressedFakeS3 { Key: string; Body?: NodeJS.ReadableStream; Metadata?: Record; + CopySource?: string; + MetadataDirective?: string; + ChecksumAlgorithm?: string; ChecksumSHA256?: string; }; }; const uri = `s3://${value.input.Bucket}/${value.input.Key}`; + if (value.constructor.name === "CopyObjectCommand") { + const { CopySource, MetadataDirective, Metadata, ChecksumAlgorithm } = value.input; + this.copies.push({ uri, CopySource, MetadataDirective, Metadata, ChecksumAlgorithm }); + const source = this.objects.get(`s3://${CopySource}`); + if (!source) throw new Error("missing copy source"); + this.objects.set(uri, { + bytes: source.bytes, + sha256: Metadata?.sha256 ?? "", + lastModified: new Date(), + }); + return {}; + } if (value.constructor.name === "HeadObjectCommand") { const object = this.objects.get(uri); if (!object) { @@ -230,6 +329,7 @@ class ContentAddressedFakeS3 { return { ContentLength: object.bytes.length, Metadata: { sha256: object.sha256 }, + LastModified: object.lastModified, }; } if (value.constructor.name === "GetObjectCommand") { diff --git a/packages/aws-lambda/src/s3Transport.ts b/packages/aws-lambda/src/s3Transport.ts index 379fa147ab..3469749b80 100644 --- a/packages/aws-lambda/src/s3Transport.ts +++ b/packages/aws-lambda/src/s3Transport.ts @@ -29,12 +29,15 @@ import { createHash } from "node:crypto"; import { dirname } from "node:path"; import { pipeline } from "node:stream/promises"; import { + CopyObjectCommand, GetObjectCommand, HeadObjectCommand, PutObjectCommand, type S3Client, } from "@aws-sdk/client-s3"; +import { getSignedUrl } from "@aws-sdk/s3-request-presigner"; import * as tar from "tar"; +import { CONTENT_STORE_REFRESH_AFTER_DAYS, contentStoreUri } from "./contentStore.js"; /** Parsed `s3://bucket/key` URI. */ export interface S3Location { @@ -101,6 +104,35 @@ export async function downloadS3ObjectToFileVerified( } } +/** + * Pre-sign a GET for a plan-v2 artifact, after S3 attests the object's stored + * SHA-256 and size are the ones the manifest pins. + */ +export async function presignVerifiedS3GetUrl( + client: S3Client, + uri: string, + expectedSha256: string, + expectedSize: number, + expiresInSeconds: number, +): Promise { + assertSha256(expectedSha256); + const { bucket, key } = parseS3Uri(uri); + const head = await client.send( + new HeadObjectCommand({ Bucket: bucket, Key: key, ChecksumMode: "ENABLED" }), + ); + const expectedChecksum = Buffer.from(expectedSha256, "hex").toString("base64"); + if (head.ChecksumSHA256 !== expectedChecksum || head.ContentLength !== expectedSize) { + const error = new Error( + `[s3Transport] PLAN_ARTIFACT_DIGEST_MISMATCH: ${uri} expected ${expectedSha256} (${expectedSize} bytes), S3 holds checksum ${String(head.ChecksumSHA256)} (${String(head.ContentLength)} bytes)`, + ); + error.name = "PLAN_ARTIFACT_DIGEST_MISMATCH"; + throw error; + } + return getSignedUrl(client, new GetObjectCommand({ Bucket: bucket, Key: key }), { + expiresIn: expiresInSeconds, + }); +} + /** * Upload a local file's contents to an S3 URI using a streaming * `PutObjectCommand`. PutObject's 5 GB cap comfortably exceeds the @@ -142,6 +174,16 @@ export async function uploadContentAddressedFileToS3( expectedSha256: string, contentType?: string, ): Promise<"uploaded" | "reused"> { + return (await putImmutableFile(client, localPath, uri, expectedSha256, contentType)).outcome; +} + +async function putImmutableFile( + client: S3Client, + localPath: string, + uri: string, + expectedSha256: string, + contentType?: string, +): Promise<{ outcome: "uploaded" | "reused"; lastModified?: Date }> { assertSha256(expectedSha256); if (!existsSync(localPath)) { throw new Error(`[s3Transport] upload source missing: ${localPath}`); @@ -149,8 +191,9 @@ export async function uploadContentAddressedFileToS3( const { bucket, key } = parseS3Uri(uri); const size = statSync(localPath).size; const existing = await inspectContentAddressedObject(client, bucket, key, size, expectedSha256); - if (existing === "matching") return "reused"; - if (existing === "conflict") throwImmutableObjectConflict(uri); + if (existing.state === "matching") + return { outcome: "reused", lastModified: existing.lastModified }; + if (existing.state === "conflict") throwImmutableObjectConflict(uri); const body = createReadStream(localPath); try { @@ -169,12 +212,12 @@ export async function uploadContentAddressedFileToS3( IfNoneMatch: "*", }), ); - return "uploaded"; + return { outcome: "uploaded" }; } catch (error) { if (!isS3PreconditionFailed(error)) throw error; const raced = await inspectContentAddressedObject(client, bucket, key, size, expectedSha256); - if (raced === "matching") return "reused"; - if (raced === "conflict") throwImmutableObjectConflict(uri); + if (raced.state === "matching") return { outcome: "reused" }; + if (raced.state === "conflict") throwImmutableObjectConflict(uri); // The winning object was deleted between the conditional failure and // verification. Preserve the service error so the orchestrator may retry. throw error; @@ -186,6 +229,30 @@ export async function uploadContentAddressedFileToS3( } } +export async function mapConcurrent( + values: readonly T[], + concurrency: number, + fn: (value: T) => Promise, +): Promise { + let cursor = 0; + async function worker(): Promise { + while (cursor < values.length) { + const index = cursor++; + await fn(values[index]!); + } + } + const results = await Promise.allSettled( + Array.from({ length: Math.min(concurrency, values.length) }, () => worker()), + ); + const failure = results.find( + (result): result is PromiseRejectedResult => result.status === "rejected", + ); + // Do not reject while sibling workers may still be writing into invocation + // scratch. The caller removes that directory in `finally`; draining the pool + // first prevents late S3 streams from racing cleanup after another GET fails. + if (failure) throw failure.reason; +} + export async function sha256File(path: string): Promise { const hash = createHash("sha256"); for await (const chunk of createReadStream(path)) { @@ -202,7 +269,9 @@ function assertSha256(value: string): void { } } -type ContentAddressedObjectState = "missing" | "matching" | "conflict"; +type ContentAddressedObjectState = + | { state: "missing" | "conflict" } + | { state: "matching"; lastModified?: Date }; async function inspectContentAddressedObject( client: S3Client, @@ -216,14 +285,82 @@ async function inspectContentAddressedObject( new HeadObjectCommand({ Bucket: bucket, Key: key, ChecksumMode: "ENABLED" }), ); return existing.ContentLength === expectedSize && existing.Metadata?.sha256 === expectedSha256 - ? "matching" - : "conflict"; + ? { state: "matching", lastModified: existing.LastModified } + : { state: "conflict" }; } catch (error) { - if (isS3NotFound(error)) return "missing"; + if (isS3NotFound(error)) return { state: "missing" }; throw error; } } +/** Put a file into the shared content store once; reusing an aging object restarts its expiry. */ +export async function putContentStoreFile( + client: S3Client, + localPath: string, + bucket: string, + sha256: string, +): Promise<"uploaded" | "reused"> { + const uri = contentStoreUri(bucket, sha256); + const { outcome, lastModified } = await putImmutableFile(client, localPath, uri, sha256); + if (lastModified) await refreshAgingObject(client, uri, sha256, lastModified); + return outcome; +} + +/** Confirm an object the caller never uploaded is in the content store, restarting an aging one. */ +export async function ensureContentStoreObject( + client: S3Client, + bucket: string, + sha256: string, + sizeBytes: number, +): Promise { + const uri = contentStoreUri(bucket, sha256); + const { key } = parseS3Uri(uri); + const existing = await inspectContentAddressedObject(client, bucket, key, sizeBytes, sha256); + if (existing.state === "matching") { + if (existing.lastModified) await refreshAgingObject(client, uri, sha256, existing.lastModified); + return; + } + if (existing.state === "conflict") throwImmutableObjectConflict(uri); + const error = new Error( + `[s3Transport] PLAN_ARTIFACT_MISSING: ${uri} is not in the content store`, + ); + error.name = "PLAN_ARTIFACT_MISSING"; + throw error; +} + +export async function presignS3GetUrl( + client: S3Client, + uri: string, + expiresInSeconds: number, +): Promise { + const { bucket, key } = parseS3Uri(uri); + return getSignedUrl(client, new GetObjectCommand({ Bucket: bucket, Key: key }), { + expiresIn: expiresInSeconds, + }); +} + +// S3 lifecycle expiry counts from the last write, not the last read, so reuse has to write. +async function refreshAgingObject( + client: S3Client, + uri: string, + sha256: string, + lastModified: Date, +): Promise { + const { bucket, key } = parseS3Uri(uri); + const ageMs = Date.now() - lastModified.getTime(); + if (ageMs < CONTENT_STORE_REFRESH_AFTER_DAYS * 86_400_000) return; + await client.send( + new CopyObjectCommand({ + Bucket: bucket, + Key: key, + CopySource: `${bucket}/${key}`, + MetadataDirective: "REPLACE", + Metadata: { sha256 }, + ChecksumAlgorithm: "SHA256", + }), + ); +} + function throwImmutableObjectConflict(uri: string): never { const error = new Error( `[s3Transport] PLAN_ARTIFACT_DIGEST_MISMATCH: immutable object ${uri} already exists with different digest metadata or size`, @@ -232,7 +369,18 @@ function throwImmutableObjectConflict(uri: string): never { throw error; } -function isS3NotFound(error: unknown): boolean { +export async function s3ObjectExists(client: S3Client, uri: string): Promise { + const { bucket, key } = parseS3Uri(uri); + try { + await client.send(new HeadObjectCommand({ Bucket: bucket, Key: key })); + return true; + } catch (error) { + if (isS3NotFound(error)) return false; + throw error; + } +} + +export function isS3NotFound(error: unknown): boolean { if (!isRecord(error)) return false; const metadata = isRecord(error.$metadata) ? error.$metadata : undefined; return ( @@ -256,12 +404,20 @@ function isRecord(value: unknown): value is Record { * binary — the AWS Lambda Node 22 base image ships a minimal set of * userland tools and does NOT include `tar` in `/usr/bin`. */ -export async function tarDirectory(sourceDir: string, destTarball: string): Promise { +export async function tarDirectory( + sourceDir: string, + destTarball: string, + files?: string[], +): Promise { if (!existsSync(sourceDir) || !statSync(sourceDir).isDirectory()) { throw new Error(`[s3Transport] tar source must be an existing directory: ${sourceDir}`); } mkdirSync(dirname(destTarball), { recursive: true }); - await tar.create({ gzip: true, file: destTarball, cwd: sourceDir }, ["."]); + // `files` (relative to `sourceDir`) packs exactly those paths, with symlinks followed. + await tar.create( + { gzip: true, file: destTarball, cwd: sourceDir, follow: files !== undefined }, + files ?? ["."], + ); } /** diff --git a/packages/aws-lambda/src/sdk/__fixtures__/fakeS3.ts b/packages/aws-lambda/src/sdk/__fixtures__/fakeS3.ts index 8221b9c797..f47b27f4af 100644 --- a/packages/aws-lambda/src/sdk/__fixtures__/fakeS3.ts +++ b/packages/aws-lambda/src/sdk/__fixtures__/fakeS3.ts @@ -8,6 +8,8 @@ * `deploySite`'s `finally` block has rmSync'd it. */ +import { Readable } from "node:stream"; +import { buffer } from "node:stream/consumers"; import type { S3Client } from "@aws-sdk/client-s3"; export interface FakeS3Op { @@ -19,31 +21,49 @@ export interface FakeS3Op { export class FakeS3 { ops: FakeS3Op[] = []; existing = new Set(); + bodies = new Map(); + metadata = new Map | undefined>(); async send(command: unknown): Promise { const cmdName = (command as { constructor: { name: string } }).constructor.name; - const input = (command as { input: { Bucket: string; Key: string } }).input; - if (cmdName === "HeadObjectCommand") { - this.ops.push({ kind: "head", bucket: input.Bucket, key: input.Key }); - if (this.existing.has(`${input.Bucket}/${input.Key}`)) { - return { ContentLength: 1, LastModified: new Date("2026-05-16T00:00:00Z") }; + const input = ( + command as { + input: { Bucket: string; Key: string; Metadata?: Record; Body?: unknown }; } - const err = new Error("Not Found") as Error & { - $metadata: { httpStatusCode: number }; - name: string; - }; - err.name = "NotFound"; - err.$metadata = { httpStatusCode: 404 }; - throw err; - } - if (cmdName === "PutObjectCommand") { - this.ops.push({ kind: "put", bucket: input.Bucket, key: input.Key }); - this.existing.add(`${input.Bucket}/${input.Key}`); - await drainBody((command as { input: { Body: NodeJS.ReadableStream | Buffer } }).input.Body); - return {}; - } - throw new Error(`FakeS3: unexpected command ${cmdName}`); + ).input; + const id = `${input.Bucket}/${input.Key}`; + const handlers: Record Promise> = { + HeadObjectCommand: async () => { + this.ops.push({ kind: "head", bucket: input.Bucket, key: input.Key }); + return this.head(id); + }, + GetObjectCommand: async () => { + if (!this.existing.has(id)) throw notFound("NoSuchKey"); + return { Body: Readable.from([this.bodies.get(id) ?? Buffer.alloc(0)]) }; + }, + PutObjectCommand: async () => { + this.ops.push({ kind: "put", bucket: input.Bucket, key: input.Key }); + this.existing.add(id); + this.metadata.set(id, input.Metadata); + this.bodies.set(id, await drainBody(input.Body as NodeJS.ReadableStream | Buffer)); + return {}; + }, + }; + const handler = handlers[cmdName]; + if (!handler) throw new Error(`FakeS3: unexpected command ${cmdName}`); + return handler(); } + + private head(id: string): unknown { + if (this.existing.has(id)) { + return { ContentLength: this.bodies.get(id)?.length, Metadata: this.metadata.get(id) }; + } + throw notFound("NotFound"); + } +} + +function notFound(name: string): Error { + return Object.assign(new Error(name), { name, $metadata: { httpStatusCode: 404 } }); } /** Convenience cast so `deploySite({ s3: makeFakeS3() })` reads cleanly. */ @@ -58,12 +78,6 @@ export function asS3Client(fake: FakeS3): S3Client { * the workdir — surfacing as an "Unhandled error between tests" * teardown noise that bun's runner attributes to the next test. */ -async function drainBody(body: NodeJS.ReadableStream | Buffer): Promise { - if (Buffer.isBuffer(body)) return; - await new Promise((resolve, reject) => { - body.on("data", () => {}); - body.on("end", () => resolve()); - body.on("close", () => resolve()); - body.on("error", reject); - }); +async function drainBody(body: NodeJS.ReadableStream | Buffer): Promise { + return Buffer.isBuffer(body) ? body : buffer(body); } diff --git a/packages/aws-lambda/src/sdk/deploySite.test.ts b/packages/aws-lambda/src/sdk/deploySite.test.ts index 9863b05187..125d54c15e 100644 --- a/packages/aws-lambda/src/sdk/deploySite.test.ts +++ b/packages/aws-lambda/src/sdk/deploySite.test.ts @@ -1,15 +1,19 @@ import { afterEach, beforeEach, describe, expect, it } from "bun:test"; -import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { mkdirSync, mkdtempSync, rmSync, symlinkSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import { createHash } from "node:crypto"; import type { S3Client } from "@aws-sdk/client-s3"; +import { parseSiteManifest } from "../siteManifest.js"; import { asS3Client, FakeS3 } from "./__fixtures__/fakeS3.js"; import { deploySite } from "./deploySite.js"; let projectDir: string; +let scratch: string; beforeEach(() => { projectDir = mkdtempSync(join(tmpdir(), "hf-deploy-site-test-")); + scratch = mkdtempSync(join(tmpdir(), "hf-deploy-site-scratch-")); mkdirSync(join(projectDir, "assets")); writeFileSync(join(projectDir, "index.html"), "hi"); writeFileSync(join(projectDir, "assets", "style.css"), "body { color: red; }"); @@ -17,10 +21,28 @@ beforeEach(() => { afterEach(() => { rmSync(projectDir, { recursive: true, force: true }); + rmSync(scratch, { recursive: true, force: true }); }); +function sha256(text: string): string { + return createHash("sha256").update(text).digest("hex"); +} + +function storeKey(text: string): string { + const digest = sha256(text); + return `cas/sha256/${digest.slice(0, 2)}/${digest}`; +} + +function sitePaths(s3: FakeS3, siteId: string): string[] { + const manifest = s3.bodies.get(`test-bucket/sites/${siteId}/files.json`)?.toString() ?? ""; + return parseSiteManifest(manifest).map((file) => file.path); +} + +const HTML = "hi"; +const CSS = "body { color: red; }"; + describe("deploySite", () => { - it("uploads the tarball when no matching object exists", async () => { + it("uploads each file once under its content hash, then the site manifest", async () => { const s3 = new FakeS3(); const result = await deploySite({ projectDir, @@ -31,14 +53,33 @@ describe("deploySite", () => { expect(result.uploaded).toBe(true); expect(result.siteId).toMatch(/^[0-9a-f]{16}$/); expect(result.bucketName).toBe("test-bucket"); - expect(result.projectS3Uri).toBe(`s3://test-bucket/sites/${result.siteId}/project.tar.gz`); - expect(result.bytes).toBeGreaterThan(0); - expect(s3.ops).toEqual([ - { kind: "head", bucket: "test-bucket", key: `sites/${result.siteId}/project.tar.gz` }, - { kind: "put", bucket: "test-bucket", key: `sites/${result.siteId}/project.tar.gz` }, + expect(result.projectS3Uri).toBe(`s3://test-bucket/sites/${result.siteId}/files.json`); + expect(result.bytes).toBe(HTML.length + CSS.length); + const puts = s3.ops.filter((op) => op.kind === "put").map((op) => op.key); + expect(puts.slice(0, 2).sort()).toEqual([storeKey(HTML), storeKey(CSS)].sort()); + expect(puts[2]).toBe(`sites/${result.siteId}/files.json`); + expect(s3.bodies.get(`test-bucket/${storeKey(CSS)}`)?.toString()).toBe(CSS); + expect(parseSiteManifest(s3.bodies.get(`test-bucket/${puts[2]}`)?.toString() ?? "")).toEqual([ + { path: "assets/style.css", sha256: sha256(CSS), sizeBytes: CSS.length }, + { path: "index.html", sha256: sha256(HTML), sizeBytes: HTML.length }, ]); }); + it("skips a file another site already stored", async () => { + const s3 = new FakeS3(); + const cssId = `test-bucket/${storeKey(CSS)}`; + s3.existing.add(cssId); + s3.bodies.set(cssId, Buffer.from(CSS)); + s3.metadata.set(cssId, { sha256: sha256(CSS) }); + + await deploySite({ projectDir, bucketName: "test-bucket", s3: asS3Client(s3) }); + + const puts = s3.ops.filter((op) => op.kind === "put").map((op) => op.key); + expect(puts).toHaveLength(2); + expect(puts).not.toContain(storeKey(CSS)); + expect(puts).toContain(storeKey(HTML)); + }); + it("yields a stable siteId across re-runs of the same tree", async () => { const s3a = new FakeS3(); const a = await deploySite({ @@ -73,7 +114,7 @@ describe("deploySite", () => { expect(after.siteId).not.toBe(before.siteId); }); - it("short-circuits on HEAD 200 (skips PUT)", async () => { + it("uploads nothing when the same tree is deployed again", async () => { const s3 = new FakeS3(); const first = await deploySite({ projectDir, @@ -89,9 +130,9 @@ describe("deploySite", () => { expect(second.uploaded).toBe(false); expect(second.siteId).toBe(first.siteId); - // Only one PUT total, plus two HEADs. - expect(s3.ops.filter((op) => op.kind === "put")).toHaveLength(1); - expect(s3.ops.filter((op) => op.kind === "head")).toHaveLength(2); + // Two files plus the manifest, each put once and HEAD-checked on both deploys. + expect(s3.ops.filter((op) => op.kind === "put")).toHaveLength(3); + expect(s3.ops.filter((op) => op.kind === "head")).toHaveLength(6); }); it("honours a caller-supplied siteId", async () => { @@ -103,7 +144,7 @@ describe("deploySite", () => { s3: asS3Client(s3), }); expect(result.siteId).toBe("release-v1.2.3"); - expect(result.projectS3Uri).toBe("s3://test-bucket/sites/release-v1.2.3/project.tar.gz"); + expect(result.projectS3Uri).toBe("s3://test-bucket/sites/release-v1.2.3/files.json"); }); it("propagates non-404 S3 errors", async () => { @@ -145,4 +186,25 @@ describe("deploySite", () => { // node_modules contents shouldn't move the hash. expect(after.siteId).toBe(before.siteId); }); + + it("ships a symlinked project and its symlinked files and folders as content", async () => { + mkdirSync(join(scratch, "media")); + writeFileSync(join(scratch, "media", "clip.txt"), "clip"); + writeFileSync(join(scratch, "logo.txt"), "logo"); + symlinkSync(join(scratch, "media"), join(projectDir, "media")); + symlinkSync(join(scratch, "logo.txt"), join(projectDir, "logo.txt")); + symlinkSync(projectDir, join(scratch, "fx")); + const s3 = new FakeS3(); + const result = await deploySite({ + projectDir: join(scratch, "fx"), + bucketName: "test-bucket", + s3: asS3Client(s3), + }); + expect(sitePaths(s3, result.siteId)).toEqual([ + "assets/style.css", + "index.html", + "logo.txt", + "media/clip.txt", + ]); + }); }); diff --git a/packages/aws-lambda/src/sdk/deploySite.ts b/packages/aws-lambda/src/sdk/deploySite.ts index c325a0b6a1..338b43556a 100644 --- a/packages/aws-lambda/src/sdk/deploySite.ts +++ b/packages/aws-lambda/src/sdk/deploySite.ts @@ -1,23 +1,41 @@ /** - * `deploySite` — upload a project directory to S3 once per content hash - * and return a reusable handle. + * `deploySite` — upload a project directory to S3 and return a reusable handle. * * `renderToLambda` calls this implicitly when no `siteHandle` is passed, - * but exposing it as a standalone verb lets adopters bundle a project - * ahead of time and reuse the handle across many renders without - * re-tarring the project tree on every call. + * but exposing it as a standalone verb lets adopters upload a project + * ahead of time and reuse the handle across many renders. * - * The handle is **content-addressed**: `siteId` is derived from a SHA-256 - * over the project files. Two `deploySite` calls on an unchanged tree - * produce the same `siteId` and `HeadObject`-short-circuit the upload. + * Every file goes into the bucket's shared content store under its SHA-256, + * once: a file any earlier site or render already stored is skipped. The + * site itself is a small `sites//files.json` listing path → hash, + * where `siteId` is derived from a SHA-256 over the project files. */ -import { mkdtempSync, rmSync, statSync } from "node:fs"; +import { mkdtempSync, readFileSync, rmSync, statSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import { HeadObjectCommand, S3Client } from "@aws-sdk/client-s3"; -import { hashProjectDir } from "@hyperframes/producer/distributed"; -import { formatS3Uri, tarDirectory, uploadFileToS3 } from "../s3Transport.js"; +import { S3Client } from "@aws-sdk/client-s3"; +import { CONTENT_STORE_RETENTION_DAYS } from "../contentStore.js"; +import { hashProjectDir, listSiteFiles } from "@hyperframes/producer/distributed"; +import { + downloadS3ObjectToFile, + ensureContentStoreObject, + isS3NotFound, + mapConcurrent, + parseS3Uri, + putContentStoreFile, + s3ObjectExists, + sha256File, + uploadContentAddressedFileToS3, +} from "../s3Transport.js"; +import { + isSiteManifestUri, + parseSiteManifest, + serializeSiteManifest, + siteManifestUri, + siteTarballUri, + type SiteFile, +} from "../siteManifest.js"; /** Options for {@link deploySite}. */ export interface DeploySiteOptions { @@ -43,94 +61,122 @@ export interface SiteHandle { siteId: string; /** Bucket the site landed in. Surfaced separately so callers don't have to re-parse `projectS3Uri`. */ bucketName: string; - /** Full `s3://bucket/sites//project.tar.gz` URI; pass through to `renderToLambda`. */ + /** Full `s3://bucket/sites//files.json` URI; pass through to `renderToLambda`. */ projectS3Uri: string; - /** Tarball size in bytes; useful for "did we actually skip the upload?" assertions. */ + /** Total size of the project's files in bytes. */ bytes: number; - /** ISO timestamp of the most recent upload OR the existing object the short-circuit found. */ + /** ISO timestamp of this `deploySite` call. */ uploadedAt: string; - /** `false` if the object already existed and we skipped the PUT. */ + /** `false` if every file and the site manifest were already in the bucket. */ uploaded: boolean; } +/** Handle for a site an earlier {@link deploySite} uploaded; `bytes` and `uploadedAt` are unknown. */ +export function siteHandleFromId(bucketName: string, siteId: string): SiteHandle { + const projectS3Uri = siteManifestUri(bucketName, siteId); + return { siteId, bucketName, projectS3Uri, bytes: 0, uploadedAt: "", uploaded: false }; +} + /** - * Upload `projectDir` to `s3://bucketName/sites//project.tar.gz`. - * - * Short-circuits when an object with the same key already exists in the - * bucket — `siteId` derives from the project's content hash, so the same - * bytes produce the same key, and re-uploading would be redundant. + * Before a render reuses `site`, re-store any file the content store expired, from `projectDir`. + * Returns the handle to render: a site uploaded before site manifests resolves to its tarball. */ -export async function deploySite(opts: DeploySiteOptions): Promise { - if (!statSync(opts.projectDir).isDirectory()) { - throw new Error(`[deploySite] projectDir is not a directory: ${opts.projectDir}`); - } - - const siteId = opts.siteId ?? hashProjectDir(opts.projectDir); - const key = `sites/${siteId}/project.tar.gz`; - const projectS3Uri = formatS3Uri({ bucket: opts.bucketName, key }); - const s3 = opts.s3 ?? new S3Client({ region: opts.region }); - - // HeadObject short-circuit. Adopters re-rendering the same project on - // a tight inner loop (CI smoke, demo flows) save the tar+gzip+PUT pass - // on every iteration. - const existing = await headObject(s3, opts.bucketName, key); - if (existing) { - return { - siteId, - bucketName: opts.bucketName, - projectS3Uri, - bytes: existing.bytes, - uploadedAt: existing.lastModified, - uploaded: false, - }; - } - - const workdir = mkdtempSync(join(tmpdir(), "hf-deploy-site-")); +export async function ensureSiteStored( + site: SiteHandle, + s3: S3Client, + projectDir?: string, +): Promise { + if (!isSiteManifestUri(site.projectS3Uri)) return site; + const { bucket } = parseS3Uri(site.projectS3Uri); + const workdir = mkdtempSync(join(tmpdir(), "hf-site-check-")); try { - const tarball = join(workdir, "project.tar.gz"); - await tarDirectory(opts.projectDir, tarball); - // Note: tarDirectory packs *everything* under `cwd`. We don't need to - // re-implement the skip list inside the tar pack because the - // producer's plan stage applies the same skip during its copy; the - // archive is slightly bigger than the planDir's compiled/ subtree - // but the cost is bounded by the project's user-authored content. - const size = statSync(tarball).size; - await uploadFileToS3(s3, tarball, projectS3Uri, "application/gzip"); - return { - siteId, - bucketName: opts.bucketName, - projectS3Uri, - bytes: size, - uploadedAt: new Date().toISOString(), - uploaded: true, - }; + const manifestPath = join(workdir, "files.json"); + try { + await downloadS3ObjectToFile(s3, site.projectS3Uri, manifestPath); + } catch (error) { + if (!isS3NotFound(error)) throw error; + return await tarballSite(site, s3, bucket); + } + const files = parseSiteManifest(readFileSync(manifestPath, "utf-8")); + const missing: string[] = []; + await mapConcurrent(files, 16, async ({ path, sha256, sizeBytes }) => { + try { + await ensureContentStoreObject(s3, bucket, sha256, sizeBytes); + } catch (error) { + if ((error as Error).name !== "PLAN_ARTIFACT_MISSING") throw error; + if (!projectDir) return void missing.push(path); + await putContentStoreFile(s3, join(projectDir, path), bucket, sha256); + } + }); + if (missing.length > 0) { + const error = new Error( + `[deploySite] SITE_FILES_MISSING: ${missing.length} of ${files.length} files of site ` + + `${site.siteId} are no longer stored (unused for ${CONTENT_STORE_RETENTION_DAYS} days). ` + + "Run `hyperframes lambda sites create`, or pass projectDir to re-upload them.", + ); + error.name = "SITE_FILES_MISSING"; + throw error; + } + return site; } finally { rmSync(workdir, { recursive: true, force: true }); } } -async function headObject( - s3: S3Client, - bucket: string, - key: string, -): Promise<{ bytes: number; lastModified: string } | null> { +async function tarballSite(site: SiteHandle, s3: S3Client, bucket: string): Promise { + const projectS3Uri = siteTarballUri(bucket, site.siteId); + if (await s3ObjectExists(s3, projectS3Uri)) return { ...site, projectS3Uri }; + const error = new Error( + `[deploySite] SITE_NOT_FOUND: site ${site.siteId} has neither ${site.projectS3Uri} nor ` + + `${projectS3Uri}. Run \`hyperframes lambda sites create\` again and use the new site id.`, + ); + error.name = "SITE_NOT_FOUND"; + throw error; +} + +/** Upload `projectDir`'s files to the content store and write the site manifest. */ +export async function deploySite(opts: DeploySiteOptions): Promise { + const s3 = opts.s3 ?? new S3Client({ region: opts.region }); + if (!statSync(opts.projectDir).isDirectory()) { + throw new Error(`[deploySite] projectDir is not a directory: ${opts.projectDir}`); + } + const files = listSiteFiles(opts.projectDir); + const siteId = opts.siteId ?? hashProjectDir(opts.projectDir, files); + const projectS3Uri = siteManifestUri(opts.bucketName, siteId); + + const stored: SiteFile[] = []; + let uploaded = false; + await mapConcurrent(files, 8, async (path) => { + const localPath = join(opts.projectDir, path); + const sha256 = await sha256File(localPath); + stored.push({ path, sha256, sizeBytes: statSync(localPath).size }); + const outcome = await putContentStoreFile(s3, localPath, opts.bucketName, sha256); + uploaded ||= outcome === "uploaded"; + }); + stored.sort((a, b) => (a.path < b.path ? -1 : 1)); + + const workdir = mkdtempSync(join(tmpdir(), "hf-deploy-site-")); try { - const res = await s3.send(new HeadObjectCommand({ Bucket: bucket, Key: key })); - return { - bytes: typeof res.ContentLength === "number" ? res.ContentLength : 0, - lastModified: - res.LastModified instanceof Date - ? res.LastModified.toISOString() - : new Date().toISOString(), - }; - } catch (err) { - // The SDK throws different error shapes for 404 vs 403 vs network; - // a 404 means "needs upload" and is the most common case. Anything - // else propagates so callers see auth / network failures. - const status = (err as { $metadata?: { httpStatusCode?: number } }).$metadata?.httpStatusCode; - if (status === 404) return null; - const name = (err as { name?: string }).name; - if (name === "NotFound" || name === "NoSuchKey") return null; - throw err; + const manifestPath = join(workdir, "files.json"); + const manifest = serializeSiteManifest(stored); + writeFileSync(manifestPath, manifest); + const outcome = await uploadContentAddressedFileToS3( + s3, + manifestPath, + projectS3Uri, + await sha256File(manifestPath), + "application/json", + ); + uploaded ||= outcome === "uploaded"; + } finally { + rmSync(workdir, { recursive: true, force: true }); } + return { + siteId, + bucketName: opts.bucketName, + projectS3Uri, + bytes: stored.reduce((sum, file) => sum + file.sizeBytes, 0), + uploadedAt: new Date().toISOString(), + uploaded, + }; } diff --git a/packages/aws-lambda/src/sdk/index.ts b/packages/aws-lambda/src/sdk/index.ts index f0e600afae..ad7f842151 100644 --- a/packages/aws-lambda/src/sdk/index.ts +++ b/packages/aws-lambda/src/sdk/index.ts @@ -8,7 +8,12 @@ * AWS-SDK only — safe to load in any Node environment. */ -export { deploySite, type DeploySiteOptions, type SiteHandle } from "./deploySite.js"; +export { + deploySite, + type DeploySiteOptions, + type SiteHandle, + siteHandleFromId, +} from "./deploySite.js"; export { renderToLambda, type RenderHandle, type RenderToLambdaOptions } from "./renderToLambda.js"; export { getRenderProgress, diff --git a/packages/aws-lambda/src/sdk/renderToLambda.test.ts b/packages/aws-lambda/src/sdk/renderToLambda.test.ts index 9e36eeaf15..0b3292938d 100644 --- a/packages/aws-lambda/src/sdk/renderToLambda.test.ts +++ b/packages/aws-lambda/src/sdk/renderToLambda.test.ts @@ -1,11 +1,12 @@ import { afterEach, beforeEach, describe, expect, it } from "bun:test"; -import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import type { SFNClient } from "@aws-sdk/client-sfn"; import type { SerializableDistributedRenderConfig } from "../events.js"; +import { SITE_MANIFEST_INVALID } from "../siteManifest.js"; import { asS3Client, FakeS3 } from "./__fixtures__/fakeS3.js"; -import type { SiteHandle } from "./deploySite.js"; +import { deploySite, type SiteHandle, siteHandleFromId } from "./deploySite.js"; import { renderToLambda } from "./renderToLambda.js"; import { InvalidConfigError } from "./validateConfig.js"; @@ -15,10 +16,16 @@ interface CapturedStart { input: unknown; } +const CURRENT_DEFINITION = JSON.stringify({ + States: { PlanV2: { Retry: [{ ErrorEquals: [SITE_MANIFEST_INVALID], MaxAttempts: 0 }] } }, +}); + class FakeSFN { starts: CapturedStart[] = []; + definition = CURRENT_DEFINITION; async send(command: unknown): Promise { const cmdName = (command as { constructor: { name: string } }).constructor.name; + if (cmdName === "DescribeStateMachineCommand") return { definition: this.definition }; if (cmdName === "StartExecutionCommand") { const input = (command as { input: { stateMachineArn: string; name: string; input: string } }) .input; @@ -76,9 +83,7 @@ describe("renderToLambda", () => { expect(handle.executionArn).toContain("smoke-1"); expect(handle.bucketName).toBe("test-bucket"); expect(handle.outputS3Uri).toBe("s3://test-bucket/renders/smoke-1/output.mp4"); - expect(handle.projectS3Uri).toMatch( - /^s3:\/\/test-bucket\/sites\/[0-9a-f]{16}\/project\.tar\.gz$/, - ); + expect(handle.projectS3Uri).toMatch(/^s3:\/\/test-bucket\/sites\/[0-9a-f]{16}\/files\.json$/); expect(Object.keys(handle)).toEqual([ "renderId", "executionArn", @@ -165,6 +170,102 @@ describe("renderToLambda", () => { expect(s3.existing.size).toBe(0); }); + it("refuses a stack deployed before files.json sites before uploading or starting", async () => { + const sfn = new FakeSFN(); + sfn.definition = JSON.stringify({ States: { PlanV2: { Retry: [] } } }); + const s3 = new FakeS3(); + await expect( + renderToLambda({ + projectDir, + bucketName: "test-bucket", + stateMachineArn: "arn:aws:states:us-east-1:1234:stateMachine:hf", + config: baseConfig, + sfn: asSFNClient(sfn), + s3: asS3Client(s3), + }), + ).rejects.toMatchObject({ + name: "STACK_TOO_OLD", + message: expect.stringContaining("hyperframes lambda deploy"), + }); + expect(s3.ops).toHaveLength(0); + expect(sfn.starts).toHaveLength(0); + }); + + describe("with a site id from before files.json", () => { + const renderSite = (s3: FakeS3, sfn: FakeSFN, siteId: string) => + renderToLambda({ + siteHandle: siteHandleFromId("test-bucket", siteId), + bucketName: "test-bucket", + stateMachineArn: "arn:aws:states:us-east-1:1234:stateMachine:hf", + config: baseConfig, + sfn: asSFNClient(sfn), + s3: asS3Client(s3), + }); + + it("renders its project.tar.gz", async () => { + const s3 = new FakeS3(); + const sfn = new FakeSFN(); + s3.existing.add("test-bucket/sites/oldsite/project.tar.gz"); + const handle = await renderSite(s3, sfn, "oldsite"); + expect(handle.projectS3Uri).toBe("s3://test-bucket/sites/oldsite/project.tar.gz"); + expect(sfn.starts[0]?.input).toMatchObject({ ProjectS3Uri: handle.projectS3Uri }); + }); + + it("names both keys and the fix when neither exists, before starting", async () => { + const s3 = new FakeS3(); + const sfn = new FakeSFN(); + await expect(renderSite(s3, sfn, "gone")).rejects.toMatchObject({ + name: "SITE_NOT_FOUND", + message: expect.stringMatching( + /sites\/gone\/files\.json.*sites\/gone\/project\.tar\.gz.*sites create/, + ), + }); + expect(sfn.starts).toHaveLength(0); + }); + }); + + describe("with a site whose files expired from the content store", () => { + async function deployThenExpire(s3: FakeS3): Promise<{ site: SiteHandle; expiredKey: string }> { + const site = await deploySite({ projectDir, bucketName: "test-bucket", s3: asS3Client(s3) }); + const expiredKey = [...s3.existing].find((id) => id.includes("/cas/sha256/"))!; + s3.existing.delete(expiredKey); + s3.ops.length = 0; + return { site, expiredKey }; + } + const renderWith = (s3: FakeS3, sfn: FakeSFN, site: SiteHandle, dir?: string) => + renderToLambda({ + siteHandle: site, + projectDir: dir, + bucketName: "test-bucket", + stateMachineArn: "arn:aws:states:us-east-1:1234:stateMachine:hf", + config: baseConfig, + sfn: asSFNClient(sfn), + s3: asS3Client(s3), + }); + + it("re-uploads the expired file from the project dir, then renders", async () => { + const s3 = new FakeS3(); + const sfn = new FakeSFN(); + const { site, expiredKey } = await deployThenExpire(s3); + await renderWith(s3, sfn, site, projectDir); + expect( + s3.ops.filter((op) => op.kind === "put").map((op) => `${op.bucket}/${op.key}`), + ).toEqual([expiredKey]); + expect(sfn.starts).toHaveLength(1); + }); + + it("without a project dir, refuses with the missing-file count before starting", async () => { + const s3 = new FakeS3(); + const sfn = new FakeSFN(); + const { site } = await deployThenExpire(s3); + await expect(renderWith(s3, sfn, site)).rejects.toMatchObject({ + name: "SITE_FILES_MISSING", + message: expect.stringContaining("1 of 1 files"), + }); + expect(sfn.starts).toHaveLength(0); + }); + }); + it("rejects invalid configs synchronously before any AWS call", async () => { const sfn = new FakeSFN(); const s3 = new FakeS3(); @@ -277,10 +378,23 @@ describe("renderToLambda", () => { expect(sfn.starts).toHaveLength(0); }); + it("has a troubleshooting entry for every error it raises before starting", () => { + const docs = readFileSync( + new URL("../../../../docs/deploy/aws-lambda.mdx", import.meta.url), + "utf8", + ); + for (const name of ["STACK_TOO_OLD", "SITE_FILES_MISSING", "SITE_NOT_FOUND"]) { + expect({ name, documented: new RegExp(`^### .*\`${name}\``, "m").test(docs) }).toEqual({ + name, + documented: true, + }); + } + }); + it("propagates a missing executionArn as an error", async () => { const sfn = { async send(_cmd: unknown): Promise { - return { executionArn: undefined }; + return { executionArn: undefined, definition: CURRENT_DEFINITION }; }, }; const s3 = new FakeS3(); diff --git a/packages/aws-lambda/src/sdk/renderToLambda.ts b/packages/aws-lambda/src/sdk/renderToLambda.ts index 1232e68761..a0c80ebb65 100644 --- a/packages/aws-lambda/src/sdk/renderToLambda.ts +++ b/packages/aws-lambda/src/sdk/renderToLambda.ts @@ -18,12 +18,13 @@ */ import { randomUUID } from "node:crypto"; -import { SFNClient, StartExecutionCommand } from "@aws-sdk/client-sfn"; -import type { S3Client } from "@aws-sdk/client-s3"; +import { DescribeStateMachineCommand, SFNClient, StartExecutionCommand } from "@aws-sdk/client-sfn"; +import { S3Client } from "@aws-sdk/client-s3"; import type { LambdaPlanProtocol, SerializableDistributedRenderConfig } from "../events.js"; import { formatExtension } from "../formatExtension.js"; import { formatS3Uri } from "../s3Transport.js"; -import { deploySite, type SiteHandle } from "./deploySite.js"; +import { SITE_MANIFEST_INVALID } from "../siteManifest.js"; +import { deploySite, ensureSiteStored, type SiteHandle } from "./deploySite.js"; import { validateDistributedRenderConfig, validateStepFunctionsInputSize, @@ -31,7 +32,7 @@ import { /** Options for {@link renderToLambda}. */ export interface RenderToLambdaOptions { - /** Local project directory. Required when `siteHandle` is not supplied. */ + /** Local project directory. Required without `siteHandle`; with one, re-uploads expired files. */ projectDir?: string; /** Re-use an existing `deploySite` upload (skips tar+S3 PUT). */ siteHandle?: SiteHandle; @@ -92,6 +93,9 @@ export async function renderToLambda(opts: RenderToLambdaOptions): Promise { + const { definition } = await sfn.send(new DescribeStateMachineCommand({ stateMachineArn })); + if (definition?.includes(`"${SITE_MANIFEST_INVALID}"`)) return; + const error = new Error( + `[renderToLambda] STACK_TOO_OLD: ${stateMachineArn} was deployed by an earlier ` + + "@hyperframes/aws-lambda that cannot read this version's sites. Redeploy the stack first " + + "(`hyperframes lambda deploy`, or the SAM template / CDK construct from this version).", + ); + error.name = "STACK_TOO_OLD"; + throw error; +} diff --git a/packages/aws-lambda/src/siteManifest.ts b/packages/aws-lambda/src/siteManifest.ts new file mode 100644 index 0000000000..36e632e2a6 --- /dev/null +++ b/packages/aws-lambda/src/siteManifest.ts @@ -0,0 +1,106 @@ +// A deployed site is `sites//files.json`: the project's files by path, each stored once +// in the shared content store under its SHA-256 (see `contentStoreUri`). + +import { readFileSync } from "node:fs"; +import { extname, join } from "node:path"; +import type { S3Client } from "@aws-sdk/client-s3"; +import type { RemoteProjectFile } from "@hyperframes/producer/distributed"; +import { contentStoreUri } from "./contentStore.js"; +import { + downloadS3ObjectToFile, + downloadS3ObjectToFileVerified, + formatS3Uri, + mapConcurrent, + parseS3Uri, + presignS3GetUrl, +} from "./s3Transport.js"; + +const SITE_MANIFEST_NAME = "files.json"; +/** Stacks that read site manifests list this as a terminal plan error; the SDK checks for it. */ +export const SITE_MANIFEST_INVALID = "SITE_MANIFEST_INVALID"; +const VIDEO_EXTENSIONS_PLANNED_IN_PLACE = new Set([".mp4", ".mov", ".m4v", ".webm", ".mkv"]); +const REMOTE_URL_TTL_SECONDS = 3600; + +export interface SiteFile { + /** `/`-separated, relative to the project root. */ + readonly path: string; + readonly sha256: string; + readonly sizeBytes: number; +} + +export function siteManifestUri(bucket: string, siteId: string): string { + return formatS3Uri({ bucket, key: `sites/${siteId}/${SITE_MANIFEST_NAME}` }); +} + +/** Sites uploaded before site manifests are one tarball; the plan step still reads them. */ +export function siteTarballUri(bucket: string, siteId: string): string { + return formatS3Uri({ bucket, key: `sites/${siteId}/project.tar.gz` }); +} + +export function isSiteManifestUri(uri: string): boolean { + return uri.endsWith(`/${SITE_MANIFEST_NAME}`); +} + +export function serializeSiteManifest(files: readonly SiteFile[]): string { + return JSON.stringify({ files }); +} + +export function parseSiteManifest(text: string): SiteFile[] { + const value = JSON.parse(text) as { files?: unknown }; + if (!Array.isArray(value.files)) throw siteManifestError("files must be an array"); + const seen = new Set(); + return value.files.map((entry: Partial, index) => { + const { path, sha256, sizeBytes } = entry ?? {}; + if (typeof path !== "string" || !isSafeRelativePath(path) || seen.has(path)) { + throw siteManifestError(`files[${index}].path is not a unique relative path`); + } + if (typeof sha256 !== "string" || !/^[a-f0-9]{64}$/.test(sha256)) { + throw siteManifestError(`files[${index}].sha256 is not a lowercase SHA-256 digest`); + } + if (!Number.isSafeInteger(sizeBytes) || (sizeBytes as number) < 0) { + throw siteManifestError(`files[${index}].sizeBytes is not a byte count`); + } + seen.add(path); + return { path, sha256, sizeBytes: sizeBytes as number }; + }); +} + +/** Rebuild the project under `projectDir`, SHA-256 verified; videos can stay remote, as URLs. */ +export async function downloadSiteFiles( + s3: S3Client, + manifestUri: string, + projectDir: string, + leaveVideosRemote: boolean, +): Promise> { + const manifestPath = `${projectDir}.${SITE_MANIFEST_NAME}`; + await downloadS3ObjectToFile(s3, manifestUri, manifestPath); + const files = parseSiteManifest(readFileSync(manifestPath, "utf-8")); + const { bucket } = parseS3Uri(manifestUri); + const remoteFiles: Record = {}; + await mapConcurrent(files, 16, async ({ path, sha256, sizeBytes }) => { + const uri = contentStoreUri(bucket, sha256); + if (leaveVideosRemote && VIDEO_EXTENSIONS_PLANNED_IN_PLACE.has(extname(path).toLowerCase())) { + remoteFiles[path] = { + url: await presignS3GetUrl(s3, uri, REMOTE_URL_TTL_SECONDS), + sha256, + sizeBytes, + }; + return; + } + await downloadS3ObjectToFileVerified(s3, uri, join(projectDir, ...path.split("/")), sha256); + }); + return remoteFiles; +} + +function isSafeRelativePath(path: string): boolean { + return ( + path.split("/").every((segment) => segment !== "" && segment !== "." && segment !== "..") && + !path.includes("\\") + ); +} + +function siteManifestError(detail: string): Error { + const error = new Error(`[siteManifest] ${SITE_MANIFEST_INVALID}: ${detail}`); + error.name = SITE_MANIFEST_INVALID; + return error; +} diff --git a/packages/cli/src/commands/lambda/render-batch.ts b/packages/cli/src/commands/lambda/render-batch.ts index 6296fb6801..c5bff7f846 100644 --- a/packages/cli/src/commands/lambda/render-batch.ts +++ b/packages/cli/src/commands/lambda/render-batch.ts @@ -226,12 +226,12 @@ export async function runRenderBatch(args: RenderBatchArgs): Promise { // Deploy the site once and reuse it across every entry. --site-id and // --dry-run both skip the deploy via a synthesised handle. let siteHandle: SiteHandle; + const { deploySite, siteHandleFromId } = await loadSDK(); if (args.siteId) { - siteHandle = makePlaceholderSiteHandle(args.siteId, stack.bucketName); + siteHandle = siteHandleFromId(stack.bucketName, args.siteId); } else if (args.dryRun) { - siteHandle = makePlaceholderSiteHandle("dry-run-site", stack.bucketName); + siteHandle = siteHandleFromId(stack.bucketName, "dry-run-site"); } else { - const { deploySite } = await loadSDK(); siteHandle = await deploySite({ projectDir, bucketName: stack.bucketName, @@ -275,6 +275,7 @@ export async function runRenderBatch(args: RenderBatchArgs): Promise { } try { const handle = await renderToLambda({ + projectDir, siteHandle, bucketName: stack.bucketName, stateMachineArn: stack.stateMachineArn, @@ -333,23 +334,6 @@ export async function runRenderBatch(args: RenderBatchArgs): Promise { if (failed > 0) setCommandExitCode(1); } -/** - * Synthesise a SiteHandle from a known siteId for paths that skip the - * `deploySite` upload (`--site-id` and `--dry-run`). The SDK reads only - * `siteId` + `projectS3Uri` when `uploaded: false`, so the byte / time - * fields are intentional placeholders. - */ -function makePlaceholderSiteHandle(siteId: string, bucketName: string): SiteHandle { - return { - siteId, - bucketName, - projectS3Uri: `s3://${bucketName}/sites/${siteId}/project.tar.gz`, - bytes: 0, - uploadedAt: "", - uploaded: false, - }; -} - /** * Build the shared wire {@link SerializableDistributedRenderConfig} used * for every entry in a `hyperframes lambda render-batch` run. Extracted for diff --git a/packages/cli/src/commands/lambda/render.ts b/packages/cli/src/commands/lambda/render.ts index f4075ae58b..3d936b5e64 100644 --- a/packages/cli/src/commands/lambda/render.ts +++ b/packages/cli/src/commands/lambda/render.ts @@ -130,24 +130,12 @@ export async function runRender(args: RenderArgs): Promise { stack.reservedConcurrency, ); - // When the caller passes only --site-id, synthesise the minimum-shape - // SiteHandle pointing at the deterministic content-addressed key. The - // `bytes` / `uploadedAt` fields are intentionally placeholders — the - // SDK reads only `siteId` + `projectS3Uri` when `uploaded: false`. - const siteHandle = args.siteId - ? { - siteId: args.siteId, - bucketName: stack.bucketName, - projectS3Uri: `s3://${stack.bucketName}/sites/${args.siteId}/project.tar.gz`, - bytes: 0, - uploadedAt: "", - uploaded: false, - } - : undefined; + // --site-id renders a site an earlier `lambda sites create` uploaded. + const { renderToLambda, siteHandleFromId } = await loadSDK(); + const siteHandle = args.siteId ? siteHandleFromId(stack.bucketName, args.siteId) : undefined; - const { renderToLambda } = await loadSDK(); const handle = await renderToLambda({ - projectDir: siteHandle ? undefined : projectDir, + projectDir, siteHandle, bucketName: stack.bucketName, stateMachineArn: stack.stateMachineArn, diff --git a/packages/cli/src/commands/lambda/sites.ts b/packages/cli/src/commands/lambda/sites.ts index 3f949afa48..f033b1810f 100644 --- a/packages/cli/src/commands/lambda/sites.ts +++ b/packages/cli/src/commands/lambda/sites.ts @@ -1,12 +1,8 @@ /** - * `hyperframes lambda sites create ` — tar + upload a project - * to the deployed render bucket at a content-addressed S3 prefix. Used - * by adopters who want to pre-stage a project so multiple subsequent - * renders share the upload. - * - * Without this verb, every `render` call re-tars the same tree on every - * invocation. With it, the same `siteId` resolves to a `HeadObject`-204 - * short-circuit inside `deploySite`. + * `hyperframes lambda sites create ` — upload a project's files + * to the deployed render bucket's shared content store. Used by adopters + * who want to pre-stage a project so multiple subsequent renders share the + * upload; files already in the bucket are skipped. */ import { resolve as resolvePath } from "node:path"; diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index ee90c67f79..50c3c5bcdf 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -299,7 +299,11 @@ export { quantizeTimeToFrame, type MediaVisualStyleProperty, } from "./inline-scripts/parityContract"; -export { redactKnownPaths, redactTelemetryString } from "./telemetryRedaction"; +export { + redactKnownPaths, + redactTelemetryString, + redactUrlQueryStrings, +} from "./telemetryRedaction"; export { folderGone, isProjectRootMissing, diff --git a/packages/core/src/telemetryRedaction.test.ts b/packages/core/src/telemetryRedaction.test.ts index ee98b4b0df..6a6a9a5daf 100644 --- a/packages/core/src/telemetryRedaction.test.ts +++ b/packages/core/src/telemetryRedaction.test.ts @@ -17,6 +17,14 @@ describe("redactTelemetryString", () => { ).toBe("[path] [path] [path] [path] [file-url] https://example.com/video.mp4?…"); }); + it("redacts a signing parameter cut off from its URL", () => { + expect( + redactTelemetryString( + "z-Signature=supersecret&X-Amz-Credential=AKID x-goog-signature=abc -t 1", + ), + ).toBe("z-Signature=…&X-Amz-Credential=… x-goog-signature=… -t 1"); + }); + // The redactor used to enumerate roots (/Users, /home, /opt, /tmp, …). Any // root outside that list reached telemetry verbatim, which is most of them. it.each([ diff --git a/packages/core/src/telemetryRedaction.ts b/packages/core/src/telemetryRedaction.ts index 423a9d29e9..84b12f90b4 100644 --- a/packages/core/src/telemetryRedaction.ts +++ b/packages/core/src/telemetryRedaction.ts @@ -5,8 +5,11 @@ function truncateTelemetryString(value: string, maxLength: number): string { return `${value.slice(0, maxLength - 1)}…`; } -function redactUrlQueryStrings(value: string): string { - return value.replace(/\b(https?:\/\/[^\s?]+)\?[^\s]*/g, "$1?…"); +/** Strips http(s) URL queries, and any signing parameter left without its URL (e.g. after a cut). */ +export function redactUrlQueryStrings(value: string): string { + return value + .replace(/\b(https?:\/\/[^\s?]+)\?[^\s]*/g, "$1?…") + .replace(/((?:X-(?:Amz|Goog)-[\w-]*|Signature)=)[^&\s]*/gi, "$1…"); } /** diff --git a/packages/engine/src/services/audioMixer.ts b/packages/engine/src/services/audioMixer.ts index e641fd6a6c..e33baaf377 100644 --- a/packages/engine/src/services/audioMixer.ts +++ b/packages/engine/src/services/audioMixer.ts @@ -1112,6 +1112,7 @@ export async function processCompositionAudio( config?: Partial> & { /** Serialised master bus chain, from `parseMasterFxChain`. */ masterFxChain?: string; + urlByAbsentLocalPath?: ReadonlyMap; }, compiledDir?: string, ): Promise { @@ -1166,6 +1167,7 @@ export async function processCompositionAudio( if (!isHttpUrl(srcPath)) { // Same browser-URL path semantics as videos. srcPath = resolveProjectRelativeSrc(element.src, baseDir, compiledDir); + srcPath = config?.urlByAbsentLocalPath?.get(srcPath) ?? srcPath; } if (isHttpUrl(srcPath)) { diff --git a/packages/engine/src/services/videoFrameExtractor.ts b/packages/engine/src/services/videoFrameExtractor.ts index 03038a5f5d..282a6959d4 100644 --- a/packages/engine/src/services/videoFrameExtractor.ts +++ b/packages/engine/src/services/videoFrameExtractor.ts @@ -45,6 +45,7 @@ import { type VideoMetadata, } from "../utils/ffprobe.js"; import { inputAlphaOpaqueWarning, probeInputAlphaPlane } from "../utils/alphaPlaneProbe.js"; +import { redactUrlQueriesInError, urlInputArgs } from "../utils/urlInput.js"; import { analyzeCompositionHdr, isHdrColorSpace as isHdrColorSpaceUtil, @@ -247,6 +248,7 @@ export interface ExtractionOptions { deferRangeExtraction?: boolean; /** Directories the caller rewrites on every render (downloaded media); their sources are cached by content. */ contentKeyedDirs?: readonly string[]; + urlByAbsentLocalPath?: ReadonlyMap; } const DOWNLOAD_SUBDIR = "_downloads"; @@ -815,7 +817,9 @@ export function parseImageElements(html: string): ImageElement[] { /** The first shown frame's colour tags; see {@link firstFrameColourArgs}. */ async function readFirstFrameColour(videoPath: string, signal?: AbortSignal) { - const result = await runFfmpeg(firstFrameColourArgs(videoPath), { signal }); + const result = await runFfmpeg([...urlInputArgs(videoPath), ...firstFrameColourArgs(videoPath)], { + signal, + }); return parseFirstFrameColour(result.stderr); } @@ -936,6 +940,16 @@ async function runSegmentedExtraction(job: SegmentedExtraction): Promise +): Promise { + try { + return await extractVideoFramesRangeUnredacted(...args); + } catch (error) { + throw redactUrlQueriesInError(error); + } +} + +async function extractVideoFramesRangeUnredacted( videoPath: string, videoId: string, startTime: number, @@ -995,7 +1009,7 @@ export async function extractVideoFramesRange( throw new Error("frameRanges extraction needs a constant-frame-rate source"); } - const args: string[] = []; + const args: string[] = urlInputArgs(videoPath); if (decodeWithVideoToolbox) { args.push("-hwaccel", "videotoolbox"); } @@ -1816,6 +1830,11 @@ export async function extractAllVideoFrames( if (!isHttpUrl(videoPath)) { videoPath = resolveProjectRelativeSrc(video.src, baseDir, compiledDir); } + const remoteUrl = options.urlByAbsentLocalPath?.get(videoPath); + if (remoteUrl) { + resolvedVideos.push({ video, videoPath: remoteUrl }); + continue; + } if (isHttpUrl(videoPath)) { const downloadDir = join(options.outputDir, DOWNLOAD_SUBDIR); diff --git a/packages/engine/src/utils/ffprobe.test.ts b/packages/engine/src/utils/ffprobe.test.ts index 33e0cc3797..969028ad2f 100644 --- a/packages/engine/src/utils/ffprobe.test.ts +++ b/packages/engine/src/utils/ffprobe.test.ts @@ -82,6 +82,38 @@ describe("extractMediaMetadata nb_frames duration cross-check", () => { vi.resetModules(); }); + it("never reads a non-PNG video's bytes looking for PNG colour metadata", async () => { + const { spawn } = createSpawnSpy([ + { + kind: "exit", + code: 0, + stdout: JSON.stringify({ + streams: [ + { + codec_type: "video", + codec_name: "h264", + width: 16, + height: 16, + r_frame_rate: "30/1", + avg_frame_rate: "30/1", + }, + ], + format: { duration: "1" }, + }), + }, + ]); + vi.resetModules(); + vi.doMock("child_process", () => ({ spawn })); + const { extractMediaMetadata: mocked } = await import("./ffprobe.js"); + // Bytes that carry PNG cICP, reported by ffprobe as an untagged H.264 stream. + const dir = mkdtempSync(`${tmpdir()}/hf-not-png-`); + const path = `${dir}/clip.mp4`; + writeFileSync(path, buildMinimalPng()); + const meta = await mocked(path); + rmSync(dir, { recursive: true, force: true }); + expect(meta.colorSpace).toBeNull(); + }); + it("uses nb_frames to bound video duration when stream duration is absent and audio is longer", async () => { const { spawn } = createSpawnSpy([ { diff --git a/packages/engine/src/utils/ffprobe.ts b/packages/engine/src/utils/ffprobe.ts index 76eb27acff..7659580761 100644 --- a/packages/engine/src/utils/ffprobe.ts +++ b/packages/engine/src/utils/ffprobe.ts @@ -5,6 +5,7 @@ import { chunkCrc32 } from "./crc32.js"; import { StringDecoder } from "node:string_decoder"; import { basename } from "path"; import { redactTelemetryString } from "@hyperframes/core"; +import { urlInputArgs } from "./urlInput.js"; import { FFPROBE_PATH_ENV, getFfprobeBinary } from "./ffmpegBinaries.js"; import { ManagedChildProcess } from "./managedChildProcess.js"; import { trackChildProcess } from "./processTracker.js"; @@ -65,7 +66,15 @@ async function runFfprobe( } const command = getFfprobeBinary(); - const proc = spawn(command, ["-v", "error", ...argsWithoutInput, "--", filePath], { + const args = [ + "-v", + "error", + ...urlInputArgs(filePath, { tool: "ffprobe" }), + ...argsWithoutInput, + "--", + filePath, + ]; + const proc = spawn(command, args, { // Nothing is ever written to the child's stdin; leaving it as a pipe is // what lets a stdin-reading invocation block indefinitely. stdio: ["ignore", "pipe", "pipe"], @@ -769,7 +778,11 @@ export async function extractMediaMetadata(filePath: string): Promise { }); }); + it("keeps no part of a signed URL's query when the tail cut lands inside it", async () => { + const child = childProcess(); + const managed = new ManagedChildProcess(child, { stderrMaxBytes: 40 }); + child.stderr.emit( + "data", + Buffer.from("Error opening https://b.s3.amazonaws.com/v.mp4?X-Amz-Sig"), + ); + child.stderr.emit( + "data", + Buffer.from("nature=supersecret&X-Amz-Credential=AKIDEXAMPLE -t 0.5"), + ); + child.emit("close", 1, null); + + const { stderr } = await managed.wait(); + expect(stderr).toBe(" -t 0.5"); + for (const leaked of ["secret", "AKID"]) expect(stderr).not.toContain(leaked); + }); + + it.each([ + ["inside the signature value", 66, 64], + ["right after the query mark", 48, 64], + ["after a cut severed the URL's head", 66, 30], + ])("keeps no part of a signature split across chunks %s", async (_where, at, maxBytes) => { + const child = childProcess(); + const managed = new ManagedChildProcess(child, { stderrMaxBytes: maxBytes }); + const line = + "Error opening https://b.s3.amazonaws.com/v.mp4?X-Amz-Signature=abcdef1234 -t 0.5\n"; + child.stderr.emit("data", Buffer.from(line.slice(0, at))); + child.stderr.emit("data", Buffer.from(line.slice(at))); + child.emit("close", 1, null); + + const { stderr } = await managed.wait(); + expect(stderr).toContain(" -t 0.5"); + for (const leaked of ["abc", "def1234", "1234"]) expect(stderr).not.toContain(leaked); + }); + + it("keeps no signature fragment when the kept tail itself starts inside one", async () => { + const child = childProcess(); + const managed = new ManagedChildProcess(child, { stderrMaxBytes: 16 }); + const signature = `${"a".repeat(30)}0123456789`; + child.stderr.emit( + "data", + Buffer.from(`x https://b.s3.amazonaws.com/v.mp4?X-Amz-Signature=${signature} -t 0.5\n`), + ); + child.emit("close", 1, null); + + const { stderr } = await managed.wait(); + expect(stderr).toContain(" -t 0.5"); + for (const leaked of ["aaa", "6789"]) expect(stderr).not.toContain(leaked); + }); + it("escalates abort from SIGTERM to SIGKILL and resolves only after close", async () => { vi.useFakeTimers(); const child = childProcess(); diff --git a/packages/engine/src/utils/managedChildProcess.ts b/packages/engine/src/utils/managedChildProcess.ts index 3778e412d6..6963027b9f 100644 --- a/packages/engine/src/utils/managedChildProcess.ts +++ b/packages/engine/src/utils/managedChildProcess.ts @@ -1,4 +1,6 @@ import type { ChildProcess } from "node:child_process"; +import { StringDecoder } from "node:string_decoder"; +import { redactUrlQueryStrings } from "@hyperframes/core"; export type ManagedProcessTerminationReason = | "exit" @@ -28,6 +30,7 @@ export interface ManagedChildProcessOptions { const DEFAULT_TERMINATION_GRACE_MS = 2_000; const DEFAULT_STDERR_MAX_BYTES = 64 * 1024; +const RAW_STDERR_SLACK_FACTOR = 2; /** Owns cancellation, escalation, stderr and reaping for one child process. */ export class ManagedChildProcess { @@ -37,7 +40,9 @@ export class ManagedChildProcess { private resolveOutcome!: (outcome: ManagedChildProcessOutcome) => void; private requestedReason: Exclude | null = null; - private stderrTail = Buffer.alloc(0); + private rawStderrTail = ""; + private rawStderrTailCut = false; + private readonly stderrDecoder = new StringDecoder("utf8"); private spawned = false; private settled = false; private deadlineTimer: NodeJS.Timeout | null = null; @@ -83,14 +88,23 @@ export class ManagedChildProcess { private readonly onStderr = (data: Buffer | string): void => { const chunk = Buffer.isBuffer(data) ? data : Buffer.from(data); - const maxBytes = this.options.stderrMaxBytes ?? DEFAULT_STDERR_MAX_BYTES; - this.stderrTail = Buffer.concat([this.stderrTail, chunk]); - if (this.stderrTail.byteLength > maxBytes) { - this.stderrTail = this.stderrTail.subarray(this.stderrTail.byteLength - maxBytes); - } + const joined = this.rawStderrTail + this.stderrDecoder.write(chunk); + this.rawStderrTail = lastBytes(joined, RAW_STDERR_SLACK_FACTOR * this.maxBytes); + if (this.rawStderrTail !== joined) this.rawStderrTailCut = true; this.options.onStderr?.(chunk.toString()); }; + private get maxBytes(): number { + return this.options.stderrMaxBytes ?? DEFAULT_STDERR_MAX_BYTES; + } + + private redactedStderrTail(): string { + const raw = this.rawStderrTailCut + ? withoutTokenSeveredByCut(this.rawStderrTail) + : this.rawStderrTail; + return lastBytes(redactUrlQueryStrings(raw), this.maxBytes); + } + private readonly onSpawn = (): void => { this.spawned = true; }; @@ -100,7 +114,7 @@ export class ManagedChildProcess { reason: this.requestedReason ?? "exit", exitCode, signal, - stderr: this.stderrTail.toString(), + stderr: this.redactedStderrTail(), durationMs: this.now() - this.startedAtMs, }); }; @@ -111,7 +125,7 @@ export class ManagedChildProcess { reason: "spawn_error", exitCode: null, signal: null, - stderr: this.stderrTail.length > 0 ? this.stderrTail.toString() : error.message, + stderr: this.rawStderrTail.length > 0 ? this.redactedStderrTail() : error.message, durationMs: this.now() - this.startedAtMs, error, }); @@ -184,3 +198,14 @@ export class ManagedChildProcess { this.escalationTimer = null; } } + +function lastBytes(text: string, maxBytes: number): string { + return Buffer.byteLength(text) > maxBytes + ? Buffer.from(text).subarray(-maxBytes).toString() + : text; +} + +function withoutTokenSeveredByCut(text: string): string { + const firstSpace = text.search(/\s/); + return firstSpace < 0 ? "" : text.slice(firstSpace); +} diff --git a/packages/engine/src/utils/urlInput.test.ts b/packages/engine/src/utils/urlInput.test.ts new file mode 100644 index 0000000000..c51711420b --- /dev/null +++ b/packages/engine/src/utils/urlInput.test.ts @@ -0,0 +1,62 @@ +import { describe, expect, it } from "vitest"; +import { VideoSourceExtractionError } from "../services/videoFrameExtractor.js"; +import { redactUrlQueriesInError, urlInputArgs } from "./urlInput.js"; + +const signed = + "https://bucket.s3.us-east-2.amazonaws.com/a/b?X-Amz-Credential=AKID&X-Amz-Signature=deadbeef"; + +describe("urlInputArgs", () => { + const caBundles = ["/nonexistent.crt", import.meta.filename]; + const verifyAndReconnect = [ + "-tls_verify", + "1", + "-ca_file", + import.meta.filename, + "-rw_timeout", + "30000000", + "-reconnect", + "1", + "-reconnect_delay_max", + "2", + ]; + + it("makes ffmpeg verify the certificate of an https input and reconnect on network errors", () => { + expect(urlInputArgs(signed, { caBundles })).toEqual([ + ...verifyAndReconnect, + "-reconnect_on_network_error", + "1", + ]); + }); + + it("gives ffprobe only the options its 4.0 build accepts", () => { + expect(urlInputArgs(signed, { tool: "ffprobe", caBundles })).toEqual(verifyAndReconnect); + }); + + it("adds nothing for local paths and plain http", () => { + expect(urlInputArgs("/plan/compiled/bg.mp4", { caBundles: [] })).toEqual([]); + expect(urlInputArgs("http://127.0.0.1:8080/bg.mp4", { caBundles: [] })).toEqual([]); + }); + + it("refuses an https input when no CA bundle exists", () => { + expect(() => urlInputArgs(signed, { caBundles: ["/nonexistent.crt"] })).toThrow(/no CA bundle/); + }); +}); + +describe("redactUrlQueriesInError", () => { + it("strips a signed URL's query from the message, diagnostic and stack", () => { + const error = redactUrlQueriesInError( + new VideoSourceExtractionError("ffmpeg_failed", false, `read ${signed}`, `${signed}: 403`), + ); + for (const text of [error.message, error.diagnostic, error.stack]) { + expect(text).toContain("https://bucket.s3.us-east-2.amazonaws.com/a/b?"); + expect(text).not.toContain("deadbeef"); + } + }); + + it("also strips it from a wrapped cause", () => { + const error = redactUrlQueriesInError( + new Error("render failed", { cause: new Error(`read ${signed}`) }), + ); + expect(String((error.cause as Error).message)).not.toContain("deadbeef"); + }); +}); diff --git a/packages/engine/src/utils/urlInput.ts b/packages/engine/src/utils/urlInput.ts new file mode 100644 index 0000000000..32cff701da --- /dev/null +++ b/packages/engine/src/utils/urlInput.ts @@ -0,0 +1,42 @@ +import { existsSync } from "node:fs"; +import { redactUrlQueryStrings } from "@hyperframes/core"; + +// The Lambda nodejs image (Amazon Linux 2023) ships the first; Debian/Ubuntu the second; macOS the third. +const CA_BUNDLES = [ + "/etc/pki/tls/certs/ca-bundle.crt", + "/etc/ssl/certs/ca-certificates.crt", + "/etc/ssl/cert.pem", +]; + +// Lambda's ffprobe 4.0 accepts these but not -reconnect_on_network_error; a stalled read fails after 30 s. +const RECONNECT_ARGS = ["-rw_timeout", "30000000", "-reconnect", "1", "-reconnect_delay_max", "2"]; + +/** ffmpeg and ffprobe skip TLS certificate checks unless told to; an https input must be verified. */ +export function urlInputArgs( + input: string, + { + tool = "ffmpeg", + caBundles = CA_BUNDLES, + }: { tool?: "ffmpeg" | "ffprobe"; caBundles?: readonly string[] } = {}, +): string[] { + if (!/^https:\/\//i.test(input)) return []; + const caFile = caBundles.find((path) => existsSync(path)); + if (!caFile) { + throw new Error( + `[FFmpeg] cannot verify TLS for an https input: no CA bundle at ${caBundles.join(", ")}`, + ); + } + const networkRetry = tool === "ffmpeg" ? ["-reconnect_on_network_error", "1"] : []; + return ["-tls_verify", "1", "-ca_file", caFile, ...RECONNECT_ARGS, ...networkRetry]; +} + +/** A signed URL grants access until it expires: strip query strings from every string field of an error and its causes. */ +export function redactUrlQueriesInError(error: T): T { + if (!(error instanceof Error)) return error; + const fields = error as unknown as Record; + for (const key of ["message", "stack", ...Object.keys(error)]) { + if (typeof fields[key] === "string") fields[key] = redactUrlQueryStrings(fields[key]); + } + if (error.cause !== error) redactUrlQueriesInError(error.cause); + return error; +} diff --git a/packages/gcp-cloud-run/src/gcsTransport.ts b/packages/gcp-cloud-run/src/gcsTransport.ts index 4147ad09ae..32055ca568 100644 --- a/packages/gcp-cloud-run/src/gcsTransport.ts +++ b/packages/gcp-cloud-run/src/gcsTransport.ts @@ -241,12 +241,20 @@ function isGcsPreconditionFailed(error: unknown): boolean { * package (pure JS over `node:zlib`) rather than spawning a system tar * binary so the archive format is independent of the container's userland. */ -export async function tarDirectory(sourceDir: string, destTarball: string): Promise { +export async function tarDirectory( + sourceDir: string, + destTarball: string, + files?: string[], +): Promise { if (!existsSync(sourceDir) || !statSync(sourceDir).isDirectory()) { throw new Error(`[gcsTransport] tar source must be an existing directory: ${sourceDir}`); } mkdirSync(dirname(destTarball), { recursive: true }); - await tar.create({ gzip: true, file: destTarball, cwd: sourceDir }, ["."]); + // `files` (relative to `sourceDir`) packs exactly those paths, with symlinks followed. + await tar.create( + { gzip: true, file: destTarball, cwd: sourceDir, follow: files !== undefined }, + files ?? ["."], + ); } /** diff --git a/packages/gcp-cloud-run/src/sdk/deploySite.test.ts b/packages/gcp-cloud-run/src/sdk/deploySite.test.ts index 6c631ef6c9..b7156b421e 100644 --- a/packages/gcp-cloud-run/src/sdk/deploySite.test.ts +++ b/packages/gcp-cloud-run/src/sdk/deploySite.test.ts @@ -4,9 +4,10 @@ */ import { afterEach, describe, expect, it } from "bun:test"; -import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { mkdtempSync, rmSync, symlinkSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import * as tar from "tar"; import { asStorage, FakeGcs } from "../__fixtures__/fakeGcs.js"; import { deploySite } from "./deploySite.js"; @@ -83,4 +84,24 @@ describe("deploySite", () => { expect(handle.siteId).toBe("my-git-sha"); expect(handle.projectGcsUri).toBe("gs://b/sites/my-git-sha/project.tar.gz"); }); + + it("packs a symlinked project and its symlinked files as content", async () => { + const gcs = new FakeGcs(); + const scratch = mkdtempSync(join(tmpdir(), "hf-site-scratch-")); + tmpDirs.push(scratch); + const real = mkProject("linked"); + writeFileSync(join(scratch, "logo.txt"), "logo"); + symlinkSync(join(scratch, "logo.txt"), join(real, "logo.txt")); + symlinkSync(real, join(scratch, "fx")); + const handle = await deploySite({ + projectDir: join(scratch, "fx"), + bucketName: "b", + storage: asStorage(gcs), + }); + const file = join(scratch, "uploaded.tar.gz"); + writeFileSync(file, gcs.objects.get(handle.projectGcsUri) ?? ""); + const entries: string[] = []; + await tar.list({ file, onReadEntry: (entry) => entries.push(`${entry.type} ${entry.path}`) }); + expect(entries).toEqual(["File index.html", "File logo.txt"]); + }); }); diff --git a/packages/gcp-cloud-run/src/sdk/deploySite.ts b/packages/gcp-cloud-run/src/sdk/deploySite.ts index dfd9f63ab6..75e0aab3f3 100644 --- a/packages/gcp-cloud-run/src/sdk/deploySite.ts +++ b/packages/gcp-cloud-run/src/sdk/deploySite.ts @@ -17,7 +17,7 @@ import { mkdtempSync, rmSync, statSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { Storage } from "@google-cloud/storage"; -import { hashProjectDir } from "@hyperframes/producer/distributed"; +import { hashProjectDir, listSiteFiles } from "@hyperframes/producer/distributed"; import { formatGcsUri, tarDirectory, uploadFileToGcs } from "../gcsTransport.js"; /** Options for {@link deploySite}. */ @@ -65,7 +65,8 @@ export async function deploySite(opts: DeploySiteOptions): Promise { throw new Error(`[deploySite] projectDir is not a directory: ${opts.projectDir}`); } - const siteId = opts.siteId ?? hashProjectDir(opts.projectDir); + const files = listSiteFiles(opts.projectDir); + const siteId = opts.siteId ?? hashProjectDir(opts.projectDir, files); const key = `sites/${siteId}/project.tar.gz`; const projectGcsUri = formatGcsUri({ bucket: opts.bucketName, key }); const storage = opts.storage ?? new Storage(); @@ -89,7 +90,7 @@ export async function deploySite(opts: DeploySiteOptions): Promise { const workdir = mkdtempSync(join(tmpdir(), "hf-deploy-site-")); try { const tarball = join(workdir, "project.tar.gz"); - await tarDirectory(opts.projectDir, tarball); + await tarDirectory(opts.projectDir, tarball, files); const size = statSync(tarball).size; await uploadFileToGcs(storage, tarball, projectGcsUri, "application/gzip"); return { diff --git a/packages/producer/scripts/test-classification.mjs b/packages/producer/scripts/test-classification.mjs index 3bf6f6b093..cf3937ba88 100644 --- a/packages/producer/scripts/test-classification.mjs +++ b/packages/producer/scripts/test-classification.mjs @@ -21,6 +21,7 @@ const INTEGRATION_TEST_FILES = new Set([ "src/services/distributed/plan.masterBus.test.ts", "src/services/distributed/plan.test.ts", "src/services/distributed/planSizeCap.test.ts", + "src/services/distributed/planV2RemoteFiles.test.ts", "src/services/distributed/renderChunk.test.ts", "src/services/fileServer.test.ts", "src/services/fileServer.file-race.test.ts", diff --git a/packages/producer/src/distributed.ts b/packages/producer/src/distributed.ts index 2abb295a93..98b5f79530 100644 --- a/packages/producer/src/distributed.ts +++ b/packages/producer/src/distributed.ts @@ -41,6 +41,7 @@ export { // Types type DistributedRenderConfig, type PlanResult, + type RemoteProjectFile, // Constants DEFAULT_CHUNK_SIZE, DEFAULT_MAX_PARALLEL_CHUNKS, @@ -61,6 +62,7 @@ export { createPlanV2FromExecutionPlan, createPlanV2FromV1, getPlanV2ExecutionPlanHash, + isPlanV2DeferredSourceArtifact, listPlanV2ArtifactsForTarget, materializePlanV2Target, planV2, @@ -75,8 +77,10 @@ export { type PlanV2Artifact, type PlanV2Limitations, type PlanV2Manifest, + type PlanV2MaterializeOptions, type PlanV2MaterializationResult, type PlanV2MaterializationTarget, + type PlanV2RemoteSources, type PlanV2Result, type PlanV2WithPublisherOptions, } from "./services/distributed/planV2.js"; @@ -128,7 +132,7 @@ export { validateDistributedRenderConfig, validateVariablesPayload, } from "./services/distributed/renderConfigValidation.js"; -export { hashProjectDir } from "./services/distributed/projectHash.js"; +export { hashProjectDir, listSiteFiles } from "./services/distributed/projectHash.js"; // ── Plan protocol compatibility ──────────────────────────────────────────── // Workers validate this descriptor before consuming layout-specific diff --git a/packages/producer/src/plan-parity-analysis.ts b/packages/producer/src/plan-parity-analysis.ts index bc15285ca8..b8061bbb82 100644 --- a/packages/producer/src/plan-parity-analysis.ts +++ b/packages/producer/src/plan-parity-analysis.ts @@ -2,6 +2,7 @@ import { createHash } from "node:crypto"; import { createReadStream, readdirSync, statSync, type Dirent } from "node:fs"; import { relative, resolve } from "node:path"; import { spawn, spawnSync } from "node:child_process"; +import { ManagedChildProcess } from "@hyperframes/engine"; import type { PlanParityDriverResult, PlanParityMediaMeasurement, @@ -199,60 +200,45 @@ async function hashPath(path: string): Promise<{ sha256: string; bytes: number } async function canonicalPcmAudio( outputPath: string, ): Promise { - return new Promise((resolvePromise, reject) => { - const child = spawn( - "ffmpeg", - [ - "-hide_banner", - "-loglevel", - "error", - "-i", - outputPath, - "-map", - "0:a:0", - "-vn", - "-ac", - "2", - "-ar", - "48000", - "-f", - "s16le", - "-", - ], - { stdio: ["ignore", "pipe", "pipe"], windowsHide: true }, - ); - const hash = createHash("sha256"); - const stderr: Buffer[] = []; - let bytes = 0; - child.stdout.on("data", (chunk: Buffer) => { - hash.update(chunk); - bytes += chunk.length; - }); - child.stderr.on("data", (chunk: Buffer) => stderr.push(chunk)); - child.on("error", reject); - child.on("close", (code) => { - if (code !== 0) { - const message = Buffer.concat(stderr).toString("utf-8"); - // A video-only fixture is a valid parity input. ffmpeg uses this - // wording when the optional audio map has no match. - if (/matches no streams|does not contain any stream/iu.test(message)) { - resolvePromise(null); - return; - } - reject(new Error(`ffmpeg PCM decode exited ${code}: ${message.trim()}`)); - return; - } - if (bytes % 4 !== 0) { - reject(new Error(`canonical stereo s16le byte count ${bytes} is not divisible by 4`)); - return; - } - resolvePromise({ - sha256: hash.digest("hex"), - sampleCount: bytes / 4, - bytes, - }); - }); + const child = spawn( + "ffmpeg", + [ + "-hide_banner", + "-loglevel", + "error", + "-i", + outputPath, + "-map", + "0:a:0", + "-vn", + "-ac", + "2", + "-ar", + "48000", + "-f", + "s16le", + "-", + ], + { stdio: ["ignore", "pipe", "pipe"], windowsHide: true }, + ); + const hash = createHash("sha256"); + let bytes = 0; + child.stdout.on("data", (chunk: Buffer) => { + hash.update(chunk); + bytes += chunk.length; }); + const outcome = await new ManagedChildProcess(child).wait(); + if (outcome.reason === "spawn_error") throw outcome.error; + if (outcome.exitCode !== 0) { + // A video-only fixture is a valid parity input. ffmpeg uses this + // wording when the optional audio map has no match. + if (/matches no streams|does not contain any stream/iu.test(outcome.stderr)) return null; + throw new Error(`ffmpeg PCM decode exited ${outcome.exitCode}: ${outcome.stderr.trim()}`); + } + if (bytes % 4 !== 0) { + throw new Error(`canonical stereo s16le byte count ${bytes} is not divisible by 4`); + } + return { sha256: hash.digest("hex"), sampleCount: bytes / 4, bytes }; } export async function analyzePlanParityDriverResult( diff --git a/packages/producer/src/regression-harness-lambda-local.ts b/packages/producer/src/regression-harness-lambda-local.ts index 84dd83e553..c475fc704e 100644 --- a/packages/producer/src/regression-harness-lambda-local.ts +++ b/packages/producer/src/regression-harness-lambda-local.ts @@ -46,6 +46,7 @@ import type { RenderChunkLambdaResult, SerializableDistributedRenderConfig, } from "@hyperframes/aws-lambda"; +import { listSiteFiles } from "./services/distributed/projectHash.js"; export type { LambdaLocalRenderResult, @@ -79,7 +80,7 @@ export async function runLambdaLocalRender( const projectKey = `sites/harness/${Date.now()}/project.tar.gz`; const projectS3Path = join(s3Root, projectKey); mkdirSync(dirname(projectS3Path), { recursive: true }); - await tarDirectory(input.projectDir, projectS3Path); + await tarDirectory(input.projectDir, projectS3Path, listSiteFiles(input.projectDir)); const fakeS3 = new FilesystemBackedFakeS3(s3Root); const deps: HandlerDeps = { diff --git a/packages/producer/src/services/audioExtractor.ts b/packages/producer/src/services/audioExtractor.ts deleted file mode 100644 index 42208f1a11..0000000000 --- a/packages/producer/src/services/audioExtractor.ts +++ /dev/null @@ -1,324 +0,0 @@ -// fallow-ignore-file unused-file code-duplication complexity -/** - * Audio Extractor Service - * - * Extracts audio from media elements in the composition HTML, - * applies timeline positioning, and mixes into a single audio track. - */ - -import { spawn } from "node:child_process"; -import { existsSync, mkdirSync, rmSync, readFileSync } from "node:fs"; -import { join, dirname } from "node:path"; -import { getFfmpegBinary, trackChildProcess } from "@hyperframes/engine"; - -export interface AudioElement { - id: string; - src: string; - start: number; - duration: number; - mediaStart: number; - volume: number; - tagName: "audio" | "video"; -} - -export interface AudioTrack { - id: string; - srcPath: string; - start: number; - duration: number; - mediaStart: number; - volume: number; -} - -/** - * Parse audio/video elements from HTML to find media with audio. - */ -export function parseAudioElements(html: string): AudioElement[] { - const elements: AudioElement[] = []; - - // Match