librarian: toleruj złe bajty w chunkach + jawne logowanie wysyłki wyników #4

Merged
gitea merged 1 commits from librarian-unicode-tolerance into main 2026-07-30 15:22:41 +00:00
3 changed files with 71 additions and 17 deletions
+38 -8
View File
@@ -451,18 +451,48 @@ class BackgroundTaskSearch(threading.Thread):
json.dump(database, s_file) json.dump(database, s_file)
self.app.logger.info("FINISHED") self.app.logger.info("FINISHED")
self.app.logger.info(result) # Send the result back to the bot. Log EXACTLY what goes out (target,
coroutine = asyncio.to_thread( # uuid, how many DOIs and which) so the librarian log makes it plain a
# result was sent and what was in it.
payload = result # shape: {uuid: {DOI: {"Title": ..., "type": ...}}}
hits = payload.get(librarian.uuid, {}) if isinstance(payload, dict) else {}
target = f"{MAIN_BOT_ADDRESS}{SEND_RESULTS}"
self.app.logger.info(
"SENDING result for %s to %s: %d DOI(s): %s",
librarian.uuid,
target,
len(hits),
list(hits.keys()),
)
# A failed send must NOT kill this worker - otherwise a bot that is
# momentarily down stalls every future query until the librarian is
# restarted. Log and carry on to the next queued search.
try:
response = await asyncio.to_thread(
requests.post, requests.post,
f"{MAIN_BOT_ADDRESS}{SEND_RESULTS}", target,
json=result, json=payload,
headers=_service_headers(), headers=_service_headers(),
timeout=360, timeout=360,
) )
self.app.logger.info("SENT") if response.status_code == 200:
result = await coroutine self.app.logger.info(
self.app.logger.info(result.status_code) "SENT result for %s -> HTTP 200 (bot accepted)", librarian.uuid
self.app.logger.info("SEND CONFIRMED") )
else:
self.app.logger.warning(
"SENT result for %s but bot returned HTTP %s: %s",
librarian.uuid,
response.status_code,
response.text[:500],
)
except requests.exceptions.RequestException as exc:
self.app.logger.error(
"FAILED to send result for %s to %s: %s",
librarian.uuid,
target,
exc,
)
await asyncio.sleep(1) await asyncio.sleep(1)
+13 -6
View File
@@ -90,7 +90,12 @@ def producer(out_q, control_q, filename, _logger):
_logger: Logger object for logging. _logger: Logger object for logging.
""" """
try: try:
with open(DATABASE_PATH + filename, "r", encoding=ENCODING) as operated_file: # errors="replace" so a stray non-UTF-8 byte in a chunk (they happen in
# scraped DOI dumps) becomes U+FFFD instead of raising UnicodeDecodeError
# mid-file. Without it the readline() below would blow up, killing the
# 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} ") print(f"Worker {filename} ")
line_no = 0 line_no = 0
while True: while True:
@@ -113,11 +118,13 @@ def producer(out_q, control_q, filename, _logger):
control_q.put(check) control_q.put(check)
break break
print(f"Worker finished: {filename}") print(f"Worker finished: {filename}")
except OSError as exc: except Exception as exc: # pylint: disable=broad-except
# A missing or unreadable chunk must not take the whole search down with # No per-file error (missing/unreadable chunk, a decode edge case that
# it - log and move on. The sentinel below still fires (finally), so the # slips past errors="replace", anything unforeseen) may take the whole
# consumers' count stays correct and nothing deadlocks. # search down or crash the thread with a traceback. Log it and move on;
_logger.warning("Chunk %s unreadable, skipping: %s", filename, exc) # 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}") print(f"Worker {filename} failed: {exc}")
finally: finally:
# ALWAYS emit exactly one sentinel per producer, on every exit path (EOF, # ALWAYS emit exactly one sentinel per producer, on every exit path (EOF,
+17
View File
@@ -77,6 +77,23 @@ def test_no_chunks_returns_immediately(tmp_path, monkeypatch):
assert result == [{"DOI": "10.0/x", "exists": False, "data": "DATA"}] assert result == [{"DOI": "10.0/x", "exists": False, "data": "DATA"}]
def test_survives_invalid_utf8_byte_and_still_finds_later_doi(tmp_path, monkeypatch):
# A chunk with a stray non-UTF-8 byte (0x96, the one from the field report)
# must not crash the producer or abort the file mid-read: DOIs AFTER the bad
# byte still have to be found.
monkeypatch.setattr(search_bot, "DATABASE_PATH", str(tmp_path) + "/")
target = "10.1234/after.the.bad.byte"
(tmp_path / "0_chunk.txt").write_bytes(
b"10.0000/before\n" + b"\x96 broken \x96 line \x96\n" + target.encode() + b"\n"
)
finished, result = _run_bounded([(target, "DATA")])
assert finished, "an invalid UTF-8 byte hung or crashed the search"
hit = [r for r in result if r["DOI"] == target and r["exists"]]
assert hit, "DOI after the bad byte was not found - the file was aborted mid-read"
def test_discover_chunk_files_sorted_numerically(tmp_path, monkeypatch): def test_discover_chunk_files_sorted_numerically(tmp_path, monkeypatch):
monkeypatch.setattr(search_bot, "DATABASE_PATH", str(tmp_path) + "/") monkeypatch.setattr(search_bot, "DATABASE_PATH", str(tmp_path) + "/")
for n in (0, 2, 10, 1): for n in (0, 2, 10, 1):