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 <noreply@anthropic.com>

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(
+        # Publish this scan as "the running search" so the heartbeat can report
-            search_bot.search_for_doi,
+        # it; cleared in finally so a finished/crashed scan never lingers there.
-            dois, self.live_results, self.app.logger, SHUTDOWN_EVENT, resume,
+        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.

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,

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"]]