Librarian: bound the DOI search work queue to stop OOM-killing the pod
CI / compile (pull_request) Successful in 9s
CI / unit (pull_request) Successful in 17s
CI / integration (pull_request) Successful in 26s
build / build (push) Successful in 23s
CI / compile (push) Successful in 7s
CI / unit (push) Successful in 22s
CI / integration (push) Successful in 26s

The pod restarted spontaneously mid-search (no liveness probe is set, so
it was the kernel OOM-killer against the 1Gi limit). Cause: search_bot
built its work queue with maxsize 35_500_000. The producers stream the
WHOLE DOI database (tens of millions of lines across chunks) into it while
a few consumers drain, so the queue could buffer gigabytes of lines -
blowing the 1Gi container and taking the whole in-flight search with it.

Bound the queue (default 100k lines, env CONJURER_LIBRARIAN_WORKQ_SIZE),
so producers backpressure to consumers and RAM stays in the low MB.
Because a bounded queue means a producer can now block on a FULL queue,
make the producer's put timeout-poll the TERM sentinel, so a full queue
whose consumers have already finished (all DOIs found) can never deadlock
it. New test pins that: tiny queue + target on line 1 + thousands of
trailing decoys still terminates and finds the target.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit was merged in pull request #15.
This commit is contained in:
2026-08-02 19:34:56 +02:00
parent 8e18071bb6
commit 40605b959f
2 changed files with 47 additions and 3 deletions
+28 -3
View File
@@ -21,7 +21,7 @@ Global Variables:
# TODO: Wpiemdolić to wszystko w klasę z loggerem przysłanym z góry # TODO: Wpiemdolić to wszystko w klasę z loggerem przysłanym z góry
import os import os
import re import re
from queue import Empty, Queue from queue import Empty, Full, Queue
from threading import Thread from threading import Thread
import time import time
q = Queue() q = Queue()
@@ -44,7 +44,13 @@ CHUNK = os.getenv("CONJURER_LIBRARIAN_CHUNK", "_chunk.txt")
MAXTHREADS = int(os.getenv("CONJURER_LIBRARIAN_MAXTHREADS", "0")) MAXTHREADS = int(os.getenv("CONJURER_LIBRARIAN_MAXTHREADS", "0"))
_sentinel = object() _sentinel = object()
WORK_Q_SIZE = 35500000 # BOUNDED work queue. The producers stream the WHOLE DOI database (potentially
# tens of millions of lines across chunks) into this queue; the previous cap of
# 35_500_000 items was effectively unbounded (~3.5 GB of buffered lines), which
# OOM-killed the 1 GiB container mid-search. A small bound makes the producers
# backpressure to the consumers, keeping RAM to a few MB. The producer put below
# stays responsive to the TERM sentinel so a full queue can never deadlock it.
WORK_Q_SIZE = int(os.getenv("CONJURER_LIBRARIAN_WORKQ_SIZE", "100000"))
# Idle backstop: after this many consecutive empty seconds a consumer assumes # Idle backstop: after this many consecutive empty seconds a consumer assumes
# the producers are done (or dead) and exits, so the search can never hang even # the producers are done (or dead) and exits, so the search can never hang even
# if a sentinel were somehow lost. The primary, correct termination is still the # if a sentinel were somehow lost. The primary, correct termination is still the
@@ -107,7 +113,26 @@ def producer(out_q, control_q, filename, _logger):
print(f"EOF {filename}") print(f"EOF {filename}")
break break
out_q.put(line) # Backpressure-safe put onto the BOUNDED queue: wait for room,
# but keep polling the TERM sentinel so a full queue whose
# consumers have already finished can never deadlock us here.
stopped = False
while True:
try:
out_q.put(line, timeout=1)
break
except Full:
try:
if control_q.get(block=False) is _sentinel:
control_q.put(_sentinel)
stopped = True
break
except Empty:
pass
if stopped:
print("TERM signal received (queue full)")
break
try: try:
check = control_q.get(block=False) check = control_q.get(block=False)
except Empty: except Empty:
+19
View File
@@ -131,3 +131,22 @@ def test_discover_chunk_files_sorted_numerically(tmp_path, monkeypatch):
# Numeric order (10 after 2, not lexicographic), and non-chunk files ignored. # Numeric order (10 after 2, not lexicographic), and non-chunk files ignored.
assert found == ["0_chunk.txt", "1_chunk.txt", "2_chunk.txt", "10_chunk.txt"] assert found == ["0_chunk.txt", "1_chunk.txt", "2_chunk.txt", "10_chunk.txt"]
def test_bounded_queue_does_not_deadlock_on_early_termination(tmp_path, monkeypatch):
# The OOM fix bounds the work queue. That means a producer can block on a
# FULL queue - and if the consumers have already finished (all DOIs found)
# it must notice the TERM sentinel instead of hanging forever. Tiny queue +
# target on the first line + thousands of trailing decoys the producer still
# holds is exactly that situation.
monkeypatch.setattr(search_bot, "DATABASE_PATH", str(tmp_path) + "/")
monkeypatch.setattr(search_bot, "WORK_Q_SIZE", 3) # force the producer to block
target = "10.1234/found.on.line.one"
lines = [target + "\n"] + [f"10.0000/decoy-{i}\n" for i in range(5000)]
(tmp_path / "0_chunk.txt").write_text("".join(lines), encoding="utf-8")
finished, result = _run_bounded([(target, "DATA")], timeout=20)
assert finished, "a full bounded queue deadlocked the producer on early termination"
hit = [r for r in result if r["DOI"] == target and r["exists"]]
assert hit, "the target on the first line should have been found"