danvics/docling-studio
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

feat(#203): per (document, store) ingestion state

Browse source 
Introduces the data layer for multi-store ingestion. Documents can now
live in multiple stores, each with its own Ingested/Stale/Failed state.
The doc-level lifecycle (#202) becomes the aggregate over all per-store
links, computed by a pure domain function.

Domain
- Store entity (name, slug, kind, embedder, config, is_default)
- DocumentStoreLink entity with mark_ingested / mark_stale / mark_failed
  helpers
- StoreKind and DocumentStoreLinkState enums
- aggregate_lifecycle(): pure function — Failed > Stale > Ingested
  > fallback (the doc's pre-link Uploaded/Parsed/Chunked state)

Persistence
- New tables 'stores' and 'document_store_links' with the right indexes
  (doc_id, store_id, state) and a UNIQUE (doc, store) on the link
- Default 'opensearch' store seeded idempotently in init_db, embedder
  pulled from DEFAULT_EMBEDDER (fallback bge-m3)
- SqliteStoreRepository (find_by_slug, find_by_id, get_default, …)
- SqliteDocumentStoreLinkRepository with ON CONFLICT … DO UPDATE upsert

Ports
- StoreRepository and DocumentStoreLinkRepository protocols added

Tests
- 14 new tests: seed idempotency, insert/find round-trips, UNIQUE
  constraint, cascade delete with the document, every link state
  round-trips, aggregation rule with all branches

Refs #203
This commit is contained in:
Pier-Jean Malandrino 2026-04-29 15:38:56 +02:00
parent 16861c50cf
commit 4c30e5eb8f
9 changed files with 671 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

58
document-parser/domain/lifecycle_aggregation.py Normal file
View file

@ -0,0 +1,58 @@
"""Aggregate the per-(document, store) link states into a single
document-level lifecycle state.
The doc lifecycle column is the materialized result of this rule. It is
recomputed any time a link write happens. Read paths use the stored
column directly (cheap) — they do not call this rule on every GET.
The rule prefers "more concerning" states first:
any link FAILED -> Document FAILED
any link STALE -> Document STALE
any link INGESTED -> Document INGESTED
no links -> keep the document's current pre-link state
(Uploaded / Parsed / Chunked) — the caller is
responsible for not overwriting that.
If you find yourself wanting a fourth case, you probably want a new
link state, not a new aggregation branch.
This module is pure: no I/O, no datetime — just data in / data out.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from domain.value_objects import DocumentLifecycleState, DocumentStoreLinkState
if TYPE_CHECKING:
from collections.abc import Iterable
from domain.models import DocumentStoreLink
def aggregate_lifecycle(
links: Iterable[DocumentStoreLink],
*,
fallback: DocumentLifecycleState,
) -> DocumentLifecycleState:
"""Compute the aggregate document lifecycle state.
Args:
links: every `DocumentStoreLink` for the document. May be empty.
fallback: the lifecycle state to return when there are no links —
typically the document's current pre-link state (`Uploaded`,
`Parsed`, or `Chunked`).
Returns:
The aggregate `DocumentLifecycleState`.
"""
states = {link.state for link in links}
if DocumentStoreLinkState.FAILED in states:
return DocumentLifecycleState.FAILED
if DocumentStoreLinkState.STALE in states:
return DocumentLifecycleState.STALE
if DocumentStoreLinkState.INGESTED in states:
return DocumentLifecycleState.INGESTED
return fallback

73
document-parser/domain/models.py
View file

@ -9,7 +9,11 @@ from datetime import UTC, datetime
from domain.events import DocumentLifecycleChanged
from domain.lifecycle import assert_transition
from domain.value_objects import DocumentLifecycleState
from domain.value_objects import (
DocumentLifecycleState,
DocumentStoreLinkState,
StoreKind,
)
class AnalysisStatus(enum.StrEnum):
@ -130,3 +134,70 @@ class AnalysisJob:
self.status = AnalysisStatus.FAILED
self.error_message = error
self.completed_at = _utcnow()
@dataclass
class Store:
"""A logical destination for ingested chunks.
A `Store` represents a named, configurable target for ingestion (e.g.
`rh-corpus-v3`, `legal-v1`). It is decoupled from the underlying
`VectorStore` adapter (which represents the *technology*, like
OpenSearch). One adapter can serve many stores by namespacing the
physical index name on `slug`.
See design doc `docs/design/203-per-store-ingestion-state.md`.
"""
id: str = field(default_factory=_new_id)
name: str = ""
slug: str = ""
kind: StoreKind = StoreKind.OPENSEARCH
embedder: str = ""
config: dict = field(default_factory=dict)
is_default: bool = False
created_at: datetime = field(default_factory=_utcnow)
@dataclass
class DocumentStoreLink:
"""A live record of a document's presence in a single store.
The link carries the per-pair state (Ingested / Stale / Failed) plus
metadata used by the auto-stale detection (#204) and the chunks
editor (#205). One row per (document, store) pair — enforced by a
UNIQUE constraint at the persistence layer.
"""
id: str = field(default_factory=_new_id)
document_id: str = ""
store_id: str = ""
state: DocumentStoreLinkState = DocumentStoreLinkState.INGESTED
chunkset_hash: str | None = None
last_push_at: datetime | None = None
last_run_id: str | None = None
error_message: str | None = None
def mark_ingested(
self,
*,
hash_: str,
at: datetime,
run_id: str | None = None,
) -> None:
"""Record a successful push: chunkset hash + timestamp + run id."""
self.state = DocumentStoreLinkState.INGESTED
self.chunkset_hash = hash_
self.last_push_at = at
self.last_run_id = run_id
self.error_message = None
def mark_stale(self) -> None:
"""Mark the link as stale (source chunkset drifted from pushed)."""
self.state = DocumentStoreLinkState.STALE
self.error_message = None
def mark_failed(self, *, error: str) -> None:
"""Record a failed push attempt."""
self.state = DocumentStoreLinkState.FAILED
self.error_message = error

32
document-parser/domain/ports.py
View file

@ -11,7 +11,7 @@ from typing import TYPE_CHECKING, Protocol, runtime_checkable
if TYPE_CHECKING:
from datetime import datetime
from domain.models import AnalysisJob, Document
from domain.models import AnalysisJob, Document, DocumentStoreLink, Store
from domain.value_objects import (
ChunkingOptions,
ChunkResult,
@ -98,6 +98,36 @@ class DocumentRepository(Protocol):
async def delete(self, doc_id: str) -> bool: ...
class StoreRepository(Protocol):
"""Port for `Store` persistence (introduced by #203)."""
async def insert(self, store: Store) -> None: ...
async def find_all(self) -> list[Store]: ...
async def find_by_slug(self, slug: str) -> Store | None: ...
async def find_by_id(self, store_id: str) -> Store | None: ...
async def get_default(self) -> Store | None: ...
class DocumentStoreLinkRepository(Protocol):
"""Port for `DocumentStoreLink` persistence (introduced by #203)."""
async def upsert(self, link: DocumentStoreLink) -> None:
"""Insert or update by (document_id, store_id)."""
...
async def find_for_document(self, document_id: str) -> list[DocumentStoreLink]: ...
async def find_for_store(self, store_id: str) -> list[DocumentStoreLink]: ...
async def find_one(self, document_id: str, store_id: str) -> DocumentStoreLink | None: ...
async def delete(self, document_id: str, store_id: str) -> bool: ...
class AnalysisRepository(Protocol):
"""Port for analysis job persistence."""

22
document-parser/domain/value_objects.py
View file

@ -39,6 +39,28 @@ class DocumentLifecycleState(StrEnum):
FAILED = "Failed"
class StoreKind(StrEnum):
"""Backing technology of a Store. Today only OpenSearch is implemented;
the enum is here so future backends (Pinecone, Qdrant, pgvector) can be
added without touching the persistence schema."""
OPENSEARCH = "opensearch"
class DocumentStoreLinkState(StrEnum):
"""State of a (document, store) ingestion link.
Distinct from `DocumentLifecycleState` — the document lifecycle is the
aggregate over all per-store links. A link is `Ingested` when its
chunkset hash matches the source; `Stale` when the source has drifted
after the last push; `Failed` when the last push attempt errored.
"""
INGESTED = "Ingested"
STALE = "Stale"
FAILED = "Failed"
@dataclass(frozen=True)
class PageElement:
type: str

45
document-parser/persistence/database.py
View file

@ -42,6 +42,34 @@ CREATE TABLE IF NOT EXISTS analysis_jobs (
CREATE INDEX IF NOT EXISTS idx_analysis_jobs_status ON analysis_jobs(status);
CREATE INDEX IF NOT EXISTS idx_analysis_jobs_created_at ON analysis_jobs(created_at);
CREATE INDEX IF NOT EXISTS idx_documents_created_at ON documents(created_at);
-- 0.6.0 — Per (document, store) ingestion state (#203).
CREATE TABLE IF NOT EXISTS stores (
id TEXT PRIMARY KEY,
name TEXT NOT NULL UNIQUE,
slug TEXT NOT NULL UNIQUE,
kind TEXT NOT NULL,
embedder TEXT NOT NULL,
config TEXT NOT NULL DEFAULT '{}',
is_default INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS document_store_links (
id TEXT PRIMARY KEY,
document_id TEXT NOT NULL REFERENCES documents(id) ON DELETE CASCADE,
store_id TEXT NOT NULL REFERENCES stores(id) ON DELETE CASCADE,
state TEXT NOT NULL,
chunkset_hash TEXT,
last_push_at TEXT,
last_run_id TEXT,
error_message TEXT,
UNIQUE (document_id, store_id)
);
CREATE INDEX IF NOT EXISTS idx_dsl_doc ON document_store_links(document_id);
CREATE INDEX IF NOT EXISTS idx_dsl_store ON document_store_links(store_id);
CREATE INDEX IF NOT EXISTS idx_dsl_state ON document_store_links(state);
"""
@ -107,10 +135,27 @@ async def init_db() -> None:
async with aiosqlite.connect(DB_PATH) as db:
await db.executescript(_SCHEMA)
await _run_migrations(db)
await _seed_default_store(db)
await db.commit()
logger.info("Database initialized at %s", DB_PATH)
async def _seed_default_store(db: aiosqlite.Connection) -> None:
"""Insert the canonical `default` store on first boot.
Idempotent — uses INSERT OR IGNORE keyed on the unique slug. The
embedder is read from the DEFAULT_EMBEDDER env var with a sensible
fallback so existing single-index deployments keep working.
"""
embedder = os.environ.get("DEFAULT_EMBEDDER", "bge-m3")
await db.execute(
"""INSERT OR IGNORE INTO stores
(id, name, slug, kind, embedder, config, is_default, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, datetime('now'))""",
("default", "default", "default", "opensearch", embedder, "{}", 1),
)
async def get_db() -> aiosqlite.Connection:
"""Open a new database connection with row factory and FK enforcement."""
db = await aiosqlite.connect(DB_PATH)

98
document-parser/persistence/document_store_link_repo.py Normal file
View file

@ -0,0 +1,98 @@
"""Document-Store link repository — SQLite CRUD on `document_store_links`."""
from __future__ import annotations
from datetime import UTC, datetime
from domain.models import DocumentStoreLink
from domain.value_objects import DocumentStoreLinkState
from persistence.database import get_connection
def _parse_iso(value: str | None) -> datetime | None:
if value is None or value == "":
return None
parsed = datetime.fromisoformat(value)
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=UTC)
return parsed
def _row_to_link(row) -> DocumentStoreLink:
return DocumentStoreLink(
id=row["id"],
document_id=row["document_id"],
store_id=row["store_id"],
state=DocumentStoreLinkState(row["state"]),
chunkset_hash=row["chunkset_hash"],
last_push_at=_parse_iso(row["last_push_at"]),
last_run_id=row["last_run_id"],
error_message=row["error_message"],
)
class SqliteDocumentStoreLinkRepository:
"""SQLite implementation of the DocumentStoreLinkRepository port."""
async def upsert(self, link: DocumentStoreLink) -> None:
"""Insert a new link or update the existing (document_id, store_id) row."""
async with get_connection() as db:
await db.execute(
"""INSERT INTO document_store_links
(id, document_id, store_id, state, chunkset_hash,
last_push_at, last_run_id, error_message)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(document_id, store_id) DO UPDATE SET
state = excluded.state,
chunkset_hash = excluded.chunkset_hash,
last_push_at = excluded.last_push_at,
last_run_id = excluded.last_run_id,
error_message = excluded.error_message""",
(
link.id,
link.document_id,
link.store_id,
link.state.value,
link.chunkset_hash,
str(link.last_push_at) if link.last_push_at else None,
link.last_run_id,
link.error_message,
),
)
await db.commit()
async def find_for_document(self, document_id: str) -> list[DocumentStoreLink]:
async with get_connection() as db:
cursor = await db.execute(
"SELECT * FROM document_store_links WHERE document_id = ?",
(document_id,),
)
rows = await cursor.fetchall()
return [_row_to_link(r) for r in rows]
async def find_for_store(self, store_id: str) -> list[DocumentStoreLink]:
async with get_connection() as db:
cursor = await db.execute(
"SELECT * FROM document_store_links WHERE store_id = ?",
(store_id,),
)
rows = await cursor.fetchall()
return [_row_to_link(r) for r in rows]
async def find_one(self, document_id: str, store_id: str) -> DocumentStoreLink | None:
async with get_connection() as db:
cursor = await db.execute(
"SELECT * FROM document_store_links WHERE document_id = ? AND store_id = ?",
(document_id, store_id),
)
row = await cursor.fetchone()
return _row_to_link(row) if row else None
async def delete(self, document_id: str, store_id: str) -> bool:
async with get_connection() as db:
cursor = await db.execute(
"DELETE FROM document_store_links WHERE document_id = ? AND store_id = ?",
(document_id, store_id),
)
await db.commit()
return cursor.rowcount > 0

83
document-parser/persistence/store_repo.py Normal file
View file

@ -0,0 +1,83 @@
"""Store repository — SQLite CRUD for the `stores` table."""
from __future__ import annotations
import json
from datetime import UTC, datetime
from domain.models import Store
from domain.value_objects import StoreKind
from persistence.database import get_connection
def _row_to_store(row) -> Store:
created = row["created_at"]
if isinstance(created, str):
created = datetime.fromisoformat(created)
if created.tzinfo is None:
created = created.replace(tzinfo=UTC)
config_raw = row["config"] or "{}"
try:
config = json.loads(config_raw)
except (TypeError, ValueError):
config = {}
return Store(
id=row["id"],
name=row["name"],
slug=row["slug"],
kind=StoreKind(row["kind"]),
embedder=row["embedder"],
config=config,
is_default=bool(row["is_default"]),
created_at=created,
)
class SqliteStoreRepository:
"""SQLite implementation of the StoreRepository port."""
async def insert(self, store: Store) -> None:
async with get_connection() as db:
await db.execute(
"""INSERT INTO stores
(id, name, slug, kind, embedder, config, is_default, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
(
store.id,
store.name,
store.slug,
store.kind.value,
store.embedder,
json.dumps(store.config),
1 if store.is_default else 0,
str(store.created_at),
),
)
await db.commit()
async def find_all(self) -> list[Store]:
async with get_connection() as db:
cursor = await db.execute("SELECT * FROM stores ORDER BY is_default DESC, name ASC")
rows = await cursor.fetchall()
return [_row_to_store(r) for r in rows]
async def find_by_slug(self, slug: str) -> Store | None:
async with get_connection() as db:
cursor = await db.execute("SELECT * FROM stores WHERE slug = ?", (slug,))
row = await cursor.fetchone()
return _row_to_store(row) if row else None
async def find_by_id(self, store_id: str) -> Store | None:
async with get_connection() as db:
cursor = await db.execute("SELECT * FROM stores WHERE id = ?", (store_id,))
row = await cursor.fetchone()
return _row_to_store(row) if row else None
async def get_default(self) -> Store | None:
"""Return the seeded `default` store, if any."""
async with get_connection() as db:
cursor = await db.execute(
"SELECT * FROM stores WHERE is_default = 1 ORDER BY created_at ASC LIMIT 1"
)
row = await cursor.fetchone()
return _row_to_store(row) if row else None

72
document-parser/tests/test_lifecycle_aggregation.py Normal file
View file

@ -0,0 +1,72 @@
"""Tests for the document lifecycle aggregation rule (#203)."""
from __future__ import annotations
from domain.lifecycle_aggregation import aggregate_lifecycle
from domain.models import DocumentStoreLink
from domain.value_objects import DocumentLifecycleState, DocumentStoreLinkState
def _link(state: DocumentStoreLinkState) -> DocumentStoreLink:
return DocumentStoreLink(
id=f"link-{state.value}",
document_id="doc-1",
store_id=f"store-{state.value}",
state=state,
)
def test_no_links_returns_fallback() -> None:
assert (
aggregate_lifecycle([], fallback=DocumentLifecycleState.CHUNKED)
== DocumentLifecycleState.CHUNKED
)
def test_single_ingested_link_makes_doc_ingested() -> None:
assert (
aggregate_lifecycle(
[_link(DocumentStoreLinkState.INGESTED)],
fallback=DocumentLifecycleState.CHUNKED,
)
== DocumentLifecycleState.INGESTED
)
def test_any_stale_link_makes_doc_stale() -> None:
"""Stale outranks Ingested — a doc with one stale store is considered stale."""
assert (
aggregate_lifecycle(
[
_link(DocumentStoreLinkState.INGESTED),
_link(DocumentStoreLinkState.STALE),
],
fallback=DocumentLifecycleState.CHUNKED,
)
== DocumentLifecycleState.STALE
)
def test_any_failed_link_makes_doc_failed() -> None:
"""Failed outranks every other link state."""
assert (
aggregate_lifecycle(
[
_link(DocumentStoreLinkState.INGESTED),
_link(DocumentStoreLinkState.STALE),
_link(DocumentStoreLinkState.FAILED),
],
fallback=DocumentLifecycleState.CHUNKED,
)
== DocumentLifecycleState.FAILED
)
def test_only_failed_link_makes_doc_failed() -> None:
assert (
aggregate_lifecycle(
[_link(DocumentStoreLinkState.FAILED)],
fallback=DocumentLifecycleState.CHUNKED,
)
== DocumentLifecycleState.FAILED
)

190
document-parser/tests/test_store_repo.py Normal file
View file

@ -0,0 +1,190 @@
"""Tests for the Store and DocumentStoreLink repositories (#203)."""
from __future__ import annotations
from datetime import UTC, datetime
import pytest
from domain.models import Document, DocumentStoreLink, Store
from domain.value_objects import DocumentStoreLinkState, StoreKind
from persistence.database import init_db
from persistence.document_repo import SqliteDocumentRepository
from persistence.document_store_link_repo import SqliteDocumentStoreLinkRepository
from persistence.store_repo import SqliteStoreRepository
@pytest.fixture(autouse=True)
async def setup_db(monkeypatch, tmp_path):
db_path = str(tmp_path / "test.db")
monkeypatch.setattr("persistence.database.DB_PATH", db_path)
await init_db()
yield
@pytest.fixture
def store_repo():
return SqliteStoreRepository()
@pytest.fixture
def link_repo():
return SqliteDocumentStoreLinkRepository()
@pytest.fixture
def document_repo():
return SqliteDocumentRepository()
class TestStoreRepo:
async def test_default_store_seeded_on_init(self, store_repo):
"""init_db() must seed exactly one `default` store on a fresh DB."""
default = await store_repo.get_default()
assert default is not None
assert default.slug == "default"
assert default.is_default is True
assert default.kind == StoreKind.OPENSEARCH
async def test_seed_is_idempotent(self, store_repo):
"""Re-running init_db() must not create a second default store."""
await init_db()
await init_db()
all_stores = await store_repo.find_all()
slugs = [s.slug for s in all_stores]
assert slugs.count("default") == 1
async def test_insert_and_find_by_slug(self, store_repo):
store = Store(
id="s-rh",
name="rh-corpus-v3",
slug="rh-corpus-v3",
kind=StoreKind.OPENSEARCH,
embedder="bge-m3",
config={"index_name": "rh-corpus-v3"},
)
await store_repo.insert(store)
found = await store_repo.find_by_slug("rh-corpus-v3")
assert found is not None
assert found.embedder == "bge-m3"
assert found.config == {"index_name": "rh-corpus-v3"}
async def test_find_all_orders_default_first(self, store_repo):
await store_repo.insert(
Store(
id="s-rh",
name="rh",
slug="rh",
kind=StoreKind.OPENSEARCH,
embedder="bge-m3",
)
)
await store_repo.insert(
Store(
id="s-legal",
name="legal",
slug="legal",
kind=StoreKind.OPENSEARCH,
embedder="bge-m3",
)
)
all_stores = await store_repo.find_all()
assert all_stores[0].slug == "default" # seeded default comes first
class TestDocumentStoreLinkRepo:
async def _seed_doc(self, document_repo, doc_id: str = "doc-1") -> Document:
doc = Document(id=doc_id, filename="t.pdf", storage_path="/tmp/t.pdf")
await document_repo.insert(doc)
return doc
async def test_upsert_creates_then_updates(self, link_repo, document_repo):
await self._seed_doc(document_repo)
link = DocumentStoreLink(
id="l-1",
document_id="doc-1",
store_id="default",
state=DocumentStoreLinkState.INGESTED,
)
await link_repo.upsert(link)
# Update same (doc, store) — no second row.
link.mark_ingested(
hash_="abc123",
at=datetime(2026, 4, 29, 12, tzinfo=UTC),
run_id="run-1",
)
await link_repo.upsert(link)
all_for_doc = await link_repo.find_for_document("doc-1")
assert len(all_for_doc) == 1
assert all_for_doc[0].chunkset_hash == "abc123"
assert all_for_doc[0].last_run_id == "run-1"
async def test_unique_constraint_on_doc_store_pair(self, link_repo, document_repo):
"""Two links with the same (doc, store) collapse to one via upsert."""
await self._seed_doc(document_repo)
await link_repo.upsert(
DocumentStoreLink(
id="l-1",
document_id="doc-1",
store_id="default",
state=DocumentStoreLinkState.INGESTED,
)
)
await link_repo.upsert(
DocumentStoreLink(
id="l-2",
document_id="doc-1",
store_id="default",
state=DocumentStoreLinkState.STALE,
)
)
rows = await link_repo.find_for_document("doc-1")
assert len(rows) == 1
assert rows[0].state == DocumentStoreLinkState.STALE
async def test_find_one_returns_none_when_absent(self, link_repo, document_repo):
await self._seed_doc(document_repo)
assert await link_repo.find_one("doc-1", "default") is None
async def test_cascade_delete_when_doc_removed(self, link_repo, document_repo):
await self._seed_doc(document_repo)
await link_repo.upsert(
DocumentStoreLink(
id="l-1",
document_id="doc-1",
store_id="default",
state=DocumentStoreLinkState.INGESTED,
)
)
await document_repo.delete("doc-1")
rows = await link_repo.find_for_document("doc-1")
assert rows == []
async def test_state_round_trips(self, link_repo, document_repo):
await self._seed_doc(document_repo)
for state in DocumentStoreLinkState:
link = DocumentStoreLink(
id=f"l-{state.value}",
document_id="doc-1",
store_id=f"store-{state.value}",
state=state,
)
# Need a store row for FK; seed minimal stores.
from persistence.store_repo import SqliteStoreRepository
await SqliteStoreRepository().insert(
Store(
id=f"store-{state.value}",
name=f"s-{state.value}",
slug=f"s-{state.value}",
kind=StoreKind.OPENSEARCH,
embedder="bge-m3",
)
)
await link_repo.upsert(link)
found = await link_repo.find_one("doc-1", f"store-{state.value}")
assert found is not None
assert found.state == state