Coverage for src/lilbee/data/store/crawl_pages.py: 100%
268 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-11-06 08:17 +0000
1"""The crawl page table: which file each crawled page owns, and the earlier copies to retire."""
3from __future__ import annotations
5import json
6import logging
7import threading
8import uuid
9from collections.abc import Callable, Iterator
10from contextlib import contextmanager
11from dataclasses import asdict, dataclass, fields, replace
12from datetime import timedelta
13from pathlib import Path
14from typing import TYPE_CHECKING, Any
16import pyarrow as pa
18from lilbee.core.config import CRAWL_PAGES_TABLE
19from lilbee.core.web_names import WEB_PREFIX
20from lilbee.runtime.lock import LOCK_TIMEOUT, crawl_pages_lock
22from .lance_helpers import ensure_table, escape_sql_string, open_committed
24if TYPE_CHECKING:
25 from lancedb.db import LanceDBConnection
26 from lancedb.table import LanceTable
28log = logging.getLogger(__name__)
30# The file an older lilbee kept its crawled pages in, under the data directory.
31CRAWL_META_FILE = "crawl_meta.json"
32# The column that orders the rows of one [address, file] key: the highest one counts.
33_SEQ = "seq"
34# The schema metadata key of an id each new table gets, so a table made again reads as new.
35_TABLE_ID = b"lilbee.table_id"
36# The commits a store makes between two folds of the table, and the keys one delete names.
37_FOLD_EVERY = 100
38_KEYS_PER_DELETE = 200
39# How long a version a fold replaces stays readable: far longer than one read of the
40# table, which takes milliseconds. Kept short, as each kept version slows the check for
41# the latest one (about 1 ms per 1,000 versions on a laptop).
42_KEEP_VERSIONS = timedelta(seconds=10)
45class CrawlPagesError(OSError):
46 """The crawl page table refused a read or a write; the crawl skips that one page."""
49@dataclass(frozen=True)
50class CrawlPage:
51 """One crawled page's file: the current one the page owns, or an earlier copy of it."""
53 address: str
54 file: str
55 content_hash: str
56 crawled_at: str
57 # The hash of the text the file held when the page claimed it; empty for a new file.
58 replaces: str = ""
59 # The page's own title; empty when the page has none or an older lilbee saved it.
60 title: str = ""
61 # False for an earlier copy: a file an earlier claim or name rule gave the page.
62 current: bool = True
65def _schema() -> pa.Schema:
66 return pa.schema(
67 [
68 pa.field("address", pa.utf8()),
69 pa.field("file", pa.utf8()),
70 pa.field("content_hash", pa.utf8()),
71 pa.field("crawled_at", pa.utf8()),
72 pa.field("replaces", pa.utf8()),
73 pa.field("title", pa.utf8()),
74 pa.field("current", pa.bool_()),
75 pa.field(_SEQ, pa.int64()),
76 ],
77 metadata={_TABLE_ID: uuid.uuid4().hex.encode()},
78 )
81_COLUMNS = [f.name for f in fields(CrawlPage)]
82_Key = tuple[str, str]
83# The id of a table and the version of it a view holds.
84_Stamp = tuple[bytes, int]
87def _stamp(table: LanceTable) -> _Stamp:
88 """Which table *table* is, and at which version: a reset that reaches the version differs."""
89 return (table.schema.metadata or {}).get(_TABLE_ID, b""), table.version
92def _where(**columns: str) -> str:
93 return " AND ".join(f"{name} = '{escape_sql_string(value)}'" for name, value in columns.items())
96def _index(index: dict[str, dict[_Key, CrawlPage]], value: str, page: CrawlPage) -> None:
97 index.setdefault(value, {})[(page.address, page.file)] = page
100def _unindex(index: dict[str, dict[_Key, CrawlPage]], value: str, key: _Key) -> None:
101 entries = index.get(value, {})
102 entries.pop(key, None)
103 if not entries:
104 index.pop(value, None)
107class _View:
108 """The rows of the table at one version, found by address, by owned file or by text.
110 A save adds a row and never rewrites one, so a key can have several rows: the one
111 with the highest ``seq`` counts, and the others wait for :meth:`CrawlPages.compact`.
112 """
114 def __init__(self, stamp: _Stamp | None, rows: list[dict[str, Any]]) -> None:
115 self.stamp = stamp
116 # The next seq to write: above every seq the table holds.
117 self.next_seq = 0
118 self._rows: dict[_Key, tuple[int, CrawlPage]] = {}
119 # Keys with rows a later row of the key replaced.
120 self.replaced: set[_Key] = set()
121 self._by_address: dict[str, dict[_Key, CrawlPage]] = {}
122 # Current rows by their file, and earlier copies by the hash of their text.
123 self._owners: dict[str, dict[_Key, CrawlPage]] = {}
124 self._copies: dict[str, dict[_Key, CrawlPage]] = {}
125 for row in rows:
126 self.put(CrawlPage(**{name: row[name] for name in _COLUMNS}), row[_SEQ])
128 def put(self, page: CrawlPage, seq: int) -> None:
129 """Note *page*, written at *seq*; a lower seq of its key loses to the row held."""
130 key = (page.address, page.file)
131 self.next_seq = max(self.next_seq, seq + 1)
132 held = self._rows.get(key)
133 if held is not None:
134 self.replaced.add(key)
135 if held[0] > seq:
136 return
137 self.drop(page)
138 self._rows[key] = (seq, page)
139 _index(self._by_address, page.address, page)
140 if page.current:
141 _index(self._owners, page.file, page)
142 else:
143 _index(self._copies, page.content_hash, page)
145 def drop(self, page: CrawlPage) -> None:
146 key = (page.address, page.file)
147 held = self._rows.pop(key, None)
148 if held is None:
149 return
150 old = held[1]
151 _unindex(self._by_address, old.address, key)
152 if old.current:
153 _unindex(self._owners, old.file, key)
154 else:
155 _unindex(self._copies, old.content_hash, key)
157 def seq(self, key: _Key) -> int:
158 return self._rows[key][0]
160 def row(self, key: _Key) -> CrawlPage | None:
161 held = self._rows.get(key)
162 return None if held is None else held[1]
164 def of(self, address: str) -> list[CrawlPage]:
165 return list(self._by_address.get(address, {}).values())
167 def owners(self, file: str) -> list[CrawlPage]:
168 return list(self._owners.get(file, {}).values())
170 def copies(self, content_hash: str) -> list[CrawlPage]:
171 return list(self._copies.get(content_hash, {}).values())
173 def all(self) -> list[CrawlPage]:
174 return [page for _seq, page in self._rows.values()]
177def _read(table: LanceTable | None, stamp: _Stamp | None = None) -> _View:
178 """Every row of *table* at the version it has checked out, under *stamp* when given."""
179 if table is None:
180 return _View(None, [])
181 return _View(_stamp(table) if stamp is None else stamp, table.to_arrow().to_pylist())
184def _add(table: LanceTable, view: _View, pages: list[CrawlPage]) -> int:
185 """Add *pages* in one commit above every row *view* holds; the version it made."""
186 seq = view.next_seq
187 rows = [{**asdict(page), _SEQ: seq + i} for i, page in enumerate(pages)]
188 version: int = table.add(rows).version
189 for i, page in enumerate(pages):
190 view.put(page, seq + i)
191 return version
194class CrawlPages:
195 """The crawl page table of one library. A write runs inside :meth:`held`.
197 The crawl lock comes before the store's write lock and never after it, so a holder
198 may remove index sources inside :meth:`held`.
200 The crawl lock orders the holds, but not every commit: the import of an older
201 release's files commits under the upgrade lock. So a hold checks the table's version
202 and reads the rows again when another commit, of any process, came since, or when a
203 commit of its own skipped a version. A hold keeps the rows in memory, with the table
204 open, and its own writes update both.
206 A LanceDB failure raises :class:`CrawlPagesError`, an ``OSError``, so a crawl skips
207 the one page, as it skips a page the disk refuses.
208 """
210 def __init__(self, open_db: Callable[[], LanceDBConnection], lancedb_dir: Path) -> None:
211 self._open_db = open_db
212 self._lancedb_dir = lancedb_dir
213 self._titles: tuple[_Stamp, dict[str, str]] | None = None
214 # The table and its rows as the last hold left them; used only inside a hold.
215 self._table: LanceTable | None = None
216 self._view = _View(None, [])
217 # The thread that holds the crawl lock through this object, or None.
218 self._holder: int | None = None
219 # The commits this store made since it last folded the table.
220 self._unfolded = 0
222 @contextmanager
223 def held(self, timeout: float | None = LOCK_TIMEOUT) -> Iterator[None]:
224 """Hold the crawl lock for the block; the library is open, and upgraded, first."""
225 self._open_db()
226 with crawl_pages_lock(self._lancedb_dir, timeout):
227 with self._lancedb_failure():
228 self._catch_up()
229 self._holder = threading.get_ident()
230 try:
231 yield
232 finally:
233 self._holder = None
235 @contextmanager
236 def _lancedb_failure(self) -> Iterator[None]:
237 """Raise a LanceDB failure as :class:`CrawlPagesError`.
239 The rows in memory change only after a commit lands, so the next hold's version
240 check reads them again when a failed write committed part of its work.
241 """
242 try:
243 yield
244 except RuntimeError as exc:
245 raise CrawlPagesError(f"The crawl page table failed: {exc}") from exc
247 def _catch_up(self) -> None:
248 """Read the table again when a commit since the last hold changed it."""
249 if self._table is not None:
250 try:
251 self._table.checkout_latest()
252 except (RuntimeError, OSError):
253 # A reset deleted the library under this store; the open table is gone.
254 self._table = None
255 if self._table is None:
256 self._view = _View(None, [])
257 self._table = open_committed(self._open_db(), CRAWL_PAGES_TABLE)
258 if self._table is not None:
259 stamp = _stamp(self._table)
260 if stamp != self._view.stamp:
261 self._view = _read(self._table, stamp)
263 def _rows_now(self) -> _View:
264 """The rows of the hold, or outside one the rows of the last commit."""
265 if self._holder == threading.get_ident():
266 return self._view
267 return _read(open_committed(self._open_db(), CRAWL_PAGES_TABLE))
269 def _committed(self, table: LanceTable, version: int) -> None:
270 """Note that the hold's own write made *version* of *table*.
272 A version the hold did not expect means a writer without the crawl lock, an
273 older lilbee's files imported at store open, committed in between: read again.
274 """
275 expected = self._view.stamp is not None and version == self._view.stamp[1] + 1
276 self._table = table
277 self._unfolded += 1
278 if expected and self._view.stamp is not None:
279 self._view.stamp = (self._view.stamp[0], version)
280 else:
281 self._view = _read(table)
283 def rows(self) -> list[CrawlPage]:
284 """Every row of the table."""
285 return self._rows_now().all()
287 def current(self, address: str) -> CrawlPage | None:
288 """The file the page at *address* owns, as the last commit holds it."""
289 return next((row for row in self._rows_now().of(address) if row.current), None)
291 def earlier(self, address: str) -> list[CrawlPage]:
292 """The earlier copies of the page at *address*."""
293 return [row for row in self._rows_now().of(address) if not row.current]
295 def owners(self, file: str) -> list[CrawlPage]:
296 """The pages whose current file is *file*."""
297 return self._rows_now().owners(file)
299 def copies_of(self, content_hash: str) -> list[CrawlPage]:
300 """The earlier copies that hold the text *content_hash* names."""
301 return self._rows_now().copies(content_hash)
303 def claim(self, page: CrawlPage) -> None:
304 """Write *page*; when it is current, the page's current file before it becomes earlier.
306 One commit adds both rows. Caller holds :meth:`held`.
307 """
308 before = [
309 replace(row, current=False)
310 for row in self._view.of(page.address)
311 if page.current and row.current and row.file != page.file
312 ]
313 with self._lancedb_failure():
314 # The truth value of a table counts its rows, so the test is for None.
315 table = self._table
316 if table is None:
317 table = ensure_table(self._open_db(), CRAWL_PAGES_TABLE, _schema())
318 self._committed(table, _add(table, self._view, [page, *before]))
320 def forget(self, page: CrawlPage) -> None:
321 """Delete the rows of *page*'s key. Caller holds :meth:`held`."""
322 if self._table is not None:
323 with self._lancedb_failure():
324 deleted = self._table.delete(_where(address=page.address, file=page.file))
325 self._committed(self._table, deleted.version)
326 self._view.drop(page)
328 def rekey(
329 self, canonical: Callable[[str], str], holds_text: Callable[[CrawlPage], bool]
330 ) -> None:
331 """Key each earlier copy by the address *canonical* gives. Caller holds :meth:`held`.
333 An older lilbee keyed earlier copies by the address as the crawl was given it, so
334 two spellings of one page can name one file. Of the copies that meet on one row,
335 one whose file *holds_text* says holds its text is kept. A current row is never
336 made an earlier copy.
337 """
338 table = self._table
339 moved = [r for r in self._view.all() if not r.current and canonical(r.address) != r.address]
340 if table is None or not moved:
341 return
342 kept: dict[_Key, CrawlPage] = {}
343 for row in moved:
344 copy = replace(row, address=canonical(row.address))
345 key = (copy.address, copy.file)
346 held = kept.get(key) or self._view.row(key)
347 if held is None or (not held.current and not holds_text(held) and holds_text(copy)):
348 kept[key] = copy
349 with self._lancedb_failure():
350 if kept:
351 _add(table, self._view, list(kept.values()))
352 table.delete(" OR ".join(f"({_where(address=r.address, file=r.file)})" for r in moved))
353 self._view = _read(table)
355 def compact(self) -> None:
356 """Delete the rows later rows replaced, then fold the table's commits into one version.
358 A version the fold replaces stays readable for ``_KEEP_VERSIONS``, so a reader of
359 another process at that version still reads it. A store that committed nothing
360 since its last fold, and holds no replaced rows, has nothing to fold. Caller holds
361 :meth:`held`.
362 """
363 table = self._table
364 if table is None or not (self._unfolded or self._view.replaced):
365 return
366 with self._lancedb_failure():
367 keys = sorted(self._view.replaced)
368 for start in range(0, len(keys), _KEYS_PER_DELETE):
369 table.delete(
370 " OR ".join(
371 f"({_where(address=address, file=file)} AND {_SEQ} < {self._view.seq(key)})"
372 for key in keys[start : start + _KEYS_PER_DELETE]
373 for address, file in [key]
374 if self._view.row(key) is not None
375 )
376 or "false"
377 )
378 table.optimize(cleanup_older_than=_KEEP_VERSIONS)
379 self._view.replaced.clear()
380 # One row per key is left; a count that differs means a writer without the
381 # crawl lock committed in between, so the rows are read again.
382 if table.count_rows() == len(self._view.all()):
383 self._view.stamp = _stamp(table)
384 else:
385 self._view = _read(table)
386 self._unfolded = 0
388 def compact_when_due(self) -> None:
389 """Fold the table after every ``_FOLD_EVERY`` commits; a failure only warns.
391 A long crawl folds as it goes, so the table never holds more than about that
392 many fragments, and each new version stays small. Caller holds :meth:`held`.
393 """
394 if self._unfolded < _FOLD_EVERY:
395 return
396 try:
397 self.compact()
398 except CrawlPagesError:
399 log.warning("Folding the crawl page table failed", exc_info=True)
401 def titles(self) -> dict[str, str]:
402 """The title of each crawled page that has one, by source name."""
403 table = open_committed(self._open_db(), CRAWL_PAGES_TABLE)
404 if table is None:
405 return {}
406 stamp = _stamp(table)
407 if self._titles is None or self._titles[0] != stamp:
408 rows = _read(table, stamp).all()
409 self._titles = (
410 stamp,
411 {f"{WEB_PREFIX}{r.file}": r.title for r in rows if r.current and r.title},
412 )
413 return self._titles[1]
416def legacy_crawl_files(data_dir: Path) -> list[Path]:
417 """The crawl metadata file of an older lilbee, when it is still under *data_dir*."""
418 path = data_dir / CRAWL_META_FILE
419 return [path] if path.exists() else []
422def _entry(address: str, data: Any) -> CrawlPage | None:
423 """The earlier copy one legacy entry describes, or None when it is malformed.
425 Every field an older lilbee wrote is a string; any other value is malformed.
426 """
427 strings = isinstance(data, dict) and all(isinstance(value, str) for value in data.values())
428 try:
429 if strings:
430 return CrawlPage(address=address, current=False, **data)
431 except TypeError:
432 pass
433 log.warning("Skipping a malformed crawl record entry of %s", address)
434 return None
437def _meta_pages(path: Path) -> list[CrawlPage] | None:
438 """The earlier copies the crawl metadata at *path* names, keyed as it wrote them.
440 None when the disk refuses the read, so the file stays for a later open.
441 """
442 try:
443 raw = json.loads(path.read_text(encoding="utf-8"))
444 except OSError as exc:
445 log.warning("Could not read crawl metadata %s: %s", path, exc)
446 return None
447 except (json.JSONDecodeError, UnicodeDecodeError) as exc:
448 log.warning("Discarding unreadable crawl metadata %s: %s", path, exc)
449 return []
450 if not isinstance(raw, dict):
451 return []
452 entries = (_entry(str(key), data) for key, data in raw.items())
453 return [page for page in entries if page is not None]
456def import_legacy_crawl_files(db: LanceDBConnection, data_dir: Path) -> None:
457 """Move the crawl metadata under *data_dir* into the table as earlier copies, then delete it.
459 Store write lock held. A file the disk refuses to read stays, with a warning. One
460 commit adds every entry the table has no row for; the file goes last, so an open
461 killed before then, or a file the disk will not delete, commits nothing the next time.
462 """
463 path = data_dir / CRAWL_META_FILE
464 entries = _meta_pages(path)
465 if entries is None:
466 return
467 pages = {(page.address, page.file): page for page in entries}
468 table = ensure_table(db, CRAWL_PAGES_TABLE, _schema())
469 view = _read(table)
470 new = [page for key, page in pages.items() if view.row(key) is None]
471 if new:
472 _add(table, view, new)
473 try:
474 path.unlink(missing_ok=True)
475 except OSError as exc:
476 log.warning("Could not delete %s: %s", path, exc)
477 log.info("Moved the crawl record of %s into the index", data_dir)