Let checkouts at different versions share the telemetry spool

README.md

@@ -49,8 +49,19 @@ 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. Checkouts at different
+versions share the spool too. Its schema migrations only add to it, and each
+records the revisions it builds on, so an older checkout keeps using a spool
+that a newer one migrated, and a service delivers the events queued by any
+version. Two kinds of checkout are refused instead, in one line, and build
+without telemetry: one on a branch whose spool migration differs from the one
+applied, and one from before October 2026, which cannot read that record, once
+the spool has been migrated past the revision it knows. 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-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.

changelog.d/telemetry-spool-shared-versions.fixed.md

@@ -0,0 +1,5 @@
+Checkouts at different microcosm versions on one host keep sharing the local telemetry spool (`~/.cache/microcosm/telemetry/events.sqlite3`). Before, once a newer checkout applied a spool migration, every older checkout's emitter service refused the spool, and its builds ran without hosted telemetry until the checkout was updated.
+
+Spool migrations after the first must now be additive: to a table that already exists they may only append ordinary nullable or defaulted columns and add plain non-unique indexes, never rebuild it, change its rows or give it a trigger. A test holds every packaged revision to that. Each migration also records the spool's lineage, the migrating checkout's history. An older checkout whose head is in that lineage uses the spool as it is, without migrating it or taking the write lock, and services of every version deliver each other's queued events. A checkout whose history diverges from the spool's, such as a branch with a different migration, is still refused in one line. The revision is now read again once the migration lock is held, so an opener that waited while a newer checkout migrated no longer fails with "Can't locate revision".
+
+This covers checkouts from this change on. One from before it still refuses a spool migrated past `20261007_01`, and if it migrates a spool itself it records no lineage, which the next checkout at that revision then records. So the next spool migration should land after this change has reached build hosts, on a branch that includes it.

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(

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/alembic/env.py

@@ -3,13 +3,20 @@
 from alembic import context
 from sqlalchemy.engine import Connection
 
+from microcosm.build.telemetry_emitter_service.constants import SPOOL_LINEAGE_TABLE
 from microcosm.build.telemetry_emitter_service.models import SpoolModel
 
 _MISSING_CONNECTION_ERROR = (
     "telemetry spool migrations require a programmatically supplied connection"
 )
 
 
+def _include_name(name: str | None, type_: str, parent_names: object) -> bool:
+    # The migration runner keeps the lineage table, as Alembic keeps
+    # alembic_version, so the models do not describe it.
+    return not (type_ == "table" and name == SPOOL_LINEAGE_TABLE)
+
+
 def run_migrations() -> None:
     """Run migrations on the connection supplied by the emitter service."""
 
@@ -20,6 +27,7 @@ def run_migrations() -> None:
         connection=connection,
         target_metadata=SpoolModel.metadata,
         render_as_batch=True,
+        include_name=_include_name,
     )
     with context.begin_transaction():
         context.run_migrations()

packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/alembic/script.py.mako

@@ -20,7 +20,7 @@ depends_on: str | Sequence[str] | None = ${repr(depends_on)}
 
 
 def upgrade() -> None:
-    """Apply this schema revision."""
+    """Apply this schema revision, additively: see ``migrations.py``."""
 
     ${upgrades if upgrades else "pass"}
 

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

@@ -58,7 +58,26 @@
 MAX_RETRY_SECONDS: Final = 60.0
 
 DATABASE_TIMEOUT_SECONDS: Final = 5
+# Kept by the migration runner beside alembic_version: every revision in the
+# history of the checkout that last migrated the spool.
+SPOOL_LINEAGE_TABLE: Final = "spool_lineage"
 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
 
 DEFAULT_HEARTBEAT_SECONDS: Final = 60.0
 DEFAULT_DRAIN_SECONDS: Final = 15.0
@@ -73,3 +92,26 @@
 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}, and the spool does not record "
+    "it as descending from {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."
+)
+MAX_WARNING_ERROR_CHARS: Final = 300

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

@@ -0,0 +1,75 @@
+"""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_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})
+
+
+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,
+    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. 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
+    ``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.
+    """
+
+    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 >= deadline:
+                raise
+            contended = error
+        sleep(wait)
+        if clock() >= deadline:
+            # The sleep overran, as it can on a loaded host.
+            raise contended
+        delay = min(SPOOL_RETRY_MAX_SECONDS, delay * 2)

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

