Librarian: graceful shutdown with resumable search state

A restart of the librarian used to throw away an in-flight search (and any
searches still queued). Now search state survives a restart:

* Resumable DB scan (search_bot): each producer records a tell()-cookie
  watermark per chunk file as it goes (safe because search_for_doi drains
  the work queue before returning), and can seek back to it. search_for_doi
  now takes stop_event + resume and returns (result_list, positions,
  interrupted).

* Persisted requests: /query writes the accepted request to a disk queue
  before enqueuing; replay_requests re-enqueues unfinished ones on startup.
  So even a search still waiting in the queue survives a restart.

* Checkpoints: when a graceful shutdown interrupts a scan, the librarian
  writes {dois, found-so-far, per-file offsets}. On restart answer_query
  loads it, skips the (already done) Crossref+refine, and continues the
  scan from the saved offsets with the found DOIs pre-marked - no line is
  read twice and none is missed. A finished or crashed search forgets its
  request+checkpoint (no poison-pill replay).

* Graceful shutdown: SIGTERM/SIGINT set a shutdown event; the running scan
  checkpoints and the worker stops. The main thread then exits within a
  BOUNDED window (CONJURER_LIBRARIAN_GRACEFUL_TIMEOUT, default 45s) so the
  pod can never become an un-killable zombie. Needs terminationGracePeriod
  >= that in the deploy (separate PR).

Tests: search_bot resume correctness (seek past scanned, don't miss/re-scan;
stop_event -> interrupted) and librarian state mechanics (request replay,
forget, checkpoint round-trip, poison-pill drop). Suite: 58 unit + 49
integration green.

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

conjurer_librarian/search_bot.py

@@ -85,15 +85,17 @@ def discover_chunk_files(_logger):
     return ordered
 
 
-def producer(out_q, control_q, filename, _logger):
-    """
-    Produces items from the output queue and puts them into the control queue.
+def producer(out_q, control_q, filename, _logger, stop_event=None, positions=None,
+             start_offsets=None):
+    """Stream a chunk file's lines onto the work queue, resumably.
 
