Remove unreachable code and fix two defects it surfaced

Delete repository methods with no callers (SettingsRepository CRUD,
ChunkRepository.update/delete/get_chunks_in_range), the DataFrame branch of
_process_search_results whose only caller always passes a query, and guards
that cannot be reached from their call sites: the rebuild mode=None default,
the staging drop already performed by _resolve_rebuild_recovery, the empty
batch skip, two context fast paths, the doctor prefix guard, the poller
_task attribute that is never assigned, and a docling caption fallback for
a field name no item class defines.

set_haiku_version built a recreated settings row from the process-global
Config rather than the store's own, so a store opened with a custom config
stamped global settings into the database.

check_source_accessible called urlparse outside its try block, so a stored
URI with a malformed IPv6 host raised ValueError instead of reporting the
source as inaccessible, aborting the whole rebuild sweep.
This commit is contained in:
Yiorgis Gozadinos 2026-07-26 18:45:48 +03:00
parent e2b273ee2a
commit dddaf0f84c
No known key found for this signature in database
12 changed files with 71 additions and 146 deletions

View file

@ -1,6 +1,15 @@
# Changelog # Changelog
## [Unreleased] ## [Unreleased]
### Fixed
- `Store.set_haiku_version` stamps the store's own config into a recreated settings row instead of the process-global `Config`.
- `check_source_accessible` returns `False` for a URI that fails to parse instead of raising `ValueError`.
### Removed
- `SettingsRepository.create`, `get_by_id`, `update`, `delete`, `list_all` and `ChunkRepository.update`, `delete`, `get_chunks_in_range`.
## [0.70.0] - 2026-07-25 ## [0.70.0] - 2026-07-25
### Added ### Added

View file

