diff --git a/CHANGELOG.md b/CHANGELOG.md index ae707cfa..b9aa8f36 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/ingester.md b/docs/ingester.md index 1aa187ae..aded0fd8 100644 --- a/docs/ingester.md +++ b/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. diff --git a/haiku_rag_slim/haiku/rag/config/models.py b/haiku_rag_slim/haiku/rag/config/models.py index b4ed9010..8c00a0bf 100644 --- a/haiku_rag_slim/haiku/rag/config/models.py +++ b/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): diff --git a/haiku_rag_slim/haiku/rag/ingester/app.py b/haiku_rag_slim/haiku/rag/ingester/app.py index 2e0e5d3d..c6767b14 100644 --- a/haiku_rag_slim/haiku/rag/ingester/app.py +++ b/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. diff --git a/haiku_rag_slim/haiku/rag/ingester/queue/repository.py b/haiku_rag_slim/haiku/rag/ingester/queue/repository.py index 3cbdac4f..85852b5f 100644 --- a/haiku_rag_slim/haiku/rag/ingester/queue/repository.py +++ b/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 diff --git a/haiku_rag_slim/haiku/rag/ingester/workers/pool.py b/haiku_rag_slim/haiku/rag/ingester/workers/pool.py index 25881ec9..08589cbc 100644 --- a/haiku_rag_slim/haiku/rag/ingester/workers/pool.py +++ b/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: diff --git a/tests/ingester/test_queue.py b/tests/ingester/test_queue.py index db376406..524806ab 100644 --- a/tests/ingester/test_queue.py +++ b/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 --- diff --git a/tests/ingester/test_workers.py b/tests/ingester/test_workers.py index bedc453e..60596d19 100644 --- a/tests/ingester/test_workers.py +++ b/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