From c5643aa28f1b65637a68506dd278fcd7729418fc Mon Sep 17 00:00:00 2001 From: Polish Hammer Date: Sun, 2 Aug 2026 20:19:49 +0200 Subject: [PATCH] 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 --- .gitignore | 5 ++ conjurer_librarian/conjurer_librarian.py | 87 +++++++++---------- conjurer_librarian/search_bot.py | 34 ++++---- .../test_result_delivery_contract.py | 7 +- 4 files changed, 68 insertions(+), 65 deletions(-) diff --git a/.gitignore b/.gitignore index c3ac2f8..5cc5244 100644 --- a/.gitignore +++ b/.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/ diff --git a/conjurer_librarian/conjurer_librarian.py b/conjurer_librarian/conjurer_librarian.py index 7dc7f79..376fa87 100644 --- a/conjurer_librarian/conjurer_librarian.py +++ b/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: - # 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) + _dump_debug(lib_paths.CR_RESULTS, self.uuid, self.search_result_from_cr) 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: - # 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) + _dump_debug(lib_paths.RR_RESULTS, self.uuid, refined_result) 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 - with open(lib_paths.S_RESULTS, "r+", encoding="utf-8") as s_file: - database = {} - 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") + # Optional debug dump of the final result (off by default). + _dump_debug(lib_paths.S_RESULTS, librarian.uuid, result[librarian.uuid]) + self.app.logger.info("Search %s finished", librarian.uuid) # 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)) diff --git a/conjurer_librarian/search_bot.py b/conjurer_librarian/search_bot.py index 831f0db..214e455 100644 --- a/conjurer_librarian/search_bot.py +++ b/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}") - _logger.info("HIT %s", line_doi) + # HIT can fire thousands of times for a deep search -> DEBUG. + _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 diff --git a/tests/integration/test_result_delivery_contract.py b/tests/integration/test_result_delivery_contract.py index 13e474f..69af460 100644 --- a/tests/integration/test_result_delivery_contract.py +++ b/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)