diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f0f568d..bdf5dbe 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -41,6 +41,13 @@ jobs: - run: CELLD_PACKAGE_INTEGRATION=1 bun run test:package:prebuilt - run: bun run test:celld - run: NATIVE_WORKER_CELLD=1 bun test scripts/native-worker cloudflare/native-worker.test.ts --timeout 180000 + - name: Test native Worker apps on a two-node celld fleet + run: | + docker run --rm -d --name artifacts-fleet-test-s3 -p 127.0.0.1:5117:9000 \ + -e MINIO_ROOT_USER=artifacts-fixture -e MINIO_ROOT_PASSWORD=artifacts-fixture-secret \ + cgr.dev/chainguard/minio:latest server /data --address :9000 + trap 'docker stop artifacts-fleet-test-s3' EXIT + NATIVE_WORKER_FLEET=1 bun test scripts/native-worker/fleet.test.ts --timeout 240000 executable: strategy: @@ -97,11 +104,31 @@ jobs: env: ARTIFACTS_DOCKER_IMAGE: artifacts:test run: bun test scripts/docker.integration.test.ts --timeout 240000 + - name: Build fleet publisher + uses: docker/build-push-action@v6 + with: + context: . + target: fleet-operator + platforms: ${{ matrix.platform }} + load: true + tags: artifacts:fleet-test + cache-from: type=gha,scope=docker-${{ matrix.arch }} + - name: Test authenticated host release command + env: + FLEET_OPERATOR_DOCKER_IMAGE: artifacts:fleet-test + run: bun test scripts/native-worker/fleet-image.test.ts --timeout 30000 docker-publish: needs: [check, executable, docker-test] if: github.repository == 'sidequery/artifacts' && github.event_name != 'pull_request' && (github.ref == 'refs/heads/main' || startsWith(github.ref, 'refs/tags/v')) runs-on: ubuntu-24.04 + strategy: + matrix: + include: + - target: runtime + prefix: "" + - target: fleet-operator + prefix: fleet- timeout-minutes: 30 permissions: contents: read @@ -121,6 +148,7 @@ jobs: id: metadata with: images: ghcr.io/sidequery/artifacts + flavor: prefix=${{ matrix.prefix }} tags: | type=raw,value=latest,enable={{is_default_branch}} type=sha,format=long @@ -128,6 +156,7 @@ jobs: - uses: docker/build-push-action@v6 with: context: . + target: ${{ matrix.target }} platforms: linux/amd64,linux/arm64 push: true tags: ${{ steps.metadata.outputs.tags }} @@ -137,7 +166,7 @@ jobs: type=gha,scope=docker-arm64 - name: Verify anonymous access env: - IMAGE: ghcr.io/sidequery/artifacts:sha-${{ github.sha }} + IMAGE: ghcr.io/sidequery/artifacts:${{ matrix.prefix }}sha-${{ github.sha }} run: | mkdir -p "$RUNNER_TEMP/public-docker" docker --config "$RUNNER_TEMP/public-docker" manifest inspect "$IMAGE" > /dev/null || { diff --git a/Dockerfile b/Dockerfile index 477b492..afbb3da 100644 --- a/Dockerfile +++ b/Dockerfile @@ -6,9 +6,23 @@ COPY patches ./patches RUN bun install --frozen-lockfile COPY . . RUN bun run build:cloudflare --minify && bun run scripts/prepare-container.ts \ + && bun build scripts/native-worker/fleet.ts --target=bun --outfile=dist/container/fleet.js \ && find dist/worker-app -name '*.map' -delete \ && mkdir -p /state +FROM oven/bun:1.4.0 AS fleet-operator +WORKDIR /app +COPY --from=build /src/dist/container/bin/ /usr/local/bin/ +COPY --from=build /src/dist/container/fleet.js /app/fleet.js +COPY --from=build /src/dist/worker-app/ /app/dist/worker-app/ +COPY --from=build /src/dist/cloudflare/assets/ /app/dist/cloudflare/assets/ +COPY --from=build /src/dist/container/wrangler.jsonc /app/wrangler.jsonc +ENV CELLD_BIN=/usr/local/bin/celld CELLD_ESBUILD=/usr/local/bin/esbuild +USER 65532:65532 +EXPOSE 4792 +STOPSIGNAL SIGTERM +ENTRYPOINT ["bun", "/app/fleet.js"] + FROM gcr.io/distroless/cc-debian13:nonroot AS runtime LABEL org.opencontainers.image.source="https://github.com/sidequery/artifacts" \ org.opencontainers.image.description="Sidequery Artifacts with celld" \ diff --git a/bun.lock b/bun.lock index 2c2973a..0308add 100644 --- a/bun.lock +++ b/bun.lock @@ -4,6 +4,7 @@ "workspaces": { "": { "dependencies": { + "@aws-sdk/client-s3": "3.1150.0", "@base-ui/react": "^1.8.0", "@better-auth/oauth-provider": "1.7.3", "@codemirror/commands": "^6.11.0", @@ -47,6 +48,42 @@ "@cloudflare/worker-bundler@0.2.3": "patches/@cloudflare%2Fworker-bundler@0.2.3.patch", }, "packages": { + "@aws-sdk/checksums": ["@aws-sdk/checksums@3.1001.3", "", { "dependencies": { "@aws-sdk/core": "^3.978.3", "@aws-sdk/types": "^3.974.6", "@smithy/core": "^3.35.2", "@smithy/types": "^4.19.0", "tslib": "^2.6.2" } }, "sha512-CcAkVyEUKGngbEqJdVdSc53Iy52+XzVL5NIcTen2BlK8UByNC1DThVtf5VKSrNzyA+SqyTvs+3S+knVBbOUizQ=="], + + "@aws-sdk/client-s3": ["@aws-sdk/client-s3@3.1150.0", "", { "dependencies": { "@aws-sdk/checksums": "^3.1001.3", "@aws-sdk/core": "^3.978.3", "@aws-sdk/credential-provider-node": "^3.972.86", "@aws-sdk/middleware-sdk-s3": "^3.972.79", "@aws-sdk/signature-v4-multi-region": "^3.996.48", "@aws-sdk/types": "^3.974.6", "@smithy/core": "^3.35.2", "@smithy/fetch-http-handler": "^5.8.0", "@smithy/node-http-handler": "^4.12.1", "@smithy/types": "^4.19.0", "tslib": "^2.6.2" } }, "sha512-tv7KzK7PY7GqBwA6nzAVAsazIY/tAagDjlqvmCwlIYpH2SMfUuD6pJklg++LGcMaNy5p0VMvz+F+/es1BpoyRw=="], + + "@aws-sdk/core": ["@aws-sdk/core@3.978.3", "", { "dependencies": { "@aws-sdk/types": "^3.974.6", "@aws-sdk/xml-builder": "^3.972.41", "@aws/lambda-invoke-store": "^0.3.0", "@smithy/core": "^3.35.2", "@smithy/signature-v4": "^5.7.4", "@smithy/types": "^4.19.0", "bowser": "^2.11.0", "tslib": "^2.6.2" } }, "sha512-nslpzUCTI9b2IQMWSd6mSAXldBexPc8rr0d3dXUutXYFLFaTdsHa2GD7sJdsn3PhyjK0O5K8ypd6zi+RXXPVlg=="], + + "@aws-sdk/credential-provider-env": ["@aws-sdk/credential-provider-env@3.972.74", "", { "dependencies": { "@aws-sdk/core": "^3.978.3", "@aws-sdk/types": "^3.974.6", "@smithy/core": "^3.35.2", "@smithy/types": "^4.19.0", "tslib": "^2.6.2" } }, "sha512-rsRJcW2iFUhCgclyD6OSbMEaPl8Eu0qSeRZb2uuCK1ViZ66hl5wOe/p6Vr6MO3b+ydyKgFeqza2p6tZThvILlw=="], + + "@aws-sdk/credential-provider-http": ["@aws-sdk/credential-provider-http@3.972.76", "", { "dependencies": { "@aws-sdk/core": "^3.978.3", "@aws-sdk/types": "^3.974.6", "@smithy/core": "^3.35.2", "@smithy/fetch-http-handler": "^5.8.0", "@smithy/node-http-handler": "^4.12.1", "@smithy/types": "^4.19.0", "tslib": "^2.6.2" } }, "sha512-pnKdh0Br7OJPj+7qKjzgTfDTFW0RhEMKXNDR2jv9CAM4oHuG/4z3hRh8KtutLfpmItuRtDbKBmwWkDqHwwAWXw=="], + + "@aws-sdk/credential-provider-ini": ["@aws-sdk/credential-provider-ini@3.973.19", "", { "dependencies": { "@aws-sdk/core": "^3.978.3", "@aws-sdk/credential-provider-env": "^3.972.74", "@aws-sdk/credential-provider-http": "^3.972.76", "@aws-sdk/credential-provider-login": "^3.972.81", "@aws-sdk/credential-provider-process": "^3.972.74", "@aws-sdk/credential-provider-sso": "^3.973.18", "@aws-sdk/credential-provider-web-identity": "^3.972.80", "@aws-sdk/nested-clients": "^3.997.48", "@aws-sdk/types": "^3.974.6", "@smithy/core": "^3.35.2", "@smithy/credential-provider-imds": "^4.5.2", "@smithy/types": "^4.19.0", "tslib": "^2.6.2" } }, "sha512-kcjvAxqQEV8yuwkX58RelNn3ne65o7JgRWf3Z/gVvbi7Xuw1qSDLJlatrLD76+40cmRxmNkLxBoyp/vAQChQ5g=="], + + "@aws-sdk/credential-provider-login": ["@aws-sdk/credential-provider-login@3.972.81", "", { "dependencies": { "@aws-sdk/core": "^3.978.3", "@aws-sdk/nested-clients": "^3.997.48", "@aws-sdk/types": "^3.974.6", "@smithy/core": "^3.35.2", "@smithy/types": "^4.19.0", "tslib": "^2.6.2" } }, "sha512-CeQHSbE2CnVdaHIJ/NayRaCecIER+Pm9faUPkmRATO54rCoMTEPSbVieGHzEwvm1dZM2HIs6qQZCznhayNTWJQ=="], + + "@aws-sdk/credential-provider-node": ["@aws-sdk/credential-provider-node@3.972.86", "", { "dependencies": { "@aws-sdk/credential-provider-env": "^3.972.74", "@aws-sdk/credential-provider-http": "^3.972.76", "@aws-sdk/credential-provider-ini": "^3.973.19", "@aws-sdk/credential-provider-process": "^3.972.74", "@aws-sdk/credential-provider-sso": "^3.973.18", "@aws-sdk/credential-provider-web-identity": "^3.972.80", "@aws-sdk/types": "^3.974.6", "@smithy/core": "^3.35.2", "@smithy/credential-provider-imds": "^4.5.2", "@smithy/types": "^4.19.0", "tslib": "^2.6.2" } }, "sha512-WNttSyr6e8innPtBSty//CdnqJcgZN6eAbtw3v/CR2IObz6MYZEo8eusdtuCHJ2ujwFqC2bSPHVyhXGHx8HruQ=="], + + "@aws-sdk/credential-provider-process": ["@aws-sdk/credential-provider-process@3.972.74", "", { "dependencies": { "@aws-sdk/core": "^3.978.3", "@aws-sdk/types": "^3.974.6", "@smithy/core": "^3.35.2", "@smithy/types": "^4.19.0", "tslib": "^2.6.2" } }, "sha512-Yj6kZFJQdbR+PygTkqdVLPHHpdDpOS9GCYVcDCuweCRRig9LF24ELQBMfyUw1vspvrZJUfnOGVyMjhaL094ahw=="], + + "@aws-sdk/credential-provider-sso": ["@aws-sdk/credential-provider-sso@3.973.18", "", { "dependencies": { "@aws-sdk/core": "^3.978.3", "@aws-sdk/nested-clients": "^3.997.48", "@aws-sdk/token-providers": "3.1150.0", "@aws-sdk/types": "^3.974.6", "@smithy/core": "^3.35.2", "@smithy/types": "^4.19.0", "tslib": "^2.6.2" } }, "sha512-FnpPhdY4qUGtX/8X6eKxh+uD+9Y19i9FYibgk4SBe0M935s9vocHkHzaxaMNk+YkHmvzgdu8cHA5SCoeU+kkew=="], + + "@aws-sdk/credential-provider-web-identity": ["@aws-sdk/credential-provider-web-identity@3.972.80", "", { "dependencies": { "@aws-sdk/core": "^3.978.3", "@aws-sdk/nested-clients": "^3.997.48", "@aws-sdk/types": "^3.974.6", "@smithy/core": "^3.35.2", "@smithy/types": "^4.19.0", "tslib": "^2.6.2" } }, "sha512-TJF25Ndwy2HxcVj6oxMgp8qVlJAawlExHYBXqhuC6Zt2m+A7WpdkjBD6QglNbiArlteYljZHEY3R7KKmno6ISA=="], + + "@aws-sdk/middleware-sdk-s3": ["@aws-sdk/middleware-sdk-s3@3.972.79", "", { "dependencies": { "@aws-sdk/core": "^3.978.3", "@aws-sdk/signature-v4-multi-region": "^3.996.48", "@aws-sdk/types": "^3.974.6", "@smithy/core": "^3.35.2", "@smithy/types": "^4.19.0", "tslib": "^2.6.2" } }, "sha512-UfgAQosf4IEXldMsyntVuHsJnaQzlH6pmkOr6dQy7CaOYSlGz3mWiW4Di7XfQ/kPJP6xE7oMshz1V0E8RqGdRg=="], + + "@aws-sdk/nested-clients": ["@aws-sdk/nested-clients@3.997.48", "", { "dependencies": { "@aws-sdk/core": "^3.978.3", "@aws-sdk/signature-v4-multi-region": "^3.996.48", "@aws-sdk/types": "^3.974.6", "@smithy/core": "^3.35.2", "@smithy/fetch-http-handler": "^5.8.0", "@smithy/node-http-handler": "^4.12.1", "@smithy/types": "^4.19.0", "tslib": "^2.6.2" } }, "sha512-B01jdXJtAW/XOiUzGdI2mz67WyEHbGEc9X6Euzw3bjLAa+txTqdZDER4X2FsIEcXzV2ABqudVZ5thali8tCCzQ=="], + + "@aws-sdk/signature-v4-multi-region": ["@aws-sdk/signature-v4-multi-region@3.996.48", "", { "dependencies": { "@aws-sdk/types": "^3.974.6", "@smithy/signature-v4": "^5.7.4", "@smithy/types": "^4.19.0", "tslib": "^2.6.2" } }, "sha512-cFDXfEbXaGwbnpvR/DyqfxfjxpjXVRVthQLgkwvcfDS5u3+8G9WSj1X1Ag/pTRg/QOeOno1Tq3ycEdrKsCUcOA=="], + + "@aws-sdk/token-providers": ["@aws-sdk/token-providers@3.1150.0", "", { "dependencies": { "@aws-sdk/core": "^3.978.3", "@aws-sdk/nested-clients": "^3.997.48", "@aws-sdk/types": "^3.974.6", "@smithy/core": "^3.35.2", "@smithy/types": "^4.19.0", "tslib": "^2.6.2" } }, "sha512-wDU36ofhabZYu+d2KSriyDbwzxNoaOp+4dG0p3qkabk7HeALR5nIWyQbjoyJZLn3iqWkv/xNdyUfbfevDVDzzg=="], + + "@aws-sdk/types": ["@aws-sdk/types@3.974.6", "", { "dependencies": { "@smithy/types": "^4.19.0", "tslib": "^2.6.2" } }, "sha512-v/clNZzZnDxGyvpHMOGpJKVXFAExJzUNAAjaWGdcx8QAcXLGwTaOkw33p5SHAi0YAioK32xB3hWwOekRVfmfKg=="], + + "@aws-sdk/xml-builder": ["@aws-sdk/xml-builder@3.972.41", "", { "dependencies": { "@smithy/types": "^4.19.0", "tslib": "^2.6.2" } }, "sha512-ctjVSyCMegrWfXlx6VqzSBFI6UqmQ5ZlnfMhdLIiWmhoH8UAQxSCP5N3OpG7X3k4LnS7ou74C4mt20+bfTW2aQ=="], + + "@aws/lambda-invoke-store": ["@aws/lambda-invoke-store@0.3.0", "", {}, "sha512-sl4Bm6yiMNYrZKkqqDFWN0UfnWhlS8ivKxrYl+6t0gCLrqr8y3B2IqZZbFRkfaVVp7C/baApyh71P+LeE1A2sQ=="], + "@babel/runtime": ["@babel/runtime@7.29.7", "", {}, "sha512-Nq8OhGWiZIZGV6hLHoyAKLLcJihP/xFeBMGJoUrxTX2psI8dCifzLhZISFb+VWS3wFMRDmCGw5R+dOySCqPLhw=="], "@base-ui/react": ["@base-ui/react@1.8.0", "", { "dependencies": { "@babel/runtime": "^7.29.7", "@base-ui/utils": "0.4.0", "@floating-ui/react-dom": "^2.1.9", "@floating-ui/utils": "^0.2.12", "use-sync-external-store": "^1.6.0" }, "peerDependencies": { "@date-fns/tz": "^1.2.0", "@types/react": "^17 || ^18 || ^19", "date-fns": "^4.0.0", "react": "^17 || ^18 || ^19", "react-dom": "^17 || ^18 || ^19" }, "optionalPeers": ["@date-fns/tz", "@types/react", "date-fns"] }, "sha512-P0/1sxo6SBVZOklKMIedvTWqw2s2IQzi9x5bIVsXu980cuSOD4NeuRSs+/L7LZQfDkZP/uRZyGPyfFl/B1oH+Q=="], @@ -273,6 +310,18 @@ "@sindresorhus/is": ["@sindresorhus/is@7.2.0", "", {}, "sha512-P1Cz1dWaFfR4IR+U13mqqiGsLFf1KbayybWwdd2vfctdV6hDpUkgCY0nKOLLTMSoRd/jJNjtbqzf13K8DCCXQw=="], + "@smithy/core": ["@smithy/core@3.35.2", "", { "dependencies": { "@smithy/types": "^4.19.0", "tslib": "^2.6.2" } }, "sha512-ejTukBuP/Pg3t9N8Ux+0zKsXeHWm7p6W3HbXFOyltKasSxZwCFmvTut7kBdZSCF3m9y1bXXyVu6p2xuMFcsTyQ=="], + + "@smithy/credential-provider-imds": ["@smithy/credential-provider-imds@4.5.2", "", { "dependencies": { "@smithy/core": "^3.33.2", "@smithy/types": "^4.17.2", "tslib": "^2.6.2" } }, "sha512-A9uSdn72ozbRUSit0eib0TW7nXuNPlaeM0zcGkJ+nE6tFcSDbnmtwoxbTCFBukVQcszDAyvsd7+rTduPTXpygg=="], + + "@smithy/fetch-http-handler": ["@smithy/fetch-http-handler@5.8.0", "", { "dependencies": { "@smithy/core": "^3.33.3", "@smithy/types": "^4.18.0", "tslib": "^2.6.2" } }, "sha512-ycSJu3tFAQ4v04CBB0agqFMVsSQ1iG3yw+SpgxRqKfaURpQD4CZ8Wn0zPMmSnOuTpTh65Vz+EA0rMrw089wvkA=="], + + "@smithy/node-http-handler": ["@smithy/node-http-handler@4.12.1", "", { "dependencies": { "@smithy/core": "^3.33.3", "@smithy/types": "^4.18.0", "tslib": "^2.6.2" } }, "sha512-ThMkboGeONWXAelq9FvGsuJC4rOi+qyC4/zhUF58xYpxUg5sQKx2VXZYJmtNjr4dSuBJ1HeJXETQILCz3wOHvw=="], + + "@smithy/signature-v4": ["@smithy/signature-v4@5.7.4", "", { "dependencies": { "@smithy/core": "^3.35.0", "@smithy/types": "^4.19.0", "tslib": "^2.6.2" } }, "sha512-tHy0K0VtqNd5Y7Y41h0a0Lhh0L1GzC08dTWg0F7vRJWFtTENg7IZikf3wQkanYIRdb7ngoIPMTmqgUi401fEeQ=="], + + "@smithy/types": ["@smithy/types@4.19.0", "", { "dependencies": { "tslib": "^2.6.2" } }, "sha512-r7jh49VJxGerfAcTQA6gXcKc+98zOp/tqRwzYjgOE+iSQsP6cEU1hq2QzbuipmP68QtYdY9wKEhiCQZIzHgZ4Q=="], + "@speed-highlight/core": ["@speed-highlight/core@1.2.24", "", {}, "sha512-qeW2e1l78afw8VhRPfPQ1Gjj+KU5XFQ/OFV5ti6eTa9bruO7mJyZtA4vw0ofqmA3tKCkROE9xLk3VZoeRc98nw=="], "@standard-schema/spec": ["@standard-schema/spec@1.1.0", "", {}, "sha512-l2aFy5jALhniG5HgqrD6jXLi/rUWrKvqN/qJx6yoJsgKhblVd+iqqU4RCXavm/jPityDo5TCvKMnpjKnOriy0w=="], @@ -305,6 +354,8 @@ "body-parser": ["body-parser@2.3.0", "", { "dependencies": { "bytes": "^3.1.2", "content-type": "^2.0.0", "debug": "^4.4.3", "http-errors": "^2.0.1", "iconv-lite": "^0.7.2", "on-finished": "^2.4.1", "qs": "^6.15.2", "raw-body": "^3.0.2", "type-is": "^2.1.0" } }, "sha512-2cGmJupaNgg+QUwVLAucDuWuoMZ6EX9iHDRswZ5lsNYEmwPaRknMPCLZz07yTzVq/83p4o/wzbDZbBrTvGGTIw=="], + "bowser": ["bowser@2.14.1", "", {}, "sha512-tzPjzCxygAKWFOJP011oxFHs57HzIhOEracIgAePE4pqB3LikALKnSzUyU4MGs9/iCEUuHlAJTjTc5M+u7YEGg=="], + "bun-types": ["bun-types@1.3.14", "", { "dependencies": { "@types/node": "*" } }, "sha512-4N0ig0fEomHt5R0KCFWjovxow98rIoRwKolrYdCcknNwMekCXRnWEUvgu5soYV8QXtVsrUD8B95MBOZGPvr6KQ=="], "bytes": ["bytes@3.1.2", "", {}, "sha512-/Nf7TyzTx6S3yRJObOAV7956r8cr2+Oj8AC5dt8wSP3BQAoeX58NoHyCU8P8zGkNXStjTSi6fzO6F0pBdcYbEg=="], diff --git a/cloudflare/celld.test.ts b/cloudflare/celld.test.ts index edb1439..6c00c9b 100644 --- a/cloudflare/celld.test.ts +++ b/cloudflare/celld.test.ts @@ -165,6 +165,33 @@ celldTest("gallery subscriptions deliver MCP writes on celld and reconnect after } finally { reconnected.close(); } }, 120_000); +celldTest("gallery HTTP events deliver changes on celld without a public WebSocket upgrade", async () => { + const abort = new AbortController(); + const response = await fetch(`${baseUrl}/api/gallery/events`, { signal: abort.signal }); + expect(response.status).toBe(200); + expect(response.headers.get("content-type")).toBe("text/event-stream"); + const reader = response.body!.pipeThrough(new TextDecoderStream()).getReader(); + async function receive(value: string) { + let received = ""; + while (!received.includes(`data: ${value}\n\n`)) { + const chunk = await reader.read(); + if (chunk.done) throw new Error("Live event stream ended"); + received += chunk.value; + } + } + try { + await receive("ready"); + const changed = receive("changed"); + await withClient(async client => { + const result = await client.callTool({ name: "script_write", arguments: { name: "http-live-gallery", contents: 'export default { fetch() { return new Response("hello"); } };' } }); + expect(result.isError, JSON.stringify(result.content)).not.toBe(true); + }); + await changed; + const denied = await fetch(`${baseUrl}/api/gallery/events`, { headers: { Origin: "https://untrusted.example" } }); + expect(denied.status).toBe(403); + } finally { abort.abort(); await reader.cancel().catch(() => {}); } +}, 30_000); + celldTest("runs native artifact SQLite and KV across code update and celld restart", async () => { await withClient(async client => { await write(client, serverV1); diff --git a/cloudflare/gallery-events.ts b/cloudflare/gallery-events.ts new file mode 100644 index 0000000..7f31186 --- /dev/null +++ b/cloudflare/gallery-events.ts @@ -0,0 +1,60 @@ +/** Bridge native subscriptions to HTTP for proxies that cannot carry upgrades. */ +export async function galleryEvents( + request: Request, + subscribe: () => Promise, +): Promise { + const response = await subscribe(); + const socket = response.webSocket; + if (response.status !== 101 || !socket) + return new Response("Live updates unavailable", { status: 503 }); + const encoder = new TextEncoder(); + let heartbeat: ReturnType; + let lease: ReturnType; + let stopped = false; + let controller: ReadableStreamDefaultController; + const close = () => { + if (stopped) return; + stopped = true; + clearInterval(heartbeat); + clearTimeout(lease); + request.signal.removeEventListener("abort", close); + socket.close(1000, "Renew subscription"); + controller.close(); + }; + const stream = new ReadableStream({ + start(output) { + controller = output; + socket.addEventListener("message", (event) => { + if (!stopped && event.data === "changed") + controller.enqueue(encoder.encode("data: changed\n\n")); + }); + socket.addEventListener("close", close); + socket.addEventListener("error", close); + socket.accept(); + controller.enqueue(encoder.encode("retry: 1000\ndata: ready\n\n")); + // Stay below common proxy idle timeouts; periodically reauthenticate. + heartbeat = setInterval(() => { + if (!stopped) controller.enqueue(encoder.encode(": heartbeat\n\n")); + }, 20_000); + lease = setTimeout(close, 4 * 60_000); + request.signal.addEventListener("abort", close, { once: true }); + if (request.signal.aborted) close(); + }, + cancel() { + // The reader already owns closure when it cancels the response. + if (stopped) return; + stopped = true; + clearInterval(heartbeat); + clearTimeout(lease); + request.signal.removeEventListener("abort", close); + socket.close(1000, "Client disconnected"); + }, + }); + return new Response(stream, { + headers: { + "Content-Type": "text/event-stream", + "Cache-Control": "private, no-store", + "X-Accel-Buffering": "no", + }, + }); +} diff --git a/cloudflare/managed-service.ts b/cloudflare/managed-service.ts index fcfbce4..a793f39 100644 --- a/cloudflare/managed-service.ts +++ b/cloudflare/managed-service.ts @@ -106,7 +106,7 @@ export class ManagedArtifactService { return this.env.NATIVE_APPS.getByName("deployment"); } private async nativeTool(tool: string, args: Record): Promise { - if (tool === "app_guide") return text({ guide: "Native apps preserve the default Worker and named Durable Object exports. app_write accepts source, a strict app-owned manifest and optional project files/dependencies. Manifest fields: main (relative JS/TS path), compatibility_date, compatibility_flags, vars, secrets (names only), bindings (name -> {type:kv|r2|d1|queue|durable-object, resource:stable-name, class_name:DO export}), triggers:{crons:[],queues:[resource-name]}. Resource names permanently identify storage; binding/export renames preserve it and removed resources remain reserved. Set secrets with app_secrets. app_read/history/restore/reconcile manage persistent revisions and interrupted deployments. Restores retain current data and current secrets. Providers cannot change on an existing app. celld-local accepts trusted code only. URLs are private /apps//?workspace=&library=, require library authentication, stream HTTP and strip management credentials/cookies. Cloudflare uses native queue/cron delivery. celld-local operator does not support WebSockets. No destructive resource removal or schema/data rollback is provided." }); + if (tool === "app_guide") return text({ guide: "Native apps preserve the default Worker and named Durable Object exports. app_write accepts source, a strict app-owned manifest and optional project files/dependencies. Manifest fields: main (relative JS/TS path), compatibility_date, compatibility_flags, vars, secrets (names only), bindings (name -> {type:kv|r2|d1|queue|durable-object, resource:stable-name, class_name:DO export}), triggers:{crons:[],queues:[resource-name]}. Resource names permanently identify storage; binding/export renames preserve it and removed resources remain reserved. Set secrets with app_secrets. app_read/history/restore/reconcile manage persistent revisions and interrupted deployments. Restores retain current data and current secrets. Providers cannot change on an existing app. celld-local and celld-fleet accept trusted code only; celld-fleet uses the existing bucket-backed fleet and its coordinated publisher. URLs are private /apps//?workspace=&library=, require library authentication, stream HTTP and strip management credentials/cookies. Cloudflare uses native queue/cron delivery. celld-fleet uses native queues and host-coordinated cron dispatch into each app's own environment. celld-local operator does not support WebSockets. No destructive resource removal or schema/data rollback is provided." }); const controller = this.nativeController(), input = { owner: this.libraryKey, workspace: this.workspace, name: args.name as string }; let result: unknown; switch (tool) { @@ -188,7 +188,7 @@ export class ManagedArtifactService { ? await this.nativeController().list({ owner: this.libraryKey, workspace: all ? undefined : this.workspace, offset }) : null; hasMore ||= workers?.next_offset != null; - return { libraryScope: this.libraryKey === "team" ? "team" : "private", capabilities: { editing: true, scripts: true, links: true, moves: true, nativeApps: !!this.env.NATIVE_APPS, subscriptions: true }, workspace: this.workspace, + return { libraryScope: this.libraryKey === "team" ? "team" : "private", capabilities: { editing: true, scripts: true, links: true, moves: true, nativeApps: !!this.env.NATIVE_APPS, subscriptions: true, subscriptionTransport: "events" as const }, workspace: this.workspace, ...(workers ? { workerApps: workers.apps.map(app => ({ ...app, kind: "worker" as const, key: JSON.stringify(["worker", app.workspace, app.name]) })), nativeAppProviders: workers.providers } : {}), artifacts: [...artifacts.values()].sort((a, b) => a.name.localeCompare(b.name) || a.workspace.localeCompare(b.workspace)), nextOffset: hasMore ? offset + 100 : null }; } diff --git a/cloudflare/native-worker/controller.ts b/cloudflare/native-worker/controller.ts index b60f95f..fa436d2 100644 --- a/cloudflare/native-worker/controller.ts +++ b/cloudflare/native-worker/controller.ts @@ -6,22 +6,26 @@ import { artifactApiResponse } from "../artifact-api"; import { appScope, parseManifest, planResources, type AppManifest, type ResourceLedger } from "./manifest"; import { CloudflareNativeProvider, type CloudflareProviderState } from "./cloudflare"; import type { NativeDeployment, NativeRevision, ProviderResult } from "./types"; +import { fleetBinding, fleetScheduleBinding } from "./fleet-contract"; export type NativeEnvironment = { NATIVE_CF_ACCOUNT_ID?: string; NATIVE_CF_API_TOKEN?: string; NATIVE_CELLD_OPERATOR_URL?: string; NATIVE_CELLD_OPERATOR_TOKEN?: string; NATIVE_CELLD_TRUSTED_APPS?: string; + NATIVE_CELLD_FLEET_OPERATOR_URL?: string; NATIVE_CELLD_FLEET_OPERATOR_TOKEN?: string; + [binding: `ARTIFACTS_APP_${string}`]: Fetcher | undefined; }; type Selection = { owner: string; workspace: string; name: string }; type Draft = { source: string; manifest: AppManifest; project: ArtifactProject }; type ProviderState = CloudflareProviderState & { operatorOrigin?: string }; -type Row = Selection & { id: string; provider: "cloudflare" | "celld-local"; desired: string | null; active: string | null; +type Row = Selection & { id: string; provider: "cloudflare" | "celld-local" | "celld-fleet"; desired: string | null; active: string | null; status: string; stage: string; resources: string; provider_state: string; secrets: string; endpoint: string | null; draft: string; operation: string; attempts: number; error: string | null; updated_at: string }; const bytes = (value: string) => new TextEncoder().encode(value).byteLength; export function nativeProviders(env: NativeEnvironment) { return [env.NATIVE_CF_ACCOUNT_ID && env.NATIVE_CF_API_TOKEN ? "cloudflare" : null, - env.NATIVE_CELLD_OPERATOR_URL && env.NATIVE_CELLD_OPERATOR_TOKEN && env.NATIVE_CELLD_TRUSTED_APPS === "true" ? "celld-local" : null].filter((value): value is "cloudflare" | "celld-local" => value !== null); + env.NATIVE_CELLD_FLEET_OPERATOR_URL && env.NATIVE_CELLD_FLEET_OPERATOR_TOKEN && env.NATIVE_CELLD_TRUSTED_APPS === "true" ? "celld-fleet" : null, + env.NATIVE_CELLD_OPERATOR_URL && env.NATIVE_CELLD_OPERATOR_TOKEN && env.NATIVE_CELLD_TRUSTED_APPS === "true" ? "celld-local" : null].filter((value): value is Row["provider"] => value !== null); } function selection(input: Selection): Selection { for (const key of ["owner", "workspace", "name"] as const) { @@ -47,6 +51,8 @@ export class NativeApps extends DurableObject { unique(owner,workspace,name))`); this.sql.exec(`create table if not exists native_revisions ( app text not null, id text not null, snapshot text not null, created_at text not null, primary key(app,id))`); + this.sql.exec(`create table if not exists native_cron_runs ( + app text not null, cron text not null, scheduled_time integer not null, primary key(app,cron,scheduled_time))`); // A restarted controller resumes persisted intent. Alarms never create a new logical deployment. ctx.blockConcurrencyWhile(async () => { if (this.sql.exec("select id from native_apps where status='deploying'").toArray().length) await ctx.storage.setAlarm(Date.now() + 1000); @@ -107,7 +113,7 @@ export class NativeApps extends DurableObject { this.sql.exec("insert into native_apps(id,owner,workspace,name,provider,status,stage,provider_state,draft,operation,updated_at) values(?,?,?,?,?,'draft','draft',?,?,?,?)", id, input.owner, input.workspace, input.name, provider, JSON.stringify({ ids: {}, migrations: {}, gatewayToken: Array.from(randomBytes(32), byte => byte.toString(16).padStart(2, "0")).join(""), - ...(provider === "cloudflare" ? { accountId: this.env.NATIVE_CF_ACCOUNT_ID } : { operatorOrigin: this.operatorUrl().origin }), + ...(provider === "cloudflare" ? { accountId: this.env.NATIVE_CF_ACCOUNT_ID } : { operatorOrigin: this.operatorUrl(provider === "celld-fleet").href }), } satisfies ProviderState), JSON.stringify({ source: input.source, manifest, project: emptyProject() }), crypto.randomUUID(), new Date().toISOString()); row = this.row(input); @@ -203,9 +209,19 @@ export class NativeApps extends DurableObject { result = await new CloudflareNativeProvider(this.env).reconcile(deployment, state, async (stage, providerState) => { this.sql.exec("update native_apps set stage=?,provider_state=?,updated_at=? where id=?", stage, JSON.stringify(providerState), new Date().toISOString(), id); }); + } else if (row.provider === "celld-fleet") { + const endpoint = this.operatorUrl(true); + if (state.operatorOrigin !== endpoint.href) throw new Error("App fleet operator changed; restore its configured operator before reconciling"); + this.sql.exec("update native_apps set stage='fleet-publication' where id=?", id); + const response = await fetch(new URL("deploy", endpoint), { method: "POST", headers: { Authorization: `Bearer ${this.env.NATIVE_CELLD_FLEET_OPERATOR_TOKEN}`, "Content-Type": "application/json" }, body: JSON.stringify(deployment), signal: AbortSignal.timeout(120_000), redirect: "manual" }); + if (!response.ok) throw new Error(`celld fleet publication failed (${response.status})`); + result = await response.json() as ProviderResult; + if (result.revision !== revision.id) throw new Error("celld fleet operator returned an invalid revision"); + // Native service bindings carry requests; no provider URL is trusted. + result.endpoint = `https://native-app.internal/${id}/`; } else { const endpoint = this.operatorUrl(); - if (state.operatorOrigin !== endpoint.origin) throw new Error("App operator origin changed; restore its configured operator before reconciling"); + if (state.operatorOrigin !== endpoint.origin && state.operatorOrigin !== endpoint.href) throw new Error("App operator origin changed; restore its configured operator before reconciling"); this.sql.exec("update native_apps set stage='operator-reconcile' where id=?", id); const response = await fetch(new URL("deploy", endpoint), { method: "POST", headers: { Authorization: `Bearer ${this.env.NATIVE_CELLD_OPERATOR_TOKEN}`, "Content-Type": "application/json" }, body: JSON.stringify(deployment), signal: AbortSignal.timeout(120_000), redirect: "manual" }); if (!response.ok) throw new Error(`celld operator deployment failed (${response.status})`); @@ -222,13 +238,34 @@ export class NativeApps extends DurableObject { return { ok: false, ...this.summary(this.byId(id)) }; } } - private operatorUrl() { - if (!nativeProviders(this.env).includes("celld-local")) throw new Error("Trusted celld native provider is not configured"); - const endpoint = new URL(this.env.NATIVE_CELLD_OPERATOR_URL!); - if (endpoint.username || endpoint.password || endpoint.search || endpoint.hash || endpoint.pathname !== "/" - || endpoint.protocol !== "https:" && !(endpoint.protocol === "http:" && ["localhost", "127.0.0.1", "[::1]"].includes(endpoint.hostname))) throw new Error("Operator URL must be HTTPS or loopback HTTP, without a path or credentials"); + private operatorUrl(fleet = false) { + if (!nativeProviders(this.env).includes(fleet ? "celld-fleet" : "celld-local")) throw new Error("Trusted celld native provider is not configured"); + const endpoint = new URL(fleet ? this.env.NATIVE_CELLD_FLEET_OPERATOR_URL! : this.env.NATIVE_CELLD_OPERATOR_URL!); + const validPath = endpoint.pathname === "/" || fleet && /^\/[a-z0-9/-]+\/$/.test(endpoint.pathname); + const loopbackHttp = endpoint.protocol === "http:" && ["localhost", "127.0.0.1", "[::1]"].includes(endpoint.hostname); + if (endpoint.username || endpoint.password || endpoint.search || endpoint.hash || !validPath || !(endpoint.protocol === "https:" || loopbackHttp)) throw new Error("Operator URL must be HTTPS or loopback HTTP, without credentials or a query"); return endpoint; } + async fireCron(event: { cron: string; scheduledTime: number }) { + if (typeof event.cron !== "string" || !Number.isSafeInteger(event.scheduledTime)) throw new Error("Invalid cron event"); + let failed = false; + const apps = this.sql.exec("select * from native_apps where provider='celld-fleet' and status='active' and active is not null").toArray(); + for (const app of apps) { + if (!this.revision(app.id, app.active!).manifest.triggers.crons.includes(event.cron)) continue; + await this.serial(app.id, async () => { + if (this.sql.exec("select 1 from native_cron_runs where app=? and cron=? and scheduled_time=?", app.id, event.cron, event.scheduledTime).toArray().length) return; + const scheduler = this.env[fleetScheduleBinding(app.id)] as unknown as { runScheduled(event: { cron: string; scheduledTime: number }): Promise<{ outcome: string; noRetry: boolean }> } | undefined; + try { + if (!scheduler) throw new Error("Fleet app scheduler unavailable"); + const result = await scheduler.runScheduled(event); + if (result.outcome !== "ok" && !result.noRetry) { failed = true; return; } + this.sql.exec("insert or ignore into native_cron_runs values(?,?,?)", app.id, event.cron, event.scheduledTime); + this.sql.exec("delete from native_cron_runs where app=? and scheduled_time("select * from native_apps where status='deploying' or (status='recovery-required' and attempts<3)").toArray(); for (const row of rows) await this.serial(row.id, async () => { @@ -248,11 +285,17 @@ export class NativeApps extends DurableObject { const target = new URL(request.url), endpoint = new URL(row.endpoint); endpoint.pathname = endpoint.pathname.replace(/\/$/, "") + target.pathname; endpoint.search = target.search; - headers.set("x-artifacts-target-url", request.url); - if (row.provider === "cloudflare") { - headers.set("x-artifacts-gateway-token", (JSON.parse(row.provider_state) as CloudflareProviderState).gatewayToken); - } else headers.set("authorization", `Bearer ${this.env.NATIVE_CELLD_OPERATOR_TOKEN}`); - const response = await fetch(endpoint, { method: request.method, headers, body: request.body, redirect: "manual" }); + let response: Response; + if (row.provider === "celld-fleet") { + const worker = this.env[fleetBinding(row.id)]; + if (!worker) return new Response("Fleet app deployment is not available on this node yet", { status: 503 }); + response = await worker.fetch(new Request(request.url, { method: request.method, headers, body: request.body, redirect: "manual" })); + } else { + headers.set("x-artifacts-target-url", request.url); + if (row.provider === "cloudflare") headers.set("x-artifacts-gateway-token", (JSON.parse(row.provider_state) as CloudflareProviderState).gatewayToken); + else headers.set("authorization", `Bearer ${this.env.NATIVE_CELLD_OPERATOR_TOKEN}`); + response = await fetch(endpoint, { method: request.method, headers, body: request.body, redirect: "manual" }); + } if (response.status === 101) { const output = new Headers(response.headers); output.delete("set-cookie"); return new Response(null, { status: 101, headers: output, webSocket: response.webSocket }); diff --git a/cloudflare/native-worker/fleet-contract.ts b/cloudflare/native-worker/fleet-contract.ts new file mode 100644 index 0000000..9fbedd8 --- /dev/null +++ b/cloudflare/native-worker/fleet-contract.ts @@ -0,0 +1,57 @@ +export const FLEET_REGISTRY_VAR = "ARTIFACTS_NATIVE_FLEET_REGISTRY"; +export const FLEET_PUBLICATION_VAR = "ARTIFACTS_NATIVE_FLEET_PUBLICATION"; +export const FLEET_SCHEDULER_EXPORT = "ArtifactsNativeScheduler"; +export const fleetBinding = (app: string) => + `ARTIFACTS_APP_${app.replaceAll("-", "")}` as const; +export const fleetScheduleBinding = (app: string) => + `${fleetBinding(app)}_SCHEDULE` as const; + +export type FleetPointer = { + script_name: string; + version: string; + prefix: string; + rollout: { percent: number }; +}; +export type FleetApp = { + app: string; + revision: string; + pointer: FleetPointer; + crons: string[]; + queues: string[]; +}; +export type FleetRegistry = { schema_version: 1; apps: FleetApp[] }; + +export function readFleetRegistry(value: unknown): FleetRegistry { + if (value === undefined) return { schema_version: 1, apps: [] }; + const registry = typeof value === "string" ? JSON.parse(value) : value; + if ( + !registry || + registry.schema_version !== 1 || + !Array.isArray(registry.apps) + ) + throw new Error("Invalid native fleet registry"); + const ids = new Set(); + for (const app of registry.apps) { + if ( + !app || + !/^[0-9a-f-]{36}$/.test(app.app) || + ids.has(app.app) || + !/^[0-9a-f]{64}$/.test(app.revision) || + app.pointer?.script_name !== `art-app-${app.app.replaceAll("-", "")}` || + !/^[0-9a-f]{16,64}$/.test(app.pointer?.version ?? "") || + app.pointer.prefix !== + `deploy/${app.pointer.script_name}/${app.pointer.version}` || + !Array.isArray(app.crons) || + !app.crons.every((value: unknown) => typeof value === "string") || + !Array.isArray(app.queues) || + !app.queues.every( + (value: unknown) => + typeof value === "string" && /^art-[0-9a-f]{48}$/.test(value), + ) + ) { + throw new Error("Invalid native fleet app"); + } + ids.add(app.app); + } + return registry; +} diff --git a/cloudflare/tool-contract.ts b/cloudflare/tool-contract.ts index 4057ec1..764e52f 100644 --- a/cloudflare/tool-contract.ts +++ b/cloudflare/tool-contract.ts @@ -153,7 +153,7 @@ addScriptTool("app_guide", "Read the native Worker manifest, ownership, private addScriptTool("app_list", "List native Worker apps and deployment status in the authenticated library/workspace. Returns 100 per page and configured providers; never runs app code.", pageProperty); addScriptTool("app_write", "Create or update an ordinary native Worker. Preserve default handler and named DO exports. Strict manifest declares app-owned native bindings; provider defaults to a configured provider and is fixed for an existing app. Stable resource names retain data on binding/class rename, removal and restoration. Persisted intent recovers provider mutations; failures require app_reconcile. Call app_guide first.", { ...appName, source: { type: "string", maxLength: 262144 }, manifest: { type: "object", maxProperties: 8 }, project: projectProperty, - provider: { type: "string", enum: ["cloudflare", "celld-local"] }, expected_revision: { ...revisionId, type: ["string", "null"], description: "Full source/manifest/project revision_token from app_read. Null requires a new app; omit for unconditional write." }, + provider: { type: "string", enum: ["cloudflare", "celld-local", "celld-fleet"] }, expected_revision: { ...revisionId, type: ["string", "null"], description: "Full source/manifest/project revision_token from app_read. Null requires a new app; omit for unconditional write." }, }, ["name", "source", "manifest"]); addScriptTool("app_read", "Read native Worker source, manifest, helper/dependency snapshot and deployment status. Optional revision_id reads immutable source; secrets and provider credentials are never returned.", { ...appName, revision_id: revisionId }, ["name"]); addScriptTool("app_history", "List immutable compiled native Worker revisions, 100 per page. Secrets and resource data are separate from source history.", { ...appName, ...pageProperty }, ["name"]); diff --git a/cloudflare/worker.ts b/cloudflare/worker.ts index 23c11b6..813a237 100644 --- a/cloudflare/worker.ts +++ b/cloudflare/worker.ts @@ -20,9 +20,12 @@ import type { PluginRequest } from "../src/plugins/types"; import galleryBridge from "../dist/cloudflare/gallery-request.json"; import { ArtifactLinks } from "./links"; +import { galleryEvents } from "./gallery-events"; import { ScriptLibrary } from "./scripts"; import { ScriptBackend } from "./script-backend"; import { NativeApps, type NativeEnvironment } from "./native-worker/controller"; +import { createHash, timingSafeEqual } from "node:crypto"; +import { FLEET_PUBLICATION_VAR } from "./native-worker/fleet-contract"; import { artifactRoute } from "./artifact-routes"; import * as toolValidators from "../dist/cloudflare/tool-validators.js"; export { ArtifactLibrary, ArtifactBackend, ArtifactFiles, ArtifactLinks, ScriptLibrary, ScriptBackend, NativeApps }; @@ -39,10 +42,17 @@ export type Env = AuthEnvironment & BetterAuthEnvironment & NativeEnvironment & ASSETS: Fetcher; DEFAULT_WORKSPACE?: string; ARTIFACTS_PUBLIC_ORIGIN?: string; + ARTIFACTS_NATIVE_FLEET_PUBLICATION?: string; }; const app = new Hono<{ Bindings: Env; Variables: { service: ManagedArtifactService; libraryScope: "private" | "team"; user: ArtifactUser | null } }>(); app.get("/health", c => c.json({ ok: true, runtime: navigator.userAgent })); +app.get("/api/native-fleet/ready", c => { + const expected = c.env.NATIVE_CELLD_FLEET_OPERATOR_TOKEN; + const supplied = c.req.header("Authorization")?.replace(/^Bearer /, "") ?? ""; + if (!expected || !timingSafeEqual(createHash("sha256").update(expected).digest(), createHash("sha256").update(supplied).digest())) return c.json({ error: "Unauthorized" }, 401); + return c.json({ publication: c.env[FLEET_PUBLICATION_VAR] }); +}); app.get("/sign-in", c => { if (c.env.AUTH_MODE !== "better-auth") return c.redirect("/"); c.header("Cache-Control", "no-store"); @@ -138,6 +148,7 @@ app.get("/api/gallery/subscribe", c => { if (c.req.header("Upgrade")?.toLowerCase() !== "websocket") return c.text("WebSocket required", 426); return c.get("service").subscribeGallery(c.req.query("all") === "1"); }); +app.get("/api/gallery/events", c => galleryEvents(c.req.raw, () => c.get("service").subscribeGallery(c.req.query("all") === "1"))); app.post("/api/library/move", async c => { let input: { kind: "artifact" | "script"; name: string; library: "private" | "team" }; try { input = JSON.parse(await readRequestText(c.req.raw, 8192)); } diff --git a/docs/native-worker-fleet.md b/docs/native-worker-fleet.md new file mode 100644 index 0000000..f6caa5c --- /dev/null +++ b/docs/native-worker-fleet.md @@ -0,0 +1,87 @@ +# Native Worker apps on a celld fleet + +The `celld-fleet` provider runs native apps on the same bucket-backed fleet as +Artifacts. Use pinned celld 0.6.1 and one authenticated publisher. The publisher +has no local app databases: native storage and the deployment registry remain +in the fleet bucket, and the `NativeApps` controller retains source, secrets, +resource identities and desired deployment state. + +## Deployment + +Build the Docker `fleet-operator` target alongside the ordinary `runtime` +target. CI publishes `ghcr.io/sidequery/artifacts:fleet-sha-` and +`ghcr.io/sidequery/artifacts:sha-` for the same source commit. + +Start the publisher with these environment variables: + +| Variable | Purpose | +| --- | --- | +| `CELLD_BUCKET` | Existing S3 fleet bucket, including its prefix when used. | +| `AWS_REGION` | Bucket region; normal AWS credentials or workload identity apply. | +| `NATIVE_CELLD_FLEET_OPERATOR_TOKEN` | Private service credential, at least 32 characters. | +| `NATIVE_CELLD_FLEET_CONFIG` | Prepared Artifacts Wrangler configuration; default `/app/wrangler.jsonc`. | +| `NATIVE_CELLD_FLEET_NODE_SERVICE` | Headless DNS name returning all node IPs. | +| `NATIVE_CELLD_FLEET_NODE_PORT` | Public Worker port for adoption checks; default `4786`. | +| `NATIVE_CELLD_FLEET_NODES` | Alternative: comma-separated node URLs for adoption checks. | +| `NATIVE_CELLD_FLEET_HOST` | Listener interface; default loopback. | +| `NATIVE_CELLD_FLEET_PORT` | Listener port; default `4792`. | +| `NATIVE_CELLD_FLEET_BASE_PATH` | Optional HTTP path prefix. | + +Controller vars must include `NATIVE_CELLD_TRUSTED_APPS=true`, the same private +token in `NATIVE_CELLD_FLEET_OPERATOR_TOKEN`, and its URL in +`NATIVE_CELLD_FLEET_OPERATOR_URL`. Use HTTPS or loopback HTTP. Declare native +apps only for admitted, trusted users; the runtime's isolation contract is +unchanged by this provider. + +Run one publisher replica with a Recreate rollout. It serializes both app +updates and host releases. A release hook sends authenticated `POST /host` +through that service using the publisher image's `--request-host` command and +`NATIVE_CELLD_FLEET_RELEASE_URL`. The hook runs after the publisher becomes +ready and before celld nodes start. Host publication does not wait for nodes, +so a fresh empty bucket can bootstrap. App publication verifies adoption on +every discovered node before reporting activation. + +All publisher HTTP requests require its bearer token. `GET /health` reports +publisher readiness; `POST /deploy` accepts the existing `NativeDeployment` +contract. Management credentials and root host configuration never enter an +authored app's environment. + +## Publication and recovery + +celld 0.6.1 has one fleet-wide default pointer. The publisher runs candidate +`celld deploy` commands in separate staging prefixes, copies their immutable +modules and manifests, then publishes the named app pointers and native queue +attachments. It adds private service bindings to the host, preserves Artifacts +as the default, and conditionally commits the host pointer. Its deployment +registry is part of that host manifest, so later host releases preserve apps. +An intervening default-pointer change fails publication instead of overwriting +another release. Do not publish directly into this fleet with plain `celld +deploy`; all changes must use the coordinated publisher. + +The controller persists desired revisions before calling the publisher. +Interrupted or failed publication remains `recovery-required`; use +`app_reconcile` to resume the same revision and retained resources. Source +restoration retains current data and current app secrets. Source revisions do +not roll back databases, files, queue delivery or application schema changes. +Native queue attachment and code publication can be partial until recovery +finishes; event handlers must tolerate retries. + +The host owns the union of app cron expressions because celld 0.6.1 does not +automatically fire a cohosted Worker's cron. It invokes generated scheduler +entrypoints in each app's native environment and waits for `waitUntil` work. +The controller remembers successful/no-retry occurrences, so another app's +retry does not rerun already completed cron handlers. Interrupted occurrences +retain the normal possibility of repeated side effects. + +HTTP and WebSocket responses use native service bindings. Public upgrades +still require an ingress that supports WebSockets. The gallery uses an HTTP +event stream for its live updates, including behind HTTP-only load balancers. + +## Validation + +`NATIVE_WORKER_FLEET=1 bun test scripts/native-worker/fleet.test.ts --timeout +240000` uses a disposable S3-compatible fixture and two pinned celld nodes. +It verifies native storage, queues and cron delivery, private/team isolation, +credential stripping, invalid drafts, source/host updates, and node/publisher +restart. Configure `FLEET_TEST_S3_ENDPOINT` for the fixture. These tests do not +write to the production bucket. diff --git a/docs/native-workers.md b/docs/native-workers.md index 748a3b8..abb1e7e 100644 --- a/docs/native-workers.md +++ b/docs/native-workers.md @@ -68,6 +68,12 @@ Set controller variables `NATIVE_CELLD_OPERATOR_URL` and `NATIVE_CELLD_TRUSTED_A This provider uses pinned celld 0.6.1 and accepts trusted applications only. celld does not offer hostile multitenant isolation. Each app uses one supervised loopback process and a stable persistent store. The operator stages changes, stops the old runtime, restarts against retained storage and checks celld runtime readiness without invoking the app handler. Updates can cause brief downtime. Failed startup reinstalls previous code/configuration while retaining current data. The operator rejects WebSocket upgrades with `501`. +### celld-fleet + +Use the [fleet publisher](native-worker-fleet.md) for an existing bucket-backed celld deployment. It keeps Artifacts as the fleet's default Worker and adds native apps as named service-binding components. Apps use native DO, KV, R2, D1 and queue bindings in the same fleet. Their resource identities survive source updates, host releases and node restarts. Native cron delivery is coordinated by the host and dispatched into each app's own isolate and environment. + +Configure `NATIVE_CELLD_FLEET_OPERATOR_URL`, `NATIVE_CELLD_FLEET_OPERATOR_TOKEN` and `NATIVE_CELLD_TRUSTED_APPS=true`. The authenticated publisher must have fleet bucket access and discover the serving nodes. Host releases must go through the same publisher; a plain `celld deploy` of the host would remove its native app service graph. The publisher's endpoint may include a path prefix, uses HTTPS or loopback HTTP, and is fixed for an existing app. Native app requests pass through private library admission and native service bindings. They receive only their declared app bindings, variables and secrets. + ## Recovery and restoration The controller journals desired code, complete reserved resource identities, provider IDs and DO migration plans before external mutations. Each app has a serialized mutation stream. Resource creation reconciles deterministic names after ambiguous responses; it never blindly allocates a second resource or deletes storage as compensation. Automatic controller retries are bounded to three attempts; `app_reconcile` starts a fresh attempt against the same desired revision. An operator restart also reconciles its persisted desired app state. @@ -76,7 +82,7 @@ The controller journals desired code, complete reserved resource identities, pro Code restoration does not restore data or undo DO lifecycle migrations. There is no managed D1 SQL migration runner: schema changes belong to application code and must remain compatible with retained data and restored revisions. Queue/cron changes and Worker publication are not a single provider transaction; `recovery-required` explicitly means the observed deployment may be partial. Native event side effects can occur while publication is incomplete and are not replayed or rolled back by the controller. Queue handlers must tolerate retries. -Providers cannot change on an existing app. Destructive storage deletion, fleet bucket publication, cross-provider data transfer, app log collection, workflows and other undeclared bindings/events are outside this delivery. +Providers cannot change on an existing app. Destructive storage deletion, cross-provider data transfer, app log collection, workflows and other undeclared bindings/events are outside this delivery. ## Validation diff --git a/e2e/gallery-connection.test.ts b/e2e/gallery-connection.test.ts new file mode 100644 index 0000000..6550422 --- /dev/null +++ b/e2e/gallery-connection.test.ts @@ -0,0 +1,200 @@ +import { expect, test } from "bun:test"; +import { chromium } from "playwright"; +import { galleryHtml } from "../src/gallery/server"; +import type { GalleryArtifact } from "../src/gallery/types"; + +async function clientBundle() { + const build = await Bun.build({ + entrypoints: [ + new URL("../src/gallery/client.tsx", import.meta.url).pathname, + ], + target: "browser", + }); + if (!build.success) throw new Error(build.logs.join("\n")); + return build.outputs[0]!.text(); +} + +test("rejected upgrades retain the empty view and copy feedback without reloading on every retry", async () => { + const js = await clientBundle(); + let galleryRequests = 0; + const server = Bun.serve({ + port: 0, + hostname: "127.0.0.1", + fetch(request) { + const path = new URL(request.url).pathname; + if (path === "/gallery.js") + return new Response(js, { + headers: { "content-type": "text/javascript" }, + }); + if (path === "/api/session") return Response.json({}); + if (path === "/api/gallery/subscribe") + return new Response("WebSocket required", { status: 426 }); + if (path === "/api/gallery") { + galleryRequests++; + return Response.json({ + workspace: "default", + artifacts: [], + capabilities: { links: true, subscriptions: true }, + }); + } + return new Response(galleryHtml(), { + headers: { "content-type": "text/html" }, + }); + }, + }); + const browser = await chromium.launch({ headless: true }); + try { + const page = await browser.newPage({ + permissions: ["clipboard-read", "clipboard-write"], + }); + await page.clock.install(); + await page.goto(server.url.href); + const onboarding = page.locator(".agent-onboarding"); + await onboarding.waitFor(); + await page + .getByRole("button", { name: "Copy MCP URL", exact: true }) + .click(); + await page.getByText("MCP URL copied", { exact: true }).waitFor(); + const before = await page + .getByPlaceholder("Search", { exact: true }) + .boundingBox(); + await onboarding.evaluate((node) => { + node.setAttribute("data-retained", "true"); + }); + await page.clock.runFor(10_000); + await page + .getByText("Reconnecting to live updates…", { exact: true }) + .waitFor(); + expect(galleryRequests).toBe(1); + expect(await onboarding.getAttribute("data-retained")).toBe("true"); + expect( + await page.getByText("MCP URL copied", { exact: true }).isVisible(), + ).toBe(true); + expect( + await page.getByPlaceholder("Search", { exact: true }).boundingBox(), + ).toEqual(before); + const polled = page.waitForResponse((response) => + response.url().includes("/api/gallery?"), + ); + await page.clock.runFor(30_000); + await polled; + expect(galleryRequests).toBe(2); + expect(await onboarding.getAttribute("data-retained")).toBe("true"); + } finally { + await browser.close(); + server.stop(true); + } +}, 30_000); + +test("hosted HTTP live updates retain onboarding during refresh and reconcile after reconnect", async () => { + const js = await clientBundle(); + const streams = new Set>(); + const artifacts: GalleryArtifact[] = []; + const encoder = new TextEncoder(); + let galleryGate: Promise | undefined; + let reachedGallery: (() => void) | undefined; + let failedConnect = false; + const server = Bun.serve({ + port: 0, + hostname: "127.0.0.1", + async fetch(request) { + const path = new URL(request.url).pathname; + if (path === "/gallery.js") + return new Response(js, { + headers: { "content-type": "text/javascript" }, + }); + if (path === "/api/session") return Response.json({}); + if (path === "/api/gallery/events") { + if (!failedConnect) { + failedConnect = true; + return new Response("Restarting", { status: 503 }); + } + let controller: ReadableStreamDefaultController; + return new Response( + new ReadableStream({ + start(output) { + controller = output; + streams.add(output); + output.enqueue(encoder.encode("retry: 100\ndata: ready\n\n")); + }, + cancel() { + streams.delete(controller); + }, + }), + { headers: { "content-type": "text/event-stream" } }, + ); + } + if (path === "/api/gallery") { + reachedGallery?.(); + await galleryGate; + return Response.json({ + workspace: "default", + artifacts, + capabilities: { + links: true, + subscriptions: true, + subscriptionTransport: "events", + }, + }); + } + if (path === "/gallery/preview") return new Response("Preview"); + return new Response(galleryHtml(), { + headers: { "content-type": "text/html" }, + }); + }, + }); + const browser = await chromium.launch({ headless: true }); + try { + const page = await browser.newPage(); + const errors: string[] = []; + page.on("pageerror", (error) => errors.push(error.message)); + const connected = page.waitForResponse( + (response) => + response.url().includes("/api/gallery/events") && + response.status() === 200, + ); + await page.goto(server.url.href); + await connected; + const onboarding = page.locator(".agent-onboarding"); + await onboarding.waitFor(); + await onboarding.evaluate((node) => + node.setAttribute("data-retained", "true"), + ); + let releaseGallery: () => void; + galleryGate = new Promise((resolve) => { + releaseGallery = resolve; + }); + const refreshed = new Promise((resolve) => { + reachedGallery = resolve; + }); + for (const stream of streams) + stream.enqueue(encoder.encode("data: changed\n\n")); + await refreshed; + expect(await onboarding.getAttribute("data-retained")).toBe("true"); + expect( + await page.getByText("Loading your library…", { exact: true }).count(), + ).toBe(0); + galleryGate = undefined; + reachedGallery = undefined; + releaseGallery!(); + artifacts.push({ + key: "missed", + name: "missed", + workspace: "default", + working: true, + versions: [], + draftRevision: "one", + }); + for (const stream of streams) stream.close(); + streams.clear(); + await page.getByTitle("default/missed", { exact: true }).waitFor(); + await page + .frameLocator('iframe[title="Preview of missed"]') + .getByText("Preview", { exact: true }) + .waitFor(); + expect(errors).toEqual([]); + } finally { + await browser.close(); + server.stop(true); + } +}, 30_000); diff --git a/e2e/gallery-native-apps.test.ts b/e2e/gallery-native-apps.test.ts index c3c4d28..65c6a91 100644 --- a/e2e/gallery-native-apps.test.ts +++ b/e2e/gallery-native-apps.test.ts @@ -377,6 +377,33 @@ test("Worker gallery deploys drafts, manages revisions/secrets and preserves edi } }, 30000); +test("a newly configured fleet provider enables Worker creation without reloading the page", async () => { + const build = await Bun.build({ entrypoints: [new URL("../src/gallery/client.tsx", import.meta.url).pathname], target: "browser" }); + if (!build.success) throw new Error(build.logs.join("\n")); + const js = await build.outputs[0]!.text(); + let configured = false; + const server = Bun.serve({ port: 0, hostname: "127.0.0.1", fetch(request) { + const path = new URL(request.url).pathname; + if (path === "/gallery.js") return new Response(js, { headers: { "content-type": "text/javascript" } }); + if (path === "/api/session") return Response.json({}); + if (path === "/api/gallery") return Response.json({ workspace: "default", artifacts: [], workerApps: [], nativeAppProviders: configured ? ["celld-fleet"] : [], capabilities: { nativeApps: true, subscriptions: true, subscriptionTransport: "events" } }); + if (path === "/api/gallery/events") return new Response(new ReadableStream({ start(controller) { controller.enqueue(new TextEncoder().encode("data: ready\n\n")); } }), { headers: { "content-type": "text/event-stream" } }); + return new Response(galleryHtml(), { headers: { "content-type": "text/html" } }); + } }); + const browser = await chromium.launch({ headless: true }); + try { + const page = await browser.newPage(); + await page.goto(`${server.url.href}?view=workers`); + await page.getByRole("heading", { name: "Worker apps are unavailable" }).waitFor(); + configured = true; + await page.evaluate(() => window.dispatchEvent(new Event("online"))); + await page.getByLabel("Worker app name").fill("fleet-app"); + await page.getByRole("button", { name: "Continue to editor", exact: true }).click(); + await page.getByRole("textbox", { name: "worker.ts", exact: true }).waitFor(); + expect(await page.getByRole("button", { name: "Save and deploy Worker", exact: true }).isEnabled()).toBe(true); + } finally { await browser.close(); server.stop(true); } +}, 30_000); + test("Worker provider empty state has documentation and no creation editor", async () => { const build = await Bun.build({ entrypoints: [new URL("../src/gallery/client.tsx", import.meta.url).pathname], diff --git a/e2e/gallery-onboarding.test.ts b/e2e/gallery-onboarding.test.ts index 6579b16..2e4bf45 100644 --- a/e2e/gallery-onboarding.test.ts +++ b/e2e/gallery-onboarding.test.ts @@ -19,14 +19,23 @@ test("empty hosted library supplies copyable deployment MCP URL and starter prom await page.goto(new URL("?library=team", server.url).href); const url = await page.getByLabel("Deployment MCP URL").inputValue(); expect(url).toBe(new URL("/mcp?workspace=my-project&library=team", server.url).href); + const before = await page.locator(".agent-onboarding").boundingBox(); await page.getByRole("button", { name: "Copy MCP URL", exact: true }).click(); await page.getByText("MCP URL copied", { exact: true }).waitFor(); expect(await page.evaluate(() => navigator.clipboard.readText())).toBe(url); - const prompt = await page.getByLabel("Starter prompt").inputValue(); + const prompt = await page.getByLabel("Starter prompt", { exact: true }).inputValue(); expect(prompt).toContain("artifact_guide"); expect(prompt).toContain("private access"); await page.getByRole("button", { name: "Copy starter prompt", exact: true }).click(); await page.getByText("Starter prompt copied", { exact: true }).waitFor(); expect(await page.evaluate(() => navigator.clipboard.readText())).toBe(prompt); + expect(await page.locator(".agent-onboarding").boundingBox()).toEqual(before); + if (process.env.GALLERY_SCREENSHOT_DIR) { + await page.screenshot({ path: `${process.env.GALLERY_SCREENSHOT_DIR}/onboarding-light.png` }); + await page.getByRole("button", { name: "Switch to dark theme", exact: true }).click(); + await page.screenshot({ path: `${process.env.GALLERY_SCREENSHOT_DIR}/onboarding-dark.png` }); + } + await page.getByRole("button", { name: "New artifact", exact: true }).click(); + await page.getByRole("textbox", { name: "Artifact name", exact: true }).waitFor(); } finally { await browser.close(); server.stop(true); } }, 30000); diff --git a/package.json b/package.json index f422c9a..b3d0e18 100644 --- a/package.json +++ b/package.json @@ -40,6 +40,7 @@ "docs/runtime-plugins.md", "docs/native-workers.md", "docs/native-worker-operator.md", + "docs/native-worker-fleet.md", "docs/routing.md", "docs/files.md", "src/args.ts", @@ -111,7 +112,10 @@ "cloudflare/project.ts", "cloudflare/native-worker/manifest.ts", "cloudflare/native-worker/types.ts", + "cloudflare/native-worker/fleet-contract.ts", "scripts/native-worker/operator.ts", + "scripts/native-worker/fleet.ts", + "scripts/native-worker/fleet-store.ts", "cloudflare/http.ts", "cloudflare/schedule.ts", "src/gallery/execution-controls.tsx", @@ -151,7 +155,7 @@ "test:native-worker": "tsc -p scripts/native-worker/tsconfig.json --noEmit && bun test scripts/native-worker --timeout 30000", "native-worker:operator": "bun run scripts/native-worker/operator.ts", "test:e2e": "bun test e2e/herdr.e2e.test.ts", - "test:mcp-ui": "bun test e2e/gallery-transfer.test.ts e2e/gallery-creation.test.ts e2e/gallery-onboarding.test.ts e2e/gallery-drafts.test.ts e2e/gallery-move.test.ts e2e/project-editor.test.ts e2e/mcp-app.test.ts e2e/mcp-file-editor.test.ts e2e/gallery-scripts.test.ts e2e/gallery-secrets.test.ts e2e/gallery-native-apps.test.ts e2e/gallery-navigation.test.ts e2e/gallery-live.test.ts e2e/execution-controls.test.ts e2e/browser-plugins.test.ts e2e/plugin-bridge.test.ts e2e/routing.test.ts e2e/runner-status.test.ts --timeout 30000", + "test:mcp-ui": "bun test e2e/gallery-transfer.test.ts e2e/gallery-creation.test.ts e2e/gallery-onboarding.test.ts e2e/gallery-drafts.test.ts e2e/gallery-move.test.ts e2e/project-editor.test.ts e2e/mcp-app.test.ts e2e/mcp-file-editor.test.ts e2e/gallery-scripts.test.ts e2e/gallery-secrets.test.ts e2e/gallery-native-apps.test.ts e2e/gallery-navigation.test.ts e2e/gallery-live.test.ts e2e/gallery-connection.test.ts e2e/execution-controls.test.ts e2e/browser-plugins.test.ts e2e/plugin-bridge.test.ts e2e/routing.test.ts e2e/runner-status.test.ts --timeout 30000", "typecheck": "tsc --noEmit && tsc -p examples/runner-status/tsconfig.json --noEmit && tsc -p examples/runner-status/tsconfig.local.json --noEmit", "dev:cloudflare": "wrangler dev --local --var ENVIRONMENT:local", "prepare:celld": "bun run scripts/prepare-celld.ts", @@ -172,6 +176,7 @@ "build:executable": "bun run scripts/build-executable.ts" }, "dependencies": { + "@aws-sdk/client-s3": "3.1150.0", "@base-ui/react": "^1.8.0", "@better-auth/oauth-provider": "1.7.3", "@codemirror/commands": "^6.11.0", diff --git a/scripts/native-worker/fleet-image.test.ts b/scripts/native-worker/fleet-image.test.ts new file mode 100644 index 0000000..b401597 --- /dev/null +++ b/scripts/native-worker/fleet-image.test.ts @@ -0,0 +1,66 @@ +import { expect, test } from "bun:test"; + +test.skipIf(!process.env.FLEET_OPERATOR_DOCKER_IMAGE)( + "fleet publisher image runs pinned celld and sends host releases through the authenticated singleton", + async () => { + const image = process.env.FLEET_OPERATOR_DOCKER_IMAGE!; + const token = "fleet-image-fixture-token-with-32-characters"; + let received = false; + const server = Bun.serve({ + hostname: "0.0.0.0", + port: 0, + fetch(request) { + expect(new URL(request.url).pathname).toBe("/api/native-operator/host"); + expect(request.method).toBe("POST"); + expect(request.headers.get("authorization")).toBe(`Bearer ${token}`); + received = true; + return Response.json({ ok: true }); + }, + }); + try { + const version = Bun.spawn( + [ + "docker", + "run", + "--rm", + "--entrypoint", + "/usr/local/bin/celld", + image, + "--version", + ], + { stdout: "pipe", stderr: "pipe" }, + ); + const [output] = await Promise.all([ + new Response(version.stdout).text(), + new Response(version.stderr).text(), + ]); + expect(await version.exited).toBe(0); + expect(output.trim()).toBe("celld 0.6.1"); + const job = Bun.spawn( + [ + "docker", + "run", + "--rm", + "--add-host", + "host.docker.internal:host-gateway", + "-e", + `NATIVE_CELLD_FLEET_OPERATOR_TOKEN=${token}`, + "-e", + `NATIVE_CELLD_FLEET_RELEASE_URL=http://host.docker.internal:${server.port}/api/native-operator/host`, + image, + "--request-host", + ], + { stdout: "pipe", stderr: "pipe" }, + ); + const [, diagnostic] = await Promise.all([ + new Response(job.stdout).text(), + new Response(job.stderr).text(), + ]); + expect(await job.exited, diagnostic).toBe(0); + expect(received).toBe(true); + } finally { + server.stop(true); + } + }, + 30_000, +); diff --git a/scripts/native-worker/fleet-store.ts b/scripts/native-worker/fleet-store.ts new file mode 100644 index 0000000..5837355 --- /dev/null +++ b/scripts/native-worker/fleet-store.ts @@ -0,0 +1,123 @@ +import { + CopyObjectCommand, + GetObjectCommand, + HeadObjectCommand, + ListObjectsV2Command, + PutObjectCommand, + S3Client, +} from "@aws-sdk/client-s3"; + +export type StoredObject = { body: Uint8Array; etag: string }; + +/** Bucket-relative keys use the same prefix as the celld nodes. */ +export class FleetStore { + readonly bucket: string; + readonly prefix: string; + constructor( + readonly client: S3Client, + location: string, + ) { + const url = new URL(location); + if ( + url.protocol !== "s3:" || + !url.hostname || + url.username || + url.password || + url.search || + url.hash + ) + throw new Error("Fleet operator requires an S3 bucket URL"); + this.bucket = url.hostname; + this.prefix = decodeURIComponent(url.pathname).replace(/^\/|\/$/g, ""); + if (this.prefix.split("/").some((part) => part === "." || part === "..")) + throw new Error("Invalid fleet bucket prefix"); + } + key(key: string) { + return this.prefix ? `${this.prefix}/${key}` : key; + } + async read(key: string): Promise { + try { + const result = await this.client.send( + new GetObjectCommand({ Bucket: this.bucket, Key: this.key(key) }), + ); + if (!result.Body || !result.ETag) + throw new Error("Fleet object has no body or ETag"); + return { + body: await result.Body.transformToByteArray(), + etag: result.ETag, + }; + } catch (error) { + if ( + (error as { $metadata?: { httpStatusCode?: number } }).$metadata + ?.httpStatusCode === 404 + ) + return undefined; + throw error; + } + } + async json(key: string): Promise<{ value: T; etag: string } | undefined> { + const result = await this.read(key); + return result + ? { + value: JSON.parse(new TextDecoder().decode(result.body)) as T, + etag: result.etag, + } + : undefined; + } + async put(key: string, value: unknown, expected?: string | null) { + await this.client.send( + new PutObjectCommand({ + Bucket: this.bucket, + Key: this.key(key), + Body: JSON.stringify(value), + ContentType: "application/json", + ...(expected === null + ? { IfNoneMatch: "*" } + : expected + ? { IfMatch: expected } + : {}), + }), + ); + } + async keys(prefix: string) { + const keys: string[] = []; + let continuation: string | undefined; + do { + const page = await this.client.send( + new ListObjectsV2Command({ + Bucket: this.bucket, + Prefix: this.key(prefix), + ContinuationToken: continuation, + }), + ); + for (const object of page.Contents ?? []) + if (object.Key) + keys.push(object.Key.slice(this.prefix ? this.prefix.length + 1 : 0)); + continuation = page.IsTruncated ? page.NextContinuationToken : undefined; + if (page.IsTruncated && !continuation) + throw new Error("Invalid fleet bucket pagination"); + } while (continuation); + return keys; + } + async copyImmutable(source: string, target: string) { + try { + await this.client.send( + new HeadObjectCommand({ Bucket: this.bucket, Key: this.key(target) }), + ); + return; + } catch (error) { + if ( + (error as { $metadata?: { httpStatusCode?: number } }).$metadata + ?.httpStatusCode !== 404 + ) + throw error; + } + await this.client.send( + new CopyObjectCommand({ + Bucket: this.bucket, + Key: this.key(target), + CopySource: `${this.bucket}/${this.key(source).split("/").map(encodeURIComponent).join("/")}`, + }), + ); + } +} diff --git a/scripts/native-worker/fleet.test.ts b/scripts/native-worker/fleet.test.ts new file mode 100644 index 0000000..c2b90e1 --- /dev/null +++ b/scripts/native-worker/fleet.test.ts @@ -0,0 +1,484 @@ +import { expect, test } from "bun:test"; +import { randomUUID } from "node:crypto"; +import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import { + CreateBucketCommand, + ListBucketsCommand, + S3Client, +} from "@aws-sdk/client-s3"; +import { CelldFleetOperator, fleetHostConfig } from "./fleet"; +import { FleetStore } from "./fleet-store"; +import { + fleetBinding, + readFleetRegistry, + type FleetPointer, +} from "../../cloudflare/native-worker/fleet-contract"; +import { prepareCelldConfig } from "../prepare-celld"; + +test("fleet operator authenticates requests before parsing deployments", async () => { + const operator = new CelldFleetOperator({ + bucket: "s3://fixture", + config: "/unused", + token: "fleet-fixture-token-with-32-characters", + }); + expect( + ( + await operator.fetch( + new Request("https://operator.invalid/deploy", { + method: "POST", + body: "invalid JSON", + }), + ) + ).status, + ).toBe(401); +}); + +test("fleet configuration keeps host bindings and rejects conflicting app identities", () => { + const app = "11111111-1111-4111-8111-111111111111"; + const script = `art-app-${app.replaceAll("-", "")}`; + const registry = readFleetRegistry({ + schema_version: 1, + apps: [ + { + app, + revision: "a".repeat(64), + pointer: { + script_name: script, + version: "b".repeat(16), + prefix: `deploy/${script}/${"b".repeat(16)}`, + rollout: { percent: 100 }, + }, + crons: ["0 * * * *"], + queues: [], + }, + ], + }); + const config = fleetHostConfig( + { + name: "canvas", + main: "worker.js", + services: [{ binding: "EXISTING", service: "existing" }], + vars: { HOST_ONLY: "host" }, + }, + registry, + "publication", + ); + expect(config.services?.map((item) => item.binding)).toEqual([ + "EXISTING", + fleetBinding(app), + `${fleetBinding(app)}_SCHEDULE`, + ]); + expect(config.triggers?.crons).toEqual(["0 * * * *"]); + expect(config.vars?.HOST_ONLY).toBe("host"); + expect(() => + fleetHostConfig( + { + name: "canvas", + main: "worker.js", + services: [{ binding: fleetBinding(app), service: "forged" }], + }, + registry, + "publication", + ), + ).toThrow("conflicts"); + expect(() => + readFleetRegistry({ + ...registry, + apps: [...registry.apps, registry.apps[0]], + }), + ).toThrow("Invalid native fleet app"); +}); + +function reservePort() { + const listener = Bun.listen({ + hostname: "127.0.0.1", + port: 0, + socket: { data() {} }, + }); + const port = listener.port; + listener.stop(true); + return port; +} + +test.skipIf(process.env.NATIVE_WORKER_FLEET !== "1")( + "native apps run on two celld fleet nodes while host, storage, queues, secrets and revisions survive updates", + async () => { + const directory = await mkdtemp(join(tmpdir(), "artifacts-native-fleet-")); + const binary = process.env.CELLD_BIN!; + const endpoint = + process.env.FLEET_TEST_S3_ENDPOINT ?? "http://127.0.0.1:5117"; + const bucket = `artifacts-native-test-${randomUUID().slice(0, 8)}`; + const region = "us-east-1"; + const credentials = { + accessKeyId: "artifacts-fixture", + secretAccessKey: "artifacts-fixture-secret", + }; + const client = new S3Client({ + endpoint, + region, + credentials, + forcePathStyle: true, + }); + const environment = { + AWS_ACCESS_KEY_ID: credentials.accessKeyId, + AWS_SECRET_ACCESS_KEY: credentials.secretAccessKey, + AWS_SESSION_TOKEN: "", + AWS_REGION: region, + S3_ENDPOINT: endpoint, + }; + const store = new FleetStore(client, `s3://${bucket}`); + const nodePorts = [reservePort(), reservePort()]; + const nodes = nodePorts.map((port) => `http://127.0.0.1:${port}`); + const operatorPort = reservePort(); + const operatorToken = "fleet-fixture-token-with-32-characters"; + const configPath = join(directory, "wrangler.jsonc"); + const children: ReturnType[] = []; + let logs = ""; + let operator: CelldFleetOperator | undefined; + try { + // A newly started fixture container can accept TCP before S3 is ready. + const fixtureDeadline = Date.now() + 30_000; + while (true) { + try { + await client.send(new ListBucketsCommand({})); + break; + } catch (error) { + if (Date.now() >= fixtureDeadline) throw error; + await Bun.sleep(250); + } + } + await client.send(new CreateBucketCommand({ Bucket: bucket })); + const config = await prepareCelldConfig( + resolve(import.meta.dir, "../../wrangler.jsonc"), + configPath, + { + main: resolve(import.meta.dir, "../../dist/worker-app/worker.js"), + assets: resolve(import.meta.dir, "../../dist/cloudflare/assets"), + }, + ); + config.vars = { + ...(config.vars as object), + NATIVE_CELLD_TRUSTED_APPS: "true", + NATIVE_CELLD_FLEET_OPERATOR_URL: `http://127.0.0.1:${operatorPort}/`, + NATIVE_CELLD_FLEET_OPERATOR_TOKEN: operatorToken, + HOST_ONLY_SECRET: "host-fixture-secret", + }; + await writeFile(configPath, JSON.stringify(config)); + function startNode(index: number) { + const port = nodePorts[index]!; + const peerPort = reservePort(); + const child = Bun.spawn( + [ + binary, + "--bucket", + `s3://${bucket}`, + "--endpoint", + endpoint, + "--region", + region, + "--listen", + `127.0.0.1:${port}`, + "--internal-listen", + `127.0.0.1:${peerPort}`, + "--advertise", + `127.0.0.1:${peerPort}`, + ], + { + cwd: directory, + env: { + ...process.env, + ...environment, + CELLD_NODE: `fixture-${index}-${randomUUID()}`, + CELLD_WATCH: join(directory, `node-${index}`), + // Keep one node slower than Bun's default HTTP idle timeout. + CELLD_DEPLOY_POLL_S: index === 1 ? "15" : "1", + CELLD_READY_FLEET_GATE_MS: "0", + CELLD_IDLE_EVICT_S: "60", + RUST_LOG: "warn", + }, + stdout: "pipe", + stderr: "pipe", + }, + ); + children.push(child); + for (const stream of [child.stdout, child.stderr]) + if (stream && typeof stream !== "number") + void (async () => { + for await (const chunk of stream) + logs = (logs + new TextDecoder().decode(chunk)).slice(-100_000); + })(); + return child; + } + for (const index of nodePorts.keys()) startNode(index); + operator = new CelldFleetOperator({ + bucket: `s3://${bucket}`, + config: configPath, + token: operatorToken, + binary, + endpoint, + region, + client, + environment, + nodes, + readinessTimeoutMs: 45_000, + }); + await operator.start(); + operator.serve({ port: operatorPort }); + await operator.publishHost(); + for (const node of nodes) + expect((await fetch(`${node}/health`)).status, logs).toBe(200); + expect( + (await store.json("deploy/current.json"))?.value + .script_name, + ).toBe("canvas"); + const source = ( + await readFile( + resolve(import.meta.dir, "../../examples/native-worker/worker.ts"), + "utf8", + ) + ).replace( + " const counter =", + ' if (url.pathname === "/inspect") return Response.json({ url: request.url, headers: Object.fromEntries(request.headers), bindings: Object.keys(env).sort() });\n const counter =', + ); + const manifest = JSON.parse( + await readFile( + resolve(import.meta.dir, "../../examples/native-worker/app.json"), + "utf8", + ), + ); + manifest.triggers.crons = ["* * * * *"]; + async function tool( + name: string, + args: Record, + index = 0, + library = "private", + ) { + const response = await fetch( + `${nodes[index]}/api/tools?workspace=test&library=${library}`, + { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ name, arguments: args }), + }, + ); + const result = (await response.json()) as { + isError?: boolean; + structuredContent?: Record; + content?: { text: string }[]; + }; + const payload = + result.structuredContent ?? + JSON.parse(result.content?.[0]?.text ?? "{}"); + return { result, payload }; + } + const invoke = (path = "/", index = 0, init?: RequestInit) => + fetch(`${nodes[index]}/apps/fixture${path}?workspace=test`, init); + const missing = await tool("app_write", { + name: "fixture", + source, + manifest, + provider: "celld-fleet", + expected_revision: null, + }); + expect(missing.result.isError, JSON.stringify(missing)).toBe(true); + expect(missing.payload.error).toContain("Missing declared app secret"); + const saved = await tool("app_secrets", { + name: "fixture", + secrets: { API_TOKEN: "app-fixture-secret" }, + }); + expect(saved.result.isError, JSON.stringify(saved)).not.toBe(true); + expect(await (await invoke()).json(), logs).toMatchObject({ + count: 1, + visits: 1, + hasToken: true, + }); + expect(await (await invoke("/rpc", 1)).json(), logs).toMatchObject({ + count: 2, + visits: 2, + }); + const inspection = (await ( + await invoke("/inspect", 1, { + headers: { + authorization: "Bearer forged", + cookie: "management=private", + "x-artifacts-gateway-token": "forged", + "x-artifacts-target-url": "https://forged.invalid", + }, + }) + ).json()) as any; + expect(inspection.url).toBe(`${nodes[1]}/inspect`); + expect(inspection.bindings).toEqual([ + "API_TOKEN", + "CACHE", + "COUNTERS", + "DB", + "FILES", + "GREETING", + "JOBS", + ]); + for (const name of [ + "authorization", + "cookie", + "x-artifacts-gateway-token", + "x-artifacts-target-url", + "x-artifacts-app-selection", + ]) + expect(inspection.headers[name]).toBeUndefined(); + expect(await (await invoke("/queue")).text()).toBe("queued"); + const queueDeadline = Date.now() + 15_000; + let queued: any; + do { + queued = await (await invoke("/events", 1)).json(); + if (queued.queued) break; + await Bun.sleep(100); + } while (Date.now() < queueDeadline); + expect(queued.queued?.bodies, logs).toEqual([{ message: "from fetch" }]); + const cronDeadline = Date.now() + 70_000; + let scheduled: any; + do { + scheduled = await (await invoke("/events", 1)).json(); + if (scheduled.scheduled) break; + await Bun.sleep(500); + } while (Date.now() < cronDeadline); + expect(scheduled.scheduled, logs).toMatchObject({ + cron: "* * * * *", + time: expect.any(Number), + }); + const read = (await tool("app_read", { name: "fixture" })).payload; + expect(JSON.stringify(read)).not.toContain("app-fixture-secret"); + expect(JSON.stringify(read)).not.toContain(operatorToken); + const update = await tool( + "app_write", + { + name: "fixture", + source: source.replace( + "greeting: env.GREETING", + 'greeting: "updated"', + ), + manifest, + expected_revision: read.revision_token, + }, + 1, + ); + expect(update.result.isError, JSON.stringify(update)).not.toBe(true); + expect(await (await invoke()).json(), logs).toMatchObject({ + greeting: "updated", + count: 3, + visits: 3, + }); + expect( + (await store.json("deploy/current.json"))?.value + .script_name, + ).toBe("canvas"); + await operator.publishHost(); + expect(await (await invoke("/rpc", 1)).json(), logs).toMatchObject({ + count: 4, + visits: 4, + }); + const active = (await tool("app_read", { name: "fixture" })).payload; + const invalid = await tool("app_write", { + name: "fixture", + source: "export default { fetch( {", + manifest, + expected_revision: active.revision_token, + }); + expect(invalid.result.isError, JSON.stringify(invalid)).toBe(true); + expect( + (await tool("app_read", { name: "fixture" })).payload.active_revision, + ).toBe(active.active_revision); + const team = await tool( + "app_write", + { + name: "fixture", + source, + manifest: { + ...manifest, + secrets: [], + triggers: { ...manifest.triggers, crons: [] }, + }, + provider: "celld-fleet", + expected_revision: null, + }, + 1, + "team", + ); + expect(team.result.isError, JSON.stringify(team)).not.toBe(true); + expect( + await ( + await fetch(`${nodes[0]}/apps/fixture/?workspace=test&library=team`) + ).json(), + ).toMatchObject({ count: 1, visits: 1, hasToken: false }); + expect(await (await invoke("/rpc")).json()).toMatchObject({ + count: 5, + visits: 5, + hasToken: true, + }); + const oldNode = children[0]!; + oldNode.kill("SIGTERM"); + await oldNode.exited; + startNode(0); + const restartDeadline = Date.now() + 30_000; + let ready = false; + do { + try { + ready = ( + await fetch(`${nodes[0]}/health`, { + signal: AbortSignal.timeout(1000), + }) + ).ok; + } catch {} + if (!ready) await Bun.sleep(100); + } while (!ready && Date.now() < restartDeadline); + expect(ready, logs).toBe(true); + expect(await (await invoke("/rpc")).json(), logs).toMatchObject({ + count: 6, + visits: 6, + }); + await operator.close(); + operator = new CelldFleetOperator({ + bucket: `s3://${bucket}`, + config: configPath, + token: operatorToken, + binary, + endpoint, + region, + client, + environment, + nodes, + readinessTimeoutMs: 45_000, + }); + await operator.start(); + operator.serve({ port: operatorPort }); + await operator.publishHost(); + expect(await (await invoke("/storage", 1)).json()).toMatchObject({ + cached: "6", + file: "6", + visits: 6, + }); + } catch (error) { + if (error instanceof Error && error.cause) console.error(error.cause); + throw error; + } finally { + await operator?.close(); + for (const child of children) + if (child.exitCode === null) child.kill("SIGTERM"); + await Promise.all( + children.map(async (child) => { + const stopped = await Promise.race([ + child.exited.then(() => true), + Bun.sleep(45_000).then(() => false), + ]); + if (!stopped) { + child.kill("SIGKILL"); + await child.exited; + } + }), + ); + client.destroy(); + await rm(directory, { recursive: true, force: true }); + } + }, + 240_000, +); diff --git a/scripts/native-worker/fleet.ts b/scripts/native-worker/fleet.ts new file mode 100644 index 0000000..6c18c11 --- /dev/null +++ b/scripts/native-worker/fleet.ts @@ -0,0 +1,567 @@ +import { createHash, randomUUID, timingSafeEqual } from "node:crypto"; +import { cp, mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { basename, dirname, join, resolve } from "node:path"; +import { tmpdir } from "node:os"; +import { resolve4 } from "node:dns/promises"; +import { S3Client } from "@aws-sdk/client-s3"; +import { + CELLD_VERSION, + ensureCelldRuntime, +} from "../../src/local/celld-runtime"; +import { celldConfig, validateDeployment } from "./operator"; +import { FleetStore } from "./fleet-store"; +import { + FLEET_REGISTRY_VAR, + FLEET_PUBLICATION_VAR, + FLEET_SCHEDULER_EXPORT, + fleetBinding, + fleetScheduleBinding, + readFleetRegistry, + type FleetApp, + type FleetPointer, + type FleetRegistry, +} from "../../cloudflare/native-worker/fleet-contract"; +import { + resourceName, + workerName, + type NativeDeployment, + type ProviderResult, +} from "../../cloudflare/native-worker/types"; + +type Manifest = { + script_name: string; + version: string; + raw_metadata: { bindings: { name: string; type: string; text?: string }[] }; +}; +type HostConfig = { + name: string; + main: string; + assets?: { directory: string }; + vars?: Record; + services?: { binding: string; service: string; entrypoint?: string }[]; + triggers?: { crons?: string[] }; + [key: string]: unknown; +}; +export type FleetOperatorOptions = { + bucket: string; + config: string; + token: string; + binary?: string; + esbuild?: string; + endpoint?: string; + region?: string; + client?: S3Client; + nodes?: string[]; + nodeService?: { hostname: string; port: number }; + basePath?: string; + readinessTimeoutMs?: number; + environment?: NodeJS.ProcessEnv; +}; + +// The fleet fires the primary script's crons. Invoke each app's native handler +// through a generated WorkerEntrypoint in that app's own isolate and env. +export const fleetAppWrapper = `import app from "./app.js"; +import { WorkerEntrypoint } from "cloudflare:workers"; +export * from "./app.js"; +export default app; +export class ${FLEET_SCHEDULER_EXPORT} extends WorkerEntrypoint { + async runScheduled(event) { + let noRetry = false; + const pending = []; + const context = new Proxy(this.ctx, { get(target, key) { + if (key === "waitUntil") return promise => { const task = Promise.resolve(promise); pending.push(task); target.waitUntil(task); }; + const value = Reflect.get(target, key, target); + return typeof value === "function" ? value.bind(target) : value; + } }); + const controller = { ...event, noRetry() { noRetry = true; } }; + const handler = typeof app === "function" ? new app(context, this.env) : app; + try { + if (typeof handler.scheduled !== "function") throw new Error("Missing scheduled handler"); + await handler.scheduled(controller, this.env, context); + await Promise.all(pending); + return { outcome: "ok", noRetry }; + } catch { return { outcome: "failed", noRetry }; } + } +} +`; + +export function fleetHostConfig( + config: HostConfig, + registry: FleetRegistry, + publication: string, +): HostConfig { + readFleetRegistry(registry); + const services = config.services ?? []; + if (services.some((binding) => binding.binding.startsWith("ARTIFACTS_APP_"))) + throw new Error( + "Native fleet service binding conflicts with host configuration", + ); + return { + ...config, + vars: { + ...config.vars, + [FLEET_REGISTRY_VAR]: JSON.stringify(registry), + [FLEET_PUBLICATION_VAR]: publication, + }, + services: [ + ...services, + ...registry.apps.flatMap((app) => [ + { binding: fleetBinding(app.app), service: app.pointer.script_name }, + { + binding: fleetScheduleBinding(app.app), + service: app.pointer.script_name, + entrypoint: FLEET_SCHEDULER_EXPORT, + }, + ]), + ], + triggers: { + ...config.triggers, + crons: [ + ...new Set([ + ...(config.triggers?.crons ?? []), + ...registry.apps.flatMap((app) => app.crons), + ]), + ], + }, + }; +} + +/** A single publisher owns native components and the primary Artifacts graph. */ +export class CelldFleetOperator { + private readonly store: FleetStore; + private readonly tokenHash: Buffer; + private readonly basePath: string; + private binary = ""; + private esbuild = ""; + private host!: HostConfig; + private pending: Promise = Promise.resolve(); + private stopped = false; + private server?: ReturnType; + constructor(private readonly options: FleetOperatorOptions) { + if (options.token.length < 32) + throw new Error("Operator token must contain at least 32 characters"); + this.tokenHash = createHash("sha256").update(options.token).digest(); + this.basePath = options.basePath?.replace(/\/$/, "") ?? ""; + if (this.basePath && !/^\/[a-z0-9/-]+$/.test(this.basePath)) + throw new Error("Invalid operator base path"); + this.store = new FleetStore( + options.client ?? + new S3Client({ + region: options.region ?? process.env.AWS_REGION ?? "us-east-1", + endpoint: options.endpoint, + forcePathStyle: !!options.endpoint, + }), + options.bucket, + ); + } + async start() { + const configPath = resolve(this.options.config); + this.host = Bun.JSONC.parse( + await readFile(configPath, "utf8"), + ) as HostConfig; + if (!this.host.name || !this.host.main) + throw new Error("Fleet operator requires the host Worker configuration"); + this.host.main = resolve(dirname(configPath), this.host.main); + if (this.host.assets) + this.host.assets.directory = resolve( + dirname(configPath), + this.host.assets.directory, + ); + this.binary = + this.options.binary ?? + (await ensureCelldRuntime({ + dataRoot: join(tmpdir(), "artifacts-fleet-runtime"), + })); + const version = Bun.spawn([this.binary, "--version"], { + stdout: "pipe", + stderr: "ignore", + }); + if ( + !(await new Response(version.stdout).text()) + .split(/\r?\n/) + .includes(`celld ${CELLD_VERSION}`) || + (await version.exited) !== 0 + ) + throw new Error(`Fleet operator requires celld ${CELLD_VERSION}`); + this.esbuild = + this.options.esbuild ?? + Bun.resolveSync( + `@esbuild/${process.platform}-${process.arch}/bin/esbuild`, + dirname(Bun.resolveSync("esbuild/package.json", import.meta.dir)), + ); + } + serve(options: { hostname?: string; port?: number } = {}) { + if (!this.binary || this.server) + throw new Error("Start the fleet operator once before serving"); + this.server = Bun.serve({ + hostname: options.hostname ?? "127.0.0.1", + port: options.port ?? 4792, + maxRequestBodySize: 16 * 1024 * 1024, + // Publication waits for node adoption (up to 90s) after compiling apps. + idleTimeout: 255, + fetch: (request) => this.fetch(request), + }); + return this.server; + } + async fetch(request: Request): Promise { + const supplied = + request.headers.get("authorization")?.replace(/^Bearer /, "") ?? ""; + if ( + !timingSafeEqual( + this.tokenHash, + createHash("sha256").update(supplied).digest(), + ) + ) + return Response.json({ error: "Unauthorized" }, { status: 401 }); + if (this.stopped) + return Response.json({ error: "Operator is stopping" }, { status: 503 }); + const url = new URL(request.url); + if (url.pathname === `${this.basePath}/health` && request.method === "GET") + return Response.json({ ok: true }); + try { + if ( + url.pathname === `${this.basePath}/deploy` && + request.method === "POST" + ) { + const input = validateDeployment(await request.json()); + return Response.json(await this.deploy(input)); + } + if ( + url.pathname === `${this.basePath}/host` && + request.method === "POST" + ) { + // The release hook runs before the nodes on a fresh fleet. + return Response.json(await this.publishHost(false)); + } + return Response.json({ error: "Not found" }, { status: 404 }); + } catch { + // Configuration, compiled modules and platform errors can contain secrets. + return Response.json( + { error: "Fleet publication failed; reconciliation is required" }, + { status: 503 }, + ); + } + } + private serial(operation: () => Promise) { + const result = this.pending.catch(() => {}).then(operation); + this.pending = result; + return result; + } + private async current() { + const pointer = await this.store.json("deploy/current.json"); + if (!pointer) return { etag: null, registry: readFleetRegistry(undefined) }; + if (pointer.value.script_name !== this.host.name) + throw new Error( + "Fleet default Worker differs from the configured Artifacts host", + ); + const manifest = await this.store.json( + `${pointer.value.prefix}/manifest.json`, + ); + if ( + !manifest || + manifest.value.script_name !== this.host.name || + manifest.value.version !== pointer.value.version + ) + throw new Error("Invalid fleet host deployment"); + const binding = manifest.value.raw_metadata.bindings.find( + (binding) => binding.name === FLEET_REGISTRY_VAR, + ); + return { etag: pointer.etag, registry: readFleetRegistry(binding?.text) }; + } + private async stage( + directory: string, + ): Promise<{ prefix: string; pointer: FleetPointer }> { + const prefix = `.artifacts-native-staging/${randomUUID()}`; + const command = [ + this.binary, + "deploy", + directory, + "--bucket", + `${this.options.bucket.replace(/\/$/, "")}/${prefix}`, + "--json", + ...(this.options.endpoint ? ["--endpoint", this.options.endpoint] : []), + ...(this.options.region ? ["--region", this.options.region] : []), + ]; + const process = Bun.spawn(command, { + env: { + ...globalThis.process.env, + ...this.options.environment, + CELLD_ESBUILD: this.esbuild, + }, + stdout: "pipe", + stderr: "pipe", + }); + // Drain build output without logging configuration or authored values. + const [, diagnostic] = await Promise.all([ + new Response(process.stdout).arrayBuffer(), + new Response(process.stderr).text(), + ]); + if ((await process.exited) !== 0) + throw new Error("celld candidate deployment failed", { + cause: diagnostic, + }); + const pointer = await this.store.json( + `${prefix}/deploy/current.json`, + ); + if (!pointer) throw new Error("celld candidate pointer is missing"); + const keys = await this.store.keys(`${prefix}/deploy/`); + for (const key of keys) { + const target = key.slice(prefix.length + 1); + if ( + target.endsWith("/current.json") || + target.startsWith("deploy/queues/") + ) + continue; + await this.store.copyImmutable(key, target); + } + return { prefix, pointer: pointer.value }; + } + private async candidateHost( + registry: FleetRegistry, + publication: string, + directory: string, + ) { + const config = fleetHostConfig(this.host, registry, publication); + await cp(dirname(this.host.main), join(directory, "host-code"), { + recursive: true, + }); + if (config.assets) { + await cp(config.assets.directory, join(directory, "assets"), { + recursive: true, + }); + config.assets = { ...config.assets, directory: "assets" }; + } + const module = `./host-code/${basename(this.host.main)}`; + const entry = `import host from ${JSON.stringify(module)}; +export * from ${JSON.stringify(module)}; +export default { + fetch(request, env, ctx) { return host.fetch(request, env, ctx); }, + async scheduled(controller, env, ctx) { + ${this.host.triggers?.crons?.length ? `if (${JSON.stringify(this.host.triggers.crons)}.includes(controller.cron) && host.scheduled) await host.scheduled(controller, env, ctx);` : ""} + const result = await env.NATIVE_APPS.getByName("deployment").fireCron({ cron: controller.cron, scheduledTime: controller.scheduledTime }); + if (result.failed) throw new Error("Worker app cron handler failed"); + } +};`; + await writeFile(join(directory, "host.ts"), entry, { mode: 0o600 }); + await writeFile( + join(directory, "wrangler.jsonc"), + JSON.stringify({ ...config, main: "host.ts", no_bundle: false }), + { mode: 0o600 }, + ); + return this.stage(directory); + } + private async ready(publication: string) { + if (!this.options.nodes?.length && !this.options.nodeService) return; + const deadline = Date.now() + (this.options.readinessTimeoutMs ?? 90_000); + while (Date.now() < deadline) { + const discovery = this.options.nodeService; + const nodes = discovery + ? (await resolve4(discovery.hostname)).map( + (address) => `http://${address}:${discovery.port}`, + ) + : this.options.nodes!; + const results = await Promise.all( + nodes.map(async (node) => { + try { + const response = await fetch( + new URL("/api/native-fleet/ready", node), + { + headers: { Authorization: `Bearer ${this.options.token}` }, + signal: AbortSignal.timeout(2000), + }, + ); + return ( + response.ok && + ((await response.json()) as { publication?: string }) + .publication === publication + ); + } catch { + return false; + } + }), + ); + if (results.length && results.every(Boolean)) return; + await Bun.sleep(500); + } + throw new Error("Fleet nodes did not adopt the publication"); + } + private async commit( + registry: FleetRegistry, + expected: string | null, + directory: string, + waitForNodes = true, + ) { + const publication = createHash("sha256") + .update(JSON.stringify(registry)) + .update(JSON.stringify(this.host)) + .update(await readFile(this.host.main)) + .digest("hex"); + const candidate = await this.candidateHost( + registry, + publication, + directory, + ); + await this.store.put( + `deploy/${this.host.name}/current.json`, + candidate.pointer, + ); + // Only this pointer selects the fleet's public application. Never point it + // at an authored app, and reject an intervening host deployment. + await this.store.put("deploy/current.json", candidate.pointer, expected); + if (waitForNodes) await this.ready(publication); + return publication; + } + async publishHost(waitForNodes = true) { + return this.serial(async () => { + const current = await this.current(); + const directory = await mkdtemp(join(tmpdir(), "artifacts-fleet-host-")); + try { + return { + ok: true, + publication: await this.commit( + current.registry, + current.etag, + directory, + waitForNodes, + ), + }; + } finally { + await rm(directory, { recursive: true, force: true }); + } + }); + } + async deploy(input: NativeDeployment): Promise { + if (!this.options.nodes?.length && !this.options.nodeService) + throw new Error("Configure fleet node discovery before deploying apps"); + const deployment = validateDeployment(input); + return this.serial(async () => { + const current = await this.current(); + const directory = await mkdtemp(join(tmpdir(), "artifacts-fleet-app-")); + try { + const appDirectory = join(directory, "app"); + const hostDirectory = join(directory, "host"); + await mkdir(appDirectory); + await mkdir(hostDirectory); + await writeFile( + join(appDirectory, "app.js"), + deployment.revision.code, + { mode: 0o600 }, + ); + await writeFile(join(appDirectory, "worker.js"), fleetAppWrapper, { + mode: 0o600, + }); + const config = celldConfig(deployment); + await writeFile( + join(appDirectory, "wrangler.jsonc"), + JSON.stringify({ + ...config, + vars: { ...config.vars, ...deployment.secrets }, + triggers: { crons: [] }, + }), + { mode: 0o600 }, + ); + const candidate = await this.stage(appDirectory); + if (candidate.pointer.script_name !== workerName(deployment.app)) + throw new Error("Invalid native app candidate"); + const app: FleetApp = { + app: deployment.app, + revision: deployment.revision.id, + pointer: candidate.pointer, + crons: deployment.revision.manifest.triggers.crons, + queues: Object.values(deployment.resources) + .filter((resource) => resource.type === "queue") + .map((resource) => resourceName(resource.id)), + }; + const previous = current.registry.apps.find( + (item) => item.app === app.app, + ); + // Publish native queue attachments before their service pointers, in + // the same order as celld deploy. Removed consumers retain tombstones. + for (const queue of new Set([ + ...(previous?.queues ?? []), + ...app.queues, + ])) { + const key = `deploy/queues/${queue}/consumer.json`; + const attachment = await this.store.json( + `${candidate.prefix}/${key}`, + ); + await this.store.put( + key, + attachment?.value ?? { schema_version: 1, queue, consumer: null }, + ); + } + await this.store.put( + `deploy/${app.pointer.script_name}/current.json`, + app.pointer, + ); + const registry: FleetRegistry = { + schema_version: 1, + apps: [ + ...current.registry.apps.filter((item) => item.app !== app.app), + app, + ].sort((a, b) => a.app.localeCompare(b.app)), + }; + await this.commit(registry, current.etag, hostDirectory); + return { + revision: deployment.revision.id, + endpoint: `https://native-app.internal/${deployment.app}/`, + }; + } finally { + await rm(directory, { recursive: true, force: true }); + } + }); + } + async close() { + this.stopped = true; + await this.pending.catch(() => {}); + await this.server?.stop(true); + } +} + +if (import.meta.main) { + const token = process.env.NATIVE_CELLD_FLEET_OPERATOR_TOKEN; + if (!token) throw new Error("Set NATIVE_CELLD_FLEET_OPERATOR_TOKEN"); + if (process.argv.includes("--request-host")) { + const endpoint = process.env.NATIVE_CELLD_FLEET_RELEASE_URL; + if (!endpoint) throw new Error("Set NATIVE_CELLD_FLEET_RELEASE_URL"); + const response = await fetch(endpoint, { + method: "POST", + headers: { Authorization: `Bearer ${token}` }, + signal: AbortSignal.timeout(240_000), + redirect: "manual", + }); + if (!response.ok || !((await response.json()) as { ok?: boolean }).ok) + throw new Error("Fleet host publication failed"); + process.exit(0); + } + const operator = new CelldFleetOperator({ + bucket: process.env.CELLD_BUCKET ?? "", + config: process.env.NATIVE_CELLD_FLEET_CONFIG ?? "/app/wrangler.jsonc", + token, + binary: process.env.CELLD_BIN, + esbuild: process.env.CELLD_ESBUILD, + endpoint: process.env.S3_ENDPOINT, + region: process.env.AWS_REGION, + basePath: process.env.NATIVE_CELLD_FLEET_BASE_PATH, + nodes: process.env.NATIVE_CELLD_FLEET_NODES?.split(",").filter(Boolean), + ...(process.env.NATIVE_CELLD_FLEET_NODE_SERVICE + ? { + nodeService: { + hostname: process.env.NATIVE_CELLD_FLEET_NODE_SERVICE, + port: Number(process.env.NATIVE_CELLD_FLEET_NODE_PORT ?? 4786), + }, + } + : {}), + }); + await operator.start(); + operator.serve({ + hostname: process.env.NATIVE_CELLD_FLEET_HOST ?? "127.0.0.1", + port: Number(process.env.NATIVE_CELLD_FLEET_PORT ?? 4792), + }); + const stop = async () => { + await operator.close(); + process.exit(0); + }; + process.on("SIGTERM", () => void stop()); + process.on("SIGINT", () => void stop()); +} diff --git a/src/gallery/client.tsx b/src/gallery/client.tsx index 678872e..26ce7a0 100644 --- a/src/gallery/client.tsx +++ b/src/gallery/client.tsx @@ -17,7 +17,7 @@ import { DraftProvider, confirmLeavingDrafts, useDrafts } from "./drafts"; import { ProjectImportPanel } from "./project-transfer"; import { NativeAppEditor, type NativeAppEditorHandle } from "./native-apps"; import { galleryStyles } from "./styles"; -import { subscribeGallery } from "./subscription"; +import { subscribeGallery, subscribeGalleryEvents } from "./subscription"; type Scope = "current" | "all"; type DetailTab = "preview" | "source" | "schedule" | "requests" | "settings"; @@ -144,6 +144,7 @@ function App() { const [accountError, setAccountError] = useState(""); const createdRemix = useRef<{name:string;workspace:string;kind:string} | null>(null); const galleryController = useRef(null); + const galleryLoaded = useRef(false); const galleryRequest = useRef(0); const sourceRequest = useRef(0); const previewFrame = useRef(null); @@ -166,8 +167,8 @@ function App() { const controller = new AbortController(); galleryController.current = controller; const request = ++galleryRequest.current; - setLoading(true); - setGalleryError(""); + // Background reconciliation must keep the loaded library and editors mounted. + setLoading(!galleryLoaded.current); try { let offset = 0; @@ -195,6 +196,8 @@ function App() { offset = page.nextOffset; } while (!controller.signal.aborted); if (request !== galleryRequest.current) return; + galleryLoaded.current = true; + setGalleryError(""); setGallery(previous => { const existing = new Map(previous?.artifacts.map(item => [item.key, item])); @@ -227,10 +230,14 @@ function App() { useEffect(() => { if (!gallery?.capabilities?.subscriptions) return; const url = new URL(galleryUrl(scope), window.location.href); + if (gallery.capabilities.subscriptionTransport === "events") { + url.pathname = "/api/gallery/events"; + return subscribeGalleryEvents(url.href, () => void loadGallery(), setLiveConnected); + } url.pathname = "/api/gallery/subscribe"; url.protocol = url.protocol === "https:" ? "wss:" : "ws:"; return subscribeGallery(url.href, () => void loadGallery(), setLiveConnected); - }, [gallery?.capabilities?.subscriptions, scope, loadGallery]); + }, [gallery?.capabilities?.subscriptions, gallery?.capabilities?.subscriptionTransport, scope, loadGallery]); useEffect(() => { const controller = new AbortController(); @@ -621,7 +628,7 @@ function App() { {creatingKind === "artifact" ? setCreatingKind(null)} onSaved={name => createdItem(name, "artifact")} /> : creatingKind === "script" ? setCreatingKind(null)} onSaved={name => createdItem(name, "script")} /> : selectedArtifact?.kind === "script" ? { setSelectedVersion("working"); await loadGallery(); }} /> - : !selectedArtifact && !loading && gallery?.capabilities?.links ? + : !selectedArtifact && gallery?.capabilities?.links ? startCreating("artifact")} /> : !selectedArtifact ?

{loading ? "Loading your library…" : "Your library"}

{loading ? "Your saved artifacts will appear shortly." : "Select an artifact or script to open it."}

: !resolvedVersion ?

No readable version

This artifact has no saved source available to preview.

: <> @@ -635,7 +642,7 @@ function App() { : source.status === "error" ?

{source.error}

:

Loading source…

} : null} - {gallery?.capabilities?.links ? : null} + {gallery?.capabilities?.links ? : null} {gallery?.capabilities?.links ? : null} } diff --git a/src/gallery/execution-controls.tsx b/src/gallery/execution-controls.tsx index a6f0ea5..12ae16d 100644 --- a/src/gallery/execution-controls.tsx +++ b/src/gallery/execution-controls.tsx @@ -96,7 +96,7 @@ export function ExecutionControls({ workspace, name, kind = "script", active = t {duplicates ?

Repeated headers are preserved until you edit this field.

: null} {method !== "GET" && method !== "HEAD" ?