@@ -2,10 +2,11 @@
 
 from __future__ import annotations
 
+from collections.abc import Callable
 from pathlib import Path
 
-from sqlalchemy import URL, create_engine
-from sqlalchemy.engine import Engine
+from sqlalchemy import URL, create_engine, event
+from sqlalchemy.engine import Connection, Engine
 from sqlalchemy.orm import Session, sessionmaker
 
 from microcosm.build.telemetry_emitter_service.constants import (
@@ -19,13 +20,49 @@ def sqlite_database_url(path: Path | str) -> URL:
     return URL.create("sqlite+pysqlite", database=str(Path(path)))
 
 
-def create_spool_engine(path: Path | str) -> Engine:
-    """Create the SQLAlchemy engine used by one emitter service process."""
+def create_spool_engine(
+    path: Path | str,
+    *,
+    immediate_transactions: bool = False,
+    busy_timeout_seconds: Callable[[], float] | None = None,
+) -> Engine:
+    """Create the SQLAlchemy engine used by one emitter service process.
 
-    return create_engine(
+    ``busy_timeout_seconds`` is read each time a connection is checked out and
+    caps how long one statement waits in SQLite's busy handler for another
+    process's lock; by default that is ``DATABASE_TIMEOUT_SECONDS``.
+
+    By default Python's sqlite3 module opens a transaction only before a data
+    change, never before DDL. With ``immediate_transactions`` every SQLAlchemy
+    transaction is instead one ``BEGIN IMMEDIATE`` transaction: it takes the
+    write lock at the start, waiting in SQLite's busy handler, and holds it
+    until commit. This is SQLAlchemy's documented pysqlite recipe for
+    transactional DDL.
+    """
+
+    engine = create_engine(
         sqlite_database_url(path),
         connect_args={"timeout": DATABASE_TIMEOUT_SECONDS},
     )
+    if busy_timeout_seconds is not None:
+
+        def apply_busy_timeout(dbapi_connection, connection_record, proxy) -> None:
+            milliseconds = max(0, round(busy_timeout_seconds() * 1000))
+            dbapi_connection.execute(f"PRAGMA busy_timeout = {milliseconds}")
+
+        event.listen(engine, "checkout", apply_busy_timeout)
+    if immediate_transactions:
+        event.listen(engine, "connect", _disable_driver_transactions)
+        event.listen(engine, "begin", _begin_immediate)
+    return engine
+
+
+def _disable_driver_transactions(dbapi_connection, connection_record) -> None:
+    dbapi_connection.isolation_level = None
+
+
+def _begin_immediate(connection: Connection) -> None:
+    connection.exec_driver_sql("BEGIN IMMEDIATE")
 
 
 def create_spool_session_factory(engine: Engine) -> sessionmaker[Session]:

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

@@ -0,0 +1,38 @@
+"""One-line warnings on the build's stderr, which the service inherits."""
+
+from __future__ import annotations
+
+import sys
+
+from sqlalchemy.exc import DBAPIError
+
+from microcosm.build.telemetry_emitter_service.constants import (
+    MAX_WARNING_ERROR_CHARS,
+)
+
+
+def describe_error(error: BaseException) -> str:
+    """Return one short line describing an error, for a warning on stderr.
+
+    SQLAlchemy's message repeats the SQL statement and its parameters, which
+    can hold a registration or event payload, and a documentation link on later
+    lines, so a wrapped driver error is described by the driver's own message.
+    """
+
+    if isinstance(error, DBAPIError) and error.orig is not None:
+        error = error.orig
+    lines = str(error).strip().splitlines()
+    return (lines[0] if lines else type(error).__name__)[:MAX_WARNING_ERROR_CHARS]
+
+
+def write_warning(message: str) -> None:
+    """Print one line to stderr without ever raising.
+
+    The service runs in its own session, so it can outlive a build whose
+    stderr was a pipe that has since closed; a failed write must not stop it.
+    """
+
+    try:
+        print(message, file=sys.stderr, flush=True)
+    except (OSError, ValueError):
+        pass