Librarian: drop write-only result dumps + tame search logging

Two hygiene fixes on top of the work-queue OOM bound:

Result dumps: cr_results / rr_results / s_results.json were write-only
(nothing reads them) yet accumulated EVERY search forever and json.load'd
the whole growing file on each write - unbounded RAM and PVC growth, and
for a deep search the raw cr_results dump is hundreds of MB. They are now
off by default (CONJURER_LIBRARIAN_DEBUG_DUMPS) and, when enabled, are
overwritten with just the latest search - never loaded or accumulated.
not_in_db.json is untouched: it's a real queue the scraper drains.

Search logging: search_bot logged via print(), including a per-line
carriage-return progress line that flooded stdout / the log file with
millions of entries - fine for a desktop app, unreadable and bloating in
a container. All of it is now proper logging at DEBUG (with coarse
per-500k-line progress), so a normal run is quiet. The librarian log
level is configurable (CONJURER_LIBRARIAN_LOG_LEVEL, default INFO) and a
stdout handler is added so  stays useful now that the
search no longer prints straight to stdout. Set DEBUG for full verbosity.

Also: make test_result_delivery_contract hermetic (point the durable
spool at a temp dir so it can't pollute or be poisoned by the real
result_inbox/ between runs) and gitignore the runtime spool dirs.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.gitignore

@@ -202,3 +202,8 @@ not_in_db.json
 rr_results.json
 s_results.json
 *.bak.DS_Store
+
+# Durable result-delivery spool (runtime, per-deploy)
+result_inbox/
+delivered_uuids/
+outbox/

conjurer_librarian/conjurer_librarian.py

@@ -76,6 +76,30 @@ RESULT_SEND_BACKOFF = float(_env("CONJURER_RESULT_SEND_BACKOFF", "2"))
 OUTBOX_RESEND_SECONDS = int(_env("CONJURER_OUTBOX_RESEND_SECONDS", "60"))
 _outbox = DiskQueue(OUTBOX_DIR)
 
+# cr_results/rr_results/s_results.json are write-only debug dumps (nothing reads
+# them). They used to accumulate EVERY search forever AND json.load the whole
+# growing file on each write - unbounded RAM + disk, and for a deep search the
+# raw dump is hundreds of MB. Off by default now; when explicitly enabled they
+# are overwritten with just the latest search (no load, no accumulation).
+DEBUG_DUMPS = _env("CONJURER_LIBRARIAN_DEBUG_DUMPS", "0").lower() in ("1", "true", "yes")
+# Log level: INFO keeps normal runs readable (the desktop-era per-line/per-file
+# chatter is now DEBUG); set DEBUG to get the full verbosity back.
+LOG_LEVEL = _env("CONJURER_LIBRARIAN_LOG_LEVEL", "INFO").upper()
+
+
+def _dump_debug(path, uuid, data) -> None:
+    """Optionally dump the latest search's data for debugging.
+
+    Overwrites (never accumulates) and does nothing unless DEBUG_DUMPS is on, so
+    it can't grow RAM or the state volume in normal operation."""
+    if not DEBUG_DUMPS:
+        return
+    try:
+        with open(path, "w", encoding="utf-8") as handle:
+            json.dump({uuid: data}, handle)
+    except OSError as exc:
+        logging.getLogger("conjurer_librarian").warning("Debug dump to %s failed: %s", path, exc)
+
 app = Flask(__name__)
 
 librarian_queue = Queue()
@@ -311,20 +335,7 @@ class Librarian(object):
             self.app.logger.info("CROSSREF DONE")
 
             self.app.logger.info("CROSSREF DONE")
-        with open(lib_paths.CR_RESULTS, "r+", encoding="utf-8") as data_file:
+        _dump_debug(lib_paths.CR_RESULTS, self.uuid, self.search_result_from_cr)
-            # First we load existing data into a dict.
-            try:
-                file_data = json.load(data_file)
-            except JSONDecodeError:
-                file_data = {}
-            data_file.truncate(0)
-            data_file.seek(0)
-            tmp = {self.uuid : self.search_result_from_cr}
-            if file_data:
-                file_data.update(tmp)
-            else:
-                file_data = tmp
-            json.dump(file_data, data_file, indent=4)
         return cr_result
 
 
@@ -375,20 +386,7 @@ class Librarian(object):
 
         for item in temp:
             refined_result[item["DOI"]]=  item
