Isolate telemetry delivery failures per run

[MaxGhenis]

Stacked on #1168: the base branch is fix/telemetry-spool-lock-startup. I'll retarget to main once #1168 merges. This is the #1168 follow-up "Isolate delivery failures per run".

Problem

Every emitter service on a host delivers every pending run in the host-wide shared spool, oldest updated_at first. On #1168, CollectorDelivery.flush_once wraps the whole pass:

• An unexpected error from one run ended the pass at that run. It deferred delivery for every run (1 s, doubling to 60 s) and re-raised. The next pass reached the same run first again: a stuck run whose build has finished keeps its updated_at, while live builds' runs move to the back with each event. So runs after it were never delivered, by any service.
• A registration row that cannot be decoded failed pending_runs() itself. That query decodes every row's registration_json, so no run at all was delivered.
• Any failure delayed everyone. An ordinary one (a 503 or a timeout on one run's request) deferred every run.
• A 401 had no backoff. On a 401 to a registration or events POST, the code cleared the token and retried on the next tick. With one-second ticks, the Keep the telemetry emitter service alive through spool lock contention #1168 review measured 21 requests in 10 ticks for events and 20 for registration. A batch the collector accepted, whose acknowledge then hit a lock error, was re-POSTed on every pass: 12 requests in 10 ticks.

Persistent per-run failures I checked on this branch:

• A run id with a space. --staging-run-id is free text, TelemetryRun does not validate run_id, and urllib raises http.client.InvalidURL for an events URL whose run id contains a space. InvalidURL is an HTTPException, not an OSError, so on Keep the telemetry emitter service alive through spool lock contention #1168 every pass raised at that run.
• A malformed registration row. A registration_json that is not a JSON object makes JsonObjectText raise ValueError on read. Another checkout sharing the spool could write one.
• A collector that answers one run's events with 4xx or 5xx every time.

The task description also listed a collector response that makes int(response["expires_in"]) raise as a per-run state. It isn't one: the token exchange sends {} with the Hugging Face credential, and neither its request nor its response depends on the run. This PR treats it as shared (item 2 below).

Change

1. Each run keeps its own retry delay (RetryDelay: 1 s, doubling to 60 s, kept in memory per service).

   • A run fails on a transport error, a timeout, a 401, any unexpected exception while delivering it, or any other status except a success, a 403, or a 409 to its registration (those two still make it local-only). A failed run waits out its own delay, and the pass moves on to the next run.
   • A delivered batch clears the delay.
   • A run that leaves the queue (delivered, local-only or pruned) is forgotten, so the table never outgrows the queue.

2. The token exchange has one shared delay. Its failures (transport error, a status other than 200/401/403, a 200 without a usable token, or an exception parsing it) defer further exchanges and charge no run. Runs that need a token skip the pass and are attempted again next pass. A 401/403 from the exchange still makes that run local-only, as before.

3. One run's stored registration can fail only that run.

   • EventSpool.pending_run_keys() lists (run_id, producer_id) from the key columns only: same filter and order as pending_runs(), with ties broken by key.
   • EventSpool.registration(run_id, producer_id) reads one row. Delivery calls it only when the run needs registering, inside that run's error handling.
   • pending_runs() now shares the query builder, using EXISTS instead of JOIN … DISTINCT. A differential test checks that it lists the same runs.

4. A 401 counts as a failure of that run. The token is still cleared. The next run that needs a token exchanges once, so a refused run costs one exchange per attempt, on its own doubling schedule.

5. An accepted batch is never sent twice. If the acknowledge after a 202 hits lock contention, the event ids stay in memory. On the run's next attempt they are removed from the spool before the run sends anything else.

6. Reporting. flush_once no longer raises for one run's error or the exchange's. Each unexpected error type prints one line per scope through write_warning, without a traceback:

   • for a run: "…hit ValueError (…) delivering one queued run; it will retry that run later and keep delivering the others."
   • for the exchange: "…hit TypeError (…) exchanging the Hugging Face credential for a collector token; it will retry later."

   Lock contention still ends the pass and propagates, as on Keep the telemetry emitter service alive through spool lock contention #1168: the worker stays silent and retries on the next tick. So does an error listing the queue, which the worker reports once per type.

Request counts in 10 one-second ticks, from the #1168 review's probe (adapted to the new spool API; scripts and output in ~/reviews/microcosm-telemetry-per-run-delivery-2026-10-09/probe/):

Scenario	#1168 head	This branch
events answered 401	21	9
registration answered 401	20	8
202, then acknowledge hits a lock	12	3
exchange expires_in: null	4	4, now one warning line instead of an exception

Per-run retry state vs a single run quarantine

Single run quarantine: on an unexpected error, set the run aside, either in memory for the service's life or as local_only in the spool.

• It needs the least state, and the bad run then costs no requests.
• It cannot tell a transient error from a persistent one. One dropped connection or one 500 during a collector deploy would park a healthy run for good.
  • In memory, only some later build's service would deliver it, and a build's own service is often the last one alive on the host.
  • As local_only, its telemetry is lost for every service.
• It also treats every error as the run's own, though the token exchange is shared.

Per-run retry state:

• Transient failures recover by themselves.
• A persistent failure settles to one attempt a minute per service. That is at most two requests (registration and events) plus, after a 401, one exchange. A registration that cannot be decoded costs no request of its own.
• It needs no schema change, and it keeps Keep the telemetry emitter service alive through spool lock contention #1168's backoff schedule, scoped per run.

Decision: per-run retry state. A run that fails for 7 days has its events pruned by retention and leaves the queue.

Evaluated and not done

• Persisting retry state in the spool. That would need a migration. Keep the telemetry emitter service alive through spool lock contention #1168 refuses a spool stamped with a revision the checkout doesn't know (test_a_spool_from_an_unknown_migration_is_refused_without_the_write_lock), so older checkouts sharing the host's spool would be locked out. Each failure would also become a write on the contended spool. Instead, the state is per service: S services on a host each retry a failing run, so at most S attempts a minute.
• Making persistent 4xx (400/404/413/422) local-only. A collector misdeploy returning 4xx would then discard every run's telemetry permanently. Backoff costs at most one attempt a minute.
• Stopping the pass at the first transport failure. This would cap an outage pass at one 5 s timeout. But failing runs whose builds have finished sort first, so they could keep a live build's run from ever being attempted. As on Keep the telemetry emitter service alive through spool lock contention #1168, a pass tries each due run once.
• Rotating where the pass starts after lock contention. Contention still ends the pass for every run, as on Keep the telemetry emitter service alive through spool lock contention #1168. After Keep the telemetry emitter service alive through spool lock contention #1168's batched prune, it takes another process holding the write lock past a 5 s busy wait.

Known limits

• A 401 from one run delays others only when the exchange is also failing. A run whose requests are refused 401 clears the shared token. If the exchange is failing at that moment, the other runs wait for the exchange too. A 401 refuses the token, not the run, and the cost is bounded by that run's own delay.
• Lock contention ends the pass for every run (unchanged).

Invariants (each has a test in test_telemetry_delivery_isolation.py)

• Retry delay (Hypothesis). After a scope's k-th failure in a row, it waits exactly min(2^(k−1), 60) s: it is not due a moment earlier, and is due then.
• Simulated host (Hypothesis, 300 examples plus 3 pinned).
  • Inputs:
    • Runs: up to 6, each healthy or scripted per request (ok, transport error, timeout, 503, 400, 401, 403, 409, an unexpected exception), then a fixed outcome for every later request. A run may also have a corrupt registration.
    • Events: appended on random ticks. Batch size: 1, 2 or 100.
    • Token exchange: outcome by time (ok, transport error, 503, expires_in: null, no token, an exception, 403). Token lifetime: 60 s, 1 h, or none given.
    • Lock contention: listing or acknowledging, on random ticks.
  • Contention only: flush_once raises only lock contention.
  • No starvation:
    • On every pass that completes, every pending healthy run is attempted.
    • When contention ends a pass, every healthy run listed before the run it stopped at was attempted.
    • A healthy run given a token sends its batch on that pass.
  • Bounded requests:
    • At most one registration and one events request per run per pass.
    • After a run's k-th failure in a row, its next request comes at least min(2^(k−1), 60) s later. The same holds for the exchange, except after a 403.
    • A pass makes more than one exchange only after a 401 or a 403 in that pass.
    • No event is accepted twice.
  • Exact schedule (reference model of one delay per run): each run is attempted on exactly the passes its own delay allows. This catches an extra delay (a missing reset) as well as a missing one.
  • Reports:
    • One line per error type and scope, for exactly the types raised.
    • Every other line is the once-per-run rejection notice.
    • No traceback.
  • Verdicts: exactly the runs the collector refused for good are local-only: registration 403/409, events 403, or exchange 403 for that run.
• Counterfactual (Hypothesis, 200 examples). With a working exchange and contention only on listing, healthy runs deliver the same events on the same ticks with or without the other runs present.
• Concrete bounds.
  • A run that fails every request makes 66 requests in an hour, not 3,600.
  • A run refused 401 costs one exchange per attempt, on its own schedule, for both registration and events.
  • A working exchange resets the exchange delay.
• Real spool.
  • A corrupt registration fails only its run: no request, one ValueError line, retried on its own delay.
  • In a run of 130 ticks, a run answered 500 is retried at ticks 0, 1, 3, 7, 15, 31, 63 and 123, while a live run delivers on every tick.
  • A malformed token response charges no run.
  • A locked listing is retried on the next tick.
  • Contention after an accepted batch never resends it.
  • Delivery forgets runs that leave the queue.
• Listing (Hypothesis, differential). pending_run_keys() lists exactly the runs the old join-and-distinct query did, in update order. pending_runs() equals registration() of each key.

Mutation check

Each mutant is an exact edit to collector.py or spool.py, injected through a pytest plugin that swaps the module's code without touching tracked files. It ran against this file plus test_telemetry_emitter.py, which leaves out the subprocess test. All 26 were killed, each by a failing assertion, not a collection or fixture error. An identity mutant passed. Runner, plugin and logs: ~/reviews/microcosm-telemetry-per-run-delivery-2026-10-09/mutation/.

Mutant	Killed by (first failing test)	Reason
one_delay_for_all_runs	test_a_failing_run_never_holds_up_another_and_costs_bounded_requests (8 tests)	AssertionError: assert {('building', 'p')} <= {('corrupt', 'p')}
failure_not_deferred	test_a_failing_run_never_holds_up_another_and_costs_bounded_requests (5 tests)	AssertionError: (('buggy', 'p'), [1, 2, 3, 4])
delivery_does_not_reset	test_a_failing_run_never_holds_up_another_and_costs_bounded_requests (1 test)	AssertionError: (3, ('run-0', 'producer-0'))
run_error_ends_pass	test_a_failing_run_never_holds_up_another_and_costs_bounded_requests (3 tests)	AssertionError: assert (RuntimeError('collector bug') is None or False)
exchange_failure_not_deferred	test_a_failing_run_never_holds_up_another_and_costs_bounded_requests (2 tests)	AssertionError: assert (3 - 3) >= 1.0
exchange_error_charged_to_run	test_a_failing_run_never_holds_up_another_and_costs_bounded_requests (2 tests)	AssertionError: (2, ('buggy', 'p'))
exchange_delay_ignored	test_a_failing_run_never_holds_up_another_and_costs_bounded_requests (2 tests)	AssertionError: assert (0 - 0) >= 1.0
exchange_delay_not_reset	test_a_working_exchange_resets_the_exchange_delay (1 test)	AssertionError: assert [(0, 'transpo...), (33, 'ok')] == [(0, 'transpo...), (32, 'ok')]
events_401_keeps_token	test_a_rejected_token_from_one_run_costs_one_exchange_per_attempt[events] (1 test)	assert 1 == (1 + 6)
events_403_retried	test_a_failing_run_never_holds_up_another_and_costs_bounded_requests (1 test)	_check_runs_retry_exactly_when_due: (2, ('run-0', 'producer-0')); _check_verdicts: assert set() == {('run-0', 'producer-0')}
exchange_rejection_retried	test_a_failing_run_never_holds_up_another_and_costs_bounded_requests (1 test)	AssertionError: assert set() == {('run-0', 'producer-0')}
report_every_time	test_a_failing_run_never_holds_up_another_and_costs_bounded_requests (2 tests)	AssertionError: assert ['RuntimeErro...RuntimeError'] == ['RuntimeError']
report_once_across_scopes	test_a_failing_run_never_holds_up_another_and_costs_bounded_requests (1 test)	AssertionError: assert [] == ['RuntimeError']
lock_error_charged_to_run	test_a_failing_run_never_holds_up_another_and_costs_bounded_requests (2 tests)	AssertionError: assert ['OperationalError'] == []
never_forget_runs	test_delivery_forgets_runs_that_leave_the_queue (1 test)	AssertionError: assert {('run-a', 'p... seconds=2.0)} == {}
due_strictly_after	test_a_persistently_failing_run_settles_to_one_request_a_minute (11 tests)	assert [1, 1, 3, 6, 11, 20, ...] == [0, 0, 1, 3, 7, 15, ...]
double_before_waiting	test_retry_delay_doubles_to_its_cap_from_each_failure (8 tests)	assert 2.0 == (0.0 + 1.0)
no_delay_cap	test_retry_delay_doubles_to_its_cap_from_each_failure (4 tests)	assert 64.0 == (0.0 + 60.0)
registration_read_before_token	test_a_failing_run_never_holds_up_another_and_costs_bounded_requests (1 test)	AssertionError: (2, ('corrupt', 'p'))
waiting_for_token_charges_run	test_a_failing_run_never_holds_up_another_and_costs_bounded_requests (1 test)	AssertionError: (2, ('buggy', 'p'))
register_401_keeps_token	test_a_rejected_token_from_one_run_costs_one_exchange_per_attempt[registration] (1 test)	assert 1 == (1 + 6)
accepted_batch_sent_again	test_a_failing_run_never_holds_up_another_and_costs_bounded_requests (2 tests)	AssertionError: assert 2 == 1
removed_batch_not_counted_delivered	test_a_failing_run_never_holds_up_another_and_costs_bounded_requests (1 test)	AssertionError: (4, ('flaky', 'p'))
listing_includes_runs_without_events	test_a_corrupt_registration_fails_only_its_own_run (4 tests)	AssertionError: assert [('corrupt', ...'producer-a')] == [('corrupt', 'producer-a')]
listing_newest_first	test_a_corrupt_registration_fails_only_its_own_run (3 tests)	AssertionError: assert ('healthy-b', 'producer-a') == ('corrupt', 'producer-a')
listing_ignores_state	test_delivery_forgets_runs_that_leave_the_queue (5 tests)	AssertionError: assert {('run-a', 'p... seconds=2.0)} == {}

Tests run (touched files only, -p no:cacheprovider --basetemp=<scratch>)

• Merged tree 845f5e02f (this branch with Keep the telemetry emitter service alive through spool lock contention #1168's head d7fda0def): test_telemetry_delivery_isolation.py, test_telemetry_spool_contention.py and test_telemetry_emitter.py. All 80 passed (15 + 32 + 33), including test_subprocess_exchanges_ambient_token_and_delivers_events.
• Head 5bacdba3a (one more test-only commit): test_telemetry_delivery_isolation.py, 15 passed.
• Lint: ruff check . passes. ruff format --check passes on the touched files.

Follow-ups (not in this PR)

• Validate or encode telemetry run ids before they reach the events URL. A space in --staging-run-id now fails only that run, but that run never delivers.

axiom: n/a: telemetry infrastructure, no policy encoding

🤖 Generated with Claude Code