[anth-volk]

Fixes #217

Merge with "Create a merge commit", not squash or rebase. PolicyEngine/microcosm-collector#11 carries this branch's head (3dfbe05) into microcosm-collector, and its record (tools/extraction-source.env) names the extraction of that commit (ae1dc45). Every later carry-over runs tools/extract-from-calibration-diagnostics.sh on this repo's main and exits with an error unless that recorded extraction is in the new extraction's history. That holds only if main contains this branch's own commits. A squash or rebase merge puts their changes on main under new commits instead, so the next extraction would not contain ae1dc45, and every later carry-over to the collector would be refused. For the same reason, adding a commit to this branch after 3dfbe05 means updating microcosm-collector#11 first.

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, so a batch's cost no longer grows with the run's stored events. It grows with the batch and with the number of producers registered for the run: every nonempty batch reads each registered producer's highest stored sequence (one row and one index lookup per producer; a producer is one build attempt, so a run that never restarted has one). Stored events are read only in the cases listed below. 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 folds nothing, but it still takes the run's row lock, reads every registered producer's highest sequence and runs the insert.

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, because the test's run has one producer. 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.

3. bbe4ea5 Merge main

A mechanical merge, so no red step. None of main's 12 new commits touch telemetry-service/.

4. 15bb5b3 Test incremental batch application with non-UTC timestamps

A follow-up from code review: every earlier test sent UTC timestamps, so nothing checked the claim that the fold sees the same database values as a full re-read. The submitted value and the row returned by the insert differ only in their offset. work.updated_at is stored as text, so a fold over the submitted values would store the producer's offset where a full re-read gives the database's.

Test: test_incremental_batch_apply_stores_database_values_for_non_utc_timestamps in tests/test_storage_postgres.py. It sends each schedule twice, once in UTC and once in alternating +14:00 and −12:00 offsets. With those offsets, every second pair of events has wall-clock times in the opposite order to their instants. The schedules:

• in-order batches of every event kind;
• a late event, which replays the events stored after it;
• an earlier attempt that first reports after a later attempt (the earliest-timestamp read).

After every batch the stored row must equal a full re-materialization. Each non-UTC run must also equal its UTC run.

Red: 0 failed against this branch, as expected. The test guards behaviour commit 2 already implements: the fold uses the rows the insert returns. To check that the test can fail, I ran it against a store that folds the submitted timestamps instead. It failed at the second batch, on the work column. The 7 existing Postgres tests passed against that store.

Green: 1 passed.

5. 3dfbe05 Document that the producer read must follow the run row lock

The second review follow-up, and comments only, so no red step. apply_event_batch is exact only if each producer's highest stored sequence, the run's stored columns and the targeted event reads all describe every event stored before the batch. append_events gets that from the run's FOR UPDATE lock, which producer registration also takes. A comment at the producer read and a paragraph in apply_event_batch's docstring now say so. If the read moved before the lock, a concurrent batch could commit in between, and this batch's earlier sequences would be folded on top of its events instead of replayed as late events.

After commits 4 and 5:

• tests/test_storage_postgres.py and tests/test_incremental_batch_apply.py with TEST_DATABASE_URL (postgres:16-alpine): 16 passed;
• ruff check and ruff format --check on the four changed files: passed.

Locally I ran only the test files this PR adds or changes. CI's telemetry-service.yml validation job runs the whole telemetry-service suite, the migration and alembic check.

Review follow-up (2026-10-20): description only, no commits

A review found that this description overstated the cost bound. It said a batch's cost was proportional to the batch, and that a duplicate re-send did no work. But _stored_producers loads every registered producer and computes each one's highest stored sequence for every nonempty batch, so the cost also grows with the number of producers. The Summary, "Normal delivery reads no stored events", commit 2's green result and the Risks now say so, and the merge-method note at the top is new.

No code change. The fold needs every producer's highest sequence only to find the active producer (the latest-registered one with stored events) and to tell whether it is the only one with events. Reading less would need a different producer query and a different input to apply_event_batch, which the equivalence and cost tests would have to cover again, in a change that deploys straight to production. It would also add a commit after 3dfbe05, which microcosm-collector#11 would then have to carry over again. With one producer per build attempt, that is not a cheap, safe fix, so the head stays 3dfbe05.

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.
• Cost grows with the producer count. _stored_producers loads every producer registered for the run and computes each one's highest stored sequence, for every nonempty batch, including a batch of duplicates. That is one indexed lookup per producer (an Index Only Scan Backward under a Limit, per the query plans above), so a run restarted many times pays for every earlier attempt on each batch. I did not measure how many producers production runs have.
• 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.
• Concurrent batches rely on the run's row lock. Batches for one run are applied one at a time because each takes the run's FOR UPDATE lock before reading its producers. The code now says the read must stay after the lock (commit 5).
• 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, until calibration-diagnostics#233 removes the push trigger. 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. Carry over calibration-diagnostics#219: apply each event batch incrementally microcosm-collector#11 carries this branch's head 3dfbe05 over, so the extracted collector doesn't regress to full re-materialization. Merge this PR with a merge commit (see the note at the top).

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 19, 2026 8:18pm UTC