-        with open(lib_paths.RR_RESULTS, "r+", encoding="utf-8") as data_file:
+        _dump_debug(lib_paths.RR_RESULTS, self.uuid, refined_result)
-            # First we load existing data into a dict.
-            try:
-                file_data = json.load(data_file)
-            except JSONDecodeError:
-                file_data = {}
-            data_file.truncate(0)
-            data_file.seek(0)
-            tmp = {self.uuid: refined_result}
-            if file_data:
-                file_data.update(tmp)
-            else:
-                file_data = tmp
-            json.dump(file_data, data_file, indent=4)
         return refined_result
 
     async def check_if_exists(self, refined_result):
@@ -568,24 +566,9 @@ class BackgroundTaskSearch(threading.Thread):
                     ndb_file.seek(0)
                     json.dump(ndb_database, ndb_file)
 
-                # Save results to "s_results.json" file
+                # Optional debug dump of the final result (off by default).
-                with open(lib_paths.S_RESULTS, "r+", encoding="utf-8") as s_file:
+                _dump_debug(lib_paths.S_RESULTS, librarian.uuid, result[librarian.uuid])
-                    database = {}
+                self.app.logger.info("Search %s finished", librarian.uuid)
-                    try:
-                        database = json.load(s_file)
-                    except JSONDecodeError:
-                        pass
-                    if database:
-                        self.app.logger.info(database)
-                        self.app.logger.info(result)
-                        database.update(result)
-                    else:
-                        database = result
-                    self.app.logger.info("DUMPING DATA")
-                    s_file.truncate(0)
-                    s_file.seek(0)
-                    json.dump(database, s_file)
-                self.app.logger.info("FINISHED")
 
                 # Persist the result to the durable OUTBOX FIRST, then try to
                 # deliver it. Writing to disk before sending is the whole point:
@@ -741,7 +724,10 @@ async def get_partial():
 
 # =======================================MAIN===================================================
 if __name__ == "__main__":
-    app.logger.setLevel(logging.DEBUG)
+    # Default INFO (readable). Set CONJURER_LIBRARIAN_LOG_LEVEL=DEBUG for the
+    # full per-file / per-line search chatter.
+    app.logger.setLevel(LOG_LEVEL)
+    _fmt = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
     LOGFILE_PATH.parent.mkdir(parents=True, exist_ok=True)
     h1 = handlers.RotatingFileHandler(
         filename=str(LOGFILE_PATH),
@@ -750,8 +736,13 @@ if __name__ == "__main__":
         maxBytes=6 * 1024 * 1024,
         backupCount=6,
     )
-
+    h1.setFormatter(_fmt)
     app.logger.addHandler(h1)
+    # Console handler so `kubectl logs` shows what's happening (k8s reads stdout);
+    # the search internals no longer print() straight to stdout.
+    _console = logging.StreamHandler()
+    _console.setFormatter(_fmt)
+    app.logger.addHandler(_console)
     threads = []
     threads.append(threading.Thread(target=waitress_run, daemon=True))
     # threads.append(threading.Thread(target=flask_debug))

conjurer_librarian/search_bot.py

@@ -102,15 +102,18 @@ def producer(out_q, control_q, filename, _logger):
         # producer partway and leaving every DOI after the bad byte unsearched.
         # DOIs are ASCII, so a replaced byte can only affect junk, never a match.
         with open(DATABASE_PATH + filename, "r", encoding=ENCODING, errors="replace") as operated_file:
-            print(f"Worker {filename} ")
+            _logger.debug("Producer started: %s", filename)
             line_no = 0
             while True:
                 line = operated_file.readline()
                 line_no += 1
-                print(f"\t \t \t \t \t \t W{filename}{line_no}\r", end="")
+                # Coarse progress at DEBUG only - the old per-line carriage-return
+                # print flooded stdout / the log file with millions of lines.
+                if line_no % 500000 == 0:
+                    _logger.debug("Producer %s: %d lines read", filename, line_no)
 
                 if not line:
-                    print(f"EOF {filename}")
+                    _logger.debug("Producer %s: EOF at %d lines", filename, line_no)
                     break
 
                 # Backpressure-safe put onto the BOUNDED queue: wait for room,
@@ -130,7 +133,7 @@ def producer(out_q, control_q, filename, _logger):
                         except Empty:
                             pass
                 if stopped:
-                    print("TERM signal received (queue full)")
+                    _logger.debug("Producer %s: TERM (queue full)", filename)
                     break
 
                 try:
@@ -139,10 +142,10 @@ def producer(out_q, control_q, filename, _logger):
                     check = False
 
                 if check is _sentinel:
-                    print("TERM signal received")
+                    _logger.debug("Producer %s: TERM signal", filename)
                     control_q.put(check)
                     break
