Decouple the telemetry emitter's event path from spool writes

README.md

@@ -49,8 +49,21 @@ until you restart the kernel, load it by path with
 Supported US and UK build commands always start the local telemetry emitter
 service. It reports live progress to the hosted collector when the operator's
 existing Hugging Face login is accepted; otherwise it retains the events
-locally and the build continues. This live event path is independent of the
-staging run files described below.
+locally and the build continues. Every build on a host shares one local
+event spool; a service that finds it locked by another process waits for it
+until shortly before the build stops waiting. One that still cannot register
+prints one warning line, without a traceback, and the build adds its own
+warning and runs without telemetry, hosted or local. Once it is running, the
+service replies to each event as soon as the event is queued in its memory,
+and a separate writer thread appends queued events to the spool in the order
+they arrived, retrying for as long as another process holds the spool's lock,
+so a busy spool never delays the build. The queue holds at most 10,000 events
+or 16 MiB, keeping room for the run's started, completed, failed and blocked
+events; when it is full the service refuses further updates and prints one
+warning line. When the build closes or dies, the service spends up to 15 s
+writing what is still queued and reports in one line anything it could not
+write. This live event path is independent of the staging run files described
+below.
 
 US fiscal refresh builds also write pre-release staging run files **by
 default**. Progress JSON is uploaded to `policyengine/populace-us-staging`

changelog.d/telemetry-event-queue.fixed.md

@@ -0,0 +1,3 @@
+A build's telemetry sends no longer wait for the shared local spool. Before, the emitter service appended each event to the SQLite spool before replying. When another build held the spool's write lock, a send waited up to SQLite's 5 s busy wait while the build's client gave up after 0.2 s and printed "could not queue an update", and an append still locked after 5 s dropped the event. Now the service replies once the event is queued in memory, and one writer thread appends queued events in arrival order, up to 100 per transaction, retrying lock errors for as long as they last. Heartbeats and the record of a killed build go through the same queue, so the killed-build record follows every event the build sent.
+
+The queue holds at most 10,000 events and 16 MiB of their encoded JSON, with 16 events and 256 KiB kept for run events (started, completed, failed, blocked), so a queue that progress updates have filled still records how the build ended. A full queue refuses updates and prints one warning line. One 15 s shutdown window opens when the build closes or dies. The writer drains the queue inside it and starts no append in its last 0.25 s. The service does not exit while the writer may still be writing, and it reports in one line how many events it could not write.

changelog.d/telemetry-spool-lock-startup.fixed.md

@@ -0,0 +1,5 @@
+A build on a busy host no longer loses its hosted telemetry when another build holds the shared local spool (`~/.cache/microcosm/telemetry/events.sqlite3`) while its emitter service starts. Before, the service's registration gave up after SQLite's 5 s busy wait, the service exited, and a SQLAlchemy traceback landed on the build's stderr. Now the build passes the time it will stop waiting (`--ready-deadline`). The service waits for the lock until 0.5 s before then: each SQLite wait ends by that point and lasts at most 5 s, a failed attempt is retried with jittered backoff, and the whole wait is capped at 60 s. Without a deadline it makes one 5 s attempt, as before. If the spool is still locked, it prints one warning line and exits with status 75 before the build gives up. Any other startup failure is also one line: status 2 for invalid arguments, 1 otherwise. SIGINT ends the service quietly, as SIGTERM does.
+
+Opening a spool already at the current schema takes no write lock. Retention pruning moved from spool open and every append to the delivery worker. It deletes in batches of 500 rows, each in its own short transaction, pauses 25 ms between batches so other builds' writers get the lock, and works at most 1 s per worker tick. A prune with nothing to remove only reads. Migrations run in one `BEGIN IMMEDIATE` transaction. Concurrent first opens of a new spool no longer fail with "table already exists"; about half of eight simultaneous opens did before. An interrupted migration leaves no partial schema, and a spool stamped by a migration this checkout does not know is refused without taking the lock.
+
+The delivery worker skips a failed step for one tick instead of dying. Heartbeats, delivery, and the check that records a killed build keep running. Lock contention is retried silently, and any other error prints one line per error type. An unexpected delivery error backs off like a collector failure. Warnings never raise on a closed stderr, and the service stops serving if its worker ever dies. A build that closes and exits while a worker step is still running is no longer recorded as an unexpected exit after its completion.

packages/microcosm-build/src/microcosm/build/telemetry_emitter.py

