From 9f22dbf94b04df506a412b1b1121ca8f3c10de12 Mon Sep 17 00:00:00 2001 From: Polish Hammer Date: Tue, 4 Aug 2026 12:09:46 +0200 Subject: [PATCH] Librarian: 'still searching' heartbeat every 20 min A deep scan runs for hours with nothing in the log between start and finish, so it's impossible to tell a working search from a wedged one. Every CONJURER_LIBRARIAN_HEARTBEAT_SECONDS (default 1200 = 20 min) a running search now logs that it is still going, with its uuid, the search phrase, hits so far, elapsed minutes, and a rough how-far-along. The estimate is deliberately cheap: the producers ALREADY record a byte offset per chunk file (the resume watermarks), and the total size is stat()'d once per search when the chunk list is discovered. A reading is then just a sum over ~40 ints - nothing extra happens per line, and no cycles are spent estimating how many cycles are left. search_for_doi takes an optional progress dict it fills with the live positions dict + total_bytes; the librarian publishes the running search (uuid/query/progress/live hits) while the scan runs and clears it in finally. Nothing running => the heartbeat stays quiet. Tests: percentage maths incl. unknown-total and >100% clamping, the register/clear round-trip, and an end-to-end check that a real scan fills progress so the offsets cover the chunk files on disk. Suite: 58 unit + 65 integration green. Co-Authored-By: Claude Opus 4.8 --- conjurer_librarian/conjurer_librarian.py | 79 ++++++++++++++++- conjurer_librarian/search_bot.py | 21 ++++- tests/integration/test_librarian_heartbeat.py | 84 +++++++++++++++++++ 3 files changed, 179 insertions(+), 5 deletions(-) create mode 100644 tests/integration/test_librarian_heartbeat.py diff --git a/conjurer_librarian/conjurer_librarian.py b/conjurer_librarian/conjurer_librarian.py index f080020..de4e136 100644 --- a/conjurer_librarian/conjurer_librarian.py +++ b/conjurer_librarian/conjurer_librarian.py @@ -173,6 +173,65 @@ def _cache_put(query, deep_search, final_result) -> None: _cache.prune(CACHE_MAX_ENTRIES) +# ---- "Still alive" heartbeat for the running search ------------------------ +# A deep scan runs for hours with nothing in the log between start and finish. +# Every HEARTBEAT_SECONDS the running search says it is still going, with its +# uuid, the phrase, and a ROUGH how-far-along. The estimate is deliberately +# cheap: producers already record a byte offset per chunk file, and the total +# size is stat()'d once at search start - so it costs a sum over ~40 ints. +HEARTBEAT_SECONDS = int(_env("CONJURER_LIBRARIAN_HEARTBEAT_SECONDS", "1200")) # 20 min +_current_search: Dict[str, object] = {} +_current_lock = threading.Lock() + + +def _set_current_search(uuid, query, progress, live_results) -> None: + with _current_lock: + _current_search.clear() + _current_search.update({ + "uuid": str(uuid), "query": str(query), "started": time.monotonic(), + "progress": progress, "live": live_results, + }) + + +def _clear_current_search() -> None: + with _current_lock: + _current_search.clear() + + +def _progress_summary(progress): + """(done_bytes, total_bytes, percent) from a live progress dict. Cheap: a + sum over one int per chunk file. Percent is 0.0 when the total is unknown.""" + progress = progress or {} + positions = progress.get("positions") or {} + total = progress.get("total_bytes") or 0 + done = sum(positions.values()) + if total > 0: + done = min(done, total) # a partially-buffered tail can nudge past 100% + return done, total, 100.0 * done / total + return done, total, 0.0 + + +def search_heartbeat(app_logger) -> None: + """Log a 'still searching' line every HEARTBEAT_SECONDS while one runs.""" + while not SHUTDOWN_EVENT.wait(HEARTBEAT_SECONDS): + try: + with _current_lock: + snapshot = dict(_current_search) if _current_search else None + if not snapshot: + continue # nothing running - stay quiet + done, total, percent = _progress_summary(snapshot.get("progress")) + app_logger.info( + "SEARCH ALIVE %s | '%s' | ~%.1f%% przeskanowane (%.2f/%.2f GB, " + "~%.2f GB do końca) | %d trafień | %.0f min", + snapshot["uuid"], snapshot["query"], percent, + done / 1e9, total / 1e9, max(0, total - done) / 1e9, + len(snapshot.get("live") or []), + (time.monotonic() - snapshot["started"]) / 60.0, + ) + except Exception as exc: # pylint: disable=broad-exception-caught + app_logger.warning("Heartbeat tick failed: %s", exc) + + def _forget_search(uuid) -> None: """A search is fully done (or abandoned): drop its persisted request and any checkpoint so it is never replayed or resumed again.""" @@ -516,10 +575,18 @@ class Librarian(object): dois = [] for item, value in refined_result.items(): dois.append([item, value]) - result, positions, interrupted = await asyncio.to_thread( - search_bot.search_for_doi, - dois, self.live_results, self.app.logger, SHUTDOWN_EVENT, resume, - ) + # Publish this scan as "the running search" so the heartbeat can report + # it; cleared in finally so a finished/crashed scan never lingers there. + progress = {} + _set_current_search(self.uuid, self.query, progress, self.live_results) + try: + result, positions, interrupted = await asyncio.to_thread( + search_bot.search_for_doi, + dois, self.live_results, self.app.logger, SHUTDOWN_EVENT, resume, + progress, + ) + finally: + _clear_current_search() if interrupted: # Graceful shutdown hit mid-scan: checkpoint found-so-far + per-file # resume offsets + the DOI list, so a restart continues instead of @@ -951,6 +1018,10 @@ if __name__ == "__main__": threads.append( threading.Thread(target=outbox_resender, args=(app.logger,), daemon=True) ) + # "Still searching" heartbeat, so an hours-long scan isn't radio silence. + threads.append( + threading.Thread(target=search_heartbeat, args=(app.logger,), daemon=True) + ) for worker in threads: worker.start() # Re-enqueue searches that were accepted/in-progress before the last stop. diff --git a/conjurer_librarian/search_bot.py b/conjurer_librarian/search_bot.py index ba75e79..dfb1a0a 100644 --- a/conjurer_librarian/search_bot.py +++ b/conjurer_librarian/search_bot.py @@ -247,7 +247,8 @@ def consumer(in_q, control_q, doi, live_results, result_list, control_dict, expe -def search_for_doi(doi, live_results, _logger, stop_event=None, resume=None): +def search_for_doi(doi, live_results, _logger, stop_event=None, resume=None, + progress=None): """Search for DOI in live_results, resumably. Returns ``(result_list, positions, interrupted)``: @@ -262,6 +263,11 @@ def search_for_doi(doi, live_results, _logger, stop_event=None, resume=None): ``resume`` is ``{"positions": {...}, "found": [doi, ...]}`` from a previous interrupted run: already-found DOIs are pre-marked and each producer seeks to its saved offset, so no already-scanned line is read twice. + + ``progress``, if given, is a dict this fills with ``positions`` (the LIVE + dict, updated as producers read) and ``total_bytes`` (summed once, up front). + That makes a rough "how far along" reading free: sum the offsets, divide by + the total - no counting, no extra work in the read loop. """ control_dict = {"sentinels":0} result_list = [] @@ -293,6 +299,19 @@ def search_for_doi(doi, live_results, _logger, stop_event=None, resume=None): ) return result_list, positions, bool(stop_event and stop_event.is_set()) + if progress is not None: + # One stat() per chunk file, ONCE - then progress is just sum(positions) + # / total_bytes, with nothing extra happening per line. + total_bytes = 0 + for name in chunk_files: + try: + total_bytes += os.path.getsize(DATABASE_PATH + name) + except OSError: + pass + progress["positions"] = positions # live dict, updated by the producers + progress["total_bytes"] = total_bytes + progress["chunk_files"] = expected + for i in range (0, (len(doi)//1000)+2): t_cons = Thread( target=consumer, diff --git a/tests/integration/test_librarian_heartbeat.py b/tests/integration/test_librarian_heartbeat.py new file mode 100644 index 0000000..6935772 --- /dev/null +++ b/tests/integration/test_librarian_heartbeat.py @@ -0,0 +1,84 @@ +"""Integration: the 'still searching' heartbeat and its cheap progress estimate. + +The estimate must stay free: producers already record a byte offset per chunk +file and the total is stat()'d once, so a reading is just a sum over ~40 ints. +""" +import logging +import sys +import types + +import pytest + +if "habanero" not in sys.modules: + _habanero = types.ModuleType("habanero") + _habanero.Crossref = object + sys.modules["habanero"] = _habanero + +import conjurer_librarian as lib # noqa: E402 +import search_bot # noqa: E402 + +_LOG = logging.getLogger("test-heartbeat") +_LOG.addHandler(logging.NullHandler()) + + +def test_progress_summary_percentages(): + progress = {"positions": {"0_chunk.txt": 250, "1_chunk.txt": 250}, "total_bytes": 1000} + done, total, percent = lib._progress_summary(progress) + assert (done, total) == (500, 1000) + assert percent == pytest.approx(50.0) + + +def test_progress_summary_unknown_total_is_zero_percent(): + done, total, percent = lib._progress_summary({"positions": {"a": 10}}) + assert (done, total, percent) == (10, 0, 0.0) + + +def test_progress_summary_handles_empty_and_none(): + assert lib._progress_summary(None) == (0, 0, 0.0) + assert lib._progress_summary({}) == (0, 0, 0.0) + + +def test_progress_summary_is_clamped_to_100(): + # A partially-buffered tail can push the summed offsets past the total. + _done, _total, percent = lib._progress_summary( + {"positions": {"a": 1500}, "total_bytes": 1000} + ) + assert percent == pytest.approx(100.0) + + +def test_current_search_registration_round_trip(): + progress = {"positions": {"a": 5}, "total_bytes": 10} + live = [{"DOI": "10.1/x"}] + lib._set_current_search("uuid-1", "kwas foliowy", progress, live) + with lib._current_lock: + snapshot = dict(lib._current_search) + assert snapshot["uuid"] == "uuid-1" + assert snapshot["query"] == "kwas foliowy" + assert lib._progress_summary(snapshot["progress"])[2] == pytest.approx(50.0) + lib._clear_current_search() + with lib._current_lock: + assert not lib._current_search + + +def test_search_fills_progress_with_live_positions_and_total(tmp_path, monkeypatch): + # End to end against the real scan: total_bytes matches the chunk files on + # disk, and once finished the recorded offsets cover them. + monkeypatch.setattr(search_bot, "DATABASE_PATH", str(tmp_path) + "/") + (tmp_path / "0_chunk.txt").write_text("10.1/a\n10.1/b\n", encoding="utf-8") + (tmp_path / "1_chunk.txt").write_text("10.1/c\n", encoding="utf-8") + expected_total = sum( + (tmp_path / name).stat().st_size for name in ("0_chunk.txt", "1_chunk.txt") + ) + + progress = {} + result, _positions, _interrupted = search_bot.search_for_doi( + [("10.1/c", "DATA")], [], _LOG, progress=progress + ) + + assert progress["total_bytes"] == expected_total + assert progress["chunk_files"] == 2 + done, total, percent = lib._progress_summary(progress) + assert total == expected_total + assert done == expected_total # whole DB scanned + assert percent == pytest.approx(100.0) + assert [r for r in result if r["DOI"] == "10.1/c" and r["exists"]]