Surface engine cooldown and classify network-level blocks on /monitor

README.md

@@ -183,10 +183,10 @@ http://localhost:5000
 A built-in observability console at **`/monitor`** (linked from the wire masthead) gives a per-engine, per-cycle view of scraper health — no Prometheus/Grafana stack required, since the data already lives in Postgres. It polls [`GET /api/v1/metrics`](docs/API_REFERENCE.md#metrics) over a selectable window (1h / 6h / 24h) and shows:
 
 - **Overall strip** - active topics, total filed, scrape success rate, feed freshness, last-cycle duration, and scrapes-in-window (blocked / failed).
-- **Engines table** - one row per engine with a health label (`healthy` / `degraded` / `blocked` / `parsing` / `idle`), success %, fetch latency (avg / p95), items parsed, 0-parse count (selector-rot signal), blocks (429/403/503), and last HTTP status.
+- **Engines table** - one row per engine with a health label (`healthy` / `degraded` / `blocked` / `parsing` / `cooldown` / `idle`), success %, fetch latency (avg / p95), items parsed, 0-parse count (selector-rot signal), blocks (429/403/503), and last HTTP status. `blocked` now also covers connection-level teardowns with no HTTP status (e.g. Yahoo's `ERR_CONNECTION_CLOSED`).
 - **Recent cycles** - a sparkline of per-cycle durations plus a list (duration, topics, parsed, new events).
 
-A throttled engine shows up here rather than silently vanishing from the feed — below, Brave reads `degraded` with its block count while Google/Bing/Yahoo stay `healthy`. See [Observability](docs/OBSERVABILITY.md) for details.
+A throttled engine shows up here rather than silently vanishing from the feed. When the scraper benches an engine, the table shows `cooldown` with a countdown to the next probe — so an engine that produces no scrapes while cooling stays visible instead of disappearing. See [Observability](docs/OBSERVABILITY.md) for details.
 
 <p align="center">
 <img src="docs/pic/ui_monitor_screenshot.png" alt="TopicStreams /monitor ops page - per-engine health table and recent-cycle timeline" width="600"/>

api/static/monitor.html

@@ -103,7 +103,7 @@ <h2>Recent failures</h2>
     <template id="engine-row-template">
         <tr>
             <td class="c-engine"><span class="edot"></span><span class="ename"></span></td>
-            <td class="c-health"><span class="ehealth"></span></td>
+            <td class="c-health"><span class="ehealth"></span><span class="ecooldown"></span></td>
             <td class="c-scrapes num r"></td>
             <td class="c-success num r"></td>
             <td class="c-latency num r"></td>

api/static/monitor.js

@@ -110,6 +110,18 @@ class MonitorApp {
             hl.dataset.health = e.health;
             row.querySelector('.c-health').dataset.health = e.health;
 
+            // Cooldown: show the countdown to the next probe beside the label.
+            const cd = row.querySelector('.ecooldown');
+            if (e.cooldown_seconds_remaining != null) {
+                cd.textContent = `~${this.fmtCountdown(e.cooldown_seconds_remaining)}`;
+                const blocks = e.cooldown_failures
+                    ? `${e.cooldown_failures} consecutive block${e.cooldown_failures === 1 ? '' : 's'}; `
+                    : '';
+                cd.title = `${blocks}probing again in ~${this.fmtCountdown(e.cooldown_seconds_remaining)}`;
+            } else {
+                cd.textContent = '';
+            }
+
             row.querySelector('.c-scrapes').textContent = this.fmtInt(e.scrapes);
             row.querySelector('.c-success').textContent = this.fmtPct(e.success_rate);
             row.querySelector('.c-latency').textContent = `${this.fmtMs(e.avg_latency_ms)} / ${this.fmtMs(e.p95_latency_ms)}`;
@@ -233,6 +245,13 @@ class MonitorApp {
         return ms >= 1000 ? `${(ms / 1000).toFixed(1)}s` : `${ms}ms`;
     }
 
+    fmtCountdown(s) {
+        if (s == null) return '–';
+        if (s < 90) return `${Math.round(s)}s`;
+        if (s < 3600) return `${Math.round(s / 60)}m`;
+        return `${(s / 3600).toFixed(1)}h`;
+    }
+
     fmtFresh(s) {
         if (s == null) return '–';
         if (s < 60) return `${Math.round(s)}s`;

api/static/styles.css

@@ -507,14 +507,21 @@ a { color: inherit; }
 .edot[data-health="healthy"] { background: var(--good); }
 .edot[data-health="degraded"] { background: var(--warn); }
 .edot[data-health="blocked"], .edot[data-health="parsing"] { background: var(--signal); }
+.edot[data-health="cooldown"] { background: var(--warn); }
 .edot[data-health="idle"] { background: var(--ink-faint); }
 
 .ehealth { font-size: 10.5px; letter-spacing: 0.08em; text-transform: uppercase; }
 .ehealth[data-health="healthy"] { color: var(--good); }
 .ehealth[data-health="degraded"] { color: var(--warn); }
 .ehealth[data-health="blocked"], .ehealth[data-health="parsing"] { color: var(--signal); }
+.ehealth[data-health="cooldown"] { color: var(--warn); }
 .ehealth[data-health="idle"] { color: var(--ink-faint); }
 
+/* Probe countdown shown beside a "cooldown" label, e.g. "~286s". Not uppercased
+   (it's data, not a status word) and de-emphasized vs. the label. */
+.ecooldown { margin-left: 6px; font-size: 10.5px; color: var(--ink-soft);
+  font-variant-numeric: tabular-nums; }
+
 td.hot { color: var(--signal); font-weight: 600; }
 td.warm { color: var(--warn); font-weight: 600; }
 

api/v1/metrics.py

@@ -14,6 +14,7 @@
 from starlette.concurrency import run_in_threadpool
 
 from common import database as db
+from common.block_signals import is_network_block
 
 router = APIRouter(prefix="/metrics", tags=["metrics"])
 
@@ -27,6 +28,12 @@
 # unavailable). Drives the per-engine "blocked" health label.
 _BLOCK_STATUSES = (429, 403, 503)
 
+# A persisted cooldown snapshot older than this is ignored: the scraper writes
+# one every cycle, so a stale snapshot means the scraper is down (the cycle
+# timeline surfaces that) and the cooldown countdown is meaningless. Generous
+# vs. the scrape interval so a slow cycle doesn't blink the indicator off.
+_COOLDOWN_STALE_SECONDS = 600
+
 
 def _naive_utc_now() -> datetime:
     return datetime.now(timezone.utc).replace(tzinfo=None)
@@ -40,7 +47,9 @@ def classify_engine(row: dict) -> str:
     """Per-engine health label from an aggregate row.
 
     idle    - no scrapes in the window
-    blocked - the most recent scrape was a throttle/block (429/403/503)
+    blocked - the most recent scrape was a throttle/block: a 429/403/503 status,
+              or a connection-level teardown with no HTTP status (e.g. Yahoo's
+              ERR_CONNECTION_CLOSED — see common/block_signals.py)
     parsing - sustained selector rot: every successful scrape parsed nothing
               (>=3 scrapes, >=1 success, all successes 0-entry)
     degraded- success rate below 0.75 (includes a total failure: 0%)
@@ -53,9 +62,9 @@ def classify_engine(row: dict) -> str:
     scrapes = row["scrapes"]
     if scrapes == 0:
         return "idle"
-    if (
-        row.get("last_success") is False
-        and row.get("last_http_status") in _BLOCK_STATUSES
+    if row.get("last_success") is False and (
+        row.get("last_http_status") in _BLOCK_STATUSES
+        or is_network_block(row.get("last_error_message"))
     ):
         return "blocked"
     successes = row["successes"]
@@ -65,6 +74,46 @@ def classify_engine(row: dict) -> str:
     return "degraded" if rate < 0.75 else "healthy"
 
 
+def _live_cooldown_seconds(cd: dict | None, now: datetime) -> float | None:
+    """Seconds until the next probe if this engine is *currently* benched.
+
+    None when not cooling, when the snapshot is stale (scraper likely down), or
+    when the probe is already due (remaining <= 0) — in those cases the
+    log-derived health label stands on its own.
+    """
+    if not cd or cd["failures"] <= 0 or cd["next_probe_at"] is None:
+        return None
+    if (now - cd["updated_at"]).total_seconds() > _COOLDOWN_STALE_SECONDS:
+        return None
+    remaining = (cd["next_probe_at"] - now).total_seconds()
+    return remaining if remaining > 0 else None
+
+
+def _empty_engine_row(engine: str) -> dict:
+    """A zeroed aggregate row for an engine with no scrapes in the window.
+
+    Lets an engine that is benched (hence producing no logs) still appear on the
+    monitor — otherwise a long cooldown would make it vanish from the table.
+    """
+    return {
+        "engine": engine,
+        "scrapes": 0,
+        "successes": 0,
+        "entries_parsed": 0,
+        "zero_parse": 0,
+        "failures": 0,
+        "blocked": 0,
+        "avg_latency_ms": None,
+        "p50_latency_ms": None,
+        "p95_latency_ms": None,
+        "last_scrape_at": None,
+        "last_success": None,
+        "last_http_status": None,
+        "last_error_message": None,
+        "http_status_breakdown": {},
+    }
+
+
 class OverallMetrics(BaseModel):
     scrapes: int = Field(..., description="Page-attempts in the window")
     successes: int = Field(..., description="Successful page-attempts")
@@ -87,7 +136,10 @@ class EngineMetrics(BaseModel):
     engine: str = Field(..., description="Search engine name")
     health: str = Field(
         ...,
-        description="healthy | degraded | blocked | parsing | idle (see classify_engine)",
+        description=(
+            "healthy | degraded | blocked | parsing | idle (see classify_engine), "
+            "or cooldown when the scraper has the engine benched"
+        ),
     )
     scrapes: int
     successes: int
@@ -107,6 +159,15 @@ class EngineMetrics(BaseModel):
     http_status_breakdown: dict[str, int] = Field(
         ..., description="Counts per monitored HTTP status (200/429/403/503)"
     )
+    cooldown_seconds_remaining: float | None = Field(
+        None,
+        description="Seconds until the scraper next probes a benched engine "
+        "(null when not cooling down)",
+    )
+    cooldown_failures: int = Field(
+        0,
+        description="Consecutive block signals driving the current cooldown (0 if none)",
+    )
 
 
 class CycleMetrics(BaseModel):
@@ -173,10 +234,23 @@ def _build_overall(row: dict) -> OverallMetrics:
     )
 
 
-def _build_engine(row: dict) -> EngineMetrics:
+def _build_engine(row: dict, cooldown: dict | None = None) -> EngineMetrics:
+    """Shape one engine row. ``cooldown`` is its live ``{failures, remaining}``
+    when currently benched, which overrides the log-derived health label."""
+    health = classify_engine(row)
+    cooldown_remaining = None
+    cooldown_failures = 0
+    if cooldown is not None:
+        cooldown_remaining = cooldown["remaining"]
+        cooldown_failures = cooldown["failures"]
+        # The engine is benched right now; that supersedes a stale log-based
+        # label (often "idle" — no logs are produced while skipped).
+        health = "cooldown"
     return EngineMetrics(
         engine=row["engine"],
-        health=classify_engine(row),
+        health=health,
+        cooldown_seconds_remaining=cooldown_remaining,
+        cooldown_failures=cooldown_failures,
         scrapes=row["scrapes"],
         successes=row["successes"],
         success_rate=_rate(row["successes"], row["scrapes"]),
@@ -209,9 +283,24 @@ async def get_metrics(
     metrics = await run_in_threadpool(db.get_scrape_metrics, window)
     cycles = await run_in_threadpool(db.get_recent_cycles, _RECENT_CYCLES)
     failures = await run_in_threadpool(db.get_recent_scrape_failures, _RECENT_FAILURES)
+    cooldowns = await run_in_threadpool(db.get_engine_cooldowns)
 
     overall = _build_overall(metrics["overall"])
-    engines = [_build_engine(row) for row in metrics["engines"]]
+
+    # Resolve which engines are *currently* benched (fresh snapshot, probe still
+    # in the future), then overlay that onto the per-engine rows.
+    now = _naive_utc_now()
+    live = {
+        engine: {"remaining": remaining, "failures": cd["failures"]}
+        for engine, cd in cooldowns.items()
+        if (remaining := _live_cooldown_seconds(cd, now)) is not None
+    }
+    rows = {row["engine"]: row for row in metrics["engines"]}
+    # A long cooldown means no scrapes in the window, so a benched engine may be
+    # absent from the aggregates entirely — synthesize a row so it still shows.
+    for engine in live:
+        rows.setdefault(engine, _empty_engine_row(engine))
+    engines = [_build_engine(rows[engine], live.get(engine)) for engine in sorted(rows)]
 
     return MetricsResponse(
         active_topics=len(topics),

common/block_signals.py

@@ -0,0 +1,42 @@
+"""Shared block-signal detection used by both the scraper and the API.
+
+Some engines block at the network layer rather than with an HTTP status: the
+server (or an upstream proxy) tears down the TCP connection, so the browser
+nav fails with a ``net::ERR_CONNECTION_*`` error and there is *no* HTTP
+response to read a 429/403/503 from. Yahoo does exactly this — after a burst it
+serves a persistent empty HTTP 500 (Server: ATS, Connection: close) and a real
+browser nav then fails ``net::ERR_CONNECTION_CLOSED`` (see
+docs/BLOCK_SIGNAL_FINDINGS.md). Treating these as a block (not a transient
+glitch) lets the scraper bench the engine and the monitor label it correctly.
+
+Deliberately narrow: only connection-level *refusals/teardowns* count. A plain
+navigation timeout (``TimeoutError``) is a transient failure we neither punish
+(cooldown) nor flag (health) — it stays "other"/non-block, matching the
+existing contract in scraper/cooldown.py.
+"""
+
+# Substrings matched case-insensitively against a scrape's error message
+# (``"{ExceptionType}: {message}"``). These are Chromium/Playwright net errors
+# that mean the connection was refused or torn down before a usable response —
+# a server actively closing the door, i.e. a block.
+NETWORK_BLOCK_PATTERNS = (
+    "err_connection_closed",
+    "err_connection_reset",
+    "err_connection_refused",
+    "err_connection_aborted",
+    "err_connection_failed",
+    "err_socket_not_connected",
+    "err_empty_response",
+)
+
+
+def is_network_block(error_message: str | None) -> bool:
+    """True if ``error_message`` looks like a connection-level block/teardown.
+
+    Returns False for ``None``/empty and for transient errors such as
+    navigation timeouts.
+    """
+    if not error_message:
+        return False
+    haystack = error_message.lower()
+    return any(pattern in haystack for pattern in NETWORK_BLOCK_PATTERNS)

common/database.py

@@ -176,6 +176,16 @@ def ensure_schema() -> None:
                 "CREATE INDEX IF NOT EXISTS idx_scraper_cycles_started_at "
                 "ON scraper_cycles(started_at DESC)"
             )
+            cursor.execute(
+                """
+                CREATE TABLE IF NOT EXISTS engine_cooldowns (
+                    engine VARCHAR(32) PRIMARY KEY,
+                    failures INTEGER NOT NULL DEFAULT 0,
+                    next_probe_at TIMESTAMP,
+                    updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+                )
+                """
+            )
         _schema_ready = True
 
 
@@ -665,6 +675,61 @@ def get_recent_cycles(limit: int = 24) -> list[dict]:
         return [dict(row) for row in cursor.fetchall()]
 
 
+# ── Engine cooldown snapshot (engine_cooldowns) ──────────────────────────────
+
+
+# Idempotent: each row is a full overwrite keyed by engine, so a retry after a
+# committed write just reapplies the same absolute state — safe.
+@retry_on_transient_error()
+def upsert_engine_cooldowns(
+    states: "list[tuple[str, int, float]]",
+) -> None:
+    """Persist the scraper's per-engine cooldown snapshot for the monitor.
+
+    ``states`` is ``(engine, failures, remaining_seconds)`` per tracked engine.
+    ``next_probe_at`` is stored as absolute UTC (``NOW() + remaining``) using the
+    *DB* clock so the API can derive a fresh countdown without trusting the
+    scraper's wall-clock; a non-cooling engine (failures == 0) stores NULL.
+    """
+    if not states:
+        return
+    with _Connection() as conn:
+        cursor = conn.cursor()
+        cursor.executemany(
+            "INSERT INTO engine_cooldowns (engine, failures, next_probe_at, updated_at) "
+            "VALUES (%s, %s, "
+            "        CASE WHEN %s > 0 THEN NOW() + make_interval(secs => %s) END, "
+            "        NOW()) "
+            "ON CONFLICT (engine) DO UPDATE SET "
+            "  failures = EXCLUDED.failures, "
+            "  next_probe_at = EXCLUDED.next_probe_at, "
+            "  updated_at = EXCLUDED.updated_at",
+            [(e, f, f, max(0.0, r)) for (e, f, r) in states],
+        )
+
+
+@retry_on_transient_error()
+def get_engine_cooldowns() -> dict[str, dict]:
+    """Per-engine cooldown state keyed by engine name (for the monitor).
+
+    Returns ``{engine: {failures, next_probe_at, updated_at}}``. The endpoint
+    decides what is still "live" (freshness + remaining); this is a plain read.
+    """
+    with _Connection() as conn:
+        cursor = conn.cursor()
+        cursor.execute(
+            "SELECT engine, failures, next_probe_at, updated_at FROM engine_cooldowns"
+        )
+        return {
+            row["engine"]: {
+                "failures": row["failures"],
+                "next_probe_at": row["next_probe_at"],
+                "updated_at": row["updated_at"],
+            }
+            for row in cursor.fetchall()
+        }
+
+
 # ── Scrape metrics aggregation (scraper_logs) ────────────────────────────────
 
 # Shared aggregate columns for the overall + per-engine rollups. duration_ms is
@@ -687,7 +752,8 @@ def get_recent_cycles(limit: int = 24) -> list[dict]:
     percentile_cont(0.95) WITHIN GROUP (ORDER BY duration_ms) AS p95_latency_ms,
     MAX(scraped_at) AS last_scrape_at,
     (ARRAY_AGG(success ORDER BY scraped_at DESC))[1] AS last_success,
-    (ARRAY_AGG(http_status_code ORDER BY scraped_at DESC))[1] AS last_http_status
+    (ARRAY_AGG(http_status_code ORDER BY scraped_at DESC))[1] AS last_http_status,
+    (ARRAY_AGG(error_message ORDER BY scraped_at DESC))[1] AS last_error_message
 """
 
 _METRICS_WHERE = "scraped_at >= NOW() - make_interval(secs => %s)"
@@ -717,6 +783,7 @@ def _ms(value) -> int | None:
         "last_scrape_at": row["last_scrape_at"],
         "last_success": row["last_success"],
         "last_http_status": row["last_http_status"],
+        "last_error_message": row["last_error_message"],
         "http_status_breakdown": breakdown,
     }