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.
This commit is contained in:
parent
b8bf846bb7
commit
206d29b74a
8 changed files with 122 additions and 26 deletions
|
|
@ -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.
|
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
|
## Multiple Databases
|
||||||
|
|
||||||
Use `lancedb.databases` to name local or remote databases that should be searched
|
Use `lancedb.databases` to name local or remote databases that should be searched
|
||||||
|
|
|
||||||
|
|
@ -1,5 +1,6 @@
|
||||||
from fastapi import APIRouter, Depends, HTTPException, status
|
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.ingester.api.server import APIState, get_state
|
||||||
from haiku.rag.store.info import DatabaseInfo, gather_database_info
|
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
|
Opens a fresh read-only connection per call; not cached and not part of
|
||||||
the dashboard's polling loop."""
|
the dashboard's polling loop."""
|
||||||
if state.db_path is None:
|
if state.scope is None:
|
||||||
raise HTTPException(
|
raise HTTPException(
|
||||||
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
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))
|
||||||
|
|
|
||||||
|
|
@ -1,5 +1,4 @@
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from pathlib import Path
|
|
||||||
from typing import TYPE_CHECKING
|
from typing import TYPE_CHECKING
|
||||||
|
|
||||||
from fastapi import Depends, FastAPI, Request
|
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
|
from haiku.rag.ingester.api.auth import require_auth
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
|
from haiku.rag.client.scope import DatabaseScope
|
||||||
from haiku.rag.config import AppConfig
|
from haiku.rag.config import AppConfig
|
||||||
from haiku.rag.ingester.pollers.manager import PollerManager
|
from haiku.rag.ingester.pollers.manager import PollerManager
|
||||||
from haiku.rag.ingester.queue.repository import JobRepo, SyncStateRepo
|
from haiku.rag.ingester.queue.repository import JobRepo, SyncStateRepo
|
||||||
|
|
@ -23,7 +23,7 @@ class APIState:
|
||||||
sync_repo: "SyncStateRepo"
|
sync_repo: "SyncStateRepo"
|
||||||
pool: "WorkerPool | None" = None
|
pool: "WorkerPool | None" = None
|
||||||
pollers: "PollerManager | None" = None
|
pollers: "PollerManager | None" = None
|
||||||
db_path: Path | None = None
|
scope: "DatabaseScope | None" = None
|
||||||
|
|
||||||
|
|
||||||
def get_state(request: Request) -> APIState:
|
def get_state(request: Request) -> APIState:
|
||||||
|
|
|
||||||
|
|
@ -72,9 +72,13 @@ class IngesterApp:
|
||||||
WorkerPool, and a HaikuRAG client for the worker pool to ingest through.
|
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._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._engine: AsyncEngine | None = None
|
||||||
self._jobs: JobRepo | None = None
|
self._jobs: JobRepo | None = None
|
||||||
self._sync: SyncStateRepo | 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
|
# The ingester is the sole writer for its LanceDB target; create on
|
||||||
# first start so docker-compose / fresh deployments don't require a
|
# first start so docker-compose / fresh deployments don't require a
|
||||||
# manual `haiku-rag init`.
|
# manual `haiku-rag init`.
|
||||||
async with HaikuRAG(
|
async with HaikuRAG._covering(
|
||||||
self._db_path, config=self._config, create=True
|
self._scope, self._config, create=True
|
||||||
) as client:
|
) as client:
|
||||||
self._pollers = PollerManager(
|
self._pollers = PollerManager(
|
||||||
configs=ingester_cfg.sources,
|
configs=ingester_cfg.sources,
|
||||||
|
|
@ -471,7 +475,7 @@ class IngesterApp:
|
||||||
sync_repo=self._sync,
|
sync_repo=self._sync,
|
||||||
pool=self._pool,
|
pool=self._pool,
|
||||||
pollers=self._pollers,
|
pollers=self._pollers,
|
||||||
db_path=self._db_path,
|
scope=self._scope,
|
||||||
)
|
)
|
||||||
if ingester_cfg.api.auth_token is None:
|
if ingester_cfg.api.auth_token is None:
|
||||||
logger.warning("API auth_token is unset — control plane is unauthenticated")
|
logger.warning("API auth_token is unset — control plane is unauthenticated")
|
||||||
|
|
|
||||||
|
|
@ -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.ingester.queue.migrations import open_queue # noqa: E402
|
||||||
from haiku.rag.logging import configure_cli_logging # noqa: E402
|
from haiku.rag.logging import configure_cli_logging # noqa: E402
|
||||||
from haiku.rag.store.exceptions import ( # noqa: E402
|
from haiku.rag.store.exceptions import ( # noqa: E402
|
||||||
|
AmbiguousDatabaseError,
|
||||||
MigrationRequiredError,
|
MigrationRequiredError,
|
||||||
ReadOnlyError,
|
ReadOnlyError,
|
||||||
)
|
)
|
||||||
|
|
@ -70,7 +71,7 @@ def cli() -> None:
|
||||||
"""Entry point that translates store-state errors into a clean exit."""
|
"""Entry point that translates store-state errors into a clean exit."""
|
||||||
try:
|
try:
|
||||||
_cli()
|
_cli()
|
||||||
except (MigrationRequiredError, ReadOnlyError) as e:
|
except (AmbiguousDatabaseError, MigrationRequiredError, ReadOnlyError) as e:
|
||||||
typer.echo(f"Error: {e}", err=True)
|
typer.echo(f"Error: {e}", err=True)
|
||||||
sys.exit(1)
|
sys.exit(1)
|
||||||
|
|
||||||
|
|
@ -149,10 +150,6 @@ def queue_migrate(
|
||||||
typer.echo(f"Queue at {_queue_target(queue_config)} is up to date")
|
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:
|
def _default_manifest_path() -> Path:
|
||||||
datestamp = datetime.now(UTC).strftime("%Y%m%d-%H%M%SZ")
|
datestamp = datetime.now(UTC).strftime("%Y%m%d-%H%M%SZ")
|
||||||
return Path(f"manifest-{datestamp}.yaml")
|
return Path(f"manifest-{datestamp}.yaml")
|
||||||
|
|
@ -216,7 +213,7 @@ def serve(
|
||||||
db: Path | None = typer.Option(
|
db: Path | None = typer.Option(
|
||||||
None,
|
None,
|
||||||
"--db",
|
"--db",
|
||||||
help="LanceDB path (overrides config.storage.data_dir).",
|
help="LanceDB path (overrides the configured database location).",
|
||||||
),
|
),
|
||||||
host: str | None = typer.Option(
|
host: str | None = typer.Option(
|
||||||
None,
|
None,
|
||||||
|
|
@ -252,8 +249,7 @@ def serve(
|
||||||
app_config.ingester.api.port = port
|
app_config.ingester.api.port = port
|
||||||
if root_path is not None:
|
if root_path is not None:
|
||||||
app_config.ingester.api.root_path = root_path
|
app_config.ingester.api.root_path = root_path
|
||||||
db_path = _resolve_db_path(app_config, db)
|
app = IngesterApp(config=app_config, db_path=db)
|
||||||
app = IngesterApp(config=app_config, db_path=db_path)
|
|
||||||
asyncio.run(app.serve(api=not no_api))
|
asyncio.run(app.serve(api=not no_api))
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -262,7 +258,7 @@ def run_batch(
|
||||||
db: Path | None = typer.Option(
|
db: Path | None = typer.Option(
|
||||||
None,
|
None,
|
||||||
"--db",
|
"--db",
|
||||||
help="LanceDB path (overrides config.storage.data_dir).",
|
help="LanceDB path (overrides the configured database location).",
|
||||||
),
|
),
|
||||||
dry_run: bool = typer.Option(
|
dry_run: bool = typer.Option(
|
||||||
False,
|
False,
|
||||||
|
|
@ -312,8 +308,7 @@ async def _run_batch(
|
||||||
) -> None:
|
) -> None:
|
||||||
from haiku.rag.ingester.app import IngesterApp
|
from haiku.rag.ingester.app import IngesterApp
|
||||||
|
|
||||||
db = _resolve_db_path(app_config, db_path)
|
app = IngesterApp(config=app_config, db_path=db_path)
|
||||||
app = IngesterApp(config=app_config, db_path=db)
|
|
||||||
if dry_run:
|
if dry_run:
|
||||||
report = await app.run_batch_dry_run()
|
report = await app.run_batch_dry_run()
|
||||||
if report.failed_sweeps:
|
if report.failed_sweeps:
|
||||||
|
|
|
||||||
|
|
@ -4,7 +4,9 @@ import httpx
|
||||||
import pytest
|
import pytest
|
||||||
from httpx import ASGITransport
|
from httpx import ASGITransport
|
||||||
|
|
||||||
|
from haiku.rag.client.scope import DatabaseScope
|
||||||
from haiku.rag.config import AppConfig
|
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.api.server import APIState, build_app
|
||||||
from haiku.rag.ingester.queue.models import JobOp, JobStatus
|
from haiku.rag.ingester.queue.models import JobOp, JobStatus
|
||||||
from haiku.rag.sources.base import (
|
from haiku.rag.sources.base import (
|
||||||
|
|
@ -12,6 +14,7 @@ from haiku.rag.sources.base import (
|
||||||
SourceEvent,
|
SourceEvent,
|
||||||
SourceEventKind,
|
SourceEventKind,
|
||||||
)
|
)
|
||||||
|
from tests.conftest import for_path
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
|
|
@ -808,7 +811,9 @@ async def _seed_lancedb(path):
|
||||||
async def test_database_reports_info(tmp_path, jobs, sync):
|
async def test_database_reports_info(tmp_path, jobs, sync):
|
||||||
db_path = tmp_path / "docs.lancedb"
|
db_path = tmp_path / "docs.lancedb"
|
||||||
await _seed_lancedb(db_path)
|
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:
|
async with _client(state) as client:
|
||||||
resp = await client.get("/database")
|
resp = await client.get("/database")
|
||||||
assert resp.status_code == 200
|
assert resp.status_code == 200
|
||||||
|
|
@ -825,12 +830,33 @@ async def test_database_reports_info(tmp_path, jobs, sync):
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@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:
|
async with _client(state) as client:
|
||||||
resp = await client.get("/database")
|
resp = await client.get("/database")
|
||||||
assert resp.status_code == 503
|
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
|
@pytest.mark.asyncio
|
||||||
async def test_database_requires_auth(state):
|
async def test_database_requires_auth(state):
|
||||||
async with _client(state, auth_token="secret") as client:
|
async with _client(state, auth_token="secret") as client:
|
||||||
|
|
|
||||||
|
|
@ -11,7 +11,8 @@ import pytest
|
||||||
import yaml
|
import yaml
|
||||||
from typer.testing import CliRunner
|
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.app import BatchProgress, BatchReport
|
||||||
from haiku.rag.ingester.batch import (
|
from haiku.rag.ingester.batch import (
|
||||||
BatchChange,
|
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 _cli as cli
|
||||||
from haiku.rag.ingester.cli import _resolve_queue_config
|
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
|
from haiku.rag.ingester.queue.models import JobOp
|
||||||
|
|
||||||
runner = CliRunner()
|
runner = CliRunner()
|
||||||
|
|
@ -481,3 +483,61 @@ def test_cli_entry_point_exits_on_migration_error(monkeypatch):
|
||||||
with pytest.raises(SystemExit) as exc_info:
|
with pytest.raises(SystemExit) as exc_info:
|
||||||
cli_entry()
|
cli_entry()
|
||||||
assert exc_info.value.code == 1
|
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
|
||||||
|
|
|
||||||
|
|
@ -7,7 +7,7 @@ import asyncio
|
||||||
from contextlib import asynccontextmanager
|
from contextlib import asynccontextmanager
|
||||||
from datetime import UTC, datetime
|
from datetime import UTC, datetime
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from unittest.mock import AsyncMock
|
from unittest.mock import AsyncMock, MagicMock
|
||||||
from urllib.parse import unquote, urlparse
|
from urllib.parse import unquote, urlparse
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
@ -103,7 +103,11 @@ def use_client(monkeypatch):
|
||||||
async def _cm(*_, **__):
|
async def _cm(*_, **__):
|
||||||
yield client
|
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
|
return _install
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue