Registry hashes chunked uploads as they arrive; the Live test runs on the deployed runner image (#138)
The registry hashes each chunk of an upload as it arrives and saves the hash state in registry_uploads.hash_state (migration 0023), so the final PUT of a chunked upload no longer reads the whole blob back from R2. That read is what made docker push time out on a 150 MB layer. The relay now logs each chunked upload, and a failed docker push prints the relay log's last lines. The deploy sets the Worker var RUNNER_IMAGE, which GET /meta returns as runner_image. Before its Docker and registry checks, the Live test dispatches a probe workflow until three jobs in a row run on that image. It then asserts that every job in its main run reported that image. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Ruyx3Qr4jYrZEpPi466deb
Canberk Demirci authored · GitHub committed signed1 parent e15c34ecommit d10062fc561a7a13fc5b4084a725ab6195704c2bBrowse files
16 files changed+305−25
Files (16)
- .github/workflows/deploy.yml+1−1
- .github/workflows/live-test.yml+61−0
- apps/diffinery/docs/actions/runners.md+1−1
- apps/diffinery/migrations/0023_upload_hash_state.sql+3−0
- apps/diffinery/src/docs/api-meta.ts+1−1
- apps/diffinery/src/env.ts+2−0
- apps/diffinery/src/http/api.ts+1−1
- apps/diffinery/src/lib/sha256.ts+107−0
- apps/diffinery/src/registry/store.ts+5−3
- apps/diffinery/src/registry/upload.ts+9−11
- apps/diffinery/test/unit/registry-upload.test.ts+83−0
- containers/runner/image/docker+11−0
- containers/runner/src/relay.ts+7−0
- containers/runner/test/relay.test.ts+5−1
- scripts/render-config.mjs+2−0
- scripts/wait-container-rollout.sh+6−6
modified.github/workflows/deploy.yml+1−1View file
| @@ -68,7 +68,7 @@ | ||
| 68 | 68 | run: pnpm exec wrangler deploy -c wrangler.deploy.jsonc |
| 69 | 69 | |
| 70 | 70 | - name: Wait for new containers to get the new image |
| 71 | - # The Live test follows this workflow, and its jobs must not start on the old image. | |
| 71 | + # The Live test follows this workflow, and its checks must run on the new image. | |
| 72 | 72 | # This job runs on an old-image container, so finishing after the rollout's last step |
| 73 | 73 | # also shows that a deploy doesn't stop a running job. |
| 74 | 74 | run: scripts/wait-container-rollout.sh |
modified.github/workflows/live-test.yml+61−0View file
| @@ -164,6 +164,61 @@ | ||
| 164 | 164 | api POST "/repos/$OWNER/$REPO/hooks" "{\"config\":{\"url\":\"https://example.com/diffinery-live-test\"},\"events\":[\"push\"]}" >/dev/null |
| 165 | 165 | echo "created $OWNER/$REPO" |
| 166 | 166 | |
| 167 | + - name: Wait for jobs to start on the deployed runner image | |
| 168 | + if: steps.token.outputs.skip != 'true' | |
| 169 | + run: | | |
| 170 | + set -euo pipefail | |
| 171 | + # A deploy's container rollout updates instances over a while after it starts, and an | |
| 172 | + # instance not yet updated starts jobs on the previous runner image. Jobs report their | |
| 173 | + # image (`runner.image`), so a probe workflow runs until three jobs in a row land on the | |
| 174 | + # image the deploy set (/meta's runner_image) before the Docker and registry checks. | |
| 175 | + api() { curl -fsS -X "$1" "$DIFFINERY/api/v3$2" -H "Authorization: Bearer $TOKEN" -H "content-type: application/json" ${3:+--data "$3"}; } | |
| 176 | + expected=$(api GET /meta | jq -r '.runner_image // empty') | |
| 177 | + if [ -z "$expected" ]; then | |
| 178 | + echo "::warning::/meta doesn't name the deployed runner image, so jobs aren't checked against it" | |
| 179 | + exit 0 | |
| 180 | + fi | |
| 181 | + echo "DEPLOYED_RUNNER_IMAGE=$expected" >> "$GITHUB_ENV" | |
| 182 | + mkdir -p src/.github/workflows && cd src | |
| 183 | + git init -q -b main | |
| 184 | + git config user.name "Live test" && git config user.email "live-test@diffinery.com" | |
| 185 | + cat > .github/workflows/probe.yml <<'EOF' | |
| 186 | + on: workflow_dispatch | |
| 187 | + jobs: | |
| 188 | + probe: | |
| 189 | + runs-on: ubuntu-latest | |
| 190 | + strategy: | |
| 191 | + matrix: | |
| 192 | + n: [1, 2, 3] | |
| 193 | + steps: | |
| 194 | + - run: echo "probe ok" | |
| 195 | + EOF | |
| 196 | + git add -A && git commit -q -m "Runner image probe" | |
| 197 | + git push -q "https://x-token:${TOKEN}@${DIFFINERY#https://}/$OWNER/$REPO.git" main | |
| 198 | + deadline=$(( $(date +%s) + 15 * 60 )) | |
| 199 | + while :; do | |
| 200 | + before=$(api GET "/repos/$OWNER/$REPO/actions/runs?event=workflow_dispatch" | jq '.workflow_runs | length') | |
| 201 | + api POST "/repos/$OWNER/$REPO/actions/workflows/probe.yml/dispatches" '{"ref":"main"}' >/dev/null | |
| 202 | + run="" | |
| 203 | + for i in $(seq 1 60); do | |
| 204 | + runs=$(api GET "/repos/$OWNER/$REPO/actions/runs?event=workflow_dispatch") | |
| 205 | + if [ "$(jq '.workflow_runs | length' <<<"$runs")" -gt "$before" ]; then | |
| 206 | + run=$(jq -c '.workflow_runs[0]' <<<"$runs") | |
| 207 | + [ "$(jq -r .status <<<"$run")" = completed ] && break | |
| 208 | + fi | |
| 209 | + sleep 5 | |
| 210 | + done | |
| 211 | + [ -n "$run" ] || { echo "::error::the probe workflow didn't start"; exit 1; } | |
| 212 | + images=$(api GET "/repos/$OWNER/$REPO/actions/runs/$(jq -r .id <<<"$run")/jobs" | jq -r '.jobs[] | .runner.image // "no image reported"') | |
| 213 | + echo "probe jobs ran on: $(sort <<<"$images" | uniq -c | sed 's/^ *//' | paste -sd ',')" | |
| 214 | + [ "$(grep -cxF "$expected" <<<"$images" || true)" = 3 ] && { echo "jobs start on $expected"; exit 0; } | |
| 215 | + if [ "$(date +%s)" -ge "$deadline" ]; then | |
| 216 | + echo "::error::15 minutes after the deploy, jobs still start on a runner image other than the deployed $expected" | |
| 217 | + exit 1 | |
| 218 | + fi | |
| 219 | + sleep 20 | |
| 220 | + done | |
| 221 | + | |
| 167 | 222 | - name: Push workflows and an LFS file |
| 168 | 223 | id: push |
| 169 | 224 | if: steps.token.outputs.skip != 'true' |
| @@ -361,6 +416,12 @@ | ||
| 361 | 416 | echo "::error::the push run did not succeed" |
| 362 | 417 | exit 1 |
| 363 | 418 | fi |
| 419 | + if [ -n "${DEPLOYED_RUNNER_IMAGE:-}" ]; then | |
| 420 | + # Every job ran the deployed runner, so these checks test what was just deployed. | |
| 421 | + stale=$(jq -r --arg want "$DEPLOYED_RUNNER_IMAGE" '.jobs[] | select(.runner.image != $want) | "\(.name) on \(.runner.image // "no image reported")"' <<<"$jobs") | |
| 422 | + [ -z "$stale" ] || { echo "::error::jobs ran on a runner image other than the deployed $DEPLOYED_RUNNER_IMAGE: $(paste -sd ';' <<<"$stale")"; exit 1; } | |
| 423 | + echo "every job ran on $DEPLOYED_RUNNER_IMAGE" | |
| 424 | + fi | |
| 364 | 425 | for job in $(jq -r '.jobs[] | select(.name | startswith("hosted")) | .id' <<<"$jobs"); do |
| 365 | 426 | log=$(api "/repos/$OWNER/$REPO/actions/jobs/$job/logs") |
| 366 | 427 | grep -F "live ok" <<<"$log" | grep -F "# $REPO" || { echo "::error::job $job did not print its line (checkout or run step failed)"; exit 1; } |
modifiedapps/diffinery/docs/actions/runners.md+1−1View file
| @@ -146,7 +146,7 @@ | ||
| 146 | 146 | |
| 147 | 147 | If the container stops unexpectedly (for example it runs out of memory), the job fails with `The runner stopped unexpectedly (exit code N).` A container that stops before its runner fetched the job is replaced instead, since no step ran. |
| 148 | 148 | |
| 149 | -The log's first lines name where the job ran, for example `Runner: image diffinery-runner:96dbbe31790f, in fra06, EEUR, container 6f1c…`. The jobs API returns the same as `runner` (`image`, `location`, `instance`), next to `runner_name`. | |
| 149 | +The log's first lines name where the job ran, for example `Runner: image diffinery-runner:96dbbe31790f, in fra06, EEUR, container 6f1c…`. The jobs API returns the same as `runner` (`image`, `location`, `instance`), next to `runner_name`. `GET /meta` names the image the last deploy set as `runner_image`; while a deploy's rollout is updating instances, a job can still start on the previous one. | |
| 150 | 150 | |
| 151 | 151 | ### Windows and macOS |
| 152 | 152 |
addedapps/diffinery/migrations/0023_upload_hash_state.sql+3−0View file
| @@ -1,0 +1,3 @@ | ||
| 1 | +-- SHA-256 progress of a blob upload in progress (registry/upload.ts), so a chunked upload's | |
| 2 | +-- last request doesn't read the whole blob back to hash it. | |
| 3 | +ALTER TABLE registry_uploads ADD COLUMN hash_state TEXT; |
modifiedapps/diffinery/src/docs/api-meta.ts+1−1View file
| @@ -122,7 +122,7 @@ | ||
| 122 | 122 | export const API_DOCS: Record<string, OpDoc> = { |
| 123 | 123 | // ------------------------------------------------------------ meta, users and organizations |
| 124 | 124 | "GET /": { summary: "API root", description: "Names the API and links to these docs and the MCP endpoint.", access: "public" }, |
| 125 | - "GET /meta": { summary: "Instance metadata", description: "Whether this instance issues OIDC tokens to workflows.", access: "public" }, | |
| 125 | + "GET /meta": { summary: "Instance metadata", description: "Whether this instance issues OIDC tokens to workflows, and the runner image its last deploy set (`runner_image`; each job reports the image it ran on as `runner.image`).", access: "public" }, | |
| 126 | 126 | "GET /rate_limit": { |
| 127 | 127 | summary: "Get rate limit status", |
| 128 | 128 | description: "For GitHub clients that check their budget before calling. It always reports a full, constant budget (15000 core requests an hour). Diffinery's real limits are per minute (2,000 requests per account, 60 new repositories, issues, pull requests, comments and the like, 300 requests per IP address without signing in) and answer 429 with `retry-after: 60`; see /docs/reference/rest-api#rate-limits.", |
modifiedapps/diffinery/src/env.ts+2−0View file
| @@ -74,6 +74,8 @@ | ||
| 74 | 74 | ACCESS_AUD: string; |
| 75 | 75 | /** Artifact storage per repository in bytes (live artifacts plus uploads in progress); default 10 GiB. */ |
| 76 | 76 | ARTIFACT_QUOTA_BYTES?: string; |
| 77 | + /** The runner image the deploy pointed containers at, as jobs report it (scripts/render-config.mjs). */ | |
| 78 | + RUNNER_IMAGE?: string; | |
| 77 | 79 | /** Local development only: act as this email when Access is not configured. */ |
| 78 | 80 | DEV_AUTH?: string; |
| 79 | 81 | /** |
modifiedapps/diffinery/src/http/api.ts+1−1View file
| @@ -64,7 +64,7 @@ | ||
| 64 | 64 | // ---------------------------------------------------------------- meta & users |
| 65 | 65 | |
| 66 | 66 | api.get("/", (c) => c.json({ name: "Diffinery", api: "v3", docs: `${c.env.PUBLIC_URL}/docs/api`, mcp: `${c.env.PUBLIC_URL}/mcp` })); |
| 67 | -api.get("/meta", (c) => c.json({ name: "Diffinery", oidc: !!c.env.OIDC_PRIVATE_KEY })); | |
| 67 | +api.get("/meta", (c) => c.json({ name: "Diffinery", oidc: !!c.env.OIDC_PRIVATE_KEY, runner_image: c.env.RUNNER_IMAGE ?? null })); | |
| 68 | 68 | /** Site admin: the workflow scheduler's state, to tell a scheduler that stopped ticking from a run that failed to start. */ |
| 69 | 69 | api.get("/diffinery/admin/scheduler", async (c) => { |
| 70 | 70 | if (!requireAccount(c).site_admin) throw new HttpError(403, "site administrators only"); |
addedapps/diffinery/src/lib/sha256.ts+107−0View file
| @@ -1,0 +1,107 @@ | ||
| 1 | +/** | |
| 2 | + * SHA-256 whose progress can be saved between requests. WebCrypto's digest can't be resumed, | |
| 3 | + * so a blob uploaded in several requests (the registry's chunked uploads) would otherwise be | |
| 4 | + * read back whole to hash it when the last request arrives. | |
| 5 | + */ | |
| 6 | +const K = new Uint32Array([ | |
| 7 | + 0x428a2f98, 0x71374491, 0xb5c0fbcf, 0xe9b5dba5, 0x3956c25b, 0x59f111f1, 0x923f82a4, 0xab1c5ed5, 0xd807aa98, 0x12835b01, 0x243185be, 0x550c7dc3, 0x72be5d74, 0x80deb1fe, 0x9bdc06a7, 0xc19bf174, | |
| 8 | + 0xe49b69c1, 0xefbe4786, 0x0fc19dc6, 0x240ca1cc, 0x2de92c6f, 0x4a7484aa, 0x5cb0a9dc, 0x76f988da, 0x983e5152, 0xa831c66d, 0xb00327c8, 0xbf597fc7, 0xc6e00bf3, 0xd5a79147, 0x06ca6351, 0x14292967, | |
| 9 | + 0x27b70a85, 0x2e1b2138, 0x4d2c6dfc, 0x53380d13, 0x650a7354, 0x766a0abb, 0x81c2c92e, 0x92722c85, 0xa2bfe8a1, 0xa81a664b, 0xc24b8b70, 0xc76c51a3, 0xd192e819, 0xd6990624, 0xf40e3585, 0x106aa070, | |
| 10 | + 0x19a4c116, 0x1e376c08, 0x2748774c, 0x34b0bcb5, 0x391c0cb3, 0x4ed8aa4a, 0x5b9cca4f, 0x682e6ff3, 0x748f82ee, 0x78a5636f, 0x84c87814, 0x8cc70208, 0x90befffa, 0xa4506ceb, 0xbef9a3f7, 0xc67178f2, | |
| 11 | +]); | |
| 12 | +const INIT = [0x6a09e667, 0xbb67ae85, 0x3c6ef372, 0xa54ff53a, 0x510e527f, 0x9b05688c, 0x1f83d9ab, 0x5be0cd19]; | |
| 13 | + | |
| 14 | +export class Sha256 { | |
| 15 | + private h = new Uint32Array(INIT); | |
| 16 | + private w = new Uint32Array(64); | |
| 17 | + /** Bytes of an unfinished 64-byte block. */ | |
| 18 | + private block = new Uint8Array(64); | |
| 19 | + private fill = 0; | |
| 20 | + private length = 0; | |
| 21 | + | |
| 22 | + /** Continues from `save()`'s output. */ | |
| 23 | + static restore(saved: string): Sha256 { | |
| 24 | + const s = JSON.parse(saved) as { h: number[]; n: number; b: string }; | |
| 25 | + if (!Array.isArray(s.h) || s.h.length !== 8 || !Number.isSafeInteger(s.n) || s.n < 0) throw new Error("invalid SHA-256 state"); | |
| 26 | + const x = new Sha256(); | |
| 27 | + x.h.set(s.h); | |
| 28 | + x.length = s.n; | |
| 29 | + const b = Uint8Array.from(atob(s.b), (ch) => ch.charCodeAt(0)); | |
| 30 | + if (b.length !== s.n % 64) throw new Error("invalid SHA-256 state"); | |
| 31 | + x.block.set(b); | |
| 32 | + x.fill = b.length; | |
| 33 | + return x; | |
| 34 | + } | |
| 35 | + | |
| 36 | + save(): string { | |
| 37 | + return JSON.stringify({ h: [...this.h], n: this.length, b: btoa(String.fromCharCode(...this.block.subarray(0, this.fill))) }); | |
| 38 | + } | |
| 39 | + | |
| 40 | + update(data: Uint8Array): this { | |
| 41 | + this.length += data.length; | |
| 42 | + let off = 0; | |
| 43 | + if (this.fill) { | |
| 44 | + const n = Math.min(64 - this.fill, data.length); | |
| 45 | + this.block.set(data.subarray(0, n), this.fill); | |
| 46 | + this.fill += n; | |
| 47 | + off = n; | |
| 48 | + if (this.fill < 64) return this; | |
| 49 | + this.compress(this.block, 0); | |
| 50 | + this.fill = 0; | |
| 51 | + } | |
| 52 | + for (; off + 64 <= data.length; off += 64) this.compress(data, off); | |
| 53 | + if (off < data.length) { | |
| 54 | + this.block.set(data.subarray(off), 0); | |
| 55 | + this.fill = data.length - off; | |
| 56 | + } | |
| 57 | + return this; | |
| 58 | + } | |
| 59 | + | |
| 60 | + /** The digest as lowercase hex. The object can't be updated afterwards. */ | |
| 61 | + hex(): string { | |
| 62 | + const bits = this.length * 8; | |
| 63 | + const pad = new Uint8Array((this.fill < 56 ? 56 : 120) - this.fill + 8); | |
| 64 | + pad[0] = 0x80; | |
| 65 | + const v = new DataView(pad.buffer); | |
| 66 | + v.setUint32(pad.length - 8, Math.floor(bits / 2 ** 32)); | |
| 67 | + v.setUint32(pad.length - 4, bits >>> 0); | |
| 68 | + this.update(pad); | |
| 69 | + return [...this.h].map((x) => x.toString(16).padStart(8, "0")).join(""); | |
| 70 | + } | |
| 71 | + | |
| 72 | + private compress(p: Uint8Array, o: number): void { | |
| 73 | + const w = this.w; | |
| 74 | + for (let i = 0; i < 16; i++, o += 4) w[i] = (p[o]! << 24) | (p[o + 1]! << 16) | (p[o + 2]! << 8) | p[o + 3]!; | |
| 75 | + for (let i = 16; i < 64; i++) { | |
| 76 | + const a = w[i - 15]!; | |
| 77 | + const b = w[i - 2]!; | |
| 78 | + const s0 = ((a >>> 7) | (a << 25)) ^ ((a >>> 18) | (a << 14)) ^ (a >>> 3); | |
| 79 | + const s1 = ((b >>> 17) | (b << 15)) ^ ((b >>> 19) | (b << 13)) ^ (b >>> 10); | |
| 80 | + w[i] = (w[i - 16]! + s0 + w[i - 7]! + s1) | 0; | |
| 81 | + } | |
| 82 | + const h = this.h; | |
| 83 | + let a = h[0]!, b = h[1]!, c = h[2]!, d = h[3]!, e = h[4]!, f = h[5]!, g = h[6]!, hh = h[7]!; | |
| 84 | + for (let i = 0; i < 64; i++) { | |
| 85 | + const S1 = ((e >>> 6) | (e << 26)) ^ ((e >>> 11) | (e << 21)) ^ ((e >>> 25) | (e << 7)); | |
| 86 | + const t1 = (hh + S1 + ((e & f) ^ (~e & g)) + K[i]! + w[i]!) | 0; | |
| 87 | + const S0 = ((a >>> 2) | (a << 30)) ^ ((a >>> 13) | (a << 19)) ^ ((a >>> 22) | (a << 10)); | |
| 88 | + const t2 = (S0 + ((a & b) ^ (a & c) ^ (b & c))) | 0; | |
| 89 | + hh = g; | |
| 90 | + g = f; | |
| 91 | + f = e; | |
| 92 | + e = (d + t1) | 0; | |
| 93 | + d = c; | |
| 94 | + c = b; | |
| 95 | + b = a; | |
| 96 | + a = (t1 + t2) | 0; | |
| 97 | + } | |
| 98 | + h[0] = (h[0]! + a) | 0; | |
| 99 | + h[1] = (h[1]! + b) | 0; | |
| 100 | + h[2] = (h[2]! + c) | 0; | |
| 101 | + h[3] = (h[3]! + d) | 0; | |
| 102 | + h[4] = (h[4]! + e) | 0; | |
| 103 | + h[5] = (h[5]! + f) | 0; | |
| 104 | + h[6] = (h[6]! + g) | 0; | |
| 105 | + h[7] = (h[7]! + hh) | 0; | |
| 106 | + } | |
| 107 | +} |
modifiedapps/diffinery/src/registry/store.ts+5−3View file
| @@ -42,6 +42,8 @@ | ||
| 42 | 42 | size: number; |
| 43 | 43 | tail: number; |
| 44 | 44 | digest: string | null; |
| 45 | + /** SHA-256 progress over the bytes so far (lib/sha256.ts), null once finished or for older uploads. */ | |
| 46 | + hash_state: string | null; | |
| 45 | 47 | created_at: string; |
| 46 | 48 | updated_at: string; |
| 47 | 49 | } |
| @@ -201,7 +203,7 @@ | ||
| 201 | 203 | async createUpload(packageId: string): Promise<UploadRow> { |
| 202 | 204 | const id = newId(); |
| 203 | 205 | const t = now(); |
| 204 | - const row: UploadRow = { id, package_id: packageId, r2_key: blobKey(id), multipart_id: null, parts: "[]", size: 0, tail: 0, digest: null, created_at: t, updated_at: t }; | |
| 206 | + const row: UploadRow = { id, package_id: packageId, r2_key: blobKey(id), multipart_id: null, parts: "[]", size: 0, tail: 0, digest: null, hash_state: null, created_at: t, updated_at: t }; | |
| 205 | 207 | await this.db.prepare("INSERT INTO registry_uploads (id, package_id, r2_key, parts, size, tail, created_at, updated_at) VALUES (?, ?, ?, '[]', 0, 0, ?, ?)").bind(id, packageId, row.r2_key, t, t).run(); |
| 206 | 208 | return row; |
| 207 | 209 | } |
| @@ -212,8 +214,8 @@ | ||
| 212 | 214 | |
| 213 | 215 | async saveUpload(u: UploadRow): Promise<void> { |
| 214 | 216 | await this.db |
| 215 | - .prepare("UPDATE registry_uploads SET multipart_id = ?, parts = ?, size = ?, tail = ?, digest = ?, updated_at = ? WHERE id = ?") | |
| 216 | - .bind(u.multipart_id, u.parts, u.size, u.tail, u.digest, now(), u.id) | |
| 217 | + .prepare("UPDATE registry_uploads SET multipart_id = ?, parts = ?, size = ?, tail = ?, digest = ?, hash_state = ?, updated_at = ? WHERE id = ?") | |
| 218 | + .bind(u.multipart_id, u.parts, u.size, u.tail, u.digest, u.hash_state, now(), u.id) | |
| 217 | 219 | .run(); |
| 218 | 220 | } |
| 219 | 221 |
modifiedapps/diffinery/src/registry/upload.ts+9−11View file
| @@ -1,4 +1,5 @@ | ||
| 1 | 1 | import type { Env } from "../env.js"; |
| 2 | +import { Sha256 } from "../lib/sha256.js"; | |
| 2 | 3 | import { tailKey, type UploadRow } from "./store.js"; |
| 3 | 4 | |
| 4 | 5 | /** |
| @@ -15,8 +16,10 @@ | ||
| 15 | 16 | const hex = (b: ArrayBuffer) => [...new Uint8Array(b)].map((x) => x.toString(16).padStart(2, "0")).join(""); |
| 16 | 17 | |
| 17 | 18 | /** |
| 18 | - * Appends a request body to an upload. When the upload started with this request the | |
| 19 | - * SHA-256 is computed on the way through; chunked uploads are hashed when they finish. | |
| 19 | + * Appends a request body to an upload, hashing it on the way through. The hash's progress | |
| 20 | + * is kept with the upload, so a chunked upload's last request doesn't read the whole blob | |
| 21 | + * back to hash it (for a big layer that took longer than clients wait for a response). | |
| 22 | + * Uploads started before the state was kept are hashed when they finish. | |
| 20 | 23 | */ |
| 21 | 24 | export async function ingest(env: Env, u: UploadRow, body: ReadableStream<Uint8Array> | null, final: boolean): Promise<UploadRow> { |
| 22 | 25 | const bucket = env.BUCKET; |
| @@ -35,8 +38,7 @@ | ||
| 35 | 38 | fill = bytes.length; |
| 36 | 39 | } |
| 37 | 40 | |
| 38 | - const hasher = u.size === 0 ? new crypto.DigestStream("SHA-256") : null; | |
| 39 | - const hw = hasher?.getWriter(); | |
| 41 | + const hasher = u.size === 0 ? new Sha256() : u.hash_state ? Sha256.restore(u.hash_state) : null; | |
| 40 | 42 | let added = 0; |
| 41 | 43 | const flush = async () => { |
| 42 | 44 | mp ??= await bucket.createMultipartUpload(u.r2_key, { httpMetadata: { contentType: "application/octet-stream" } }); |
| @@ -52,7 +54,7 @@ | ||
| 52 | 54 | if (done) break; |
| 53 | 55 | if (!value?.length) continue; |
| 54 | 56 | added += value.length; |
| 55 | - if (hw) await hw.write(value); | |
| 57 | + hasher?.update(value); | |
| 56 | 58 | for (let off = 0; off < value.length; ) { |
| 57 | 59 | const n = Math.min(PART_SIZE - fill, value.length - off); |
| 58 | 60 | buf.set(value.subarray(off, off + n), fill); |
| @@ -62,11 +64,7 @@ | ||
| 62 | 64 | } |
| 63 | 65 | } |
| 64 | 66 | } |
| 65 | - let digest = u.digest; | |
| 66 | - if (hw && hasher) { | |
| 67 | - await hw.close(); | |
| 68 | - digest = `sha256:${hex(await hasher.digest)}`; | |
| 69 | - } else if (added > 0) digest = null; | |
| 67 | + const digest = final && hasher ? `sha256:${hasher.hex()}` : null; | |
| 70 | 68 | |
| 71 | 69 | const hadTail = !!u.multipart_id && u.tail > 0; |
| 72 | 70 | if (final) { |
| @@ -85,7 +83,7 @@ | ||
| 85 | 83 | } |
| 86 | 84 | // A small upload that grew into a multipart one: its first bytes went into part 1, so the |
| 87 | 85 | // single object written by an earlier chunk is replaced when the multipart upload completes. |
| 88 | - return { ...u, multipart_id: mp?.uploadId ?? null, parts: JSON.stringify(parts), size: u.size + added, tail: mp && !final ? fill : 0, digest }; | |
| 86 | + return { ...u, multipart_id: mp?.uploadId ?? null, parts: JSON.stringify(parts), size: u.size + added, tail: mp && !final ? fill : 0, digest, hash_state: hasher && !final ? hasher.save() : null }; | |
| 89 | 87 | } |
| 90 | 88 | |
| 91 | 89 | /** SHA-256 of a stored object, for uploads that arrived in several requests. */ |
addedapps/diffinery/test/unit/registry-upload.test.ts+83−0View file
| @@ -1,0 +1,83 @@ | ||
| 1 | +import { createHash } from "node:crypto"; | |
| 2 | +import { describe, expect, it } from "vitest"; | |
| 3 | +import type { Env } from "../../src/env.js"; | |
| 4 | +import { Sha256 } from "../../src/lib/sha256.js"; | |
| 5 | +import { tailKey, type UploadRow } from "../../src/registry/store.js"; | |
| 6 | +import { ingest, PART_SIZE } from "../../src/registry/upload.js"; | |
| 7 | + | |
| 8 | +const sha = (b: Uint8Array) => createHash("sha256").update(b).digest("hex"); | |
| 9 | +const filled = (n: number, seed: number) => { | |
| 10 | + const b = new Uint8Array(n); | |
| 11 | + for (let i = 0; i < n; i++) b[i] = (i * 131 + seed * 7 + (i >> 8)) & 255; | |
| 12 | + return b; | |
| 13 | +}; | |
| 14 | + | |
| 15 | +describe("SHA-256 with saved progress", () => { | |
| 16 | + it("matches node's SHA-256 however the input is split and resumed", () => { | |
| 17 | + for (const n of [0, 1, 55, 56, 63, 64, 65, 119, 120, 1000, 100_000]) { | |
| 18 | + const data = filled(n, n); | |
| 19 | + let h = new Sha256(); | |
| 20 | + for (let o = 0, step = 1; o < n; o += step, step = (step * 7 + 3) % 97 || 1) h = Sha256.restore(h.update(data.subarray(o, o + step)).save()); | |
| 21 | + expect(h.hex(), `${n} bytes`).toBe(sha(data)); | |
| 22 | + } | |
| 23 | + expect(() => Sha256.restore(JSON.stringify({ h: [1], n: 0, b: "" }))).toThrow(); | |
| 24 | + }); | |
| 25 | +}); | |
| 26 | + | |
| 27 | +describe("chunked blob uploads", () => { | |
| 28 | + it("hash each chunk as it arrives, so the final request doesn't read the blob back", async () => { | |
| 29 | + const objects = new Map<string, Uint8Array>(); | |
| 30 | + const parts = new Map<string, Map<number, Uint8Array>>(); | |
| 31 | + const reads: string[] = []; | |
| 32 | + const bytes = async (v: ReadableStream | ArrayBuffer | ArrayBufferView) => (v instanceof ArrayBuffer ? new Uint8Array(v) : ArrayBuffer.isView(v) ? new Uint8Array(v.buffer, v.byteOffset, v.byteLength).slice() : new Uint8Array(await new Response(v).arrayBuffer())); | |
| 33 | + const handle = (key: string, uploadId: string) => ({ | |
| 34 | + uploadId, | |
| 35 | + async uploadPart(n: number, v: ArrayBufferView) { | |
| 36 | + parts.get(uploadId)!.set(n, await bytes(v)); | |
| 37 | + return { partNumber: n, etag: `p${n}` }; | |
| 38 | + }, | |
| 39 | + async complete(list: Array<{ partNumber: number }>) { | |
| 40 | + const all = list.map((p) => parts.get(uploadId)!.get(p.partNumber)!); | |
| 41 | + const out = new Uint8Array(all.reduce((a, b) => a + b.length, 0)); | |
| 42 | + let at = 0; | |
| 43 | + for (const p of all) out.set(p, (at += p.length) - p.length); | |
| 44 | + objects.set(key, out); | |
| 45 | + }, | |
| 46 | + }); | |
| 47 | + const bucket = { | |
| 48 | + async get(key: string) { | |
| 49 | + reads.push(key); | |
| 50 | + const o = objects.get(key); | |
| 51 | + return o ? { arrayBuffer: async () => o.slice().buffer, body: new Response(o).body, size: o.length } : null; | |
| 52 | + }, | |
| 53 | + async put(key: string, v: ArrayBufferView) { | |
| 54 | + objects.set(key, await bytes(v)); | |
| 55 | + }, | |
| 56 | + async delete(keys: string | string[]) { | |
| 57 | + for (const k of [keys].flat()) objects.delete(k); | |
| 58 | + }, | |
| 59 | + async createMultipartUpload(key: string) { | |
| 60 | + parts.set("mp1", new Map()); | |
| 61 | + return handle(key, "mp1"); | |
| 62 | + }, | |
| 63 | + resumeMultipartUpload: (key: string, id: string) => handle(key, id), | |
| 64 | + }; | |
| 65 | + const env = { BUCKET: bucket } as unknown as Env; | |
| 66 | + const blob = filled(2 * PART_SIZE + 3 * 1024 * 1024 + 17, 1); | |
| 67 | + // Chunks as the registry relay sends them: PATCHes, then the rest with the final PUT. | |
| 68 | + const cuts = [PART_SIZE + 12345, 2 * PART_SIZE - 1, blob.length]; | |
| 69 | + let u: UploadRow = { id: "up1", package_id: "p", r2_key: "registry/blobs/up1", multipart_id: null, parts: "[]", size: 0, tail: 0, digest: null, hash_state: null, created_at: "", updated_at: "" }; | |
| 70 | + let from = 0; | |
| 71 | + for (const [i, to] of cuts.entries()) { | |
| 72 | + u = await ingest(env, u, new Response(blob.subarray(from, to)).body, i === cuts.length - 1); | |
| 73 | + // Each request starts from the row as stored, so only saved state carries over. | |
| 74 | + u = JSON.parse(JSON.stringify(u)) as UploadRow; | |
| 75 | + from = to; | |
| 76 | + } | |
| 77 | + expect(u.digest).toBe(`sha256:${sha(blob)}`); | |
| 78 | + expect(u.size).toBe(blob.length); | |
| 79 | + expect(sha(objects.get(u.r2_key)!)).toBe(sha(blob)); | |
| 80 | + // Only the unfinished part's bytes were read back between requests, never the blob. | |
| 81 | + expect(reads.every((k) => k === tailKey("up1"))).toBe(true); | |
| 82 | + }); | |
| 83 | +}); |
modifiedcontainers/runner/image/docker+11−0View file
| @@ -164,4 +164,15 @@ | ||
| 164 | 164 | exit "$status" |
| 165 | 165 | fi |
| 166 | 166 | |
| 167 | +# docker push: when it fails, show what the registry relay saw (its error alone doesn't say). | |
| 168 | +if { [ "${1:-}" = push ] || { [ "${1:-}" = image ] && [ "${2:-}" = push ]; }; } && [ -s "$state/relay.host" ]; then | |
| 169 | + "$real" "$@" | |
| 170 | + status=$? | |
| 171 | + if [ "$status" != 0 ]; then | |
| 172 | + echo "The registry relay's last lines:" >&2 | |
| 173 | + tail -n 20 /var/log/diffinery-relay.log >&2 2>/dev/null | |
| 174 | + fi | |
| 175 | + exit "$status" | |
| 176 | +fi | |
| 177 | + | |
| 167 | 178 | exec "$real" "$@" |
modifiedcontainers/runner/src/relay.ts+7−0View file
| @@ -87,6 +87,9 @@ | ||
| 87 | 87 | */ |
| 88 | 88 | async function chunkedUpload(req: http.IncomingMessage, res: http.ServerResponse): Promise<void> { |
| 89 | 89 | const url = new URL(req.url!, "http://relay"); |
| 90 | + const started = Date.now(); | |
| 91 | + let requests = 0; | |
| 92 | + const secs = (since: number) => `${((Date.now() - since) / 1000).toFixed(1)}s`; | |
| 90 | 93 | const headers = forwardHeaders(req); |
| 91 | 94 | delete headers["content-range"]; |
| 92 | 95 | delete headers["content-type"]; |
| @@ -113,7 +116,9 @@ | ||
| 113 | 116 | const flush = async (data: Buffer) => { |
| 114 | 117 | for (let attempt = 1; ; attempt++) { |
| 115 | 118 | try { |
| 119 | + requests++; | |
| 116 | 120 | const r = await send("PATCH", location, { ...headers, "content-type": "application/octet-stream", "content-range": `${offset}-${offset + data.length - 1}` }, data); |
| 121 | + if (r.status !== 202) log(`relay: PATCH ${url.pathname} (${data.length} bytes at ${offset}) answered ${r.status} (attempt ${attempt})`); | |
| 117 | 122 | if (r.status === 202) { |
| 118 | 123 | location = new URL(String(r.headers.location ?? location), "http://relay").pathname; |
| 119 | 124 | offset += data.length; |
| @@ -144,7 +149,9 @@ | ||
| 144 | 149 | } |
| 145 | 150 | const buf = take(held); |
| 146 | 151 | if (req.method === "PUT") { |
| 152 | + const put = Date.now(); | |
| 147 | 153 | const r = await send("PUT", `${location}${url.search}`, { ...headers, ...(buf.length ? { "content-type": "application/octet-stream", "content-range": `${offset}-${offset + buf.length - 1}` } : {}) }, buf); |
| 154 | + log(`relay: upload ${url.pathname}: ${offset + buf.length} bytes in ${requests + 1} requests over ${secs(started)}; the final PUT answered ${r.status} after ${secs(put)}`); | |
| 148 | 155 | return reply(res, r); |
| 149 | 156 | } |
| 150 | 157 | if (buf.length && !(await flush(buf))) return reply(res, last!); |
modifiedcontainers/runner/test/relay.test.ts+5−1View file
| @@ -69,6 +69,7 @@ | ||
| 69 | 69 | let reg: ReturnType<typeof fakeRegistry>; |
| 70 | 70 | let relay: Relay; |
| 71 | 71 | let cert: string; |
| 72 | +let logged: string[]; | |
| 72 | 73 | |
| 73 | 74 | beforeEach(async () => { |
| 74 | 75 | const dir = tmp(); |
| @@ -77,7 +78,8 @@ | ||
| 77 | 78 | reg = fakeRegistry(); |
| 78 | 79 | await new Promise<void>((r) => reg.server.listen(0, "127.0.0.1", () => r())); |
| 79 | 80 | const upstream = `http://127.0.0.1:${(reg.server.address() as net.AddressInfo).port}`; |
| 80 | - relay = await startRelay({ port: 0, host: HOST, upstream, cert, key: readFileSync(path.join(dir, "k.pem"), "utf8"), chunkSize: 1024 * 1024, log: () => {} }); | |
| 81 | + logged = []; | |
| 82 | + relay = await startRelay({ port: 0, host: HOST, upstream, cert, key: readFileSync(path.join(dir, "k.pem"), "utf8"), chunkSize: 1024 * 1024, log: (l) => void logged.push(l) }); | |
| 81 | 83 | }); |
| 82 | 84 | |
| 83 | 85 | afterEach(async () => { |
| @@ -126,6 +128,8 @@ | ||
| 126 | 128 | ["PUT", 502848], |
| 127 | 129 | ]); |
| 128 | 130 | expect(sent[1]!.range).toBe("1048576-2097151"); |
| 131 | + // Its log (shown when a docker push fails) says how long the registry took to answer. | |
| 132 | + expect(logged.at(-1)).toMatch(/^relay: upload \/v2\/o\/img\/blobs\/uploads\/\S+: 2600000 bytes in 3 requests over [\d.]+s; the final PUT answered 201 after [\d.]+s$/); | |
| 129 | 133 | }); |
| 130 | 134 | |
| 131 | 135 | it("answers a streamed PATCH (docker) like the registry would, then completes with an empty PUT", async () => { |
modifiedscripts/render-config.mjs+2−0View file
| @@ -54,6 +54,8 @@ | ||
| 54 | 54 | // A runner image already in the registry (scripts/runner-image.sh): containers use it |
| 55 | 55 | // instead of building the Dockerfile on every deploy. |
| 56 | 56 | if (process.env.RUNNER_IMAGE) { |
| 57 | + // Named as the runner reports it (`runner.image` on a job), for /meta and the Live test. | |
| 58 | + cfg.vars.RUNNER_IMAGE = process.env.RUNNER_IMAGE.split("/").pop(); | |
| 57 | 59 | for (const c of cfg.containers) { |
| 58 | 60 | c.image = process.env.RUNNER_IMAGE; |
| 59 | 61 | delete c.image_build_context; |
modifiedscripts/wait-container-rollout.sh+6−6View file
| @@ -1,9 +1,9 @@ | ||
| 1 | 1 | #!/usr/bin/env bash |
| 2 | -# Waits after `wrangler deploy` until the container rollouts it started reach their last step, | |
| 3 | -# when every new container starts on the new image. `wrangler deploy` returns as soon as a | |
| 4 | -# rollout starts, and a job started before its last step can land on the old image and be | |
| 5 | -# replaced (with its steps) moments later. Running containers keep the old image until they | |
| 6 | -# exit (rollout_active_grace_period), so the rollouts don't need to complete. | |
| 2 | +# Waits after `wrangler deploy` until the container rollouts it started reach their last step. | |
| 3 | +# `wrangler deploy` returns as soon as a rollout starts. Even in the last step, an instance the | |
| 4 | +# rollout hasn't updated yet starts jobs on the old image, so the Live test waits until jobs | |
| 5 | +# report the deployed image. Running containers keep the old image until they exit | |
| 6 | +# (rollout_active_grace_period), so the rollouts don't need to complete. | |
| 7 | 7 | # |
| 8 | 8 | # Run from the repository root with CLOUDFLARE_API_TOKEN, CLOUDFLARE_ACCOUNT_ID and |
| 9 | 9 | # RUNNER_IMAGE (scripts/runner-image.sh) set. Gives up with a warning after |
| @@ -28,7 +28,7 @@ | ||
| 28 | 28 | # Steps started out of all steps, from the steps themselves (progress.current_step is 1-based in practice). |
| 29 | 29 | state=$(jq -r '(.steps // []) as $s | "\(.status), step \([$s[] | select(.status != "pending")] | length) of \($s | length) \($s | last | .status // ""), \(.progress.updated_instances // "?")/\(.progress.total_instances // "?") instances updated"' <<<"$r") |
| 30 | 30 | states+="$name: $state; " |
| 31 | - # Done once the rollout ended, or its last step is under way (new containers get the new image). | |
| 31 | + # Done once the rollout ended, or its last step is under way. | |
| 32 | 32 | jq -e '(.status | IN("completed", "reverted", "replaced")) or ((.steps // []) | length > 0 and (last | .status | IN("progressing", "completed")))' <<<"$r" >/dev/null || pending+="$name " |
| 33 | 33 | done |
| 34 | 34 | [ "$states" = "$last" ] || { echo "${states:-no container rollout for $RUNNER_IMAGE}"; last=$states; } |