From 4aee18dcbeafcda56fa3b28ed5e455a3f86c8759 Mon Sep 17 00:00:00 2001 From: Yiorgis Gozadinos Date: Mon, 25 May 2026 17:21:48 +0300 Subject: [PATCH] Operator dashboard at GET /; tighten Logfire span shape MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Self-contained HTML status page served from the ingester's FastAPI app. Polls /health, /sources, /stats, /jobs?status={claimed,dead,succeeded} every 3s from the browser and renders queue chips, sources with last-poll/skip-reason/circuit state, active jobs with cancel, recent failures with retry, and recently-completed feed with op badges so DELETE rows are visually distinct from UPSERTs. Zero external deps — single static HTML, no CDN, no fonts, no images. Works offline. To support the dashboard: - New /stats endpoint exposing rolling throughput (5m/30m/1h), worker occupancy, oldest-queued age, and per-source DLQ + queue-depth breakdowns. Each field is a single SQL aggregation against the queue. - JobRepo gains count_succeeded_since, oldest_queued_age_seconds, counts_by_source. - SourceSummary gains last_skip_reason. BasePoller now records the reason the most recent sweep attempt was skipped ("pending_work" / "circuit_open"), cleared on the next successful poll. Closes the gap where operators couldn't tell from /sources alone why a source wasn't picking up new work. Auth: dashboard route is unauthenticated (markup only). The JS attaches the bearer to its own JSON fetches; on 401 it prompts once and stashes the token in localStorage. Two Logfire fixes that landed alongside: - Drop logfire.instrument_fastapi() and the [fastapi] extra. The control plane is polled frequently (dashboard + docker healthcheck), so every endpoint became a span and drowned the useful traces. logfire itself stays — pulled in transitively via pydantic-ai-slim[logfire] — so ingester.poller.* / ingester.job / document.* spans keep emitting. - Wrap FSPoller._handle_watch_change in an ingester.poller.watch_event span and pass _enqueue_extra. Without this, the watchfiles callback ran with no active context, the _otel carrier in job.extra was empty, and the worker's ingester.job span surfaced as an orphan trace root instead of nesting under the FS event that caused it. --- .../rag/ingester/api/routes/dashboard.py | 18 + .../haiku/rag/ingester/api/routes/sources.py | 1 + .../haiku/rag/ingester/api/routes/stats.py | 38 ++ .../haiku/rag/ingester/api/schemas.py | 26 + .../haiku/rag/ingester/api/server.py | 19 +- .../haiku/rag/ingester/api/static/index.html | 553 ++++++++++++++++++ .../haiku/rag/ingester/pollers/base.py | 11 + .../haiku/rag/ingester/pollers/fs.py | 58 +- .../haiku/rag/ingester/queue/repository.py | 43 ++ haiku_rag_slim/pyproject.toml | 1 - tests/ingester/test_api.py | 76 +++ tests/ingester/test_pollers.py | 22 + tests/ingester/test_queue.py | 77 +++ uv.lock | 46 -- 14 files changed, 914 insertions(+), 75 deletions(-) create mode 100644 haiku_rag_slim/haiku/rag/ingester/api/routes/dashboard.py create mode 100644 haiku_rag_slim/haiku/rag/ingester/api/routes/stats.py create mode 100644 haiku_rag_slim/haiku/rag/ingester/api/static/index.html diff --git a/haiku_rag_slim/haiku/rag/ingester/api/routes/dashboard.py b/haiku_rag_slim/haiku/rag/ingester/api/routes/dashboard.py new file mode 100644 index 00000000..892c2fc2 --- /dev/null +++ b/haiku_rag_slim/haiku/rag/ingester/api/routes/dashboard.py @@ -0,0 +1,18 @@ +from pathlib import Path + +from fastapi import APIRouter +from fastapi.responses import HTMLResponse + +router = APIRouter(tags=["dashboard"]) + +_STATIC_DIR = Path(__file__).resolve().parent.parent / "static" +_INDEX_HTML = (_STATIC_DIR / "index.html").read_text(encoding="utf-8") + + +@router.get("/", response_class=HTMLResponse, include_in_schema=False) +async def dashboard() -> HTMLResponse: + """Serve the operator dashboard. Static page that polls /health, /sources, + /stats and /jobs from the browser. Auth happens via the bearer header the + JS attaches to its fetches — the dashboard route itself is unauthenticated + so an operator can open it in a browser and paste the token on demand.""" + return HTMLResponse(content=_INDEX_HTML) diff --git a/haiku_rag_slim/haiku/rag/ingester/api/routes/sources.py b/haiku_rag_slim/haiku/rag/ingester/api/routes/sources.py index 5bd45082..b26f2e3e 100644 --- a/haiku_rag_slim/haiku/rag/ingester/api/routes/sources.py +++ b/haiku_rag_slim/haiku/rag/ingester/api/routes/sources.py @@ -20,6 +20,7 @@ async def list_sources( type=type(poller.config).__name__, last_polled_at=poller.last_polled_at, circuit_breaker_open=poller.is_circuit_open, + last_skip_reason=poller.last_skip_reason, ) ) return summaries diff --git a/haiku_rag_slim/haiku/rag/ingester/api/routes/stats.py b/haiku_rag_slim/haiku/rag/ingester/api/routes/stats.py new file mode 100644 index 00000000..5f0294aa --- /dev/null +++ b/haiku_rag_slim/haiku/rag/ingester/api/routes/stats.py @@ -0,0 +1,38 @@ +from fastapi import APIRouter, Depends + +from haiku.rag.ingester.api.schemas import ( + StatsResponse, + ThroughputStats, + WorkerStats, +) +from haiku.rag.ingester.api.server import APIState, get_state + +router = APIRouter(tags=["stats"]) + + +@router.get("/stats", response_model=StatsResponse) +async def stats(state: APIState = Depends(get_state)) -> StatsResponse: + """Dashboard summary: rolling throughput, worker occupancy, backlog age, + and per-source DLQ / queue depth. Each field is a single SQL aggregation + against the queue file — cheap to call every few seconds.""" + jobs = state.job_repo + + counts = await jobs.counts_by_status() + worker_total = ( + state.config.ingester.workers.worker_count if state.pool is not None else 0 + ) + + return StatsResponse( + throughput=ThroughputStats( + succeeded_5m=await jobs.count_succeeded_since(300), + succeeded_30m=await jobs.count_succeeded_since(1800), + succeeded_1h=await jobs.count_succeeded_since(3600), + ), + workers=WorkerStats( + busy=counts.get("claimed", 0), + total=worker_total, + ), + oldest_queued_age_s=await jobs.oldest_queued_age_seconds(), + dlq_by_source=await jobs.counts_by_source("dead"), + queue_depth_by_source=await jobs.counts_by_source("queued", "claimed"), + ) diff --git a/haiku_rag_slim/haiku/rag/ingester/api/schemas.py b/haiku_rag_slim/haiku/rag/ingester/api/schemas.py index c57742dc..6a94ea24 100644 --- a/haiku_rag_slim/haiku/rag/ingester/api/schemas.py +++ b/haiku_rag_slim/haiku/rag/ingester/api/schemas.py @@ -15,6 +15,10 @@ class SourceSummary(BaseModel): type: str last_polled_at: datetime | None circuit_breaker_open: bool + # Reason the most recent sweep attempt was skipped (e.g. "pending_work"), + # or None when the most recent attempt actually polled. Lets operators + # see at a glance why a source isn't picking up new work. + last_skip_reason: str | None = None class RefreshResponse(BaseModel): @@ -25,3 +29,25 @@ class RefreshResponse(BaseModel): class CancelResponse(BaseModel): job_id: str cancelled: bool + + +class ThroughputStats(BaseModel): + succeeded_5m: int + succeeded_30m: int + succeeded_1h: int + + +class WorkerStats(BaseModel): + busy: int + total: int + + +class StatsResponse(BaseModel): + """Aggregated counters and per-source breakdowns that drive the dashboard. + Cheap to compute (all SQL aggregations against the queue file).""" + + throughput: ThroughputStats + workers: WorkerStats + oldest_queued_age_s: float | None + dlq_by_source: dict[str, int] + queue_depth_by_source: dict[str, int] diff --git a/haiku_rag_slim/haiku/rag/ingester/api/server.py b/haiku_rag_slim/haiku/rag/ingester/api/server.py index 6a04b34f..f29f981c 100644 --- a/haiku_rag_slim/haiku/rag/ingester/api/server.py +++ b/haiku_rag_slim/haiku/rag/ingester/api/server.py @@ -1,7 +1,6 @@ from dataclasses import dataclass from typing import TYPE_CHECKING -import logfire from fastapi import Depends, FastAPI, Request from haiku.rag.ingester.api.auth import require_auth @@ -35,7 +34,14 @@ def build_app( auth_token: str | None = None, ) -> FastAPI: """Construct the ingester's FastAPI control plane.""" - from haiku.rag.ingester.api.routes import dlq, health, jobs, sources + from haiku.rag.ingester.api.routes import ( + dashboard, + dlq, + health, + jobs, + sources, + stats, + ) app = FastAPI( title="haiku-ingester", @@ -47,12 +53,13 @@ def build_app( auth_dep = [Depends(require_auth)] app.include_router(health.router) # /health is unauthenticated by design + # Dashboard route is markup-only; the JS it serves attaches the bearer + # token to its own fetches. Keeping the route unauthenticated lets an + # operator open it in a browser and paste the token on demand. + app.include_router(dashboard.router) app.include_router(jobs.router, dependencies=auth_dep) app.include_router(sources.router, dependencies=auth_dep) app.include_router(dlq.router, dependencies=auth_dep) + app.include_router(stats.router, dependencies=auth_dep) - # Every request becomes a span when logfire is configured; no-op otherwise. - # /health is the docker healthcheck endpoint — polled every few seconds - # by Compose, would otherwise drown the trace stream in idle GETs. - logfire.instrument_fastapi(app, excluded_urls=r"^.*/health$") return app diff --git a/haiku_rag_slim/haiku/rag/ingester/api/static/index.html b/haiku_rag_slim/haiku/rag/ingester/api/static/index.html new file mode 100644 index 00000000..9fdc3261 --- /dev/null +++ b/haiku_rag_slim/haiku/rag/ingester/api/static/index.html @@ -0,0 +1,553 @@ + + + + + + haiku-ingester · status + + + +
+
+ +

