Measure public cache effectiveness and invalidation causes

apps/api/src/cache.py

@@ -27,11 +27,16 @@ def identity(request: Request) -> str:
 
 
 def tagged(
-    request: Request, response: Response, status: str, reason: str | None = None
+    request: Request,
+    response: Response,
+    status: str,
+    reason: str | None = None,
+    *,
+    route: str,
 ) -> Response:
     request.state.cache_status = status
     request.state.cache_bypass_reason = reason
-    record_cache_read("public_response", status.lower(), reason)
+    record_cache_read("public_response", status.lower(), reason, route=route)
     response.headers["X-Cache"] = status
     if reason:
         response.headers["X-Cache-Bypass-Reason"] = reason
@@ -78,7 +83,7 @@ async def render():
             elif len(request.scope.get("query_string", b"")) > 4096:
                 reason = "query_too_long"
             if reason:
-                return tagged(request, await render(), "BYPASS", reason)
+                return tagged(request, await render(), "BYPASS", reason, route=self.path)
             ttl = (
                 settings.cache_metadata_ttl_seconds
                 if self.path.startswith(("/v1/sources", "/v1/tags", "/v1/topics"))
@@ -94,13 +99,18 @@ async def render():
                     )
                     if lookup.body is not None:
                         return tagged(
-                            request, Response(lookup.body, media_type="application/json"), "HIT"
+                            request,
+                            Response(lookup.body, media_type="application/json"),
+                            "HIT",
+                            route=self.path,
                         )
                     if lookup.token or time.monotonic() >= deadline:
                         break
                     await anyio.sleep(POLL_SECONDS)
             except CacheUnavailable:
-                return tagged(request, await render(), "BYPASS", "cache_unavailable")
+                return tagged(
+                    request, await render(), "BYPASS", "cache_unavailable", route=self.path
+                )
             try:
                 response = await render()
                 body = getattr(response, "body", None)
@@ -114,7 +124,7 @@ async def render():
                 ):
                     with suppress(CacheUnavailable):
                         await run_in_threadpool(cache.publish, lookup, body, ttl)
-                return tagged(request, response, "MISS")
+                return tagged(request, response, "MISS", route=self.path)
             finally:
                 # Bounded, owner-checked release even on endpoint failure/cancellation.
                 with anyio.CancelScope(shield=True), suppress(CacheUnavailable):

apps/cli/src/commands.py

@@ -41,8 +41,8 @@ class InputError(ValueError):
 
 def cache_clear(args):
     cache = get_cache()
-    cache.invalidate()
-    cache.invalidate("operations")
+    cache.invalidate(reasons=("manual",))
+    cache.invalidate("operations", reasons=("manual",))
     return {"cleared": True, "scope": "api_get_responses"}
 
 

apps/cli/src/runtime.py

@@ -16,6 +16,7 @@
 from devfeed_core.feeds.validation import FeedValidationError
 from devfeed_core.logging import configure_logging, elapsed_ms, log_context
 from devfeed_core.services import OperationConflict, RecordNotFound
+from devfeed_core.telemetry import current, start_runtime, stop_runtime
 from pydantic import ValidationError
 from redis.exceptions import RedisError
 from sqlalchemy.exc import IntegrityError, SQLAlchemyError
@@ -64,9 +65,18 @@ def invoke(ctx: typer.Context, handler, values: dict, **overrides) -> None:
 
 
 def execute(args) -> int:
+    telemetry = None
     try:
         settings = get_settings()
         configure_logging("cli", settings.log_level, settings.log_format)
+        # Cache-writing commands need a final OTLP flush. Keep help offline and
+        # leave worker/scheduler runtime ownership with their existing entrypoints.
+        if (
+            args.command
+            in {"sources", "articles", "tags", "topics", "images", "discovery", "cache"}
+            and current() is None
+        ):
+            telemetry = start_runtime("cli", serve_metrics=False, profiling=False)
         logger.info("command_started")
         result = args.execute(args)
         fields = {}
@@ -177,5 +187,7 @@ def execute(args) -> int:
         print("error: Unexpected command failure; see diagnostic logs.", file=sys.stderr)
         return 1
     finally:
+        if telemetry is not None:
+            stop_runtime(telemetry)
         if get_engine.cache_info().currsize:
             get_engine().dispose()

apps/user-api/src/avatars.py

