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