diff --git a/apps/admin-api/src/dependencies.py b/apps/admin-api/src/dependencies.py index ec649e40f..d3934fbba 100644 --- a/apps/admin-api/src/dependencies.py +++ b/apps/admin-api/src/dependencies.py @@ -9,4 +9,5 @@ # Each service owns its cached clients and dependency identity. get_session = session_dependency(session_factory) get_redis = redis_dependency(get_settings) -DB = Annotated[Session, Depends(get_session, scope="function")] +database_session = Depends(get_session, scope="function") +DB = Annotated[Session, database_session] diff --git a/apps/admin-api/src/main.py b/apps/admin-api/src/main.py index 1ce1e1678..f22bd4a22 100644 --- a/apps/admin-api/src/main.py +++ b/apps/admin-api/src/main.py @@ -2,22 +2,17 @@ import asyncio import logging -from contextlib import asynccontextmanager +from contextlib import AsyncExitStack, asynccontextmanager from devfeed_core.client_lifecycle import close_shared_clients from devfeed_core.config import get_settings as core_settings from devfeed_core.db import database_revision from devfeed_core.logging import configure_logging -from devfeed_core.telemetry import start_runtime, stop_runtime from devfeed_core.version import SCHEMA_REVISION, __version__ -from devfeed_http.admission import AdmissionMiddleware -from devfeed_http.errors import register_error_handlers -from devfeed_http.health import readiness_response -from devfeed_http.logging import RequestLoggingMiddleware -from devfeed_http.schemas import ERROR_RESPONSES, HealthResponse, UnhealthyResponse +from devfeed_http.schemas import ERROR_RESPONSES +from devfeed_http.service import HTTPService from devfeed_http.telemetry import fastapi_telemetry from fastapi import FastAPI -from starlette.concurrency import run_in_threadpool from devfeed_admin_api import ( ai_connection, @@ -44,31 +39,31 @@ ) from devfeed_admin_api.codex_connection import CodexConnection from devfeed_admin_api.config import get_settings -from devfeed_admin_api.dependencies import DB, get_redis +from devfeed_admin_api.dependencies import database_session, get_redis from devfeed_admin_api.reporting import close_reporting logger = logging.getLogger(__name__) def close_clients(): - close_reporting() - close_shared_clients(get_redis) + try: + close_reporting() + finally: + close_shared_clients(get_redis) @asynccontextmanager -async def lifespan(app): - telemetry = start_runtime("admin-api") - logger.info("admin_api_started") - app.state.codex.start() - try: +async def admin_resources(app): + async with AsyncExitStack() as stack: + stack.push_async_callback(app.state.codex.close) + app.state.codex.start() + stack.push_async_callback(overview_panels.close_panel_tasks, app) + stack.push_async_callback(overview.close_snapshot_tasks, app) yield - finally: - await overview.close_snapshot_tasks(app) - await overview_panels.close_panel_tasks(app) - await app.state.codex.close() - await run_in_threadpool(close_clients) - await run_in_threadpool(stop_runtime, telemetry) - logger.info("admin_api_stopped") + + +service = HTTPService("admin-api", logger, lambda: close_clients(), admin_resources) +lifespan = service.lifespan def create_app() -> FastAPI: @@ -91,37 +86,15 @@ def create_app() -> FastAPI: app.state.overview_tasks = {} app.state.overview_retry_at = {} # No cross-origin cookie access: the Next.js admin service proxies same-origin requests. - app.add_middleware( - AdmissionMiddleware, - requests=settings.api_max_concurrent_requests, - streams=settings.api_max_concurrent_streams, - ) - app.add_middleware( - RequestLoggingMiddleware, - service="admin-api", - logger=logger, - response_headers=((b"cache-control", b"no-store"), (b"referrer-policy", b"no-referrer")), - ) - - register_error_handlers(app, logger, admin=True) - - @app.get("/health/live", tags=["health"], response_model=HealthResponse) - async def live(): - return {"status": "ok"} - - @app.get( - "/health/ready", - tags=["health"], - response_model=HealthResponse, - responses={503: {"model": UnhealthyResponse, "description": "Not ready"}}, + service.configure(app, settings, private=True) + + service.register_health( + app, + session_dependency=database_session, + get_redis=lambda: get_redis(), + revision_reader=lambda session: database_revision(session), + schema_revision=SCHEMA_REVISION, ) - def ready(session: DB): - return readiness_response( - session, - get_redis(), - revision_reader=database_revision, - schema_revision=SCHEMA_REVISION, - ) for router in ( auth.router, diff --git a/apps/aggregator/pyproject.toml b/apps/aggregator/pyproject.toml index 7a00da45f..990baa6db 100644 --- a/apps/aggregator/pyproject.toml +++ b/apps/aggregator/pyproject.toml @@ -5,15 +5,16 @@ description = "Scheduled RSS and Atom ingestion workers" requires-python = ">=3.12" dependencies = [ "devfeed-core", - "devfeed-notifications", "rq>=2.8,<3", ] [project.optional-dependencies] +notifications = ["devfeed-notifications"] article-enrichment = ["lingua-language-detector>=2.2,<3", "trafilatura>=2.2,<3"] images = ["boto3>=1.40,<2", "pillow>=11.3,<13", "cairosvg>=2.8,<3"] ai = ["websockets>=15,<18"] all-workers = [ + "devfeed-notifications", "lingua-language-detector>=2.2,<3", "trafilatura>=2.2,<3", "websockets>=15,<18", diff --git a/apps/aggregator/src/analysis_tasks.py b/apps/aggregator/src/analysis_tasks.py index a2e64a470..0cf4f0e63 100644 --- a/apps/aggregator/src/analysis_tasks.py +++ b/apps/aggregator/src/analysis_tasks.py @@ -3,6 +3,9 @@ import logging import time import uuid +from collections.abc import Callable +from dataclasses import dataclass, field +from typing import Protocol from devfeed_core.ai_capacity import CAPACITY_ERRORS, safe_pause from devfeed_core.ai_content import eligible_article @@ -25,17 +28,19 @@ from devfeed_core.analysis_wire import compact_request, restore_identities from devfeed_core.article_jobs import approved_sources from devfeed_core.catalog_cache import reuse_snapshots, snapshot_current -from devfeed_core.config import get_settings +from devfeed_core.config import Settings, get_settings from devfeed_core.db import session_factory from devfeed_core.inference_validation import feedback_prompt, validation_feedback -from devfeed_core.job_lifecycle import finish_job, start_job +from devfeed_core.job_lifecycle import fail_or_retry, finish_job, start_job from devfeed_core.job_logs import job_log_context from devfeed_core.jobs import owned_job from devfeed_core.logging import log_context from devfeed_core.models import Article, ArticleAnalysisJob, ArticleContent, utcnow +from devfeed_core.publication_policy import apply_publication_policy +from devfeed_core.topics import lock_topics from pydantic import ValidationError from sqlalchemy import select -from sqlalchemy.orm import lazyload +from sqlalchemy.orm import Session, lazyload, sessionmaker from devfeed_aggregator.analysis_telemetry import record_attempt from devfeed_aggregator.codex_client import AnalysisError, CodexClient @@ -142,126 +147,225 @@ def _analyze(identifier): snapshot, article_id = job.input_snapshot, job.article_id attempt = job.attempts with log_context(article_id=article_id, attempt=attempt): - _analyze_claimed(settings, factory, identifier, token, snapshot, article_id) + ArticleAnalysisService(settings, factory, CodexClient).run( + AnalysisContext(identifier, token, snapshot, article_id) + ) + + +@dataclass(frozen=True) +class AnalysisContext: + identifier: uuid.UUID + token: uuid.UUID + snapshot: dict + article_id: uuid.UUID + + +class AnalysisClient(Protocol): + operation: str + job_id: uuid.UUID | None + attempt: int | None + reason: str | None + quality_failure: bool + + def complete(self, prompt: str, schema: dict) -> dict: ... + + +@dataclass(frozen=True) +class PreparedAnalysis: + taxonomy: dict + feedback: dict | None + compact: bool + reason: str + + +@dataclass +class AnalysisAttempt: + started: float = field(default_factory=time.perf_counter) + client: AnalysisClient | None = None + number: int | None = None + output: dict | None = None + + +@dataclass(frozen=True) +class ArticleAnalysisService: + """Own one claimed job's I/O and transaction boundaries. + Inference never holds a transaction. Application and retries always verify + the original lease; transaction retries never repeat external inference. + """ -def _analyze_claimed(settings, factory, identifier, token, snapshot, article_id): - logger.info("article_analysis_started") - started, client, attempt = time.perf_counter(), None, None - output = None - try: - # No transaction remains open while waiting for Codex. - with factory() as session: + settings: Settings + factory: sessionmaker[Session] + client_factory: Callable[[Settings], AnalysisClient] + + def run(self, context: AnalysisContext) -> None: + logger.info("article_analysis_started") + attempt = AnalysisAttempt() + try: + prepared = self.prepare(context, attempt) + if prepared is None: + return + result, topic_match_status = self.infer(context, prepared, attempt) + self.propose_topics(context, result) + outcome = _apply_prepared( + self.factory, + context.article_id, + context.snapshot, + lambda session, taxonomy, current: self.apply( + session, context, result, topic_match_status, attempt, taxonomy, current + ), + ) + logger.info("article_analysis_completed", extra={"outcome": outcome}) + except Exception as exc: + self.fail(context, attempt, exc) + finally: + if attempt.client is not None and attempt.number is not None: + record_attempt( + self.factory, + ArticleAnalysisJob, + context.identifier, + attempt.client, + attempt.started, + attempt=attempt.number, + ) + + def prepare( + self, context: AnalysisContext, attempt: AnalysisAttempt + ) -> PreparedAnalysis | None: + with self.factory() as session: inference_catalog = catalog(session) - taxonomy = analysis_candidates(inference_catalog, snapshot) + taxonomy = analysis_candidates(inference_catalog, context.snapshot) del inference_catalog - with factory.begin() as session: - job = owned_job(session, ArticleAnalysisJob, identifier, token) + with self.factory.begin() as session: + job = owned_job(session, ArticleAnalysisJob, context.identifier, context.token) if job is None: logger.warning("article_analysis_lease_lost") - return + return None job.catalog_snapshot = taxonomy job.catalog_hash = snapshot_hash(taxonomy) - attempt = job.attempts + attempt.number = job.attempts feedback = (job.result or {}).get("validation_feedback") - compact = getattr(settings, "ai_compact_article_prompts", False) + compact = getattr(self.settings, "ai_compact_article_prompts", False) job.usage = { **(job.usage or {}), "prompt_format": "compact-evidence-v3" if compact else "original", } - reason = job.usage.get("requested_reason", "queued_analysis") - client = CodexClient(settings) + return PreparedAnalysis( + taxonomy, feedback, compact, job.usage.get("requested_reason", "queued_analysis") + ) + + def infer( + self, context: AnalysisContext, prepared: PreparedAnalysis, attempt: AnalysisAttempt + ) -> tuple[AnalysisResult, str | None]: + client = attempt.client = self.client_factory(self.settings) client.operation = "article_analysis" - client.job_id = identifier - client.attempt = attempt - client.reason = reason - client.quality_failure = bool(feedback) and attempt == 2 - if compact: - prompt, schema, identities = compact_request(snapshot, taxonomy, evidence_refs=True) + client.job_id, client.attempt = context.identifier, attempt.number + client.reason = prepared.reason + client.quality_failure = bool(prepared.feedback) and attempt.number == 2 + if prepared.compact: + prompt, schema, identities = compact_request( + context.snapshot, prepared.taxonomy, evidence_refs=True + ) else: - prompt, schema = analysis_prompt(snapshot, taxonomy), analysis_output_schema(taxonomy) - correction = feedback_prompt(feedback) - if compact: + prompt, schema = ( + analysis_prompt(context.snapshot, prepared.taxonomy), + analysis_output_schema(prepared.taxonomy), + ) + correction = feedback_prompt(prepared.feedback) + if prepared.compact: correction = correction.replace( "exact verbatim evidence", "source passage IDs for evidence" ) - output = client.complete(prompt + correction, schema) - if compact: - output = restore_identities(output, identities) - result = AnalysisResult.model_validate(output) - result, topic_match_status = require_primary_topic(result) + attempt.output = client.complete(prompt + correction, schema) + if prepared.compact: + attempt.output = restore_identities(attempt.output, identities) + result, topic_match_status = require_primary_topic( + AnalysisResult.model_validate(attempt.output) + ) validate_english_ai_prose(result.ai_summary, result.ai_description) - validate_evidence(result, snapshot, taxonomy) - if ( - settings.full_automation + validate_evidence(result, context.snapshot, prepared.taxonomy) + return result, topic_match_status + + def propose_topics(self, context: AnalysisContext, result: AnalysisResult) -> None: + if not ( + self.settings.full_automation and result.developer_relevance == "relevant" and result.page_kind == "article" ): - # Proposal creation precedes application in a short transaction. Holding - # that lock throughout classification would serialize unrelated articles. - from devfeed_core.article_automation import propose_source_topics - - with factory.begin() as session: - job = owned_job(session, ArticleAnalysisJob, identifier, token) - if job is None: - logger.warning("article_analysis_lease_lost") - return - source_ids = approved_sources(session, article_id, lock=True) - article = session.scalar( - select(Article).where(Article.id == article_id).with_for_update(of=Article) - ) - if article is not None and source_ids and article.review_status != "rejected": - propose_source_topics(session, article) + return + from devfeed_core.article_automation import propose_source_topics - def apply_result(session, current_catalog, prepared_snapshot): - job = owned_job(session, ArticleAnalysisJob, identifier, token) + # Release proposal locks before classification and publication. + with self.factory.begin() as session: + job = owned_job(session, ArticleAnalysisJob, context.identifier, context.token) if job is None: logger.warning("article_analysis_lease_lost") return - job.result = result.model_dump(mode="json") - if topic_match_status: - job.result["topic_match_status"] = topic_match_status - job.model = getattr(client, "model", settings.codex_model) - # Lock source review before Article, matching ingestion lock order. - if not approved_sources(session, article_id, lock=True): - finish_job(job, "unapproved", utcnow()) - else: - article = session.scalar( - select(Article).where(Article.id == article_id).with_for_update(of=Article) - ) - if article is None: - finish_job(job, "superseded", utcnow()) - elif not eligible_article(article): - finish_job(job, "content_date_deferred", utcnow()) - else: - # Deleted catalog IDs or changed topic status cannot bypass - # application validation between inference and application. - from devfeed_core.topics import lock_topics - - lock_topics(session, read=True) - if source_snapshot( - article, session.get(ArticleContent, article_id) - ) != prepared_snapshot or not snapshot_current( - session, ("topics", "tags"), current_catalog - ): - raise _ApplicationInputsChanged() - if not analysis_catalog_current(job, current_catalog, snapshot): - finish_job(job, "superseded", utcnow()) - else: - validate_evidence(result, snapshot, current_catalog) - apply_analysis(session, article, job, result) - if job.outcome in {"applied", "insufficient_evidence"}: - from devfeed_core.publication_policy import apply_publication_policy - - apply_publication_policy( - session, article, job, taxonomy=current_catalog - ) - refresh_superseded_analysis(session, article, job, taxonomy=current_catalog) + source_ids = approved_sources(session, context.article_id, lock=True) + article = self.lock_article(session, context) + if article is not None and source_ids and article.review_status != "rejected": + propose_source_topics(session, article) + + @staticmethod + def lock_article(session: Session, context: AnalysisContext) -> Article | None: + return session.scalar( + select(Article).where(Article.id == context.article_id).with_for_update(of=Article) + ) + + def apply( + self, + session: Session, + context: AnalysisContext, + result: AnalysisResult, + topic_match_status: str | None, + attempt: AnalysisAttempt, + current_catalog: dict, + prepared_snapshot: dict | None, + ) -> str | None: + job = owned_job(session, ArticleAnalysisJob, context.identifier, context.token) + if job is None: + logger.warning("article_analysis_lease_lost") + return None + job.result = result.model_dump(mode="json") + if topic_match_status: + job.result["topic_match_status"] = topic_match_status + job.model = getattr(attempt.client, "model", self.settings.codex_model) + # Match ingestion's Source -> Article -> catalog lock order. + if not approved_sources(session, context.article_id, lock=True): + finish_job(job, "unapproved", utcnow()) return job.outcome + article = self.lock_article(session, context) + if article is None: + finish_job(job, "superseded", utcnow()) + elif not eligible_article(article): + finish_job(job, "content_date_deferred", utcnow()) + else: + self.apply_current( + session, context, result, article, job, current_catalog, prepared_snapshot + ) + return job.outcome - outcome = _apply_prepared(factory, article_id, snapshot, apply_result) - logger.info("article_analysis_completed", extra={"outcome": outcome}) - except Exception as exc: + @staticmethod + def apply_current( + session, context, result, article, job, current_catalog, prepared_snapshot + ) -> None: + lock_topics(session, read=True) + if source_snapshot( + article, session.get(ArticleContent, context.article_id) + ) != prepared_snapshot or not snapshot_current( + session, ("topics", "tags"), current_catalog + ): + raise _ApplicationInputsChanged() + if not analysis_catalog_current(job, current_catalog, context.snapshot): + finish_job(job, "superseded", utcnow()) + else: + validate_evidence(result, context.snapshot, current_catalog) + apply_analysis(session, article, job, result) + if job.outcome in {"applied", "insufficient_evidence"}: + apply_publication_policy(session, article, job, taxonomy=current_catalog) + refresh_superseded_analysis(session, article, job, taxonomy=current_catalog) + + def fail(self, context: AnalysisContext, attempt: AnalysisAttempt, exc: Exception) -> None: feedback = validation_feedback(exc) reason = ( str(exc) @@ -275,30 +379,17 @@ def apply_result(session, current_catalog, prepared_snapshot): if reason in CAPACITY_ERRORS else 0 ) - with factory.begin() as session: - job = owned_job(session, ArticleAnalysisJob, identifier, token) + with self.factory.begin() as session: + job = owned_job(session, ArticleAnalysisJob, context.identifier, context.token) if job is not None: - repeated_invalid_output = False - if feedback is not None: - if isinstance(output, dict): - digest = snapshot_hash(output) - repeated_invalid_output = (job.usage or {}).get( - "invalid_output_hash" - ) == digest - job.usage = {**(job.usage or {}), "invalid_output_hash": digest} - job.result = {**(job.result or {}), "validation_feedback": feedback} + repeated = self.save_feedback(job, attempt, feedback) fail_analysis(job, reason, retry_after=cooldown) if ( feedback - and (attempt or 0) >= 2 + and (attempt.number or 0) >= 2 and reason not in CAPACITY_ERRORS - and ( - getattr(settings, "ai_tiered_routing_enabled", False) - or repeated_invalid_output - ) + and (getattr(self.settings, "ai_tiered_routing_enabled", False) or repeated) ): - from devfeed_core.job_lifecycle import fail_or_retry - fail_or_retry(job, reason, utcnow(), retryable=False) logger.warning( "article_analysis_failed", @@ -309,8 +400,15 @@ def apply_result(session, current_catalog, prepared_snapshot): }, exc_info=True, ) - finally: - if client is not None and attempt is not None: - record_attempt( - factory, ArticleAnalysisJob, identifier, client, started, attempt=attempt - ) + + @staticmethod + def save_feedback(job, attempt: AnalysisAttempt, feedback: dict | None) -> bool: + if feedback is None: + return False + repeated = False + if isinstance(attempt.output, dict): + digest = snapshot_hash(attempt.output) + repeated = (job.usage or {}).get("invalid_output_hash") == digest + job.usage = {**(job.usage or {}), "invalid_output_hash": digest} + job.result = {**(job.result or {}), "validation_feedback": feedback} + return repeated diff --git a/apps/aggregator/src/scheduler.py b/apps/aggregator/src/scheduler.py index 9af341662..3576c33fd 100644 --- a/apps/aggregator/src/scheduler.py +++ b/apps/aggregator/src/scheduler.py @@ -176,7 +176,7 @@ def _tick() -> dict[str, int]: notifications_dispatched = notifications_recovered = 0 if get_settings().notifications_enabled: from devfeed_core.feed_notifications import expand_feed_notifications - from devfeed_notifications.delivery import recover_notifications + from devfeed_core.notification_delivery import recover_notifications expand_feed_notifications(factory) diff --git a/apps/aggregator/src/worker.py b/apps/aggregator/src/worker.py index ecd315162..f4800ca49 100644 --- a/apps/aggregator/src/worker.py +++ b/apps/aggregator/src/worker.py @@ -1,4 +1,5 @@ import argparse +import importlib.util import logging import os import uuid @@ -70,6 +71,11 @@ def run( ai_enabled=settings.ai_enabled, notifications_enabled=settings.notifications_enabled, ) + if "notifications" in names and importlib.util.find_spec("devfeed_notifications") is None: + raise OperationConflict( + "Notification delivery requires devfeed-aggregator[notifications] " + "or devfeed-aggregator[all-workers]" + ) if settings.image_storage_enabled and queue_name in {"all", "background"}: # R2 credentials belong only to the dedicated Images pipeline worker. names.remove("images") diff --git a/apps/api/src/dependencies.py b/apps/api/src/dependencies.py index 1af0a34cc..70e3d1eeb 100644 --- a/apps/api/src/dependencies.py +++ b/apps/api/src/dependencies.py @@ -17,4 +17,5 @@ def get_session(request: Request): get_redis = redis_dependency(get_settings) # Return the connection after serialization, before route-cache Redis I/O or # response transmission. Only this public API has no streaming DB consumers. -DB = Annotated[Session, Depends(get_session, scope="function")] +database_session = Depends(get_session, scope="function") +DB = Annotated[Session, database_session] diff --git a/apps/api/src/main.py b/apps/api/src/main.py index 773c89d16..1e2514d9b 100644 --- a/apps/api/src/main.py +++ b/apps/api/src/main.py @@ -1,33 +1,25 @@ import logging -from contextlib import asynccontextmanager from devfeed_core.client_lifecycle import close_shared_clients from devfeed_core.config import get_settings from devfeed_core.db import database_revision from devfeed_core.feeds.validation import FeedValidationError from devfeed_core.logging import configure_logging -from devfeed_core.telemetry import start_runtime, stop_runtime from devfeed_core.version import SCHEMA_REVISION, __version__ -from devfeed_http.admission import AdmissionMiddleware -from devfeed_http.errors import register_error_handlers -from devfeed_http.health import readiness_response -from devfeed_http.logging import RequestLoggingMiddleware from devfeed_http.schemas import ( ERROR_RESPONSES, ErrorResponse, FeedValidationResponse, - HealthResponse, - UnhealthyResponse, VersionResponse, ) +from devfeed_http.service import HTTPService from devfeed_http.telemetry import fastapi_telemetry from fastapi import FastAPI, Request, Response from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse -from starlette.concurrency import run_in_threadpool from devfeed_api import feed, search, sitemaps, sources, taxonomy, topics -from devfeed_api.dependencies import DB, get_redis +from devfeed_api.dependencies import database_session, get_redis logger = logging.getLogger(__name__) @@ -36,18 +28,8 @@ def close_clients(): close_shared_clients(get_redis) -@asynccontextmanager -async def lifespan(app): - telemetry = start_runtime("api") - settings = get_settings() - configure_logging("api", settings.log_level, settings.log_format, non_blocking=True) - logger.info("api_started") - try: - yield - finally: - await run_in_threadpool(close_clients) - await run_in_threadpool(stop_runtime, telemetry) - logger.info("api_stopped") +service = HTTPService("api", logger, lambda: close_clients()) +lifespan = service.lifespan def create_app() -> FastAPI: @@ -61,22 +43,17 @@ def create_app() -> FastAPI: responses={**ERROR_RESPONSES, 422: {"model": ErrorResponse | FeedValidationResponse}}, description="Developer article ingestion, taxonomy and discovery. No account layer.", ) - app.add_middleware( - AdmissionMiddleware, - requests=settings.api_max_concurrent_requests, - streams=settings.api_max_concurrent_streams, - ) - app.add_middleware( - CORSMiddleware, - allow_origins=settings.cors_origins, - allow_methods=["GET", "POST", "PUT", "PATCH"], - allow_headers=["Content-Type"], - allow_credentials=False, - ) - app.add_middleware(RequestLoggingMiddleware, service="api", logger=logger) + def cors(app): + app.add_middleware( + CORSMiddleware, + allow_origins=settings.cors_origins, + allow_methods=["GET", "POST", "PUT", "PATCH"], + allow_headers=["Content-Type"], + allow_credentials=False, + ) - register_error_handlers(app, logger) + service.configure(app, settings, middleware=cors) @app.exception_handler(FeedValidationError) async def invalid_feed(request: Request, exc: FeedValidationError): @@ -87,24 +64,13 @@ async def invalid_feed(request: Request, exc: FeedValidationError): ).model_dump(), ) - @app.get("/health/live", tags=["health"], response_model=HealthResponse) - async def live(): - return {"status": "ok"} - - @app.get( - "/health/ready", - tags=["health"], - response_model=HealthResponse, - responses={503: {"model": UnhealthyResponse, "description": "Not ready"}}, + service.register_health( + app, + session_dependency=database_session, + get_redis=lambda: get_redis(), + revision_reader=lambda session: database_revision(session), + schema_revision=SCHEMA_REVISION, ) - def ready(session: DB): - return readiness_response( - session, - get_redis(), - logger, - revision_reader=database_revision, - schema_revision=SCHEMA_REVISION, - ) @app.get("/version", tags=["operations"], response_model=VersionResponse) def app_version(response: Response): diff --git a/apps/notifications/src/delivery.py b/apps/notifications/src/delivery.py index 18a3298ad..2be753b6c 100644 --- a/apps/notifications/src/delivery.py +++ b/apps/notifications/src/delivery.py @@ -2,50 +2,23 @@ import logging import uuid -from datetime import timedelta import httpx from devfeed_core.db import session_factory -from devfeed_core.job_definitions import JOB_DEFINITIONS from devfeed_core.job_lifecycle import start_job from devfeed_core.job_logs import job_log_context from devfeed_core.jobs import LEASE_SECONDS, owned_job from devfeed_core.models import NotificationDelivery, utcnow +from devfeed_core.notification_delivery import DELIVERY_POLICY +from devfeed_core.notification_delivery import recover_notifications as recover_notifications +from devfeed_core.notification_delivery import retry_delivery as retry_delivery from sqlalchemy import select from devfeed_notifications.config import get_settings logger = logging.getLogger(__name__) -MAX_ATTEMPTS = 20 -# Chimely defaults to 30-day idempotency retention. Never silently re-send a -# delivery that may already exist once that server-side protection has expired. -MAX_DELIVERY_AGE = timedelta(days=28) - - -def retry_delivery(job, error: str, retry_after: int = 0): - job.error = error - job.dispatched_at = job.lease_until = job.lease_token = None - if job.attempts >= MAX_ATTEMPTS or utcnow() - job.created_at >= MAX_DELIVERY_AGE: - job.status, job.finished_at = "failed", utcnow() - else: - job.status = "queued" - delay = min(3600, max(30 * 2 ** min(job.attempts - 1, 7), retry_after)) - job.available_at = utcnow() + timedelta(seconds=delay) - - -def recover_notifications(factory, batch, now): - with factory.begin() as session: - jobs = session.scalars( - select(NotificationDelivery) - .options(*JOB_DEFINITIONS["notifications"].metadata_options()) - .where(NotificationDelivery.status == "running", NotificationDelivery.lease_until < now) - .order_by(NotificationDelivery.lease_until) - .limit(batch) - .with_for_update(skip_locked=True) - ).all() - for job in jobs: - retry_delivery(job, "Delivery worker interrupted; lease recovered") - return len(jobs) +MAX_ATTEMPTS = DELIVERY_POLICY.max_attempts +MAX_DELIVERY_AGE = DELIVERY_POLICY.max_age def deliver_notification(job_id: str): @@ -70,7 +43,7 @@ def _deliver(identifier: uuid.UUID): ) if job is None or job.status != "queued" or job.available_at > utcnow(): return - if utcnow() - job.created_at >= MAX_DELIVERY_AGE: + if DELIVERY_POLICY.expired(job, utcnow()): job.status, job.finished_at = "failed", utcnow() job.error = "Delivery exceeded the safe idempotency window; manual review required" return diff --git a/apps/user-api/src/dependencies.py b/apps/user-api/src/dependencies.py index ec649e40f..d3934fbba 100644 --- a/apps/user-api/src/dependencies.py +++ b/apps/user-api/src/dependencies.py @@ -9,4 +9,5 @@ # Each service owns its cached clients and dependency identity. get_session = session_dependency(session_factory) get_redis = redis_dependency(get_settings) -DB = Annotated[Session, Depends(get_session, scope="function")] +database_session = Depends(get_session, scope="function") +DB = Annotated[Session, database_session] diff --git a/apps/user-api/src/main.py b/apps/user-api/src/main.py index c1fb40056..1eb9d3cd8 100644 --- a/apps/user-api/src/main.py +++ b/apps/user-api/src/main.py @@ -1,22 +1,16 @@ """Independently deployable user API; public traffic never loads this service.""" import logging -from contextlib import asynccontextmanager from devfeed_core.client_lifecycle import close_shared_clients from devfeed_core.config import get_settings as core_settings from devfeed_core.db import database_revision from devfeed_core.logging import configure_logging -from devfeed_core.telemetry import start_runtime, stop_runtime from devfeed_core.version import SCHEMA_REVISION, __version__ -from devfeed_http.admission import AdmissionMiddleware -from devfeed_http.errors import register_error_handlers -from devfeed_http.health import readiness_response -from devfeed_http.logging import RequestLoggingMiddleware -from devfeed_http.schemas import ERROR_RESPONSES, HealthResponse, UnhealthyResponse +from devfeed_http.schemas import ERROR_RESPONSES +from devfeed_http.service import HTTPService from devfeed_http.telemetry import fastapi_telemetry from fastapi import FastAPI -from starlette.concurrency import run_in_threadpool from devfeed_user_api import ( auth, @@ -33,7 +27,7 @@ sources, ) from devfeed_user_api.config import get_settings -from devfeed_user_api.dependencies import DB, get_redis +from devfeed_user_api.dependencies import database_session, get_redis logger = logging.getLogger(__name__) @@ -42,16 +36,8 @@ def close_clients(): close_shared_clients(get_redis) -@asynccontextmanager -async def lifespan(app): - telemetry = start_runtime("user-api") - logger.info("user_api_started") - try: - yield - finally: - await run_in_threadpool(close_clients) - await run_in_threadpool(stop_runtime, telemetry) - logger.info("user_api_stopped") +service = HTTPService("user-api", logger, lambda: close_clients()) +lifespan = service.lifespan def create_app() -> FastAPI: @@ -67,41 +53,20 @@ def create_app() -> FastAPI: description="Optional personalization API. OIDC sessions and CSRF protection required.", ) # No cross-origin cookie access: the Next.js user service proxies same-origin requests. - app.add_middleware( - AdmissionMiddleware, - requests=settings.api_max_concurrent_requests, - streams=settings.api_max_concurrent_streams, + service.configure( + app, + settings, + private=True, + middleware=lambda app: app.add_middleware(avatars.AvatarUploadBodyLimit), ) - app.add_middleware(avatars.AvatarUploadBodyLimit) - app.add_middleware( - RequestLoggingMiddleware, - service="user-api", - logger=logger, - response_headers=( - (b"cache-control", b"no-store"), - (b"referrer-policy", b"no-referrer"), - ), - ) - - register_error_handlers(app, logger, admin=True) - - @app.get("/health/live", tags=["health"], response_model=HealthResponse) - async def live(): - return {"status": "ok"} - @app.get( - "/health/ready", - tags=["health"], - response_model=HealthResponse, - responses={503: {"model": UnhealthyResponse, "description": "Not ready"}}, + service.register_health( + app, + session_dependency=database_session, + get_redis=lambda: get_redis(), + revision_reader=lambda session: database_revision(session), + schema_revision=SCHEMA_REVISION, ) - def ready(session: DB): - return readiness_response( - session, - get_redis(), - revision_reader=database_revision, - schema_revision=SCHEMA_REVISION, - ) app.include_router(auth.router) app.include_router(mcp.router) diff --git a/packages/core/src/client_lifecycle.py b/packages/core/src/client_lifecycle.py index 040788fd5..f422a4ded 100644 --- a/packages/core/src/client_lifecycle.py +++ b/packages/core/src/client_lifecycle.py @@ -1,5 +1,6 @@ """Shutdown shared cache, database, and Redis clients without opening them.""" +from contextlib import ExitStack from typing import Protocol from sqlalchemy.engine import Engine @@ -23,10 +24,18 @@ def cache_clear(self) -> None: ... def close_shared_clients(redis_provider: CachedRedisProvider) -> None: """Close initialized shared clients; preserve each app's Redis provider.""" - close_cache() - if get_engine.cache_info().currsize: - engine: Engine = get_engine() - engine.dispose() - if redis_provider.cache_info().currsize: - redis_provider().close() - redis_provider.cache_clear() + + def close_engine(): + if get_engine.cache_info().currsize: + engine: Engine = get_engine() + engine.dispose() + + def close_redis(): + if redis_provider.cache_info().currsize: + redis_provider().close() + + with ExitStack() as stack: + stack.callback(redis_provider.cache_clear) + stack.callback(close_redis) + stack.callback(close_engine) + stack.callback(close_cache) diff --git a/packages/core/src/job_retries.py b/packages/core/src/job_retries.py index 5cd942000..b45402e30 100644 --- a/packages/core/src/job_retries.py +++ b/packages/core/src/job_retries.py @@ -1,7 +1,5 @@ """Explicit retries use current eligibility and the pipelines' coalescing requests.""" -from datetime import timedelta - from sqlalchemy import and_, case, or_, select, tuple_ from sqlalchemy.orm import aliased @@ -10,6 +8,7 @@ from devfeed_core.config import get_settings from devfeed_core.image_jobs import retry_image from devfeed_core.models import ArticleImageJob, NotificationDelivery, TopicAnalysisJob, utcnow +from devfeed_core.notification_delivery import DELIVERY_POLICY from devfeed_core.services import OperationConflict, RecordNotFound, retry_job from devfeed_core.source_enrichment import request_enrichment from devfeed_core.topic_analysis import request_topic_analysis @@ -77,7 +76,7 @@ def retry_notification(session, identifier): raise RecordNotFound("Notification delivery not found") if job.status != "failed": raise OperationConflict("Only failed deliveries can be retried") - if utcnow() - job.created_at >= timedelta(days=28): + if DELIVERY_POLICY.expired(job, utcnow()): raise OperationConflict( "Delivery is beyond the safe retry window; review it before publishing a new event" ) diff --git a/packages/core/src/notification_delivery.py b/packages/core/src/notification_delivery.py new file mode 100644 index 000000000..0a08d8c53 --- /dev/null +++ b/packages/core/src/notification_delivery.py @@ -0,0 +1,55 @@ +"""Durable delivery policy; scheduler and admin never load the HTTP adapter.""" + +from dataclasses import dataclass +from datetime import datetime, timedelta + +from sqlalchemy import select + +from devfeed_core.job_definitions import JOB_DEFINITIONS +from devfeed_core.job_lifecycle import clear_lease +from devfeed_core.models import NotificationDelivery, utcnow + + +@dataclass(frozen=True) +class DeliveryPolicy: + max_attempts: int = 20 + # Leave a margin below Chimely's 30-day idempotency retention. + max_age: timedelta = timedelta(days=28) + + def expired(self, job: NotificationDelivery, now: datetime) -> bool: + return now - job.created_at >= self.max_age + + def retry( + self, job: NotificationDelivery, error: str, now: datetime, retry_after: int = 0 + ) -> None: + job.error = error + job.dispatched_at = None + clear_lease(job) + if job.attempts >= self.max_attempts or self.expired(job, now): + job.status, job.finished_at = "failed", now + else: + job.status = "queued" + delay = min(3600, max(30 * 2 ** min(job.attempts - 1, 7), retry_after)) + job.available_at = now + timedelta(seconds=delay) + + +DELIVERY_POLICY = DeliveryPolicy() + + +def retry_delivery(job, error: str, retry_after: int = 0): + DELIVERY_POLICY.retry(job, error, utcnow(), retry_after) + + +def recover_notifications(factory, batch, now): + with factory.begin() as session: + jobs = session.scalars( + select(NotificationDelivery) + .options(*JOB_DEFINITIONS["notifications"].metadata_options()) + .where(NotificationDelivery.status == "running", NotificationDelivery.lease_until < now) + .order_by(NotificationDelivery.lease_until) + .limit(batch) + .with_for_update(skip_locked=True) + ).all() + for job in jobs: + DELIVERY_POLICY.retry(job, "Delivery worker interrupted; lease recovered", now) + return len(jobs) diff --git a/packages/http/src/service.py b/packages/http/src/service.py new file mode 100644 index 000000000..c9af4b335 --- /dev/null +++ b/packages/http/src/service.py @@ -0,0 +1,103 @@ +"""Process lifecycle and HTTP policy shared by independently deployed APIs.""" + +import logging +from collections.abc import Callable +from contextlib import AbstractAsyncContextManager, AsyncExitStack, asynccontextmanager +from dataclasses import dataclass +from typing import Annotated + +from devfeed_core.config import Settings +from devfeed_core.telemetry import start_runtime, stop_runtime +from devfeed_core.version import SCHEMA_REVISION +from fastapi import FastAPI +from fastapi.params import Depends +from sqlalchemy.orm import Session +from starlette.concurrency import run_in_threadpool + +from devfeed_http.admission import AdmissionMiddleware +from devfeed_http.errors import register_error_handlers +from devfeed_http.health import readiness_response +from devfeed_http.logging import RequestLoggingMiddleware +from devfeed_http.schemas import HealthResponse, UnhealthyResponse + + +@dataclass(frozen=True) +class HTTPService: + """Compose process infrastructure without inheriting account or admin policy. + + Each app supplies its own clients, dependencies and optional resources. + Cleanup runs in reverse acquisition order, including after startup failures. + """ + + name: str + logger: logging.Logger + close_clients: Callable[[], None] + resources: Callable[[FastAPI], AbstractAsyncContextManager] | None = None + + @asynccontextmanager + async def lifespan(self, app: FastAPI): + runtime = start_runtime(self.name) + async with AsyncExitStack() as stack: + event_prefix = self.name.replace("-", "_") + stack.callback(self.logger.info, event_prefix + "_stopped") + stack.push_async_callback(run_in_threadpool, stop_runtime, runtime) + stack.push_async_callback(run_in_threadpool, self.close_clients) + self.logger.info(event_prefix + "_started") + if self.resources is not None: + await stack.enter_async_context(self.resources(app)) + yield + + def configure( + self, + app: FastAPI, + settings: Settings, + *, + private: bool = False, + middleware: Callable[[FastAPI], None] | None = None, + ) -> None: + app.add_middleware( + AdmissionMiddleware, + requests=settings.api_max_concurrent_requests, + streams=settings.api_max_concurrent_streams, + ) + if middleware is not None: + middleware(app) + app.add_middleware( + RequestLoggingMiddleware, + service=self.name, + logger=self.logger, + response_headers=( + ((b"cache-control", b"no-store"), (b"referrer-policy", b"no-referrer")) + if private + else () + ), + ) + register_error_handlers(app, self.logger, admin=private) + + def register_health( + self, + app: FastAPI, + *, + session_dependency: Depends, + get_redis: Callable, + revision_reader: Callable, + schema_revision: str = SCHEMA_REVISION, + ) -> None: + @app.get("/health/live", tags=["health"], response_model=HealthResponse) + async def live(): + return {"status": "ok"} + + @app.get( + "/health/ready", + tags=["health"], + response_model=HealthResponse, + responses={503: {"model": UnhealthyResponse, "description": "Not ready"}}, + ) + def ready(session: Annotated[Session, session_dependency]): + return readiness_response( + session, + get_redis(), + self.logger, + revision_reader=revision_reader, + schema_revision=schema_revision, + ) diff --git a/tests/test_analysis_tasks.py b/tests/test_analysis_tasks.py index 41f50f204..c5546f387 100644 --- a/tests/test_analysis_tasks.py +++ b/tests/test_analysis_tasks.py @@ -8,6 +8,7 @@ from devfeed_aggregator.codex_client import AnalysisError from devfeed_core import analysis from devfeed_core.models import Article, ArticleAnalysisJob, utcnow +from sqlalchemy.exc import OperationalError TOPIC_ID = uuid.uuid4() SUPPORTED_TOPIC = { @@ -114,6 +115,75 @@ def test_worker_claims_waiting_lock_and_persists_analysis_without_publication(ru assert "FOR UPDATE" in str(statements[0]) and "SKIP LOCKED" not in str(statements[0]) +def test_application_deadlock_retry_never_repeats_inference_or_holds_a_transaction( + runtime, monkeypatch +): + article, job, _, _ = runtime + factory = analysis_tasks.session_factory() + state = {"transactions": 0, "inferences": 0, "applies": 0} + original_apply = analysis_tasks.apply_analysis + client = analysis_tasks.CodexClient(None) + original_complete = client.complete + + class TrackingFactory: + @contextmanager + def begin(self): + state["transactions"] += 1 + try: + with factory.begin() as session: + yield session + finally: + state["transactions"] -= 1 + + __call__ = begin + + def complete(*args): + assert state["transactions"] == 0 + state["inferences"] += 1 + return original_complete(*args) + + class Deadlock(Exception): + sqlstate = "40P01" + + def apply(*args): + assert state["transactions"] == 1 + state["applies"] += 1 + if state["applies"] == 1: + raise OperationalError("test", {}, Deadlock()) + original_apply(*args) + + monkeypatch.setattr(analysis_tasks, "session_factory", TrackingFactory) + monkeypatch.setattr(analysis_tasks, "CodexClient", lambda _: client) + monkeypatch.setattr(client, "complete", complete) + monkeypatch.setattr(analysis_tasks, "apply_analysis", apply) + analysis_tasks._analyze(job.id) + assert state == {"transactions": 0, "inferences": 1, "applies": 2} + assert job.outcome == "applied" and article.ai_summary == "Generated preview" + assert list(job.usage["attempts"]) == ["1"] + + +@pytest.mark.parametrize("invalid_output", [False, True]) +def test_lost_lease_prevents_both_result_application_and_failure_updates( + runtime, monkeypatch, invalid_output +): + article, job, _, _ = runtime + telemetry = [] + original_complete = analysis_tasks.CodexClient(None).complete + + def complete(*args): + job.lease_token = uuid.uuid4() # Another process recovered this lease. + if invalid_output: + raise ValueError("invalid") + return original_complete(*args) + + monkeypatch.setattr(analysis_tasks, "CodexClient", lambda _: SimpleNamespace(complete=complete)) + monkeypatch.setattr(analysis_tasks, "record_attempt", lambda *args, **kw: telemetry.append(kw)) + analysis_tasks._analyze(job.id) + assert job.status == "running" and job.result is None and job.error is None + assert article.ai_summary is None and article.review_status == "pending" + assert telemetry == [{"attempt": 1}] + + def test_worker_keeps_relevant_analysis_incomplete_without_a_primary_topic(runtime, monkeypatch): article, job, _, _ = runtime monkeypatch.setattr( diff --git a/tests/test_backend_boundaries.py b/tests/test_backend_boundaries.py new file mode 100644 index 000000000..81476cd77 --- /dev/null +++ b/tests/test_backend_boundaries.py @@ -0,0 +1,74 @@ +"""Keep deployment owners separate while permitting explicit worker/CLI adapters.""" + +import ast +import json +import os +import subprocess +import sys +import tomllib +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +# These apps are launchers/operators for existing worker entry points. Domain +# libraries and HTTP services must never depend on another application package. +ADAPTER_IMPORTS = { + "devfeed_cli": {"devfeed_aggregator", "devfeed_search_indexer"}, + "devfeed_article_enrichment_worker": {"devfeed_aggregator"}, + "devfeed_images_worker": {"devfeed_aggregator"}, + "devfeed_source_discovery_worker": {"devfeed_aggregator"}, +} + + +def test_backend_imports_follow_service_ownership(): + violations = [] + workspace = tomllib.loads((ROOT / "pyproject.toml").read_text())["tool"]["uv"]["workspace"] + roots = [ROOT / member / "src" for member in workspace["members"]] + for root in roots: + configuration = tomllib.loads((root.parent / "pyproject.toml").read_text()) + package_dirs = configuration.get("tool", {}).get("setuptools", {}).get("package-dir", {}) + if not package_dirs: + continue + owner = next(iter(package_dirs)) + allowed = {owner, "devfeed_core"} + if owner != "devfeed_core": + allowed.add("devfeed_http") + allowed.update(ADAPTER_IMPORTS.get(owner, set())) + for path in root.rglob("*.py"): + for node in ast.walk(ast.parse(path.read_text())): + names = ( + [node.module] + if isinstance(node, ast.ImportFrom) and node.module + else [alias.name for alias in node.names] + if isinstance(node, ast.Import) + else [] + ) + for name in names: + target = name.split(".")[0] + if target.startswith("devfeed_") and target not in allowed: + violations.append(f"{path.relative_to(ROOT)}:{node.lineno}: {target}") + assert not violations, "Application coupling:\n" + "\n".join(violations) + + +@pytest.mark.parametrize("package", ["devfeed_api", "devfeed_admin_api", "devfeed_user_api"]) +def test_loading_an_api_does_not_load_another_application(package, tmp_path): + code = ( + "import importlib,json,sys; " + f"importlib.import_module('{package}.main'); " + "print(json.dumps(sorted({name.split('.')[0] for name in sys.modules " + "if name.startswith('devfeed_')})))" + ) + result = subprocess.run( + [sys.executable, "-c", code], + capture_output=True, + text=True, + check=True, + cwd=tmp_path, + env={ + **os.environ, + "DEVFEED_DATABASE_URL": "postgresql+psycopg://ci@database.invalid/ci", + "DEVFEED_REDIS_URL": "redis://redis.invalid/15", + }, + ) + assert set(json.loads(result.stdout)) <= {package, "devfeed_core", "devfeed_http"} diff --git a/tests/test_client_lifecycle.py b/tests/test_client_lifecycle.py index 89a2c5a4a..16f299517 100644 --- a/tests/test_client_lifecycle.py +++ b/tests/test_client_lifecycle.py @@ -1,5 +1,6 @@ from types import SimpleNamespace +import pytest from devfeed_core import client_lifecycle @@ -45,3 +46,22 @@ def test_close_shared_clients_does_not_create_unused_clients(monkeypatch): client_lifecycle.close_shared_clients(redis_provider) assert events == ["cache", "clear"] + + +@pytest.mark.parametrize("failure", ["cache", "engine", "redis"]) +def test_one_failed_close_cannot_leave_other_clients_open(failure, monkeypatch): + events = [] + + def close(name): + events.append(name) + if name == failure: + raise RuntimeError(name) + + engine_provider = CachedProvider(SimpleNamespace(dispose=lambda: close("engine")), events) + redis_provider = CachedProvider(SimpleNamespace(close=lambda: close("redis")), events) + monkeypatch.setattr(client_lifecycle, "close_cache", lambda: close("cache")) + monkeypatch.setattr(client_lifecycle, "get_engine", engine_provider) + with pytest.raises(RuntimeError, match=failure): + client_lifecycle.close_shared_clients(redis_provider) + assert events == ["cache", "get", "engine", "get", "redis", "clear"] + assert not redis_provider.cached diff --git a/tests/test_http_service.py b/tests/test_http_service.py new file mode 100644 index 000000000..b2df15362 --- /dev/null +++ b/tests/test_http_service.py @@ -0,0 +1,186 @@ +"""Process ownership survives partial startup and failed cleanup.""" + +import asyncio +import importlib +import threading +from contextlib import asynccontextmanager +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest +from devfeed_http import service as runtime +from fastapi import FastAPI +from fastapi.responses import JSONResponse + + +@pytest.mark.parametrize("api", ["devfeed_api", "devfeed_user_api", "devfeed_admin_api"]) +@pytest.mark.parametrize("healthy", [True, False]) +def test_readiness_releases_overridden_database_session_before_response(api, healthy, monkeypatch): + module = importlib.import_module(f"{api}.main") + dependencies = importlib.import_module(f"{api}.dependencies") + session, redis = object(), object() + events = [] + + def get_session(): + events.append("opened") + try: + yield session + finally: + events.append("closed") + + def readiness(current_session, current_redis, *args, **kwargs): + assert current_session is session + assert current_redis is redis + assert events == ["opened"] + if healthy: + return {"status": "ok"} + return JSONResponse({"status": "unhealthy"}, status_code=503) + + monkeypatch.setattr(module, "get_redis", lambda: redis) + monkeypatch.setattr(runtime, "readiness_response", readiness) + app = module.create_app() + app.dependency_overrides[dependencies.get_session] = get_session + responses = [] + + async def receive(): + return {"type": "http.request", "body": b"", "more_body": False} + + async def send(message): + assert events == ["opened", "closed"] + responses.append(message) + + asyncio.run( + app( + { + "type": "http", + "asgi": {"version": "3.0", "spec_version": "2.3"}, + "http_version": "1.1", + "method": "GET", + "scheme": "http", + "path": "/health/ready", + "raw_path": b"/health/ready", + "query_string": b"", + "root_path": "", + "headers": [(b"host", b"localhost")], + "client": ("127.0.0.1", 12345), + "server": ("localhost", 80), + }, + receive, + send, + ) + ) + assert responses[0]["type"] == "http.response.start" + assert responses[0]["status"] == (200 if healthy else 503) + + +@pytest.mark.parametrize( + "failure", [None, "startup", "request", "resources", "clients", "telemetry"] +) +def test_lifecycle_always_releases_resources_clients_and_telemetry(failure, monkeypatch): + events, blocking_threads = [], [] + telemetry = object() + monkeypatch.setattr(runtime, "start_runtime", lambda name: telemetry) + + def close_clients(): + blocking_threads.append(threading.get_ident()) + events.append("clients") + if failure == "clients": + raise RuntimeError("clients") + + def stop(token): + assert token is telemetry + blocking_threads.append(threading.get_ident()) + events.append("telemetry") + if failure == "telemetry": + raise RuntimeError("telemetry") + + monkeypatch.setattr(runtime, "stop_runtime", stop) + + @asynccontextmanager + async def resources(app): + events.append("start") + try: + if failure == "startup": + raise RuntimeError("startup") + yield + finally: + events.append("resources") + if failure == "resources": + raise RuntimeError("resources") + + logger = Mock() + service = runtime.HTTPService("test-api", logger, close_clients, resources) + + async def scenario(): + loop_thread = threading.get_ident() + async with service.lifespan(FastAPI()): + events.append("request") + if failure == "request": + raise RuntimeError("request") + assert all(thread != loop_thread for thread in blocking_threads) + + if failure: + with pytest.raises(RuntimeError, match=failure): + asyncio.run(scenario()) + else: + asyncio.run(scenario()) + assert events[-3:] == ["resources", "clients", "telemetry"] + assert len(blocking_threads) == 2 + logger.info.assert_any_call("test_api_started") + logger.info.assert_any_call("test_api_stopped") + + +@pytest.mark.parametrize("failure", ["start", "snapshots", "panels", "codex"]) +def test_admin_owned_resources_cannot_prevent_process_cleanup(failure, monkeypatch): + module = importlib.import_module("devfeed_admin_api.main") + app = module.create_app() + events = [] + monkeypatch.setattr(runtime, "start_runtime", lambda name: None) + monkeypatch.setattr(runtime, "stop_runtime", lambda token: events.append("telemetry")) + monkeypatch.setattr(module, "close_clients", lambda: events.append("clients")) + + def start(): + events.append("start") + if failure == "start": + raise RuntimeError("start") + + async def close(name, *args): + events.append(name) + if failure == name: + raise RuntimeError(name) + + app.state.codex = SimpleNamespace(start=start, close=lambda: close("codex")) + monkeypatch.setattr(module.overview, "close_snapshot_tasks", lambda app: close("snapshots")) + monkeypatch.setattr(module.overview_panels, "close_panel_tasks", lambda app: close("panels")) + + async def scenario(): + async with module.lifespan(app): + events.append("request") + + with pytest.raises(RuntimeError, match=failure): + asyncio.run(scenario()) + assert events[-3:] == ["codex", "clients", "telemetry"] + if failure != "start": + assert events == [ + "start", + "request", + "snapshots", + "panels", + "codex", + "clients", + "telemetry", + ] + + +def test_failed_reporting_cleanup_still_closes_admin_shared_clients(monkeypatch): + module = importlib.import_module("devfeed_admin_api.main") + closed = [] + + def fail(): + raise RuntimeError("reporting") + + monkeypatch.setattr(module, "close_reporting", fail) + monkeypatch.setattr(module, "close_shared_clients", closed.append) + with pytest.raises(RuntimeError, match="reporting"): + module.close_clients() + assert closed == [module.get_redis] diff --git a/tests/test_notification_delivery_policy.py b/tests/test_notification_delivery_policy.py new file mode 100644 index 000000000..c49ebdf51 --- /dev/null +++ b/tests/test_notification_delivery_policy.py @@ -0,0 +1,92 @@ +"""Delivery recovery, delivery attempts and manual retries share one safety window.""" + +import uuid +from datetime import timedelta + +import pytest +from devfeed_core.models import NotificationDelivery, utcnow +from devfeed_core.notification_delivery import DELIVERY_POLICY, recover_notifications +from sqlalchemy import select + + +@pytest.mark.parametrize( + "age", [timedelta(days=28) - timedelta(microseconds=1), timedelta(days=28)] +) +def test_delivery_expiry_has_an_exact_inclusive_boundary(age): + now = utcnow() + job = NotificationDelivery(created_at=now - age) + assert DELIVERY_POLICY.expired(job, now) == (age >= timedelta(days=28)) + + +@pytest.mark.parametrize( + "attempts,hint,delay", [(1, 0, 30), (2, 120, 120), (19, 0, 3600), (2, 9000, 3600)] +) +def test_retry_keeps_identity_and_clears_the_owned_lease(attempts, hint, delay): + now, identifier = utcnow(), uuid.uuid4() + job = NotificationDelivery( + id=identifier, + status="running", + attempts=attempts, + created_at=now, + lease_token=uuid.uuid4(), + lease_until=now + timedelta(minutes=5), + dispatched_at=now, + dedup_key="same-delivery", + ) + DELIVERY_POLICY.retry(job, "unavailable", now, hint) + assert job.id == identifier and job.dedup_key == "same-delivery" + assert job.status == "queued" and job.error == "unavailable" + assert job.available_at == now + timedelta(seconds=delay) + assert job.lease_token is job.lease_until is job.dispatched_at is None + + +@pytest.mark.parametrize("age,attempts", [(28, 1), (0, 20)]) +def test_retry_exhaustion_requires_manual_review(age, attempts): + now = utcnow() + job = NotificationDelivery(created_at=now - timedelta(days=age), attempts=attempts) + DELIVERY_POLICY.retry(job, "unavailable", now) + assert job.status == "failed" and job.finished_at == now + + +@pytest.mark.integration +def test_recovery_is_bounded_skips_locked_rows_and_preserves_payload(database): + now = utcnow() + identifiers = [] + with database.begin() as session: + for index, (age, attempts, lease) in enumerate( + [(0, 1, -4), (28, 1, -3), (0, 20, -2), (0, 1, 1)] + ): + job = NotificationDelivery( + event_key=f"event-{index}", + dedup_key=f"delivery-{index}", + audience="admin", + category="product.release", + payload={"private": "never needed for recovery"}, + status="running", + attempts=attempts, + created_at=now - timedelta(days=age), + lease_until=now + timedelta(minutes=lease), + lease_token=uuid.uuid4(), + dispatched_at=now, + ) + session.add(job) + session.flush() + identifiers.append(job.id) + # A delivery owned by another transaction must not delay other recovery. + with database.begin() as locked: + locked.scalar( + select(NotificationDelivery) + .where(NotificationDelivery.id == identifiers[0]) + .with_for_update() + ) + assert recover_notifications(database, 1, now) == 1 + with database() as session: + assert session.get(NotificationDelivery, identifiers[1]).status == "failed" + assert session.get(NotificationDelivery, identifiers[2]).status == "running" + assert recover_notifications(database, 10, now) == 2 + assert recover_notifications(database, 10, now) == 0 + with database() as session: + jobs = [session.get(NotificationDelivery, identifier) for identifier in identifiers] + assert [job.status for job in jobs] == ["queued", "failed", "failed", "running"] + assert all(job.payload == {"private": "never needed for recovery"} for job in jobs) + assert jobs[0].available_at == now + timedelta(seconds=30) diff --git a/tests/test_notifications.py b/tests/test_notifications.py index 045d3c0cd..f53fac2bb 100644 --- a/tests/test_notifications.py +++ b/tests/test_notifications.py @@ -370,3 +370,17 @@ def test_relationship_research_has_its_own_event_preferences(): event = notifications.job_notification(job) assert event.category == "jobs.relationship-research.success" assert event.action_url == f"/jobs/analysis/topics/{job.id}" + + +def test_lightweight_worker_rejects_missing_delivery_adapter_before_dequeue(monkeypatch): + from devfeed_aggregator import worker + from devfeed_core.services import OperationConflict + + monkeypatch.setattr(worker.importlib.util, "find_spec", lambda name: None) + monkeypatch.setattr(worker, "get_queue", lambda *args: pytest.fail("Opened a queue")) + stopped = [] + monkeypatch.setattr(worker, "start_runtime", lambda *args, **kw: "runtime") + monkeypatch.setattr(worker, "stop_runtime", stopped.append) + with pytest.raises(OperationConflict, match=r"devfeed-aggregator\[notifications\]"): + worker.run(burst=True, queue_name="notifications") + assert stopped == ["runtime"] diff --git a/tests/test_scheduler_unit.py b/tests/test_scheduler_unit.py index e399bd49a..7b96ba137 100644 --- a/tests/test_scheduler_unit.py +++ b/tests/test_scheduler_unit.py @@ -10,8 +10,7 @@ import pytest from devfeed_aggregator import quota_monitor, scheduler -from devfeed_core import feed_notifications, job_retention -from devfeed_notifications import delivery +from devfeed_core import feed_notifications, job_retention, notification_delivery NOW = datetime(2026, 10, 4, tzinfo=UTC) @@ -174,7 +173,7 @@ def transaction(): monkeypatch.setattr(scheduler, "request_ingestion", Mock(return_value=NS(id=uuid.UUID(int=5)))) monkeypatch.setattr(scheduler, "recover_jobs", Mock(return_value=1)) monkeypatch.setattr(feed_notifications, "expand_feed_notifications", Mock()) - monkeypatch.setattr(delivery, "recover_notifications", Mock(return_value=2)) + monkeypatch.setattr(notification_delivery, "recover_notifications", Mock(return_value=2)) monkeypatch.setattr(scheduler, "dispatch_discovery", Mock()) monkeypatch.setattr(scheduler, "dispatch_recommendations", Mock(return_value=3)) queues = [] diff --git a/uv.lock b/uv.lock index abf465c33..57d6f8991 100644 --- a/uv.lock +++ b/uv.lock @@ -876,7 +876,6 @@ version = "0.0.50" source = { editable = "apps/aggregator" } dependencies = [ { name = "devfeed-core" }, - { name = "devfeed-notifications" }, { name = "rq" }, ] @@ -887,6 +886,7 @@ ai = [ all-workers = [ { name = "boto3" }, { name = "cairosvg" }, + { name = "devfeed-notifications" }, { name = "lingua-language-detector" }, { name = "pillow" }, { name = "trafilatura" }, @@ -901,6 +901,9 @@ images = [ { name = "cairosvg" }, { name = "pillow" }, ] +notifications = [ + { name = "devfeed-notifications" }, +] [package.metadata] requires-dist = [ @@ -909,7 +912,8 @@ requires-dist = [ { name = "cairosvg", marker = "extra == 'all-workers'", specifier = ">=2.8,<3" }, { name = "cairosvg", marker = "extra == 'images'", specifier = ">=2.8,<3" }, { name = "devfeed-core", editable = "packages/core" }, - { name = "devfeed-notifications", editable = "apps/notifications" }, + { name = "devfeed-notifications", marker = "extra == 'all-workers'", editable = "apps/notifications" }, + { name = "devfeed-notifications", marker = "extra == 'notifications'", editable = "apps/notifications" }, { name = "lingua-language-detector", marker = "extra == 'all-workers'", specifier = ">=2.2,<3" }, { name = "lingua-language-detector", marker = "extra == 'article-enrichment'", specifier = ">=2.2,<3" }, { name = "pillow", marker = "extra == 'all-workers'", specifier = ">=11.3,<13" }, @@ -920,7 +924,7 @@ requires-dist = [ { name = "websockets", marker = "extra == 'ai'", specifier = ">=15,<18" }, { name = "websockets", marker = "extra == 'all-workers'", specifier = ">=15,<18" }, ] -provides-extras = ["article-enrichment", "images", "ai", "all-workers"] +provides-extras = ["notifications", "article-enrichment", "images", "ai", "all-workers"] [[package]] name = "devfeed-api"