@@ -95,7 +95,7 @@ def upload_avatar(
         session.flush()
         result = profile_value(session, account)
         session.commit()
-    invalidate_public_cache()
+    invalidate_public_cache(reasons=("avatar",))
     return result
 
 
@@ -116,5 +116,5 @@ def remove_avatar(user: User, session: DB):
         session.flush()
         result = profile_value(session, account)
         session.commit()
-    invalidate_public_cache()
+    invalidate_public_cache(reasons=("avatar",))
     return result

apps/user-api/src/profile.py

@@ -214,7 +214,7 @@ def save_profile(payload: UserProfileUpdate, user: User, session: DB):
         if objects is not None:
             objects.remove()
         session.commit()
-    invalidate_public_cache()
+    invalidate_public_cache(reasons=("profile",))
     return result
 
 

packages/core/OBSERVABILITY.md

@@ -46,7 +46,9 @@ Avoid an alert per route or an alert for every cache miss.
 | `http.server.active_requests`                         | Requests still sending responses.                                                                                                                   |
 | `devfeed.admission.active`, `devfeed.admission.limit` | Actual occupied handler slots versus the configured concurrency budget, separately for requests and streams. Slots include background/cleanup work. |
 | `devfeed.admission.rejections`                        | Overload rejections before database checkout. These also appear as HTTP 503s.                                                                       |
-| `devfeed.cache.reads`                                 | Hits, misses, bypasses, and bounded reasons by cache name.                                                                                          |
+| `devfeed.cache.reads`                                 | Hits, misses, bypasses, and bounded reasons by cache name; public response decisions also identify the route template.                              |
+| `devfeed.cache.invalidations`                         | Generation invalidation attempts by domain and outcome, across API, CLI, and background writers.                                                    |
+| `devfeed.cache.invalidation.causes`                   | Distinct bounded causes of each invalidation attempt; one attempt can have several causes.                                                          |
 | `db.client.operation.duration` (seconds)              | SQL operation latency and failures, with operation type and an error marker.                                                                        |
 | `db.client.connection.wait_time` (seconds)            | Connection acquisition, including pool wait, connect, and pre-ping; distinguishes `ok`, `timeout`, and `error`.                                     |
 | `db.client.connection.count`                          | Checked-out connections (`state=used`).                                                                                                             |
@@ -83,12 +85,106 @@ have their own cache name, with stale served snapshots marked `stale`. Feed sequ
 reads count actual lookup attempts. Catalog revision mismatches count as misses;
 cache failures fall back to the existing database path and count as bypasses.
 Only aggregate outcomes are recorded, never cache keys or user identifiers.
+Public route decisions also include `http.route` from an allowlist of the cached
+route templates. Unknown templates become `other`; concrete paths, search terms,
+IDs, and query parameters cannot become labels. Shared low-level lookup metrics
+and other cache names have no route label. Responses that raise before the final
+cache decision remain visible in the HTTP error metrics, rather than as cache misses.
 
 Hit ratio is `hits / (hits + misses)`. Track bypass share separately: including
 bypasses in hit ratio hides whether caching is disabled or unavailable. A miss is
 normal; investigate a sustained change together with request latency and database
 pressure. A cache hit does not bypass authentication or visibility checks.
 
+### Measuring public cache effectiveness
+
+Keep the existing generation boundaries and TTLs until measurements identify a
+specific cause. `devfeed.cache.invalidations` records one attempt with
+`cache.domain=public|operations` and
+`cache.outcome=success|cache_unavailable|disabled`. Only `success` means a generation
+was replaced. Failures and the Redis circuit breaker are counted without failing
+the committed database write; disabled caching does not contact Redis.
+
+`devfeed.cache.invalidation.causes` adds `cache.invalidation_reason`. The bounded
+reasons are `sources`, `articles`, `article_origins`, `article_tags`, `tags`, `topics`,
+`topic_relations`, `article_topics`, `profile`, `avatar`, `search_index`, `manual`,
+`unspecified`, and `other`. ORM and bulk writes use the existing dirty-write rules,
+collect table names, and report them at the outer commit. Duplicate upserts,
+zero-row updates, private writes, and source polling retain their existing exclusions.
+An outer rollback discards the causes. Multiple writes to the same table in a
+transaction contribute one cause; a transaction touching several tables still
+counts as one invalidation. Explicit profile, avatar, search-index, and CLI clears
+have their own reasons. Never sum the causes counter to calculate invalidation count.
+
+Use a consistent environment and time window. Calculate rates before summing replicas
+so counter resets are handled correctly. These examples use the production namespace
+for reader scrapes and the OTLP environment for writer totals. Do not combine scraped
+and OTLP copies of the same metric: that would double count long-lived services.
+
+Route hit ratio, with zero hits represented for routes that only miss:
+
+```promql
+(
+  sum by (service_name, http_route) (
+    rate(devfeed_cache_reads_total{namespace="devfeed",cache_name="public_response",cache_outcome="hit",http_route!=""}[15m])
+  )
+  or
+  0 * sum by (service_name, http_route) (
+    rate(devfeed_cache_reads_total{namespace="devfeed",cache_name="public_response",cache_outcome=~"hit|miss",http_route!=""}[15m])
+  )
+)
+/
+sum by (service_name, http_route) (
+  rate(devfeed_cache_reads_total{namespace="devfeed",cache_name="public_response",cache_outcome=~"hit|miss",http_route!=""}[15m])
+)
+```
+
+Route traffic and bypass reasons, kept separate from hit ratio:
+
+```promql
+sum by (service_name, http_route, cache_outcome, cache_bypass_reason) (
+  rate(devfeed_cache_reads_total{namespace="devfeed",cache_name="public_response",http_route!=""}[15m])
+)
+```
+
+Successful invalidations across **all writers** sharing the cache, and their causes:
+
+```promql
+sum by (cache_domain) (
+  rate(devfeed_cache_invalidations_total{service_name=~"devfeed-.*",deployment_environment_name="production",cache_outcome="success"}[15m])
+)
+
+sum by (service_name, cache_domain, cache_invalidation_reason) (
+  rate(devfeed_cache_invalidation_causes_total{service_name=~"devfeed-.*",deployment_environment_name="production",cache_outcome="success"}[15m])
+)
+```
+
+RQ work horses and CLI writers export counter increments using OTLP delta
+temporality, including the final shutdown flush. Consecutive children reuse their pod identity, so their
+collector must accumulate deltas before exposing cumulative Prometheus counters.
+The production collector already uses `deltatocumulative` with stream-affine routing.
+Parent Prometheus scrapes and long-lived service counters retain their existing
+temporality. Worker child metrics require `DEVFEED_OTLP_ENDPOINT`; they are not
+served by the parent scrape listener and lack its Kubernetes `namespace` label.
+Use the OTLP-only environment selector above for all-writer totals. Verify collector
+delivery and writer coverage after rollout, including short jobs and quiet writers;
+absent series are not evidence of zero invalidations. CLI source, article, tag,
+topic, image, discovery, and cache commands initialize and flush telemetry when
+configured. Help and completion stay offline; worker and scheduler commands own their runtimes.
+
+Graph invalidations alongside route misses, request volume, HTTP latency, and
+database pressure using the same window. Review the other invalidation outcomes
+and bypass reasons as well. Sparse traffic, many distinct request variants, TTL
+expiry, and repeated invalidation can all lower hit ratio; aggregate counters
+cannot attribute an individual miss to a specific cause. Correlation is evidence
+for a targeted follow-up, not proof that a cache boundary should change.
+
+The route label creates new series on rollout. Exclude unlabeled historical
+lookups from route comparisons, allow two scrapes before calculating rates, and
+use a full post-rollout window after updating **all** writer services. Missing
+series or a zero request denominator do not mean a healthy hit ratio. No database
+migration, cache clear, key-format change, or TTL change is required.
+
 ## Background services
 
 Workers, scheduler, and indexer use the same provider/exporters and a small set of
@@ -213,6 +309,8 @@ indexes remain; no schema migration is needed.
 - [OpenTelemetry HTTP metric conventions](https://opentelemetry.io/docs/specs/semconv/http/http-metrics/)
 - [Google SRE: monitoring distributed systems](https://sre.google/sre-book/monitoring-distributed-systems/)
 - [OpenTelemetry Prometheus exporter](https://opentelemetry-python.readthedocs.io/en/latest/exporter/prometheus/prometheus.html)
+- [OpenTelemetry Python OTLP exporters](https://opentelemetry-python.readthedocs.io/en/latest/exporter/otlp/otlp.html)
+- [Delta to cumulative collector processor](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/processor/deltatocumulativeprocessor/README.md)
 
 ## Investigating database query totals
 

packages/core/src/cache.py

@@ -7,6 +7,7 @@
 import threading
 import time
 import uuid
+from collections.abc import Collection
 from contextlib import suppress
 from dataclasses import dataclass
 from functools import lru_cache
@@ -45,18 +46,72 @@
         "none",
     }
 )
+PUBLIC_CACHE_ROUTES = frozenset(
+    {
+        "/v1/feed",
+        "/v1/feed/options",
+        "/v1/articles/{article_id}",
+        "/v1/sources",
+        "/v1/sources/{source_id}",
+        "/v1/tags",
+        "/v1/tags/{slug}",
+        "/v1/topics",
+        "/v1/topics/{slug}",
+        "/v1/topics/{slug}/relations",
+        "/v1/search",
+        "/v1/search/suggestions",
+    }
+)
+INVALIDATION_REASONS = frozenset(
+    {
+        "sources",
+        "articles",
+        "article_origins",
+        "article_tags",
+        "tags",
+        "topics",
+        "topic_relations",
+        "article_topics",
+        "profile",
+        "avatar",
+        "search_index",
+        "manual",
+        "unspecified",
+        "other",
+    }
+)
 
 
-def record_cache_read(name: str, outcome: str, reason: str | None = None) -> None:
+def record_cache_read(
+    name: str, outcome: str, reason: str | None = None, *, route: str | None = None
+) -> None:
     if (runtime := current()) and (instrument := runtime.instruments.get("cache")):
-        instrument.add(
-            1,
-            {
-                "cache.name": name if name in CACHE_NAMES else "other",
-                "cache.outcome": outcome if outcome in {"hit", "miss", "bypass"} else "other",
-                "cache.bypass_reason": reason if reason in CACHE_REASONS else "none",
-            },
-        )
+        attributes = {
+            "cache.name": name if name in CACHE_NAMES else "other",
+            "cache.outcome": outcome if outcome in {"hit", "miss", "bypass"} else "other",
+            "cache.bypass_reason": reason if reason in CACHE_REASONS else "none",
+        }
+        if name == "public_response" and route is not None:
+            attributes["http.route"] = route if route in PUBLIC_CACHE_ROUTES else "other"
+        instrument.add(1, attributes)
+
+
+def record_cache_invalidation(domain: str, outcome: str, reasons: Collection[str]) -> None:
+    if runtime := current():
+        attributes = {
+            "cache.domain": domain if domain in {"public", "operations"} else "other",
+            "cache.outcome": (
+                outcome if outcome in {"success", "cache_unavailable", "disabled"} else "other"
+            ),
+        }
+        if instrument := runtime.instruments.get("cache_invalidations"):
+            instrument.add(1, attributes)
+        if instrument := runtime.instruments.get("cache_invalidation_causes"):
+            # One invalidation can have several causes. Deduplicate within the
+            # transaction; never count this counter as the number of invalidations.
+            bounded = {reason if reason in INVALIDATION_REASONS else "other" for reason in reasons}
+            for reason in bounded or {"unspecified"}:
+                instrument.add(1, {**attributes, "cache.invalidation_reason": reason})
 
 
 # A late user may not repopulate a generation invalidated by a committed write,
@@ -183,10 +238,19 @@ def release(self, lookup: CacheLookup) -> None:
         if lookup.token:
             self._run(lambda: self.redis.eval(RELEASE, 1, lookup.lock_key, lookup.token))
 
-    def invalidate(self, domain: str = "public") -> None:
+    def invalidate(
+        self, domain: str = "public", *, reasons: Collection[str] = ("unspecified",)
+    ) -> None:
         if domain not in {"public", "operations"}:
             raise ValueError("Unknown cache domain")
-        self._run(lambda: self.redis.set(f"{self.namespace}:{domain}:generation", uuid.uuid4().hex))
+        try:
+            self._run(
+                lambda: self.redis.set(f"{self.namespace}:{domain}:generation", uuid.uuid4().hex)
+            )
+        except CacheUnavailable:
+            record_cache_invalidation(domain, "cache_unavailable", reasons)
+            raise
+        record_cache_invalidation(domain, "success", reasons)
         logger.debug("cache_invalidated")
 
     def read_sequence(self, identity: str, start: int, stop: int) -> list[bytes] | None:
@@ -249,9 +313,11 @@ def close_cache() -> None:
     _cache_for_process.cache_clear()
 
 
-def invalidate_public_cache() -> None:
+def invalidate_public_cache(*, reasons: Collection[str] = ("unspecified",)) -> None:
     if get_settings().cache_enabled:
         # A successful database commit must never be reported as failed just
         # because disposable cache invalidation failed. TTL bounds staleness.
         with suppress(CacheUnavailable):
-            get_cache().invalidate()
+            get_cache().invalidate(reasons=reasons)
+    else:
+        record_cache_invalidation("public", "disabled", reasons)