diff --git a/communication_subroutine.py b/communication_subroutine.py index 49867c1..0f7827b 100644 --- a/communication_subroutine.py +++ b/communication_subroutine.py @@ -13,6 +13,9 @@ import requests from flask import Flask, abort, jsonify, request from waitress import serve +from constants import DELIVERED_DIR, DELIVERED_MAX, RESULT_INBOX_DIR +from durable_queue import DiskQueue + HOST_ADDRESS = os.getenv("CONJURER_DISCORD_HOST", "192.168.1.92") PORT_ADDRESS = int(os.getenv("CONJURER_DISCORD_PORT", "5000")) ICECAST_ADDRESS = os.getenv("CONJURER_ICECAST", "http://192.168.1.12:8000") @@ -33,9 +36,43 @@ incoming_q = Queue() # records older than this. Kept well above the ping timeout so a slow-but-alive # round-trip is never swept out from under a waiter. PING_TTL_SECONDS = 30 + +# Durable result delivery. Every incoming search result is persisted to _inbox +# before we ack the librarian, and only removed once it has actually been +# rendered to the user (its uuid recorded in _delivered). This makes /conjurer +# idempotent (the librarian can safely resend until acked) and lets an accepted +# result survive a bot restart mid-flight (replayed from _inbox on startup). +_inbox = DiskQueue(RESULT_INBOX_DIR) +_delivered = DiskQueue(DELIVERED_DIR) app = Flask(__name__) +def mark_delivered(query_uuid) -> None: + """Record that a result was rendered to the user. + + After this, the librarian's resends of that uuid are dropped as duplicates + and it is never replayed from the inbox again. Called by the cog once it has + actually posted the result to Discord.""" + _delivered.put(query_uuid, {}) + _delivered.prune(DELIVERED_MAX) + _inbox.remove(query_uuid) + + +def replay_inbox() -> None: + """Re-queue INBOX results that were accepted but not yet rendered. + + Recovers an expensive result that reached the bot (and was acked to the + librarian, so it won't be resent) but whose render was lost to a bot restart. + """ + logger = logging.getLogger("discord") + for query_uuid, payload, _ts in _inbox.items(): + if _delivered.contains(query_uuid): + _inbox.remove(query_uuid) + continue + logger.info("Replaying un-rendered result %s from INBOX", query_uuid) + incoming_q.put(payload) + + def _authorize_request() -> None: """Reject inbound calls lacking the shared key (no-op if key is unset).""" if API_KEY and request.headers.get("X-Conjurer-Api-Key") != API_KEY: @@ -113,11 +150,30 @@ def answer_external_command(): """ _authorize_request() logger = logging.getLogger("discord") - logger.info(request) record = json.loads(request.data) - logger.info(record) - logger.info("DATA RECEIVED") - incoming_q.put(record) + logger.info("DATA RECEIVED: %s", record) + # Health-check pongs are ephemeral - never persisted or deduped. + if isinstance(record, dict) and "__pong__" in record: + incoming_q.put(record) + return jsonify("SUCCESS") + # Search results: idempotent, durable intake. Persist each uuid to the inbox + # before acking, and queue only a uuid we have NOT already delivered or + # accepted. This lets the librarian's resender retry safely (a duplicate is + # dropped, never double-rendered) and lets an accepted-but-unrendered result + # be replayed after a bot restart. + if isinstance(record, dict): + for query_uuid in list(record.keys()): + if _delivered.contains(query_uuid): + logger.info("Result %s already delivered - dropping duplicate", query_uuid) + continue + if _inbox.contains(query_uuid): + logger.info("Result %s already pending - dropping duplicate", query_uuid) + continue + single = {query_uuid: record[query_uuid]} + _inbox.put(query_uuid, single) + incoming_q.put(single) + else: + incoming_q.put(record) # unexpected shape - preserve old behaviour return jsonify("SUCCESS") @@ -405,6 +461,13 @@ def comm_subroutine(stop_event: Optional[threading.Event] = None): for worker in threads: worker.start() + # Recover any result that was accepted before a previous shutdown but never + # rendered - re-queue it now that scan_incoming is running. + try: + replay_inbox() + except Exception: # pylint: disable=broad-exception-caught + logger.exception("INBOX replay on startup failed") + try: while any(thread.is_alive() for thread in threads): if stop_event and stop_event.is_set(): diff --git a/conjurer_librarian/conjurer_librarian.py b/conjurer_librarian/conjurer_librarian.py index 5b27c6c..0b2c666 100644 --- a/conjurer_librarian/conjurer_librarian.py +++ b/conjurer_librarian/conjurer_librarian.py @@ -18,6 +18,7 @@ import json import logging import os import threading +import time from json.decoder import JSONDecodeError from logging import handlers from pathlib import Path @@ -29,6 +30,7 @@ import lib_paths import scrape_bot import search_bot # import search_bot2 as search_bot +from durable_queue import DiskQueue from flask import Flask, jsonify, request, abort from habanero import Crossref from waitress import serve @@ -64,6 +66,16 @@ LOGFILE_PATH = _env_path( "CONJURER_LIBRARIAN_LOG", str(BASE_DIR / "librarian.log") ) +# Durable OUTBOX for finished results. A result is expensive (hours of compute), +# so it is written here and only removed once the bot ACKs it (HTTP 200). Lives +# on the librarian's persistent state volume, so it survives a librarian restart +# and a transient bot outage; the resender thread keeps retrying until delivered. +OUTBOX_DIR = _env("CONJURER_LIBRARIAN_OUTBOX", os.path.join(lib_paths.STATE_DIR, "outbox")) +RESULT_SEND_ATTEMPTS = int(_env("CONJURER_RESULT_SEND_ATTEMPTS", "3")) +RESULT_SEND_BACKOFF = float(_env("CONJURER_RESULT_SEND_BACKOFF", "2")) +OUTBOX_RESEND_SECONDS = int(_env("CONJURER_OUTBOX_RESEND_SECONDS", "60")) +_outbox = DiskQueue(OUTBOX_DIR) + app = Flask(__name__) librarian_queue = Queue() @@ -113,6 +125,63 @@ def _post_pong(app_logger, ping_uuid) -> None: app_logger.warning("PING pong send failed for %s: %s", ping_uuid, exc) +def _deliver_result(uuid, payload, app_logger, attempts=RESULT_SEND_ATTEMPTS) -> bool: + """POST one result to the bot, retrying with backoff. True only on HTTP 200. + + The bot's /conjurer is idempotent (dedups by uuid), so re-POSTing a result + it already has is safe - it just answers 200 again. That is what lets the + OUTBOX keep retrying until the result is truly acknowledged, without ever + double-delivering to the user. + """ + target = f"{MAIN_BOT_ADDRESS}{SEND_RESULTS}" + for attempt in range(1, max(1, attempts) + 1): + try: + response = requests.post( + target, json=payload, headers=_service_headers(), timeout=60 + ) + if response.status_code == 200: + app_logger.info("Result %s delivered (HTTP 200) on attempt %d", uuid, attempt) + return True + app_logger.warning( + "Result %s: bot returned HTTP %s (attempt %d/%d): %s", + uuid, response.status_code, attempt, attempts, response.text[:300], + ) + except requests.exceptions.RequestException as exc: + app_logger.warning( + "Result %s delivery failed (attempt %d/%d): %s", uuid, attempt, attempts, exc + ) + if attempt < attempts: + time.sleep(RESULT_SEND_BACKOFF * attempt) + return False + + +def _resend_once(app_logger) -> None: + """One sweep of the OUTBOX: try to deliver every un-acked result, once each. + + Removes each entry only after a positive ACK, so nothing is dropped until + the bot has it. Corrupt/unreadable entries are skipped by DiskQueue.items(). + """ + for uuid, payload, _ts in _outbox.items(): + if _deliver_result(uuid, payload, app_logger, attempts=1): + _outbox.remove(uuid) + + +def outbox_resender(app_logger) -> None: + """Background loop: periodically flush the OUTBOX until the bot is reachable. + + This is what makes an expensive result survive a transient bot outage or a + librarian restart - on restart the persisted OUTBOX is simply resent.""" + pending = len(_outbox) + if pending: + app_logger.info("OUTBOX has %d un-acked result(s) on startup - will resend", pending) + while True: + try: + _resend_once(app_logger) + except Exception as exc: # pylint: disable=broad-exception-caught + app_logger.exception("OUTBOX resend sweep failed: %s", exc) + time.sleep(OUTBOX_RESEND_SECONDS) + + # trunk-ignore(pylint/R0902) class Librarian(object): """ @@ -503,47 +572,24 @@ class BackgroundTaskSearch(threading.Thread): json.dump(database, s_file) self.app.logger.info("FINISHED") - # Send the result back to the bot. Log EXACTLY what goes out - # (target, uuid, how many DOIs and which) so the librarian log - # makes it plain a result was sent and what was in it. + # Persist the result to the durable OUTBOX FIRST, then try to + # deliver it. Writing to disk before sending is the whole point: + # an expensive (hours-long) result now survives a failed send, a + # bot outage, or a librarian restart - the resender keeps + # retrying until the bot ACKs, and only then is it removed. payload = result # shape: {uuid: {DOI: {"Title": ..., "type": ...}}} + uuid = str(librarian.uuid) hits = payload.get(librarian.uuid, {}) if isinstance(payload, dict) else {} - target = f"{MAIN_BOT_ADDRESS}{SEND_RESULTS}" + _outbox.put(uuid, payload) self.app.logger.info( - "SENDING result for %s to %s: %d DOI(s): %s", - librarian.uuid, - target, - len(hits), - list(hits.keys()), + "SENDING result for %s: %d DOI(s): %s (queued to OUTBOX)", + uuid, 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 search. - try: - response = await asyncio.to_thread( - requests.post, - target, - json=payload, - headers=_service_headers(), - timeout=360, - ) - if response.status_code == 200: - self.app.logger.info( - "SENT result for %s -> HTTP 200 (bot accepted)", librarian.uuid - ) - 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, + if await asyncio.to_thread(_deliver_result, uuid, payload, self.app.logger): + _outbox.remove(uuid) + else: + self.app.logger.warning( + "Result %s not acked yet - left in OUTBOX for the resender", uuid ) except Exception as exc: # pylint: disable=broad-exception-caught # A crashing search must not kill the worker thread (which would @@ -703,6 +749,11 @@ if __name__ == "__main__": target=scrape_bot.scraper, args=(app.logger,), daemon=True ) ) + # Durable delivery: keep flushing the OUTBOX so any result not yet acked by + # the bot (transient outage, or left over from before a restart) is resent. + threads.append( + threading.Thread(target=outbox_resender, args=(app.logger,), daemon=True) + ) i = 0 try: for worker in threads: diff --git a/constants.py b/constants.py index 686c21d..9602697 100644 --- a/constants.py +++ b/constants.py @@ -230,6 +230,21 @@ MEMORY_COMPACT_THRESHOLD = int(os.getenv("CONJURER_MEMORY_COMPACT_THRESHOLD", "4 MEMORY_KEEP_RECENT = int(os.getenv("CONJURER_MEMORY_KEEP_RECENT", "200")) MEMORY_COMPACT_HOURS = float(os.getenv("CONJURER_MEMORY_COMPACT_HOURS", "6")) +# Durable result-delivery spool (librarian -> bot). The bot persists every +# incoming search result to RESULT_INBOX_DIR before acking and only forgets it +# once rendered (uuid recorded in DELIVERED_DIR), so an expensive (hours-long) +# result survives a bot restart mid-flight and duplicate resends are idempotent. +# Rooted under CONJURER_DATA_DIR when set (a mounted volume), else next to the +# log file. DELIVERED_MAX bounds the remembered-uuid set. +_STATE_ROOT = _DATA_DIR or (os.path.dirname(LOGFILE) or ".") +RESULT_INBOX_DIR = os.getenv( + "CONJURER_RESULT_INBOX", os.path.join(_STATE_ROOT, "result_inbox") +) +DELIVERED_DIR = os.getenv( + "CONJURER_DELIVERED_DIR", os.path.join(_STATE_ROOT, "delivered_uuids") +) +DELIVERED_MAX = int(os.getenv("CONJURER_DELIVERED_MAX", "10000")) + FILE_SERVICE_ADDRESS = os.getenv("CONJURER_FILE_SERVICE", "http://192.168.1.15:5000") RADIO_HARBOR_ADDRESS = os.getenv("CONJURER_RADIO_HARBOR", "http://192.168.1.15:54321") # Betoniarka (radio-operator service colocated with Liquidsoap). Falls back to diff --git a/docker/Dockerfile.librarian b/docker/Dockerfile.librarian index 327532b..01aced9 100644 --- a/docker/Dockerfile.librarian +++ b/docker/Dockerfile.librarian @@ -14,6 +14,9 @@ RUN pip install --no-cache-dir --upgrade pip \ && pip install --no-cache-dir -r requirements_librarian.txt requests COPY conjurer_librarian/ ./ +# durable_queue lives at the repo root and is shared with the bot; the librarian +# imports it for the durable result OUTBOX. +COPY durable_queue.py ./ ENV PYTHONUNBUFFERED=1 \ CONJURER_LIBRARIAN_HOST=0.0.0.0 \ diff --git a/durable_queue.py b/durable_queue.py new file mode 100644 index 0000000..671a85e --- /dev/null +++ b/durable_queue.py @@ -0,0 +1,104 @@ +"""Dependency-free, disk-backed queue for durable message delivery. + +One JSON file per key under a directory. Used on both sides of the +librarian <-> bot result path so an expensive (hours-long) search result is +never lost to a transient network failure or a restart: + +* the librarian keeps a result in its OUTBOX until the bot acks it, +* the bot keeps a result in its INBOX until it is actually rendered, and + remembers delivered uuids so duplicate resends are idempotent. + +Only ``json`` + ``os`` are imported, so the logic is unit-testable without +flask, discord, or the network. Writes are atomic (temp file + ``os.replace``) +so a crash mid-write can never leave a half-written record that poisons replay. +""" +import json +import os +import tempfile +import time + + +def _safe_name(key: str) -> str: + """Filesystem-safe file stem for a key (uuids are safe; be defensive).""" + stem = "".join(c for c in str(key) if c.isalnum() or c in "-_.") + return stem or "_" + + +class DiskQueue: + """A directory of ``.json`` records, each ``{key, payload, ts}``.""" + + def __init__(self, directory: str): + # No disk touch here on purpose: constructing a DiskQueue at import time + # must not create directories (tests, read-only default paths). The + # directory is created lazily on the first put(). + self.directory = directory + + def _path(self, key) -> str: + return os.path.join(self.directory, _safe_name(key) + ".json") + + def put(self, key, payload) -> None: + """Atomically write (overwrite) the record for ``key``.""" + os.makedirs(self.directory, exist_ok=True) + record = {"key": str(key), "payload": payload, "ts": time.time()} + fd, tmp = tempfile.mkstemp(dir=self.directory, suffix=".tmp") + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump(record, handle) + os.replace(tmp, self._path(key)) # atomic on POSIX + finally: + if os.path.exists(tmp): + os.remove(tmp) + + def contains(self, key) -> bool: + return os.path.exists(self._path(key)) + + def remove(self, key) -> None: + try: + os.remove(self._path(key)) + except FileNotFoundError: + pass + + def items(self): + """Return ``[(key, payload, ts), ...]`` oldest-first. + + Unreadable / half-written / corrupt files are skipped (never raise), + so one bad file can't stall replay of the rest. + """ + out = [] + try: + names = os.listdir(self.directory) + except FileNotFoundError: + return out + for name in names: + if not name.endswith(".json"): + continue + try: + with open(os.path.join(self.directory, name), encoding="utf-8") as handle: + record = json.load(handle) + out.append((record["key"], record["payload"], record.get("ts", 0))) + except (OSError, ValueError, KeyError, TypeError): + continue + out.sort(key=lambda triple: triple[2]) + return out + + def keys(self): + return [key for key, _payload, _ts in self.items()] + + def __len__(self) -> int: + return len(self.items()) + + def prune(self, max_entries: int) -> int: + """Keep only the newest ``max_entries`` (by ts); drop the rest. + + Used for the delivered-uuid set so it cannot grow without bound. + Returns how many were dropped. + """ + if max_entries < 0: + return 0 + entries = self.items() # oldest first + excess = len(entries) - max_entries + dropped = 0 + for key, _payload, _ts in entries[: max(0, excess)]: + self.remove(key) + dropped += 1 + return dropped diff --git a/librarian_commands.py b/librarian_commands.py index 9cf195c..85fd80d 100644 --- a/librarian_commands.py +++ b/librarian_commands.py @@ -15,7 +15,13 @@ import requests from discord.ext import commands, tasks from ai_functions import handle_response -from communication_subroutine import IN_COMM_Q, OUT_COMM_Q, QueryControl, submit_ai_query +from communication_subroutine import ( + IN_COMM_Q, + OUT_COMM_Q, + QueryControl, + mark_delivered, + submit_ai_query, +) from constants import ( DIR_PATH_SADOX, LIBRARIAN_SERVICE_ADDRESS, @@ -169,6 +175,12 @@ class DataModule(commands.Cog): await ctx.send(message) message = "" + # The result is now on screen: mark it delivered so the + # librarian's resends become no-ops and it is dropped from the + # durable inbox (never replayed again). Done after the core + # render but before the optional AI review, which is a bonus. + mark_delivered(str(fresh_data.uuid)) + # Optional AI pass: re-rank the (already Crossref-relevance- # sorted) DOI list and review the sources. Enqueued to the AI # worker so it runs on whatever backend $gadaj_teraz selected; diff --git a/tests/integration/test_librarian_outbox.py b/tests/integration/test_librarian_outbox.py new file mode 100644 index 0000000..dc472a8 --- /dev/null +++ b/tests/integration/test_librarian_outbox.py @@ -0,0 +1,99 @@ +"""Integration: the librarian's durable result OUTBOX + retrying delivery. + +An 8-hour search result must not be lost to a transient bot outage. The result +is written to the OUTBOX before sending; delivery retries with backoff; the +entry is removed only on a positive ACK; and the resender keeps flushing the +OUTBOX (across restarts, since it is on the persistent state volume). +""" +import logging +import sys +import types + +import pytest + +# conjurer_librarian imports `from habanero import Crossref` at import time; the +# integration job doesn't install habanero. Stub it (we never build a real +# Librarian here). +if "habanero" not in sys.modules: + _habanero = types.ModuleType("habanero") + _habanero.Crossref = object + sys.modules["habanero"] = _habanero + +import conjurer_librarian as lib # noqa: E402 +from durable_queue import DiskQueue # noqa: E402 + +_LOG = logging.getLogger("test-outbox") +_LOG.addHandler(logging.NullHandler()) + + +class _Resp: + def __init__(self, status_code, text=""): + self.status_code = status_code + self.text = text + + +@pytest.fixture(autouse=True) +def _no_backoff(monkeypatch): + # Never actually sleep during retry backoff in tests. + monkeypatch.setattr(lib.time, "sleep", lambda _s: None) + + +@pytest.fixture +def outbox(tmp_path, monkeypatch): + box = DiskQueue(str(tmp_path / "outbox")) + monkeypatch.setattr(lib, "_outbox", box) + return box + + +def test_deliver_succeeds_first_try(monkeypatch): + calls = [] + monkeypatch.setattr(lib.requests, "post", lambda *a, **k: calls.append(1) or _Resp(200)) + assert lib._deliver_result("u1", {"u1": {}}, _LOG, attempts=3) is True + assert len(calls) == 1 # no needless retries after a 200 + + +def test_deliver_retries_then_succeeds(monkeypatch): + responses = iter([_Resp(503), _Resp(500), _Resp(200)]) + monkeypatch.setattr(lib.requests, "post", lambda *a, **k: next(responses)) + assert lib._deliver_result("u2", {"u2": {}}, _LOG, attempts=3) is True + + +def test_deliver_returns_false_when_all_attempts_fail(monkeypatch): + def boom(*_a, **_k): + raise lib.requests.exceptions.RequestException("bot down") + + monkeypatch.setattr(lib.requests, "post", boom) + assert lib._deliver_result("u3", {"u3": {}}, _LOG, attempts=2) is False + + +def test_resend_once_removes_only_acked_entries(outbox, monkeypatch): + outbox.put("ok", {"ok": {}}) + outbox.put("bad", {"bad": {}}) + + def fake_deliver(query_uuid, _payload, _logger, attempts=1): + return query_uuid == "ok" + + monkeypatch.setattr(lib, "_deliver_result", fake_deliver) + lib._resend_once(_LOG) + + assert not outbox.contains("ok") # acked -> dropped + assert outbox.contains("bad") # not acked -> kept for the next sweep + + +def test_resend_keeps_result_until_bot_recovers(outbox, monkeypatch): + # Simulate: bot down for the first sweep, up for the second. The result must + # survive the outage and be delivered on recovery. + outbox.put("u9", {"u9": {"10.1/x": {"Title": ["P"], "type": "article"}}}) + state = {"up": False} + + def flaky_post(*_a, **_k): + return _Resp(200) if state["up"] else _Resp(502) + + monkeypatch.setattr(lib.requests, "post", flaky_post) + + lib._resend_once(_LOG) # bot down + assert outbox.contains("u9") # preserved, not lost + + state["up"] = True + lib._resend_once(_LOG) # bot recovered + assert not outbox.contains("u9") # now delivered and cleared diff --git a/tests/integration/test_result_durable_delivery.py b/tests/integration/test_result_durable_delivery.py new file mode 100644 index 0000000..d9210b2 --- /dev/null +++ b/tests/integration/test_result_durable_delivery.py @@ -0,0 +1,92 @@ +"""Integration: the bot's DURABLE, idempotent result intake (/conjurer). + +The expensive-result guarantees on the bot side: +* every result is persisted to the inbox before it is acked, +* a resend of a not-yet-delivered result is dropped (no double render), +* once rendered (mark_delivered) further resends are dropped and it leaves the + inbox, +* on startup, an accepted-but-unrendered result is replayed from the inbox, +* health-check pongs are never persisted. +""" +import pytest + +import communication_subroutine as cs +from durable_queue import DiskQueue + + +@pytest.fixture +def spool(tmp_path, monkeypatch): + inbox = DiskQueue(str(tmp_path / "inbox")) + delivered = DiskQueue(str(tmp_path / "delivered")) + monkeypatch.setattr(cs, "_inbox", inbox) + monkeypatch.setattr(cs, "_delivered", delivered) + cs.API_KEY = None + while not cs.incoming_q.empty(): + cs.incoming_q.get() + return inbox, delivered + + +def _drain_incoming(): + out = [] + while not cs.incoming_q.empty(): + out.append(cs.incoming_q.get()) + return out + + +def _payload(uuid): + return {uuid: {"10.1000/xyz": {"Title": ["A Real Paper"], "type": "journal-article"}}} + + +def test_result_is_persisted_then_queued(spool): + inbox, _delivered = spool + resp = cs.app.test_client().post("/conjurer", json=_payload("u1")) + assert resp.status_code == 200 + assert inbox.contains("u1") # durable before ack + assert _drain_incoming() == [_payload("u1")] + + +def test_resend_while_pending_is_not_requeued(spool): + client = cs.app.test_client() + client.post("/conjurer", json=_payload("u2")) + _drain_incoming() # consume the first queueing + # Resend before it was rendered: inbox still holds it -> dropped, not doubled. + client.post("/conjurer", json=_payload("u2")) + assert _drain_incoming() == [] + + +def test_resend_after_delivery_is_dropped(spool): + inbox, delivered = spool + client = cs.app.test_client() + client.post("/conjurer", json=_payload("u3")) + _drain_incoming() + cs.mark_delivered("u3") + assert not inbox.contains("u3") + assert delivered.contains("u3") + # A late resend of an already-delivered result must not re-render. + client.post("/conjurer", json=_payload("u3")) + assert _drain_incoming() == [] + + +def test_replay_requeues_only_undelivered(spool): + inbox, delivered = spool + inbox.put("u4", _payload("u4")) + inbox.put("u5", _payload("u5")) + delivered.put("u5", {}) # u5 already shown to the user + cs.replay_inbox() + keys = [list(p.keys())[0] for p in _drain_incoming()] + assert keys == ["u4"] # only the un-rendered one replayed + assert not inbox.contains("u5") # the delivered one is cleaned from the inbox + + +def test_empty_result_still_persisted_and_delivered(spool): + inbox, _delivered = spool + cs.app.test_client().post("/conjurer", json={"u6": {}}) + assert inbox.contains("u6") + assert _drain_incoming() == [{"u6": {}}] + + +def test_pong_is_not_persisted(spool): + inbox, _delivered = spool + cs.app.test_client().post("/conjurer", json={"__pong__": "ping-1"}) + assert len(inbox) == 0 + assert _drain_incoming() == [{"__pong__": "ping-1"}] diff --git a/tests/unit/test_durable_queue.py b/tests/unit/test_durable_queue.py new file mode 100644 index 0000000..78b9a1e --- /dev/null +++ b/tests/unit/test_durable_queue.py @@ -0,0 +1,76 @@ +"""Unit tests for the disk-backed durable queue used by result delivery.""" +import json + +from durable_queue import DiskQueue + + +def test_put_contains_remove(tmp_path): + q = DiskQueue(str(tmp_path / "q")) + assert not q.contains("a") + q.put("a", {"hello": 1}) + assert q.contains("a") + q.remove("a") + assert not q.contains("a") + q.remove("a") # idempotent - no error on missing + + +def test_put_overwrites_and_roundtrips_payload(tmp_path): + q = DiskQueue(str(tmp_path / "q")) + q.put("uuid-1", {"uuid-1": {"10.1/x": {"Title": ["P"], "type": "article"}}}) + q.put("uuid-1", {"uuid-1": {"changed": True}}) + items = q.items() + assert len(items) == 1 + key, payload, _ts = items[0] + assert key == "uuid-1" + assert payload == {"uuid-1": {"changed": True}} + + +def test_items_sorted_oldest_first(tmp_path, monkeypatch): + q = DiskQueue(str(tmp_path / "q")) + import durable_queue + + times = iter([100.0, 200.0, 300.0]) + monkeypatch.setattr(durable_queue.time, "time", lambda: next(times)) + q.put("c", {}) + q.put("a", {}) + q.put("b", {}) + assert [k for k, _p, _ts in q.items()] == ["c", "a", "b"] + + +def test_corrupt_file_is_skipped_not_fatal(tmp_path): + directory = tmp_path / "q" + q = DiskQueue(str(directory)) + q.put("good", {"ok": 1}) + (directory / "broken.json").write_text("{ this is not json", encoding="utf-8") + keys = q.keys() + assert keys == ["good"] # broken file skipped, good one survives + + +def test_prune_keeps_newest(tmp_path, monkeypatch): + q = DiskQueue(str(tmp_path / "q")) + import durable_queue + + times = iter([1.0, 2.0, 3.0, 4.0, 5.0]) + monkeypatch.setattr(durable_queue.time, "time", lambda: next(times)) + for key in ("k1", "k2", "k3", "k4", "k5"): + q.put(key, {}) + dropped = q.prune(2) + assert dropped == 3 + assert set(q.keys()) == {"k4", "k5"} + + +def test_atomic_write_leaves_no_tmp_files(tmp_path): + directory = tmp_path / "q" + q = DiskQueue(str(directory)) + q.put("a", {"x": 1}) + leftover = [p.name for p in directory.iterdir() if p.suffix == ".tmp"] + assert leftover == [] + + +def test_key_with_slashes_is_sanitised(tmp_path): + q = DiskQueue(str(tmp_path / "q")) + q.put("../../etc/passwd", {"evil": 1}) + # Stays inside the directory (no traversal), and round-trips by key. + files = list((tmp_path / "q").iterdir()) + assert all(f.parent == tmp_path / "q" for f in files) + assert q.items()[0][1] == {"evil": 1}