From 206d29b74ab16afc802d16a415d2d02d813bb1fe Mon Sep 17 00:00:00 2001 From: Yiorgis Gozadinos Date: Thu, 27 Aug 2026 13:39:22 +0300 Subject: [PATCH] Let the configuration place the ingester's database The ingester manufactured `data_dir / haiku.rag.lancedb` whenever `--db` was absent and passed it to HaikuRAG. A path is now an explicit override that clears the configured URI, so a `lancedb.uri` deployment wrote to local disk while the control plane reported on the remote. `IngesterApp` resolves the databases it works on once, in its constructor, and both the client it opens and the control plane read that scope. `--db` names one directly and nothing stands in for it, so a configured set is refused rather than guessed, and `AmbiguousDatabaseError` joins the errors `cli()` turns into a clean exit. --- docs/configuration/storage.md | 2 + .../haiku/rag/ingester/api/routes/database.py | 11 +++- .../haiku/rag/ingester/api/server.py | 4 +- haiku_rag_slim/haiku/rag/ingester/app.py | 14 +++-- haiku_rag_slim/haiku/rag/ingester/cli.py | 17 ++--- tests/ingester/test_api.py | 30 ++++++++- tests/ingester/test_cli.py | 62 ++++++++++++++++++- tests/ingester/test_run_batch.py | 8 ++- 8 files changed, 122 insertions(+), 26 deletions(-) diff --git a/docs/configuration/storage.md b/docs/configuration/storage.md index 87d63e4c..d3396877 100644 --- a/docs/configuration/storage.md +++ b/docs/configuration/storage.md @@ -198,6 +198,8 @@ The recommended layout for production is "different buckets, same account, separ Each process picks up its own credentials from the AWS default chain (env vars, IAM instance role, AWS profile), so no credentials are hard-coded in the configuration files. +`haiku-ingester` writes the database the configuration places, so a `lancedb.uri` needs no further option. `--db PATH` overrides it. When `lancedb.databases` contains more than one database the ingester has no way to name which it writes, and refuses to start with `AmbiguousDatabaseError`: give each database its own ingester process, each with a configuration naming a single database, or select one with `--db PATH`. A one-entry mapping is unambiguous and is accepted. + ## Multiple Databases Use `lancedb.databases` to name local or remote databases that should be searched diff --git a/haiku_rag_slim/haiku/rag/ingester/api/routes/database.py b/haiku_rag_slim/haiku/rag/ingester/api/routes/database.py index 0750db46..fa2f355c 100644 --- a/haiku_rag_slim/haiku/rag/ingester/api/routes/database.py +++ b/haiku_rag_slim/haiku/rag/ingester/api/routes/database.py @@ -1,5 +1,6 @@ from fastapi import APIRouter, Depends, HTTPException, status +from haiku.rag.client.session import default_db_path from haiku.rag.ingester.api.server import APIState, get_state from haiku.rag.store.info import DatabaseInfo, gather_database_info @@ -14,9 +15,13 @@ async def database(state: APIState = Depends(get_state)) -> DatabaseInfo: Opens a fresh read-only connection per call; not cached and not part of the dashboard's polling loop.""" - if state.db_path is None: + if state.scope is None: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, - detail="database path not configured", + detail="database not configured", ) - return await gather_database_info(state.config, state.db_path) + # The one database the ingester writes: a configured set is refused when the + # client opens, before this app is built. + [ref] = state.scope.databases + one, db_path = ref.connection(state.config) + return await gather_database_info(one, db_path or default_db_path(one)) diff --git a/haiku_rag_slim/haiku/rag/ingester/api/server.py b/haiku_rag_slim/haiku/rag/ingester/api/server.py index bafa7b0e..ece74ca3 100644 --- a/haiku_rag_slim/haiku/rag/ingester/api/server.py +++ b/haiku_rag_slim/haiku/rag/ingester/api/server.py @@ -1,5 +1,4 @@ from dataclasses import dataclass -from pathlib import Path from typing import TYPE_CHECKING from fastapi import Depends, FastAPI, Request @@ -7,6 +6,7 @@ from fastapi import Depends, FastAPI, Request from haiku.rag.ingester.api.auth import require_auth if TYPE_CHECKING: + from haiku.rag.client.scope import DatabaseScope from haiku.rag.config import AppConfig from haiku.rag.ingester.pollers.manager import PollerManager from haiku.rag.ingester.queue.repository import JobRepo, SyncStateRepo @@ -23,7 +23,7 @@ class APIState: sync_repo: "SyncStateRepo" pool: "WorkerPool | None" = None pollers: "PollerManager | None" = None - db_path: Path | None = None + scope: "DatabaseScope | None" = None def get_state(request: Request) -> APIState: diff --git a/haiku_rag_slim/haiku/rag/ingester/app.py b/haiku_rag_slim/haiku/rag/ingester/app.py index 321845c0..900b8880 100644 --- a/haiku_rag_slim/haiku/rag/ingester/app.py +++ b/haiku_rag_slim/haiku/rag/ingester/app.py @@ -72,9 +72,13 @@ class IngesterApp: WorkerPool, and a HaikuRAG client for the worker pool to ingest through. """ - def __init__(self, *, config: AppConfig, db_path: Path): + def __init__(self, *, config: AppConfig, db_path: Path | None = None): + from haiku.rag.client.scope import DatabaseScope + self._config = config - self._db_path = db_path + # `--db` names the database directly, and nothing stands in for it: a + # manufactured default would override a configured `lancedb.uri`. + self._scope = DatabaseScope.resolve(config, database_path=db_path) self._engine: AsyncEngine | None = None self._jobs: JobRepo | None = None self._sync: SyncStateRepo | None = None @@ -107,8 +111,8 @@ class IngesterApp: # The ingester is the sole writer for its LanceDB target; create on # first start so docker-compose / fresh deployments don't require a # manual `haiku-rag init`. - async with HaikuRAG( - self._db_path, config=self._config, create=True + async with HaikuRAG._covering( + self._scope, self._config, create=True ) as client: self._pollers = PollerManager( configs=ingester_cfg.sources, @@ -471,7 +475,7 @@ class IngesterApp: sync_repo=self._sync, pool=self._pool, pollers=self._pollers, - db_path=self._db_path, + scope=self._scope, ) if ingester_cfg.api.auth_token is None: logger.warning("API auth_token is unset — control plane is unauthenticated") diff --git a/haiku_rag_slim/haiku/rag/ingester/cli.py b/haiku_rag_slim/haiku/rag/ingester/cli.py index eb6d62e6..f8f1eee3 100644 --- a/haiku_rag_slim/haiku/rag/ingester/cli.py +++ b/haiku_rag_slim/haiku/rag/ingester/cli.py @@ -33,6 +33,7 @@ from haiku.rag.ingester.batch import BatchManifest # noqa: E402 from haiku.rag.ingester.queue.migrations import open_queue # noqa: E402 from haiku.rag.logging import configure_cli_logging # noqa: E402 from haiku.rag.store.exceptions import ( # noqa: E402 + AmbiguousDatabaseError, MigrationRequiredError, ReadOnlyError, ) @@ -70,7 +71,7 @@ def cli() -> None: """Entry point that translates store-state errors into a clean exit.""" try: _cli() - except (MigrationRequiredError, ReadOnlyError) as e: + except (AmbiguousDatabaseError, MigrationRequiredError, ReadOnlyError) as e: typer.echo(f"Error: {e}", err=True) sys.exit(1) @@ -149,10 +150,6 @@ def queue_migrate( typer.echo(f"Queue at {_queue_target(queue_config)} is up to date") -def _resolve_db_path(config: AppConfig, override: Path | None) -> Path: - return override or (config.storage.data_dir / "haiku.rag.lancedb") - - def _default_manifest_path() -> Path: datestamp = datetime.now(UTC).strftime("%Y%m%d-%H%M%SZ") return Path(f"manifest-{datestamp}.yaml") @@ -216,7 +213,7 @@ def serve( db: Path | None = typer.Option( None, "--db", - help="LanceDB path (overrides config.storage.data_dir).", + help="LanceDB path (overrides the configured database location).", ), host: str | None = typer.Option( None, @@ -252,8 +249,7 @@ def serve( app_config.ingester.api.port = port if root_path is not None: app_config.ingester.api.root_path = root_path - db_path = _resolve_db_path(app_config, db) - app = IngesterApp(config=app_config, db_path=db_path) + app = IngesterApp(config=app_config, db_path=db) asyncio.run(app.serve(api=not no_api)) @@ -262,7 +258,7 @@ def run_batch( db: Path | None = typer.Option( None, "--db", - help="LanceDB path (overrides config.storage.data_dir).", + help="LanceDB path (overrides the configured database location).", ), dry_run: bool = typer.Option( False, @@ -312,8 +308,7 @@ async def _run_batch( ) -> None: from haiku.rag.ingester.app import IngesterApp - db = _resolve_db_path(app_config, db_path) - app = IngesterApp(config=app_config, db_path=db) + app = IngesterApp(config=app_config, db_path=db_path) if dry_run: report = await app.run_batch_dry_run() if report.failed_sweeps: diff --git a/tests/ingester/test_api.py b/tests/ingester/test_api.py index 301435fc..876aa0bd 100644 --- a/tests/ingester/test_api.py +++ b/tests/ingester/test_api.py @@ -4,7 +4,9 @@ import httpx import pytest from httpx import ASGITransport +from haiku.rag.client.scope import DatabaseScope from haiku.rag.config import AppConfig +from haiku.rag.config.models import LanceDBConfig from haiku.rag.ingester.api.server import APIState, build_app from haiku.rag.ingester.queue.models import JobOp, JobStatus from haiku.rag.sources.base import ( @@ -12,6 +14,7 @@ from haiku.rag.sources.base import ( SourceEvent, SourceEventKind, ) +from tests.conftest import for_path @pytest.fixture @@ -808,7 +811,9 @@ async def _seed_lancedb(path): async def test_database_reports_info(tmp_path, jobs, sync): db_path = tmp_path / "docs.lancedb" await _seed_lancedb(db_path) - state = APIState(config=AppConfig(), job_repo=jobs, sync_repo=sync, db_path=db_path) + state = APIState( + config=AppConfig(), job_repo=jobs, sync_repo=sync, scope=for_path(db_path) + ) async with _client(state) as client: resp = await client.get("/database") assert resp.status_code == 200 @@ -825,12 +830,33 @@ async def test_database_reports_info(tmp_path, jobs, sync): @pytest.mark.asyncio -async def test_database_503_when_db_path_unset(state): +async def test_database_503_when_no_database_is_configured(state): async with _client(state) as client: resp = await client.get("/database") assert resp.status_code == 503 +@pytest.mark.asyncio +async def test_the_report_follows_a_configured_uri(tmp_path, jobs, sync): + """A configured `lancedb.uri` places the database, so the report opens that + and not the local default.""" + db_path = tmp_path / "configured.lancedb" + await _seed_lancedb(db_path) + config = AppConfig(lancedb=LanceDBConfig(uri=str(db_path))) + state = APIState( + config=config, + job_repo=jobs, + sync_repo=sync, + scope=DatabaseScope.resolve(config), + ) + + async with _client(state) as client: + resp = await client.get("/database") + + assert resp.status_code == 200 + assert resp.json()["stored_version"] == "1.2.3" + + @pytest.mark.asyncio async def test_database_requires_auth(state): async with _client(state, auth_token="secret") as client: diff --git a/tests/ingester/test_cli.py b/tests/ingester/test_cli.py index 793fd7e6..5e6fb3d3 100644 --- a/tests/ingester/test_cli.py +++ b/tests/ingester/test_cli.py @@ -11,7 +11,8 @@ import pytest import yaml from typer.testing import CliRunner -from haiku.rag.config import QueueConfig +from haiku.rag.config import AppConfig, QueueConfig +from haiku.rag.config.models import LanceDBConfig from haiku.rag.ingester.app import BatchProgress, BatchReport from haiku.rag.ingester.batch import ( BatchChange, @@ -21,6 +22,7 @@ from haiku.rag.ingester.batch import ( ) from haiku.rag.ingester.cli import _cli as cli from haiku.rag.ingester.cli import _resolve_queue_config +from haiku.rag.ingester.cli import cli as ingester_cli from haiku.rag.ingester.queue.models import JobOp runner = CliRunner() @@ -481,3 +483,61 @@ def test_cli_entry_point_exits_on_migration_error(monkeypatch): with pytest.raises(SystemExit) as exc_info: cli_entry() assert exc_info.value.code == 1 + + +class TestPlacingTheIngesterDatabase: + """The ingester writes wherever the configuration places the database, and + resolves that once. A manufactured local default would silently redirect a + remote deployment to the local disk, because a path is an explicit override + of a configured `lancedb.uri`.""" + + @staticmethod + def _app(config: AppConfig, db_path=None): + from haiku.rag.ingester.app import IngesterApp + + return IngesterApp(config=config, db_path=db_path) + + def test_a_configured_uri_becomes_the_scope(self, tmp_path): + config = AppConfig(lancedb=LanceDBConfig(uri="s3://bucket/prod.lancedb")) + + [ref] = self._app(config)._scope.databases + + assert ref.uri == "s3://bucket/prod.lancedb" + assert ref.db_path is None + + def test_an_override_names_the_database(self, tmp_path): + config = AppConfig(lancedb=LanceDBConfig(uri="s3://bucket/prod.lancedb")) + override = tmp_path / "local.lancedb" + + [ref] = self._app(config, override)._scope.databases + + assert ref.db_path == override + assert ref.uri == "" + + def test_one_configured_database_is_accepted(self, tmp_path): + """A one-entry mapping names which database to write.""" + config = AppConfig( + lancedb=LanceDBConfig(databases={"docs": str(tmp_path / "docs.lancedb")}) + ) + + scope = self._app(config)._scope + + assert scope.names == ("docs",) + assert not scope.covers_multiple + + def test_several_configured_databases_exit_cleanly(self, tmp_path, monkeypatch): + """No selector names one of a set, so the CLI reports it rather than + printing a traceback.""" + config_file = tmp_path / "haiku.rag.yaml" + config_file.write_text( + "lancedb:\n databases:\n" + f" a: {tmp_path / 'a.lancedb'}\n" + f" b: {tmp_path / 'b.lancedb'}\n" + ) + monkeypatch.setattr(sys, "argv", ["haiku-ingester", "run-batch"]) + monkeypatch.setenv("HAIKU_RAG_CONFIG_PATH", str(config_file)) + + with pytest.raises(SystemExit) as exit_info: + ingester_cli() + + assert exit_info.value.code == 1 diff --git a/tests/ingester/test_run_batch.py b/tests/ingester/test_run_batch.py index eb4ee956..5ea4fa85 100644 --- a/tests/ingester/test_run_batch.py +++ b/tests/ingester/test_run_batch.py @@ -7,7 +7,7 @@ import asyncio from contextlib import asynccontextmanager from datetime import UTC, datetime from pathlib import Path -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, MagicMock from urllib.parse import unquote, urlparse import pytest @@ -103,7 +103,11 @@ def use_client(monkeypatch): async def _cm(*_, **__): yield client - monkeypatch.setattr("haiku.rag.client.HaikuRAG", lambda *a, **k: _cm()) + # The app opens the databases its scope covers, so the double stands in + # for `_covering` rather than the public constructor. + stub = MagicMock() + stub._covering = lambda *a, **k: _cm() + monkeypatch.setattr("haiku.rag.client.HaikuRAG", stub) return _install