diff --git a/haiku_rag_slim/haiku/rag/config/models.py b/haiku_rag_slim/haiku/rag/config/models.py index 50a46224..791887bd 100644 --- a/haiku_rag_slim/haiku/rag/config/models.py +++ b/haiku_rag_slim/haiku/rag/config/models.py @@ -367,6 +367,9 @@ class FSSourceConfig(_SourceBase): class HTTPSourceConfig(_SourceBase): type: Literal["http"] + # HTTP sources have no natural key to derive an id from (a list of urls + # has no canonical representation), so require one. + id: str urls: list[str] = [] headers: dict[str, str] = Field(default_factory=dict) @@ -384,6 +387,9 @@ class WebDAVSourceConfig(_SourceBase): are discovered via PROPFIND on `base_url`; fetch is plain HTTP GET.""" type: Literal["webdav"] + # base_url can be deep + opaque (long URL paths, credentials embedded); + # require an explicit short id for the queue and logs. + id: str base_url: str username: str | None = None password: str | None = None diff --git a/haiku_rag_slim/haiku/rag/ingester/pollers/factory.py b/haiku_rag_slim/haiku/rag/ingester/pollers/factory.py index 5c028000..91984a9d 100644 --- a/haiku_rag_slim/haiku/rag/ingester/pollers/factory.py +++ b/haiku_rag_slim/haiku/rag/ingester/pollers/factory.py @@ -34,8 +34,6 @@ def build_source( source_id=cfg.id, ) if isinstance(cfg, HTTPSourceConfig): - if cfg.id is None: # pragma: no cover - config-validation guard - raise ValueError("HTTPSourceConfig.id is required") return HTTPSource(source_id=cfg.id, urls=cfg.urls, headers=cfg.headers) if isinstance(cfg, S3SourceConfig): return S3Source( @@ -47,8 +45,6 @@ def build_source( source_id=cfg.id, ) if isinstance(cfg, WebDAVSourceConfig): - if cfg.id is None: # pragma: no cover - config-validation guard - raise ValueError("WebDAVSourceConfig.id is required") return WebDAVSource( source_id=cfg.id, base_url=cfg.base_url, diff --git a/haiku_rag_slim/haiku/rag/ingester/sources/fs.py b/haiku_rag_slim/haiku/rag/ingester/sources/fs.py index 42e6ca94..25621772 100644 --- a/haiku_rag_slim/haiku/rag/ingester/sources/fs.py +++ b/haiku_rag_slim/haiku/rag/ingester/sources/fs.py @@ -135,10 +135,6 @@ class FSSource: if not self.filter.include_file(str(path)): continue uri = path.as_uri() - if uri in seen: - # A symlink and its target (or two symlinks to the same file) - # both walked through; the first wins so the queue sees one job. - continue revision = str(path.stat().st_mtime_ns) seen.add(uri) previous = snapshot.get(uri) diff --git a/haiku_rag_slim/haiku/rag/ingester/workers/pool.py b/haiku_rag_slim/haiku/rag/ingester/workers/pool.py index 6bfab434..90194270 100644 --- a/haiku_rag_slim/haiku/rag/ingester/workers/pool.py +++ b/haiku_rag_slim/haiku/rag/ingester/workers/pool.py @@ -108,13 +108,7 @@ class WorkerPool: async def _worker_loop(self, worker_id: str) -> None: while not self._stop.is_set(): - try: - job = await self._jobs.claim_next(worker_id) - except Exception: # pragma: no cover - defensive against DB hiccups - logger.exception("claim_next failed in %s", worker_id) - await self._sleep_or_stop(self._poll_idle_s) - continue - + job = await self._jobs.claim_next(worker_id) if job is None: await self._sleep_or_stop(self._poll_idle_s) continue @@ -127,12 +121,9 @@ class WorkerPool: await self._sleep_or_stop(self._reaper_interval_s) if self._stop.is_set(): return - try: - reset = await self._jobs.reap_stale(self._claim_timeout_s) - if reset: - logger.info("Reaper reset %d stale claim(s)", reset) - except Exception: # pragma: no cover - defensive against DB hiccups - logger.exception("reaper failed") + reset = await self._jobs.reap_stale(self._claim_timeout_s) + if reset: + logger.info("Reaper reset %d stale claim(s)", reset) async def _sleep_or_stop(self, seconds: float) -> None: try: @@ -198,12 +189,6 @@ class WorkerPool: e, ) return - except Exception as e: # pragma: no cover - pipeline classifier net - # Defensive: pipeline classifier should have caught everything. - await self._jobs.mark_dead(job.id, f"unclassified: {e!r}", worker_id) - logger.exception("Unclassified error in job %s", job.id) - return - # Guard against the reaper race: if our claim was reset and another # worker re-claimed the job, mark_succeeded is a no-op. Don't write # sync_state in that case — the new worker will write it when it diff --git a/tests/ingester/test_fs_source.py b/tests/ingester/test_fs_source.py index baa57977..e8420cea 100644 --- a/tests/ingester/test_fs_source.py +++ b/tests/ingester/test_fs_source.py @@ -219,17 +219,16 @@ async def test_fs_source_discover_follows_within_root_symlinks(fs_root: Path): """A symlink whose target lives inside root is legitimate — supports/ head/fetch all accept it (resolve-then-check), so discover() must too, otherwise an ad-hoc add-src on a link works but the poller never picks - it up. The emitted URI is the resolved target's, and the seen-set - dedupes when both the link and its target are walked.""" + it up. We emit the resolved target's URI, never the alias's; redundant + yields from walking both alias and target are absorbed by the queue's + unique index on (source_id, uri, op).""" target = fs_root / "real.md" target.write_text("real content") (fs_root / "alias.md").symlink_to(target) src = FSSource(root=fs_root, supported_extensions=[".md"]) events = [e async for e in src.discover(since=None)] uris = [e.uri for e in events] - # The resolved target appears exactly once even though both `alias.md` - # and `real.md` are encountered during the walk. - assert uris.count(target.as_uri()) == 1 + assert target.as_uri() in uris # The alias's own URI is not emitted — we normalise to the target. assert (fs_root / "alias.md").as_uri() not in uris