-            print(f"Worker finished: {filename}")
+            _logger.debug("Producer finished: %s", filename)
     except Exception as exc:  # pylint: disable=broad-except
         # No per-file error (missing/unreadable chunk, a decode edge case that
         # slips past errors="replace", anything unforeseen) may take the whole
@@ -150,7 +153,6 @@ def producer(out_q, control_q, filename, _logger):
         # the sentinel below still fires (finally), so the consumers' count stays
         # correct and nothing deadlocks or silently loses a producer.
         _logger.warning("Chunk %s failed, skipping rest of it: %s", filename, exc)
-        print(f"Worker {filename} failed: {exc}")
     finally:
         # ALWAYS emit exactly one sentinel per producer, on every exit path (EOF,
         # early TERM, or crash). This is what lets the consumers count producers
@@ -170,7 +172,7 @@ def consumer(in_q, control_q, doi, live_results, result_list, control_dict, expe
         live_results (list): List to store the search results.
         _logger: Logger object for logging.
     """
-    print(f"Consumer thread started: {no} no")
+    _logger.debug("Consumer %s started", no)
     empty_counter = 0
     alive_no = 0
     # DOI -> result item, so a line is matched with one O(1) dict lookup instead
@@ -182,12 +184,14 @@ def consumer(in_q, control_q, doi, live_results, result_list, control_dict, expe
             data = in_q.get(block=True, timeout = 1)
             if data is _sentinel:
                 control_dict["sentinels"] += 1
-                print(f"Workers finished: {control_dict['sentinels']} reported by consumer {no}")
+                _logger.debug(
+                    "Consumer %s: producer done (%d/%d)",
+                    no, control_dict["sentinels"], expected_sentinels,
+                )
 
             else:
                 empty_counter = 0
                 alive_no += 1
-                print(f"C{no}__{alive_no}\r", end="")
 
                 # Each DB line is a DOI (optionally followed by metadata). Match
                 # the WHOLE first token exactly - the old `item["DOI"] in data`
@@ -197,8 +201,8 @@ def consumer(in_q, control_q, doi, live_results, result_list, control_dict, expe
                 line_doi = parts[0] if parts else ""
                 item = doi_index.get(line_doi)
                 if item is not None and not item["exists"]:
-                    print(f"HIT in {no}: {line_doi}")
+                    # HIT can fire thousands of times for a deep search -> DEBUG.
-                    _logger.info("HIT %s", line_doi)
+                    _logger.debug("HIT %s (consumer %s)", line_doi, no)
                     item["exists"] = True
                     live_results.append(item)
                     # All found? Signal producers to stop early (rare -> cheap).
@@ -207,20 +211,18 @@ def consumer(in_q, control_q, doi, live_results, result_list, control_dict, expe
         except Empty:
             empty_counter += 1
             time.sleep(1)
-            print(f"Consumer {no} empty")
             # Order matters: the >EMPTY_LIMIT break must be checked BEFORE the
             # lesser threshold, otherwise (as in the original) the first branch
             # always wins and the break is dead code, leaving the sentinel count
             # as the only exit - which is exactly what used to hang the search.
             if empty_counter > EMPTY_LIMIT:
-                print(f"Consumer thread finished {no} (idle backstop)")
+                _logger.debug("Consumer %s finished (idle backstop)", no)
                 break
             if empty_counter > 5:
-                print(f"Consumer {no} empty lvl 2")
                 time.sleep(2)
 
         if control_dict["sentinels"] >= expected_sentinels:
-            _logger.info(f"All workers finished {no}")
+            _logger.debug("Consumer %s: all producers finished", no)
             break
 
 

tests/integration/test_result_delivery_contract.py

@@ -16,6 +16,7 @@ import time
 import pytest
 
 import communication_subroutine as cs
+from durable_queue import DiskQueue
 
 
 def _drain(queue):
@@ -24,7 +25,11 @@ def _drain(queue):
 
 
 @pytest.fixture
-def comm_threads():
+def comm_threads(tmp_path, monkeypatch):
+    # Point the durable spool at a temp dir so /conjurer's dedup/persist can't
+    # leak into (or be poisoned by) the real result_inbox/ between runs.
+    monkeypatch.setattr(cs, "_inbox", DiskQueue(str(tmp_path / "inbox")))
+    monkeypatch.setattr(cs, "_delivered", DiskQueue(str(tmp_path / "delivered")))
     cs.awaiting_q.clear()
     _drain(cs.incoming_q)
     _drain(cs.OUT_COMM_Q)