@@ -196,6 +196,11 @@ def start(
                 str(os.getpid()),
                 "--heartbeat-seconds",
                 str(heartbeat_seconds),
+                # The service retries spool lock contention until just before
+                # this wall-clock time, when the wait below gives up. One token,
+                # so a past or negative deadline is not read as an option.
+                "--ready-deadline="
+                + repr(datetime.now(UTC).timestamp() + startup_timeout_seconds),
             ]
             if development_collector_url is not None:
                 command.extend(
@@ -261,7 +266,12 @@ def emit(
         message: str | None = None,
         details: Mapping[str, Any] | None = None,
     ) -> None:
-        """Queue one event locally, never raising into the build."""
+        """Queue one event with the local service, never raising into the build.
+
+        The service acknowledges the event once it is queued in the service's
+        memory, behind every event acknowledged before it, without waiting for
+        the shared spool; see ``_send``.
+        """
 
         if not self.available:
             return
@@ -464,6 +474,17 @@ def _close_transition_stage(
         self.stage(stage_id, status=status, message=message, **details)
 
     def _send(self, payload: Mapping[str, Any]) -> None:
+        """Send one message and wait at most the send timeout for its reply.
+
+        ``ok`` to an event means the service has queued it in memory behind
+        the events acknowledged before it; a writer thread then appends it to
+        the spool in that order, retrying while another build holds the
+        spool's lock. The reply never waits for the spool, so a locked spool
+        does not delay the build. ``error``, or no reply within the timeout,
+        means the event may not be queued: the first such failure prints one
+        warning, and the build continues either way.
+        """
+
         if self._socket_path is None:
             return
         try:

packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/__main__.py

@@ -1,5 +1,11 @@
 """Module execution entry point for the telemetry emitter service."""
 
+import signal
+
 from microcosm.build.telemetry_emitter_service.main import main
 
+# The service runs in its own session, so an interrupt reaches it only when
+# sent on purpose. End at once, as SIGTERM does, rather than print a
+# KeyboardInterrupt traceback on the build's stderr.
+signal.signal(signal.SIGINT, signal.SIG_DFL)
 raise SystemExit(main())

packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/collector.py

@@ -4,7 +4,6 @@
 
 import json
 import os
-import sys
 import time
 import urllib.error
 import urllib.request
@@ -42,6 +41,10 @@
     TOKEN_EXCHANGE_PATH,
     TOKEN_REFRESH_MARGIN_SECONDS,
 )
+from microcosm.build.telemetry_emitter_service.contention import (
+    is_transient_spool_error,
+)
+from microcosm.build.telemetry_emitter_service.diagnostics import write_warning
 from microcosm.build.telemetry_emitter_service.spool import EventSpool
 
 
@@ -168,6 +171,17 @@ def flush_once(self) -> bool:
 
         if time.monotonic() < self._next_attempt_at:
             return False
+        try:
+            return self._flush_pending_runs()
+        except Exception as error:
+            # The worker runs this every second and survives the error, so an
+            # unexpected one backs off like a collector failure rather than
+            # repeating requests every tick. A busy spool is retried next tick.
+            if not is_transient_spool_error(error):
+                self._defer_retry()
+            raise
+
+    def _flush_pending_runs(self) -> bool:
         made_progress = False
         for registration in self.spool.pending_runs():
             run_id = str(registration["run_id"])
@@ -220,7 +234,7 @@ def _collector_token(self, registration: Mapping[str, Any]) -> str | None:
         hf_token = _huggingface_token()
         if not hf_token:
             if not self._warned_no_token:
-                print(NO_CREDENTIAL_MESSAGE, file=sys.stderr, flush=True)
+                write_warning(NO_CREDENTIAL_MESSAGE)
                 self._warned_no_token = True
             self._make_local_only(registration, LOCAL_ONLY_MISSING_CREDENTIAL)
             return None
@@ -293,5 +307,5 @@ def _defer_retry(self) -> None:
     def _warn_denied(self, registration_key: str, message: str) -> None:
         if registration_key in self._warned_denied:
             return
-        print(message, file=sys.stderr, flush=True)
+        write_warning(message)
         self._warned_denied.add(registration_key)

packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/constants.py

@@ -59,6 +59,42 @@
 
 DATABASE_TIMEOUT_SECONDS: Final = 5
 PRUNE_INTERVAL_SECONDS: Final = 60.0
+PRUNE_BATCH_ROWS: Final = 500
+# SQLite's busy handler polls rather than queueing, so another process rarely
+# gets the lock between two back-to-back transactions. Prune pauses between
+# batches and works at most PRUNE_STEP_SECONDS per worker tick.
+PRUNE_BATCH_PAUSE_SECONDS: Final = 0.025
+PRUNE_STEP_SECONDS: Final = 1.0
+# Every concurrent build on a host shares one spool, so a write can find it
+# locked for longer than SQLite's own busy wait. Startup waits for the lock,
+# and retries lock errors with jittered, doubling pauses, until the margin
+# before the build stops waiting for readiness; every SQLite wait in that time
+# also ends by then. The margin covers binding the socket and the build's ping.
+SPOOL_RETRY_INITIAL_SECONDS: Final = 0.05
+SPOOL_RETRY_MAX_SECONDS: Final = 0.25
+READY_DEADLINE_MARGIN_SECONDS: Final = 0.5
+STARTUP_RETRY_LIMIT_SECONDS: Final = 60.0
+SPOOL_LOCKED_EXIT_STATUS: Final = 75  # EX_TEMPFAIL from sysexits.h
+
+# The socket hands each accepted event to a writer thread through an ordered
+# in-memory queue, so a spool that another process keeps locked never delays
+# the build's sends. The queue holds at most QUEUE_MAX_EVENTS events and
+# QUEUE_MAX_BYTES of their encoded JSON. The last QUEUE_RESERVED_EVENTS and
+# QUEUE_RESERVED_BYTES of that room take only run events (started, completed,
+# failed, blocked), so a queue that progress updates have filled still records
+# how the build ended.
+QUEUE_MAX_EVENTS: Final = 10_000
+QUEUE_MAX_BYTES: Final = 16 * 1024 * 1024
+QUEUE_RESERVED_EVENTS: Final = 16
+QUEUE_RESERVED_BYTES: Final = 256 * 1024
+# The writer appends up to WRITER_BATCH_EVENTS queued events per transaction.
+# One append's statements together wait at most WRITER_BUSY_TIMEOUT_SECONDS
+# for other processes' locks, and the append waits no longer than that for
+# another thread of this process to release the spool. So the writer's retry
+# loop decides how long to keep trying, and during a shutdown drain no append
+# touches the database after the deadline less that wait.
+WRITER_BATCH_EVENTS: Final = 100
+WRITER_BUSY_TIMEOUT_SECONDS: Final = 0.25
 
 DEFAULT_HEARTBEAT_SECONDS: Final = 60.0
 DEFAULT_DRAIN_SECONDS: Final = 15.0
@@ -70,6 +106,51 @@
 DRAIN_RETRY_SECONDS: Final = 0.5
 
 EVENT_OBJECT_ERROR: Final = "event must be an object"
+EVENT_FIELDS_ERROR: Final = "event must have an event_type and a status"
+QUEUE_FULL_ERROR: Final = "local telemetry queue is full"
+QUEUE_CLOSED_ERROR: Final = "local telemetry queue is closed"
+SPOOL_BUSY_ERROR: Final = "the spool is busy in this process"
 UNSUPPORTED_ACTION_ERROR: Final = "unsupported local telemetry action"
 LOCAL_MESSAGE_TOO_LARGE_ERROR: Final = "local telemetry message exceeds 1 MiB"
 FAILURE_CLASS_UNEXPECTED_PROCESS_EXIT: Final = "unexpected_process_exit"
+
+READY_DEADLINE_ERROR: Final = "ready deadline must be a number of seconds"
+UNKNOWN_SPOOL_REVISION_ERROR: Final = (
+    "telemetry spool schema revision {revision!r} is not in this microcosm's "
+    "migration history, whose head is {head!r}"
+)
+SPOOL_LOCKED_WARNING: Final = (
+    "warning: the local telemetry emitter service could not register this build "
+    "in its spool {spool}: another process kept the spool locked for "
+    "{waited_seconds:.1f} s ({error})."
+)
+SERVICE_ARGUMENTS_WARNING: Final = (
+    "warning: the local telemetry emitter service could not start: {error}"
+)
+SERVICE_FAILED_WARNING: Final = (
+    "warning: the local telemetry emitter service stopped: {error_type}: {error}"
+)
+WORKER_STEP_WARNING: Final = (
+    "warning: the local telemetry emitter service's delivery worker hit "
+    "{error_type} ({error}) and will keep running."
+)
+QUEUE_FULL_WARNING: Final = (
+    "warning: the local telemetry emitter service's queue is full ({events} "
+    "updates, {megabytes:.1f} MiB, waiting for its spool); it will refuse "
+    "updates until the spool takes them."
+)
+UNWRITTEN_EVENTS_WARNING: Final = (
+    "warning: the local telemetry emitter service could not write {count} "
+    "queued update(s) to its spool before its {seconds:g} s shutdown drain "
+    "ended ({reason})."
+)
+DRAIN_TIME_REASON: Final = "the drain ran out of time"
+EVENT_DROPPED_WARNING: Final = (
+    "warning: the local telemetry emitter service's spool refused an update, "
+    "which was dropped: {error_type} ({error})."
+)
+WRITER_STOPPED_WARNING: Final = (
+    "warning: the local telemetry emitter service's spool writer stopped: "
+    "{error_type}: {error}"
+)
+MAX_WARNING_ERROR_CHARS: Final = 300

packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/contention.py

@@ -0,0 +1,95 @@
+"""Bounded retry for lock contention on the shared telemetry spool."""
+
+from __future__ import annotations
+
+import random
+import sqlite3
+import time
+from collections.abc import Callable
+
+from sqlalchemy.exc import DBAPIError
+
+from microcosm.build.telemetry_emitter_service.constants import (
+    SPOOL_BUSY_ERROR,
+    SPOOL_RETRY_INITIAL_SECONDS,
+    SPOOL_RETRY_MAX_SECONDS,
+)
+
+# SQLite reports extended result codes; the low byte is the primary code.
+_PRIMARY_RESULT_CODE_MASK = 0xFF
+_TRANSIENT_RESULT_CODES = frozenset({sqlite3.SQLITE_BUSY, sqlite3.SQLITE_LOCKED})
+
+
+class SpoolBusyError(sqlite3.OperationalError):
+    """Another thread of this process kept the spool past the caller's wait.
+
+    It carries SQLite's busy code, so it is retried like a lock another
+    process holds.
+    """
+
+    def __init__(self) -> None:
+        super().__init__(SPOOL_BUSY_ERROR)
+        self.sqlite_errorcode = sqlite3.SQLITE_BUSY
+        self.sqlite_errorname = "SQLITE_BUSY"
+
+
+def is_transient_spool_error(error: BaseException) -> bool:
+    """Return whether an error is SQLite lock contention that a retry can clear.
+
+    That is SQLITE_BUSY ("database is locked") or SQLITE_LOCKED, raised
+    directly by ``sqlite3`` or wrapped by SQLAlchemy. Schema, constraint and
+    I/O errors are not transient.
+    """
+
+    if isinstance(error, DBAPIError):
+        error = error.orig
+    if not isinstance(error, sqlite3.Error):
+        return False
+    code = getattr(error, "sqlite_errorcode", None)
+    return (
+        isinstance(code, int)
+        and code & _PRIMARY_RESULT_CODE_MASK in _TRANSIENT_RESULT_CODES
+    )
+
+
+def retry_spool_contention[T](
+    operation: Callable[[], T],
+    *,
+    deadline: float | Callable[[], float],
+    clock: Callable[[], float] = time.monotonic,
+    sleep: Callable[[float], None] = time.sleep,
+    jitter: Callable[[float, float], float] = random.uniform,
+) -> T:
+    """Run ``operation``, retrying spool lock contention until ``deadline``.
+
+    ``deadline`` is a ``clock`` reading, or a function returning one. A
+    function is read again after every lock error and after every wait, so its
+    caller can bring the deadline forward, from infinity say, while this
+    waits. The first attempt always runs. After a transient lock error the
+    next attempt waits a jittered backoff that starts at
+    ``SPOOL_RETRY_INITIAL_SECONDS`` and doubles up to
+    ``SPOOL_RETRY_MAX_SECONDS``. When that attempt would not start before the
+    deadline, or the wait overran it, the last lock error is raised instead.
+    Any other error is raised at once. Each attempt may itself wait in
+    SQLite's busy handler, so this bounds when attempts start, not when the
+    last one ends.
+    """
+
+    current_deadline = deadline if callable(deadline) else lambda: deadline
+    delay = SPOOL_RETRY_INITIAL_SECONDS
+    while True:
+        try:
+            return operation()
+        except Exception as error:
+            if not is_transient_spool_error(error):
+                raise
+            wait = jitter(delay / 2, delay)
+            if clock() + wait >= current_deadline():
+                raise
+            contended = error
+        sleep(wait)
+        if clock() >= current_deadline():
+            # The sleep overran, as it can on a loaded host, or the deadline
+            # moved forward during it.
+            raise contended
+        delay = min(SPOOL_RETRY_MAX_SECONDS, delay * 2)