Consolidate tag primitives

This commit is contained in:
Yiorgis Gozadinos 2026-07-15 15:54:26 +03:00
parent 0e271eaf4b
commit f997d7adc9
No known key found for this signature in database
3 changed files with 96 additions and 17 deletions

View file

@ -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."""

View file

@ -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.

View file

@ -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