-    Args:
-        out_q (Queue): Output queue.
-        control_q (Queue): Control queue.
-        filename (str): Name of the file.
-        _logger: Logger object for logging.
+    ``start_offsets[filename]`` (a tell() cookie) is where to RESUME reading from
+    - so a search continued after a restart skips the part already scanned.
+    ``positions[filename]`` is updated to the tell() cookie just PAST each line
+    successfully enqueued; because search_for_doi drains the queue before it
+    returns, that cookie is a safe "everything up to here is processed" watermark
+    to checkpoint. ``stop_event`` (graceful shutdown) makes the producer stop
+    reading and record its watermark, mirroring the early-TERM path.
     """
     try:
         # errors="replace" so a stray non-UTF-8 byte in a chunk (they happen in
@@ -102,9 +104,16 @@ 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:
-            _logger.debug("Producer started: %s", filename)
+            if start_offsets and filename in start_offsets:
+                operated_file.seek(start_offsets[filename])
+                _logger.debug("Producer %s: resuming at offset %s", filename, start_offsets[filename])
+            else:
+                _logger.debug("Producer started: %s", filename)
             line_no = 0
             while True:
+                if stop_event is not None and stop_event.is_set():
+                    _logger.debug("Producer %s: stop requested (graceful)", filename)
+                    break
                 line = operated_file.readline()
                 line_no += 1
                 # Coarse progress at DEBUG only - the old per-line carriage-return
@@ -113,11 +122,15 @@ def producer(out_q, control_q, filename, _logger):
                     _logger.debug("Producer %s: %d lines read", filename, line_no)
 
                 if not line:
+                    # EOF: record the end offset so a resume seeks here and stops
+                    # immediately (the file is fully scanned).
+                    if positions is not None:
+                        positions[filename] = operated_file.tell()
                     _logger.debug("Producer %s: EOF at %d lines", filename, line_no)
                     break
 
-                # Backpressure-safe put onto the BOUNDED queue: wait for room,
-                # but keep polling the TERM sentinel so a full queue whose
+                # Backpressure-safe put onto the BOUNDED queue: wait for room, but
+                # keep polling stop_event / the TERM sentinel so a full queue whose
                 # consumers have already finished can never deadlock us here.
                 stopped = False
                 while True:
@@ -125,6 +138,9 @@ def producer(out_q, control_q, filename, _logger):
                         out_q.put(line, timeout=1)
                         break
                     except Full:
+                        if stop_event is not None and stop_event.is_set():
+                            stopped = True
+                            break
                         try:
                             if control_q.get(block=False) is _sentinel:
                                 control_q.put(_sentinel)
@@ -133,8 +149,12 @@ def producer(out_q, control_q, filename, _logger):
                         except Empty:
                             pass
                 if stopped:
-                    _logger.debug("Producer %s: TERM (queue full)", filename)
+                    _logger.debug("Producer %s: stop while enqueuing", filename)
                     break
+                # Watermark AFTER a successful enqueue: safe to resume past here
+                # once the queue drains (which it does before search_for_doi ends).
+                if positions is not None:
+                    positions[filename] = operated_file.tell()
 
                 try:
                     check = control_q.get(block=False)
@@ -227,23 +247,41 @@ def consumer(in_q, control_q, doi, live_results, result_list, control_dict, expe
 
 
 
-def search_for_doi(doi, live_results, _logger):
-    """
-    Search for DOI in live_results using _logger for logging.
+def search_for_doi(doi, live_results, _logger, stop_event=None, resume=None):
+    """Search for DOI in live_results, resumably.
 
-    Args:
-        doi (list): List of DOI to search for.
-        live_results (list): List to store the search results.
-        _logger: Logger object for logging.
+    Returns ``(result_list, positions, interrupted)``:
+    * ``result_list`` - the queried DOIs with their ``exists`` flag,
+    * ``positions`` - ``{filename: tell()-cookie}`` safe-to-resume watermarks
+      (the queue is drained before return, so everything up to each cookie is
+      processed),
+    * ``interrupted`` - True if ``stop_event`` fired (the scan is PARTIAL; check
+      point ``positions`` + the found DOIs and call again with ``resume=`` to
+      continue where it left off).
+
+    ``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.
     """
     control_dict = {"sentinels":0}
     result_list = []
     threads = []
     work_q = Queue(maxsize=WORK_Q_SIZE)
     control_q = Queue()
+    positions = {}
+
+    resume = resume or {}
+    already_found = set(resume.get("found", []))
+    start_offsets = resume.get("positions", {})
 
     for item in doi:
-        result_list.append({"DOI": item[0], "exists": False, "data": item[1]})
+        entry = {"DOI": item[0], "exists": False, "data": item[1]}
+        if item[0] in already_found:
+            # Pre-mark hits from the previous (interrupted) run so we neither
+            # re-scan for them nor drop them from live_results.
+            entry["exists"] = True
+            live_results.append(entry)
+        result_list.append(entry)
 
     # One producer per chunk file that actually exists; the sentinel threshold is
     # that same count, so the two can never drift apart the way MAXTHREADS did.
@@ -253,25 +291,29 @@ def search_for_doi(doi, live_results, _logger):
         _logger.error(
             "No '<n>%s' chunk files in %s - DOI search cannot run", CHUNK, DATABASE_PATH
         )
-        return result_list
+        return result_list, positions, bool(stop_event and stop_event.is_set())
 
     for i in range (0, (len(doi)//1000)+2):
         t_cons = Thread(
             target=consumer,
             args=(work_q, control_q, doi, live_results, result_list, control_dict, expected, i, _logger),
         )
-        _logger.info("Consumer thread created")
+        _logger.debug("Consumer thread created")
         threads.append(t_cons)
     for filename in chunk_files:
-        _logger.info("Creating worker thread for %s", filename)
+        _logger.debug("Creating worker thread for %s", filename)
         threads.append(
-            Thread(target=producer, args=(work_q, control_q, filename, _logger))
+            Thread(
+                target=producer,
+                args=(work_q, control_q, filename, _logger, stop_event, positions, start_offsets),
+            )
         )
     for worker in threads:
         worker.start()
     for worker in threads:
         worker.join()
-    return result_list
+    interrupted = bool(stop_event and stop_event.is_set())
+    return result_list, positions, interrupted
 
 
 if __name__ == "__main__":