diff --git a/apps/api/src/cache.py b/apps/api/src/cache.py index 2238de3b5..5bdf507b3 100644 --- a/apps/api/src/cache.py +++ b/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): diff --git a/apps/cli/src/commands.py b/apps/cli/src/commands.py index cb1d02340..a786da9e9 100644 --- a/apps/cli/src/commands.py +++ b/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"} diff --git a/apps/cli/src/runtime.py b/apps/cli/src/runtime.py index 96de97a64..7bf237e79 100644 --- a/apps/cli/src/runtime.py +++ b/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() diff --git a/apps/user-api/src/avatars.py b/apps/user-api/src/avatars.py index a285af7de..cde61b4cd 100644 --- a/apps/user-api/src/avatars.py +++ b/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 diff --git a/apps/user-api/src/profile.py b/apps/user-api/src/profile.py index 898f7c1a7..e68dada6d 100644 --- a/apps/user-api/src/profile.py +++ b/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 diff --git a/packages/core/OBSERVABILITY.md b/packages/core/OBSERVABILITY.md index cf86b17de..5a215c8a3 100644 --- a/packages/core/OBSERVABILITY.md +++ b/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 diff --git a/packages/core/src/cache.py b/packages/core/src/cache.py index fd47eeaa2..5aa705b8e 100644 --- a/packages/core/src/cache.py +++ b/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) diff --git a/packages/core/src/cache_events.py b/packages/core/src/cache_events.py index d4ec2a860..516bfc85c 100644 --- a/packages/core/src/cache_events.py +++ b/packages/core/src/cache_events.py @@ -34,14 +34,21 @@ } ) DIRTY = "devfeed_public_cache_dirty" +REASONS = "devfeed_public_cache_reasons" PRIVATE_ARTICLES = "devfeed_private_articles" SESSION_OPTION = "devfeed_cache_session" +TABLE_OPTION = "devfeed_cache_table" class AppSession(Session): """Only application sessions participate; migrations and raw connections don't.""" +def mark_dirty(session, table): + session.info[DIRTY] = True + session.info.setdefault(REASONS, set()).add(table) + + @event.listens_for(AppSession, "before_flush") def track_objects(session, flush_context, instances): for obj in session.new | session.deleted | session.dirty: @@ -63,10 +70,10 @@ def track_objects(session, flush_context, instances): if known_private or (obj in session.new and visibility.loaded_value != "published"): continue if obj in session.new or obj in session.deleted: - session.info[DIRTY] = True + mark_dirty(session, table) elif table == "sources": if any(state.attrs[field].history.has_changes() for field in SOURCE_FIELDS): - session.info[DIRTY] = True + mark_dirty(session, table) elif table == "articles": private = {"editorial_revision", "review_status", "classification_provenance"} if any( @@ -74,9 +81,9 @@ def track_objects(session, flush_context, instances): for key, attr in state.attrs.items() if key not in private ): - session.info[DIRTY] = True + mark_dirty(session, table) elif session.is_modified(obj, include_collections=True): - session.info[DIRTY] = True + mark_dirty(session, table) @event.listens_for(AppSession, "do_orm_execute") @@ -87,10 +94,13 @@ def track_statements(state): if getattr(state, "execution_options", {}).get("devfeed_private_write"): return table = getattr(state.statement, "table", None) - if getattr(table, "name", None) in PUBLIC_TABLES: + table_name = getattr(table, "name", None) + if table_name in PUBLIC_TABLES: # Do not invalidate merely because a write was attempted. Ignored # duplicates and conditional updates matching zero rows are no-ops. - state.update_execution_options(**{SESSION_OPTION: weakref.ref(state.session)}) + state.update_execution_options( + **{SESSION_OPTION: weakref.ref(state.session), TABLE_OPTION: table_name} + ) @event.listens_for(Engine, "after_cursor_execute") @@ -102,19 +112,21 @@ def track_affected_rows(connection, cursor, statement, parameters, context, exec # Unknown counts (-1) conservatively invalidate rather than risking stale data. session = reference() if session is not None: - session.info[DIRTY] = True + mark_dirty(session, context.execution_options[TABLE_OPTION]) @event.listens_for(AppSession, "after_commit") def committed(session): if not session.in_nested_transaction(): session.info.pop(PRIVATE_ARTICLES, None) + reasons = session.info.pop(REASONS, ()) if session.info.pop(DIRTY, False): - invalidate_public_cache() + invalidate_public_cache(reasons=reasons) @event.listens_for(AppSession, "after_soft_rollback") def rolled_back(session, previous_transaction): if previous_transaction.parent is None: session.info.pop(DIRTY, None) + session.info.pop(REASONS, None) session.info.pop(PRIVATE_ARTICLES, None) diff --git a/packages/core/src/search_index.py b/packages/core/src/search_index.py index 83eaf3817..8cfcb9c94 100644 --- a/packages/core/src/search_index.py +++ b/packages/core/src/search_index.py @@ -192,5 +192,5 @@ def sync_batch(factory, engine=None): session.execute( delete(SearchEvent).where(SearchEvent.id.in_([event.id for event in events])) ) - invalidate_public_cache() + invalidate_public_cache(reasons=("search_index",)) return len(events) diff --git a/packages/core/src/telemetry.py b/packages/core/src/telemetry.py index a25b4eb56..3f625ece9 100644 --- a/packages/core/src/telemetry.py +++ b/packages/core/src/telemetry.py @@ -15,7 +15,7 @@ from functools import wraps from opentelemetry import metrics, trace -from opentelemetry.sdk.metrics import MeterProvider +from opentelemetry.sdk.metrics import Counter, MeterProvider from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor @@ -167,13 +167,29 @@ def start_metrics(self, readers=None, *, serve_metrics=True, export_otlp=True) - from opentelemetry.exporter.otlp.proto.http.metric_exporter import ( OTLPMetricExporter, ) - from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader + from opentelemetry.sdk.metrics.export import ( + AggregationTemporality, + PeriodicExportingMetricReader, + ) readers.append( PeriodicExportingMetricReader( OTLPMetricExporter( endpoint=self.settings.otlp_endpoint.rstrip("/") + "/v1/metrics", timeout=1, + # Short-lived writers reuse the pod identity but + # reset counters every invocation. Export + # increments so the collector can accumulate + # invalidations across + # children, including their final shutdown flush. + preferred_temporality=( + {Counter: AggregationTemporality.DELTA} + if not serve_metrics + and ( + self.service.startswith("worker-") or self.service == "cli" + ) + else None + ), ), export_interval_millis=60000, export_timeout_millis=1000, @@ -340,6 +356,12 @@ def create_instruments(meter): explicit_bucket_boundaries_advisory=DURATION_BUCKETS, ), "cache": meter.create_counter("devfeed.cache.reads", unit="{read}"), + "cache_invalidations": meter.create_counter( + "devfeed.cache.invalidations", unit="{invalidation}" + ), + "cache_invalidation_causes": meter.create_counter( + "devfeed.cache.invalidation.causes", unit="{cause}" + ), "db_wait": meter.create_histogram( "db.client.connection.wait_time", unit="s", diff --git a/tests/test_cache.py b/tests/test_cache.py index 3a2bd8798..ce99308ce 100644 --- a/tests/test_cache.py +++ b/tests/test_cache.py @@ -297,21 +297,27 @@ def slow(statement): def test_commit_invalidation_is_not_called_for_rollback_or_savepoint_commit(monkeypatch): calls = [] - monkeypatch.setattr(cache_events, "invalidate_public_cache", lambda: calls.append(True)) + monkeypatch.setattr( + cache_events, "invalidate_public_cache", lambda **kwargs: calls.append(kwargs["reasons"]) + ) with cache_events.AppSession() as session: with session.begin(): with session.begin_nested(): session.info[cache_events.DIRTY] = True + session.info[cache_events.REASONS] = {"sources", "articles"} assert not calls - assert calls == [True] + assert calls == [{"sources", "articles"}] + assert cache_events.REASONS not in session.info with session.begin(): pass - assert calls == [True] + assert calls == [{"sources", "articles"}] session.begin() session.info[cache_events.DIRTY] = True + session.info[cache_events.REASONS] = {"tags"} session.rollback() session.commit() - assert calls == [True] + assert calls == [{"sources", "articles"}] + assert cache_events.REASONS not in session.info @pytest.mark.parametrize("field", ["description", "logo_url", "approval_status", "enabled"]) @@ -324,9 +330,11 @@ def test_source_public_changes_invalidate_but_polling_does_not(field): source.next_fetch_at = utcnow() cache_events.track_objects(session, None, None) assert cache_events.DIRTY not in session.info + assert cache_events.REASONS not in session.info setattr(source, field, False if field == "enabled" else "Changed") cache_events.track_objects(session, None, None) assert session.info[cache_events.DIRTY] + assert session.info[cache_events.REASONS] == {"sources"} def test_private_article_enrichment_does_not_clear_public_feed_cache(): @@ -337,6 +345,7 @@ def test_private_article_enrichment_does_not_clear_public_feed_cache(): article.ai_summary = "Private analysis output" cache_events.track_objects(session, None, None) assert cache_events.DIRTY not in session.info + assert cache_events.REASONS not in session.info def test_unpublish_invalidates_even_though_new_state_is_private(): @@ -384,6 +393,9 @@ def test_direct_worker_dml_marks_public_cache_only_after_affected_rows( False, ) assert bool(session.info.get(cache_events.DIRTY)) is (expected and affected != 0) + assert session.info.get(cache_events.REASONS, set()) == ( + {statement.table.name} if expected and affected != 0 else set() + ) def test_cache_invalidation_failure_does_not_fail_a_database_commit(response_cache, monkeypatch): diff --git a/tests/test_cache_integration.py b/tests/test_cache_integration.py index 5eeb005e1..1f9ac2e82 100644 --- a/tests/test_cache_integration.py +++ b/tests/test_cache_integration.py @@ -5,16 +5,23 @@ import pytest from devfeed_aggregator import image_tasks, source_tasks, tasks from devfeed_core import cache, services -from devfeed_core.cache_events import DIRTY +from devfeed_core.cache_events import DIRTY, REASONS from devfeed_core.config import get_settings from devfeed_core.db import get_engine from devfeed_core.feeds.fetcher import FetchResult from devfeed_core.jobs import request_ingestion -from devfeed_core.models import ArticleImageJob, Source, utcnow +from devfeed_core.models import ArticleImageJob, Source, Tag, utcnow from devfeed_core.schemas import SourceDecision +from devfeed_core.search_engine import SearchUnavailable +from devfeed_core.search_index import sync_batch from devfeed_core.source_enrichment import request_enrichment from sqlalchemy import event, select, update from sqlalchemy.dialects.postgresql import insert +from test_avatar_uploads import picture +from test_avatar_uploads import uploaded_reader as uploaded_reader +from test_observability import observed_runtime as observed_runtime +from test_observability import otel_points +from test_user_personalization import user_data as user_data pytestmark = pytest.mark.integration @@ -172,3 +179,114 @@ def test_duplicate_upsert_and_zero_row_update_do_not_invalidate(database, client assert result == identifier and session.info[DIRTY] response = client.get(path) assert response.headers["x-cache"] == "MISS" and response.json()["name"] == "Changed" + + +def test_invalidation_metrics_follow_commits_and_affected_rows(database, observed_runtime): + runtime, _ = observed_runtime + with database.begin() as session: + source = Source(name="Example", feed_url="https://example.com/rss", source_type="publisher") + session.add(source) + session.add_all([Tag(name="Python", slug="python"), Tag(name="Rust", slug="rust")]) + session.flush() + identifier = source.id + assert session.info[REASONS] == {"sources", "tags"} + assert not otel_points(runtime, "devfeed.cache.invalidations") + assert sum(point.value for point in otel_points(runtime, "devfeed.cache.invalidations")) == 1 + causes = otel_points(runtime, "devfeed.cache.invalidation.causes") + assert {point.attributes["cache.invalidation_reason"]: point.value for point in causes} == { + "sources": 1, + "tags": 1, + } + with database() as session: + session.execute(update(Source).where(Source.id == identifier).values(name="Rolled back")) + assert session.info[REASONS] == {"sources"} + session.rollback() + assert REASONS not in session.info + session.commit() + with database.begin() as session: + session.get(Source, identifier).last_success_at = utcnow() + session.execute(update(Source).where(Source.id == uuid.uuid4()).values(name="No match")) + session.execute( + insert(Source) + .values(name="Duplicate", feed_url="https://example.com/rss", source_type="publisher") + .on_conflict_do_nothing(index_elements=[Source.feed_url]) + .returning(Source.id) + ) + assert REASONS not in session.info + assert sum(point.value for point in otel_points(runtime, "devfeed.cache.invalidations")) == 1 + with database.begin() as session: + assert ( + session.scalar( + update(Source) + .where(Source.id == identifier) + .values(name="Changed") + .returning(Source.id) + ) + == identifier + ) + assert session.info[REASONS] == {"sources"} + assert sum(point.value for point in otel_points(runtime, "devfeed.cache.invalidations")) == 2 + causes = otel_points(runtime, "devfeed.cache.invalidation.causes") + assert {point.attributes["cache.invalidation_reason"]: point.value for point in causes} == { + "sources": 2, + "tags": 1, + } + + +def test_profile_edit_reports_its_invalidation_cause(user_data, observed_runtime): + runtime, _ = observed_runtime + client, *_ = user_data + assert client.put("/v1/user/settings/profile", json={"bio": "Reader"}).status_code == 200 + points = otel_points(runtime, "devfeed.cache.invalidation.causes") + assert [ + (point.attributes["cache.outcome"], point.value) + for point in points + if point.attributes["cache.invalidation_reason"] == "profile" + ] == [("success", 1)] + + +def test_avatar_causes_follow_successful_upload_and_removal( + uploaded_reader, observed_runtime, monkeypatch +): + from devfeed_user_api.avatar_storage import avatar_prefix + + runtime, _ = observed_runtime + client, storage, _, identifier = uploaded_reader + # The avatar fixture reloads settings for its disposable object store. + monkeypatch.setattr(get_settings(), "cache_enabled", True) + path = "/v1/user/settings/profile/avatar" + assert ( + client.post(path, files={"file": ("avatar.png", picture(), "image/png")}).status_code == 200 + ) + storage.fail_next = f"{avatar_prefix(identifier)}/128.webp" + assert ( + client.post(path, files={"file": ("avatar.png", picture(), "image/png")}).status_code == 503 + ) + assert client.delete(path).status_code == 200 + points = otel_points(runtime, "devfeed.cache.invalidation.causes") + assert [ + (point.attributes["cache.outcome"], point.value) + for point in points + if point.attributes["cache.invalidation_reason"] == "avatar" + ] == [("success", 2)] + + +def test_search_index_cause_requires_a_completed_nonempty_batch(database, observed_runtime): + from unittest.mock import Mock + + runtime, _ = observed_runtime + with database.begin() as session: + session.add(Tag(name="Python", slug="python")) + engine = Mock() + engine.sync.side_effect = SearchUnavailable("offline") + with pytest.raises(SearchUnavailable): + sync_batch(database, engine) + engine.sync.side_effect = None + assert sync_batch(database, engine) > 0 + assert sync_batch(database, engine) == 0 + points = otel_points(runtime, "devfeed.cache.invalidation.causes") + assert [ + (point.attributes["cache.outcome"], point.value) + for point in points + if point.attributes["cache.invalidation_reason"] == "search_index" + ] == [("success", 1)] diff --git a/tests/test_observability.py b/tests/test_observability.py index 4fd3c2333..6644ad6c8 100644 --- a/tests/test_observability.py +++ b/tests/test_observability.py @@ -504,6 +504,7 @@ def fail(*args): "cache_unavailable", } assert all(point.attributes["cache.name"] == "public_response" for point in points) + assert all(point.attributes["http.route"] == "/v1/feed" for point in points) assert "private-" not in str([point.attributes for point in points]) @@ -532,6 +533,7 @@ def test_cache_metric_is_shared_and_does_not_label_keys(observed_runtime): with pytest.raises(CacheUnavailable): cache.read_sequence("private-user", 0, 0) record_cache_read("private-key", "private-value", "private-reason") + record_cache_read("public_response", "miss", route="/v1/articles/private-id?code=private-code") points = otel_points(runtime, "devfeed.cache.reads") assert {point.attributes["cache.name"] for point in points} == { "admin_overview", @@ -539,8 +541,280 @@ def test_cache_metric_is_shared_and_does_not_label_keys(observed_runtime): "public_profile", "feed_sequence", "other", + "public_response", } assert "private-" not in str([point.attributes for point in points]) + assert ( + next( + point.attributes["http.route"] + for point in points + if point.attributes["cache.name"] == "public_response" + ) + == "other" + ) + + +def test_public_cache_routes_are_bounded_and_complete(): + from devfeed_api import feed, search, sources, taxonomy, topics + from devfeed_api.cache import CachedReadRoute + from devfeed_core.cache import PUBLIC_CACHE_ROUTES + + assert { + route.path + for module in (feed, search, sources, taxonomy, topics) + for route in module.router.routes + if isinstance(route, CachedReadRoute) and "GET" in route.methods + } == PUBLIC_CACHE_ROUTES + + +def test_cache_route_templates_separate_reads_without_query_or_identifier_labels( + observed_runtime, cached_client +): + import uuid + + from devfeed_core.models import Source, utcnow + + runtime, _ = observed_runtime + client, session, *_ = cached_client + session.get = lambda *args: Source( + id=args[1], + name="Example", + feed_url="https://example.com/rss", + source_type="publisher", + slug="example", + created_at=utcnow(), + approval_status="approved", + enabled=True, + ) + identifiers = [str(uuid.uuid4()) for _ in range(3)] + for identifier in identifiers: + path = f"/v1/sources/{identifier}?unused=private-query" + assert client.get(path).headers["X-Cache"] == "MISS" + assert client.get(path).headers["X-Cache"] == "HIT" + for path in ("/v1/feed", "/v1/tags"): + assert client.get(path).headers["X-Cache"] == "MISS" + assert client.get(path).headers["X-Cache"] == "HIT" + points = otel_points(runtime, "devfeed.cache.reads") + assert sum(point.value for point in points) == 10 + assert {point.attributes["http.route"] for point in points} == { + "/v1/sources/{source_id}", + "/v1/feed", + "/v1/tags", + } + exposition = generate_latest(runtime.metric_server.registry).decode() + assert 'http_route="/v1/sources/{source_id}"' in exposition + assert "private-query" not in exposition + assert not any(identifier in exposition for identifier in identifiers) + + +def test_coalesced_public_cache_reads_count_final_decisions_without_poll_attempts( + observed_runtime, cached_client +): + from concurrent.futures import ThreadPoolExecutor + + runtime, _ = observed_runtime + client, session, *_ = cached_client + original = session.scalars + + def slow(statement): + time.sleep(0.1) + return original(statement) + + session.scalars = slow + with ThreadPoolExecutor(max_workers=8) as pool: + responses = list(pool.map(lambda _: client.get("/v1/feed"), range(8))) + assert sum(response.headers["X-Cache"] == "MISS" for response in responses) == 1 + points = otel_points(runtime, "devfeed.cache.reads") + assert sum(point.value for point in points) == 8 + assert next(point.value for point in points if point.attributes["cache.outcome"] == "hit") == 7 + + +def test_cache_invalidations_count_once_and_deduplicate_bounded_causes( + observed_runtime, response_cache, monkeypatch +): + from devfeed_core.cache import invalidate_public_cache, record_cache_invalidation + from redis.exceptions import ConnectionError + + runtime, _ = observed_runtime + invalidate_public_cache(reasons=("sources", "articles", "articles")) + response_cache.invalidate("operations", reasons=("manual",)) + invalidate_public_cache(reasons=("private-table", "private-other-table")) + invalidate_public_cache(reasons=()) + monkeypatch.setattr(get_settings(), "cache_enabled", False) + invalidate_public_cache(reasons=("profile",)) + monkeypatch.setattr(get_settings(), "cache_enabled", True) + + def fail(*args, **kwargs): + raise ConnectionError("private-redis-credentials") + + monkeypatch.setattr(response_cache.redis, "set", fail) + # Failure is measured but remains suppressed after successful database writes. + invalidate_public_cache(reasons=("avatar",)) + invalidate_public_cache(reasons=("search_index",)) + record_cache_invalidation("private-domain", "private-outcome", ("private-reason",)) + points = otel_points(runtime, "devfeed.cache.invalidations") + assert sum(point.value for point in points) == 8 + assert {point.attributes["cache.outcome"] for point in points} == { + "success", + "cache_unavailable", + "disabled", + "other", + } + assert ( + next( + point.value + for point in points + if point.attributes == {"cache.domain": "public", "cache.outcome": "success"} + ) + == 3 + ) + causes = otel_points(runtime, "devfeed.cache.invalidation.causes") + assert sum(point.value for point in causes) == 9 + assert {point.attributes["cache.invalidation_reason"] for point in causes} == { + "sources", + "articles", + "manual", + "other", + "unspecified", + "profile", + "avatar", + "search_index", + } + exposition = generate_latest(runtime.metric_server.registry).decode() + assert "devfeed_cache_invalidations_total" in exposition + assert "devfeed_cache_invalidation_causes_total" in exposition + assert "private-" not in exposition + + +def test_manual_clear_counts_each_domain_once(observed_runtime, response_cache, monkeypatch): + from devfeed_cli import commands + + runtime, _ = observed_runtime + monkeypatch.setattr(commands, "get_cache", lambda: response_cache) + assert commands.cache_clear(None)["cleared"] + points = otel_points(runtime, "devfeed.cache.invalidations") + assert sum(point.value for point in points) == 2 + assert {point.attributes["cache.domain"] for point in points} == {"public", "operations"} + causes = otel_points(runtime, "devfeed.cache.invalidation.causes") + assert sum(point.value for point in causes) == 2 + assert all(point.attributes["cache.invalidation_reason"] == "manual" for point in causes) + + +@pytest.mark.parametrize("service", ["worker-ingestion", "cli"]) +def test_short_lived_cache_counters_export_deltas_across_flushes_and_lifetimes( + monkeypatch, response_cache, service +): + from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + + from devfeed_core.cache import invalidate_public_cache + from opentelemetry.proto.collector.metrics.v1.metrics_service_pb2 import ( + ExportMetricsServiceRequest, + ) + from opentelemetry.proto.metrics.v1.metrics_pb2 import AGGREGATION_TEMPORALITY_DELTA + + received = [] + + class Collector(BaseHTTPRequestHandler): + def do_POST(self): + message = ExportMetricsServiceRequest() + message.ParseFromString(self.rfile.read(int(self.headers["Content-Length"]))) + received.append(message) + self.send_response(200) + self.end_headers() + + def log_message(self, *args): + pass + + collector = ThreadingHTTPServer(("127.0.0.1", 0), Collector) + thread = threading.Thread(target=collector.serve_forever, daemon=True) + thread.start() + monkeypatch.setenv("DEVFEED_OTLP_ENDPOINT", f"http://127.0.0.1:{collector.server_port}") + monkeypatch.setenv("HOSTNAME", "worker-pod") + monkeypatch.setenv("OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE", "CUMULATIVE") + get_settings.cache_clear() + monkeypatch.setattr(get_settings(), "cache_enabled", True) + try: + for counts in ((1, 2), (3,)): + runtime = telemetry.start_runtime(service, serve_metrics=False, profiling=False) + try: + for count in counts: + for _ in range(count): + invalidate_public_cache(reasons=("articles",)) + if len(counts) > 1: + assert runtime.meter_provider.force_flush(timeout_millis=2000) + finally: + telemetry.stop_runtime(runtime) + for name in ("devfeed.cache.invalidations", "devfeed.cache.invalidation.causes"): + sums = [ + metric.sum + for batch in received + for resource in batch.resource_metrics + for scope in resource.scope_metrics + for metric in scope.metrics + if metric.name == name + ] + assert sums + assert all( + item.aggregation_temporality == AGGREGATION_TEMPORALITY_DELTA for item in sums + ) + assert [ + point.as_int for item in sums for point in item.data_points if point.as_int + ] == [1, 2, 3] + finally: + collector.shutdown() + collector.server_close() + thread.join(timeout=1) + + +def test_parent_cache_counters_keep_cumulative_prometheus_reads(monkeypatch): + from devfeed_core.cache import record_cache_invalidation + + monkeypatch.setattr(get_settings(), "metrics_enabled", True) + runtime = telemetry.Runtime("worker-ingestion", get_settings()) + runtime.start_metrics(serve_metrics=True, export_otlp=False) + monkeypatch.setattr(telemetry, "_runtime", runtime) + try: + for expected in (1, 2): + record_cache_invalidation("public", "success", ("articles",)) + payload = generate_latest(runtime.metric_server.registry).decode() + line = next( + line + for line in payload.splitlines() + if line.startswith("devfeed_cache_invalidations_total{") + ) + assert float(line.rsplit(" ", 1)[1]) == expected + finally: + runtime.close() + + +@pytest.mark.parametrize( + "command", ["cache", "sources", "images", "discovery", "worker", "scheduler", "search"] +) +@pytest.mark.parametrize("reuse", [False, True]) +def test_cli_writers_flush_their_runtime_without_taking_background_ownership( + monkeypatch, command, reuse +): + from types import SimpleNamespace + + from devfeed_cli import runtime + + owned = object() + starts, stops = [], [] + monkeypatch.setattr(runtime, "current", lambda: owned if reuse else None) + + def start(service, **kwargs): + starts.append((service, kwargs)) + return owned + + monkeypatch.setattr(runtime, "start_runtime", start) + monkeypatch.setattr(runtime, "stop_runtime", stops.append) + args = SimpleNamespace(command=command, action="worker", execute=lambda _: None) + assert runtime.execute(args) == 0 + if command in {"cache", "sources", "images", "discovery"} and not reuse: + assert starts == [("cli", {"serve_metrics": False, "profiling": False})] + assert stops == [owned] + else: + assert not starts and not stops @pytest.mark.parametrize("service", ["api", "admin-api", "user-api"])