gitea
44b7298a15
An 8h search result must survive a transient bot outage, an api/address misroute, or a restart of either side. Make the librarian->bot result path durably at-least-once with idempotent rendering: Shared: durable_queue.DiskQueue - a dependency-free, atomically-written, one-file-per-key disk queue (unit-tested), shared by both images (added to Dockerfile.librarian; the bot already COPYs *.py). Librarian (sender): finished results go to a persistent OUTBOX before sending; delivery retries with backoff; an entry is removed only on a positive ACK; a resender thread keeps flushing the OUTBOX, so a result survives a bot outage AND a librarian restart (OUTBOX is on the state volume) - it simply keeps trying until acked. Bot (receiver): /conjurer is now idempotent and durable - each result is persisted to an INBOX before acking and only queued if its uuid was not already delivered (dropped as a duplicate) or already pending. Once the cog actually renders it, mark_delivered() records the uuid and clears the inbox, so the librarian's resends become no-ops. On startup the bot replays any accepted-but-unrendered result from the INBOX, so a bot crash mid-flight doesn't lose it. Pongs stay ephemeral. Together: the librarian keeps a result until the bot confirms it; the bot keeps it until it is on screen; duplicates never double-render. Combined with the deploy return-path fix, an expensive result no longer vanishes. Tests: unit test_durable_queue; integration test_librarian_outbox (retry/backoff, resend survives outage) and test_result_durable_delivery (persist, dedup pending, dedup delivered, replay, pong not persisted). Suite: 55 unit + 39 integration green. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
93 lines
3.1 KiB
Python
93 lines
3.1 KiB
Python
"""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"}]
|