feat(registry): ingest signed labels in the aggregator#1959
Conversation
Adds the per-labeler subscribeLabels ingest path: a LabelIngestDO per configured labeler maintains the outbound WebSocket, verifies every label against the resolved #atproto_label key before queue acceptance, and persists its cursor to ingest_state only after durable enqueue. A new labels queue consumer writes append-only history and the label_state projection in one atomic batch. Replaces the label history identity (src, uri, val, cts) with a collision-safe digest primary key plus unique (src, source_sequence, frame_index) coordinates, and adds validated epoch-millisecond columns so SQL never orders RFC 3339 text. Tables were empty in every deployment (no writer existed; production preflight confirmed zero rows).
…rison Adversarial review findings: the search enforcement filter still matched the legacy 'security:yanked' value (canonical labels use 'security-yanked') and compared RFC 3339 expiry strings as text, which misorders across timezone offsets. Both were dormant against the empty label_state table and would have gone live with the new ingest writer. The filter now matches the canonical value and compares the validated epoch-millisecond column. Also bounds the stream client's inbound decode buffer, failing closed on overflow.
|
Scope checkThis PR changes 2,826 lines across 16 files. Large PRs are harder to review and more likely to be closed without review. If this scope is intentional, no action needed. A maintainer will review it. If not, please consider splitting this into smaller PRs. See CONTRIBUTING.md for contribution guidelines. |
github-actions
Bot
added
review/needs-review
Deploying with
|
| Status | Name | Latest Commit | Updated (UTC) |
|---|---|---|---|
| ✅ Deployment successful! View logs |
emdash-playground | 753b512 | Jul 11 2026, 06:07 PM |
Deploying with
|
| Status | Name | Latest Commit | Updated (UTC) |
|---|---|---|---|
| ✅ Deployment successful! View logs |
emdash-demo-do | 753b512 | Jul 11 2026, 06:06 PM |
Deploying with
|
| Status | Name | Latest Commit | Updated (UTC) |
|---|---|---|---|
| ✅ Deployment successful! View logs |
emdash-demo-cache | 753b512 | Jul 11 2026, 06:07 PM |
![emdashbot[bot]](https://avatars.githubusercontent.com/in/3255304?s=60&v=4){kind=small}
There was a problem hiding this comment.
I reviewed the signed-label ingest pipeline introduced by this PR: the Durable Object lifecycle and queue wiring, the streaming client, the ingestor's verification/cursor/backoff logic, the labels consumer's batch write path, the labeler resolver, the searchPackages enforcement-filter fixes, the migration, the wrangler/queue/DO configuration, and the associated unit tests.
The pipeline is well-structured and the tests cover most of the intended behavior (signature verification, cursor persistence, redelivery idempotency, coordinate conflicts, expiry, source snapshots). However, I found three logic bugs that should block merge:
-
LabelIngestorbackoff accounting swallows verification failures whenever any earlier frame succeeded.madeProgressis set totrueafter every fully processed frame, so a connection that processes some valid frames and then fails verification is treated as a successful run and resetsconsecutiveFailures. This contradicts the fail-closed intent and lets a misbehaving or compromised labeler force a tight reconnect loop. -
The cron
wakeLabelIngestDOsfire-and-forgets DO fetches inside an awaited function. The promises are not collected or returned, soctx.waitUntil(wakeLabelIngestDOs(env))may resolve before the fetches complete. If the runtime invocation is released early, some DO wakes may be dropped. -
RealLabelStreamClientaccept()s the WebSocket before attachingmessage/closelisteners. Any frames or close events sent by the labeler immediately after the upgrade can be lost, causing hangs or missed historical replay.
I also flagged two lower-priority suggestions: defending the migration's unconditional table drops, and guarding Date.parse timestamps in the consumer.
Add a regression test for case (1) — process one valid frame then an unverifiable frame in the same connection and assert that consecutiveFailures increases.
| if (this.madeProgress) this._consecutiveFailures = 0; | ||
| else this._consecutiveFailures += 1; | ||
| } catch (err) { | ||
| if (this.madeProgress) this._consecutiveFailures = 0; | ||
| else this._consecutiveFailures += 1; |
There was a problem hiding this comment.
[needs fixing] connectAndConsume() sets this.madeProgress = true (line 262) after every fully-processed frame. If a later label in the same connection fails verification and sets verificationClosed = true (lines 221/237), the method returns into this run loop, which then sees madeProgress === true and resets _consecutiveFailures to 0. That means a stream that alternates one good frame with one unverifiable frame never escalates backoff, contradicting the fail-closed comment at line 247.
The cleanest fix is to mark the connection as not having made safe progress when verificationClosed triggers:
| if (this.madeProgress) this._consecutiveFailures = 0; | |
| else this._consecutiveFailures += 1; | |
| } catch (err) { | |
| if (this.madeProgress) this._consecutiveFailures = 0; | |
| else this._consecutiveFailures += 1; | |
| if (verificationClosed) { | |
| // Fail closed: nothing from this frame is enqueued, the | |
| // cursor doesn't move, and the connection ends here (the | |
| // `finally` below closes `sub`). The run loop's backoff | |
| // counter picks this up as a failed attempt. | |
| this.madeProgress = false; | |
| break; | |
| } |
Also add a regression test: emit one verifiable frame, then an unverifiable frame in the same connection, and assert that ingestor.consecutiveFailures increases.
There was a problem hiding this comment.
Fixed in 753b512: verificationClosed now clears madeProgress so the connection counts as a failed attempt, with a regression test (good frame then unverifiable frame in one connection asserts consecutiveFailures increases, cursor stays at the good frame).
| async function wakeLabelIngestDOs(env: Env): Promise<void> { | ||
| try { | ||
| const rows = await env.DB.prepare(`SELECT did FROM labelers`).all<{ did: string }>(); | ||
| for (const row of rows.results ?? []) { | ||
| const stub = env.LABEL_INGEST_DO.getByName(row.did); | ||
| // Fire-and-forget per labeler — one slow/unreachable DO shouldn't | ||
| // block waking the others. | ||
| void stub | ||
| .fetch(`https://do.internal/wake?did=${encodeURIComponent(row.did)}`) | ||
| .catch((err: unknown) => { | ||
| console.error("[aggregator] label ingest DO wake failed", { | ||
| did: row.did, | ||
| error: err instanceof Error ? err.message : String(err), | ||
| }); | ||
| }); | ||
| } | ||
| } catch (err) { | ||
| console.error("[aggregator] label ingest DO wake pump failed", { | ||
| error: err instanceof Error ? err.message : String(err), | ||
| }); | ||
| } | ||
| } |
There was a problem hiding this comment.
[needs fixing] wakeLabelIngestDOs is passed to ctx.waitUntil at line 323, but inside the loop it does void stub.fetch(...) and returns as soon as the loop finishes. Because the per-labeler fetches are not awaited or collected, the waitUntil promise can resolve before the fetches complete, and pending work may be dropped when the cron invocation is released. The .catch() on each fetch prevents one slow labeler from rejecting the batch, but the calls still need to be awaited collectively to keep the invocation alive.
Change the function to collect the promises and return Promise.all(...):
| async function wakeLabelIngestDOs(env: Env): Promise<void> { | |
| try { | |
| const rows = await env.DB.prepare(`SELECT did FROM labelers`).all<{ did: string }>(); | |
| for (const row of rows.results ?? []) { | |
| const stub = env.LABEL_INGEST_DO.getByName(row.did); | |
| // Fire-and-forget per labeler — one slow/unreachable DO shouldn't | |
| // block waking the others. | |
| void stub | |
| .fetch(`https://do.internal/wake?did=${encodeURIComponent(row.did)}`) | |
| .catch((err: unknown) => { | |
| console.error("[aggregator] label ingest DO wake failed", { | |
| did: row.did, | |
| error: err instanceof Error ? err.message : String(err), | |
| }); | |
| }); | |
| } | |
| } catch (err) { | |
| console.error("[aggregator] label ingest DO wake pump failed", { | |
| error: err instanceof Error ? err.message : String(err), | |
| }); | |
| } | |
| } | |
| async function wakeLabelIngestDOs(env: Env): Promise<void> { | |
| try { | |
| const rows = await env.DB.prepare(`SELECT did FROM labelers`).all<{ did: string }>(); | |
| const wakes: Promise<void>[] = []; | |
| for (const row of rows.results ?? []) { | |
| const stub = env.LABEL_INGEST_DO.getByName(row.did); | |
| // Fire-and-forget per labeler — one slow/unreachable DO shouldn't | |
| // block waking the others. | |
| wakes.push( | |
| stub | |
| .fetch(`https://do.internal/wake?did=${encodeURIComponent(row.did)}`) | |
| .catch((err: unknown) => { | |
| console.error("[aggregator] label ingest DO wake failed", { | |
| did: row.did, | |
| error: err instanceof Error ? err.message : String(err), | |
| }); | |
| }), | |
| ); | |
| } | |
| await Promise.all(wakes); | |
| } catch (err) { | |
| console.error("[aggregator] label ingest DO wake pump failed", { | |
| error: err instanceof Error ? err.message : String(err), | |
| }); | |
| } | |
| } |
There was a problem hiding this comment.
Fixed in 753b512: the wake fetches are collected and awaited via Promise.all, with the per-fetch catch retained so one unreachable DO can't fail the batch.
| socket.accept(); | ||
| if (ended) { | ||
| // close() won the race against the upgrade: without this the | ||
| // accepted socket would stay open with nobody reading it. | ||
| socket.close(); | ||
| return; | ||
| } | ||
| socket.addEventListener("message", (event: MessageEvent) => { |
There was a problem hiding this comment.
[needs fixing] socket.accept() is called at line 221, but the message and close listeners are not attached until line 228+. If the labeler sends a frame or closes the connection immediately after the WebSocket upgrade, the event is delivered while no handler is listening and is silently lost. This can cause the async iterator to miss replay frames or hang because finish() is never called.
Add the listeners (and explicitly request binary frames) before accepting the socket:
| socket.accept(); | |
| if (ended) { | |
| // close() won the race against the upgrade: without this the | |
| // accepted socket would stay open with nobody reading it. | |
| socket.close(); | |
| return; | |
| } | |
| socket.addEventListener("message", (event: MessageEvent) => { | |
| socket = response.webSocket; | |
| socket.binaryType = "arraybuffer"; | |
| socket.addEventListener("message", (event: MessageEvent) => { | |
| if (!(event.data instanceof ArrayBuffer)) { | |
| deliver({ | |
| kind: "error", | |
| error: new TypeError("subscribeLabels message was not binary"), | |
| }); | |
| socket?.close(); | |
| return; | |
| } | |
| handleMessage(event.data); | |
| }); | |
| socket.addEventListener("close", () => { | |
| finish(); | |
| }); | |
| socket.accept(); | |
| if (ended) { | |
| // close() won the race against the upgrade: without this the | |
| // accepted socket would stay open with nobody reading it. | |
| socket.close(); | |
| return; | |
| } |
There was a problem hiding this comment.
Fixed in 753b512: message and close listeners attach before accept() (the close()-race guard stays after it). Left binaryType unset — workerd delivers binary frames as ArrayBuffer and the handler already fails closed on anything else.
| -- different label landing at coordinates the ingestor already used. `cts`/`exp` | ||
| -- gain epoch-millisecond columns because RFC 3339 strings compare incorrectly | ||
| -- across timezone offsets in SQL. | ||
| DROP TABLE labels; |
There was a problem hiding this comment.
[suggestion] The migration drops labels and label_state unconditionally. The comment states these tables are empty and preflight confirmed it, but a guard that aborts if rows exist would be safer against accidental data loss if that assumption ever changes in an existing deployment. Since this is still pre-1.0 and preflight-confirmed empty, treat as low-priority polish.
There was a problem hiding this comment.
Taken in 753b512 — worth more than polish given self-hosted aggregators we can't preflight: an unsatisfiable-CHECK guard table makes the migration fail if either label table has rows, tested in migrations.test.ts by seeding a row and asserting the migration rejects.
| const ctsEpochMs = Date.parse(label.cts); | ||
| const expEpochMs = label.exp === undefined ? null : Date.parse(label.exp); |
There was a problem hiding this comment.
[suggestion] Date.parse(label.cts) and Date.parse(label.exp) are inserted directly. If a malformed date ever reaches this point, the epoch columns silently become NaN, which can invert comparisons in the state upsert and search-enforcement filters. parseSignedLabel likely validates these today, but an explicit guard here would keep the consumer fail-closed: check Number.isNaN(ctsEpochMs) (and expEpochMs) and dead-letter with LABEL_INVALID rather than committing bad epoch values.
There was a problem hiding this comment.
Taken in 753b512: NaN in either parsed instant dead-letters as LABEL_INVALID before any write.
github-actions
Bot
added
review/awaiting-author
A connection ending on a verification failure no longer counts earlier frames as progress, so an alternating good/unverifiable stream escalates backoff instead of hammering. The scheduled wake pump awaits its DO fetches collectively so waitUntil keeps them alive. Stream listeners attach before accept() so an immediate frame or close can't slip past. The label migration refuses to run against a deployment that already has label rows. The consumer dead-letters any label whose datetimes fail to parse to instants rather than committing NaN epoch values.
github-actions
Bot
added
review/needs-rereview
There was a problem hiding this comment.
This PR implements the signed-label ingest pipeline (labeler DO, stream client, queue consumer, D1 schema migration, and the searchPackages enforcement-filter fixes) against the labelling-service plan. The approach is sound and consistent with the existing records-DO/ingestor architecture.
All three blockers from the previous review are fixed in this revision:
- Backoff accounting after verification failure —
LabelIngestor.connectAndConsumenow resetsmadeProgress = falseat the start of every connection and clears it again onverificationClosed, so a stream that processes some valid frames and then hits an unverifiable frame counts as a failed attempt. A regression test was added. - Cron DO wake Promise collection —
wakeLabelIngestDOsnow awaitsPromise.all(...)for the per-labelerfetchcalls, soctx.waitUntilkeeps the invocation alive until every wake lands. - WebSocket listener ordering —
RealLabelStreamClientattachesmessage/closelisteners before callingsocket.accept().
I have one remaining low-confidence suggestion. The real WebSocket binary-frame path is not exercised by the unit tests (which only call the exported decodeLabelStreamFrame directly with a Uint8Array), and RealLabelStreamClient does not explicitly set socket.binaryType = "arraybuffer". If the runtime ever delivers binary frames as Blob (the WebSocket spec default), the !(event.data instanceof ArrayBuffer) guard would close every connection. Explicitly setting binaryType removes that latent risk.
Overall the pipeline is solid; with the binary-type defensive fix the ingest surface would be robust.
| // Listeners attach before accept() so a frame or close arriving | ||
| // immediately after the upgrade can't slip past unhandled. | ||
| socket.addEventListener("message", (event: MessageEvent) => { |
There was a problem hiding this comment.
[suggestion] RealLabelStreamClient does not explicitly set the WebSocket binaryType. The ATProto label stream sends binary CBOR frames, and this code assumes they arrive as ArrayBuffer. If the runtime default is ever blob, !(event.data instanceof ArrayBuffer) at line 226 will fail and the ingestor will treat every binary frame as a fatal stream error.
Because the unit tests call decodeLabelStreamFrame directly with Uint8Array inputs, the real WebSocket binary path is currently untested, so this latent default would not be caught by the test suite.
| // Listeners attach before accept() so a frame or close arriving | |
| // immediately after the upgrade can't slip past unhandled. | |
| socket.addEventListener("message", (event: MessageEvent) => { | |
| socket = response.webSocket; | |
| // Defensive: ensure binary frames are delivered as ArrayBuffer. | |
| // The labeler sends CBOR; the decode path expects ArrayBuffer input. | |
| socket.binaryType = "arraybuffer"; | |
| socket.addEventListener("message", (event: MessageEvent) => { |
Setting binaryType before accept() (and before any listener can fire) guarantees the subsequent event.data instanceof ArrayBuffer check is reliable.
github-actions
Bot
added
review/awaiting-author
ascorbic
merged commit c5983a5
into
feat/plugin-registry-labelling-service
1 participant
What does this PR do?
Implements aggregator signed-label ingest — W4.2, W4.3, and the collision-safe history migration from the labelling-service plan (#1909).
LabelIngestDO(one per configured labeler, named by DID, woken from the existing cron): maintains the outboundcom.atproto.label.subscribeLabelsWebSocket, verifies every label against the labeler's resolved#atproto_labelkey before queue acceptance, and persists its cursor toingest_state(labeler:<did>) only after every label in a frame is durably enqueued. A fresh labeler subscribes fromcursor=0for full history (the labeler's default is "start from now"). Any unverifiable label fails closed: nothing enqueued, cursor unmoved, connection closed with backoff — after oneresolveFreshretry per connection to absorb key rotation.labels) and thelabel_stateprojection in one atomicdb.batch. History identity is the SHA-256 digest of the canonical signed bytes (exact redelivery is a silent no-op) plusUNIQUE (src, source_sequence, frame_index)(a different label at taken coordinates dead-letters as a permanent conflict). Projection winner per(src, uri, val)is decided by validated epoch-ms instants, tie-broken by stream coordinates, in a conditional upsert that stays correct under concurrent batches and out-of-order delivery.labels/label_statewith the collision-safe identity andcts_epoch_ms/exp_epoch_mscolumns. The tables are empty in every deployment: no writer existed anywhere in code, and the production zero-row preflight was confirmed on Plugin registry labelling service #1909.searchPackages' hard filter matched the legacysecurity:yankedvalue instead of canonicalsecurity-yanked, and compared RFC 3339 expiry strings as text — which misorders across timezone offsets. It now matches the canonical value and comparesexp_epoch_ms. Both regressions are covered by tests that were verified to fail against the old SQL.Review notes:
isCoordinateConflictclassifies the coordinate-unique violation by SQLite error text (both column names must appear). This is exercised against workerd/miniflare D1 in tests; if production D1 ever words it differently, the failure degrades to retry → labels DLQ → forensicsdead_lettersrow rather than silent loss.labelersrows;trustedgates enforcement, not subscription, so untrusted-but-known labelers build history without policy effect.Implementation PRs target
feat/plugin-registry-labelling-serviceper the umbrella PR. Related to #1909 and RFC #694.Type of change
Checklist
pnpm typecheckpassespnpm lintpassespnpm testpasses (or targeted tests for my change)pnpm formathas been runmessages.pochanges except in translation PRs — a workflow extracts catalogs on merge tomain.Admin i18n is n/a — no UI changes. Changeset is n/a —
@emdash-cms/aggregatoris a private package; the only published-package change is adding@emdash-cms/registry-moderationas its dependency, which doesn't alter any published package.AI-generated code disclosure
Screenshots / test output
pnpm --filter @emdash-cms/aggregator test: 13 files, 265 tests pass (44 ingestor/stream-client unit tests with real P-256 signing, 19 workerd/D1 consumer tests, 3 enforcement-filter regression tests)pnpm --filter @emdash-cms/aggregator typecheckcleanpnpm lint:quick0 diagnostics; type-awarepnpm lint:jsonhas no findings in any touched fileTry this PR
Open a fresh playground →
A full working EmDash site, deployed from this branch. Each visit gets its own session-scoped sandbox: no login needed and no shared state. Try the admin, edit content, hit the public site.
Tracks
feat/labeler-aggregator-ingest. Updated automatically when the playground redeploys.