Apply each telemetry event batch incrementally instead of re-materializing the run

[anth-volk]

Fixes #217

Summary

The telemetry collector re-read every stored event of a run, and re-materialized the run, whenever a batch added an event (PostgresTelemetryStore.append_events). Each batch's cost grew with the run's history, which is O(n²) over a run. It also held the run's row lock for longer as a build went on.

The store now folds each batch onto the run's stored columns. A batch's cost is proportional to the batch, and the resulting run state, database contents and HTTP responses are identical to the full re-materialization.

No schema change and no migration. alembic check reports no new upgrade operations.

Merging deploys to production. Merging to main runs .github/workflows/telemetry-service.yml, which deploys the collector to staging and then to production. See Deploy implications.

Why the fold is exact

materialize_run folds events in canonical order (producer registration, producer, sequence) and resets the state at each later producer's first event:

• updated_at is a maximum over every event.
• started_at is a minimum over the final producer's events. It also includes the run's registration only while that producer's events are the run's first.
• Every other column is set by the latest event that writes it, from that event alone.

So replaying any suffix of the active producer's stream that contains the new events, on top of the stored columns, gives exactly the full fold. apply_event_batch handles each case:

Batch	Stored events read
Follows the active producer's stored events (the normal case)	none
Contains a later producer's events (a restarted attempt)	none; its first event resets the state as before
Contains an earlier producer's events	none; they only advance updated_at
An earlier producer's first events	one min(emitted_at) of the active producer, at most once per run
A late event (below its producer's highest stored sequence)	that producer's events from the late one onward

The store inserts with RETURNING the stored rows. The fold therefore sees the same database values (time zone, JSONB) as a full re-read.

materialize_run and apply_event_batch share one fold loop. get_run still materializes from every event, because it returns them all.

Normal delivery reads no stored events

PolicyEngine/microcosm-emitter's telemetry delivery (telemetry/service/collector.py and spool.py) sends one batch per producer, in sequence order. It sends the next batch only after a 202.

• A normal batch is therefore a strict tail of its producer's stream, and reads no stored events.
• A re-send after a lost acknowledgement is all duplicates. It inserts nothing and does no work.

Commits

1. 6993387 Add an incremental event-batch reducer equivalent to full materialization

Tests:

• tests/test_incremental_batch_apply.py, which is pure, with an in-memory run that de-duplicates by event id and by producer and sequence, as the store does;
• the shared tests/event_schedules.py: 18 event kinds covering every materializer branch, plus scripted and seeded delivery schedules.

After every batch, the tests compare the folded columns with materialize_run over all stored events. They cover:

• every event kind, in nine orders and five batch sizes;
• out-of-order batches and events, including skewed clocks;
• duplicate re-sends and a conflicting sequence under a new event id;
• restarted, tied and early-clock producers;
• 300 random schedules that swap, merge, shuffle, re-send, withhold and conflict, while asserting that every kind is covered.

The cost tests count the stored events the reducer reads.

Red: 8 failed, for the intended reason.

• Each failed with NotImplementedError from the stub apply_event_batch. Only the interface existed, with no incremental application.
• Check on the tests themselves: a temporary reducer that replays every stored event, as the store did, passed the 5 equivalence tests and failed the 3 cost tests (for example, 36,600 stored events read where 0 were expected).
• Six deliberate reducer bugs each failed between 2 and 8 tests:
  • skipping the started_at correction;
  • skipping late replay;
  • replaying from the wrong sequence;
  • dropping updated_at from superseded events;
  • never resetting for a later producer;
  • treating a first batch as a continuation.

Green: 8 passed.

Gate:

• full telemetry-service suite with TEST_DATABASE_URL (postgres:16-alpine): 96 passed;
• ruff check .: passed;
• ruff format --check .: passed.

2. e881547 Apply each stored event batch incrementally in the Postgres store

Tests: in tests/test_storage_postgres.py:

• test_incremental_batch_apply_reads_a_constant_number_of_event_rows_per_in_order_batch
• test_incremental_batch_apply_reads_only_the_event_rows_after_a_late_event
• test_incremental_batch_apply_stores_the_fully_materialized_state. This one covers 42 scripted and seeded schedules. After every batch it checks that the stored row equals materialize_run over the stored events, which is what ingestion stored before. After each schedule it checks that list_runs matches.

Red, against the store before this change: 2 failed, for the intended reason; 1 passed.

• In order, the rows read per batch were 20, 40, …, 400, growing with the run.
• A late event read {60: 60, 600: 600} rows instead of 7.
• The equivalence test passed. It characterizes the current behavior and guards it through the change.

Green: 3 passed. Rows read per in-order batch are constant (one producer row). A late event reads 7 rows (one producer row, then the late event and the 5 stored after it), whether the run has 60 events or 600.

Gate:

• full suite: 99 passed;
• python -m telemetry_collector.migrate, then alembic check, against the disposable test database: "No new upgrade operations detected";
• ruff check .: passed;
• ruff format --check .: passed.

Query plans, on 20 runs × 2,000 events:

• The per-producer max(sequence) is an Index Only Scan Backward under a Limit: 1 row, 4 buffers.
• The late-event replay is an index range scan on telemetry_events_run_order_idx: 6 rows.

Findings, risks and deploy implications

Findings

• Problem confirmed: the full re-read and re-materialization was at storage.py lines 537–553 on main.
• Already consistent: every run that can still receive events was last written by the full re-materialization under the current rules. Every batch since the last rule change (Read UK staging contract v3: a real blocked run status, block details and failure classes #212) rebuilt its row. The incremental fold therefore continues from consistent state.
• Fake store unchanged: tests/fakes.py still materializes fully. It remains the reference for the HTTP contract tests.

Risks

• Materializer rule changes no longer self-heal. Before this change, each batch rebuilt a run under the current rules. A future change to _apply_event's rules will leave runs that are in progress during its deploy with columns computed under the old rules, unless those runs are re-materialized once. The fold also relies on each column staying a maximum, a minimum or an event-only write; _apply_event's docstring now says so.
• Pathological delivery is still quadratic. A producer stream delivered in reverse, one event per batch, would cost O(lateness) per batch. The emitter never delivers that way.
• One aggregate scan. The earliest-timestamp read (an earlier producer's first event arriving after a later producer's events) scans the active producer's index range for a one-row result, at most once per run.
• Test runtime. The new Postgres tests take 40–60 s locally against Docker on macOS. CI's Postgres service ran the whole suite in 8 s on a recent run, so expect noticeably less.

Deploy implications

• Production deploy on merge. Merging to main triggers .github/workflows/telemetry-service.yml (push to main, paths telemetry-service/**): validate, deploy staging, then deploy production. Each environment gets a maintenance revision, the migration job, a candidate revision and promotion; production also gets a Cloud SQL backup first.
• No migration. The migration job finds the database at head.
• No new configuration. No environment variables, secrets or configuration are added or read, and no workflow, Dockerfile or deploy script changes.
• Carry into the extracted repo. telemetry-service/ is being extracted to PolicyEngine/microcosm-collector. This change must be carried into that repository before cutover, so the extracted collector doesn't regress to full re-materialization.

Unverified

• Not exercised against staging or production data. Equivalence is shown on test schedules and on the disposable Postgres database.
• That production rows are consistent with the current rules is argued above, not checked against production.

🤖 Generated with Claude Code

[vercel]

The latest updates on your projects. Learn more about Vercel for GitHub.

Project	Deployment	Actions	Updated
calibration-diagnostics	Ready	Preview	Oct 15, 2026 8:06pm UTC