diff --git a/.env.example b/.env.example index 5107a14..ec10203 100644 --- a/.env.example +++ b/.env.example @@ -46,6 +46,10 @@ LOG_REQUEST_SKIP_PATHS=/healthz LOG_REDACT_FIELDS= SCHEDULER_ENABLED=1 SCHEDULER_TICK_SECONDS=60 +INGESTION_AUTOMATION_ALLOWED=1 +INGESTION_MANUAL_RUN_ALLOWED=1 +INGESTION_MIN_RUN_INTERVAL_MINUTES=15 +INGESTION_MIN_REQUEST_DELAY_SECONDS=2 INGESTION_NETWORK_ERROR_RETRIES=1 INGESTION_RETRY_BACKOFF_SECONDS=1.0 INGESTION_MAX_PAGES_PER_SCHOLAR=30 @@ -53,6 +57,8 @@ INGESTION_PAGE_SIZE=100 INGESTION_ALERT_BLOCKED_FAILURE_THRESHOLD=1 INGESTION_ALERT_NETWORK_FAILURE_THRESHOLD=2 INGESTION_ALERT_RETRY_SCHEDULED_THRESHOLD=3 +INGESTION_SAFETY_COOLDOWN_BLOCKED_SECONDS=1800 +INGESTION_SAFETY_COOLDOWN_NETWORK_SECONDS=900 INGESTION_CONTINUATION_QUEUE_ENABLED=1 INGESTION_CONTINUATION_BASE_DELAY_SECONDS=120 INGESTION_CONTINUATION_MAX_DELAY_SECONDS=3600 diff --git a/README.md b/README.md index 4489432..56ce6ce 100644 --- a/README.md +++ b/README.md @@ -70,6 +70,7 @@ docker compose -f docker-compose.yml -f docker-compose.dev.yml down - `Dockerfile`: multi-stage build (frontend + backend runtime). - `README.md`: deployment and operations reference. - `CONTRIBUTING.md`: contribution policy and merge checklist. +- `docs/ops/scrape_safety_runbook.md`: scrape cooldown and blocked-IP operations guide. - `scripts/check_no_generated_artifacts.sh`: tracked-artifact guard used by CI. ## Data Model Notes @@ -170,6 +171,10 @@ Notes: | --- | --- | --- | --- | --- | | `SCHEDULER_ENABLED` | `1` | boolean | deploy, dev | Enable background scheduler loop. | | `SCHEDULER_TICK_SECONDS` | `60` | integer >= 1 | deploy, dev | Scheduler interval in seconds. | +| `INGESTION_AUTOMATION_ALLOWED` | `1` | boolean | deploy, dev | Global safety switch for scheduled/automatic checks. When disabled, auto-run settings are forced off. | +| `INGESTION_MANUAL_RUN_ALLOWED` | `1` | boolean | deploy, dev | Global safety switch for manual checks from API/UI. | +| `INGESTION_MIN_RUN_INTERVAL_MINUTES` | `15` | integer >= 15 | deploy, dev | Server-enforced minimum for user-configured automatic check interval. | +| `INGESTION_MIN_REQUEST_DELAY_SECONDS` | `2` | integer >= 2 | deploy, dev | Server-enforced minimum delay between scholar requests. | | `SCHEDULER_QUEUE_BATCH_SIZE` | `10` | integer >= 1 | deploy, dev | Queue items processed per scheduler tick. | | `INGESTION_NETWORK_ERROR_RETRIES` | `1` | integer >= 0 | deploy, dev | Retries for transient ingestion network errors. | | `INGESTION_RETRY_BACKOFF_SECONDS` | `1.0` | float >= 0 | deploy, dev | Backoff delay for retry attempts. | @@ -178,6 +183,8 @@ Notes: | `INGESTION_ALERT_BLOCKED_FAILURE_THRESHOLD` | `1` | integer >= 1 | deploy, dev | Trigger blocked/captcha scrape alert flag when this many blocked failures occur in a run. | | `INGESTION_ALERT_NETWORK_FAILURE_THRESHOLD` | `2` | integer >= 1 | deploy, dev | Trigger network scrape alert flag when this many network failures occur in a run. | | `INGESTION_ALERT_RETRY_SCHEDULED_THRESHOLD` | `3` | integer >= 1 | deploy, dev | Trigger retry alert flag when scheduled retry count reaches this threshold in a run. | +| `INGESTION_SAFETY_COOLDOWN_BLOCKED_SECONDS` | `1800` | integer >= 60 | deploy, dev | Cooldown duration applied when blocked-failure threshold is exceeded. | +| `INGESTION_SAFETY_COOLDOWN_NETWORK_SECONDS` | `900` | integer >= 60 | deploy, dev | Cooldown duration applied when network-failure threshold is exceeded. | | `INGESTION_CONTINUATION_QUEUE_ENABLED` | `1` | boolean | deploy, dev | Enable continuation queue for long runs. | | `INGESTION_CONTINUATION_BASE_DELAY_SECONDS` | `120` | integer >= 0 | deploy, dev | Initial delay before retrying continuation queue items. | | `INGESTION_CONTINUATION_MAX_DELAY_SECONDS` | `3600` | integer >= 0 | deploy, dev | Maximum continuation retry delay. | @@ -200,6 +207,16 @@ Notes: | `SCHOLAR_NAME_SEARCH_ALERT_RETRY_COUNT_THRESHOLD` | `2` | integer >= 1 | deploy, dev | Emit retry-threshold observability warning when name-search retry count reaches this value. | | `SCHOLAR_NAME_SEARCH_ALERT_COOLDOWN_REJECTIONS_THRESHOLD` | `3` | integer >= 1 | deploy, dev | Emit cooldown-threshold observability alert after this many requests are rejected during active cooldown. | +### Scrape Safety Operations + +- Structured safety events are emitted for: + - policy/manual run blocks, + - cooldown entered/cleared transitions, + - blocked/network/retry threshold trips, + - scheduler cooldown skips/deferments. +- Use `LOG_FORMAT=json` and ship logs to your preferred collector for alerting. +- Runbook: `docs/ops/scrape_safety_runbook.md`. + ### Startup Bootstrap and DB Wait | Variable | Default | Options | Scope | Description | diff --git a/alembic/versions/20260219_0009_user_settings_scrape_safety_state.py b/alembic/versions/20260219_0009_user_settings_scrape_safety_state.py new file mode 100644 index 0000000..9ea658b --- /dev/null +++ b/alembic/versions/20260219_0009_user_settings_scrape_safety_state.py @@ -0,0 +1,66 @@ +"""Add scrape safety state fields to user settings. + +Revision ID: 20260219_0009 +Revises: 20260219_0008 +Create Date: 2026-02-19 21:20:00.000000 +""" + +from collections.abc import Sequence + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + + +# revision identifiers, used by Alembic. +revision: str = "20260219_0009" +down_revision: str | Sequence[str] | None = "20260219_0008" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +TABLE_NAME = "user_settings" + + +def upgrade() -> None: + bind = op.get_bind() + inspector = sa.inspect(bind) + columns = {column["name"] for column in inspector.get_columns(TABLE_NAME)} + + if "scrape_safety_state" not in columns: + op.add_column( + TABLE_NAME, + sa.Column( + "scrape_safety_state", + postgresql.JSONB(astext_type=sa.Text()), + nullable=False, + server_default=sa.text("'{}'::jsonb"), + ), + ) + + if "scrape_cooldown_until" not in columns: + op.add_column( + TABLE_NAME, + sa.Column("scrape_cooldown_until", sa.DateTime(timezone=True), nullable=True), + ) + + if "scrape_cooldown_reason" not in columns: + op.add_column( + TABLE_NAME, + sa.Column("scrape_cooldown_reason", sa.String(length=64), nullable=True), + ) + + +def downgrade() -> None: + bind = op.get_bind() + inspector = sa.inspect(bind) + columns = {column["name"] for column in inspector.get_columns(TABLE_NAME)} + + if "scrape_cooldown_reason" in columns: + op.drop_column(TABLE_NAME, "scrape_cooldown_reason") + + if "scrape_cooldown_until" in columns: + op.drop_column(TABLE_NAME, "scrape_cooldown_until") + + if "scrape_safety_state" in columns: + op.drop_column(TABLE_NAME, "scrape_safety_state") diff --git a/app/api/routers/runs.py b/app/api/routers/runs.py index e095cb7..ae9d803 100644 --- a/app/api/routers/runs.py +++ b/app/api/routers/runs.py @@ -22,6 +22,7 @@ from app.api.schemas import ( from app.db.models import RunStatus, RunTriggerType, User from app.db.session import get_db_session from app.services import ingestion as ingestion_service +from app.services import run_safety as run_safety_service from app.services import runs as run_service from app.services import user_settings as user_settings_service from app.settings import settings @@ -253,6 +254,7 @@ def _manual_run_payload_from_run( *, idempotency_key: str | None, reused_existing_run: bool, + safety_state: dict[str, Any], ) -> dict[str, Any]: summary = run_service.extract_run_summary(run.error_log) return { @@ -265,9 +267,37 @@ def _manual_run_payload_from_run( "new_publication_count": int(run.new_pub_count or 0), "reused_existing_run": reused_existing_run, "idempotency_key": idempotency_key, + "safety_state": safety_state, } +async def _load_safety_state( + db_session: AsyncSession, + *, + user_id: int, +) -> dict[str, Any]: + user_settings = await user_settings_service.get_or_create_settings( + db_session, + user_id=user_id, + ) + previous_safety_state = run_safety_service.get_safety_event_context(user_settings) + if run_safety_service.clear_expired_cooldown(user_settings): + await db_session.commit() + await db_session.refresh(user_settings) + logger.info( + "api.runs.safety_cooldown_cleared", + extra={ + "event": "api.runs.safety_cooldown_cleared", + "user_id": user_id, + "reason": previous_safety_state.get("cooldown_reason"), + "cooldown_until": previous_safety_state.get("cooldown_until"), + "metric_name": "api_runs_safety_cooldown_cleared_total", + "metric_value": 1, + }, + ) + return run_safety_service.get_safety_state_payload(user_settings) + + @router.get( "", response_model=RunsListEnvelope, @@ -285,10 +315,15 @@ async def list_runs( limit=limit, failed_only=failed_only, ) + safety_state = await _load_safety_state( + db_session, + user_id=current_user.id, + ) return success_payload( request, data={ "runs": [_serialize_run(run) for run in runs], + "safety_state": safety_state, }, ) @@ -318,12 +353,17 @@ async def get_run( scholar_results = error_log.get("scholar_results") if not isinstance(scholar_results, list): scholar_results = [] + safety_state = await _load_safety_state( + db_session, + user_id=current_user.id, + ) return success_payload( request, data={ "run": _serialize_run(run), "summary": run_service.extract_run_summary(error_log), "scholar_results": [_normalize_scholar_result(item) for item in scholar_results], + "safety_state": safety_state, }, ) @@ -338,6 +378,34 @@ async def run_manual( current_user: User = Depends(get_api_current_user), ingest_service: ingestion_service.ScholarIngestionService = Depends(get_ingestion_service), ): + safety_state = await _load_safety_state( + db_session, + user_id=current_user.id, + ) + if not settings.ingestion_manual_run_allowed: + logger.warning( + "api.runs.manual_blocked_policy", + extra={ + "event": "api.runs.manual_blocked_policy", + "user_id": current_user.id, + "policy": {"manual_run_allowed": False}, + "safety_state": safety_state, + "metric_name": "api_runs_manual_blocked_policy_total", + "metric_value": 1, + }, + ) + raise ApiException( + status_code=403, + code="manual_runs_disabled", + message="Manual checks are disabled by server policy.", + details={ + "policy": { + "manual_run_allowed": False, + }, + "safety_state": safety_state, + }, + ) + idempotency_key = _normalize_idempotency_key(request.headers.get(IDEMPOTENCY_HEADER)) if idempotency_key is not None: previous_run = await run_service.get_manual_run_by_idempotency_key( @@ -362,6 +430,7 @@ async def run_manual( previous_run, idempotency_key=idempotency_key, reused_existing_run=True, + safety_state=safety_state, ), ) @@ -393,6 +462,28 @@ async def run_manual( code="run_in_progress", message="A run is already in progress for this account.", ) from exc + except ingestion_service.RunBlockedBySafetyPolicyError as exc: + await db_session.rollback() + logger.info( + "api.runs.manual_blocked_safety", + extra={ + "event": "api.runs.manual_blocked_safety", + "user_id": current_user.id, + "reason": exc.safety_state.get("cooldown_reason"), + "cooldown_until": exc.safety_state.get("cooldown_until"), + "cooldown_remaining_seconds": exc.safety_state.get("cooldown_remaining_seconds"), + "metric_name": "api_runs_manual_blocked_safety_total", + "metric_value": 1, + }, + ) + raise ApiException( + status_code=429, + code=exc.code, + message=exc.message, + details={ + "safety_state": exc.safety_state, + }, + ) from exc except IntegrityError as exc: await db_session.rollback() if idempotency_key is not None: @@ -418,6 +509,10 @@ async def run_manual( existing_run, idempotency_key=idempotency_key, reused_existing_run=True, + safety_state=await _load_safety_state( + db_session, + user_id=current_user.id, + ), ), ) logger.exception( @@ -447,6 +542,10 @@ async def run_manual( message="Manual run failed.", ) from exc + current_safety_state = await _load_safety_state( + db_session, + user_id=current_user.id, + ) return success_payload( request, data={ @@ -459,6 +558,7 @@ async def run_manual( "new_publication_count": run_summary.new_publication_count, "reused_existing_run": False, "idempotency_key": idempotency_key, + "safety_state": current_safety_state, }, ) diff --git a/app/api/routers/settings.py b/app/api/routers/settings.py index 082cfe1..f4b36a5 100644 --- a/app/api/routers/settings.py +++ b/app/api/routers/settings.py @@ -11,19 +11,38 @@ from app.api.responses import success_payload from app.api.schemas import SettingsEnvelope, SettingsUpdateRequest from app.db.models import User from app.db.session import get_db_session +from app.services import run_safety as run_safety_service from app.services import user_settings as user_settings_service +from app.settings import settings as settings_module logger = logging.getLogger(__name__) router = APIRouter(prefix="/settings", tags=["api-settings"]) -def _serialize_settings(settings) -> dict[str, object]: +def _serialize_settings(user_settings) -> dict[str, object]: + min_run_interval_minutes = user_settings_service.resolve_run_interval_minimum( + settings_module.ingestion_min_run_interval_minutes + ) + min_request_delay_seconds = user_settings_service.resolve_request_delay_minimum( + settings_module.ingestion_min_request_delay_seconds + ) return { - "auto_run_enabled": bool(settings.auto_run_enabled), - "run_interval_minutes": int(settings.run_interval_minutes), - "request_delay_seconds": int(settings.request_delay_seconds), - "nav_visible_pages": list(settings.nav_visible_pages or []), + "auto_run_enabled": bool(user_settings.auto_run_enabled) and settings_module.ingestion_automation_allowed, + "run_interval_minutes": int(user_settings.run_interval_minutes), + "request_delay_seconds": int(user_settings.request_delay_seconds), + "nav_visible_pages": list(user_settings.nav_visible_pages or []), + "policy": { + "min_run_interval_minutes": min_run_interval_minutes, + "min_request_delay_seconds": min_request_delay_seconds, + "automation_allowed": bool(settings_module.ingestion_automation_allowed), + "manual_run_allowed": bool(settings_module.ingestion_manual_run_allowed), + "blocked_failure_threshold": max(1, int(settings_module.ingestion_alert_blocked_failure_threshold)), + "network_failure_threshold": max(1, int(settings_module.ingestion_alert_network_failure_threshold)), + "cooldown_blocked_seconds": max(60, int(settings_module.ingestion_safety_cooldown_blocked_seconds)), + "cooldown_network_seconds": max(60, int(settings_module.ingestion_safety_cooldown_network_seconds)), + }, + "safety_state": run_safety_service.get_safety_state_payload(user_settings), } @@ -36,13 +55,28 @@ async def get_settings( db_session: AsyncSession = Depends(get_db_session), current_user: User = Depends(get_api_current_user), ): - settings = await user_settings_service.get_or_create_settings( + user_settings = await user_settings_service.get_or_create_settings( db_session, user_id=current_user.id, ) + previous_safety_state = run_safety_service.get_safety_event_context(user_settings) + if run_safety_service.clear_expired_cooldown(user_settings): + await db_session.commit() + await db_session.refresh(user_settings) + logger.info( + "api.settings.safety_cooldown_cleared", + extra={ + "event": "api.settings.safety_cooldown_cleared", + "user_id": current_user.id, + "reason": previous_safety_state.get("cooldown_reason"), + "cooldown_until": previous_safety_state.get("cooldown_until"), + "metric_name": "api_settings_safety_cooldown_cleared_total", + "metric_value": 1, + }, + ) return success_payload( request, - data=_serialize_settings(settings), + data=_serialize_settings(user_settings), ) @@ -56,22 +90,31 @@ async def update_settings( db_session: AsyncSession = Depends(get_db_session), current_user: User = Depends(get_api_current_user), ): - settings = await user_settings_service.get_or_create_settings( + user_settings = await user_settings_service.get_or_create_settings( db_session, user_id=current_user.id, ) + min_run_interval_minutes = user_settings_service.resolve_run_interval_minimum( + settings_module.ingestion_min_run_interval_minutes + ) + min_request_delay_seconds = user_settings_service.resolve_request_delay_minimum( + settings_module.ingestion_min_request_delay_seconds + ) + try: parsed_interval = user_settings_service.parse_run_interval_minutes( - str(payload.run_interval_minutes) + str(payload.run_interval_minutes), + minimum=min_run_interval_minutes, ) parsed_delay = user_settings_service.parse_request_delay_seconds( - str(payload.request_delay_seconds) + str(payload.request_delay_seconds), + minimum=min_request_delay_seconds, ) parsed_nav_visible_pages = user_settings_service.parse_nav_visible_pages( payload.nav_visible_pages if payload.nav_visible_pages is not None - else list(settings.nav_visible_pages or user_settings_service.DEFAULT_NAV_VISIBLE_PAGES) + else list(user_settings.nav_visible_pages or user_settings_service.DEFAULT_NAV_VISIBLE_PAGES) ) except user_settings_service.UserSettingsServiceError as exc: raise ApiException( @@ -80,14 +123,33 @@ async def update_settings( message=str(exc), ) from exc + effective_auto_run_enabled = ( + bool(payload.auto_run_enabled) and settings_module.ingestion_automation_allowed + ) + updated = await user_settings_service.update_settings( db_session, - settings=settings, - auto_run_enabled=bool(payload.auto_run_enabled), + settings=user_settings, + auto_run_enabled=effective_auto_run_enabled, run_interval_minutes=parsed_interval, request_delay_seconds=parsed_delay, nav_visible_pages=parsed_nav_visible_pages, ) + previous_safety_state = run_safety_service.get_safety_event_context(updated) + if run_safety_service.clear_expired_cooldown(updated): + await db_session.commit() + await db_session.refresh(updated) + logger.info( + "api.settings.safety_cooldown_cleared", + extra={ + "event": "api.settings.safety_cooldown_cleared", + "user_id": current_user.id, + "reason": previous_safety_state.get("cooldown_reason"), + "cooldown_until": previous_safety_state.get("cooldown_until"), + "metric_name": "api_settings_safety_cooldown_cleared_total", + "metric_value": 1, + }, + ) logger.info( "api.settings.updated", extra={ diff --git a/app/api/schemas.py b/app/api/schemas.py index 73a3452..d7424f2 100644 --- a/app/api/schemas.py +++ b/app/api/schemas.py @@ -202,6 +202,7 @@ class RunListItemData(BaseModel): class RunsListData(BaseModel): runs: list[RunListItemData] + safety_state: ScrapeSafetyStateData model_config = ConfigDict(extra="forbid") @@ -227,6 +228,30 @@ class RunSummaryData(BaseModel): model_config = ConfigDict(extra="forbid") +class ScrapeSafetyCountersData(BaseModel): + consecutive_blocked_runs: int = 0 + consecutive_network_runs: int = 0 + cooldown_entry_count: int = 0 + blocked_start_count: int = 0 + last_blocked_failure_count: int = 0 + last_network_failure_count: int = 0 + last_evaluated_run_id: int | None = None + + model_config = ConfigDict(extra="forbid") + + +class ScrapeSafetyStateData(BaseModel): + cooldown_active: bool + cooldown_reason: str | None = None + cooldown_reason_label: str | None = None + cooldown_until: datetime | None = None + cooldown_remaining_seconds: int = 0 + recommended_action: str | None = None + counters: ScrapeSafetyCountersData = Field(default_factory=ScrapeSafetyCountersData) + + model_config = ConfigDict(extra="forbid") + + class RunAttemptLogData(BaseModel): attempt: int cstart: int @@ -293,6 +318,7 @@ class RunDetailData(BaseModel): run: RunListItemData summary: RunSummaryData scholar_results: list[RunScholarResultData] + safety_state: ScrapeSafetyStateData model_config = ConfigDict(extra="forbid") @@ -314,6 +340,7 @@ class ManualRunData(BaseModel): new_publication_count: int reused_existing_run: bool idempotency_key: str | None = None + safety_state: ScrapeSafetyStateData model_config = ConfigDict(extra="forbid") @@ -429,11 +456,26 @@ class AdminResetPasswordRequest(BaseModel): model_config = ConfigDict(extra="forbid") +class SettingsPolicyData(BaseModel): + min_run_interval_minutes: int + min_request_delay_seconds: int + automation_allowed: bool + manual_run_allowed: bool + blocked_failure_threshold: int + network_failure_threshold: int + cooldown_blocked_seconds: int + cooldown_network_seconds: int + + model_config = ConfigDict(extra="forbid") + + class SettingsData(BaseModel): auto_run_enabled: bool run_interval_minutes: int request_delay_seconds: int nav_visible_pages: list[str] + policy: SettingsPolicyData + safety_state: ScrapeSafetyStateData model_config = ConfigDict(extra="forbid") diff --git a/app/db/models.py b/app/db/models.py index d98cfe9..d7d8e8e 100644 --- a/app/db/models.py +++ b/app/db/models.py @@ -105,6 +105,13 @@ class UserSetting(Base): '\'["dashboard","scholars","publications","settings","style-guide","runs","users"]\'::jsonb' ), ) + scrape_safety_state: Mapped[dict] = mapped_column( + JSONB, + nullable=False, + server_default=text("'{}'::jsonb"), + ) + scrape_cooldown_until: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + scrape_cooldown_reason: Mapped[str | None] = mapped_column(String(64)) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) diff --git a/app/services/ingestion.py b/app/services/ingestion.py index 5d50811..f8d8b60 100644 --- a/app/services/ingestion.py +++ b/app/services/ingestion.py @@ -22,6 +22,8 @@ from app.db.models import ( ScholarPublication, ) from app.services import continuation_queue as queue_service +from app.services import run_safety as run_safety_service +from app.services import user_settings as user_settings_service from app.services.scholar_parser import ( ParseState, ParsedProfilePage, @@ -29,6 +31,7 @@ from app.services.scholar_parser import ( parse_profile_page, ) from app.services.scholar_source import FetchResult, ScholarSource +from app.settings import settings TITLE_ALNUM_RE = re.compile(r"[^a-z0-9]+") WORD_RE = re.compile(r"[a-z0-9]+") @@ -88,6 +91,20 @@ class RunAlreadyInProgressError(RuntimeError): """Raised when a run lock for a user is already held by another process.""" +class RunBlockedBySafetyPolicyError(RuntimeError): + def __init__( + self, + *, + code: str, + message: str, + safety_state: dict[str, Any], + ) -> None: + super().__init__(message) + self.code = code + self.message = message + self.safety_state = safety_state + + def _int_or_default(value: Any, default: int = 0) -> int: try: return int(value) @@ -134,6 +151,61 @@ class ScholarIngestionService: alert_network_failure_threshold: int = 2, alert_retry_scheduled_threshold: int = 3, ) -> RunExecutionSummary: + user_settings = await user_settings_service.get_or_create_settings( + db_session, + user_id=user_id, + ) + now_utc = datetime.now(timezone.utc) + previous_safety_state = run_safety_service.get_safety_event_context( + user_settings, + now_utc=now_utc, + ) + if run_safety_service.clear_expired_cooldown( + user_settings, + now_utc=now_utc, + ): + await db_session.commit() + await db_session.refresh(user_settings) + logger.info( + "ingestion.safety_cooldown_cleared", + extra={ + "event": "ingestion.safety_cooldown_cleared", + "user_id": user_id, + "reason": previous_safety_state.get("cooldown_reason"), + "cooldown_until": previous_safety_state.get("cooldown_until"), + "metric_name": "ingestion_safety_cooldown_cleared_total", + "metric_value": 1, + }, + ) + now_utc = datetime.now(timezone.utc) + if run_safety_service.is_cooldown_active(user_settings, now_utc=now_utc): + safety_state = run_safety_service.register_cooldown_blocked_start( + user_settings, + now_utc=now_utc, + ) + await db_session.commit() + logger.warning( + "ingestion.safety_policy_blocked_run_start", + extra={ + "event": "ingestion.safety_policy_blocked_run_start", + "user_id": user_id, + "trigger_type": trigger_type.value, + "reason": safety_state.get("cooldown_reason"), + "cooldown_until": safety_state.get("cooldown_until"), + "cooldown_remaining_seconds": safety_state.get("cooldown_remaining_seconds"), + "blocked_start_count": ( + (safety_state.get("counters") or {}).get("blocked_start_count") + ), + "metric_name": "ingestion_safety_run_start_blocked_total", + "metric_value": 1, + }, + ) + raise RunBlockedBySafetyPolicyError( + code="scrape_cooldown_active", + message="Scrape safety cooldown is active; run start is temporarily blocked.", + safety_state=safety_state, + ) + lock_acquired = await self._try_acquire_user_lock( db_session, user_id=user_id, @@ -505,6 +577,8 @@ class ScholarIngestionService: "crawl_run_id": run.id, "blocked_failure_count": blocked_failure_count, "threshold": bounded_blocked_threshold, + "metric_name": "ingestion_blocked_failure_threshold_exceeded_total", + "metric_value": 1, }, ) if alert_flags["network_failure_threshold_exceeded"]: @@ -516,6 +590,8 @@ class ScholarIngestionService: "crawl_run_id": run.id, "network_failure_count": network_failure_count, "threshold": bounded_network_threshold, + "metric_name": "ingestion_network_failure_threshold_exceeded_total", + "metric_value": 1, }, ) if alert_flags["retry_scheduled_threshold_exceeded"]: @@ -527,6 +603,56 @@ class ScholarIngestionService: "crawl_run_id": run.id, "retries_scheduled_count": retries_scheduled_count, "threshold": bounded_retry_threshold, + "metric_name": "ingestion_retry_scheduled_threshold_exceeded_total", + "metric_value": 1, + }, + ) + + pre_apply_safety_state = run_safety_service.get_safety_event_context( + user_settings, + now_utc=datetime.now(timezone.utc), + ) + safety_state, cooldown_trigger_reason = run_safety_service.apply_run_safety_outcome( + user_settings, + run_id=int(run.id), + blocked_failure_count=blocked_failure_count, + network_failure_count=network_failure_count, + blocked_failure_threshold=bounded_blocked_threshold, + network_failure_threshold=bounded_network_threshold, + blocked_cooldown_seconds=settings.ingestion_safety_cooldown_blocked_seconds, + network_cooldown_seconds=settings.ingestion_safety_cooldown_network_seconds, + now_utc=datetime.now(timezone.utc), + ) + if cooldown_trigger_reason is not None: + logger.warning( + "ingestion.safety_cooldown_entered", + extra={ + "event": "ingestion.safety_cooldown_entered", + "user_id": user_id, + "crawl_run_id": run.id, + "reason": cooldown_trigger_reason, + "blocked_failure_count": blocked_failure_count, + "network_failure_count": network_failure_count, + "blocked_failure_threshold": bounded_blocked_threshold, + "network_failure_threshold": bounded_network_threshold, + "cooldown_until": safety_state.get("cooldown_until"), + "cooldown_remaining_seconds": safety_state.get("cooldown_remaining_seconds"), + "safety_counters": safety_state.get("counters", {}), + "metric_name": "ingestion_safety_cooldown_entered_total", + "metric_value": 1, + }, + ) + elif pre_apply_safety_state.get("cooldown_active") and not safety_state.get("cooldown_active"): + logger.info( + "ingestion.safety_cooldown_cleared", + extra={ + "event": "ingestion.safety_cooldown_cleared", + "user_id": user_id, + "crawl_run_id": run.id, + "reason": pre_apply_safety_state.get("cooldown_reason"), + "cooldown_until": pre_apply_safety_state.get("cooldown_until"), + "metric_name": "ingestion_safety_cooldown_cleared_total", + "metric_value": 1, }, ) diff --git a/app/services/run_safety.py b/app/services/run_safety.py new file mode 100644 index 0000000..210c688 --- /dev/null +++ b/app/services/run_safety.py @@ -0,0 +1,239 @@ +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from typing import Any + +from app.db.models import UserSetting + +COOLDOWN_REASON_BLOCKED_FAILURE_THRESHOLD = "blocked_failure_threshold_exceeded" +COOLDOWN_REASON_NETWORK_FAILURE_THRESHOLD = "network_failure_threshold_exceeded" + +_COUNTER_CONSECUTIVE_BLOCKED_RUNS = "consecutive_blocked_runs" +_COUNTER_CONSECUTIVE_NETWORK_RUNS = "consecutive_network_runs" +_COUNTER_COOLDOWN_ENTRY_COUNT = "cooldown_entry_count" +_COUNTER_BLOCKED_START_COUNT = "blocked_start_count" +_COUNTER_LAST_BLOCKED_FAILURE_COUNT = "last_blocked_failure_count" +_COUNTER_LAST_NETWORK_FAILURE_COUNT = "last_network_failure_count" +_COUNTER_LAST_EVALUATED_RUN_ID = "last_evaluated_run_id" + + +def _utcnow() -> datetime: + return datetime.now(timezone.utc) + + +def _safe_int(value: Any, default: int = 0) -> int: + try: + return int(value) + except (TypeError, ValueError): + return default + + +def _safe_optional_int(value: Any) -> int | None: + if value is None: + return None + try: + return int(value) + except (TypeError, ValueError): + return None + + +def _normalize_datetime(value: datetime | None) -> datetime | None: + if value is None: + return None + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + +def _state_dict(settings: UserSetting) -> dict[str, Any]: + state = settings.scrape_safety_state + if isinstance(state, dict): + return state + return {} + + +def _counters_from_state(settings: UserSetting) -> dict[str, Any]: + state = _state_dict(settings) + return { + _COUNTER_CONSECUTIVE_BLOCKED_RUNS: max( + 0, + _safe_int(state.get(_COUNTER_CONSECUTIVE_BLOCKED_RUNS), 0), + ), + _COUNTER_CONSECUTIVE_NETWORK_RUNS: max( + 0, + _safe_int(state.get(_COUNTER_CONSECUTIVE_NETWORK_RUNS), 0), + ), + _COUNTER_COOLDOWN_ENTRY_COUNT: max( + 0, + _safe_int(state.get(_COUNTER_COOLDOWN_ENTRY_COUNT), 0), + ), + _COUNTER_BLOCKED_START_COUNT: max( + 0, + _safe_int(state.get(_COUNTER_BLOCKED_START_COUNT), 0), + ), + _COUNTER_LAST_BLOCKED_FAILURE_COUNT: max( + 0, + _safe_int(state.get(_COUNTER_LAST_BLOCKED_FAILURE_COUNT), 0), + ), + _COUNTER_LAST_NETWORK_FAILURE_COUNT: max( + 0, + _safe_int(state.get(_COUNTER_LAST_NETWORK_FAILURE_COUNT), 0), + ), + _COUNTER_LAST_EVALUATED_RUN_ID: _safe_optional_int( + state.get(_COUNTER_LAST_EVALUATED_RUN_ID), + ), + } + + +def _cooldown_reason_label(reason: str | None) -> str | None: + if reason == COOLDOWN_REASON_BLOCKED_FAILURE_THRESHOLD: + return "Blocked responses exceeded safety threshold" + if reason == COOLDOWN_REASON_NETWORK_FAILURE_THRESHOLD: + return "Network failures exceeded safety threshold" + return None + + +def _recommended_action(reason: str | None) -> str | None: + if reason == COOLDOWN_REASON_BLOCKED_FAILURE_THRESHOLD: + return ( + "Google Scholar appears to be blocking requests. Wait for cooldown to expire, " + "increase request delay, and avoid repeated manual retries." + ) + if reason == COOLDOWN_REASON_NETWORK_FAILURE_THRESHOLD: + return ( + "Network failures crossed the threshold. Verify connectivity and retry after cooldown." + ) + return None + + +def is_cooldown_active( + settings: UserSetting, + *, + now_utc: datetime | None = None, +) -> bool: + now = now_utc or _utcnow() + cooldown_until = _normalize_datetime(settings.scrape_cooldown_until) + if cooldown_until is None: + return False + return cooldown_until > now + + +def clear_expired_cooldown( + settings: UserSetting, + *, + now_utc: datetime | None = None, +) -> bool: + now = now_utc or _utcnow() + cooldown_until = _normalize_datetime(settings.scrape_cooldown_until) + if cooldown_until is None: + return False + if cooldown_until > now: + return False + settings.scrape_cooldown_until = None + settings.scrape_cooldown_reason = None + return True + + +def register_cooldown_blocked_start( + settings: UserSetting, + *, + now_utc: datetime | None = None, +) -> dict[str, Any]: + now = now_utc or _utcnow() + counters = _counters_from_state(settings) + counters[_COUNTER_BLOCKED_START_COUNT] = int(counters[_COUNTER_BLOCKED_START_COUNT]) + 1 + settings.scrape_safety_state = counters + return get_safety_state_payload(settings, now_utc=now) + + +def apply_run_safety_outcome( + settings: UserSetting, + *, + run_id: int, + blocked_failure_count: int, + network_failure_count: int, + blocked_failure_threshold: int, + network_failure_threshold: int, + blocked_cooldown_seconds: int, + network_cooldown_seconds: int, + now_utc: datetime | None = None, +) -> tuple[dict[str, Any], str | None]: + now = now_utc or _utcnow() + counters = _counters_from_state(settings) + + bounded_blocked_failures = max(0, int(blocked_failure_count)) + bounded_network_failures = max(0, int(network_failure_count)) + + counters[_COUNTER_LAST_BLOCKED_FAILURE_COUNT] = bounded_blocked_failures + counters[_COUNTER_LAST_NETWORK_FAILURE_COUNT] = bounded_network_failures + counters[_COUNTER_LAST_EVALUATED_RUN_ID] = int(run_id) + counters[_COUNTER_CONSECUTIVE_BLOCKED_RUNS] = ( + int(counters[_COUNTER_CONSECUTIVE_BLOCKED_RUNS]) + 1 + if bounded_blocked_failures > 0 + else 0 + ) + counters[_COUNTER_CONSECUTIVE_NETWORK_RUNS] = ( + int(counters[_COUNTER_CONSECUTIVE_NETWORK_RUNS]) + 1 + if bounded_network_failures > 0 + else 0 + ) + + reason: str | None = None + cooldown_seconds = 0 + + if bounded_blocked_failures >= max(1, int(blocked_failure_threshold)): + reason = COOLDOWN_REASON_BLOCKED_FAILURE_THRESHOLD + cooldown_seconds = max(60, int(blocked_cooldown_seconds)) + elif bounded_network_failures >= max(1, int(network_failure_threshold)): + reason = COOLDOWN_REASON_NETWORK_FAILURE_THRESHOLD + cooldown_seconds = max(60, int(network_cooldown_seconds)) + + if reason is not None: + settings.scrape_cooldown_reason = reason + settings.scrape_cooldown_until = now + timedelta(seconds=cooldown_seconds) + counters[_COUNTER_COOLDOWN_ENTRY_COUNT] = int(counters[_COUNTER_COOLDOWN_ENTRY_COUNT]) + 1 + else: + clear_expired_cooldown(settings, now_utc=now) + + settings.scrape_safety_state = counters + return get_safety_state_payload(settings, now_utc=now), reason + + +def get_safety_state_payload( + settings: UserSetting, + *, + now_utc: datetime | None = None, +) -> dict[str, Any]: + now = now_utc or _utcnow() + cooldown_until = _normalize_datetime(settings.scrape_cooldown_until) + cooldown_active = bool(cooldown_until is not None and cooldown_until > now) + cooldown_remaining_seconds = 0 + if cooldown_active and cooldown_until is not None: + cooldown_remaining_seconds = max(0, int((cooldown_until - now).total_seconds())) + + reason = settings.scrape_cooldown_reason if cooldown_active else None + + return { + "cooldown_active": cooldown_active, + "cooldown_reason": reason, + "cooldown_reason_label": _cooldown_reason_label(reason), + "cooldown_until": cooldown_until, + "cooldown_remaining_seconds": cooldown_remaining_seconds, + "recommended_action": _recommended_action(reason), + "counters": _counters_from_state(settings), + } + + +def get_safety_event_context( + settings: UserSetting, + *, + now_utc: datetime | None = None, +) -> dict[str, Any]: + payload = get_safety_state_payload(settings, now_utc=now_utc) + return { + "cooldown_active": bool(payload.get("cooldown_active")), + "cooldown_reason": payload.get("cooldown_reason"), + "cooldown_until": payload.get("cooldown_until"), + "cooldown_remaining_seconds": int(payload.get("cooldown_remaining_seconds") or 0), + "safety_counters": payload.get("counters", {}), + } diff --git a/app/services/scheduler.py b/app/services/scheduler.py index fef4ddb..96dd5f1 100644 --- a/app/services/scheduler.py +++ b/app/services/scheduler.py @@ -18,7 +18,11 @@ from app.db.models import ( ) from app.db.session import get_session_factory from app.services import continuation_queue as queue_service -from app.services.ingestion import RunAlreadyInProgressError, ScholarIngestionService +from app.services.ingestion import ( + RunAlreadyInProgressError, + RunBlockedBySafetyPolicyError, + ScholarIngestionService, +) from app.services.scholar_source import LiveScholarSource from app.settings import settings @@ -30,6 +34,8 @@ class _AutoRunCandidate: user_id: int run_interval_minutes: int request_delay_seconds: int + cooldown_until: datetime | None + cooldown_reason: str | None class SchedulerService: @@ -134,6 +140,9 @@ class SchedulerService: await self._run_candidate(candidate) async def _load_candidates(self) -> list[_AutoRunCandidate]: + if not settings.ingestion_automation_allowed: + return [] + session_factory = get_session_factory() async with session_factory() as session: result = await session.execute( @@ -141,6 +150,8 @@ class SchedulerService: UserSetting.user_id, UserSetting.run_interval_minutes, UserSetting.request_delay_seconds, + UserSetting.scrape_cooldown_until, + UserSetting.scrape_cooldown_reason, ) .join(User, User.id == UserSetting.user_id) .where( @@ -150,14 +161,35 @@ class SchedulerService: .order_by(UserSetting.user_id.asc()) ) rows = result.all() - return [ - _AutoRunCandidate( - user_id=int(user_id), - run_interval_minutes=int(run_interval_minutes), - request_delay_seconds=int(request_delay_seconds), + now_utc = datetime.now(timezone.utc) + candidates: list[_AutoRunCandidate] = [] + for user_id, run_interval_minutes, request_delay_seconds, cooldown_until, cooldown_reason in rows: + if cooldown_until is not None and cooldown_until.tzinfo is None: + cooldown_until = cooldown_until.replace(tzinfo=timezone.utc) + if cooldown_until is not None and cooldown_until > now_utc: + logger.info( + "scheduler.run_skipped_safety_cooldown_precheck", + extra={ + "event": "scheduler.run_skipped_safety_cooldown_precheck", + "user_id": int(user_id), + "reason": cooldown_reason, + "cooldown_until": cooldown_until, + "cooldown_remaining_seconds": int((cooldown_until - now_utc).total_seconds()), + "metric_name": "scheduler_run_skipped_safety_cooldown_total", + "metric_value": 1, + }, + ) + continue + candidates.append( + _AutoRunCandidate( + user_id=int(user_id), + run_interval_minutes=int(run_interval_minutes), + request_delay_seconds=int(request_delay_seconds), + cooldown_until=cooldown_until, + cooldown_reason=(str(cooldown_reason).strip() if cooldown_reason else None), + ) ) - for user_id, run_interval_minutes, request_delay_seconds in rows - ] + return candidates async def _is_due(self, candidate: _AutoRunCandidate, *, now: datetime) -> bool: session_factory = get_session_factory() @@ -210,6 +242,21 @@ class SchedulerService: }, ) return + except RunBlockedBySafetyPolicyError as exc: + await session.rollback() + logger.info( + "scheduler.run_skipped_safety_cooldown", + extra={ + "event": "scheduler.run_skipped_safety_cooldown", + "user_id": candidate.user_id, + "reason": exc.safety_state.get("cooldown_reason"), + "cooldown_until": exc.safety_state.get("cooldown_until"), + "cooldown_remaining_seconds": exc.safety_state.get("cooldown_remaining_seconds"), + "metric_name": "scheduler_run_skipped_safety_cooldown_total", + "metric_value": 1, + }, + ) + return except Exception: await session.rollback() logger.exception( @@ -354,6 +401,34 @@ class SchedulerService: }, ) return + except RunBlockedBySafetyPolicyError as exc: + await session.rollback() + cooldown_remaining_seconds = max( + self._tick_seconds, + int(exc.safety_state.get("cooldown_remaining_seconds") or 0), + ) + async with session_factory() as recovery_session: + await queue_service.reschedule_job( + recovery_session, + job_id=job.id, + delay_seconds=max(self._tick_seconds, cooldown_remaining_seconds), + reason="scrape_safety_cooldown", + error=str(exc.message), + ) + await recovery_session.commit() + logger.info( + "scheduler.queue_item_deferred_safety_cooldown", + extra={ + "event": "scheduler.queue_item_deferred_safety_cooldown", + "queue_item_id": job.id, + "user_id": job.user_id, + "reason": exc.safety_state.get("cooldown_reason"), + "cooldown_remaining_seconds": cooldown_remaining_seconds, + "metric_name": "scheduler_queue_item_deferred_safety_cooldown_total", + "metric_value": 1, + }, + ) + return except Exception as exc: await session.rollback() async with session_factory() as recovery_session: diff --git a/app/services/user_settings.py b/app/services/user_settings.py index d3314e7..3548e44 100644 --- a/app/services/user_settings.py +++ b/app/services/user_settings.py @@ -33,25 +33,49 @@ REQUIRED_NAV_PAGES = ( NAV_PAGE_SETTINGS, ) DEFAULT_NAV_VISIBLE_PAGES = list(ALLOWED_NAV_PAGES) +HARD_MIN_RUN_INTERVAL_MINUTES = 15 +HARD_MIN_REQUEST_DELAY_SECONDS = 2 -def parse_run_interval_minutes(value: str) -> int: +def resolve_run_interval_minimum(configured_minimum: int | None) -> int: + try: + parsed = int(configured_minimum) if configured_minimum is not None else HARD_MIN_RUN_INTERVAL_MINUTES + except (TypeError, ValueError): + parsed = HARD_MIN_RUN_INTERVAL_MINUTES + return max(HARD_MIN_RUN_INTERVAL_MINUTES, parsed) + + +def resolve_request_delay_minimum(configured_minimum: int | None) -> int: + try: + parsed = int(configured_minimum) if configured_minimum is not None else HARD_MIN_REQUEST_DELAY_SECONDS + except (TypeError, ValueError): + parsed = HARD_MIN_REQUEST_DELAY_SECONDS + return max(HARD_MIN_REQUEST_DELAY_SECONDS, parsed) + + +def parse_run_interval_minutes(value: str, *, minimum: int = HARD_MIN_RUN_INTERVAL_MINUTES) -> int: try: parsed = int(value) except ValueError as exc: raise UserSettingsServiceError("Check interval must be a whole number.") from exc - if parsed < 15: - raise UserSettingsServiceError("Check interval must be at least 15 minutes.") + effective_minimum = resolve_run_interval_minimum(minimum) + if parsed < effective_minimum: + raise UserSettingsServiceError( + f"Check interval must be at least {effective_minimum} minutes." + ) return parsed -def parse_request_delay_seconds(value: str) -> int: +def parse_request_delay_seconds(value: str, *, minimum: int = HARD_MIN_REQUEST_DELAY_SECONDS) -> int: try: parsed = int(value) except ValueError as exc: raise UserSettingsServiceError("Request delay must be a whole number.") from exc - if parsed < 2: - raise UserSettingsServiceError("Request delay must be at least 2 seconds.") + effective_minimum = resolve_request_delay_minimum(minimum) + if parsed < effective_minimum: + raise UserSettingsServiceError( + f"Request delay must be at least {effective_minimum} seconds." + ) return parsed diff --git a/app/settings.py b/app/settings.py index 48b875b..4299405 100644 --- a/app/settings.py +++ b/app/settings.py @@ -115,6 +115,22 @@ class Settings: log_redact_fields: str = os.getenv("LOG_REDACT_FIELDS", "") scheduler_enabled: bool = _env_bool("SCHEDULER_ENABLED", True) scheduler_tick_seconds: int = _env_int("SCHEDULER_TICK_SECONDS", 60) + ingestion_automation_allowed: bool = _env_bool( + "INGESTION_AUTOMATION_ALLOWED", + True, + ) + ingestion_manual_run_allowed: bool = _env_bool( + "INGESTION_MANUAL_RUN_ALLOWED", + True, + ) + ingestion_min_run_interval_minutes: int = _env_int( + "INGESTION_MIN_RUN_INTERVAL_MINUTES", + 15, + ) + ingestion_min_request_delay_seconds: int = _env_int( + "INGESTION_MIN_REQUEST_DELAY_SECONDS", + 2, + ) ingestion_network_error_retries: int = _env_int("INGESTION_NETWORK_ERROR_RETRIES", 1) ingestion_retry_backoff_seconds: float = _env_float( "INGESTION_RETRY_BACKOFF_SECONDS", @@ -137,6 +153,14 @@ class Settings: "INGESTION_ALERT_RETRY_SCHEDULED_THRESHOLD", 3, ) + ingestion_safety_cooldown_blocked_seconds: int = _env_int( + "INGESTION_SAFETY_COOLDOWN_BLOCKED_SECONDS", + 1800, + ) + ingestion_safety_cooldown_network_seconds: int = _env_int( + "INGESTION_SAFETY_COOLDOWN_NETWORK_SECONDS", + 900, + ) ingestion_continuation_queue_enabled: bool = _env_bool( "INGESTION_CONTINUATION_QUEUE_ENABLED", True, diff --git a/docker-compose.yml b/docker-compose.yml index a2d9a5c..eb861e7 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -40,6 +40,10 @@ services: LOG_REDACT_FIELDS: ${LOG_REDACT_FIELDS:-} SCHEDULER_ENABLED: ${SCHEDULER_ENABLED:-1} SCHEDULER_TICK_SECONDS: ${SCHEDULER_TICK_SECONDS:-60} + INGESTION_AUTOMATION_ALLOWED: ${INGESTION_AUTOMATION_ALLOWED:-1} + INGESTION_MANUAL_RUN_ALLOWED: ${INGESTION_MANUAL_RUN_ALLOWED:-1} + INGESTION_MIN_RUN_INTERVAL_MINUTES: ${INGESTION_MIN_RUN_INTERVAL_MINUTES:-15} + INGESTION_MIN_REQUEST_DELAY_SECONDS: ${INGESTION_MIN_REQUEST_DELAY_SECONDS:-2} INGESTION_NETWORK_ERROR_RETRIES: ${INGESTION_NETWORK_ERROR_RETRIES:-1} INGESTION_RETRY_BACKOFF_SECONDS: ${INGESTION_RETRY_BACKOFF_SECONDS:-1.0} INGESTION_MAX_PAGES_PER_SCHOLAR: ${INGESTION_MAX_PAGES_PER_SCHOLAR:-30} @@ -47,6 +51,8 @@ services: INGESTION_ALERT_BLOCKED_FAILURE_THRESHOLD: ${INGESTION_ALERT_BLOCKED_FAILURE_THRESHOLD:-1} INGESTION_ALERT_NETWORK_FAILURE_THRESHOLD: ${INGESTION_ALERT_NETWORK_FAILURE_THRESHOLD:-2} INGESTION_ALERT_RETRY_SCHEDULED_THRESHOLD: ${INGESTION_ALERT_RETRY_SCHEDULED_THRESHOLD:-3} + INGESTION_SAFETY_COOLDOWN_BLOCKED_SECONDS: ${INGESTION_SAFETY_COOLDOWN_BLOCKED_SECONDS:-1800} + INGESTION_SAFETY_COOLDOWN_NETWORK_SECONDS: ${INGESTION_SAFETY_COOLDOWN_NETWORK_SECONDS:-900} INGESTION_CONTINUATION_QUEUE_ENABLED: ${INGESTION_CONTINUATION_QUEUE_ENABLED:-1} INGESTION_CONTINUATION_BASE_DELAY_SECONDS: ${INGESTION_CONTINUATION_BASE_DELAY_SECONDS:-120} INGESTION_CONTINUATION_MAX_DELAY_SECONDS: ${INGESTION_CONTINUATION_MAX_DELAY_SECONDS:-3600} diff --git a/docs/ops/scrape_safety_runbook.md b/docs/ops/scrape_safety_runbook.md new file mode 100644 index 0000000..db01064 --- /dev/null +++ b/docs/ops/scrape_safety_runbook.md @@ -0,0 +1,74 @@ +# Scrape Safety Runbook + +This runbook covers operational handling for scrape safety cooldowns, threshold alerts, and blocked-IP events. + +## Key Signals + +Use structured log `event` fields (recommended with `LOG_FORMAT=json`): + +- `ingestion.safety_policy_blocked_run_start` +- `ingestion.safety_cooldown_entered` +- `ingestion.safety_cooldown_cleared` +- `ingestion.alert_blocked_failure_threshold_exceeded` +- `ingestion.alert_network_failure_threshold_exceeded` +- `ingestion.alert_retry_scheduled_threshold_exceeded` +- `api.runs.manual_blocked_policy` +- `api.runs.manual_blocked_safety` +- `scheduler.run_skipped_safety_cooldown` +- `scheduler.run_skipped_safety_cooldown_precheck` +- `scheduler.queue_item_deferred_safety_cooldown` + +Each event includes metric-style fields (`metric_name`, `metric_value`) for straightforward log-based alert rules. + +## Recommended Alert Rules + +- Cooldown enters: + - Trigger on `event=ingestion.safety_cooldown_entered`. +- Repeated start blocks: + - Trigger on high rate of `event=api.runs.manual_blocked_safety`. +- Threshold trips: + - Trigger on any of: + - `ingestion.alert_blocked_failure_threshold_exceeded` + - `ingestion.alert_network_failure_threshold_exceeded` + - `ingestion.alert_retry_scheduled_threshold_exceeded` +- Scheduler pressure: + - Trigger on sustained `scheduler.queue_item_deferred_safety_cooldown`. + +## If Your IP Appears Blocked + +Symptoms: + +- cooldown reason `blocked_failure_threshold_exceeded` +- parse state `blocked_or_captcha` +- redirects toward Google account sign-in flows + +Actions: + +1. Stop manual retries immediately. +2. Let cooldown expire; do not spam retriggers. +3. Increase `INGESTION_MIN_REQUEST_DELAY_SECONDS` and user request delay values. +4. Reduce concurrency pressure (keep one scheduler instance). +5. Keep name-search disabled/WIP for now if login-gated responses persist. +6. Resume with a small monitored run and verify blocked rate drops. + +Avoid: + +- aggressive rapid retries +- rotating through risky scraping patterns that increase challenge rates +- bypass/captcha-solving workflows that may violate source platform rules + +## Environment Controls + +Policy floors and safety controls: + +- `INGESTION_MIN_REQUEST_DELAY_SECONDS` +- `INGESTION_MIN_RUN_INTERVAL_MINUTES` +- `INGESTION_ALERT_BLOCKED_FAILURE_THRESHOLD` +- `INGESTION_ALERT_NETWORK_FAILURE_THRESHOLD` +- `INGESTION_ALERT_RETRY_SCHEDULED_THRESHOLD` +- `INGESTION_SAFETY_COOLDOWN_BLOCKED_SECONDS` +- `INGESTION_SAFETY_COOLDOWN_NETWORK_SECONDS` +- `INGESTION_MANUAL_RUN_ALLOWED` +- `INGESTION_AUTOMATION_ALLOWED` + +Apply stricter values first, then relax slowly only after sustained healthy runs. diff --git a/frontend/src/components/layout/AppNav.vue b/frontend/src/components/layout/AppNav.vue index 1e07d9c..5243973 100644 --- a/frontend/src/components/layout/AppNav.vue +++ b/frontend/src/components/layout/AppNav.vue @@ -3,7 +3,9 @@ import { computed } from "vue"; import { useRouter } from "vue-router"; import RunStatusBadge from "@/components/patterns/RunStatusBadge.vue"; +import ScrapeSafetyBadge from "@/components/patterns/ScrapeSafetyBadge.vue"; import AppButton from "@/components/ui/AppButton.vue"; +import { formatCooldownCountdown } from "@/features/safety"; import { useAuthStore } from "@/stores/auth"; import { useRunStatusStore } from "@/stores/run_status"; import { useUserSettingsStore } from "@/stores/user_settings"; @@ -31,21 +33,34 @@ const links = computed(() => { return userSettings.isPageVisible(item.id); }); }); -const navRunStatus = computed(() => { +const navSafetyText = computed(() => { + if (!userSettings.manualRunAllowed) { + return "Manual checks disabled"; + } + if (runStatus.safetyState.cooldown_active) { + return `Cooldown: ${formatCooldownCountdown(runStatus.safetyState.cooldown_remaining_seconds)}`; + } + return "Safety ready"; +}); +const navRunStateStatus = computed(() => { + if (runStatus.isSubmitting && !runStatus.isLikelyRunning) { + return "starting"; + } if (runStatus.isLikelyRunning) { return "running"; } return "idle"; }); -const navRunText = computed(() => { - const label = navRunStatus.value - .replaceAll("_", " ") - .split(" ") +const navRunStateLabel = computed(() => + navRunStateStatus.value + .split("_") .filter((segment) => segment.length > 0) .map((segment) => segment.charAt(0).toUpperCase() + segment.slice(1)) - .join(" "); - return `State: ${label}`; -}); + .join(" "), +); +const showSafetyRow = computed( + () => runStatus.safetyState.cooldown_active || !userSettings.manualRunAllowed, +); async function onLogout(): Promise { await auth.logout(); @@ -71,12 +86,18 @@ async function onLogout(): Promise {
-
- {{ navRunText }} - +
+
+ State: {{ navRunStateLabel }} + +
+
+ {{ navSafetyText }} + +
Logout
diff --git a/frontend/src/components/patterns/RunStatusBadge.vue b/frontend/src/components/patterns/RunStatusBadge.vue index b79aabf..575b5c7 100644 --- a/frontend/src/components/patterns/RunStatusBadge.vue +++ b/frontend/src/components/patterns/RunStatusBadge.vue @@ -20,6 +20,9 @@ const toneClass = computed(() => { if (props.status === "running") { return "border-state-info-border bg-state-info-bg text-state-info-text"; } + if (props.status === "starting") { + return "border-state-info-border bg-state-info-bg text-state-info-text"; + } return "border-stroke-default bg-surface-card-muted text-ink-secondary"; }); diff --git a/frontend/src/components/patterns/ScrapeSafetyBadge.vue b/frontend/src/components/patterns/ScrapeSafetyBadge.vue new file mode 100644 index 0000000..b40c1c9 --- /dev/null +++ b/frontend/src/components/patterns/ScrapeSafetyBadge.vue @@ -0,0 +1,27 @@ + + + diff --git a/frontend/src/components/patterns/ScrapeSafetyBanner.vue b/frontend/src/components/patterns/ScrapeSafetyBanner.vue new file mode 100644 index 0000000..300b05a --- /dev/null +++ b/frontend/src/components/patterns/ScrapeSafetyBanner.vue @@ -0,0 +1,102 @@ + + + diff --git a/frontend/src/features/dashboard/index.ts b/frontend/src/features/dashboard/index.ts index 94d546a..78affe4 100644 --- a/frontend/src/features/dashboard/index.ts +++ b/frontend/src/features/dashboard/index.ts @@ -4,6 +4,7 @@ import { type PublicationMode, } from "@/features/publications"; import { listQueueItems, listRuns, type RunListItem } from "@/features/runs"; +import { type ScrapeSafetyState } from "@/features/safety"; export interface QueueHealth { queued: number; @@ -19,6 +20,7 @@ export interface DashboardSnapshot { recentRuns: RunListItem[]; recentPublications: PublicationItem[]; queue: QueueHealth; + safetyState: ScrapeSafetyState; } function countQueueStatuses(statuses: string[]): QueueHealth { @@ -38,11 +40,12 @@ function countQueueStatuses(statuses: string[]): QueueHealth { } export async function fetchDashboardSnapshot(): Promise { - const [publications, runs, queueItems] = await Promise.all([ + const [publications, runsPayload, queueItems] = await Promise.all([ listPublications({ mode: "new", limit: 20 }), listRuns({ limit: 20 }), listQueueItems(200), ]); + const runs = runsPayload.runs; const queueHealth = countQueueStatuses(queueItems.map((item) => item.status)); @@ -54,5 +57,6 @@ export async function fetchDashboardSnapshot(): Promise { recentRuns: runs, recentPublications: publications.publications, queue: queueHealth, + safetyState: runsPayload.safety_state, }; } diff --git a/frontend/src/features/runs/index.ts b/frontend/src/features/runs/index.ts index 4fe4add..97d49c6 100644 --- a/frontend/src/features/runs/index.ts +++ b/frontend/src/features/runs/index.ts @@ -1,4 +1,5 @@ import { apiRequest } from "@/lib/api/client"; +import { type ScrapeSafetyState } from "@/features/safety"; export interface RunListItem { id: number; @@ -49,6 +50,7 @@ export interface RunDetail { run: RunListItem; summary: RunSummary; scholar_results: RunScholarResult[]; + safety_state: ScrapeSafetyState; } export interface QueueItem { @@ -68,6 +70,7 @@ export interface QueueItem { interface RunsListData { runs: RunListItem[]; + safety_state: ScrapeSafetyState; } interface QueueListData { @@ -79,7 +82,7 @@ export interface RunsListQuery { limit?: number; } -export async function listRuns(query: RunsListQuery = {}): Promise { +export async function listRuns(query: RunsListQuery = {}): Promise { const params = new URLSearchParams(); if (query.failedOnly) { params.set("failed_only", "true"); @@ -92,7 +95,7 @@ export async function listRuns(query: RunsListQuery = {}): Promise(`/runs${suffix ? `?${suffix}` : ""}`, { method: "GET", }); - return response.data.runs; + return response.data; } export async function getRunDetail(runId: number): Promise { @@ -118,6 +121,7 @@ export async function triggerManualRun(): Promise<{ new_publication_count: number; reused_existing_run: boolean; idempotency_key: string | null; + safety_state: ScrapeSafetyState; }> { const headers: Record = { "Idempotency-Key": generateIdempotencyKey(), @@ -133,6 +137,7 @@ export async function triggerManualRun(): Promise<{ new_publication_count: number; reused_existing_run: boolean; idempotency_key: string | null; + safety_state: ScrapeSafetyState; }>("/runs/manual", { method: "POST", headers, diff --git a/frontend/src/features/safety/index.ts b/frontend/src/features/safety/index.ts new file mode 100644 index 0000000..185540b --- /dev/null +++ b/frontend/src/features/safety/index.ts @@ -0,0 +1,112 @@ +export interface ScrapeSafetyCounters { + consecutive_blocked_runs: number; + consecutive_network_runs: number; + cooldown_entry_count: number; + blocked_start_count: number; + last_blocked_failure_count: number; + last_network_failure_count: number; + last_evaluated_run_id: number | null; +} + +export interface ScrapeSafetyState { + cooldown_active: boolean; + cooldown_reason: string | null; + cooldown_reason_label: string | null; + cooldown_until: string | null; + cooldown_remaining_seconds: number; + recommended_action: string | null; + counters: ScrapeSafetyCounters; +} + +export function createDefaultSafetyCounters(): ScrapeSafetyCounters { + return { + consecutive_blocked_runs: 0, + consecutive_network_runs: 0, + cooldown_entry_count: 0, + blocked_start_count: 0, + last_blocked_failure_count: 0, + last_network_failure_count: 0, + last_evaluated_run_id: null, + }; +} + +export function createDefaultSafetyState(): ScrapeSafetyState { + return { + cooldown_active: false, + cooldown_reason: null, + cooldown_reason_label: null, + cooldown_until: null, + cooldown_remaining_seconds: 0, + recommended_action: null, + counters: createDefaultSafetyCounters(), + }; +} + +function parseNumber(value: unknown, fallback: number): number { + if (typeof value === "number" && Number.isFinite(value)) { + return value; + } + if (typeof value === "string" && value.trim().length > 0) { + const parsed = Number(value); + if (Number.isFinite(parsed)) { + return parsed; + } + } + return fallback; +} + +function parseNullableString(value: unknown): string | null { + return typeof value === "string" && value.trim().length > 0 ? value : null; +} + +export function normalizeSafetyState(value: unknown): ScrapeSafetyState { + if (!value || typeof value !== "object") { + return createDefaultSafetyState(); + } + + const raw = value as Record; + const rawCounters = raw.counters; + const counters = rawCounters && typeof rawCounters === "object" + ? (rawCounters as Record) + : {}; + + return { + cooldown_active: Boolean(raw.cooldown_active), + cooldown_reason: parseNullableString(raw.cooldown_reason), + cooldown_reason_label: parseNullableString(raw.cooldown_reason_label), + cooldown_until: parseNullableString(raw.cooldown_until), + cooldown_remaining_seconds: Math.max(0, parseNumber(raw.cooldown_remaining_seconds, 0)), + recommended_action: parseNullableString(raw.recommended_action), + counters: { + consecutive_blocked_runs: Math.max(0, parseNumber(counters.consecutive_blocked_runs, 0)), + consecutive_network_runs: Math.max(0, parseNumber(counters.consecutive_network_runs, 0)), + cooldown_entry_count: Math.max(0, parseNumber(counters.cooldown_entry_count, 0)), + blocked_start_count: Math.max(0, parseNumber(counters.blocked_start_count, 0)), + last_blocked_failure_count: Math.max(0, parseNumber(counters.last_blocked_failure_count, 0)), + last_network_failure_count: Math.max(0, parseNumber(counters.last_network_failure_count, 0)), + last_evaluated_run_id: + counters.last_evaluated_run_id === null + ? null + : Math.max(0, parseNumber(counters.last_evaluated_run_id, 0)), + }, + }; +} + +export function formatCooldownCountdown(seconds: number): string { + const bounded = Number.isFinite(seconds) ? Math.max(0, Math.floor(seconds)) : 0; + if (bounded <= 0) { + return "0s"; + } + + const hours = Math.floor(bounded / 3600); + const minutes = Math.floor((bounded % 3600) / 60); + const secs = bounded % 60; + + if (hours > 0) { + return `${hours}h ${minutes}m`; + } + if (minutes > 0) { + return `${minutes}m ${secs}s`; + } + return `${secs}s`; +} diff --git a/frontend/src/features/settings/index.ts b/frontend/src/features/settings/index.ts index e809ada..e8ef325 100644 --- a/frontend/src/features/settings/index.ts +++ b/frontend/src/features/settings/index.ts @@ -1,10 +1,31 @@ import { apiRequest } from "@/lib/api/client"; +import { type ScrapeSafetyState } from "@/features/safety"; + +export interface UserSettingsPolicy { + min_run_interval_minutes: number; + min_request_delay_seconds: number; + automation_allowed: boolean; + manual_run_allowed: boolean; + blocked_failure_threshold: number; + network_failure_threshold: number; + cooldown_blocked_seconds: number; + cooldown_network_seconds: number; +} export interface UserSettings { auto_run_enabled: boolean; run_interval_minutes: number; request_delay_seconds: number; nav_visible_pages: string[]; + policy: UserSettingsPolicy; + safety_state: ScrapeSafetyState; +} + +export interface UserSettingsUpdate { + auto_run_enabled: boolean; + run_interval_minutes: number; + request_delay_seconds: number; + nav_visible_pages: string[]; } export interface ChangePasswordPayload { @@ -18,7 +39,7 @@ export async function fetchSettings(): Promise { return response.data; } -export async function updateSettings(payload: UserSettings): Promise { +export async function updateSettings(payload: UserSettingsUpdate): Promise { const response = await apiRequest("/settings", { method: "PUT", body: payload, diff --git a/frontend/src/pages/DashboardPage.vue b/frontend/src/pages/DashboardPage.vue index d6bf8d3..211ce9e 100644 --- a/frontend/src/pages/DashboardPage.vue +++ b/frontend/src/pages/DashboardPage.vue @@ -1,5 +1,5 @@