Coverage for src/lilbee/data/ingest/pipeline.py: 100%
719 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-14 11:46 +0000
1"""Top-level sync orchestration: discovery, dispatch, batching, post-sync hooks."""
3from __future__ import annotations
5import asyncio
6import contextlib
7import logging
8import os
9import threading
10import time
11from collections import deque
12from collections.abc import AsyncGenerator, Callable, Coroutine, Iterable, Iterator, Mapping
13from concurrent.futures import Future, ThreadPoolExecutor, as_completed
14from dataclasses import dataclass, field
15from itertools import count
16from pathlib import Path
17from typing import Any, cast
19from rich.progress import (
20 BarColumn,
21 MofNCompleteColumn,
22 Progress,
23 SpinnerColumn,
24 TextColumn,
25 TimeElapsedColumn,
26)
28from lilbee.app.services import get_services
29from lilbee.core.config import Config, active_config
30from lilbee.data.extract.document import (
31 extract_batching,
32 ingest_document,
33 ingest_markdown,
34 warn_if_table_model_ignored,
35)
36from lilbee.data.extract.trace import configure_from_env as configure_trace_from_env
37from lilbee.data.ingest.adaptive import (
38 AdaptiveController,
39 ResizableGate,
40 enumerate_fleet_devices,
41 make_signal_sampler,
42 profile_for,
43 resolve_mode,
44)
45from lilbee.data.ingest.code import ingest_code_sync
46from lilbee.data.ingest.discovery import classify_file, discover_files, file_hash
47from lilbee.data.ingest.errors import error_reason
48from lilbee.data.ingest.fanout import (
49 WORKER_LOG_NAME,
50 ShardDone,
51 ShardOptions,
52 ShardSpec,
53 aggregate_results,
54 plan_fanout,
55 run_workers,
56)
57from lilbee.data.ingest.skip_marker import (
58 clear_skip_markers,
59 load_skip_markers,
60 load_skip_reasons,
61 write_skip_markers,
62 write_skip_reasons,
63)
64from lilbee.data.offload import (
65 embed_inflight_target,
66 max_workers,
67 to_executor,
68 to_ingest_thread,
69)
70from lilbee.data.store import (
71 SOURCE_STAT_UNKNOWN,
72 ChunkWrite,
73 ConceptRecords,
74 PageTextRecord,
75 SourceMeta,
76 SourceRecord,
77 SourceStat,
78 SourceStatBackfill,
79 SourceType,
80 Store,
81 source_stat,
82)
83from lilbee.data.title import derive_title
84from lilbee.data.types import (
85 ChunkRecord,
86 FileChangePlan,
87 FileToProcess,
88 ShardId,
89 SyncResult,
90 _IngestResult,
91)
92from lilbee.runtime.asyncio_loop import is_executor_shutdown
93from lilbee.runtime.cancellation import CancelSignal, TaskCancelledError
94from lilbee.runtime.cpu import available_cpu_count, cpu_quota
95from lilbee.runtime.lock import LockTimeoutError
96from lilbee.runtime.progress import (
97 BatchProgressEvent,
98 BatchStatus,
99 DetailedProgressCallback,
100 EmbedEvent,
101 EventType,
102 ExtractEvent,
103 FileDoneEvent,
104 FileStartEvent,
105 ProgressEvent,
106 SyncDoneEvent,
107 noop_callback,
108)
110log = logging.getLogger(__name__)
113def _max_concurrent() -> int:
114 """Files allowed in their compute phase at once.
116 ``cpu_quota()`` (cpu_count // 2) keeps worker storms from starving the TUI's asyncio
117 main thread, and is the cap for text/code ingest. Vision OCR is different: every file
118 in compute holds a continuous-batching slot on a vision server, so an OCR run is bounded
119 by the vision slot capacity. Once the fleet is up that capacity is the servers' real
120 fitted ``--parallel`` slots (a memory-constrained card fits fewer than requested); until
121 then it is estimated from ``replicas x per-server pages``. Sizing to real capacity keeps
122 the OCR queue shallow instead of piling pages behind the dispatcher with their deadlines
123 ticking, and still keeps every slot fed.
124 """
125 from lilbee.providers.fleet.replicas import gpu_device_count, resolve_replica_count
126 from lilbee.providers.roles import WorkerRole
128 config = active_config()
129 if config.vision_model:
130 fitted = get_services().provider.vision_slot_capacity()
131 if fitted is not None:
132 return fitted
133 replicas = resolve_replica_count(WorkerRole.VISION, gpu_device_count())
134 return max(1, replicas * config.vision_ocr_concurrency)
135 if config.ingest_max_inflight > 0:
136 return config.ingest_max_inflight # explicit override
137 # Auto: keep every embed replica fed. The CPU-bound quota alone leaves a
138 # many-core multi-GPU box starved (~4 files/card), so scale admission with
139 # the detected fleet size -- no manual cap needed.
140 return max(cpu_quota(), embed_inflight_target())
143async def _rebuild_concept_clusters() -> None:
144 """Re-run Leiden clustering after sync. No-op if disabled."""
145 if not active_config().concept_graph:
146 return
147 from lilbee.retrieval.concepts import concepts_available
149 if not concepts_available():
150 return
151 try:
152 cg = get_services().concepts
153 if not cg.get_graph():
154 return
155 await to_ingest_thread(cg.rebuild_clusters)
156 except Exception:
157 log.warning("Concept cluster rebuild failed", exc_info=True)
160async def build_entity_records(records: list[ChunkRecord], source_name: str) -> list[dict] | None:
161 """Extract typed entities for ingested chunks. None when the mode is off.
163 Gated twice: the ``entity_extraction`` config flag, and a schema already
164 induced into the index; absent either, syncs cost nothing. Extraction
165 failures degrade to no rows for the file, mirroring concept extraction.
166 """
167 config = active_config()
168 if not config.entity_extraction or not records:
169 return None
170 from lilbee.retrieval.entities import ExtractorKind, extract_entities, load_schema
172 schema = load_schema(get_services().store)
173 if schema is None:
174 return None
175 nlp = None
176 if any(t.kind is ExtractorKind.SPACY for t in schema.types):
177 from lilbee.retrieval.concepts import concepts_available
178 from lilbee.retrieval.concepts.nlp import load_spacy_pipeline
180 if concepts_available():
181 try:
182 nlp = load_spacy_pipeline()
183 except ImportError:
184 log.warning("spaCy model unavailable; spacy-kind entity types skipped")
185 provider = None
186 if any(t.kind is ExtractorKind.LLM for t in schema.types):
187 provider = get_services().provider
188 try:
189 return await to_ingest_thread(
190 extract_entities,
191 cast("list[Mapping[str, Any]]", records),
192 schema,
193 provider=provider,
194 nlp=nlp,
195 )
196 except Exception:
197 log.warning("Entity extraction failed for %s", source_name, exc_info=True)
198 return None
201async def build_concept_records(
202 records: list[ChunkRecord], source_name: str
203) -> ConceptRecords | None:
204 """Extract concepts for ingested chunks and build their table rows. None if disabled.
206 Pure record building, no store access: the rows are buffered on the file's
207 ingest result and written once per flush (see :func:`_flush_concept_records`),
208 so a large sync pays one concept-table write per flush, not per file.
209 """
210 if not active_config().concept_graph or not records:
211 return None
212 from lilbee.retrieval.concepts import concepts_available
214 if not concepts_available():
215 return None
216 try:
217 cg = get_services().concepts
218 texts = [r["chunk"] for r in records]
219 concept_lists = await to_ingest_thread(cg.extract_concepts_batch, texts)
220 chunk_ids = [(source_name, r["chunk_index"]) for r in records]
221 return await to_ingest_thread(cg.build_concept_records, chunk_ids, concept_lists)
222 except Exception:
223 log.warning("Concept extraction failed for %s", source_name, exc_info=True)
224 return None
227async def produce_records(
228 path: Path,
229 source_name: str,
230 content_type: str,
231 *,
232 quiet: bool = False,
233 on_progress: DetailedProgressCallback = noop_callback,
234 page_texts_out: list[PageTextRecord] | None = None,
235) -> tuple[list[ChunkRecord], SourceMeta]:
236 """Extract, chunk, and embed a single file into (records, source metadata).
238 The LanceDB write is deferred: records are returned to the caller and written
239 in a batched flush (see :func:`_flush_writes`), so bulk ingest pays one
240 write-lock acquisition per batch instead of one per file. The per-page text
241 dataset rows land in ``page_texts_out`` and are written by the same flush.
242 The returned metadata (extraction-provided when available, stem-derived title
243 otherwise) stamps every record's ``title`` and updates the source row.
244 """
245 records: list[ChunkRecord]
246 page_texts: list[PageTextRecord] = page_texts_out if page_texts_out is not None else []
247 if content_type == "code":
248 records = await to_ingest_thread(ingest_code_sync, path, source_name, on_progress)
249 meta = SourceMeta(title=derive_title(source_name))
250 elif path.suffix.lower() == ".md":
251 records, meta = await ingest_markdown(
252 path, source_name, on_progress, page_texts_out=page_texts
253 )
254 else:
255 records, meta = await ingest_document(
256 path,
257 source_name,
258 content_type,
259 quiet=quiet,
260 on_progress=on_progress,
261 page_texts_out=page_texts,
262 )
264 for record in records:
265 # NULL (not "") for an absent title, so chunk rows match the migration
266 # and the _sources table, which both persist absence as NULL.
267 record["title"] = meta.title or None
268 return records, meta
271def _disk_stat(path: Path) -> SourceStat | None:
272 """Current size/mtime of *path* stamped with now, or None when it cannot be stat'd."""
273 try:
274 st = path.stat()
275 except OSError:
276 return None
277 return SourceStat(st.st_size, st.st_mtime_ns, time.time_ns())
280def _stat_unchanged(stored: SourceStat, current: SourceStat) -> bool:
281 """Whether the stored stat proves the file unchanged without hashing it.
283 Git-style racily-clean guard: a matching (size, mtime) only counts when the
284 mtime is strictly older than the time the stat was recorded; a same-size
285 edit landing in the same mtime tick is otherwise missed forever.
286 """
287 if (stored.size_bytes, stored.mtime_ns) != (current.size_bytes, current.mtime_ns):
288 return False
289 # Unknown capture or mtime >= capture hashes anyway; clock skew can only widen
290 # hashing, never widen skipping past the pre-existing same-tick window.
291 return stored.captured_ns != SOURCE_STAT_UNKNOWN and current.mtime_ns < stored.captured_ns
294@dataclass(frozen=True)
295class _FileChangeVerdict:
296 """One file's sync verdict: process it, or unchanged (optionally backfilling its stat)."""
298 to_process: FileToProcess | None = None
299 backfill: SourceStatBackfill | None = None
300 is_update: bool = False
303def _classify_file_change(
304 name: str,
305 path: Path,
306 record: SourceRecord | None,
307 skip_markers: dict[str, str],
308) -> _FileChangeVerdict:
309 """Decide one file's verdict: stat-unchanged, hash-unchanged, skip-marked, or process."""
310 content_type = classify_file(path)
311 if content_type is None:
312 raise ValueError(f"Unsupported file slipped through discovery: {name}")
313 stored_stat = source_stat(record) if record is not None else None
314 current_stat = _disk_stat(path)
315 if (
316 record is not None
317 and stored_stat is not None
318 and current_stat is not None
319 and _stat_unchanged(stored_stat, current_stat)
320 ):
321 return _FileChangeVerdict()
322 old_hash = record["file_hash"] if record is not None else None
323 current_hash = file_hash(path)
324 if old_hash == current_hash:
325 # Content verified unchanged; persist the stat pair so the next
326 # sync skips the hash entirely.
327 backfill = (
328 SourceStatBackfill(record, current_stat)
329 if record is not None and current_stat is not None
330 else None
331 )
332 return _FileChangeVerdict(backfill=backfill)
333 if skip_markers.get(name) == current_hash:
334 # Failed last sync at this exact hash; skip the retry.
335 return _FileChangeVerdict()
336 # needs_cleanup=True unconditionally: delete_by_source is idempotent,
337 # and this closes the race where a prior ingest wrote chunks but died
338 # before upsert_source, leaving orphaned chunks that would duplicate.
339 return _FileChangeVerdict(
340 to_process=FileToProcess(
341 name, path, content_type, current_hash, needs_cleanup=True, stat=current_stat
342 ),
343 is_update=old_hash is not None,
344 )
347def _plan_workers() -> int:
348 """Worker count for the parallel planning pass: config override, else auto.
350 ``config.ingest_workers`` (also set per run by ``add --max-cpus``) wins when
351 positive; otherwise size to the container-aware CPU budget so a big corpus
352 hashes on every core the pod actually has, not the host's vCPU count.
353 """
354 configured = active_config().ingest_workers
355 return configured if configured > 0 else available_cpu_count()
358# How often the plan pass logs progress. The pass can run for tens of minutes on
359# a multi-million-file corpus while the Rich bar renders nothing without a TTY
360# and stdout is block-buffered when piped; a periodic line (which logging flushes
361# per record) keeps a headless run observable instead of looking hung.
362_PLAN_LOG_INTERVAL_S = 10.0
365class _PlanProgress:
366 """Periodic progress for the plan/hash pass, with rate and ETA.
368 Emitted at warning level, not info: the default LILBEE_LOG_LEVEL is WARNING,
369 so an info line would be filtered before any handler and a headless
370 ``lilbee sync`` would show nothing during the plan pass and still look hung.
371 """
373 def __init__(self, total: int) -> None:
374 self._total = total
375 self._done = 0
376 self._started = time.monotonic()
377 self._last = self._started
379 def tick(self) -> None:
380 self._done += 1
381 now = time.monotonic()
382 if now - self._last < _PLAN_LOG_INTERVAL_S:
383 return
384 self._last = now
385 elapsed = now - self._started
386 rate = self._done / elapsed if elapsed > 0 else 0.0
387 remaining = (self._total - self._done) / rate if rate > 0 else 0.0
388 log.warning(
389 "Planning: examined %d/%d files (%.0f%%, %.0f files/s, ~%.0fs left)",
390 self._done,
391 self._total,
392 100.0 * self._done / self._total,
393 rate,
394 remaining,
395 )
398class _StreamStop:
399 """Stop signal for a streamed plan: the caller's cancel, or the stream closing.
401 Closing the stream has to reach the plan batch in flight, not just the next one.
402 Build-vs-buy: the hashers run in a thread pool, so the flag must be a
403 thread-visible ``threading.Event``; ``anyio.CancelScope`` is async-only.
404 """
406 def __init__(self, cancel: CancelSignal | None) -> None:
407 self._cancel = cancel
408 self._closed = threading.Event()
410 def close(self) -> None:
411 self._closed.set()
413 def is_set(self) -> bool:
414 return self._closed.is_set() or (self._cancel is not None and self._cancel.is_set())
417def _classify_pooled(
418 pool: ThreadPoolExecutor,
419 items: list[tuple[str, Path]],
420 classify: Callable[[str, Path], _FileChangeVerdict],
421 cancel: CancelSignal | None,
422 progress: _PlanProgress,
423) -> dict[str, _FileChangeVerdict]:
424 """Fan *items* across *pool*, returning name -> verdict for what completed."""
425 verdicts: dict[str, _FileChangeVerdict] = {}
426 futures: dict[Future[_FileChangeVerdict], str] = {}
427 for name, path in items:
428 if cancel and cancel.is_set():
429 break
430 futures[pool.submit(classify, name, path)] = name
431 for future in as_completed(futures):
432 if cancel and cancel.is_set():
433 # Drop queued-but-unstarted work; running tasks drain on their own.
434 for pending in futures:
435 pending.cancel()
436 break
437 verdicts[futures[future]] = future.result()
438 progress.tick()
439 return verdicts
442def _classify_changes(
443 items: list[tuple[str, Path]],
444 existing_sources: dict[str, SourceRecord],
445 skip_markers: dict[str, str],
446 cancel: CancelSignal | None,
447 *,
448 progress: _PlanProgress | None = None,
449 pool: ThreadPoolExecutor | None = None,
450) -> dict[str, _FileChangeVerdict]:
451 """Classify each file (stat + hash) by name, fanning across a thread pool.
453 Independent per file and side-effect-free, so pooled classification matches a
454 serial pass; ``hashlib`` releases the GIL during digest, giving real speedup
455 on a large corpus. Returns name -> verdict. A set ``cancel`` stops promptly:
456 submission halts and queued-but-unstarted work is cancelled, so a mid-pass
457 cancel over a huge corpus does not hash every remaining file. A *pool* and
458 *progress* passed in are shared across the batches of a streamed plan, so the
459 ETA covers the whole corpus and the workers are spun up once.
460 """
462 def _classify(name: str, path: Path) -> _FileChangeVerdict:
463 return _classify_file_change(name, path, existing_sources.get(name), skip_markers)
465 total = len(items)
466 progress = progress or _PlanProgress(total)
467 if pool is not None:
468 return _classify_pooled(pool, items, _classify, cancel, progress)
469 workers = _plan_workers()
470 if workers <= 1 or total <= 1:
471 verdicts: dict[str, _FileChangeVerdict] = {}
472 for name, path in items:
473 if cancel and cancel.is_set():
474 break
475 verdicts[name] = _classify(name, path)
476 progress.tick()
477 return verdicts
478 with ThreadPoolExecutor(max_workers=workers, thread_name_prefix="lilbee-plan") as owned:
479 return _classify_pooled(owned, items, _classify, cancel, progress)
482def _plan_file_changes(
483 disk_files: dict[str, Path],
484 existing_sources: dict[str, SourceRecord],
485 cancel: CancelSignal | None,
486 skip_markers: dict[str, str] | None = None,
487) -> FileChangePlan:
488 """Diff every disk file against the store in one pass (see :func:`_plan_items`)."""
489 return _plan_items(sorted(disk_files.items()), existing_sources, cancel, skip_markers or {})
492def _plan_items(
493 items: list[tuple[str, Path]],
494 existing_sources: dict[str, SourceRecord],
495 cancel: CancelSignal | None,
496 skip_markers: dict[str, str],
497 *,
498 progress: _PlanProgress | None = None,
499 pool: ThreadPoolExecutor | None = None,
500) -> FileChangePlan:
501 """Diff *items* (sorted name -> path) against the store, hashing only drifted files.
503 A tracked file whose stored (size, mtime) matches the disk stat, and whose
504 mtime predates the stat capture (see :func:`_stat_unchanged`), is unchanged
505 without reading its bytes; everything else is SHA-256 hashed. A file whose
506 current hash matches a marker in ``skip_markers`` (set by a prior failed
507 attempt) is treated as unchanged so we don't retry every sync. Edit the file
508 or run ``/sync --force-rebuild`` to clear the marker and try again.
510 Classification fans across a thread pool (see :func:`_classify_changes`); the
511 plan is assembled from the results in the original sorted order, so a partial
512 or reordered completion never yields a wrong or reordered plan -- only a
513 shorter one when cancelled mid-pass.
514 """
515 verdicts = _classify_changes(
516 items, existing_sources, skip_markers, cancel, progress=progress, pool=pool
517 )
519 files_to_process: list[FileToProcess] = []
520 added: dict[str, None] = {}
521 updated: dict[str, None] = {}
522 stat_backfills: list[SourceStatBackfill] = []
523 unchanged = 0
524 # Assemble in the original sorted order so a partial (cancelled) or reordered
525 # completion never changes the plan the serial pass would have produced.
526 for name, _path in items:
527 verdict = verdicts.get(name)
528 if verdict is None:
529 continue # cancelled before this file was classified
530 if verdict.to_process is None:
531 unchanged += 1
532 if verdict.backfill is not None:
533 stat_backfills.append(verdict.backfill)
534 continue
535 files_to_process.append(verdict.to_process)
536 if verdict.is_update:
537 updated[name] = None
538 else:
539 added[name] = None
540 return FileChangePlan(files_to_process, added, updated, unchanged, stat_backfills)
543@dataclass(frozen=True)
544class _Move:
545 """One relocated source: its old key, its new key, and the new file's stat."""
547 old: str
548 new: str
549 stat: SourceStat | None
552class _MovePool:
553 """Absent sources indexed by content hash, consumed as moves are paired.
555 Built once per sync and drained across the streamed plan's batches, so a file
556 that moved matches exactly the one old key it would have matched in a
557 single-pass plan however the corpus is sharded.
558 """
560 def __init__(self, absent: list[str], existing_sources: dict[str, SourceRecord]) -> None:
561 by_hash: dict[str, list[str]] = {}
562 for name in absent:
563 record = existing_sources.get(name)
564 if record is not None:
565 by_hash.setdefault(record["file_hash"], []).append(name)
566 for candidates in by_hash.values():
567 candidates.sort()
568 self._by_hash = by_hash
570 def take(self, file_hash: str) -> str | None:
571 """The next absent source with this content hash, or None."""
572 matches = self._by_hash.get(file_hash)
573 return matches.pop(0) if matches else None
576def _detect_moves(
577 files_to_process: list[FileToProcess],
578 added: dict[str, None],
579 pool: _MovePool,
580) -> list[_Move]:
581 """Pair brand-new files with absent sources of the same content hash.
583 Only additions (files with a new name) can be moves; an update keeps its name.
584 When several absent sources share a hash, pairing is deterministic (sorted)
585 and one-to-one, so a duplicated file that moved matches exactly one old key
586 and any leftovers stay indexed under their old key.
587 """
588 moves: list[_Move] = []
589 for entry in files_to_process:
590 if entry.name not in added:
591 continue
592 old = pool.take(entry.file_hash)
593 if old is not None:
594 moves.append(_Move(old, entry.name, entry.stat))
595 return moves
598def _apply_moves(
599 moves: list[_Move],
600 files_to_process: list[FileToProcess],
601 added: dict[str, None],
602) -> tuple[list[FileToProcess], list[str]]:
603 """Fold detected moves out of the add set after they were relocated.
605 Drops each moved file from the ingest list and the added set: its chunks were
606 re-keyed onto the new source name, not rebuilt. Returns the trimmed
607 ``(files_to_process, relocated)``.
608 """
609 moved_new = {m.new for m in moves}
610 for name in moved_new:
611 added.pop(name, None)
612 remaining = [e for e in files_to_process if e.name not in moved_new]
613 return remaining, sorted(moved_new)
616def _absent_sources(sources: list[SourceRecord], disk_files: dict[str, Path]) -> list[str]:
617 """Document sources whose backing file is not on disk this sync.
619 A vanished file is never removed on its own (its chunks stay searchable, a
620 dead path-link discovered at open time); this set exists only to pair a
621 reappeared identical file to its old key in move detection. Imported sources
622 have no backing file, so they are excluded.
623 """
624 return [
625 s["filename"]
626 for s in sources
627 if s["filename"] not in disk_files and s["source_type"] != SourceType.IMPORTED
628 ]
631# A plan batch is only done when its slowest hash is, so the first one is small (work
632# reaches the fleet within a second) and later ones amortize that barrier.
633_PLAN_SHARD_MIN_FILES = 256
634_PLAN_SHARD_MAX_FILES = 8192
637def _plan_batch_bounds(total: int) -> Iterator[tuple[int, int]]:
638 """(start, stop) slices covering *total* files, doubling up to the cap.
640 Build-vs-buy: ``itertools.batched`` is the stock slicer but is 3.12+ (floor is
641 3.11) and fixed-size, so it cannot ramp the batch size.
642 """
643 start = 0
644 size = _PLAN_SHARD_MIN_FILES
645 while start < total:
646 stop = min(start + size, total)
647 yield start, stop
648 start = stop
649 size = min(size * 2, _PLAN_SHARD_MAX_FILES)
652@dataclass
653class _StreamedPlan:
654 """Bookkeeping a streamed plan accumulates across its batches.
656 ``added`` and ``updated`` are the dicts the ingest pass mutates as files land.
657 """
659 added: dict[str, None] = field(default_factory=dict)
660 updated: dict[str, None] = field(default_factory=dict)
661 # Processed files' content hashes, for the skip markers written after the run.
662 pending_hashes: dict[str, str] = field(default_factory=dict)
663 relocated: list[str] = field(default_factory=list)
664 # Old keys of relocated sources. The wiki index is keyed by source name, so
665 # without these a move leaves the old name in it forever: its mentions
666 # double-count and its dead chunk refs occupy the per-subject cap.
667 relocated_from: list[str] = field(default_factory=list)
668 unchanged: int = 0
669 planned: int = 0
670 # Files this pass's slice holds, from the discovery walk. Fixed before the
671 # first batch is planned, so it is what progress is measured against: the
672 # plan's own running totals grow as batches land and cannot say how much of
673 # the corpus is left. Summed across a fan-out's workers it is the corpus.
674 corpus_total: int = 0
676 @property
677 def resolved(self) -> int:
678 """Files the plan disposed of without ingest, as it disposes of them.
680 Unchanged (or skip-marked) files and repointed moves are done as far as
681 the corpus is concerned, and nothing downstream reports them, so an
682 incremental sync would otherwise show a handful of changed files against
683 the whole corpus. Files a cancelled plan never classified are absent from
684 both counts and so are not claimed as done.
685 """
686 return self.unchanged + len(self.relocated)
689async def _absorb_plan_batch(
690 plan: FileChangePlan, state: _StreamedPlan, moves: _MovePool
691) -> list[FileToProcess]:
692 """Fold one batch's plan into *state* and return the files it queues for ingest.
694 Relocations and stat backfills are written per batch, so they contend with the
695 batch flushes and take the same one-shot lock retry.
696 """
697 store = get_services().store
698 entries = plan.files_to_process
699 state.unchanged += plan.unchanged
700 state.added.update(plan.added)
701 state.updated.update(plan.updated)
703 detected = _detect_moves(entries, plan.added, moves)
704 if detected:
705 relocations = [(m.old, m.new, m.stat) for m in detected]
706 await to_ingest_thread(
707 _retry_after_lock_timeout, lambda: store.relocate_sources(relocations)
708 )
709 entries, relocated = _apply_moves(detected, entries, plan.added)
710 state.relocated_from.extend(m.old for m in detected)
711 for name in relocated:
712 state.added.pop(name, None)
713 state.relocated.extend(relocated)
714 if plan.stat_backfills:
715 backfills = plan.stat_backfills
716 await to_ingest_thread(
717 _retry_after_lock_timeout, lambda: store.update_source_stats(backfills)
718 )
720 state.pending_hashes.update((entry.name, entry.file_hash) for entry in entries)
721 state.planned += len(entries)
722 return entries
725async def _plan_batches(
726 disk_files: dict[str, Path],
727 existing_sources: dict[str, SourceRecord],
728 skip_markers: dict[str, str],
729 absent: list[str],
730 state: _StreamedPlan,
731 cancel: CancelSignal | None,
732) -> AsyncGenerator[list[FileToProcess]]:
733 """Plan the corpus batch by batch, yielding each batch's files to ingest.
735 Shards are contiguous slices of one sorted item list consumed in order, so
736 this delivers the single-pass plan of :func:`_plan_items` in pieces. The next
737 batch is planned while the current one ingests, on a dedicated thread: the
738 shared ingest pool is saturated by extraction and would stall the stream it
739 feeds. Empty batches are not yielded, so the first yield means there is work.
740 """
741 items = sorted(disk_files.items())
742 moves = _MovePool(absent, existing_sources)
743 progress = _PlanProgress(len(items))
744 stop = _StreamStop(cancel)
745 workers = _plan_workers()
746 hashers = ThreadPoolExecutor(max_workers=workers, thread_name_prefix="lilbee-plan")
747 driver = ThreadPoolExecutor(max_workers=1, thread_name_prefix="lilbee-plan-driver")
749 def _plan(lo: int, hi: int) -> FileChangePlan:
750 return _plan_items(
751 items[lo:hi],
752 existing_sources,
753 stop,
754 skip_markers,
755 progress=progress,
756 pool=hashers if workers > 1 else None,
757 )
759 bounds = list(_plan_batch_bounds(len(items)))
760 try:
761 ahead = asyncio.ensure_future(to_executor(driver, _plan, *bounds[0])) if bounds else None
762 for index, _bound in enumerate(bounds):
763 if ahead is None:
764 break
765 plan = await ahead
766 ahead = (
767 asyncio.ensure_future(to_executor(driver, _plan, *bounds[index + 1]))
768 if index + 1 < len(bounds) and not stop.is_set()
769 else None
770 )
771 entries = await _absorb_plan_batch(plan, state, moves)
772 if entries:
773 yield entries
774 finally:
775 stop.close()
776 if ahead is not None:
777 ahead.cancel()
778 # Retrieve the outcome: a prefetch that had already failed would
779 # otherwise surface as an unretrieved task exception.
780 with contextlib.suppress(Exception, asyncio.CancelledError):
781 await ahead
782 # No wait: the running batch notices the stop and exits on its own, and a
783 # generator being closed must not block the event loop on a hash in flight.
784 hashers.shutdown(wait=False)
785 driver.shutdown(wait=False)
788def detect_pending() -> int:
789 """Count files in documents/ that are out of sync with the store.
791 Cheap operation: filesystem walk + stat-gated SHA-256 hashing + a single
792 sources-table read. No embedding, no writes. Returns the count of files that
793 would be ingested (added + updated), which is what the TaskBar hint surfaces.
794 A vanished file is not pending work: sync leaves it indexed, so it is not
795 counted. Reuses ``_plan_file_changes`` so the diff logic stays single-sourced.
796 Honors skip markers: a file that failed last time at this hash does
797 not show up as pending. Blocking: callers on the event loop run it via
798 ``asyncio.to_thread``.
799 """
800 config = active_config()
801 disk_files = discover_files()
802 if not disk_files:
803 return 0
804 existing_sources = {s["filename"]: s for s in get_services().store.get_sources()}
805 skip_markers = load_skip_markers(config.data_root)
806 plan = _plan_file_changes(disk_files, existing_sources, cancel=None, skip_markers=skip_markers)
807 return len(plan.files_to_process)
810def _load_sync_skip_markers(*, clear_first: bool) -> dict[str, str]:
811 """Read the skip-marker file, optionally clearing it first.
813 Entries are kept whether or not this pass discovered the file. A marker is
814 what holds a removed or unextractable file out of the next sync, so a pass
815 that cannot see the file -- its root is unmounted or moved, or this worker
816 owns only a shard of the corpus -- must not erase the record and re-offer
817 the file the moment it comes back. A marker is dropped when the file
818 ingests cleanly, or by ``retry-skipped`` / ``rebuild``.
819 """
820 data_root = active_config().data_root
821 if clear_first:
822 # Clearing the markers makes the diff re-include the skipped files.
823 clear_skip_markers(data_root)
824 return load_skip_markers(data_root)
827def _persist_skip_markers(
828 markers: dict[str, str],
829 pending_hashes: dict[str, str],
830 *,
831 succeeded: Iterable[str],
832 failed: Iterable[str],
833) -> None:
834 """Mark files that produced no chunks so the next sync skips them, clear the
835 markers for files that ingested cleanly, then write the file back."""
836 for name in succeeded:
837 markers.pop(name, None)
838 for name in failed:
839 fhash = pending_hashes.get(name)
840 if fhash:
841 markers[name] = fhash
842 write_skip_markers(active_config().data_root, markers)
845def _persist_skip_reasons(markers: dict[str, str], reasons: dict[str, str]) -> None:
846 """Write the reasons sidecar so it explains exactly the markers that survive.
848 Merged onto what is already recorded, not replaced: the reasons for files
849 this sync never touched (a removal, an earlier failure) explain markers that
850 are still in force, and dropping them would leave the user with a file held
851 out of every sync and nothing saying why. Reasons whose marker is gone are
852 dropped in the same step, so the two sidecars cannot drift apart.
853 """
854 data_root = active_config().data_root
855 merged = load_skip_reasons(data_root) | reasons
856 write_skip_reasons(data_root, {name: why for name, why in merged.items() if name in markers})
859def _force_rebuild_store(store: Any) -> None:
860 """Drop the store and re-embed the preserved memories table (blocking).
862 Run off the event loop by ``sync``. ``drop_all`` keeps the memories table, so
863 its vectors are refreshed under the (possibly changed) embedding model;
864 a no-op when empty or no embedder.
865 """
866 store.drop_all()
867 embedder = get_services().embedder
868 if embedder.embedding_available():
869 store.rebuild_memory_embeddings(lambda texts: embedder.embed_batch(texts))
872def _reconcile_missing(
873 disk_files: dict[str, Path],
874 sources: list[SourceRecord],
875 failed: Iterable[str],
876 skipped: Iterable[str],
877) -> list[str]:
878 """On-disk document files absent from the store and not reported failed/skipped.
880 A file discovery found and classified that ended up in neither the sources table
881 nor the run's failed/skipped lists was dropped with no signal -- the silent
882 data-loss case (a scanned PDF that never made it into the index yet reported no
883 error). Everything legitimately not indexed is excluded: a failed extraction is in
884 ``failed``, a zero-text file is in ``skipped``, and an unsupported type was never
885 returned by discovery in the first place.
886 """
887 accounted = {s["filename"] for s in sources} | set(failed) | set(skipped)
888 return sorted(name for name in disk_files if name not in accounted)
891def _require_embedding_model() -> None:
892 """Refuse ingest without an embedding model.
894 Ingest has no degraded mode: without one, every chunk fails to embed after
895 the run has already paid the parse and OCR cost. Search and chat fall back
896 to keyword via embedding_available() and carry on.
897 """
898 if not get_services().embedder.validate_model():
899 ref = active_config().embedding_model
900 detail = (
901 f"{ref!r} is not available: pull it, or set a different embedding_model."
902 if ref
903 else "None is configured: pull one and set embedding_model."
904 )
905 raise RuntimeError(f"Ingest needs an embedding model. {detail}")
908async def _run_post_ingest_passes(
909 store: Any,
910 *,
911 indexed_anything: bool,
912 touched: set[str],
913 cancel: CancelSignal | None,
914) -> None:
915 """Index maintenance, concept clusters, the wiki hook, and the entity lifecycle.
917 The index and wiki passes only run when this sync indexed something. The
918 entity lifecycle runs every sync (a cheap no-op when off or already current)
919 so turning the setting on takes effect without a separate operation.
920 """
921 if indexed_anything:
922 store.ensure_fts_index()
923 store.ensure_scalar_indexes()
924 store.ensure_vector_index()
925 store.optimize_sources()
926 await _rebuild_concept_clusters()
927 await _update_wiki(touched, active_config())
929 from lilbee.retrieval.entities.lifecycle import ensure_entities
931 await to_ingest_thread(ensure_entities, cancel)
934async def _update_wiki(changed_sources: set[str], config: Config) -> None:
935 """Refresh the wiki index, and regenerate pages when auto-update is on.
937 The index refresh runs whenever the wiki is enabled, because it spends no
938 LLM call and it is what lets the browse tree list a page the moment its
939 document lands. Regeneration is the expensive half and stays behind
940 wiki_auto_update.
942 Best effort: the ingest itself already succeeded and `lilbee wiki update`
943 re-runs the regeneration, so a wiki failure must not skip the post-ingest
944 entity pass or the reconciliation guard.
945 """
946 if not config.wiki:
947 return
948 # circular: lilbee.wiki imports lilbee.data.ingest.file_hash, so the
949 # post-ingest hook stays function-local at this boundary.
950 from lilbee.wiki.ingest import incremental_update
951 from lilbee.wiki.stubs import refresh_stub_index
953 try:
954 await to_ingest_thread(
955 refresh_stub_index, get_services().store, config, sources=changed_sources
956 )
957 except Exception:
958 log.warning("Wiki index refresh failed after sync", exc_info=True)
960 if not config.wiki_auto_update:
961 return
962 try:
963 await incremental_update(changed_sources, config)
964 except Exception:
965 log.warning("Wiki auto-update failed after sync", exc_info=True)
968def _worker_failure_message(failures: list[ShardDone], specs: list[ShardSpec]) -> str:
969 """What a failed fan-out reports; its workers' shards are kept for the re-run."""
970 detail = "; ".join(
971 f"worker {failure.index} ({specs[failure.index].config.data_root / WORKER_LOG_NAME}): "
972 f"{failure.error}"
973 for failure in failures
974 )
975 return (
976 f"{len(failures)} ingest worker(s) failed, so the index was not updated: {detail}. "
977 "Their work is kept: re-running sync continues from where they stopped."
978 )
981def _merge_worker_shards(store: Store, specs: list[ShardSpec], touched: set[str]) -> None:
982 """Fold every worker's shard into this index.
984 A store with no chunks of its own takes the shards whole; one that already
985 holds a corpus takes only the sources this run touched, so a re-sync replaces
986 those rows instead of appending a second copy of everything.
987 """
988 from lilbee.data.store.shard_merge import merge_shards
990 scope = touched if store.has_chunks() else None
991 merge_shards(store, [spec.config.lancedb_dir for spec in specs], sources=scope)
994async def _sync_across_workers(
995 specs: list[ShardSpec],
996 store: Store,
997 *,
998 options: ShardOptions,
999 quiet: bool,
1000 on_progress: DetailedProgressCallback,
1001 cancel: CancelSignal | None,
1002) -> SyncResult:
1003 """Ingest on one worker per GPU, then fold their shards into the one index."""
1004 verdicts = await run_workers(
1005 specs, options=options, quiet=quiet, on_progress=on_progress, cancel=cancel
1006 )
1007 if cancel is not None and cancel.is_set():
1008 raise asyncio.CancelledError
1009 if failures := [verdict for verdict in verdicts if verdict.error is not None]:
1010 raise RuntimeError(_worker_failure_message(failures, specs))
1011 result = aggregate_results(verdicts)
1012 touched = set(result.added) | set(result.updated) | set(result.relocated)
1013 await to_ingest_thread(_merge_worker_shards, store, specs, touched)
1014 await _run_post_ingest_passes(
1015 store, indexed_anything=bool(touched), touched=touched, cancel=cancel
1016 )
1017 on_progress(
1018 EventType.DONE,
1019 SyncDoneEvent(
1020 added=len(result.added),
1021 updated=len(result.updated),
1022 removed=0,
1023 failed=len(result.failed),
1024 skipped=len(result.skipped),
1025 relocated=len(result.relocated),
1026 ),
1027 )
1028 return result
1031async def sync(
1032 force_rebuild: bool = False,
1033 quiet: bool = False,
1034 *,
1035 on_progress: DetailedProgressCallback = noop_callback,
1036 cancel: CancelSignal | None = None,
1037 retry_skipped: bool = False,
1038 shard: ShardId | None = None,
1039) -> SyncResult:
1040 """Sync documents/ with the vector store.
1041 Returns a SyncResult with the added/updated/removed/unchanged/failed/skipped lists.
1042 When *quiet* is True, the Rich progress bar is suppressed (for JSON output).
1043 When *cancel* is set mid-run, planning and processing stop between files
1044 without data loss (completed work is flushed) and CancelledError is raised;
1045 a cancel already set on entry returns an empty result instead.
1046 When *retry_skipped* (or *force_rebuild*) is set, the failed-file skip
1047 markers are cleared so this sync attempts every file.
1048 A *shard* runs this sync as one worker of a multi-GPU fan-out: it sees only
1049 its slice of the corpus and leaves the corpus-wide passes to the parent.
1050 """
1051 config = active_config()
1052 _store = get_services().store
1054 if force_rebuild:
1055 # drop_all + memory re-embedding are heavy blocking store work; run them
1056 # off the event loop so a rebuild doesn't stall other admitted requests.
1057 await to_ingest_thread(_force_rebuild_store, _store)
1059 config.documents_dir.mkdir(parents=True, exist_ok=True)
1061 if shard is None and (specs := plan_fanout()):
1062 return await _sync_across_workers(
1063 specs,
1064 _store,
1065 options=ShardOptions(
1066 parent_pid=os.getpid(),
1067 force_rebuild=force_rebuild,
1068 retry_skipped=retry_skipped,
1069 ),
1070 quiet=quiet,
1071 on_progress=on_progress,
1072 cancel=cancel,
1073 )
1075 disk_files = discover_files(shard)
1076 sources = _store.get_sources()
1077 existing_sources = {s["filename"]: s for s in sources}
1078 skip_markers = _load_sync_skip_markers(clear_first=force_rebuild or retry_skipped)
1080 failed: dict[str, None] = {}
1081 skipped: dict[str, None] = {}
1082 reasons: dict[str, str] = {} # filename → why it was skipped/failed (for reporting)
1083 flush_failed: set[str] = set()
1085 # Sources whose backing file is not on disk this pass. They are NOT removed:
1086 # a vanished file stays indexed and searchable, a dead path-link the user
1087 # discovers only when they try to open it. The set exists only to pair a
1088 # reappeared identical file to its old key below.
1089 absent = _absent_sources(sources, disk_files)
1091 # The planning pass stats (and where needed hashes) every file on disk, batch by
1092 # batch off the event loop and overlapped with ingest. A brand-new file whose
1093 # content hash matches an absent source is folded in per batch as a move, not an
1094 # add: repointed in place so its chunks and embeddings are reused, not rebuilt.
1095 state = _StreamedPlan(corpus_total=len(disk_files))
1096 added, updated, pending_hashes = state.added, state.updated, state.pending_hashes
1097 plan_batches = _plan_batches(disk_files, existing_sources, skip_markers, absent, state, cancel)
1099 # Snapshot the cumulative truncation counter so the delta over this sync can
1100 # surface "N chunks truncated" instead of being lost in per-chunk debug logs.
1101 truncated_before = get_services().embedder.truncated_total
1103 # Ingest files (with optional progress bar). Only non-empty batches are yielded,
1104 # so the first one arriving is what proves there is work to do.
1105 try:
1106 first = await anext(plan_batches, None)
1107 if first is not None:
1108 # Hold the embed fleet resident for the whole batch: an unevenly loaded
1109 # replica must not idle-unload and reload cold mid-run (which snowballs
1110 # into a fleet collapse). The ContextVar propagates into the ingest
1111 # thread pool, where the fleet actually spawns on the first embed.
1112 from lilbee.providers.fleet.ingest_warmth import keep_fleet_warm
1114 with keep_fleet_warm():
1115 _require_embedding_model()
1116 await ingest_stream(
1117 _chain_plan_batches(first, plan_batches),
1118 added,
1119 updated,
1120 failed,
1121 skipped,
1122 plan=state,
1123 quiet=quiet,
1124 on_progress=on_progress,
1125 cancel=cancel,
1126 flush_failed=flush_failed,
1127 reasons=reasons,
1128 )
1129 if cancel is not None and cancel.is_set():
1130 # The stream stops feeding on cancel, so ingest can drain its
1131 # admitted files and return without raising. A cancelled run must
1132 # not go on to write skip markers or reconcile an unplanned corpus.
1133 raise asyncio.CancelledError
1134 finally:
1135 # Idempotent, and the only close when the stream is never consumed.
1136 await plan_batches.aclose()
1137 relocated = state.relocated
1139 # A flush failure is a transient store-side problem, not a verdict on the
1140 # file: leaving it unmarked re-plans it next sync instead of skipping it.
1141 marker_failed = [name for name in (*failed, *skipped) if name not in flush_failed]
1142 _persist_skip_markers(
1143 skip_markers, pending_hashes, succeeded=[*added, *updated], failed=marker_failed
1144 )
1145 # Record why each file this run skip-marked was marked (informational; the
1146 # hash markers above drive the resume logic). Only marker_failed files, so a
1147 # transient flush failure doesn't leave a stale reason behind.
1148 _persist_skip_reasons(skip_markers, {n: reasons[n] for n in marker_failed if n in reasons})
1150 if shard is None:
1151 # A worker's shard is merged before the indexes are built, so the passes
1152 # run once corpus-wide in the parent instead of once per shard.
1153 await _run_post_ingest_passes(
1154 _store,
1155 indexed_anything=bool(state.planned or relocated),
1156 # The old names of relocated sources ride along so the wiki index
1157 # subtracts them in the same pass that merges their new ones.
1158 touched=set(added) | set(updated) | set(relocated) | set(state.relocated_from),
1159 cancel=cancel,
1160 )
1162 # Reconciliation guard against silent data loss: any on-disk document file that
1163 # ended up in neither the index nor the failed/skipped lists was dropped without
1164 # a signal. Surface it loudly instead of letting a whole dataset vanish quietly.
1165 if missing := _reconcile_missing(disk_files, _store.get_sources(), failed, skipped):
1166 log.warning(
1167 "Sync reconciliation: %d document file(s) on disk are absent from the index "
1168 "with no failure reported (possible silent drop): %s",
1169 len(missing),
1170 ", ".join(missing[:20]),
1171 )
1173 result = SyncResult(
1174 added=list(added),
1175 updated=list(updated),
1176 removed=[],
1177 unchanged=state.unchanged,
1178 relocated=relocated,
1179 failed=list(failed),
1180 skipped=list(skipped),
1181 truncated=get_services().embedder.truncated_total - truncated_before,
1182 )
1183 on_progress(
1184 EventType.DONE,
1185 SyncDoneEvent(
1186 added=len(result.added),
1187 updated=len(result.updated),
1188 removed=len(result.removed),
1189 failed=len(result.failed),
1190 skipped=len(result.skipped),
1191 relocated=len(result.relocated),
1192 ),
1193 )
1194 return result
1197def _phase_progress_callback(
1198 progress: Progress, ptask: Any, chain: DetailedProgressCallback
1199) -> DetailedProgressCallback:
1200 """Wrap *chain*, updating the bar's description on per-page / per-chunk events.
1202 EXTRACT (vision OCR page i/N) and EMBED (chunk i/N) events would otherwise
1203 leave the bar frozen between file completions; surfacing them on the spinner
1204 description keeps a single large file's row visibly moving. All events still
1205 forward to *chain* so the caller's own callback (TUI / JSON) is unaffected.
1206 """
1208 def _callback(event_type: EventType, data: ProgressEvent) -> None:
1209 if event_type is EventType.EXTRACT and isinstance(data, ExtractEvent):
1210 progress.update(
1211 ptask, description=f"OCR {data.file} (page {data.page}/{data.total_pages})"
1212 )
1213 elif event_type is EventType.EMBED and isinstance(data, EmbedEvent):
1214 progress.update(
1215 ptask, description=f"Embedding {data.file} ({data.chunk}/{data.total_chunks})"
1216 )
1217 chain(event_type, data)
1219 return _callback
1222# In-flight task cap, as a multiple of _max_concurrent(): enough queued tasks to
1223# keep every compute slot fed, without materializing one task object per file.
1224_TASK_WINDOW_MULTIPLIER = 2
1227def _build_admission(
1228 baseline: int, pages_done: list[int]
1229) -> tuple[asyncio.Semaphore | ResizableGate, int, asyncio.Task[None] | None]:
1230 """The batch's admission control, plus its task-window size and controller task.
1232 Static mode (the default) returns a fixed semaphore and no controller. Adaptive
1233 mode, when a GPU fleet is present to feed, returns a resizable gate and a running
1234 :class:`AdaptiveController` that tunes it toward this box's throughput knee; with
1235 no fleet it falls back to the static path so a GPU-less host is never affected.
1236 """
1237 profile = profile_for(resolve_mode())
1238 devices = enumerate_fleet_devices() if profile is not None else []
1239 if profile is None or not devices:
1240 return asyncio.Semaphore(baseline), baseline * _TASK_WINDOW_MULTIPLIER, None
1241 permit_max = max_workers()
1242 gate = ResizableGate(min(baseline, permit_max))
1243 controller = AdaptiveController(
1244 gate,
1245 profile,
1246 make_signal_sampler(devices),
1247 lambda: pages_done[0],
1248 permit_min=1,
1249 permit_max=permit_max,
1250 )
1251 task = asyncio.ensure_future(controller.run())
1252 # warning, not info: the default LILBEE_LOG_LEVEL is WARNING, so the
1253 # auto-chosen concurrency would otherwise never surface on a headless sync.
1254 log.warning(
1255 "Adaptive ingest concurrency (%s): start %d, max %d", profile.name, gate.limit, permit_max
1256 )
1257 return gate, permit_max * _TASK_WINDOW_MULTIPLIER, task
1260def _failed_result(
1261 exc: Exception,
1262 entry: FileToProcess,
1263 *,
1264 pages_done: list[int],
1265 on_progress: DetailedProgressCallback,
1266 cancel: CancelSignal | None,
1267) -> _IngestResult:
1268 """A file's failure as a result, or cancellation when the run is stopping.
1270 During shutdown, worker pools raise RuntimeError from submit(). Those are
1271 cancellation, not ingest failures: the cancel flag is the source of truth,
1272 and the executor's shutdown message covers the race where cancel was set
1273 after the submit.
1274 """
1275 if (cancel and cancel.is_set()) or is_executor_shutdown(exc):
1276 raise asyncio.CancelledError from exc
1277 # Suppress TaskCancelledError on the FILE_DONE notice: the user already
1278 # cancelled, and re-raising here would strand sibling tasks awaiting in
1279 # _collect_results.
1280 with contextlib.suppress(TaskCancelledError):
1281 on_progress(EventType.FILE_DONE, FileDoneEvent(file=entry.name, status="error", chunks=0))
1282 pages_done[0] += 1 # cleared the gate (as a failure); still a throughput tick
1283 return _IngestResult(entry.name, entry.path, 0, error=exc)
1286async def _stream_tasks(
1287 plan_batches: AsyncGenerator[list[FileToProcess]],
1288 make_task: Callable[[FileToProcess, int], Coroutine[Any, Any, _IngestResult]],
1289) -> AsyncGenerator[list[Coroutine[Any, Any, _IngestResult]]]:
1290 """Each plan batch's per-file coroutines, in plan order."""
1291 index = count(1)
1292 async for plan_batch in plan_batches:
1293 yield [make_task(entry, next(index)) for entry in plan_batch]
1296async def _chain_plan_batches(
1297 first: list[FileToProcess], rest: AsyncGenerator[list[FileToProcess]]
1298) -> AsyncGenerator[list[FileToProcess]]:
1299 """Yield an already-pulled plan batch, then the remainder of its stream."""
1300 yield first
1301 async for plan_batch in rest:
1302 yield plan_batch
1305async def ingest_stream(
1306 plan_batches: AsyncGenerator[list[FileToProcess]],
1307 added: dict[str, None],
1308 updated: dict[str, None],
1309 failed: dict[str, None],
1310 skipped: dict[str, None],
1311 *,
1312 plan: _StreamedPlan | None = None,
1313 quiet: bool = False,
1314 on_progress: DetailedProgressCallback = noop_callback,
1315 cancel: CancelSignal | None = None,
1316 flush_failed: set[str] | None = None,
1317 reasons: dict[str, str] | None = None,
1318) -> None:
1319 """Ingest a stream of planned file batches, optionally showing a Rich progress bar.
1321 Files are admitted as their batch is planned, so ingest starts on the first
1322 batch instead of waiting for the whole corpus to be diffed. Old chunks are
1323 deleted in the same transaction as the new write, so the two are atomic per
1324 file. When *cancel* is set, pending files raise CancelledError before starting.
1326 *plan* is the bookkeeping the batches were planned into; it carries the corpus
1327 the run is measured against. Without it progress is reported with no total,
1328 since a bare stream of batches does not say what corpus it came from.
1329 """
1330 # Honor LILBEE_INGEST_TRACE once per batch: it raises the trace loggers above
1331 # the default WARNING so per-file extraction lines actually surface.
1332 configure_trace_from_env()
1333 warn_if_table_model_ignored()
1334 # Throughput is measured in OCR pages, not documents: a document's cost scales
1335 # with its page count (a 500-page scan is 500x a memo), so pages are the unbiased
1336 # unit of GPU-feeding work for the adaptive controller to hill-climb on.
1337 pages_done = [0]
1338 # Sized off the files with no source row yet, not the planned count: the plan
1339 # streams in batches and its total is unknown until the stream drains, but the
1340 # pool has to be decided before the first batch is dispatched. Unindexed files
1341 # are the one part of the plan that is known without diffing, exact for a first
1342 # ingest or a rebuild and near zero for an incremental sync, so a small sync
1343 # over a large corpus stays in-process. Undercounting only keeps a run
1344 # in-process, which is the safe direction.
1345 admission, window, controller_task = _build_admission(_max_concurrent(), pages_done)
1347 async def _process_one(entry: FileToProcess, file_index: int) -> _IngestResult:
1348 name = entry.name
1349 async with admission:
1350 if cancel and cancel.is_set():
1351 raise asyncio.CancelledError
1353 try:
1354 on_progress(
1355 EventType.FILE_START,
1356 FileStartEvent(file=name, total_files=feed.planned, current_file=file_index),
1357 )
1358 except TaskCancelledError as exc:
1359 # FILE_START itself can raise the cooperative cancel signal;
1360 # normalize so _collect_results can drain siblings cleanly.
1361 raise asyncio.CancelledError from exc
1362 try:
1363 # The source's old chunks are deleted in the same locked
1364 # transaction as the new write (see _flush_writes), so cleanup is
1365 # carried on the result rather than run eagerly here.
1366 page_texts: list[PageTextRecord] = []
1367 records, meta = await produce_records(
1368 entry.path,
1369 name,
1370 entry.content_type,
1371 quiet=quiet,
1372 on_progress=on_progress,
1373 page_texts_out=page_texts,
1374 )
1375 concept_records = await build_concept_records(records, name)
1376 entity_rows = await build_entity_records(records, name)
1377 on_progress(
1378 EventType.FILE_DONE,
1379 FileDoneEvent(file=name, status="ok", chunks=len(records)),
1380 )
1381 pages_done[0] += max(1, len(page_texts)) # OCR pages cleared: the throughput signal
1382 return _IngestResult(
1383 name,
1384 entry.path,
1385 len(records),
1386 error=None,
1387 file_hash=entry.file_hash,
1388 records=records,
1389 needs_cleanup=entry.needs_cleanup,
1390 page_texts=page_texts,
1391 stat=entry.stat,
1392 concept_records=concept_records,
1393 entity_rows=entity_rows,
1394 meta=meta,
1395 )
1396 except (asyncio.CancelledError, TaskCancelledError) as exc:
1397 # TaskCancelledError is the TUI's cooperative cancel signal raised
1398 # by reporter.check_cancelled() inside on_progress; treat it as
1399 # asyncio cancellation so _collect_results can drain siblings
1400 # cleanly instead of orphaning their pending exceptions.
1401 raise asyncio.CancelledError from exc
1402 except Exception as exc:
1403 return _failed_result(
1404 exc, entry, pages_done=pages_done, on_progress=on_progress, cancel=cancel
1405 )
1407 feed = _ResultFeed(_stream_tasks(plan_batches, _process_one), plan)
1408 collect = _collect_results if quiet else _collect_under_bar
1409 try:
1410 # extract_batching coalesces extractions into xberg batch calls when the
1411 # toggle is on (off by default); the per-file collect contract is unchanged.
1412 async with extract_batching():
1413 await collect(
1414 feed,
1415 added,
1416 updated,
1417 failed,
1418 skipped,
1419 window=window,
1420 on_progress=on_progress,
1421 flush_failed=flush_failed,
1422 reasons=reasons,
1423 )
1424 finally:
1425 # Stop the adaptive controller (if any) before returning: its background
1426 # loop must not outlive the batch it was tuning.
1427 if controller_task is not None:
1428 controller_task.cancel()
1429 with contextlib.suppress(asyncio.CancelledError):
1430 await controller_task
1433# Accumulate roughly this many chunks across documents before one batched
1434# LanceDB write. Bounds buffered-vector memory while amortizing the write lock
1435# and per-transaction overhead over many documents instead of one write per file.
1436_WRITE_FLUSH_CHUNKS = 2000
1439class _ResultFeed:
1440 """Pull-based source of per-file ingest coroutines over a streamed plan.
1442 ``take(wait=False)`` hands back an already-planned file without blocking, so
1443 the collector waits on the planner only when it has nothing left to run.
1444 Build-vs-buy: an ``asyncio.Queue`` is the stock bounded channel, but it would
1445 need a separate producer task to pump the plan-batch generator into it and a
1446 sentinel to close it; pulling ``anext`` on demand keeps the plan stream the
1447 single driver and needs neither. ``planned`` is the file count seen so far:
1448 the run's total once the stream is drained.
1449 """
1451 def __init__(
1452 self,
1453 plan_batches: AsyncGenerator[list[Coroutine[Any, Any, _IngestResult]]],
1454 plan: _StreamedPlan | None = None,
1455 ) -> None:
1456 self._plan_batches = plan_batches
1457 self._buffer: deque[Coroutine[Any, Any, _IngestResult]] = deque()
1458 self._pull: asyncio.Task[list[Coroutine[Any, Any, _IngestResult]] | None] | None = None
1459 self._drained = False
1460 self._plan = plan if plan is not None else _StreamedPlan()
1461 self.planned = 0
1463 @property
1464 def corpus_total(self) -> int:
1465 """Files the run's slice holds, or 0 when the caller declared no corpus."""
1466 return self._plan.corpus_total
1468 @property
1469 def resolved(self) -> int:
1470 """Files already disposed of by the plan, which never reach this feed."""
1471 return self._plan.resolved
1473 def pull(self) -> asyncio.Task[list[Coroutine[Any, Any, _IngestResult]] | None] | None:
1474 """The in-flight plan-batch prefetch, so a waiting collector wakes when it lands."""
1475 return self._pull
1477 async def take(self, *, wait: bool) -> Coroutine[Any, Any, _IngestResult] | None:
1478 """The next planned file, or None when the stream is drained (or, with
1479 *wait* False, when the next batch is not planned yet)."""
1480 while not self._buffer:
1481 if self._drained:
1482 return None
1483 if self._pull is None:
1484 self._pull = asyncio.ensure_future(anext(self._plan_batches, None))
1485 if not wait and not self._pull.done():
1486 return None
1487 plan_batch = await self._pull
1488 self._pull = None
1489 if plan_batch is None:
1490 self._drained = True
1491 return None
1492 self._buffer.extend(plan_batch)
1493 self.planned += len(plan_batch)
1494 return self._buffer.popleft()
1496 async def aclose(self) -> None:
1497 """Close the plan stream and discard files that were never started."""
1498 if self._pull is not None:
1499 self._pull.cancel()
1500 with contextlib.suppress(asyncio.CancelledError):
1501 # A batch that landed before the cancel took effect still owns
1502 # coroutines; recover it so they are closed rather than leaked.
1503 plan_batch = await self._pull
1504 if plan_batch:
1505 self._buffer.extend(plan_batch)
1506 self._pull = None
1507 for coro in self._buffer:
1508 coro.close()
1509 self._buffer.clear()
1510 await self._plan_batches.aclose()
1513async def _refill_window(
1514 in_flight: set[asyncio.Task[_IngestResult]],
1515 feed: _ResultFeed,
1516 window: int,
1517) -> None:
1518 """Top up the in-flight task set from *feed*, capped at *window* tasks.
1520 Waits on the planner only when nothing is running, so a slow batch never
1521 stalls files that are already planned.
1522 """
1523 while len(in_flight) < window:
1524 coro = await feed.take(wait=not in_flight)
1525 if coro is None:
1526 return
1527 in_flight.add(asyncio.ensure_future(coro))
1530async def _next_completions(
1531 in_flight: set[asyncio.Task[_IngestResult]], prefetch: asyncio.Future[Any] | None
1532) -> tuple[Iterable[asyncio.Future[Any]], set[asyncio.Task[_IngestResult]]]:
1533 """Wait for the next file to finish, returning (completed, still running).
1535 The feed's plan-batch prefetch waits alongside the running files, so work is
1536 admitted as soon as it is planned rather than on the next file completion,
1537 and it is filtered out of the still-running set here since it is not a file.
1538 """
1539 waiting: set[asyncio.Future[Any]] = set(in_flight)
1540 if prefetch is not None:
1541 waiting.add(prefetch)
1542 done, still_running = await asyncio.wait(waiting, return_when=asyncio.FIRST_COMPLETED)
1543 # Explicit loop, like _cancel_in_flight: Nuitka miscompiled the comprehension
1544 # form of this task-set filtering.
1545 remaining: set[asyncio.Task[_IngestResult]] = set()
1546 for task in still_running:
1547 if task is not prefetch:
1548 remaining.add(cast("asyncio.Task[_IngestResult]", task))
1549 return done, remaining
1552async def _collect_results(
1553 feed: _ResultFeed,
1554 added: dict[str, None],
1555 updated: dict[str, None],
1556 failed: dict[str, None],
1557 skipped: dict[str, None],
1558 *,
1559 window: int,
1560 on_progress: DetailedProgressCallback = noop_callback,
1561 progress: Progress | None = None,
1562 ptask: Any = None,
1563 flush_failed: set[str] | None = None,
1564 reasons: dict[str, str] | None = None,
1565) -> None:
1566 """Run *feed* through a bounded task window, batching writes and progress.
1568 At most *window* tasks exist at once: results are consumed as they complete
1569 and the window is refilled from the feed, so memory stays flat however many
1570 files a sync covers. Successful files are buffered and flushed to LanceDB in
1571 batches (one locked transaction per batch) rather than one write per file.
1572 The buffer is flushed on the way out too -- even on cancel -- so
1573 completed-but-unwritten work is persisted. On exception (typically
1574 asyncio.CancelledError from a user cancel), cancel every in-flight sibling
1575 and await them with ``return_exceptions=True`` so their pending
1576 CancelledErrors don't surface as "Task exception was never retrieved".
1577 """
1578 buffer: list[_IngestResult] = []
1579 buffered_chunks = 0
1580 completed_count = 0
1581 to_purge: list[str] = []
1582 in_flight: set[asyncio.Task[_IngestResult]] = set()
1583 try:
1584 await _refill_window(in_flight, feed, window)
1585 while in_flight:
1586 prefetch = feed.pull()
1587 done, in_flight = await _next_completions(in_flight, prefetch)
1588 saw_cancel = False
1589 for fut in done:
1590 if fut is prefetch:
1591 continue # a planned batch landing, not a file result
1592 try:
1593 result = fut.result()
1594 except asyncio.CancelledError:
1595 # A user cancel completes several futures together. Flag it but
1596 # keep draining `done` so a sibling that genuinely finished in
1597 # the same batch is still buffered and flushed (the
1598 # cancel-persists contract), then propagate after the loop. A
1599 # non-cancel exception still propagates immediately, as before,
1600 # so a genuine ingest bug surfaces and cancels the siblings.
1601 saw_cancel = True
1602 continue
1603 completed_count += 1
1604 status = _classify_result(result, added, updated, failed, skipped, reasons)
1605 if status is BatchStatus.INGESTED:
1606 buffered_chunks = await _buffer_and_maybe_flush(
1607 result,
1608 buffer,
1609 buffered_chunks,
1610 added,
1611 updated,
1612 failed,
1613 skipped,
1614 flush_failed,
1615 )
1616 elif status is BatchStatus.SKIPPED and result.needs_cleanup:
1617 # Zero-text result is never buffered; collect it for the
1618 # purge pass (see _purge_emptied_sources).
1619 to_purge.append(result.name)
1620 _report_file_progress(
1621 result,
1622 status,
1623 feed.resolved + completed_count,
1624 feed.corpus_total,
1625 on_progress,
1626 progress,
1627 ptask,
1628 )
1629 if saw_cancel:
1630 # Completed siblings in this batch are now buffered; propagate the
1631 # cancel so the finally flushes them and cancels still-running work.
1632 raise asyncio.CancelledError
1633 await _refill_window(in_flight, feed, window)
1634 finally:
1635 # The inner finally guarantees the sibling cancel even if the flush
1636 # itself raises (e.g. a cancellation landing on the to_thread await).
1637 try:
1638 await to_ingest_thread(
1639 _flush_writes, buffer, added, updated, failed, skipped, flush_failed
1640 )
1641 await to_ingest_thread(_purge_emptied_sources, to_purge)
1642 finally:
1643 try:
1644 await _cancel_in_flight(in_flight)
1645 finally:
1646 # Closing the feed stops the planner behind it, so a cancelled
1647 # sync does not keep hashing the rest of the corpus.
1648 await feed.aclose()
1651async def _collect_under_bar(
1652 feed: _ResultFeed,
1653 added: dict[str, None],
1654 updated: dict[str, None],
1655 failed: dict[str, None],
1656 skipped: dict[str, None],
1657 *,
1658 window: int,
1659 on_progress: DetailedProgressCallback = noop_callback,
1660 flush_failed: set[str] | None = None,
1661 reasons: dict[str, str] | None = None,
1662) -> None:
1663 """Run :func:`_collect_results` under a transient Rich progress bar."""
1664 with Progress(
1665 SpinnerColumn(),
1666 TextColumn("{task.description}"),
1667 BarColumn(),
1668 MofNCompleteColumn(),
1669 TimeElapsedColumn(),
1670 transient=True,
1671 ) as progress:
1672 # The corpus the discovery walk found, known before the first batch is
1673 # planned. None only when the caller supplied no plan to measure against.
1674 ptask = progress.add_task("Ingesting documents...", total=feed.corpus_total or None)
1675 # The bar advances once per file (in _collect_results), so a single
1676 # multi-page scanned PDF would freeze at "0/1" through its whole
1677 # OCR + embed phase. Drive the spinner's description off the same
1678 # EXTRACT (OCR page i/N) and EMBED (chunk i/N) events the TUI uses
1679 # so the row visibly moves while one file is being worked.
1680 await _collect_results(
1681 feed,
1682 added,
1683 updated,
1684 failed,
1685 skipped,
1686 window=window,
1687 on_progress=_phase_progress_callback(progress, ptask, on_progress),
1688 progress=progress,
1689 ptask=ptask,
1690 flush_failed=flush_failed,
1691 reasons=reasons,
1692 )
1695async def _cancel_in_flight(in_flight: set[asyncio.Task[_IngestResult]]) -> None:
1696 """Cancel still-running tasks and await them so their CancelledErrors are retrieved."""
1697 # Explicit loop: Nuitka miscompiled the comprehension form of this cleanup.
1698 still_pending = []
1699 for t in in_flight:
1700 if not t.done():
1701 still_pending.append(t)
1702 for task in still_pending:
1703 task.cancel()
1704 if still_pending:
1705 await asyncio.gather(*still_pending, return_exceptions=True)
1708async def _buffer_and_maybe_flush(
1709 result: _IngestResult,
1710 buffer: list[_IngestResult],
1711 buffered_chunks: int,
1712 added: dict[str, None],
1713 updated: dict[str, None],
1714 failed: dict[str, None],
1715 skipped: dict[str, None],
1716 flush_failed: set[str] | None,
1717) -> int:
1718 """Buffer one ingested file, flushing at the chunk threshold; returns the new count."""
1719 buffer.append(result)
1720 # Zero-chunk files count one unit so the buffer stays bounded.
1721 buffered_chunks += max(result.chunk_count, 1)
1722 if buffered_chunks >= _WRITE_FLUSH_CHUNKS:
1723 await to_ingest_thread(_flush_writes, buffer, added, updated, failed, skipped, flush_failed)
1724 buffered_chunks = 0
1725 return buffered_chunks
1728def _report_file_progress(
1729 result: _IngestResult,
1730 status: BatchStatus,
1731 completed_count: int,
1732 total: int,
1733 on_progress: DetailedProgressCallback,
1734 progress: Progress | None,
1735 ptask: Any,
1736) -> None:
1737 """Advance the Rich bar (when present) and emit one BATCH_PROGRESS event.
1739 *completed_count* counts every file the pass has disposed of and *total* is
1740 the corpus the discovery walk found, so the pair answers how much of the
1741 corpus is done rather than how much of the plan so far is.
1742 """
1743 if progress is not None and ptask is not None:
1744 desc = f"Ingested {result.name}" if result.error is None else f"Failed {result.name}"
1745 # Set, not advanced: files the plan resolved without ingest produce no
1746 # result of their own and would otherwise never reach the bar.
1747 progress.update(ptask, description=desc, completed=completed_count)
1748 with contextlib.suppress(TaskCancelledError):
1749 on_progress(
1750 EventType.BATCH_PROGRESS,
1751 BatchProgressEvent(
1752 file=result.name,
1753 status=status,
1754 current=completed_count,
1755 total=total,
1756 ),
1757 )
1760def _classify_result(
1761 result: _IngestResult,
1762 added: dict[str, None],
1763 updated: dict[str, None],
1764 failed: dict[str, None],
1765 skipped: dict[str, None],
1766 reasons: dict[str, str] | None = None,
1767) -> BatchStatus:
1768 """Record a completed file's outcome and return its batch status.
1770 Failures and zero-chunk files are tracked here; a successful file is reported
1771 as ``INGESTED`` and its chunks are persisted by the batched flush, so it stays
1772 in ``added`` / ``updated`` until then. When *reasons* is given, the
1773 human-readable cause is recorded there (filename → reason) for reporting.
1774 """
1775 if result.error is not None:
1776 # A traceback here would bleed into the TUI chat pane; the full trace stays at DEBUG.
1777 log.warning("Failed to ingest %s: %s", result.name, result.error)
1778 log.debug("Traceback for failed ingest of %s", result.name, exc_info=result.error)
1779 added.pop(result.name, None)
1780 updated.pop(result.name, None)
1781 failed[result.name] = None
1782 if reasons is not None:
1783 reasons[result.name] = error_reason(result.error)
1784 return BatchStatus.FAILED
1785 if result.chunk_count == 0:
1786 # No searchable chunks: never report it as added/updated. With no page
1787 # texts either, nothing is persisted and the file retries next sync. With
1788 # page texts, it stays INGESTED so its pages persist (export/recon) and it
1789 # stops replanning, but it is reported as skipped since search can't see it.
1790 added.pop(result.name, None)
1791 updated.pop(result.name, None)
1792 skipped[result.name] = None
1793 if reasons is not None:
1794 reasons[result.name] = (
1795 "no text extracted (0 chunks)"
1796 if not result.page_texts
1797 else "stored page text only (0 searchable chunks)"
1798 )
1799 return BatchStatus.SKIPPED if not result.page_texts else BatchStatus.INGESTED
1800 return BatchStatus.INGESTED
1803# Back off briefly before the single flush retry: the usual contender is a
1804# search-triggered FTS optimize holding the store lock past its 30s timeout.
1805_FLUSH_RETRY_DELAY_SECONDS = 2.0
1808def _retry_after_lock_timeout(write: Callable[[], object]) -> None:
1809 """Run one store write, retrying once after a lock timeout."""
1810 try:
1811 write()
1812 except LockTimeoutError:
1813 log.warning(
1814 "Store write lock busy; retrying batch flush in %.0fs", _FLUSH_RETRY_DELAY_SECONDS
1815 )
1816 time.sleep(_FLUSH_RETRY_DELAY_SECONDS)
1817 write()
1820def _flush_batch(buffer: list[_IngestResult]) -> None:
1821 """Persist one flush unit in a single locked ``write_chunks_batch`` transaction.
1823 Page texts travel inside each :class:`ChunkWrite` so the store writes them
1824 after the cleanup delete (which clears the source's old page-text rows) and
1825 before the source row: a page-text failure leaves the row stale and the file
1826 replans next sync instead of losing its pages forever behind the stat
1827 short-circuit. The write retries once on a lock timeout.
1828 """
1829 store = get_services().store
1830 items = [
1831 ChunkWrite(
1832 source=r.name,
1833 file_hash=r.file_hash or file_hash(r.path),
1834 records=cast(list[dict], r.records or []),
1835 needs_cleanup=r.needs_cleanup,
1836 stat=r.stat,
1837 page_texts=cast(list[dict], r.page_texts or []),
1838 meta=r.meta,
1839 )
1840 for r in buffer
1841 ]
1842 _retry_after_lock_timeout(lambda: store.write_chunks_batch(items))
1843 _flush_concept_records(buffer)
1844 _flush_entity_rows(buffer)
1847def _flush_concept_records(buffer: list[_IngestResult]) -> None:
1848 """Write the flush unit's buffered concept rows in one batched pass.
1850 Runs after the chunk write so a failed flush (files moved to ``failed``
1851 and replanned) never lands concept rows for unwritten chunks. A concept
1852 write failure is logged and never fails the files, matching the
1853 per-file extraction failure semantics.
1854 """
1855 batches = [r.concept_records for r in buffer if r.concept_records is not None]
1856 if not batches:
1857 return
1858 try:
1859 get_services().concepts.write_concept_records(ConceptRecords.merged(batches))
1860 except Exception:
1861 log.warning("Concept indexing failed for %d-file batch", len(batches), exc_info=True)
1864def _flush_entity_rows(buffer: list[_IngestResult]) -> None:
1865 """Write the flush unit's buffered entity rows in one batched pass.
1867 Runs after the chunk write, which also performed the per-source deletes,
1868 so replacement never leaves a source's stale entity rows behind. A write
1869 failure is logged and never fails the files, matching concept semantics.
1870 """
1871 rows = [row for r in buffer if r.entity_rows for row in r.entity_rows]
1872 if not rows:
1873 return
1874 try:
1875 get_services().store.add_entities(rows)
1876 except Exception:
1877 log.warning("Entity indexing failed for a %d-row batch", len(rows), exc_info=True)
1880def _purge_emptied_sources(names: list[str]) -> None:
1881 """Remove the prior index entry for files that now extract to nothing.
1883 An already-indexed file edited to yield zero chunks and zero page texts is
1884 classified SKIPPED and never buffered, so the batched cleanup delete never
1885 runs and its old chunks and source row would linger in search results. Full
1886 removal here keeps the index consistent; ``remove_documents`` is a no-op for
1887 never-indexed (brand-new empty) files, so unindexed inputs cost nothing.
1888 """
1889 if not names:
1890 return
1891 get_services().store.remove_documents(names)
1894def _flush_writes(
1895 buffer: list[_IngestResult],
1896 added: dict[str, None],
1897 updated: dict[str, None],
1898 failed: dict[str, None],
1899 skipped: dict[str, None],
1900 flush_failed: set[str] | None = None,
1901) -> None:
1902 """Flush the buffered documents to the store; track a write failure.
1904 Each buffered file's page texts, chunks, cleanup delete, and source upsert
1905 are written by :func:`_flush_batch`. If that fails, every file in the batch
1906 is moved to ``failed`` since its source row did not land, and recorded in
1907 *flush_failed* so the caller replans them next sync instead of skip-marking
1908 them; the exception never escapes, so the caller's sibling-cancel and
1909 skip-marker path always runs. The buffer is cleared either way.
1910 """
1911 if not buffer:
1912 return
1913 try:
1914 _flush_batch(buffer)
1915 except Exception as exc:
1916 for r in buffer:
1917 log.warning("Failed to write %s: %s", r.name, exc)
1918 added.pop(r.name, None)
1919 updated.pop(r.name, None)
1920 # A page-text-only file was pre-marked skipped at classification; on a
1921 # flush failure it belongs in failed only, never both.
1922 skipped.pop(r.name, None)
1923 failed[r.name] = None
1924 if flush_failed is not None:
1925 flush_failed.add(r.name)
1926 finally:
1927 buffer.clear()