From 2cd97880fd4ebef5b1ab30dcb1473fbe18297f21 Mon Sep 17 00:00:00 2001 From: Yiorgis Gozadinos Date: Mon, 1 Jun 2026 18:25:15 +0300 Subject: [PATCH] Replace sync_state batch 5-tuple with a SyncRow NamedTuple --- .../haiku/rag/ingester/pollers/base.py | 10 +++---- .../haiku/rag/ingester/queue/models.py | 12 ++++++++ .../haiku/rag/ingester/queue/repository.py | 16 +++++----- tests/ingester/test_queue.py | 30 +++++++++++-------- 4 files changed, 44 insertions(+), 24 deletions(-) diff --git a/haiku_rag_slim/haiku/rag/ingester/pollers/base.py b/haiku_rag_slim/haiku/rag/ingester/pollers/base.py index b4e6b8b0..093a8cd0 100644 --- a/haiku_rag_slim/haiku/rag/ingester/pollers/base.py +++ b/haiku_rag_slim/haiku/rag/ingester/pollers/base.py @@ -5,7 +5,7 @@ from datetime import UTC, datetime from haiku.rag.config import SourceConfig from haiku.rag.ingester.pollers.circuit_breaker import CircuitBreaker -from haiku.rag.ingester.queue.models import JobOp +from haiku.rag.ingester.queue.models import JobOp, SyncRow from haiku.rag.ingester.queue.repository import JobRepo, SyncStateRepo from haiku.rag.ingester.sources.base import ( Source, @@ -129,7 +129,7 @@ class BasePoller: SourceEventKind.DELETE: 0, SourceEventKind.UNCHANGED: 0, } - sync_batch: list[tuple[str, str, str | None, str | None, bool]] = [] + sync_batch: list[SyncRow] = [] async for event in self.source.discover( since=revisions, known_uris=known ): @@ -168,7 +168,7 @@ class BasePoller: async def _handle_event( self, event: SourceEvent, - sync_batch: list[tuple[str, str, str | None, str | None, bool]], + sync_batch: list[SyncRow], ) -> None: if event.kind is SourceEventKind.UPSERT: await self._jobs.enqueue( @@ -182,11 +182,11 @@ class BasePoller: # Don't write revision to sync_state here — the worker writes it # after a successful ingestion. last_seen_at gets bumped to keep # orphan detection accurate. - sync_batch.append((event.source_id, event.uri, None, None, False)) + sync_batch.append(SyncRow(event.source_id, event.uri, None, None, False)) elif event.kind is SourceEventKind.UNCHANGED: # Touch last_seen_at without changing the stored revision. sync_batch.append( - (event.source_id, event.uri, event.revision, None, False) + SyncRow(event.source_id, event.uri, event.revision, None, False) ) elif event.kind is SourceEventKind.DELETE: if not self.config.delete_orphans: diff --git a/haiku_rag_slim/haiku/rag/ingester/queue/models.py b/haiku_rag_slim/haiku/rag/ingester/queue/models.py index aa4e7765..ae5df4d3 100644 --- a/haiku_rag_slim/haiku/rag/ingester/queue/models.py +++ b/haiku_rag_slim/haiku/rag/ingester/queue/models.py @@ -1,5 +1,6 @@ from datetime import datetime from enum import StrEnum +from typing import NamedTuple from pydantic import BaseModel @@ -44,3 +45,14 @@ class SyncStateRow(BaseModel): content_hash: str | None = None last_seen_at: datetime last_ingested_at: datetime | None = None + + +class SyncRow(NamedTuple): + """A pending sync_state write collected during a sweep and flushed in one + transaction by SyncStateRepo.batch_upsert().""" + + source_id: str + uri: str + revision: str | None + content_hash: str | None + ingested: bool diff --git a/haiku_rag_slim/haiku/rag/ingester/queue/repository.py b/haiku_rag_slim/haiku/rag/ingester/queue/repository.py index 60070f46..3cbdac4f 100644 --- a/haiku_rag_slim/haiku/rag/ingester/queue/repository.py +++ b/haiku_rag_slim/haiku/rag/ingester/queue/repository.py @@ -5,7 +5,13 @@ from datetime import UTC, datetime, timedelta import aiosqlite -from haiku.rag.ingester.queue.models import Job, JobOp, JobStatus, SyncStateRow +from haiku.rag.ingester.queue.models import ( + Job, + JobOp, + JobStatus, + SyncRow, + SyncStateRow, +) def _utcnow_iso() -> str: @@ -504,12 +510,8 @@ class SyncStateRepo: pass await self._conn.commit() - async def batch_upsert( - self, - rows: list[tuple[str, str, str | None, str | None, bool]], - ) -> None: - """Batch insert-or-update sync_state rows in a single transaction. - Each tuple is (source_id, uri, revision, content_hash, ingested).""" + async def batch_upsert(self, rows: list[SyncRow]) -> None: + """Batch insert-or-update sync_state rows in a single transaction.""" if not rows: return now = _utcnow_iso() diff --git a/tests/ingester/test_queue.py b/tests/ingester/test_queue.py index 1577d805..db376406 100644 --- a/tests/ingester/test_queue.py +++ b/tests/ingester/test_queue.py @@ -5,7 +5,7 @@ import aiosqlite import pytest from haiku.rag.ingester.queue.migrations import apply_migrations, open_queue -from haiku.rag.ingester.queue.models import JobOp, JobStatus +from haiku.rag.ingester.queue.models import JobOp, JobStatus, SyncRow from haiku.rag.ingester.queue.repository import JobRepo, SyncStateRepo @@ -839,11 +839,13 @@ async def test_sync_state_upsert_replaces_revision_when_provided(sync): @pytest.mark.asyncio async def test_sync_state_batch_upsert_inserts_multiple_rows(sync): """batch_upsert writes many rows in a single transaction.""" - await sync.batch_upsert([ - ("s", "u1", "rev1", "hash1", False), - ("s", "u2", "rev2", "hash2", False), - ("s", "u3", "rev3", None, True), - ]) + await sync.batch_upsert( + [ + SyncRow("s", "u1", "rev1", "hash1", False), + SyncRow("s", "u2", "rev2", "hash2", False), + SyncRow("s", "u3", "rev3", None, True), + ] + ) assert await sync.get_revision_snapshot("s") == { "u1": "rev1", "u2": "rev2", @@ -858,9 +860,11 @@ async def test_sync_state_batch_upsert_inserts_multiple_rows(sync): async def test_sync_state_batch_upsert_updates_existing(sync): """batch_upsert applies ON CONFLICT update semantics like upsert().""" await sync.upsert("s", "u1", revision="old", content_hash="old-hash") - await sync.batch_upsert([ - ("s", "u1", "new", "new-hash", False), - ]) + await sync.batch_upsert( + [ + SyncRow("s", "u1", "new", "new-hash", False), + ] + ) row = await sync.get_row("s", "u1") assert row is not None assert row.revision == "new" @@ -871,9 +875,11 @@ async def test_sync_state_batch_upsert_updates_existing(sync): async def test_sync_state_batch_upsert_preserves_revision_when_none(sync): """batch_upsert with revision=None leaves existing revision in place.""" await sync.upsert("s", "u1", revision="keep", content_hash="keep-hash") - await sync.batch_upsert([ - ("s", "u1", None, None, False), - ]) + await sync.batch_upsert( + [ + SyncRow("s", "u1", None, None, False), + ] + ) row = await sync.get_row("s", "u1") assert row is not None assert row.revision == "keep"