haiku-ingester · status

+ + + never + + +
+ +
+
+
Queued
+
—
+
+
+
Claimed
+
—
+
+
+
Succeeded
+
—
+
+
+
Dead
+
—
+
+
+ +
+
+

Stats

+
+
+
Succeeded · 5m
+
—
+
+
+
Succeeded · 30m
+
—
+
+
+
Succeeded · 1h
+
—
+
+
+
Workers busy
+
—
+
+
+
Oldest queued
+
—
+
+
+
+
+ +
+

Sources

+
+
loading…
+
+
+ +
+

Active jobs

+
+
loading…
+
+
+ +
+

Recent failures

+
+
loading…
+
+
+ +
+

Recently completed

+
+
loading…
+
+
+
+
+ + + + diff --git a/haiku_rag_slim/haiku/rag/ingester/pollers/base.py b/haiku_rag_slim/haiku/rag/ingester/pollers/base.py index 26bbe0a6..eb5a0068 100644 --- a/haiku_rag_slim/haiku/rag/ingester/pollers/base.py +++ b/haiku_rag_slim/haiku/rag/ingester/pollers/base.py @@ -62,6 +62,7 @@ class BasePoller: self._stop = asyncio.Event() self._task: asyncio.Task | None = None self._last_polled_at: datetime | None = None + self._last_skip_reason: str | None = None self._default_max_attempts = default_max_attempts @property @@ -76,6 +77,13 @@ class BasePoller: def is_circuit_open(self) -> bool: return self._breaker.is_open + @property + def last_skip_reason(self) -> str | None: + """Reason the most recent sweep attempt skipped (e.g. "pending_work", + "circuit_open"), or None when the most recent attempt actually polled. + Cleared on the next successful sweep.""" + return self._last_skip_reason + async def run(self) -> None: # pragma: no cover - subclasses override raise NotImplementedError @@ -90,6 +98,7 @@ class BasePoller: breaker is open, the source has pending work already queued, or the sweep failed (and was recorded).""" if self._breaker.is_open: + self._last_skip_reason = "circuit_open" logger.debug( "Skipping discover() — circuit breaker open for %s", self.source_id ) @@ -99,6 +108,7 @@ class BasePoller: # The unique index would dedupe a re-sweep into a saturated # queue anyway; skipping saves the listing round-trip # (PROPFIND / S3 LIST / FS walk) and keeps Logfire readable. + self._last_skip_reason = "pending_work" span.set_attribute("skipped", True) span.set_attribute("skip_reason", "pending_work") logger.debug( @@ -118,6 +128,7 @@ class BasePoller: await self._handle_event(event) self._breaker.record_success() self._last_polled_at = datetime.now(UTC) + self._last_skip_reason = None span.set_attribute("upsert", counts[SourceEventKind.UPSERT]) span.set_attribute("delete", counts[SourceEventKind.DELETE]) span.set_attribute("unchanged", counts[SourceEventKind.UNCHANGED]) diff --git a/haiku_rag_slim/haiku/rag/ingester/pollers/fs.py b/haiku_rag_slim/haiku/rag/ingester/pollers/fs.py index 7fa4b2bd..22d8634d 100644 --- a/haiku_rag_slim/haiku/rag/ingester/pollers/fs.py +++ b/haiku_rag_slim/haiku/rag/ingester/pollers/fs.py @@ -3,9 +3,10 @@ import logging from pathlib import Path from typing import TYPE_CHECKING +import logfire from watchfiles import Change, awatch -from haiku.rag.ingester.pollers.base import BasePoller +from haiku.rag.ingester.pollers.base import BasePoller, _enqueue_extra from haiku.rag.ingester.queue.models import JobOp from haiku.rag.ingester.sources.filter import FileFilter @@ -98,29 +99,42 @@ class FSPoller(BasePoller): async def _handle_watch_change(self, change: Change, path: Path) -> None: uri = path.as_uri() - if change is Change.deleted: - if not self._fs_config.delete_orphans: + # Wrap in a span so the worker's `ingester.job` (and everything it + # nests) hangs off a watch-event parent. Without this the watchfiles + # callback runs with no active context, the `_otel` carrier is empty, + # and the job span surfaces at the trace root — disconnected from + # the FS event that caused it. + with logfire.span( + "ingester.poller.watch_event", + source_id=self.source_id, + change=change.name, + uri=uri, + ): + if change is Change.deleted: + if not self._fs_config.delete_orphans: + return + await self._jobs.enqueue( + self.source_id, + uri, + op=JobOp.DELETE, + max_attempts=self._max_attempts(), + extra=_enqueue_extra(self._fs_config), + ) return - await self._jobs.enqueue( - self.source_id, - uri, - op=JobOp.DELETE, - max_attempts=self._max_attempts(), - ) - return - if change in (Change.added, Change.modified): - revision = str(path.stat().st_mtime_ns) if path.exists() else None - await self._jobs.enqueue( - self.source_id, - uri, - op=JobOp.UPSERT, - revision=revision, - max_attempts=self._max_attempts(), - ) - await self._sync.upsert( - self.source_id, uri, revision=None, content_hash=None - ) + if change in (Change.added, Change.modified): + revision = str(path.stat().st_mtime_ns) if path.exists() else None + await self._jobs.enqueue( + self.source_id, + uri, + op=JobOp.UPSERT, + revision=revision, + max_attempts=self._max_attempts(), + extra=_enqueue_extra(self._fs_config), + ) + await self._sync.upsert( + self.source_id, uri, revision=None, content_hash=None + ) def _max_attempts(self) -> int: cfg = self._fs_config diff --git a/haiku_rag_slim/haiku/rag/ingester/queue/repository.py b/haiku_rag_slim/haiku/rag/ingester/queue/repository.py index f275848f..ad9a7a6d 100644 --- a/haiku_rag_slim/haiku/rag/ingester/queue/repository.py +++ b/haiku_rag_slim/haiku/rag/ingester/queue/repository.py @@ -266,6 +266,49 @@ class JobRepo: rows = await cursor.fetchall() return {row["status"]: row["n"] for row in rows} + async def count_succeeded_since(self, seconds: int) -> int: + """How many jobs reached `succeeded` in the last `seconds` seconds. + Drives the dashboard's rolling-throughput chips.""" + threshold = (datetime.now(UTC) - timedelta(seconds=seconds)).isoformat() + async with self._lock: + async with self._conn.execute( + "SELECT COUNT(*) AS n FROM jobs WHERE status='succeeded' AND completed_at >= ?", + (threshold,), + ) as cursor: + row = await cursor.fetchone() + return int(row["n"]) if row else 0 + + async def oldest_queued_age_seconds(self) -> float | None: + """Age (in seconds) of the oldest job sitting in `queued` whose + scheduled_at is in the past. Returns None when nothing is waiting. + Tells operators whether work is backing up.""" + now = datetime.now(UTC) + async with self._lock: + async with self._conn.execute( + "SELECT MIN(scheduled_at) AS oldest FROM jobs " + "WHERE status='queued' AND scheduled_at <= ?", + (now.isoformat(),), + ) as cursor: + row = await cursor.fetchone() + if not row or row["oldest"] is None: + return None + return (now - datetime.fromisoformat(row["oldest"])).total_seconds() + + async def counts_by_source(self, *statuses: str) -> dict[str, int]: + """source_id → count of jobs in any of the given statuses. Drives the + dashboard's per-source DLQ and backlog summaries.""" + if not statuses: + return {} + placeholders = ",".join("?" * len(statuses)) + async with self._lock: + async with self._conn.execute( + f"SELECT source_id, COUNT(*) AS n FROM jobs " + f"WHERE status IN ({placeholders}) GROUP BY source_id", + statuses, + ) as cursor: + rows = await cursor.fetchall() + return {row["source_id"]: row["n"] for row in rows} + async def release_if_claimed(self, job_id: str) -> bool: """Reset a still-claimed job back to queued, immediately reclaimable. Idempotent — a no-op if the job already transitioned to diff --git a/haiku_rag_slim/pyproject.toml b/haiku_rag_slim/pyproject.toml index 5ebd0dd4..d50c4661 100644 --- a/haiku_rag_slim/pyproject.toml +++ b/haiku_rag_slim/pyproject.toml @@ -59,7 +59,6 @@ ingester = [ "fastapi>=0.125", "uvicorn[standard]>=0.32", "aiosqlite>=0.20", - "logfire[fastapi]>=4.30", "haiku.rag-slim[s3]", ] # TUI (chat and inspect commands) diff --git a/tests/ingester/test_api.py b/tests/ingester/test_api.py index 5d61b770..def21535 100644 --- a/tests/ingester/test_api.py +++ b/tests/ingester/test_api.py @@ -338,3 +338,79 @@ async def test_source_refresh_503_when_pollers_absent(state): async with _client(state) as client: resp = await client.post("/sources/anything/refresh") assert resp.status_code == 503 + + +# --- /stats --- + + +@pytest.mark.asyncio +async def test_stats_returns_shape_on_empty_queue(state): + async with _client(state) as client: + resp = await client.get("/stats") + assert resp.status_code == 200 + body = resp.json() + assert body["throughput"] == { + "succeeded_5m": 0, + "succeeded_30m": 0, + "succeeded_1h": 0, + } + assert body["workers"] == {"busy": 0, "total": 0} + assert body["oldest_queued_age_s"] is None + assert body["dlq_by_source"] == {} + assert body["queue_depth_by_source"] == {} + + +@pytest.mark.asyncio +async def test_stats_aggregates_real_queue(state, jobs): + j1 = await jobs.enqueue("s1", "u1", JobOp.UPSERT) + j2 = await jobs.enqueue("s1", "u2", JobOp.UPSERT) + j3 = await jobs.enqueue("s2", "u3", JobOp.UPSERT) + assert j1 and j2 and j3 + + claimed = await jobs.claim_next("w") + assert claimed is not None + await jobs.mark_succeeded(claimed.id) + dead = await jobs.claim_next("w") + assert dead is not None + await jobs.mark_dead(dead.id, "boom") + + async with _client(state) as client: + resp = await client.get("/stats") + body = resp.json() + # One succeeded in the last 5m, 30m, 1h (we just marked it). + assert body["throughput"]["succeeded_5m"] == 1 + assert body["throughput"]["succeeded_30m"] == 1 + assert body["throughput"]["succeeded_1h"] == 1 + # Last enqueued (s2/u3) remains queued. + assert body["queue_depth_by_source"] == {"s2": 1} + # The dead job was claim_next-ed from s1. + assert body["dlq_by_source"] == {"s1": 1} + + +@pytest.mark.asyncio +async def test_stats_requires_auth(state): + async with _client(state, auth_token="secret") as client: + resp = await client.get("/stats") + assert resp.status_code == 401 + + +# --- dashboard --- + + +@pytest.mark.asyncio +async def test_dashboard_served_unauthenticated(state): + """The dashboard is markup-only. The JS it serves attaches the bearer + token to its own JSON fetches, so the page itself must load without one + even when auth is enabled.""" + async with _client(state, auth_token="secret") as client: + resp = await client.get("/") + assert resp.status_code == 200 + assert "text/html" in resp.headers["content-type"] + body = resp.text + assert "haiku-ingester · status" in body + # The JS calls the JSON endpoints; sanity-check it's wired up. + assert "/stats" in body + assert "/sources" in body + assert "/jobs?status=claimed" in body + # Op badge helper is present so DELETE rows render distinctly. + assert "opBadge" in body diff --git a/tests/ingester/test_pollers.py b/tests/ingester/test_pollers.py index 11f78569..2b527dfe 100644 --- a/tests/ingester/test_pollers.py +++ b/tests/ingester/test_pollers.py @@ -180,6 +180,28 @@ async def test_repeated_sweep_skipped_when_queue_has_pending(fs_config, jobs, sy assert source.discover_calls == 1 +@pytest.mark.asyncio +async def test_skipped_sweep_records_pending_work_reason(fs_config, jobs, sync): + """last_skip_reason surfaces 'pending_work' while the queue is saturated + and clears once the next sweep actually polls.""" + event = _event("file:///a.md", revision="r1") + source = _StubSource("src", [[event], [event], []]) + poller = _periodic(source, fs_config, jobs, sync) + + await poller._sweep_once() # first sweep enqueues, succeeds + assert poller.last_skip_reason is None + + await poller._sweep_once() # backpressure skips + assert poller.last_skip_reason == "pending_work" + + # Drain the queue, sweep again, reason clears. + claimed = await jobs.claim_next("worker") + assert claimed is not None + await jobs.mark_succeeded(claimed.id) + await poller._sweep_once() + assert poller.last_skip_reason is None + + @pytest.mark.asyncio async def test_sweep_resumes_after_queue_drains(fs_config, jobs, sync): """Once the queue clears (success, dead, or cancel), sweeps resume.""" diff --git a/tests/ingester/test_queue.py b/tests/ingester/test_queue.py index a4daec74..2e39d742 100644 --- a/tests/ingester/test_queue.py +++ b/tests/ingester/test_queue.py @@ -460,6 +460,83 @@ async def test_counts_by_status(jobs): assert j3.id # silence unused +# --- stats --- + + +@pytest.mark.asyncio +async def test_count_succeeded_since_only_includes_recent(jobs, conn): + await jobs.enqueue("s", "old", JobOp.UPSERT) + await jobs.enqueue("s", "new", JobOp.UPSERT) + + old_claim = await jobs.claim_next("w") + assert old_claim is not None + await jobs.mark_succeeded(old_claim.id) + long_ago = (datetime.now(UTC) - timedelta(hours=2)).isoformat() + await conn.execute( + "UPDATE jobs SET completed_at = ? WHERE id = ?", (long_ago, old_claim.id) + ) + await conn.commit() + + new_claim = await jobs.claim_next("w") + assert new_claim is not None + await jobs.mark_succeeded(new_claim.id) + + assert await jobs.count_succeeded_since(60) == 1 + assert await jobs.count_succeeded_since(86400) == 2 + + +@pytest.mark.asyncio +async def test_oldest_queued_age_seconds_none_when_empty(jobs): + assert await jobs.oldest_queued_age_seconds() is None + + +@pytest.mark.asyncio +async def test_oldest_queued_age_seconds_returns_oldest(jobs, conn): + old = await jobs.enqueue("s", "old", JobOp.UPSERT) + await jobs.enqueue("s", "new", JobOp.UPSERT) + backdate = (datetime.now(UTC) - timedelta(seconds=120)).isoformat() + assert old is not None + await conn.execute( + "UPDATE jobs SET scheduled_at = ? WHERE id = ?", (backdate, old.id) + ) + await conn.commit() + + age = await jobs.oldest_queued_age_seconds() + assert age is not None + assert 119 <= age <= 125 + + +@pytest.mark.asyncio +async def test_oldest_queued_age_seconds_ignores_future_scheduled(jobs, conn): + """A job whose scheduled_at is in the future (e.g. after a backoff + reschedule) isn't ready to run, so it shouldn't count toward backlog age.""" + j = await jobs.enqueue("s", "u", JobOp.UPSERT) + assert j is not None + future = (datetime.now(UTC) + timedelta(seconds=600)).isoformat() + await conn.execute("UPDATE jobs SET scheduled_at = ? WHERE id = ?", (future, j.id)) + await conn.commit() + assert await jobs.oldest_queued_age_seconds() is None + + +@pytest.mark.asyncio +async def test_counts_by_source_groups_correctly(jobs): + await jobs.enqueue("s1", "u1", JobOp.UPSERT) + await jobs.enqueue("s1", "u2", JobOp.UPSERT) + j3 = await jobs.enqueue("s2", "u3", JobOp.UPSERT) + assert j3 is not None + await jobs.mark_dead(j3.id, "boom") + + assert await jobs.counts_by_source("queued") == {"s1": 2} + assert await jobs.counts_by_source("dead") == {"s2": 1} + assert await jobs.counts_by_source("queued", "claimed") == {"s1": 2} + + +@pytest.mark.asyncio +async def test_counts_by_source_no_statuses_returns_empty(jobs): + await jobs.enqueue("s", "u", JobOp.UPSERT) + assert await jobs.counts_by_source() == {} + + # --- sync state --- diff --git a/uv.lock b/uv.lock index b002a066..753db671 100644 --- a/uv.lock +++ b/uv.lock @@ -243,15 +243,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/38/0e/27be9fdef66e72d64c0cdc3cc2823101b80585f8119b5c112c2e8f5f7dab/anyio-4.12.1-py3-none-any.whl", hash = "sha256:d405828884fc140aa80a3c667b8beed277f1dfedec42ba031bd6ac3db606ab6c", size = 113592, upload-time = "2026-01-06T11:45:19.497Z" }, ] -[[package]] -name = "asgiref" -version = "3.11.1" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/63/40/f03da1264ae8f7cfdbf9146542e5e7e8100a4c66ab48e791df9a03d3f6c0/asgiref-3.11.1.tar.gz", hash = "sha256:5f184dc43b7e763efe848065441eac62229c9f7b0475f41f80e207a114eda4ce", size = 38550, upload-time = "2026-02-03T13:30:14.33Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/5c/0a/a72d10ed65068e115044937873362e6e32fab1b7dce0046aeb224682c989/asgiref-3.11.1-py3-none-any.whl", hash = "sha256:e8667a091e69529631969fd45dc268fa79b99c92c5fcdda727757e52146ec133", size = 24345, upload-time = "2026-02-03T13:30:13.039Z" }, -] - [[package]] name = "attrs" version = "26.1.0" @@ -1606,7 +1597,6 @@ groq = [ ingester = [ { name = "aiosqlite" }, { name = "fastapi" }, - { name = "logfire", extra = ["fastapi"] }, { name = "obstore" }, { name = "uvicorn", extra = ["standard"] }, ] @@ -1653,7 +1643,6 @@ requires-dist = [ { name = "jinja2", specifier = ">=3.1.0" }, { name = "jsonpatch", specifier = ">=1.33" }, { name = "lancedb", specifier = "==0.30.2" }, - { name = "logfire", extras = ["fastapi"], marker = "extra == 'ingester'", specifier = ">=4.30" }, { name = "mxbai-rerank", marker = "extra == 'mxbai'", specifier = ">=0.1.6" }, { name = "obstore", marker = "extra == 's3'", specifier = ">=0.9,<0.10" }, { name = "opencv-python-headless", marker = "extra == 'docling'", specifier = ">=4.13.0.92" }, @@ -2262,9 +2251,6 @@ wheels = [ ] [package.optional-dependencies] -fastapi = [ - { name = "opentelemetry-instrumentation-fastapi" }, -] httpx = [ { name = "opentelemetry-instrumentation-httpx" }, ] @@ -3119,38 +3105,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/77/d2/6788e83c5c86a2690101681aeef27eeb2a6bf22df52d3f263a22cee20915/opentelemetry_instrumentation-0.60b1-py3-none-any.whl", hash = "sha256:04480db952b48fb1ed0073f822f0ee26012b7be7c3eac1a3793122737c78632d", size = 33096, upload-time = "2025-12-11T13:35:33.067Z" }, ] -[[package]] -name = "opentelemetry-instrumentation-asgi" -version = "0.60b1" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "asgiref" }, - { name = "opentelemetry-api" }, - { name = "opentelemetry-instrumentation" }, - { name = "opentelemetry-semantic-conventions" }, - { name = "opentelemetry-util-http" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/77/db/851fa88db7441da82d50bd80f2de5ee55213782e25dc858e04d0c9961d60/opentelemetry_instrumentation_asgi-0.60b1.tar.gz", hash = "sha256:16bfbe595cd24cda309a957456d0fc2523f41bc7b076d1f2d7e98a1ad9876d6f", size = 26107, upload-time = "2025-12-11T13:36:47.015Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/76/76/1fb94367cef64420d2171157a6b9509582873bd09a6afe08a78a8d1f59d9/opentelemetry_instrumentation_asgi-0.60b1-py3-none-any.whl", hash = "sha256:d48def2dbed10294c99cfcf41ebbd0c414d390a11773a41f472d20000fcddc25", size = 16933, upload-time = "2025-12-11T13:35:40.462Z" }, -] - -[[package]] -name = "opentelemetry-instrumentation-fastapi" -version = "0.60b1" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "opentelemetry-api" }, - { name = "opentelemetry-instrumentation" }, - { name = "opentelemetry-instrumentation-asgi" }, - { name = "opentelemetry-semantic-conventions" }, - { name = "opentelemetry-util-http" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/9c/e7/e7e5e50218cf488377209d85666b182fa2d4928bf52389411ceeee1b2b60/opentelemetry_instrumentation_fastapi-0.60b1.tar.gz", hash = "sha256:de608955f7ff8eecf35d056578346a5365015fd7d8623df9b1f08d1c74769c01", size = 24958, upload-time = "2025-12-11T13:36:59.35Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/7d/cc/6e808328ba54662e50babdcab21138eae4250bc0fddf67d55526a615a2ca/opentelemetry_instrumentation_fastapi-0.60b1-py3-none-any.whl", hash = "sha256:af94b7a239ad1085fc3a820ecf069f67f579d7faf4c085aaa7bd9b64eafc8eaf", size = 13478, upload-time = "2025-12-11T13:36:00.811Z" }, -] - [[package]] name = "opentelemetry-instrumentation-httpx" version = "0.60b1"