From f997d7adc9fbe27447a29fd7f110d90669deb210 Mon Sep 17 00:00:00 2001 From: Yiorgis Gozadinos Date: Wed, 15 Jul 2026 15:54:26 +0300 Subject: [PATCH] Consolidate tag primitives --- haiku_rag_slim/haiku/rag/app.py | 16 +++--- haiku_rag_slim/haiku/rag/store/engine.py | 33 +++++++++--- tests/store/test_tags.py | 64 ++++++++++++++++++++++++ 3 files changed, 96 insertions(+), 17 deletions(-) diff --git a/haiku_rag_slim/haiku/rag/app.py b/haiku_rag_slim/haiku/rag/app.py index 46bc557b..302664ca 100644 --- a/haiku_rag_slim/haiku/rag/app.py +++ b/haiku_rag_slim/haiku/rag/app.py @@ -337,20 +337,16 @@ class HaikuRAGApp: # pragma: no cover ) def _tag_write_store(self) -> "Store": - """Writable store for tag create/delete. + """Writable store for tag create/delete with normal validation and + migration checks. - Migration checks stay on: a coordinated tag is only reliable when the - database schema is current, and a writable open of a legacy database - would create missing tables as a side effect. + A coordinated tag is only reliable when the database schema is + current, and a writable open of a legacy database would create + missing tables as a side effect. """ from haiku.rag.store.engine import Store - return Store( - self.db_path, - config=self.config, - skip_validation=True, - read_only=self.read_only, - ) + return Store(self.db_path, config=self.config, read_only=self.read_only) def _tag_read_store(self) -> "Store": """Read-only store for tag inspection; works on old or drifted DBs.""" diff --git a/haiku_rag_slim/haiku/rag/store/engine.py b/haiku_rag_slim/haiku/rag/store/engine.py index 833c0aad..3dc0e091 100644 --- a/haiku_rag_slim/haiku/rag/store/engine.py +++ b/haiku_rag_slim/haiku/rag/store/engine.py @@ -1,5 +1,4 @@ import asyncio -import contextlib import json import logging from dataclasses import dataclass @@ -882,7 +881,7 @@ class Store: self._assert_not_rebuilding() tables = self._tables() - async with self._write_lock: + async with self._rebuild_lock, self._write_lock: existing = [ table_name for table_name, table in tables.items() @@ -902,10 +901,19 @@ class Store: for table_name, table in tables.items(): await table.tags.create(name, versions[table_name]) created.append(table_name) - except Exception: + except Exception as exc: + failed_cleanup: list[str] = [] for table_name in created: - with contextlib.suppress(Exception): + try: await tables[table_name].tags.delete(name) + except Exception: + failed_cleanup.append(table_name) + if failed_cleanup: + raise RuntimeError( + f"Tag '{name}' creation failed ({exc}) and cleanup " + f"failed on: {', '.join(failed_cleanup)}. A partial " + "tag may remain; delete it with delete_tag." + ) from exc raise async def list_tags(self) -> dict[str, TagInfo]: @@ -933,17 +941,28 @@ class Store: Raises: ReadOnlyError: If the store is in read-only mode. ValueError: If a rebuild is in progress or no table has the tag. + RuntimeError: If deletion failed on some tables; remnants remain + until a retry succeeds. """ self._assert_writable() self._assert_not_rebuilding() - async with self._write_lock: + async with self._rebuild_lock, self._write_lock: found = False - for table in self._tables().values(): + failed: list[str] = [] + for table_name, table in self._tables().items(): if name in await table.tags.list(): - await table.tags.delete(name) found = True + try: + await table.tags.delete(name) + except Exception: + failed.append(table_name) if not found: raise ValueError(f"Tag '{name}' does not exist") + if failed: + raise RuntimeError( + f"Tag '{name}' deletion failed on: {', '.join(failed)}. " + "Remnants remain; retry delete_tag." + ) async def list_table_versions(self, table_name: str) -> list[dict[str, Any]]: """List version history for a table. diff --git a/tests/store/test_tags.py b/tests/store/test_tags.py index f005195a..e5f867ef 100644 --- a/tests/store/test_tags.py +++ b/tests/store/test_tags.py @@ -85,6 +85,70 @@ async def test_create_tag_rolls_back_own_tags_on_failure(temp_db_path, monkeypat assert tags["keep"].complete is True +@pytest.mark.asyncio +async def test_create_tag_reports_failed_cleanup(temp_db_path, monkeypatch): + """When midway-failure cleanup also fails, the error reports both the + original failure and the remaining partial-tag risk.""" + async with Store(temp_db_path, create=True) as store: + real_create = AsyncTags.create + calls = {"n": 0} + + async def flaky_create(self, name: str, version: int) -> None: + calls["n"] += 1 + if calls["n"] == 4: + raise RuntimeError("create boom") + await real_create(self, name, version) + + async def failing_delete(self, name: str) -> None: + raise RuntimeError("delete boom") + + monkeypatch.setattr(AsyncTags, "create", flaky_create) + monkeypatch.setattr(AsyncTags, "delete", failing_delete) + + with pytest.raises(RuntimeError) as exc_info: + await store.create_tag("broken") + + msg = str(exc_info.value) + assert "create boom" in msg + assert "partial" in msg + assert exc_info.value.__cause__ is not None + + monkeypatch.undo() + tags = await store.list_tags() + assert tags["broken"].complete is False + + +@pytest.mark.asyncio +async def test_delete_tag_reports_failed_tables(temp_db_path, monkeypatch): + """delete_tag never claims success when remnants remain: it names the + tables where deletion failed.""" + async with Store(temp_db_path, create=True) as store: + await store.create_tag("release-1") + + real_delete = AsyncTags.delete + calls = {"n": 0} + + async def flaky_delete(self, name: str) -> None: + calls["n"] += 1 + if calls["n"] == 2: + raise RuntimeError("delete boom") + await real_delete(self, name) + + monkeypatch.setattr(AsyncTags, "delete", flaky_delete) + + with pytest.raises(RuntimeError) as exc_info: + await store.delete_tag("release-1") + + assert "document_meta" in str(exc_info.value) + + monkeypatch.undo() + tags = await store.list_tags() + assert set(tags["release-1"].tables) == {"document_meta"} + + await store.delete_tag("release-1") + assert await store.list_tags() == {} + + @pytest.mark.asyncio async def test_create_tag_waits_for_write_lock(temp_db_path): """create_tag serializes with client writes so a write cannot land