From 8ae7dc814d6e93b69883f81b724ecb6b528e2408 Mon Sep 17 00:00:00 2001 From: wan9chi Date: Fri, 6 Nov 2026 19:05:43 +0800 Subject: [PATCH 1/2] feat: keep several entries per key Fetch returns a key's newest live entries instead of one, and a store adds an entry instead of replacing the key's entry. Each namespace keeps `entries_per_key` entries per key (8 by default, at most 16), and publication retires the older ones. Fetch streams the values from R2. Migration 0003 clears the cache tables and keeps namespace policy, and the operator gains `--entries-per-key`. RFC 0003 describes the design. Co-Authored-By: Claude Opus 5.5 --- docs/README.md | 30 ++--- migrations/0003_entries_per_key.sql | 133 +++++++++++++++++++++ rfcs/0003-several-entries-per-key.md | 168 +++++++++++++++++++++++++++ scripts/benchmark.ts | 20 ++-- scripts/e2e/suite.ts | 97 +++++++++------- scripts/operator.ts | 18 ++- src/cbor.ts | 13 ++- src/database.ts | 50 +++++--- src/fetch.ts | 98 +++++++++++++--- src/index.ts | 5 +- test/failures.test.ts | 29 +++-- test/helpers.ts | 2 +- test/operator.test.ts | 15 ++- test/protocol.test.ts | 6 +- test/service.test.ts | 103 +++++++++++++--- 15 files changed, 641 insertions(+), 146 deletions(-) create mode 100644 migrations/0003_entries_per_key.sql create mode 100644 rfcs/0003-several-entries-per-key.md diff --git a/docs/README.md b/docs/README.md index 3e354af..ce21d72 100644 --- a/docs/README.md +++ b/docs/README.md @@ -12,19 +12,19 @@ The package contains no `vp run` client adapter. Cache keys, values, and blobs r The endpoint is `https:///projects/`. Namespaces use 1–63 lowercase letters, digits, or hyphens, starting with a letter or digit. An operator can register up to 100 namespaces per deployment. -| Request | Success | -| ------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------ | -| `POST /fetch`, `application/cbor`, `{ key: bytes, secondary_key: bytes }` | CBOR `{ kind: "exact", value: bytes, blob_id: string \| null }`, or `{ kind: "fallback", key: bytes }` | -| `GET /blob/{blob_id}` | Raw bytes with `application/octet-stream` | -| `POST /store`, `multipart/form-data` | CBOR `{ blob_id: string \| null }` | +| Request | Success | +| ------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------- | +| `POST /fetch`, `application/cbor`, `{ key: bytes, secondary_key: bytes }` | CBOR `{ kind: "exact", entries: [{ value: bytes, blob_id: string \| null }] }`, or `{ kind: "fallback", key: bytes }` | +| `GET /blob/{blob_id}` | Raw bytes with `application/octet-stream` | +| `POST /store`, `multipart/form-data` | CBOR `{ blob_id: string \| null }` | `/store` requires one `metadata` part with content type `application/cbor` and fields `{ key: bytes, secondary_key: bytes, value: bytes }`. An optional `blob` part has content type `application/octet-stream`. Either part order works. An omitted blob returns `null`; an empty blob gets an ID. Duplicate parts, unknown parts, duplicate envelope fields, extra fields, invalid types, and truncated bodies are rejected. Empty and non-UTF-8 keys work. CBOR definite and indefinite maps and byte strings work, including noncanonical lengths. The bounded envelope decoder supports only the protocol's map and string types. It never decodes `value` contents. Chunked strings have a 4,096-chunk bookkeeping limit. -Fetch gives exact matches priority. A fallback follows the latest secondary-key association and returns only the stored key, without reading R2. A store replaces both mappings atomically. Reassigning a secondary key does not delete its former target. Replacing an entry preserves the other associations to its key. +A key keeps several entries ([RFC 0003](../rfcs/0003-several-entries-per-key.md)). A store adds an entry for its key and sets the secondary-key association in one transaction; concurrent stores to a key are all kept. When a key has more than its namespace's `entries_per_key` (8 by default, at most 16), its oldest entries are retired. An exact fetch lists the key's live entries, newest first; the client uses the first one that matches its inputs. Fetch gives exact matches priority. A fallback follows the latest secondary-key association and returns only the stored key, without reading R2. Reassigning a secondary key does not delete its former target. -HTTP status codes follow the [local RFC](../rfcs/0001-remote-cache.md#4-http-api-mapping). A fetch returns `404` when neither key resolves to a live entry. An unavailable blob also returns `404`, including when its R2 object is missing. These responses use plain text. A missing or unreadable value for a live exact match is a storage failure and returns `503`, as specified in [the read design](../rfcs/0001-remote-cache.md#7-fetch-and-download-implementation). +HTTP status codes follow the [local RFC](../rfcs/0001-remote-cache.md#4-http-api-mapping). A fetch returns `404` when neither key resolves to a live entry. An unavailable blob also returns `404`, including when its R2 object is missing. These responses use plain text. A missing or unreadable value for any of a key's live entries is a storage failure and returns `503`, as specified in [the read design](../rfcs/0001-remote-cache.md#7-fetch-and-download-implementation). Errors have `Content-Type: text/plain; charset=utf-8`. Codes are `400` for invalid input, `401` for invalid/missing/expired tokens, `403` for a signed token that fails write policy, `404` for absent data or unavailable namespaces/routes, `413` for size limits, `429` for admission limits, `500` for an incomplete operation, and `503` for unavailable authorization/storage, failed publication guards, quotas, or concurrency admission. `429` and `503` include `Retry-After: 60`. @@ -66,7 +66,7 @@ Use the account's actual Workers subdomain. For a custom domain, set `--origin h Setup discovers resources by name, creates missing resources, applies migrations, resolves public repository and owner IDs through GitHub, records the default branch and exact audience, installs lifecycle/Cron settings, deploys, and prints the endpoint. Repeated setup does not duplicate resources or re-enable a withdrawn namespace. An incompatible origin or repository is rejected before setup changes existing policies, bucket settings, or saved configuration. Configuration is written to the ignored `wrangler.operator.json`; keep a secure backup of this non-secret file. -Start with Workers Free. The operator defaults to the `free` resource profile: an 8 GB budget, 20,000 entries, 20,000 associations, and 16 generations per Cron run. A paid subscription is optional. Setup does not purchase or change a Workers subscription, and R2 must be enabled separately. +Start with Workers Free. The operator defaults to the `free` resource profile: an 8 GB budget, 20,000 entry keys, 20,000 associations, and 16 generations per Cron run. A paid subscription is optional. Setup does not purchase or change a Workers subscription, and R2 must be enabled separately. Check deployed CPU and usage for your workload. Workers Free currently allows [10 ms of CPU per invocation](https://developers.cloudflare.com/workers/platform/limits/); maximum payloads are not yet validated against that allowance. Local CPU measurements are diagnostic, not Cloudflare billing measurements. Lower payload limits or select Workers Paid if your workload exceeds Free limits. @@ -82,7 +82,7 @@ Set GitHub Actions `id-token: write` only on the trusted publishing job. The tok `pull_request` tokens are denied by default. As an interim measure, an operator can opt a namespace in with `pnpm operator policy --namespace docs --pull-requests on`, or pass `--pull-requests on` to the setup that creates the namespace. `--pull-requests off` withdraws the opt-in. An opted-in namespace accepts a `pull_request` token that passes the same audience, repository/owner ID, and visibility checks as a push and also has `ref` `refs/pull//merge`, `ref_type` `branch`, and a `base_ref` equal to the registered default branch. PRs that target another branch, `pull_request_target`, `workflow_run`, tags, and workflows in fork repositories remain denied. -The opt-in carries a risk: PR entries share the namespace with default-branch entries. Code from any same-repository PR, including Renovate dependency branches, runs before review and can publish an entry that a later default-branch build restores. That is why it is off by default. The long-term fix is to [scope pull request uploads to the pull request](https://github.com/voidzero-dev/vite-plus-remote-cache-cloudflare/issues/16). +The opt-in carries a risk: PR entries share the namespace with default-branch entries. Code from any same-repository PR, including Renovate dependency branches, runs before review and can publish an entry that a later default-branch build restores. That is why it is off by default. Because a key keeps several entries, PR entries also take slots from the default branch's entries. The long-term fix is to [scope pull request uploads to the pull request](https://github.com/voidzero-dev/vite-plus-remote-cache-cloudflare/issues/16). Fork PR workflows run in the base repository, so their tokens carry its repository and owner IDs; the server cannot tell them from same-repository PRs. The policy depends on GitHub not granting `id-token: write` to fork PR workflows. GitHub grants write permissions to them only when the repository enables **Send write tokens to workflows from pull requests**, a setting [GitHub offers for private repositories only](https://docs.github.com/en/repositories/managing-your-repositorys-settings-and-features/enabling-features-for-your-repository/managing-github-actions-settings-for-a-repository#enabling-workflows-for-forks-of-private-repositories). GitHub treats Dependabot PRs like fork PRs. @@ -102,7 +102,8 @@ Fork PR workflows run in the base repository, so their tokens carry its reposito | Store request deadline | 2 minutes | | Metadata / blob lookup deadline | 15 seconds | | Upload lease | 15 minutes | -| Replacement/late-upload grace | 10 minutes | +| Entries per key | 8, at most 16 | +| Retired-entry/late-upload grace | 10 minutes | | Concurrent buffered stores/reads per isolate | 2 / 4 | The `LIMITS` Wrangler variable is a JSON object with optional keys `key`, `value`, `metadata`, `fetch`, `blob`, `store`, `headers`, and `deadlineMs`. It can lower the tested ceilings. Envelope limits must accommodate their field limits and framing. Raising ceilings requires code changes and new memory, CPU, and account-limit measurements. An HTTP request without `Content-Length` reserves the full store limit. Actual bytes replace that reservation only at publication; pending, retired, and deleting generations remain charged. @@ -124,13 +125,14 @@ pnpm operator policy --namespace docs --pull-requests on pnpm operator policy --namespace docs --enabled off pnpm operator policy --namespace docs --enabled on --writes on pnpm operator policy --namespace docs --retention-days 30 --byte-limit 30000000000 +pnpm operator policy --namespace docs --entries-per-key 16 pnpm operator deployment --byte-limit 30000000000 pnpm operator deployment --writes off pnpm operator deployment --enabled off pnpm operator upgrade ``` -`bind` refreshes repository display data, owner ID, and default branch for an existing repository, or registers a new namespace. It never silently moves existing public data to a different repository. Per-scope and deployment byte/entry/association limits all apply; use `--entry-limit` and `--association-limit` to change them explicitly. Lowering a limit below current use blocks further publication until cleanup or an operator change restores capacity. +`bind` refreshes repository display data, owner ID, and default branch for an existing repository, or registers a new namespace. It never silently moves existing public data to a different repository. Per-scope and deployment byte/entry/association limits all apply; use `--entry-limit` and `--association-limit` to change them explicitly. The entry limit counts keys, not each key's entries. `--entries-per-key` sets how many entries a key keeps; a fetch reads up to that many values. Lowering a limit below current use blocks further publication until cleanup or an operator change restores capacity. `status` reports policy, charged bytes, entry/association counts, generation states, cleanup eligibility, and actual D1 storage. It warns at 400 MB. Set provider alerts at 80% of request, CPU, R2 operation/storage, and D1 read/write/storage allowances. Monitor cleanup delay and pending bytes, and pause writes before a backlog reaches the storage ceiling. Public requests and other services in the same account can exhaust allowances even when this cache's byte budget is respected. @@ -138,9 +140,9 @@ Sampled structured logs include request ID, namespace, operation, exact/fallback ### Cleanup and withdrawal -Every five minutes Cron claims a bounded indexed set of eligible generations, aborts known multipart uploads, deletes immutable object names, then releases charges. Failed deletion is retried after ten minutes. Conditional claims and generation-specific foreign keys preserve concurrent replacement entries. An indexed cursor scans and deletes orphan associations in bounded batches. +Every five minutes Cron claims a bounded indexed set of eligible generations, aborts known multipart uploads, deletes immutable object names, then releases charges. Failed deletion is retried after ten minutes. Conditional claims and generation-specific foreign keys preserve a key's newer entries. An indexed cursor scans and deletes orphan associations in bounded batches. -Current entries expire seven days after commit by default. Replaced blob IDs retain their original bytes for ten minutes. Abandoned uploads get a late-operation grace period after their lease ends. R2 lifecycle rules abort unfinished multipart uploads after one day and expire objects after retention plus two days (9/32 days for 7/30-day retention). A retention high-water mark prevents later policy reductions from deleting older generations early. Lifecycle covers the crash window between multipart creation and recording its ID; it supplements D1 accounting. +Each entry expires seven days after its store by default. A retired entry's blob ID keeps its bytes for ten minutes. Abandoned uploads get a late-operation grace period after their lease ends. R2 lifecycle rules abort unfinished multipart uploads after one day and expire objects after retention plus two days (9/32 days for 7/30-day retention). A retention high-water mark prevents later policy reductions from deleting older generations early. Lifecycle covers the crash window between multipart creation and recording its ID; it supplements D1 accounting. To withdraw and delete one namespace: @@ -160,7 +162,7 @@ Public data can include logs, source maps, and input metadata. Only publish resu ### Upgrade and recovery -Keep the lockfile and compatibility date pinned. `upgrade` applies additive migrations before deploying the Worker. Back up `wrangler.operator.json` and the `scopes`/`deployment` policy rows separately from disposable cache data. `status` exports readable policy data. Before a migration that changes the schema contract, retain a compatible Worker bundle; do not roll back to code that predates a required schema change. +Keep the lockfile and compatibility date pinned. `upgrade` applies migrations before deploying the Worker. Migration `0003_entries_per_key` clears cache data and keeps policy rows; the bucket's lifecycle rules delete the old objects. Back up `wrangler.operator.json` and the `scopes`/`deployment` policy rows separately from disposable cache data. `status` exports readable policy data. Before a migration that changes the schema contract, retain a compatible Worker bundle; do not roll back to code that predates a required schema change. A D1 restore cannot restore deleted R2 objects. Missing live values return `503`; missing blobs return `404`. After partial recovery, create a new namespace and retire the old one, or reconcile the generation records and objects before enabling it. Do not restore stale policy over an intentional withdrawal. diff --git a/migrations/0003_entries_per_key.sql b/migrations/0003_entries_per_key.sql new file mode 100644 index 0000000..db1611d --- /dev/null +++ b/migrations/0003_entries_per_key.sql @@ -0,0 +1,133 @@ +-- Keep several entries per key (RFC 0003). Cache data is disposable, so the cache tables are +-- recreated empty. Namespace registrations, policy and limits stay. The R2 bucket's lifecycle rules +-- delete the old objects; empty the bucket to get their bytes back sooner. +DROP TABLE entries; +DROP TABLE associations; +DROP TABLE generations; +UPDATE maintenance SET scope_id = '', secondary_key = X'' WHERE id = 1; +UPDATE scopes SET charged_bytes = 0, entry_count = 0, association_count = 0; +UPDATE deployment SET charged_bytes = 0, entry_count = 0, association_count = 0; +-- How many of a key's newest entries a namespace keeps. +ALTER TABLE scopes ADD COLUMN entries_per_key INTEGER NOT NULL DEFAULT 8 CHECK (entries_per_key BETWEEN 1 AND 16); + +CREATE TABLE generations ( + generation_id TEXT PRIMARY KEY, + scope_id TEXT NOT NULL REFERENCES scopes(scope_id), + state TEXT NOT NULL CHECK (state IN ('uploading', 'ready', 'retired', 'deleting')), + policy_version INTEGER NOT NULL, + token_exp INTEGER NOT NULL, + lease_until INTEGER NOT NULL, + gc_after INTEGER NOT NULL, + gc_claim TEXT, + key BLOB, + secondary_key BLOB, + value_object TEXT NOT NULL UNIQUE, + blob_object TEXT NOT NULL UNIQUE, + blob_id TEXT, + upload_id TEXT, + value_size INTEGER NOT NULL DEFAULT 0 CHECK (value_size >= 0), + blob_size INTEGER NOT NULL DEFAULT 0 CHECK (blob_size >= 0), + charged_bytes INTEGER NOT NULL CHECK (charged_bytes >= 0), + expires_at INTEGER, + retired_at INTEGER, + UNIQUE (scope_id, generation_id) +); +CREATE UNIQUE INDEX generations_blob ON generations(scope_id, blob_id) WHERE blob_id IS NOT NULL; +CREATE INDEX generations_gc ON generations(gc_after); +-- One row per live entry. A key can have up to `entries_per_key` of them. `seq` numbers a key's +-- publications in order, so the newest entries sort first even within one second. +CREATE TABLE entries ( + scope_id TEXT NOT NULL REFERENCES scopes(scope_id), + key BLOB NOT NULL CHECK (typeof(key) = 'blob'), + generation_id TEXT NOT NULL, + seq INTEGER NOT NULL, + PRIMARY KEY (scope_id, key, generation_id), + FOREIGN KEY (scope_id, generation_id) REFERENCES generations(scope_id, generation_id) ON DELETE CASCADE +) WITHOUT ROWID; +CREATE INDEX entries_generation ON entries(generation_id); +CREATE UNIQUE INDEX entries_seq ON entries(scope_id, key, seq); +CREATE TABLE associations ( + scope_id TEXT NOT NULL REFERENCES scopes(scope_id), + secondary_key BLOB NOT NULL CHECK (typeof(secondary_key) = 'blob'), + target_key BLOB NOT NULL CHECK (typeof(target_key) = 'blob'), + PRIMARY KEY (scope_id, secondary_key) +) WITHOUT ROWID; +CREATE INDEX associations_target ON associations(scope_id, target_key); + +-- Parenthesize CASE expressions so the D1 REST parser preserves each trigger body. +CREATE TRIGGER reserve_capacity BEFORE INSERT ON generations BEGIN + SELECT (CASE WHEN NOT EXISTS ( + SELECT 1 FROM scopes s, deployment d WHERE s.scope_id = NEW.scope_id + AND s.enabled = 1 AND s.writes_enabled = 1 AND d.enabled = 1 AND d.writes_enabled = 1 + AND s.policy_version = NEW.policy_version AND NEW.token_exp > unixepoch() + AND s.charged_bytes + NEW.charged_bytes <= s.byte_limit + AND d.charged_bytes + NEW.charged_bytes <= d.byte_limit + ) THEN RAISE(ABORT, 'cache_admission_denied') END); +END; +CREATE TRIGGER charge_generation AFTER INSERT ON generations BEGIN + UPDATE scopes SET charged_bytes = charged_bytes + NEW.charged_bytes WHERE scope_id = NEW.scope_id; + UPDATE deployment SET charged_bytes = charged_bytes + NEW.charged_bytes; +END; +CREATE TRIGGER adjust_generation AFTER UPDATE OF charged_bytes ON generations BEGIN + UPDATE scopes SET charged_bytes = charged_bytes + NEW.charged_bytes - OLD.charged_bytes WHERE scope_id = NEW.scope_id; + UPDATE deployment SET charged_bytes = charged_bytes + NEW.charged_bytes - OLD.charged_bytes; +END; +CREATE TRIGGER release_generation AFTER DELETE ON generations BEGIN + UPDATE scopes SET charged_bytes = charged_bytes - OLD.charged_bytes WHERE scope_id = OLD.scope_id; + UPDATE deployment SET charged_bytes = charged_bytes - OLD.charged_bytes; +END; + +-- `entry_limit` counts keys: a key's first entry counts it, and deleting its last entry uncounts it. +CREATE TRIGGER count_entry BEFORE INSERT ON entries +WHEN NOT EXISTS (SELECT 1 FROM entries WHERE scope_id = NEW.scope_id AND key = NEW.key) +BEGIN + SELECT (CASE WHEN EXISTS ( + SELECT 1 FROM scopes s, deployment d WHERE s.scope_id = NEW.scope_id + AND (s.entry_count >= s.entry_limit OR d.entry_count >= d.entry_limit) + ) THEN RAISE(ABORT, 'cache_entry_limit') END); + UPDATE scopes SET entry_count = entry_count + 1 WHERE scope_id = NEW.scope_id; + UPDATE deployment SET entry_count = entry_count + 1; +END; +CREATE TRIGGER uncount_entry AFTER DELETE ON entries +WHEN NOT EXISTS (SELECT 1 FROM entries WHERE scope_id = OLD.scope_id AND key = OLD.key) +BEGIN + UPDATE scopes SET entry_count = entry_count - 1 WHERE scope_id = OLD.scope_id; + UPDATE deployment SET entry_count = entry_count - 1; +END; +CREATE TRIGGER count_association BEFORE INSERT ON associations +WHEN NOT EXISTS (SELECT 1 FROM associations WHERE scope_id = NEW.scope_id AND secondary_key = NEW.secondary_key) +BEGIN + SELECT (CASE WHEN EXISTS ( + SELECT 1 FROM scopes s, deployment d WHERE s.scope_id = NEW.scope_id + AND (s.association_count >= s.association_limit OR d.association_count >= d.association_limit) + ) THEN RAISE(ABORT, 'cache_association_limit') END); + UPDATE scopes SET association_count = association_count + 1 WHERE scope_id = NEW.scope_id; + UPDATE deployment SET association_count = association_count + 1; +END; +CREATE TRIGGER uncount_association AFTER DELETE ON associations BEGIN + UPDATE scopes SET association_count = association_count - 1 WHERE scope_id = OLD.scope_id; + UPDATE deployment SET association_count = association_count - 1; +END; + +-- All publication mutations inherit the guarded state change's transaction. +-- A zero-row guard fires no trigger. A quota failure rolls every mutation back. +-- Publication adds an entry, then retires the key's entries past its newest `entries_per_key`. +CREATE TRIGGER publish_generation AFTER UPDATE OF state ON generations +WHEN OLD.state = 'uploading' AND NEW.state = 'ready' +BEGIN + INSERT INTO entries (scope_id, key, generation_id, seq) + VALUES (NEW.scope_id, NEW.key, NEW.generation_id, + (SELECT coalesce(max(seq), 0) + 1 FROM entries WHERE scope_id = NEW.scope_id AND key = NEW.key)); + UPDATE generations SET state = 'retired', retired_at = unixepoch(), + gc_after = (CASE WHEN expires_at > unixepoch() THEN unixepoch() + 600 ELSE unixepoch() END) + WHERE state = 'ready' AND generation_id IN ( + SELECT generation_id FROM entries WHERE scope_id = NEW.scope_id AND key = NEW.key + ORDER BY seq DESC + LIMIT -1 OFFSET (SELECT entries_per_key FROM scopes WHERE scope_id = NEW.scope_id)); + DELETE FROM entries WHERE scope_id = NEW.scope_id AND key = NEW.key AND generation_id IN ( + SELECT generation_id FROM entries WHERE scope_id = NEW.scope_id AND key = NEW.key + ORDER BY seq DESC + LIMIT -1 OFFSET (SELECT entries_per_key FROM scopes WHERE scope_id = NEW.scope_id)); + INSERT INTO associations (scope_id, secondary_key, target_key) VALUES (NEW.scope_id, NEW.secondary_key, NEW.key) + ON CONFLICT (scope_id, secondary_key) DO UPDATE SET target_key = excluded.target_key; +END; diff --git a/rfcs/0003-several-entries-per-key.md b/rfcs/0003-several-entries-per-key.md new file mode 100644 index 0000000..663694a --- /dev/null +++ b/rfcs/0003-several-entries-per-key.md @@ -0,0 +1,168 @@ +# RFC: Several entries per key + +Status: Draft design. + +Updated: 2026-11-06. Changes the fetch and store behavior of [RFC 0001](0001-remote-cache.md) and nothing else. Writes keep RFC 0001's policy: protected pushes to the registered branch, and pull requests only where a namespace opts in. Scoped selectors, with per-PR scopes, remain a possible later design. Remote caching hasn't shipped, so this RFC keeps no compatibility with RFC 0001 clients or data. + +## 1. Motivation + +A namespace keeps one entry per key, and a store with the same key replaces it. A run hits only if its state is the last one stored. + +**Ordinary pull request runs don't suffer from this.** They build the PR merged into the current default branch. A task the PR doesn't touch has the same state as the latest default-branch run. A task the PR does touch would miss anyway. + +**Runs built on an older default-branch commit do suffer:** + +- PR runs without a merged result +- stacked PRs +- PRs into other branches +- developer machines that haven't pulled + +These runs miss whenever a task's inputs changed since that commit, even though the namespace stored that state earlier. + +**Keeping the last N entries per key lets these runs hit** without any PR uploads. + +## 2. Changes + +| Area | RFC 0001 | This RFC | +| ---------------------------- | -------------------------------------- | ----------------------------------------------------------- | +| Fetch | the exact entry, or the associated key | the key's live entries, newest first, or the associated key | +| Store | replaces the key's entry | adds an entry; the key keeps its newest N | +| N | | `entries_per_key`, per namespace, default 8, at most 16 | +| Concurrent stores to one key | the last one wins | all are kept, within N | +| Writes | protected pushes; PR opt-in | unchanged | +| Data | | cleared; no data migration | + +## 3. HTTP API + +### Fetch + +The request is unchanged: + +```text +POST {endpoint}/fetch +Content-Type: application/cbor + +{ key: bytes, secondary_key: bytes } +``` + +The response is one of: + +```text +{ kind: "exact", entries: [{ value: bytes, blob_id: string | null }] } +{ kind: "fallback", key: bytes } +``` + +- **`entries`** holds the key's live entries, newest first. There's at least one and at most `entries_per_key`. +- **The fallback and `404`** are unchanged. When the key has no live entry, the Worker resolves `secondary_key` to a stored key and returns only that key. If neither resolves, it returns `404`. + +### Store + +The request and response are unchanged. The behavior changes: + +- **A store adds an entry for `key`.** It doesn't replace the key's other entries. +- **If the key then has more than `entries_per_key` live entries,** the oldest are retired in the same transaction. Their blobs stay downloadable for the existing ten-minute grace period. +- **The association** `secondary_key → key` is set as before. +- **Each entry expires** the namespace's retention after its own store, as each generation does today. + +## 4. Storage + +The tables stay. These change: + +- **`scopes`** gains `entries_per_key INTEGER NOT NULL DEFAULT 8 CHECK (entries_per_key BETWEEN 1 AND 16)`. +- **`entries`** gets one row per live entry instead of one per key: + + ```sql + CREATE TABLE entries ( + scope_id TEXT NOT NULL REFERENCES scopes(scope_id), + key BLOB NOT NULL CHECK (typeof(key) = 'blob'), + generation_id TEXT NOT NULL, + seq INTEGER NOT NULL, + PRIMARY KEY (scope_id, key, generation_id), + FOREIGN KEY (scope_id, generation_id) REFERENCES generations(scope_id, generation_id) ON DELETE CASCADE + ) WITHOUT ROWID; + CREATE UNIQUE INDEX entries_seq ON entries(scope_id, key, seq); + ``` + + `seq` numbers a key's publications in order. A timestamp would tie when two stores publish in the same second. + +- **The publication trigger** adds the entry, then retires the entries past the newest N: + + ```sql + CREATE TRIGGER publish_generation AFTER UPDATE OF state ON generations + WHEN OLD.state = 'uploading' AND NEW.state = 'ready' + BEGIN + INSERT INTO entries (scope_id, key, generation_id, seq) + VALUES (NEW.scope_id, NEW.key, NEW.generation_id, + (SELECT coalesce(max(seq), 0) + 1 FROM entries WHERE scope_id = NEW.scope_id AND key = NEW.key)); + UPDATE generations SET state = 'retired', retired_at = unixepoch(), + gc_after = (CASE WHEN expires_at > unixepoch() THEN unixepoch() + 600 ELSE unixepoch() END) + WHERE state = 'ready' AND generation_id IN ( + SELECT generation_id FROM entries WHERE scope_id = NEW.scope_id AND key = NEW.key + ORDER BY seq DESC + LIMIT -1 OFFSET (SELECT entries_per_key FROM scopes WHERE scope_id = NEW.scope_id)); + DELETE FROM entries WHERE scope_id = NEW.scope_id AND key = NEW.key AND generation_id IN ( + SELECT generation_id FROM entries WHERE scope_id = NEW.scope_id AND key = NEW.key + ORDER BY seq DESC + LIMIT -1 OFFSET (SELECT entries_per_key FROM scopes WHERE scope_id = NEW.scope_id)); + INSERT INTO associations (scope_id, secondary_key, target_key) VALUES (NEW.scope_id, NEW.secondary_key, NEW.key) + ON CONFLICT (scope_id, secondary_key) DO UPDATE SET target_key = excluded.target_key; + END; + ``` + +- **`entry_limit` still counts keys.** `count_entry` fires only for a key's first row, as today. `uncount_entry` fires only when a key's last row is deleted (`WHEN NOT EXISTS` a remaining row for the key). Bytes stay bounded by the byte budget. +- **Fetch** reads the newest `entries_per_key` live entries for the key, joined with their generations. It reads the association only when there are none. +- **Cleanup** is unchanged. Deleting an expired generation deletes its entry row through the foreign key. + +## 5. Costs and limits + +**Fetch.** A fetch makes up to N R2 reads, at most six at a time, and returns up to N values. The value limit stays 4 MiB, so a response can reach N × 4 MiB; the Worker streams it. + +**Values must be compressed.** `vp run` should compress values with zstd before storing them. In a replay of five public repositories' CI, values as the client sends them today measured: + +| Values | Median | p90 | Max | +| ------------ | ------ | ------- | ------- | +| Uncompressed | 111 KB | 1.66 MB | 3.10 MB | +| Compressed | 9.1 KB | 123 KB | 191 KB | + +With eight entries, a fetch carries eight times that. Large repositories also exceed the 4 MiB limit with uncompressed values. + +**Storage.** A key holds up to N generations, so it holds N blobs where it held one. The byte budget, rate limits and retention are unchanged. + +**Store.** Publication writes up to N more rows, for retiring and deleting old entries. + +**Choosing N.** A run hits if its state is among the last N stored for the key: + +- A task whose inputs change on every merge reaches back N merges. +- A task that rarely changes reaches back N input changes, which is usually days or weeks. + +Most of the gain over one entry comes by 8–16. The default is 8, and operators can raise it to 16. + +## 6. Client changes in `vp run` + +- **Fetch.** Decode `entries` and validate them newest first. Restore the first match's blob. With no match, explain the miss against the newest entry. +- **Store.** Compress the value with zstd before upload, and decompress it after fetch. The request is otherwise unchanged. +- **Tests.** The test server in `vt_remote_cache` and the e2e `remote_cache_backend` fixture return entry lists. + +## 7. Rollout + +- **Migration `0003_entries_per_key` drops and recreates** `generations`, `entries` and `associations` and their triggers. It adds `entries_per_key` and resets the byte and entry counters. Namespace registrations and policy stay. + - The R2 bucket's lifecycle rules delete the old objects. Operators who want the bytes back sooner can empty the bucket. +- **The operator script** gains `--entries-per-key`. +- **The Worker and `vp run` ship together.** Older clients fail to decode the new fetch response; they treat that as a failed read and run the task. + +## 8. Alternatives considered + +- **Scoped selectors:** + - one compressed list of runs per key, sorted by path + - a version check on store + - PR scopes + + They reach further for fewer bytes and add hits on a PR's later runs, but need much more server work. They can still come later. + +- **Renewing a key's older entries on each store.** It keeps old states past their own retention. The measured gain at N ≤ 16 was small, and it costs a write per kept entry. +- **One entry per key with a version check that keeps the newer run.** Out-of-order stores turned out to cost almost nothing, so this wouldn't help. + +## 9. Open questions + +- **Whether the default N should be 8 or 16.** +- **The PR upload opt-in.** With entries added rather than replaced, PR stores would also push the default branch's entries out of a key's N slots. Should it be removed with this change? diff --git a/scripts/benchmark.ts b/scripts/benchmark.ts index 6f0f446..36a4a26 100644 --- a/scripts/benchmark.ts +++ b/scripts/benchmark.ts @@ -107,18 +107,14 @@ try { // separately; it must not be interpreted as Cloudflare's billed Worker CPU. const value = new Uint8Array(valueSize); const envelope = encodeEnvelope({ key: bytes('codec'), secondary_key: bytes('codec'), value }); - for (const operation of ['decode', 'encode']) { - const start = process.cpuUsage(); - for (let i = 0; i < 100; i++) { - if (operation === 'decode') decodeEnvelope(envelope, true, defaults); - else encodeEnvelope({ kind: 'exact', value, blob_id: null }); - } - const cpu = process.cpuUsage(start); - measurements.push({ - name: `cbor-${operation}-${valueSize}`, - node_cpu_ms_per_operation: (cpu.user + cpu.system) / 100000, - }); - } + // Fetch streams values from R2 without encoding them, so only store decoding is measured. + const start = process.cpuUsage(); + for (let i = 0; i < 100; i++) decodeEnvelope(envelope, true, defaults); + const cpu = process.cpuUsage(start); + measurements.push({ + name: `cbor-decode-${valueSize}`, + node_cpu_ms_per_operation: (cpu.user + cpu.system) / 100000, + }); await measure(`fetch-value-${valueSize}`, async () => { assert.equal( (await h.store(bytes('fetch'), bytes('fetch'), value, undefined, { token })).status, diff --git a/scripts/e2e/suite.ts b/scripts/e2e/suite.ts index 974edc3..d8eb09b 100644 --- a/scripts/e2e/suite.ts +++ b/scripts/e2e/suite.ts @@ -35,6 +35,15 @@ export interface Options { function digest(value: Uint8Array): string { return createHash('sha256').update(value).digest('hex'); } +// An exact fetch lists a key's entries, newest first. +function entries(result: Record): Record[] { + assert.equal(result['kind'], 'exact'); + assert.ok(Array.isArray(result['entries']) && result['entries'].length > 0); + return result['entries'] as Record[]; +} +function newest(result: Record): Record { + return entries(result)[0]!; +} function blobId(result: Record): string { assert.equal(typeof result['blob_id'], 'string'); return result['blob_id'] as string; @@ -182,7 +191,7 @@ export async function runSuite(options: Options): Promise { response.status === 200 && response.headers.get('X-Remote-Cache-Deployment') === deployment ) { - assert.deepEqual((await envelope(response)).value, seeded.value); + assert.deepEqual(newest(await envelope(response))['value'], seeded.value); return seeded; } await response.body?.cancel(); @@ -199,10 +208,9 @@ export async function runSuite(options: Options): Promise { 'manual', ), ); - assert.equal(manual['kind'], 'exact'); - assert.deepEqual(manual['value'], bytes(`deployment:${deployment}`)); + assert.deepEqual(newest(manual)['value'], bytes(`deployment:${deployment}`)); assert.deepEqual( - await body(await call('manual', `blob/${blobId(manual)}`, 200)), + await body(await call('manual', `blob/${blobId(newest(manual))}`, 200)), bytes('Public remote cache verification blob\n'), ); }); @@ -211,7 +219,7 @@ export async function runSuite(options: Options): Promise { // this live fixture, rather than assuming global totals cannot decrease. const accounting = `SELECT charged_bytes - (SELECT coalesce(sum(charged_bytes), 0) FROM generations) AS bytes_delta, - entry_count - (SELECT count(*) FROM entries) AS entries_delta, + entry_count - (SELECT count(*) FROM (SELECT DISTINCT scope_id, key FROM entries)) AS entries_delta, association_count - (SELECT count(*) FROM associations) AS associations_delta FROM deployment`; const before = await admin.sql( 'SELECT charged_bytes, expires_at, state FROM generations WHERE generation_id = ?', @@ -222,8 +230,7 @@ export async function runSuite(options: Options): Promise { ]); assert.deepEqual(await envelope(await lookup(fixture.key, fixture.secondary)), { kind: 'exact', - value: fixture.value, - blob_id: fixture.blobId, + entries: [{ value: fixture.value, blob_id: fixture.blobId }], }); assert.deepEqual(await envelope(await lookup(bytes('missing'), fixture.secondary)), { kind: 'fallback', @@ -291,7 +298,7 @@ export async function runSuite(options: Options): Promise { await admin.sql("UPDATE scopes SET enabled = 1 WHERE scope_id = 'e2e'"); } assert.deepEqual( - (await envelope(await lookup(fixture.key, fixture.secondary))).value, + newest(await envelope(await lookup(fixture.key, fixture.secondary)))['value'], fixture.value, ); }); @@ -325,52 +332,60 @@ export async function runSuite(options: Options): Promise { ); assert.equal(typeof empty.blob_id, 'string'); assert.equal((await body(await call('e2e', `blob/${blobId(empty)}`, 200))).length, 0); - assert.deepEqual((await envelope(await lookup(key, secondary))).value, value); - }); - await check('secondary reassignment and replacement keep both mappings coherent', async () => { - const a = bytes(`${prefix}-A`), - b = bytes(`${prefix}-B`), - s = bytes(`${prefix}-S`), - t = bytes(`${prefix}-T`); - const old = await envelope(await store(a, s, bytes('old'), bytes('old-blob'))); - await body(await store(b, s, bytes('B'))); - assert.deepEqual((await envelope(await lookup(a, s))).value, bytes('old')); - assert.deepEqual(await envelope(await lookup(bytes('missing'), s)), { - kind: 'fallback', - key: b, - }); - await body(await store(a, t, bytes('replacement'))); - assert.deepEqual(await envelope(await lookup(bytes('missing'), t)), { - kind: 'fallback', - key: a, - }); - assert.deepEqual((await envelope(await lookup(a, t))).value, bytes('replacement')); - assert.deepEqual( - await body(await call('e2e', `blob/${blobId(old)}`, 200)), - bytes('old-blob'), - ); + assert.deepEqual(newest(await envelope(await lookup(key, secondary)))['value'], value); }); + await check( + 'secondary reassignment and added entries keep both mappings coherent', + async () => { + const a = bytes(`${prefix}-A`), + b = bytes(`${prefix}-B`), + s = bytes(`${prefix}-S`), + t = bytes(`${prefix}-T`); + const old = await envelope(await store(a, s, bytes('old'), bytes('old-blob'))); + await body(await store(b, s, bytes('B'))); + assert.deepEqual(newest(await envelope(await lookup(a, s)))['value'], bytes('old')); + assert.deepEqual(await envelope(await lookup(bytes('missing'), s)), { + kind: 'fallback', + key: b, + }); + await body(await store(a, t, bytes('added'))); + assert.deepEqual(await envelope(await lookup(bytes('missing'), t)), { + kind: 'fallback', + key: a, + }); + assert.deepEqual( + entries(await envelope(await lookup(a, t))).map((entry) => entry['value']), + [bytes('added'), bytes('old')], + ); + assert.deepEqual( + await body(await call('e2e', `blob/${blobId(old)}`, 200)), + bytes('old-blob'), + ); + }, + ); await check('R2 multipart upload and download preserve bytes', async () => { const value = new Uint8Array(250000).fill(149), blob = new Uint8Array(5 * MiB + 17).fill(61); const key = bytes(`${prefix}-large`); const stored = await envelope(await store(key, key, value, blob, { blobFirst: true })); - assert.deepEqual((await envelope(await lookup(key, key))).value, value); + assert.deepEqual(newest(await envelope(await lookup(key, key)))['value'], value); assert.equal( digest(await body(await call('e2e', `blob/${blobId(stored)}`, 200))), digest(blob), ); }); - await check('concurrent HTTP stores select one complete generation', async () => { + await check('concurrent HTTP stores keep both complete generations', async () => { const key = bytes(`${prefix}-concurrent`); await Promise.all( [0, 1].map(async (i) => body(await store(key, key, bytes(String(i)), bytes(String(i))))), ); - const selected = await envelope(await lookup(key, key)); - assert.deepEqual( - await body(await call('e2e', `blob/${blobId(selected)}`, 200)), - selected.value, - ); + const kept = entries(await envelope(await lookup(key, key))); + assert.equal(kept.length, 2); + for (const entry of kept) + assert.deepEqual( + await body(await call('e2e', `blob/${blobId(entry)}`, 200)), + entry['value'], + ); }); await check('malformed uploads and exhausted quotas preserve published data', async () => { const token = await storeToken(); @@ -401,7 +416,7 @@ export async function runSuite(options: Options): Promise { await admin.sql("UPDATE scopes SET byte_limit = ? WHERE scope_id = 'e2e'", [Number(limit)]); } assert.deepEqual( - (await envelope(await lookup(fixture.key, fixture.secondary))).value, + newest(await envelope(await lookup(fixture.key, fixture.secondary)))['value'], fixture.value, ); }); @@ -416,7 +431,7 @@ export async function runSuite(options: Options): Promise { await store(key, key, value, blob, { blobFirst: i === 0 }), ); assert.equal( - digest((await envelope(await lookup(key, key))).value as Uint8Array), + digest(newest(await envelope(await lookup(key, key)))['value'] as Uint8Array), digest(value), ); assert.equal( diff --git a/scripts/operator.ts b/scripts/operator.ts index 01473ca..ecbc327 100644 --- a/scripts/operator.ts +++ b/scripts/operator.ts @@ -79,6 +79,13 @@ function positive(value: string | undefined, fallback: number): number { function optionalPositive(value: string | undefined): number | null { return value ? positive(value, 1) : null; } +// How many of a key's newest entries a namespace keeps. Each fetch reads up to this many values. +function entriesPerKey(value: string | undefined): number | null { + if (value === undefined) return null; + const number = positive(value, 1); + if (number > 16) throw new Error('A namespace keeps at most 16 entries per key'); + return number; +} function toggle(value: string | undefined): number | null { if (value === undefined) return null; if (value !== 'on' && value !== 'off') throw new Error('Use on or off'); @@ -226,6 +233,7 @@ export async function runOperator( 'byte-limit': { type: 'string' }, 'entry-limit': { type: 'string' }, 'association-limit': { type: 'string' }, + 'entries-per-key': { type: 'string' }, enabled: { type: 'string' }, writes: { type: 'string' }, 'pull-requests': { type: 'string' }, @@ -256,6 +264,7 @@ export async function runOperator( } if (values['retention-days'] && positive(values['retention-days'], 7) > 365) throw new Error('Retention cannot exceed 365 days'); + entriesPerKey(values['entries-per-key']); toggle(values.enabled); toggle(values.writes); toggle(values['pull-requests']); @@ -330,8 +339,8 @@ export async function runOperator( await query( io, config, - `INSERT INTO scopes (scope_id, endpoint, repository, repository_id, repository_owner_id, branch, pull_requests, retention_seconds, byte_limit, entry_limit, association_limit) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(scope_id) DO NOTHING`, + `INSERT INTO scopes (scope_id, endpoint, repository, repository_id, repository_owner_id, branch, pull_requests, retention_seconds, byte_limit, entry_limit, association_limit, entries_per_key) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(scope_id) DO NOTHING`, [ namespace, `${origin.origin}/projects/${namespace}`, @@ -344,6 +353,7 @@ export async function runOperator( positive(values['byte-limit'], 8000000000), positive(values['entry-limit'], 20000), positive(values['association-limit'], 20000), + entriesPerKey(values['entries-per-key']) ?? 8, ], ); await query( @@ -520,7 +530,8 @@ export async function runOperator( config, `UPDATE scopes SET enabled = coalesce(?, enabled), writes_enabled = coalesce(?, writes_enabled), pull_requests = coalesce(?, pull_requests), retention_seconds = coalesce(?, retention_seconds), byte_limit = coalesce(?, byte_limit), entry_limit = coalesce(?, entry_limit), - association_limit = coalesce(?, association_limit), policy_version = policy_version + 1 WHERE scope_id = ?`, + association_limit = coalesce(?, association_limit), entries_per_key = coalesce(?, entries_per_key), + policy_version = policy_version + 1 WHERE scope_id = ?`, [ toggle(values.enabled), toggle(values.writes), @@ -529,6 +540,7 @@ export async function runOperator( optionalPositive(values['byte-limit']), optionalPositive(values['entry-limit']), optionalPositive(values['association-limit']), + entriesPerKey(values['entries-per-key']), namespace, ], ); diff --git a/src/cbor.ts b/src/cbor.ts index 08510e3..706e9f6 100644 --- a/src/cbor.ts +++ b/src/cbor.ts @@ -107,7 +107,7 @@ export function decodeEnvelope( }; } -function header(major: number, length: number): Uint8Array { +export function header(major: number, length: number): Uint8Array { if (length < 24) return Uint8Array.of((major << 5) | length); if (length <= 255) return Uint8Array.of((major << 5) | 24, length); if (length <= 65535) return Uint8Array.of((major << 5) | 25, length >> 8, length & 255); @@ -120,14 +120,17 @@ function header(major: number, length: number): Uint8Array { ); } +// A CBOR text or byte string with its header. +export function cborString(value: string | Uint8Array): Uint8Array[] { + const bytes = typeof value === 'string' ? new TextEncoder().encode(value) : value; + return [header(typeof value === 'string' ? 3 : 2, bytes.length), bytes]; +} + export function encodeEnvelope( fields: Record, ): Uint8Array { const chunks = [header(5, Object.keys(fields).length)]; - const string = (value: string | Uint8Array) => { - const bytes = typeof value === 'string' ? new TextEncoder().encode(value) : value; - chunks.push(header(typeof value === 'string' ? 3 : 2, bytes.length), bytes); - }; + const string = (value: string | Uint8Array) => chunks.push(...cborString(value)); for (const [key, value] of Object.entries(fields)) { string(key); if (value === null) chunks.push(Uint8Array.of(246)); diff --git a/src/database.ts b/src/database.ts index e224e10..1c6e444 100644 --- a/src/database.ts +++ b/src/database.ts @@ -26,6 +26,10 @@ export interface Selection extends Generation { kind: 'exact' | 'fallback'; key: number[]; } +export type Entries = + | { kind: 'exact'; entries: Generation[] } + | { kind: 'fallback'; key: Uint8Array } + | null; // Preserve empty and non-UTF-8 keys as BLOBs, without text or hash conversions. export function binary(bytes: Uint8Array): ArrayBuffer { @@ -51,39 +55,49 @@ export async function getScope(db: D1Database, id: string, stats?: Observations) return scope; } -export async function selectEntry( +export async function selectEntries( db: D1Database, scope: string, key: Uint8Array, secondary: Uint8Array, stats?: Observations, -): Promise { - // Both branches use indexed identities in one SQLite snapshot. +): Promise { + // One statement reads the key's newest live entries and the association in one SQLite snapshot. + // The association is used only when the key has no live entry. + let results: Selection[]; try { - const { results } = await measured( + ({ results } = await measured( db .prepare(` - SELECT g.*, e.key, 'exact' AS kind, 0 AS priority - FROM entries e JOIN generations g ON g.generation_id = e.generation_id - JOIN scopes s ON s.scope_id = e.scope_id CROSS JOIN deployment d - WHERE e.scope_id = ?1 AND e.key = ?2 AND g.state = 'ready' AND g.expires_at > unixepoch() - AND s.enabled = 1 AND d.enabled = 1 + SELECT * FROM ( + SELECT g.*, e.key, 'exact' AS kind, 0 AS priority, e.seq + FROM entries e JOIN generations g ON g.generation_id = e.generation_id + JOIN scopes s ON s.scope_id = e.scope_id CROSS JOIN deployment d + WHERE e.scope_id = ?1 AND e.key = ?2 AND g.state = 'ready' AND g.expires_at > unixepoch() + AND s.enabled = 1 AND d.enabled = 1 + ORDER BY e.seq DESC + LIMIT (SELECT entries_per_key FROM scopes WHERE scope_id = ?1)) UNION ALL - SELECT g.*, e.key, 'fallback' AS kind, 1 AS priority - FROM associations a JOIN entries e ON e.scope_id = a.scope_id AND e.key = a.target_key - JOIN generations g ON g.generation_id = e.generation_id - JOIN scopes s ON s.scope_id = e.scope_id CROSS JOIN deployment d - WHERE a.scope_id = ?1 AND a.secondary_key = ?3 AND g.state = 'ready' AND g.expires_at > unixepoch() - AND s.enabled = 1 AND d.enabled = 1 - ORDER BY priority LIMIT 1`) + SELECT * FROM ( + SELECT g.*, e.key, 'fallback' AS kind, 1 AS priority, e.seq + FROM associations a JOIN entries e ON e.scope_id = a.scope_id AND e.key = a.target_key + JOIN generations g ON g.generation_id = e.generation_id + JOIN scopes s ON s.scope_id = e.scope_id CROSS JOIN deployment d + WHERE a.scope_id = ?1 AND a.secondary_key = ?3 AND g.state = 'ready' AND g.expires_at > unixepoch() + AND s.enabled = 1 AND d.enabled = 1 + LIMIT 1) + ORDER BY priority, seq DESC`) .bind(scope, binary(key), binary(secondary)) .all(), stats, - ); - return results[0] ?? null; + )); } catch { unavailable(); } + const exact = results.filter((row) => row.kind === 'exact'); + if (exact.length) return { kind: 'exact', entries: exact }; + const fallback = results[0]; + return fallback ? { kind: 'fallback', key: new Uint8Array(fallback.key) } : null; } export async function selectBlob( diff --git a/src/fetch.ts b/src/fetch.ts index 73b8283..902e162 100644 --- a/src/fetch.ts +++ b/src/fetch.ts @@ -1,23 +1,20 @@ -import { decodeEnvelope } from './cbor.ts'; -import { selectEntry } from './database.ts'; +import { cborResponse, cborString, decodeEnvelope, header } from './cbor.ts'; +import { selectEntries } from './database.ts'; import { badRequest, HttpError, unavailable } from './errors.ts'; import type { Limits } from './limits.ts'; import { parameters } from './multipart.ts'; import type { Observations } from './observations.ts'; import { contentLength, Deadline, readBody } from './streams.ts'; -type FetchResult = - | { kind: 'exact'; value: Uint8Array; blob_id: string | null } - | { kind: 'fallback'; key: Uint8Array }; - export async function fetchMetadata( request: Request, env: Pick, + ctx: Pick, scope: string, limits: Limits, deadline: Deadline, stats: Observations, -): Promise { +): Promise<{ kind: 'exact' | 'fallback'; response: Response }> { if (parameters(request.headers.get('Content-Type') ?? '').type !== 'application/cbor') badRequest(); contentLength(request, limits.fetch); @@ -25,18 +22,87 @@ export async function fetchMetadata( stats.request_bytes = body.length; const data = decodeEnvelope(body, false, limits); const selected = await deadline.run( - selectEntry(env.INDEX, scope, data.key, data.secondary_key, stats), + selectEntries(env.INDEX, scope, data.key, data.secondary_key, stats), ); if (!selected) throw new HttpError(404, 'miss'); // Fallbacks only identify the associated key. They must not depend on R2 availability. - if (selected.kind === 'fallback') return { kind: 'fallback', key: new Uint8Array(selected.key) }; + if (selected.kind === 'fallback') + return { kind: 'fallback', response: cborResponse({ kind: 'fallback', key: selected.key }) }; - stats.r2_operations++; - const object = await deadline - .run(env.ARTIFACTS.get(selected.value_object)) + const { entries } = selected; + stats.r2_operations += entries.length; + const objects = await deadline + .run(Promise.all(entries.map((entry) => env.ARTIFACTS.get(entry.value_object)))) .catch(() => unavailable()); - if (!object || object.size !== selected.value_size || object.size > limits.value) unavailable(); - const value = await readBody(object.body, limits.value, deadline).catch(() => unavailable()); - if (value.length !== selected.value_size) unavailable(); - return { kind: 'exact', value, blob_id: selected.blob_id }; + // Check every value before responding, so a missing or resized object is still a 503. + if ( + objects.some( + (object, i) => + !object || object.size !== entries[i]!.value_size || object.size > limits.value, + ) + ) { + for (const object of objects) object?.body.cancel().catch(() => {}); + unavailable(); + } + + // { kind: "exact", entries: [{ value, blob_id }] }, newest first. Values stream from R2, + // so a response with many large values isn't held in memory. + const parts: (Uint8Array | ReadableStream)[] = [ + header(5, 2), + ...cborString('kind'), + ...cborString('exact'), + ...cborString('entries'), + header(4, entries.length), + ]; + for (const [i, entry] of entries.entries()) { + parts.push(header(5, 2), ...cborString('value'), header(2, entry.value_size)); + parts.push(objects[i]!.body); + parts.push(...cborString('blob_id')); + if (entry.blob_id === null) parts.push(Uint8Array.of(246)); + else parts.push(...cborString(entry.blob_id)); + } + const length = + parts.reduce((sum, part) => sum + (part instanceof Uint8Array ? part.length : 0), 0) + + entries.reduce((sum, entry) => sum + entry.value_size, 0); + const stream = new FixedLengthStream(length); + ctx.waitUntil(pump(parts, stream.writable)); + return { + kind: 'exact', + response: new Response(stream.readable, { + headers: { + 'Content-Type': 'application/cbor', + 'Content-Length': String(length), + 'Cache-Control': 'no-store', + }, + }), + }; +} + +async function pump( + parts: (Uint8Array | ReadableStream)[], + writable: WritableStream, +): Promise { + const writer = writable.getWriter(); + let next = 0; + try { + for (; next < parts.length; next++) { + const part = parts[next]!; + if (part instanceof Uint8Array) { + await writer.write(part); + continue; + } + const reader = part.getReader(); + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + await writer.write(value); + } + } + await writer.close(); + } catch (error) { + // A truncated body fails the client's CBOR decoding, which it treats as a failed read. + await writer.abort(error).catch(() => {}); + for (const part of parts.slice(next + 1)) + if (!(part instanceof Uint8Array)) await part.cancel().catch(() => {}); + } } diff --git a/src/index.ts b/src/index.ts index 5b7c0cb..afb9706 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,5 +1,4 @@ import { authorize, githubKeys, type WriteIdentity } from './auth.ts'; -import { cborResponse } from './cbor.ts'; import { cleanup, getScope, selectBlob } from './database.ts'; import { errorResponse, HttpError, unavailable } from './errors.ts'; import { fetchMetadata } from './fetch.ts'; @@ -64,8 +63,8 @@ export default { response = await store(request, env, ctx, scope, identity, limits, deadline, stats); outcome = 'stored'; } else if (operation === 'fetch') { - const result = await fetchMetadata(request, env, scopeId, limits, deadline, stats); - response = cborResponse(result); + const result = await fetchMetadata(request, env, ctx, scopeId, limits, deadline, stats); + response = result.response; outcome = result.kind; } else { const selected = await deadline.run(selectBlob(env.INDEX, scopeId, route[3]!, stats)); diff --git a/test/failures.test.ts b/test/failures.test.ts index 23874aa..c2395ec 100644 --- a/test/failures.test.ts +++ b/test/failures.test.ts @@ -35,6 +35,7 @@ test('fallbacks return only opaque keys without R2 reads, while exact failures r const deadline = new Deadline(10000); const stats = new Observations(); const env = { ...original, ARTIFACTS: bucket }; + const ctx = { waitUntil: () => {} }; try { for (const key of [ Uint8Array.of(0, 255), @@ -44,23 +45,30 @@ test('fallbacks return only opaque keys without R2 reads, while exact failures r const stored = await h.store(key, bytes('S'), bytes('value'), bytes('blob')); assert.equal(stored.status, 200); await stored.arrayBuffer(); - assert.deepEqual( - await fetchMetadata(request(bytes('missing')), env, 'test', defaults, deadline, stats), - { kind: 'fallback', key }, + const result = await fetchMetadata( + request(bytes('missing')), + env, + ctx, + 'test', + defaults, + deadline, + stats, ); + assert.equal(result.kind, 'fallback'); + assert.deepEqual(await decodeResponse(result.response), { kind: 'fallback', key }); assert.equal(reads, 0); assert.equal(stats.r2_operations, 0); } // An exact match still wins when the secondary key points at a different live entry. await assert.rejects( - fetchMetadata(request(Uint8Array.of(0, 255)), env, 'test', defaults, deadline, stats), + fetchMetadata(request(Uint8Array.of(0, 255)), env, ctx, 'test', defaults, deadline, stats), { status: 503 }, ); assert.equal(reads, 1); assert.equal(stats.r2_operations, 1); await h.db.prepare("UPDATE generations SET expires_at = 0 WHERE state = 'ready'").run(); await assert.rejects( - fetchMetadata(request(bytes('missing')), env, 'test', defaults, deadline, stats), + fetchMetadata(request(bytes('missing')), env, ctx, 'test', defaults, deadline, stats), { status: 404 }, ); assert.equal(reads, 1); @@ -144,7 +152,7 @@ test('R2 PUT, multipart creation, part, completion, and upload-ID recording fail ); await Promise.all(pending); assert.deepEqual( - (await decodeResponse(await h.fetch(bytes('A'), bytes('S')))).value, + (await decodeResponse(await h.fetch(bytes('A'), bytes('S')))).entries[0].value, bytes('old'), ); assert.equal((await h.fetch(bytes('missing'), bytes('T'))).status, 404); @@ -164,6 +172,7 @@ test('R2 PUT, multipart creation, part, completion, and upload-ID recording fail test('failed deletion remains charged and a retry removes only the claimed generation', async () => { const h = await harness(); try { + await h.db.prepare("UPDATE scopes SET entries_per_key = 1 WHERE scope_id = 'test'").run(); await h.store(bytes('A'), bytes('S'), bytes('old'), bytes('blob')); await h.store(bytes('A'), bytes('T'), bytes('new')); await h.db.prepare("UPDATE generations SET gc_after = 0 WHERE state = 'retired'").run(); @@ -194,7 +203,7 @@ test('failed deletion remains charged and a retry removes only the claimed gener key: bytes('A'), }); assert.deepEqual( - (await decodeResponse(await h.fetch(bytes('A'), bytes('S')))).value, + (await decodeResponse(await h.fetch(bytes('A'), bytes('S')))).entries[0].value, bytes('new'), ); } finally { @@ -288,7 +297,7 @@ test('scope/deployment disable and capacity changes revoke pending publications' } }); -test('scope and deployment entry limits roll back association changes, while replacements use no new slot', async () => { +test('scope and deployment entry limits roll back association changes, while more entries for a key use no new slot', async () => { const h = await harness(); try { assert.equal((await h.store(bytes('A'), bytes('S'), bytes('old'))).status, 200); @@ -304,7 +313,7 @@ test('scope and deployment entry limits roll back association changes, while rep await h.db.prepare(`UPDATE ${table} SET association_limit = 1`).run(); assert.equal((await h.store(bytes('A'), bytes('T'), bytes('rejected'))).status, 503); assert.deepEqual( - (await decodeResponse(await h.fetch(bytes('A'), bytes('S')))).value, + (await decodeResponse(await h.fetch(bytes('A'), bytes('S')))).entries[0].value, bytes('replacement'), ); await h.db @@ -347,7 +356,7 @@ test('cleanup preserves a recreated target and claims at most 16 generations', a key: bytes('A'), }); assert.deepEqual( - (await decodeResponse(await h.fetch(bytes('A'), bytes('S')))).value, + (await decodeResponse(await h.fetch(bytes('A'), bytes('S')))).entries[0].value, bytes('new'), ); const scope = await getScope(h.db, 'test'); diff --git a/test/helpers.ts b/test/helpers.ts index a4ae97b..aa3211a 100644 --- a/test/helpers.ts +++ b/test/helpers.ts @@ -84,7 +84,7 @@ export async function harness( for (const name of (await readdir(migrations)).filter((file) => file.endsWith('.sql')).sort()) statements.push( ...(await readFile(new URL(name, migrations), 'utf8')).match( - /CREATE TRIGGER[\s\S]*?\nEND;|(?:CREATE TABLE|CREATE (?:UNIQUE )?INDEX|INSERT INTO|ALTER TABLE|DROP TRIGGER)[\s\S]*?;/g, + /CREATE TRIGGER[\s\S]*?\nEND;|(?:CREATE TABLE|CREATE (?:UNIQUE )?INDEX|INSERT INTO|UPDATE|ALTER TABLE|DROP TRIGGER|DROP TABLE)[\s\S]*?;/g, )!, ); const migrate = () => db.batch(statements.map((sql) => db.prepare(sql))); diff --git a/test/operator.test.ts b/test/operator.test.ts index 05d0425..cd0cdb8 100644 --- a/test/operator.test.ts +++ b/test/operator.test.ts @@ -120,6 +120,14 @@ test('operator setup validates before writes and preserves withdrawal on retries const opted = await h.db.prepare("SELECT * FROM scopes WHERE scope_id = 'new'").first(); assert.equal(opted!.pull_requests, 1); assert.equal(opted!.policy_version, 3); + assert.equal(opted!.entries_per_key, 8); + for (const value of ['0', '17', 'many']) + await assert.rejects( + runOperator(['policy', '--namespace', 'new', '--entries-per-key', value], io), + ); + await runOperator(['policy', '--namespace', 'new', '--entries-per-key', '16'], io); + const deeper = await h.db.prepare("SELECT * FROM scopes WHERE scope_id = 'new'").first(); + assert.equal(deeper!.entries_per_key, 16); assert.equal(config.workers_dev, false); assert.equal(config.preview_urls, false); assert.deepEqual(config.routes, [{ pattern: 'cache.example.com', custom_domain: true }]); @@ -269,6 +277,8 @@ test('operator setup creates resources and initializes an empty database', async 'https://cache.example.com', '--pull-requests', 'on', + '--entries-per-key', + '4', ], io, ); @@ -276,10 +286,13 @@ test('operator setup creates resources and initializes an empty database', async assert.equal(config.vars['GC_BATCH_SIZE'], '16'); assert.deepEqual(JSON.parse(config.vars['NAMESPACES']!), ['test']); const scope = await h.db - .prepare("SELECT endpoint, pull_requests FROM scopes WHERE scope_id = 'test'") + .prepare( + "SELECT endpoint, pull_requests, entries_per_key FROM scopes WHERE scope_id = 'test'", + ) .first(); assert.equal(scope!.endpoint, 'https://cache.example.com/projects/test'); assert.equal(scope!.pull_requests, 1); + assert.equal(scope!.entries_per_key, 4); } finally { await h.close(); } diff --git a/test/protocol.test.ts b/test/protocol.test.ts index 2ac7080..b5526c5 100644 --- a/test/protocol.test.ts +++ b/test/protocol.test.ts @@ -14,9 +14,9 @@ test('CBOR fixtures preserve opaque, empty, indefinite, and noncanonical bytes', assert.deepEqual(Array.from(actual.secondary_key), fixtures.expectedSecondaryKey, fixture.name); } const value = Uint8Array.of(0, 255, 159, 255); - assert.deepEqual(decode(encodeEnvelope({ kind: 'exact', value, blob_id: null })), { - kind: 'exact', - value, + assert.deepEqual(decode(encodeEnvelope({ kind: 'fallback', key: value, blob_id: null })), { + kind: 'fallback', + key: value, blob_id: null, }); }); diff --git a/test/service.test.ts b/test/service.test.ts index 3b4daee..a13af0d 100644 --- a/test/service.test.ts +++ b/test/service.test.ts @@ -4,7 +4,7 @@ import { encode } from 'cborg'; import { bytes, decodeResponse, endpoint, harness } from './helpers.ts'; import { getScope, publish, reserve } from '../src/database.ts'; -test('Workers D1/R2: exact/fallback, replacement, empty blobs, namespace isolation, and read-only access', async () => { +test('Workers D1/R2: exact/fallback, appended entries, empty blobs, namespace isolation, and read-only access', async () => { const h = await harness(); try { const a = Uint8Array.of(0, 255), @@ -19,16 +19,23 @@ test('Workers D1/R2: exact/fallback, replacement, empty blobs, namespace isolati assert.equal((await h.store(b, s, bytes('VB'))).status, 200); assert.deepEqual(await decodeResponse(await h.fetch(a, s)), { kind: 'exact', - value: bytes('VA'), - blob_id: firstBlob, + entries: [{ value: bytes('VA'), blob_id: firstBlob }], }); assert.deepEqual(await decodeResponse(await h.fetch(c, s)), { kind: 'fallback', key: b, }); - const replacement = await h.store(a, t, bytes('VA2'), new Uint8Array(), { blobFirst: true }); - assert.equal(replacement.status, 200); - const emptyBlob = (await decodeResponse(replacement)).blob_id; + // A second store adds an entry. Fetch returns both, newest first. + const second = await h.store(a, t, bytes('VA2'), new Uint8Array(), { blobFirst: true }); + assert.equal(second.status, 200); + const emptyBlob = (await decodeResponse(second)).blob_id; + assert.deepEqual(await decodeResponse(await h.fetch(a, s)), { + kind: 'exact', + entries: [ + { value: bytes('VA2'), blob_id: emptyBlob }, + { value: bytes('VA'), blob_id: firstBlob }, + ], + }); assert.equal(typeof emptyBlob, 'string'); assert.equal(await (await h.mf.dispatchFetch(`${endpoint}/blob/${firstBlob}`)).text(), 'old'); assert.equal((await h.mf.dispatchFetch(`${endpoint}/blob/${emptyBlob}`)).status, 200); @@ -52,8 +59,11 @@ test('Workers D1/R2: exact/fallback, replacement, empty blobs, namespace isolati ).status, 200, ); - assert.deepEqual((await decodeResponse(await h.fetch(a, s, 'other'))).value, bytes('other')); - assert.deepEqual((await decodeResponse(await h.fetch(a, s))).value, bytes('VA2')); + assert.deepEqual( + (await decodeResponse(await h.fetch(a, s, 'other'))).entries[0].value, + bytes('other'), + ); + assert.deepEqual((await decodeResponse(await h.fetch(a, s))).entries[0].value, bytes('VA2')); assert.deepEqual(await decodeResponse(await h.fetch(c, t)), { kind: 'fallback', key: a }); assert.equal((await h.fetch(c, t, 'other')).status, 404); const before = await h.db.prepare('SELECT * FROM deployment').first(); @@ -69,6 +79,56 @@ test('Workers D1/R2: exact/fallback, replacement, empty blobs, namespace isolati } }); +test('a key keeps its newest entries_per_key entries and counts as one entry', async () => { + const h = await harness(); + try { + await h.db.prepare("UPDATE scopes SET entries_per_key = 2 WHERE scope_id = 'test'").run(); + const blobs: string[] = []; + for (const value of ['one', 'two', 'three']) { + const response = await h.store(bytes('A'), bytes('S'), bytes(value), bytes(value)); + assert.equal(response.status, 200, await response.clone().text()); + blobs.push((await decodeResponse(response)).blob_id); + } + assert.deepEqual(await decodeResponse(await h.fetch(bytes('A'), bytes('S'))), { + kind: 'exact', + entries: [ + { value: bytes('three'), blob_id: blobs[2] }, + { value: bytes('two'), blob_id: blobs[1] }, + ], + }); + // The oldest entry is retired, and its blob stays downloadable for the grace period. + const states = await h.db + .prepare('SELECT blob_id, state FROM generations ORDER BY state, blob_id') + .all<{ blob_id: string; state: string }>(); + assert.deepEqual( + states.results.filter((row) => row.state === 'retired').map((row) => row.blob_id), + [blobs[0]], + ); + assert.equal(await (await h.mf.dispatchFetch(`${endpoint}/blob/${blobs[0]}`)).text(), 'one'); + assert.deepEqual( + await h.db.prepare("SELECT entry_count FROM scopes WHERE scope_id = 'test'").first(), + { entry_count: 1 }, + ); + assert.equal((await h.db.prepare('SELECT count(*) AS n FROM entries').first())!.n, 2); + // Expired entries are left out, and cleanup removes them and uncounts the key. + await h.db + .prepare('UPDATE generations SET expires_at = 0, gc_after = 0 WHERE blob_id = ?') + .bind(blobs[2]) + .run(); + assert.deepEqual((await decodeResponse(await h.fetch(bytes('A'), bytes('S')))).entries, [ + { value: bytes('two'), blob_id: blobs[1] }, + ]); + await h.db.prepare('UPDATE generations SET expires_at = 0, gc_after = 0').run(); + await (await h.mf.getWorker()).scheduled(); + assert.equal((await h.fetch(bytes('A'), bytes('S'))).status, 404); + const accounting = await h.db.prepare('SELECT * FROM deployment').first(); + assert.equal(accounting!.entry_count, 0); + assert.equal(accounting!.charged_bytes, 0); + } finally { + await h.close(); + } +}); + test('signed GitHub policy denies forks, wrong owners, private repos, non-push events, branches, audiences, and invalid times', async () => { const h = await harness(); try { @@ -250,7 +310,7 @@ test('publication guard and quotas preserve BOTH mappings and charged bytes', as 'uploading', ); assert.deepEqual( - (await decodeResponse(await h.fetch(bytes('A'), bytes('S')))).value, + (await decodeResponse(await h.fetch(bytes('A'), bytes('S')))).entries[0].value, bytes('old'), ); assert.equal((await h.fetch(bytes('missing'), bytes('T'))).status, 404); @@ -259,7 +319,7 @@ test('publication guard and quotas preserve BOTH mappings and charged bytes', as await h.db.prepare('UPDATE deployment SET association_limit = 1').run(); assert.equal((await h.store(bytes('A'), bytes('new-S'), bytes('new'))).status, 503); assert.deepEqual( - (await decodeResponse(await h.fetch(bytes('A'), bytes('S')))).value, + (await decodeResponse(await h.fetch(bytes('A'), bytes('S')))).entries[0].value, bytes('old'), ); assert.equal((await h.store(bytes('A'), bytes('S'), bytes('update'))).status, 200); @@ -287,13 +347,16 @@ test('large sequential R2 multipart stores and concurrent stores return coherent ), ); for (const result of stored) assert.equal(result.status, 200); + // Concurrent stores to one key are all kept, each with its own blob. const final = await decodeResponse(await h.fetch(bytes('race'), bytes('race'))); - assert.deepEqual( - new Uint8Array( - await (await h.mf.dispatchFetch(`${endpoint}/blob/${final.blob_id}`)).arrayBuffer(), - ), - final.value, - ); + assert.equal(final.entries.length, 2); + for (const entry of final.entries) + assert.deepEqual( + new Uint8Array( + await (await h.mf.dispatchFetch(`${endpoint}/blob/${entry.blob_id}`)).arrayBuffer(), + ), + entry.value, + ); const accounting = await h.db .prepare( 'SELECT charged_bytes, (SELECT sum(charged_bytes) FROM generations) AS expected FROM deployment', @@ -323,6 +386,7 @@ test('large sequential R2 multipart stores and concurrent stores return coherent test('retention, retirement, abandoned uploads, and bounded cleanup protect replacements', async () => { const h = await harness(); try { + await h.db.prepare("UPDATE scopes SET entries_per_key = 1 WHERE scope_id = 'test'").run(); const oldBlob = ( await decodeResponse(await h.store(bytes('A'), bytes('S'), bytes('old'), bytes('old'))) ).blob_id; @@ -331,10 +395,11 @@ test('retention, retirement, abandoned uploads, and bounded cleanup protect repl assert.equal((await h.mf.dispatchFetch(`${endpoint}/blob/${oldBlob}`)).status, 404); await (await h.mf.getWorker()).scheduled(); assert.deepEqual( - (await decodeResponse(await h.fetch(bytes('A'), bytes('S')))).value, + (await decodeResponse(await h.fetch(bytes('A'), bytes('S')))).entries[0].value, bytes('new'), ); - const expiredBlob = (await decodeResponse(await h.fetch(bytes('A'), bytes('S')))).blob_id; + const expiredBlob = (await decodeResponse(await h.fetch(bytes('A'), bytes('S')))).entries[0] + .blob_id; await h.db .prepare("UPDATE generations SET expires_at = 0, gc_after = 0 WHERE state = 'ready'") .run(); @@ -467,7 +532,7 @@ test('configured field and streamed request limits return 413 without publicatio }); assert.equal(response.status, 200, await response.clone().text()); assert.deepEqual( - (await decodeResponse(await h.fetch(bytes('A'), bytes('S')))).value, + (await decodeResponse(await h.fetch(bytes('A'), bytes('S')))).entries[0].value, bytes('V'), ); } finally { From 78764c66432639b5215171e3186cec2120d55cc2 Mon Sep 17 00:00:00 2001 From: wan9chi Date: Sat, 7 Nov 2026 17:21:04 +0800 Subject: [PATCH 2/2] fix: keep live entries and stop R2 reads on cancelled fetches Publication ranked every entry row of a key, including expired rows that Cron hasn't removed yet. After a retention cut a newer entry can expire first, and its row then pushed an older live entry out. Rank only live entries. When a client cancelled a streamed fetch, the R2 body being copied was never cancelled, so a stalled read kept the copy and the remaining bodies alive. Cancel the active body when the response errors, and cancel opened bodies when another R2 read fails. Co-Authored-By: Claude Opus 5.5 --- migrations/0003_entries_per_key.sql | 16 ++++++----- rfcs/0003-several-entries-per-key.md | 16 +++++------ src/fetch.ts | 20 +++++++++----- test/protocol.test.ts | 40 ++++++++++++++++++++++++++++ test/service.test.ts | 22 +++++++++++++++ 5 files changed, 93 insertions(+), 21 deletions(-) diff --git a/migrations/0003_entries_per_key.sql b/migrations/0003_entries_per_key.sql index db1611d..59fd35b 100644 --- a/migrations/0003_entries_per_key.sql +++ b/migrations/0003_entries_per_key.sql @@ -111,7 +111,9 @@ END; -- All publication mutations inherit the guarded state change's transaction. -- A zero-row guard fires no trigger. A quota failure rolls every mutation back. --- Publication adds an entry, then retires the key's entries past its newest `entries_per_key`. +-- Publication adds an entry, then retires the key's live entries past its newest `entries_per_key`. +-- Only live entries are ranked: after a retention cut a newer entry can expire before an older one, +-- and an expired row Cron hasn't removed yet must not push a live entry out. CREATE TRIGGER publish_generation AFTER UPDATE OF state ON generations WHEN OLD.state = 'uploading' AND NEW.state = 'ready' BEGIN @@ -120,14 +122,14 @@ BEGIN (SELECT coalesce(max(seq), 0) + 1 FROM entries WHERE scope_id = NEW.scope_id AND key = NEW.key)); UPDATE generations SET state = 'retired', retired_at = unixepoch(), gc_after = (CASE WHEN expires_at > unixepoch() THEN unixepoch() + 600 ELSE unixepoch() END) - WHERE state = 'ready' AND generation_id IN ( - SELECT generation_id FROM entries WHERE scope_id = NEW.scope_id AND key = NEW.key - ORDER BY seq DESC + WHERE generation_id IN ( + SELECT g.generation_id FROM entries e JOIN generations g ON g.generation_id = e.generation_id + WHERE e.scope_id = NEW.scope_id AND e.key = NEW.key AND g.state = 'ready' AND g.expires_at > unixepoch() + ORDER BY e.seq DESC LIMIT -1 OFFSET (SELECT entries_per_key FROM scopes WHERE scope_id = NEW.scope_id)); DELETE FROM entries WHERE scope_id = NEW.scope_id AND key = NEW.key AND generation_id IN ( - SELECT generation_id FROM entries WHERE scope_id = NEW.scope_id AND key = NEW.key - ORDER BY seq DESC - LIMIT -1 OFFSET (SELECT entries_per_key FROM scopes WHERE scope_id = NEW.scope_id)); + SELECT g.generation_id FROM entries e JOIN generations g ON g.generation_id = e.generation_id + WHERE e.scope_id = NEW.scope_id AND e.key = NEW.key AND g.state = 'retired'); INSERT INTO associations (scope_id, secondary_key, target_key) VALUES (NEW.scope_id, NEW.secondary_key, NEW.key) ON CONFLICT (scope_id, secondary_key) DO UPDATE SET target_key = excluded.target_key; END; diff --git a/rfcs/0003-several-entries-per-key.md b/rfcs/0003-several-entries-per-key.md index 663694a..ba81419 100644 --- a/rfcs/0003-several-entries-per-key.md +++ b/rfcs/0003-several-entries-per-key.md @@ -85,7 +85,7 @@ The tables stay. These change: `seq` numbers a key's publications in order. A timestamp would tie when two stores publish in the same second. -- **The publication trigger** adds the entry, then retires the entries past the newest N: +- **The publication trigger** adds the entry, then retires the live entries past the newest N. It ranks only live entries: after a retention cut, a newer entry can expire before an older one, and an expired row that Cron hasn't removed yet must not push a live entry out. ```sql CREATE TRIGGER publish_generation AFTER UPDATE OF state ON generations @@ -96,14 +96,14 @@ The tables stay. These change: (SELECT coalesce(max(seq), 0) + 1 FROM entries WHERE scope_id = NEW.scope_id AND key = NEW.key)); UPDATE generations SET state = 'retired', retired_at = unixepoch(), gc_after = (CASE WHEN expires_at > unixepoch() THEN unixepoch() + 600 ELSE unixepoch() END) - WHERE state = 'ready' AND generation_id IN ( - SELECT generation_id FROM entries WHERE scope_id = NEW.scope_id AND key = NEW.key - ORDER BY seq DESC + WHERE generation_id IN ( + SELECT g.generation_id FROM entries e JOIN generations g ON g.generation_id = e.generation_id + WHERE e.scope_id = NEW.scope_id AND e.key = NEW.key AND g.state = 'ready' AND g.expires_at > unixepoch() + ORDER BY e.seq DESC LIMIT -1 OFFSET (SELECT entries_per_key FROM scopes WHERE scope_id = NEW.scope_id)); DELETE FROM entries WHERE scope_id = NEW.scope_id AND key = NEW.key AND generation_id IN ( - SELECT generation_id FROM entries WHERE scope_id = NEW.scope_id AND key = NEW.key - ORDER BY seq DESC - LIMIT -1 OFFSET (SELECT entries_per_key FROM scopes WHERE scope_id = NEW.scope_id)); + SELECT g.generation_id FROM entries e JOIN generations g ON g.generation_id = e.generation_id + WHERE e.scope_id = NEW.scope_id AND e.key = NEW.key AND g.state = 'retired'); INSERT INTO associations (scope_id, secondary_key, target_key) VALUES (NEW.scope_id, NEW.secondary_key, NEW.key) ON CONFLICT (scope_id, secondary_key) DO UPDATE SET target_key = excluded.target_key; END; @@ -115,7 +115,7 @@ The tables stay. These change: ## 5. Costs and limits -**Fetch.** A fetch makes up to N R2 reads, at most six at a time, and returns up to N values. The value limit stays 4 MiB, so a response can reach N × 4 MiB; the Worker streams it. +**Fetch.** A fetch makes up to N R2 reads, at most six at a time, and returns up to N values. The value limit stays 4 MiB, so a response can reach N × 4 MiB; the Worker streams it. If the client cancels the response, the Worker cancels the R2 reads it hasn't finished. **Values must be compressed.** `vp run` should compress values with zstd before storing them. In a replay of five public repositories' CI, values as the client sends them today measured: diff --git a/src/fetch.ts b/src/fetch.ts index 902e162..d287057 100644 --- a/src/fetch.ts +++ b/src/fetch.ts @@ -31,10 +31,11 @@ export async function fetchMetadata( const { entries } = selected; stats.r2_operations += entries.length; - const objects = await deadline - .run(Promise.all(entries.map((entry) => env.ARTIFACTS.get(entry.value_object)))) + const settled = await deadline + .run(Promise.allSettled(entries.map((entry) => env.ARTIFACTS.get(entry.value_object)))) .catch(() => unavailable()); - // Check every value before responding, so a missing or resized object is still a 503. + const objects = settled.map((result) => (result.status === 'fulfilled' ? result.value : null)); + // Check every value before responding, so a missing, failed, or resized object is still a 503. if ( objects.some( (object, i) => @@ -78,11 +79,16 @@ export async function fetchMetadata( }; } -async function pump( +// Write `parts` in order, copying each stream's chunks. If the client cancels the response, the +// writable errors: the stream being copied is cancelled, so a pending read can't keep it open, and +// so are the streams not reached yet. +export async function pump( parts: (Uint8Array | ReadableStream)[], writable: WritableStream, ): Promise { const writer = writable.getWriter(); + let active: ReadableStreamDefaultReader | undefined; + writer.closed.catch(() => active?.cancel().catch(() => {})); let next = 0; try { for (; next < parts.length; next++) { @@ -91,17 +97,19 @@ async function pump( await writer.write(part); continue; } - const reader = part.getReader(); + active = part.getReader(); for (;;) { - const { done, value } = await reader.read(); + const { done, value } = await active.read(); if (done) break; await writer.write(value); } + active = undefined; } await writer.close(); } catch (error) { // A truncated body fails the client's CBOR decoding, which it treats as a failed read. await writer.abort(error).catch(() => {}); + await active?.cancel().catch(() => {}); for (const part of parts.slice(next + 1)) if (!(part instanceof Uint8Array)) await part.cancel().catch(() => {}); } diff --git a/test/protocol.test.ts b/test/protocol.test.ts index b5526c5..b434388 100644 --- a/test/protocol.test.ts +++ b/test/protocol.test.ts @@ -2,6 +2,7 @@ import assert from 'node:assert/strict'; import { test } from 'vite-plus/test'; import { decode, encode } from 'cborg'; import { decodeEnvelope, encodeEnvelope } from '../src/cbor.ts'; +import { pump } from '../src/fetch.ts'; import { defaults } from '../src/limits.ts'; import { multipart } from '../src/multipart.ts'; import { collect, Deadline, Input } from '../src/streams.ts'; @@ -100,3 +101,42 @@ test('multipart accepts every split point, either part order, and boundary-like } } }); + +test('a cancelled response cancels the value being copied and the ones after it', async () => { + let copying = false, + later = false; + // An R2 body whose next read never settles on its own. + const stalled = new ReadableStream({ + pull: () => new Promise(() => {}), + cancel: () => { + copying = true; + }, + }); + const rest = new ReadableStream({ + cancel: () => { + later = true; + }, + }); + const { readable, writable } = new TransformStream(); + const done = pump([Uint8Array.of(1), stalled, Uint8Array.of(2), rest], writable); + const reader = readable.getReader(); + assert.deepEqual((await reader.read()).value, Uint8Array.of(1)); + await reader.cancel(); + // Without cancellation the pending read never settles; fail instead of hanging. + let timer: ReturnType | undefined; + try { + await Promise.race([ + done, + new Promise((_, reject) => { + timer = setTimeout( + () => reject(new Error('the copy outlived the cancelled response')), + 1000, + ); + }), + ]); + } finally { + clearTimeout(timer); + } + assert.equal(copying, true); + assert.equal(later, true); +}); diff --git a/test/service.test.ts b/test/service.test.ts index a13af0d..6817133 100644 --- a/test/service.test.ts +++ b/test/service.test.ts @@ -129,6 +129,28 @@ test('a key keeps its newest entries_per_key entries and counts as one entry', a } }); +test("expired entries take no slot when a store retires a key's oldest entries", async () => { + const h = await harness(); + try { + await h.db.prepare("UPDATE scopes SET entries_per_key = 2 WHERE scope_id = 'test'").run(); + const store = async (value: string) => + (await decodeResponse(await h.store(bytes('A'), bytes('S'), bytes(value), bytes(value)))) + .blob_id; + const a = await store('a'); + const b = await store('b'); + // After a retention cut, a newer entry can expire before an older one. Until Cron removes it, + // its row must not push the older live entry out. + await h.db.prepare('UPDATE generations SET expires_at = 0 WHERE blob_id = ?').bind(b).run(); + const c = await store('c'); + assert.deepEqual((await decodeResponse(await h.fetch(bytes('A'), bytes('S')))).entries, [ + { value: bytes('c'), blob_id: c }, + { value: bytes('a'), blob_id: a }, + ]); + } finally { + await h.close(); + } +}); + test('signed GitHub policy denies forks, wrong owners, private repos, non-push events, branches, audiences, and invalid times', async () => { const h = await harness(); try {