@ -906,13 +906,17 @@ async def update_document(
def check_source_accessible(uri: str) -> bool: def check_source_accessible(uri: str) -> bool:
"""Check if a document's source URI is accessible.""" """Check if a document's source URI is accessible.
parsed_url = urlparse(uri)
A stored URI that no longer parses (``urlparse`` rejects malformed IPv6
hosts) counts as inaccessible rather than aborting the caller's sweep.
"""
try: try:
parsed_url = urlparse(uri)
if parsed_url.scheme == "file": if parsed_url.scheme == "file":
return Path(parsed_url.path).exists() return Path(parsed_url.path).exists()
elif parsed_url.scheme in ("http", "https", "s3"): elif parsed_url.scheme in ("http", "https", "s3"):
return True return True
return False return False
except Exception: except ValueError:
return False return False

View file

@ -66,7 +66,7 @@ class _StagingMarkerRecord(LanceModel):
async def rebuild_database( async def rebuild_database(
client: "HaikuRAG", mode: "RebuildMode | None" = None client: "HaikuRAG", mode: "RebuildMode"
) -> AsyncGenerator[str, None]: ) -> AsyncGenerator[str, None]:
"""Rebuild the database with the specified mode. """Rebuild the database with the specified mode.
@ -79,9 +79,6 @@ async def rebuild_database(
""" """
from haiku.rag.client import RebuildMode from haiku.rag.client import RebuildMode
if mode is None:
mode = RebuildMode.FULL
async with client.store._rebuild_lock: async with client.store._rebuild_lock:
if mode == RebuildMode.SET_EMBEDDER: if mode == RebuildMode.SET_EMBEDDER:
await _set_embedder(client) await _set_embedder(client)
@ -287,8 +284,6 @@ async def _populate_staging_table(client: "HaikuRAG") -> None:
""" """
db = client.store.db db = client.store.db
tables = (await db.list_tables()).tables tables = (await db.list_tables()).tables
if _STAGING_TABLE_NAME in tables:
await db.drop_table(_STAGING_TABLE_NAME)
staging = await db.create_table(_STAGING_TABLE_NAME, schema=_StagingChunkRecord) staging = await db.create_table(_STAGING_TABLE_NAME, schema=_StagingChunkRecord)
if "chunks" not in tables: if "chunks" not in tables:
@ -301,8 +296,6 @@ async def _populate_staging_table(client: "HaikuRAG") -> None:
) )
async for batch in stream: async for batch in stream:
rows = batch.to_pylist() rows = batch.to_pylist()
if not rows:
continue
records = [ records = [
_StagingChunkRecord( _StagingChunkRecord(
id=r["id"], id=r["id"],

View file

@ -101,8 +101,6 @@ def _evidence_anchors(content: str, max_chars: int) -> list[str]:
mid = len(content) // 2 mid = len(content) // 2
start = max(0, mid - target // 2) start = max(0, mid - target // 2)
anchors.append(content[start : start + target]) anchors.append(content[start : start + target])
if not anchors:
anchors.append(content[:max_chars] if max_chars > 0 else content)
return anchors return anchors
@ -289,8 +287,6 @@ def _add_input_pages_for_surviving_refs(
) -> None: ) -> None:
"""Fill missing item-table pages from inputs whose own refs all survived.""" """Fill missing item-table pages from inputs whose own refs all survived."""
surviving = set(refs) surviving = set(refs)
if not surviving:
return
for result in original_results: for result in original_results:
if not result.page_numbers or not result.doc_item_refs: if not result.page_numbers or not result.doc_item_refs:
continue continue

View file

@ -330,8 +330,6 @@ def _common_path_prefix(labels: list[str]) -> str:
Returns "" unless the shared prefix is long enough to be worth factoring out Returns "" unless the shared prefix is long enough to be worth factoring out
of every line (deep URI trees are otherwise unreadable). of every line (deep URI trees are otherwise unreadable).
""" """
if len(labels) < 2:
return ""
lo, hi = min(labels), max(labels) lo, hi = min(labels), max(labels)
end = 0 end = 0
while end < len(lo) and lo[end] == hi[end]: while end < len(lo) and lo[end] == hi[end]:

View file

@ -55,7 +55,6 @@ class BasePoller:
self._sync = sync_repo self._sync = sync_repo
self._breaker = breaker or CircuitBreaker(config.circuit_breaker) self._breaker = breaker or CircuitBreaker(config.circuit_breaker)
self._stop = asyncio.Event() self._stop = asyncio.Event()
self._task: asyncio.Task | None = None
self._last_polled_at: datetime | None = None self._last_polled_at: datetime | None = None
self._last_skip_reason: str | None = None self._last_skip_reason: str | None = None
self._default_max_attempts = default_max_attempts self._default_max_attempts = default_max_attempts
@ -84,9 +83,6 @@ class BasePoller:
async def stop(self) -> None: async def stop(self) -> None:
self._stop.set() self._stop.set()
if self._task is not None:
await asyncio.gather(self._task, return_exceptions=True)
self._task = None
async def _stagger_start(self) -> bool: async def _stagger_start(self) -> bool:
"""Sleep a random fraction of the interval so pollers sharing an """Sleep a random fraction of the interval so pollers sharing an

View file

@ -847,7 +847,7 @@ class Store:
) )
else: else:
# Create new settings record # Create new settings record
settings_data = Config.model_dump(mode="json") settings_data = self._config.model_dump(mode="json")
settings_data["version"] = version settings_data["version"] = version
await self.settings_table.add( await self.settings_table.add(
[SettingsRecord(id="settings", settings=json.dumps(settings_data))] [SettingsRecord(id="settings", settings=json.dumps(settings_data))]

View file

@ -105,10 +105,6 @@ def extract_item_text(
except Exception: except Exception:
pass pass
if caption := getattr(item, "caption", None):
if hasattr(caption, "text"):
return caption.text
return None return None

View file

@ -4,7 +4,6 @@ from typing import TYPE_CHECKING
from uuid import uuid4 from uuid import uuid4
if TYPE_CHECKING: if TYPE_CHECKING:
import pandas as pd
from lancedb.query import AsyncQueryBase from lancedb.query import AsyncQueryBase
from lancedb.index import FTS from lancedb.index import FTS
@ -153,40 +152,6 @@ class ChunkRepository:
order=chunk_record.order, order=chunk_record.order,
) )
async def update(self, entity: Chunk) -> Chunk:
"""Update an existing chunk.
Chunk must have embedding set before calling this method.
"""
self.store._assert_writable()
assert entity.id, "Chunk ID is required for update"
assert entity.embedding is not None, "Chunk must have an embedding"
await self.store.chunks_table.update(
{
"document_id": entity.document_id,
"content": entity.content,
"content_fts": self._contextualize_content(entity),
"metadata": json.dumps(
{k: v for k, v in entity.metadata.items() if k != "order"}
),
"order": int(entity.order),
"vector": entity.embedding,
},
where=f"id = '{entity.id}'",
)
return entity
async def delete(self, entity_id: str) -> bool:
"""Delete a chunk by its ID."""
self.store._assert_writable()
chunk = await self.get_by_id(entity_id)
if chunk is None:
return False
await self.store.chunks_table.delete(f"id = '{entity_id}'")
return True
async def list_all( async def list_all(
self, limit: int | None = None, offset: int | None = None self, limit: int | None = None, offset: int | None = None
) -> list[Chunk]: ) -> list[Chunk]:
@ -417,46 +382,10 @@ class ChunkRepository:
) )
return len(df) return len(df)
async def get_chunks_in_range(
self, document_id: str, min_order: int, max_order: int
) -> list[Chunk]:
"""Get chunks for a document within an order range.
Args:
document_id: The document ID to get chunks for.
min_order: Minimum order value (inclusive).
max_order: Maximum order value (inclusive).
Returns:
List of chunks within the order range.
"""
where = (
f"document_id = '{document_id}'"
f" AND `order` >= {min_order}"
f" AND `order` <= {max_order}"
)
results = await query_to_pydantic(
self.store.chunks_table.query().where(where), self.store.ChunkRecord
)
return [
Chunk(
id=rec.id,
document_id=rec.document_id,
content=rec.content,
metadata=json.loads(rec.metadata),
order=rec.order,
)
for rec in results
]
async def _process_search_results( async def _process_search_results(
self, query_result: "pd.DataFrame | AsyncQueryBase" self, query_result: "AsyncQueryBase"
) -> list[tuple[Chunk, float]]: ) -> list[tuple[Chunk, float]]:
"""Process search results into chunks with document info and scores. """Process search results into chunks with document info and scores."""
Args:
query_result: Either a pandas DataFrame or a LanceDB async query result
"""
import pandas as pd import pandas as pd
def extract_scores(df: pd.DataFrame) -> list[float]: def extract_scores(df: pd.DataFrame) -> list[float]:
@ -473,11 +402,6 @@ class ChunkRepository:
else: else:
raise ValueError("Unknown search result format, cannot extract scores") raise ValueError("Unknown search result format, cannot extract scores")
# Convert everything to DataFrame for uniform processing
if isinstance(query_result, pd.DataFrame):
df = query_result
else:
# Convert LanceDB query result to DataFrame
df = await query_result.to_pandas() df = await query_result.to_pandas()
# Extract scores # Extract scores

View file

@ -18,47 +18,6 @@ class SettingsRepository:
def __init__(self, store: Store) -> None: def __init__(self, store: Store) -> None:
self.store = store self.store = store
async def create(self, entity: dict) -> dict:
"""Create settings in the database."""
settings_record = SettingsRecord(id="settings", settings=json.dumps(entity))
await self.store.settings_table.add([settings_record])
return entity
async def get_by_id(self, entity_id: str) -> dict | None:
"""Get settings by ID."""
results = await query_to_pydantic(
self.store.settings_table.query().where(f"id = '{entity_id}'").limit(1),
SettingsRecord,
)
if not results:
return None
return json.loads(results[0].settings) if results[0].settings else {}
async def update(self, entity: dict) -> dict:
"""Update existing settings."""
await self.store.settings_table.update(
{"settings": json.dumps(entity)}, where="id = 'settings'"
)
return entity
async def delete(self, entity_id: str) -> bool:
"""Delete settings by ID."""
await self.store.settings_table.delete(f"id = '{entity_id}'")
return True
async def list_all(
self, limit: int | None = None, offset: int | None = None
) -> list[dict]:
"""List all settings."""
results = await query_to_pydantic(
self.store.settings_table.query(), SettingsRecord
)
return [
json.loads(record.settings) if record.settings else {} for record in results
]
async def get_current_settings(self) -> dict: async def get_current_settings(self) -> dict:
"""Get the current settings.""" """Get the current settings."""
results = await query_to_pydantic( results = await query_to_pydantic(

View file

@ -14,6 +14,7 @@ from haiku.rag.client.documents import (
DocumentImport, DocumentImport,
_prepare_document_from_docling, _prepare_document_from_docling,
_write_fetch_body, _write_fetch_body,
check_source_accessible,
) )
from haiku.rag.config import Config from haiku.rag.config import Config
from haiku.rag.store.compression import decompress_json from haiku.rag.store.compression import decompress_json
@ -2254,3 +2255,27 @@ async def test_metadata_only_update_waits_for_write_lock(temp_db_path):
assert not task.done() assert not task.done()
updated = await task updated = await task
assert updated.metadata == {"k": "v"} assert updated.metadata == {"k": "v"}
@pytest.mark.parametrize(
"uri,expected",
[
("https://example.com/doc.pdf", True),
("s3://bucket/key", True),
("mem://not-a-source", False),
# urlparse rejects a malformed IPv6 host; a stored URI that no longer
# parses must not abort the caller's rebuild sweep.
("http://[::1", False),
],
ids=["https", "s3", "unknown_scheme", "unparseable"],
)
def test_check_source_accessible(uri, expected):
assert check_source_accessible(uri) is expected
def test_check_source_accessible_file_uri(tmp_path):
existing = tmp_path / "there.txt"
existing.write_text("x")
assert check_source_accessible(existing.as_uri()) is True
assert check_source_accessible((tmp_path / "gone.txt").as_uri()) is False

View file

@ -44,6 +44,31 @@ async def test_settings_save_and_retrieve(temp_db_path):
Config.processing.chunk_size = original_chunk_size Config.processing.chunk_size = original_chunk_size
@pytest.mark.asyncio
async def test_set_haiku_version_recreates_row_from_store_config(temp_db_path):
"""Recreating a missing settings row stamps the store's own config, not the
process-global one."""
from haiku.rag.store.engine import Store
from haiku.rag.store.repositories.settings import SettingsRepository
config = AppConfig()
config.processing.chunk_size = Config.processing.chunk_size + 512
async with Store(temp_db_path, config=config, create=True) as store:
settings_repo = SettingsRepository(store)
await store.settings_table.delete("id = 'settings'")
assert await settings_repo.get_current_settings() == {}
assert await store.get_haiku_version() == "0.0.0"
await store.set_haiku_version("1.2.3")
recreated = await settings_repo.get_current_settings()
assert recreated["version"] == "1.2.3"
assert recreated["processing"]["chunk_size"] == config.processing.chunk_size
assert await store.get_haiku_version() == "1.2.3"
class TestValidateConfigCompatibility: class TestValidateConfigCompatibility:
"""Tests for validate_config_compatibility method.""" """Tests for validate_config_compatibility method."""