Keep the telemetry emitter service alive through spool lock contention

README.md

@@ -49,8 +49,12 @@ 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. 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.

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/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,22 @@
 
 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
 
 DEFAULT_HEARTBEAT_SECONDS: Final = 60.0
 DEFAULT_DRAIN_SECONDS: Final = 15.0
@@ -73,3 +89,25 @@
 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."
+)
+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