diff --git a/src/reconciler/scheduler.py b/src/reconciler/scheduler.py index e171c4fa..30f0199b 100644 --- a/src/reconciler/scheduler.py +++ b/src/reconciler/scheduler.py @@ -21,6 +21,7 @@ from src import models from src.config import settings from src.dependencies import tracked_db from src.models import QueueItem +from src.reconciler.sync_vectors import record_pending_embeddings_backlog logger = logging.getLogger(__name__) @@ -154,6 +155,13 @@ class ReconcilerScheduler: while not self._shutdown_event.is_set(): now = datetime.now(timezone.utc) + # Refresh the pending-embeddings backlog gauge on EVERY replica, + # not just whichever one wins the sync_vectors work unit — the + # count is DB-global, so a replica that never ran a cycle would + # otherwise export a stale (or zero-initialized) value forever. + # See record_pending_embeddings_backlog for the full rationale. + await record_pending_embeddings_backlog() + # Check each task and enqueue if due for task_name, task in RECONCILER_TASKS.items(): next_run = self._next_run.get(task_name, now) diff --git a/src/reconciler/sync_vectors.py b/src/reconciler/sync_vectors.py index 2166c9ab..aaf4d8f4 100644 --- a/src/reconciler/sync_vectors.py +++ b/src/reconciler/sync_vectors.py @@ -701,13 +701,21 @@ async def _cleanup_pgvector_batch( return True -async def _record_pending_embeddings_backlog() -> None: +async def record_pending_embeddings_backlog() -> None: """Set the pending-embeddings backlog gauge to the current count of MessageEmbedding rows awaiting a vector (sync_state='pending'). - Called at the end of each reconciliation cycle so the gauge reflects the - residual backlog after the sweep. Best-effort: a metrics/DB hiccup here must - never fail the reconciliation cycle. + Called from ``ReconcilerScheduler._scheduler_loop`` — deliberately NOT from + ``run_vector_reconciliation_cycle``. The cycle runs off the queue behind + work-unit dedup, so exactly one deriver replica executes it; driving the + gauge from there would leave every other replica exporting a stale value, or + (worse, given this metric is zero-initialized) a confident permanent 0 it had + never measured. The count is a property of the database, not of the process, + so every replica must refresh it on its own timer for ``max()``/``avg()`` to + mean anything. One indexed COUNT per replica per interval is negligible — + ``ix_message_embeddings_sync_state_last_sync_at`` covers it. + + Best-effort: a metrics/DB hiccup here must never break the scheduler loop. """ if not settings.METRICS.ENABLED: return @@ -754,7 +762,6 @@ async def run_vector_reconciliation_cycle() -> ReconciliationMetrics: if not (embs_work or cleanup_work): break logger.debug("Vector reconciliation cycle completed (pgvector mode)") - await _record_pending_embeddings_backlog() return metrics # External vector store mode - reconcile documents, embeddings, and cleanup @@ -782,5 +789,4 @@ async def run_vector_reconciliation_cycle() -> ReconciliationMetrics: break logger.debug("Vector reconciliation cycle completed") - await _record_pending_embeddings_backlog() return metrics diff --git a/src/telemetry/events/__init__.py b/src/telemetry/events/__init__.py index dfba4084..65dd6614 100644 --- a/src/telemetry/events/__init__.py +++ b/src/telemetry/events/__init__.py @@ -147,8 +147,14 @@ __all__ = [ # Explicit registry of CloudEvents `type` values, used to pre-materialize the # `telemetry_events_emitted` / `telemetry_events_sampled_out` counter children at # 0 so those metrics are visible in Prometheus before any event fires (see -# src/telemetry/prometheus/metrics.py:initialize_bounded_metrics and -# .meta design telemetry-counter-zero-init). +# src/telemetry/prometheus/metrics.py:initialize_bounded_metrics for why absent +# and zero are worth distinguishing). +# +# This is an explicit literal rather than a set derived from BaseEvent +# subclasses: the derived version would silently follow whatever happens to be +# imported at init time, so a type could drop out of the registry without any +# code change. Pairing a hand-maintained list with a drift-guard test keeps the +# failure loud and at the right moment. # # ⚠️ When you add a new BaseEvent subclass, add its `_event_type` here (and to # HIGH_VOLUME_EVENT_TYPES if `_volume_class == "high_volume"`). The drift-guard diff --git a/src/telemetry/prometheus/metrics.py b/src/telemetry/prometheus/metrics.py index d271b4f7..217e690c 100644 --- a/src/telemetry/prometheus/metrics.py +++ b/src/telemetry/prometheus/metrics.py @@ -5,7 +5,7 @@ from __future__ import annotations import logging from collections.abc import Iterator from enum import Enum -from typing import cast, final, get_args +from typing import cast, final from prometheus_client import ( CONTENT_TYPE_LATEST, @@ -20,7 +20,8 @@ from prometheus_client.core import GaugeMetricFamily from starlette.requests import Request from starlette.responses import Response -from src.config import ReasoningLevel, settings +from src.config import REASONING_LEVELS, settings +from src.utils.types import walk_subclasses disable_created_metrics() @@ -66,11 +67,11 @@ class DialecticComponents(Enum): TOTAL = "total" -# Bounded label domains used to zero-initialize counter children at startup (see -# initialize_bounded_metrics). REASONING_LEVELS is derived from the config -# Literal so it never drifts. -REASONING_LEVELS: tuple[str, ...] = get_args(ReasoningLevel) - +# Bounded label domains used to zero-initialize counter children at startup are +# defined below (see initialize_bounded_metrics). Reasoning levels come from +# src.config.REASONING_LEVELS, which is derived from the config Literal so it +# never drifts. +# # Valid (token_type, component) pairs for deriver_tokens_processed, per task_type. # NOT the cartesian product: input tokens only pair with input components, output # only with OUTPUT_TOTAL, and PREVIOUS_SUMMARY occurs only for summary tasks @@ -186,10 +187,16 @@ telemetry_buffer_size_gauge = NamespacedGauge( # Embedding backlog: MessageEmbedding rows still awaiting a vector # (sync_state='pending'). Distinct from embed_now_tasks_in_flight (which counts # in-flight fast-path work in the API process) — this is the durable, DB-wide -# backlog the reconciler drains. Set once per reconciliation cycle in the deriver. +# backlog the reconciler drains. Every deriver replica refreshes it on its own +# timer from ReconcilerScheduler._scheduler_loop, so replicas can disagree by at +# most one interval. The help string names the owner explicitly because this is +# the only gauge here whose value is service-wide rather than per-process, and a +# dashboard author must not reach for sum(). message_embeddings_pending_gauge = NamespacedGauge( "message_embeddings_pending", - "MessageEmbedding rows awaiting embedding (sync_state='pending')", + "MessageEmbedding rows awaiting embedding (sync_state='pending'). " + + "Service-wide DB count, reported independently by every replica — " + + "aggregate with max() or avg(), never sum()", ["namespace"], ) @@ -393,23 +400,54 @@ class PrometheusMetrics: startup keeps the metric present at 0, so a missing series signals a broken scrape rather than "no drops". - Called per-emitter from ``TelemetryEmitter.start()`` because the reason - values are prefix-dependent (the trace emitter uses a ``trace_`` prefix), - which the process-level ``initialize_bounded_metrics`` does not know. + Called per-emitter from ``TelemetryEmitter.start()`` rather than hoisted + into the process-level ``initialize_bounded_metrics``: the trace emitter + (whose reasons carry a ``trace_`` prefix) only exists when + ``TRACE_PAYLOADS_ENABLED`` is set, so hoisting would fabricate ``trace_*`` + series on deployments that run with tracing off. Args: reasons: The reason label values the calling emitter can produce. """ + if not settings.METRICS.ENABLED: + return + for reason in reasons: self._touch(telemetry_events_dropped_counter, reason=reason) def initialize_bounded_metrics(self, *, instance_type: str) -> None: """Pre-create bounded-label counter children at 0 for this process. - See .meta design telemetry-counter-zero-init. Only counters whose full - label domain is bounded, enumerable at startup, and actually emitted by - THIS process are materialized; high-cardinality labels (endpoint, - workspace_name) and impossible label tuples are deliberately left absent. + A Prometheus counter does not exist until its first increment, so a + never-yet-incremented metric is indistinguishable from a broken scrape: + you cannot graph or alert on a series that is absent. Materializing the + children at 0 inverts that — a missing series now means something is + wrong, and "no events" reads as a flat 0 instead of a gap. + + That only holds for label sets we can enumerate honestly, so a metric is + initialized here only when its full label domain is bounded, enumerable + at startup, and actually emitted by THIS process. High-cardinality labels + (endpoint, workspace_name) and impossible label tuples are deliberately + left absent — fabricating a permanently-0 series that no code path can + ever increment is the same lie in the other direction. + + Multi-instance safety splits the metrics here into three buckets: + + 1. instance-scoped (``telemetry_buffer_size``, + ``embed_now_tasks_in_flight``) — per-process by nature, so any + aggregation is meaningful and zero-init is unambiguously right. + 2. service-scoped additive (the token counters, + ``telemetry_events_emitted``) — each instance holds a partial count + and ``sum()`` reconstructs the whole, so multi-instance safe. + 3. service-scoped non-additive — every instance claims to speak for the + whole service, so no aggregation is correct once they disagree. A + metric in this bucket MUST be refreshed by every instance on its own + timer (see ``message_embeddings_pending``, refreshed per replica from + ``ReconcilerScheduler._scheduler_loop``), or it does not belong in the + app at all — it belongs in an exporter that yields one series. + + Prometheus stamps ``instance``/``job`` at scrape time, which is why + buckets 1 and 2 need no special handling. ``telemetry_events_dropped`` is handled separately, per-emitter, in ``TelemetryEmitter.start()`` (prefix-dependent — see above). @@ -462,10 +500,11 @@ class PrometheusMetrics: ) # dreamer tokens: specialist_name x token_type. Specialist names are # derived from the concrete BaseSpecialist subclasses so a new - # specialist can't silently miss init. + # specialist can't silently miss init. Walked recursively — a + # specialist that subclasses another specialist is still a specialist. from src.dreamer.specialists import BaseSpecialist - for specialist in BaseSpecialist.__subclasses__(): + for specialist in walk_subclasses(BaseSpecialist): for token_type in TokenTypes: self._touch( dreamer_tokens_processed_counter, diff --git a/src/utils/types.py b/src/utils/types.py index 33a2993d..a6b14bac 100644 --- a/src/utils/types.py +++ b/src/utils/types.py @@ -1,4 +1,4 @@ -from collections.abc import Awaitable, Callable, Generator +from collections.abc import Awaitable, Callable, Generator, Iterator from contextlib import contextmanager from contextvars import ContextVar from dataclasses import dataclass, field @@ -6,6 +6,20 @@ from typing import Any, Generic, Literal, TypeVar T = TypeVar("T") + +def walk_subclasses(cls: type[T]) -> Iterator[type[T]]: + """Yield every subclass of ``cls``, recursively. + + ``type.__subclasses__()`` is direct-children-only, so a grandchild class is + silently invisible to it. Any registry that enumerates subclasses to decide + what to initialize or validate wants the transitive closure — otherwise + subclassing a concrete class is enough to slip past the check. + """ + for subclass in cls.__subclasses__(): + yield subclass + yield from walk_subclasses(subclass) + + # Context variable for tracking current iteration in tool execution loop # This is used for telemetry to associate tool calls with their iteration _current_iteration: ContextVar[int] = ContextVar("current_iteration", default=0) diff --git a/tests/reconciler/test_pending_backlog_gauge.py b/tests/reconciler/test_pending_backlog_gauge.py new file mode 100644 index 00000000..f1008181 --- /dev/null +++ b/tests/reconciler/test_pending_backlog_gauge.py @@ -0,0 +1,79 @@ +"""The pending-embeddings backlog gauge must be refreshed per-replica. + +``message_embeddings_pending`` reports a DB-global count, so it is the one gauge +here whose value is service-wide rather than per-process. It is also zero- +initialized at startup, which makes a missing refresh actively harmful: a replica +that never measured the backlog would export a confident, permanently-healthy 0. + +The count therefore has to be driven from ``ReconcilerScheduler._scheduler_loop`` +(runs on every replica, every interval) and NOT from +``run_vector_reconciliation_cycle`` (runs off the queue behind work-unit dedup, so +exactly one replica per cycle executes it). These tests pin both halves of that. +""" + +import asyncio + +import pytest + +from src.reconciler import scheduler as scheduler_module +from src.reconciler import sync_vectors +from src.reconciler.scheduler import ReconcilerScheduler + + +@pytest.fixture(autouse=True) +def _reset_scheduler_singleton(): + ReconcilerScheduler.reset_singleton() + yield + ReconcilerScheduler.reset_singleton() + + +async def test_scheduler_loop_refreshes_backlog_gauge( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Every scheduler iteration refreshes the gauge, on every replica. + + Patched at the scheduler's own reference so this asserts the call site, not + just that the function exists. + """ + calls = 0 + refreshed = asyncio.Event() + + async def _fake_refresh() -> None: + nonlocal calls + calls += 1 + refreshed.set() + + async def _never_enqueue(_task: object) -> bool: + return False + + monkeypatch.setattr( + scheduler_module, "record_pending_embeddings_backlog", _fake_refresh + ) + monkeypatch.setattr( + ReconcilerScheduler, "_try_enqueue_task", _never_enqueue, raising=True + ) + + scheduler = ReconcilerScheduler() + await scheduler.start() + try: + await asyncio.wait_for(refreshed.wait(), timeout=5.0) + finally: + await scheduler.shutdown() + + assert calls >= 1, "scheduler loop never refreshed the backlog gauge" + + +def test_reconciliation_cycle_does_not_drive_the_gauge() -> None: + """The queue-driven cycle must not be the thing that sets the gauge. + + If the refresh moves back into ``run_vector_reconciliation_cycle``, only the + replica that wins the ``sync_vectors`` work unit would ever measure the + backlog, and the zero-init would go back to lying on all the others. + + Structural guard: the cycle is a long DB-driven coroutine, so this inspects + the global names it references rather than executing it. + """ + assert hasattr(sync_vectors, "record_pending_embeddings_backlog") + + referenced = sync_vectors.run_vector_reconciliation_cycle.__code__.co_names + assert "record_pending_embeddings_backlog" not in referenced diff --git a/tests/telemetry/test_metric_zero_init.py b/tests/telemetry/test_metric_zero_init.py index 8b5cde93..d3e9a982 100644 --- a/tests/telemetry/test_metric_zero_init.py +++ b/tests/telemetry/test_metric_zero_init.py @@ -18,18 +18,19 @@ from uuid import uuid4 import pytest from prometheus_client import REGISTRY -from src.config import settings +from src.config import REASONING_LEVELS, settings +from src.dreamer.specialists import BaseSpecialist from src.telemetry.events import ALL_EVENT_TYPES, HIGH_VOLUME_EVENT_TYPES from src.telemetry.events.base import BaseEvent from src.telemetry.prometheus.metrics import ( _DERIVER_TOKEN_COMBOS_BY_TASK, # pyright: ignore[reportPrivateUsage] - REASONING_LEVELS, DeriverComponents, DeriverTaskTypes, DialecticComponents, TokenTypes, prometheus_metrics, ) +from src.utils.types import walk_subclasses def unique_ns(tag: str) -> str: @@ -70,12 +71,6 @@ def sample(name: str, **labels: str) -> float | None: # --------------------------------------------------------------------------- -def _walk_event_subclasses(cls: type[BaseEvent]) -> Iterator[type[BaseEvent]]: - for sub in cls.__subclasses__(): - yield sub - yield from _walk_event_subclasses(sub) - - def test_all_event_types_registry_matches_subclasses(): """ALL_EVENT_TYPES must equal every BaseEvent subclass's _event_type. @@ -85,7 +80,7 @@ def test_all_event_types_registry_matches_subclasses(): """ discovered = { event_type - for cls in _walk_event_subclasses(BaseEvent) + for cls in walk_subclasses(BaseEvent) if (event_type := getattr(cls, "_event_type", None)) is not None } assert set(ALL_EVENT_TYPES) == discovered @@ -96,7 +91,7 @@ def test_high_volume_registry_matches_subclasses(): """HIGH_VOLUME_EVENT_TYPES must equal the high_volume-classed subclasses.""" discovered = { event_type - for cls in _walk_event_subclasses(BaseEvent) + for cls in walk_subclasses(BaseEvent) if (event_type := getattr(cls, "_event_type", None)) is not None and getattr(cls, "_volume_class", None) == "high_volume" } @@ -197,8 +192,18 @@ def test_deriver_init_materializes_token_and_backlog(): ) is not None ) - # dreamer specialists are derived from the concrete BaseSpecialist subclasses - for specialist_name in ("deduction", "induction"): + # dreamer specialists are derived from the concrete BaseSpecialist subclasses. + # Derived here too, rather than hardcoded: a hardcoded pair would keep passing + # when a third specialist is added (it only asserts presence), silently leaving + # the new one uncovered — which is exactly what happened when CardRefreshSpecialist + # landed. + specialist_names = { + name + for cls in walk_subclasses(BaseSpecialist) + if (name := getattr(cls, "name", None)) is not None + } + assert {"deduction", "induction", "card_refresh"} <= specialist_names + for specialist_name in specialist_names: assert ( sample( "dreamer_tokens_processed_total", @@ -206,7 +211,7 @@ def test_deriver_init_materializes_token_and_backlog(): token_type=TokenTypes.INPUT.value, ) is not None - ) + ), f"specialist {specialist_name!r} was not zero-initialized" assert sample("message_embeddings_pending") == 0.0 # gauge zero-init