Merge pull request #414 from ggozad/feat/prune-ingestor-pool

Add retention window to ingester queue

.github/workflows/test.yml

@@ -48,6 +48,8 @@ jobs:
   test:
     needs: [lint, lint-frontend]
     runs-on: ubuntu-latest
+    env:
+      HF_TOKEN: ${{ secrets.HF_TOKEN }}
     steps:
       - uses: actions/checkout@v4
       - uses: astral-sh/setup-uv@v4

CHANGELOG.md

@@ -1,6 +1,10 @@
 # Changelog
 ## [Unreleased]
 
+### Added
+
+- `ingester.queue.retention_days` (default 30): the reaper deletes succeeded/dead jobs whose `completed_at` is older than the window. `null` disables pruning.
+
 ## [0.52.0] - 2026-06-01
 
 ### Added

docs/ingester.md

@@ -358,6 +358,20 @@ haiku-ingester queue init             # create the DB and schema
 haiku-ingester queue migrate          # apply pending schema changes
 ```
 
+Terminal job rows (`succeeded` and `dead`) are kept for history and pruned by
+the reaper once they age past `retention_days`:
+
+```yaml
+ingester:
+  queue:
+    path: /var/lib/haiku-rag/ingester.db
+    retention_days: 30                # null disables pruning
+```
+
+The reaper deletes terminal rows whose `completed_at` is older than the window
+on its `reaper_interval_s` cadence. Set `retention_days: null` to keep all
+terminal rows.
+
 ### Logs
 
 The service logs via Python `logging` to stderr through a Rich handler.

haiku_rag_slim/haiku/rag/config/models.py

@@ -266,6 +266,12 @@ class QueueConfig(BaseModel):
         default_factory=lambda: get_default_data_dir() / "ingester.db",
         description="Location of the ingester's SQLite queue file.",
     )
+    retention_days: int | None = Field(
+        default=30,
+        description="Delete succeeded/dead jobs whose completed_at is older "
+        "than this many days. The reaper enforces it on reaper_interval_s. "
+        "None disables pruning (keep all terminal rows).",
+    )
 
 
 class RetryPolicyConfig(BaseModel):

haiku_rag_slim/haiku/rag/ingester/app.py

@@ -95,6 +95,11 @@ class IngesterApp:
                     poll_idle_interval_s=ingester_cfg.workers.poll_idle_interval_s,
                     claim_timeout_s=ingester_cfg.workers.claim_timeout_s,
                     reaper_interval_s=ingester_cfg.workers.reaper_interval_s,
+                    retention_s=(
+                        ingester_cfg.queue.retention_days * 86400
+                        if ingester_cfg.queue.retention_days is not None
+                        else None
+                    ),
                     # Same Source instances the pollers discover with —
                     # workers resolve URIs through them so authenticated
                     # HTTP / WebDAV / S3 fetches reuse credentials.

haiku_rag_slim/haiku/rag/ingester/queue/repository.py

@@ -404,6 +404,22 @@ class JobRepo:
             await self._conn.commit()
         return rowcount
 
+    async def prune_terminal(self, max_age_seconds: int) -> int:
+        """Delete terminal jobs (succeeded/dead) whose completed_at is older
+        than max_age_seconds. General housekeeping so the table doesn't grow
+        without bound. Returns the number of rows removed."""
+        threshold = (datetime.now(UTC) - timedelta(seconds=max_age_seconds)).isoformat()
+        async with self._lock:
+            cursor = await self._conn.execute(
+                "DELETE FROM jobs WHERE status IN ('succeeded','dead') "
+                "AND completed_at < ?",
+                (threshold,),
+            )
+            rowcount = cursor.rowcount or 0
+            await cursor.close()
+            await self._conn.commit()
+        return rowcount
+
     async def reap_stale(self, claim_timeout_seconds: int) -> int:
         """Reset claimed jobs whose claimed_at is older than the timeout
         back to `queued`. Decrements `attempts` to undo the increment from

haiku_rag_slim/haiku/rag/ingester/workers/pool.py

@@ -39,6 +39,7 @@ class WorkerPool:
         poll_idle_interval_s: float = 1.0,
         claim_timeout_s: int = 1800,
         reaper_interval_s: int = 60,
+        retention_s: int | None = None,
         sources: "list[Source] | None" = None,
     ):
         self._client = client
@@ -49,6 +50,7 @@ class WorkerPool:
         self._poll_idle_s = poll_idle_interval_s
         self._claim_timeout_s = claim_timeout_s
         self._reaper_interval_s = reaper_interval_s
+        self._retention_s = retention_s
         self._sources: list[Source] = list(sources) if sources else []
         self._stop = asyncio.Event()
         self._workers: list[asyncio.Task] = []
@@ -153,6 +155,10 @@ class WorkerPool:
             reset = await self._jobs.reap_stale(self._claim_timeout_s)
             if reset:
                 logger.info("Reaper reset %d stale claim(s)", reset)
+            if self._retention_s is not None:
+                pruned = await self._jobs.prune_terminal(self._retention_s)
+                if pruned:
+                    logger.info("Reaper pruned %d terminal job(s)", pruned)
 
     async def _sleep_or_stop(self, seconds: float) -> None:
         try:

tests/ingester/test_queue.py

@@ -567,6 +567,62 @@ async def test_prune_dead_scoped_to_matching_uri(jobs):
     assert await jobs.get_job(j2.id) is not None
 
 
+# --- prune_terminal ---
+
+
+@pytest.mark.asyncio
+async def test_prune_terminal_deletes_old_terminal_rows(jobs, conn):
+    """Succeeded and dead rows whose completed_at is older than the window
+    are deleted so the table doesn't grow without bound."""
+    ok = await jobs.enqueue("s", "u1", JobOp.UPSERT)
+    await jobs.claim_next("w")
+    await jobs.mark_succeeded(ok.id, "w")
+
+    bad = await jobs.enqueue("s", "u2", JobOp.UPSERT)
+    await jobs.claim_next("w")
+    await jobs.mark_dead(bad.id, "boom", "w")
+
+    long_ago = (datetime.now(UTC) - timedelta(hours=1)).isoformat()
+    await conn.execute(
+        "UPDATE jobs SET completed_at = ? WHERE id IN (?, ?)",
+        (long_ago, ok.id, bad.id),
+    )
+    await conn.commit()
+
+    pruned = await jobs.prune_terminal(max_age_seconds=60)
+    assert pruned == 2
+    assert await jobs.get_job(ok.id) is None
+    assert await jobs.get_job(bad.id) is None
+
+
+@pytest.mark.asyncio
+async def test_prune_terminal_keeps_recent_terminal_rows(jobs):
+    """A freshly-completed succeeded row survives — only rows past the
+    retention window are removed."""
+    job = await jobs.enqueue("s", "u", JobOp.UPSERT)
+    await jobs.claim_next("w")
+    await jobs.mark_succeeded(job.id, "w")
+
+    pruned = await jobs.prune_terminal(max_age_seconds=3600)
+    assert pruned == 0
+    refreshed = await jobs.get_job(job.id)
+    assert refreshed is not None and refreshed.status is JobStatus.SUCCEEDED
+
+
+@pytest.mark.asyncio
+async def test_prune_terminal_ignores_non_terminal_rows(jobs):
+    """Queued/claimed rows have no completed_at and are never pruned,
+    regardless of the window."""
+    queued = await jobs.enqueue("s", "u1", JobOp.UPSERT)
+    claimed = await jobs.enqueue("s", "u2", JobOp.UPSERT)
+    await jobs.claim_next("w")
+
+    pruned = await jobs.prune_terminal(max_age_seconds=0)
+    assert pruned == 0
+    assert (await jobs.get_job(queued.id)) is not None
+    assert (await jobs.get_job(claimed.id)) is not None
+
+
 # --- release_if_claimed ---
 
 

tests/ingester/test_workers.py

@@ -53,6 +53,7 @@ def _pool(client, jobs, sync, **kwargs) -> WorkerPool:
         poll_idle_interval_s=kwargs.pop("poll_idle_interval_s", 0.05),
         reaper_interval_s=kwargs.pop("reaper_interval_s", 60),
         claim_timeout_s=kwargs.pop("claim_timeout_s", 60),
+        retention_s=kwargs.pop("retention_s", None),
         retry_policy=kwargs.pop("retry_policy", RetryPolicy()),
         sources=kwargs.pop("sources", None),
     )
@@ -743,3 +744,71 @@ async def test_reaper_resets_stale_claims(client, jobs, sync, conn):
     refreshed = await jobs.get_job(job.id)
     assert refreshed is not None
     assert refreshed.status is JobStatus.QUEUED
+
+
+@pytest.mark.asyncio
+async def test_reaper_prunes_old_terminal_jobs(client, jobs, sync, conn):
+    from datetime import UTC, datetime, timedelta
+
+    job = await jobs.enqueue("src", "u", JobOp.UPSERT)
+    await jobs.claim_next("w")
+    await jobs.mark_succeeded(job.id, "w")
+
+    # Backdate completed_at so prune_terminal picks it up.
+    long_ago = (datetime.now(UTC) - timedelta(hours=1)).isoformat()
+    await conn.execute(
+        "UPDATE jobs SET completed_at = ? WHERE id = ?", (long_ago, job.id)
+    )
+    await conn.commit()
+
+    pool = _pool(
+        client,
+        jobs,
+        sync,
+        worker_count=0,
+        reaper_interval_s=0.05,
+        retention_s=1,
+    )
+    await pool.start()
+    try:
+        for _ in range(30):
+            if await jobs.get_job(job.id) is None:
+                break
+            await asyncio.sleep(0.05)
+    finally:
+        await pool.stop()
+
+    assert await jobs.get_job(job.id) is None
+
+
+@pytest.mark.asyncio
+async def test_reaper_skips_prune_when_retention_none(client, jobs, sync, conn):
+    from datetime import UTC, datetime, timedelta
+
+    job = await jobs.enqueue("src", "u", JobOp.UPSERT)
+    await jobs.claim_next("w")
+    await jobs.mark_succeeded(job.id, "w")
+
+    long_ago = (datetime.now(UTC) - timedelta(hours=1)).isoformat()
+    await conn.execute(
+        "UPDATE jobs SET completed_at = ? WHERE id = ?", (long_ago, job.id)
+    )
+    await conn.commit()
+
+    pool = _pool(
+        client,
+        jobs,
+        sync,
+        worker_count=0,
+        reaper_interval_s=0.05,
+        retention_s=None,
+    )
+    await pool.start()
+    try:
+        await asyncio.sleep(0.3)
+    finally:
+        await pool.stop()
+
+    refreshed = await jobs.get_job(job.id)
+    assert refreshed is not None
+    assert refreshed.status is JobStatus.SUCCEEDED