Nick
1c16d3d26c
A 10KB single-token error string made a 10KB dedup marker (issue titles and q= params unbounded). error_class_of now caps at 64 chars for both exception names and string tokens. Also replaces the vacuous forge_created_not_called() assertion (unconditional True) with real fake-state assertions (zero client calls, failed=1, line left spooled).
417 lines
17 KiB
Python
417 lines
17 KiB
Python
"""#122 Task 2: burst coalescing + spool reconciliation.
|
|
|
|
Burst coalescing: after the first occurrence files, subsequent occurrences
|
|
within the dedup window batch into ONE comment per batch_size occurrences
|
|
or 10 minutes (whichever first) — a 3am crash-loop must not write 200
|
|
comments. The worker additionally coalesces everything already queued
|
|
into one write per signature. Reconcile: replay unconsumed spool lines
|
|
through the same pipeline (dedup re-applies via search_issues); filed
|
|
lines are marked consumed; forge still down → spool preserved, never
|
|
raises.
|
|
"""
|
|
|
|
import json
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from rune.failure_auto_filer import (
|
|
BATCH_INTERVAL_SECONDS,
|
|
FailureAutoFiler,
|
|
JsonlSpool,
|
|
)
|
|
from rune.forge_client import NullAdapter
|
|
|
|
from tests.test_failure_auto_filer import FakeForge, FakeJournal, MARKER
|
|
|
|
|
|
# ------------------------------------------------------------ burst coalescing
|
|
|
|
|
|
def test_burst_of_50_produces_at_most_2_client_writes():
|
|
# First occurrence files; everything already queued coalesces into one
|
|
# comment; the tail (< batch_size) stays pending: 1 create + ≤1 comment.
|
|
forge = FakeForge()
|
|
filer = FailureAutoFiler(forge, FakeJournal(), batch_size=10, wait=True)
|
|
for i in range(50):
|
|
filer.submit("memory_selector", f"RuntimeError: boom {i}", tier=1)
|
|
filer.drain(timeout=5) # the whole backlog coalesces into ONE write
|
|
assert len(forge.created) == 1
|
|
assert len(forge.created) + len(forge.commented) <= 2
|
|
|
|
|
|
def test_batch_flushes_every_batch_size_occurrences():
|
|
forge = FakeForge()
|
|
filer = FailureAutoFiler(forge, FakeJournal(), batch_size=10, wait=False)
|
|
for i in range(25):
|
|
filer.file("memory_selector", f"RuntimeError: boom {i}", tier=1)
|
|
assert len(forge.created) == 1
|
|
# Comments when pending EXCEEDS the batch size (occurrences 11 and 21):
|
|
# one comment per 10 occurrences, tail pending.
|
|
assert len(forge.commented) == 2
|
|
assert "+10 occurrences" in forge.commented[0]["body"]
|
|
assert "+20 occurrences" in forge.commented[1]["body"]
|
|
|
|
|
|
def test_batch_flushes_after_ten_minutes():
|
|
forge = FakeForge()
|
|
filer = FailureAutoFiler(forge, FakeJournal(), batch_size=10, wait=False)
|
|
start = time.monotonic()
|
|
filer.file("memory_selector", "RuntimeError: boom 0", tier=1, _now=start) # files
|
|
later = start + BATCH_INTERVAL_SECONDS + 1 # 10 minutes pass
|
|
filer.file("memory_selector", "RuntimeError: boom 1", tier=1, _now=later)
|
|
assert len(forge.created) == 1
|
|
assert len(forge.commented) == 1 # window expired -> flush on next occurrence
|
|
assert "+1 occurrences" in forge.commented[0]["body"]
|
|
|
|
|
|
def test_burst_counter_is_per_signature():
|
|
forge = FakeForge()
|
|
filer = FailureAutoFiler(forge, FakeJournal(), batch_size=10, wait=False)
|
|
for i in range(12):
|
|
filer.file("memory_selector", f"RuntimeError: boom {i}", tier=1)
|
|
for i in range(3):
|
|
filer.file("warm_router", f"TimeoutError: slow {i}", tier=1)
|
|
assert len(forge.created) == 2 # one per signature
|
|
assert len(forge.commented) == 1 # memory_selector hit 10; warm_router (3) not yet
|
|
|
|
|
|
def test_burst_tail_below_batch_size_stays_pending_until_window():
|
|
forge = FakeForge()
|
|
filer = FailureAutoFiler(forge, FakeJournal(), batch_size=10, wait=False)
|
|
for i in range(7):
|
|
filer.file("memory_selector", f"RuntimeError: boom {i}", tier=1)
|
|
filer.drain(timeout=5)
|
|
assert len(forge.created) == 1
|
|
assert len(forge.commented) == 0 # 7 < batch_size, window not expired
|
|
|
|
|
|
def test_burst_batched_comment_carries_last_error_excerpt():
|
|
forge = FakeForge()
|
|
filer = FailureAutoFiler(forge, FakeJournal(), batch_size=10, wait=True)
|
|
for i in range(11): # 11th occurrence crosses the batch threshold
|
|
filer.file("memory_selector", f"RuntimeError: boom {i}", tier=1)
|
|
filer.drain(timeout=5)
|
|
assert len(forge.commented) == 1
|
|
assert "+10 occurrences" in forge.commented[0]["body"]
|
|
assert "boom 10" in forge.commented[0]["body"] # last error of the batch
|
|
|
|
|
|
def test_window_resets_after_flush():
|
|
forge = FakeForge()
|
|
filer = FailureAutoFiler(forge, FakeJournal(), batch_size=10, wait=False)
|
|
base = time.monotonic()
|
|
filer.file("memory_selector", "RuntimeError: a0", tier=1, _now=base) # files
|
|
for i in range(1, 12):
|
|
filer.file("memory_selector", f"RuntimeError: a{i}", tier=1, _now=base + i * 0.1)
|
|
assert len(forge.commented) == 1 # flush at pending=11 (> batch_size)
|
|
# The window restarted at that flush: no second comment for 3 more.
|
|
for i in range(3):
|
|
filer.file("memory_selector", f"RuntimeError: b{i}", tier=1, _now=base + 2 + i * 0.1)
|
|
assert len(forge.commented) == 1
|
|
|
|
|
|
# ---------------------------------------------------------------- reconcile
|
|
|
|
|
|
def test_reconcile_groups_spool_lines_by_signature_into_one_write():
|
|
# #122 gate: 10 spooled lines of ONE signature (a long outage) must
|
|
# produce exactly ONE client write on reconcile — not a comment flood.
|
|
forge = FakeForge()
|
|
journal = FakeJournal()
|
|
filer = FailureAutoFiler(forge, journal, batch_size=1, wait=False)
|
|
for i in range(10):
|
|
journal.append(
|
|
{
|
|
"kind": "failure_auto_file_spool",
|
|
"spool_id": f"spool-{i}",
|
|
"timestamp": "t",
|
|
"signature": ["memory_selector", "RuntimeError"],
|
|
"tier": 1,
|
|
"count": 1,
|
|
"error": f"RuntimeError: boom {i}",
|
|
}
|
|
)
|
|
report = filer.reconcile()
|
|
assert len(forge.created) + len(forge.commented) == 1 # ONE write
|
|
assert len(forge.created) == 1 # dedup still applies: issue created once
|
|
assert "occurrences=10" in forge.created[0]["body"] # aggregated count
|
|
assert "boom 9" in forge.created[0]["body"] # last error of the group
|
|
assert report["filed"] == 10
|
|
assert journal.unconsumed() == []
|
|
|
|
|
|
def test_reconcile_group_comments_once_with_offline_count():
|
|
# Same grouping when the issue already exists: ONE aggregated comment.
|
|
forge = FakeForge(search_results=[{"number": 7, "title": MARKER + " memory_selector"}])
|
|
journal = FakeJournal()
|
|
filer = FailureAutoFiler(forge, journal, batch_size=1, wait=False)
|
|
for i in range(4):
|
|
journal.append(
|
|
{
|
|
"kind": "failure_auto_file_spool",
|
|
"spool_id": f"spool-{i}",
|
|
"timestamp": "t",
|
|
"signature": ["memory_selector", "RuntimeError"],
|
|
"tier": 1,
|
|
"count": 1,
|
|
"error": f"RuntimeError: boom {i}",
|
|
}
|
|
)
|
|
report = filer.reconcile()
|
|
assert forge.created == []
|
|
assert len(forge.commented) == 1
|
|
assert "+4 occurrences while offline" in forge.commented[0]["body"]
|
|
assert "boom 3" in forge.commented[0]["body"]
|
|
assert report["filed"] == 4
|
|
assert journal.unconsumed() == []
|
|
|
|
|
|
def test_reconcile_replays_spool_and_marks_consumed():
|
|
forge = FakeForge()
|
|
journal = FakeJournal()
|
|
filer = FailureAutoFiler(forge, journal, batch_size=1, wait=False)
|
|
forge.fail_with = ConnectionError("forge down")
|
|
filer.file("memory_selector", "RuntimeError: boom", tier=1)
|
|
assert len(journal.unconsumed()) == 1
|
|
forge.fail_with = None
|
|
filer.reconcile()
|
|
assert len(forge.created) == 1
|
|
assert forge.created[0]["title"].startswith(MARKER)
|
|
assert journal.unconsumed() == [] # marked consumed
|
|
|
|
|
|
def test_reconcile_dedups_against_live_issues():
|
|
forge = FakeForge(search_results=[{"number": 7, "title": MARKER + " memory_selector"}])
|
|
journal = FakeJournal()
|
|
filer = FailureAutoFiler(forge, journal, batch_size=1, wait=False)
|
|
forge.fail_with = ConnectionError("forge down")
|
|
filer.file("memory_selector", "RuntimeError: boom", tier=1)
|
|
forge.fail_with = None
|
|
filer.reconcile()
|
|
assert forge.created == [] # dedup re-applied: comment, not a second issue
|
|
assert len(forge.commented) == 1
|
|
assert journal.unconsumed() == []
|
|
|
|
|
|
def test_reconcile_with_forge_still_down_preserves_spool():
|
|
forge = FakeForge(fail_with=ConnectionError("forge still down"))
|
|
journal = FakeJournal()
|
|
filer = FailureAutoFiler(forge, journal, batch_size=1, wait=False)
|
|
filer.file("memory_selector", "RuntimeError: boom", tier=1)
|
|
pending_before = [dict(r) for r in journal.unconsumed()]
|
|
filer.reconcile() # must not raise
|
|
assert journal.unconsumed() == pending_before # preserved, still pending
|
|
assert forge.created == []
|
|
|
|
|
|
def test_reconcile_on_jsonl_spool_marks_consumed_in_sidecar(tmp_path):
|
|
path = tmp_path / "events.jsonl"
|
|
forge = FakeForge()
|
|
spool = JsonlSpool(path)
|
|
filer = FailureAutoFiler(forge, spool, batch_size=1, wait=False)
|
|
forge.fail_with = ConnectionError("down")
|
|
filer.file("memory_selector", "RuntimeError: boom", tier=1)
|
|
assert len(spool.unconsumed()) == 1
|
|
canonical_before = path.read_bytes()
|
|
forge.fail_with = None
|
|
filer.reconcile()
|
|
assert spool.unconsumed() == []
|
|
# Canonical journal untouched — consumption lives in the sidecar only.
|
|
assert path.read_bytes() == canonical_before
|
|
assert (tmp_path / "events.jsonl.consumed").exists()
|
|
|
|
|
|
def _spool_record(spool_id: str) -> dict:
|
|
return {
|
|
"kind": "failure_auto_file_spool",
|
|
"spool_id": spool_id,
|
|
"timestamp": "t",
|
|
"signature": ["s", "E"],
|
|
"tier": 1,
|
|
"count": 1,
|
|
"error": "E: x",
|
|
}
|
|
|
|
|
|
def test_mark_consumed_never_rewrites_the_canonical_journal(tmp_path):
|
|
# #122 gate: mark_consumed must be append-only. The canonical journal
|
|
# is shared with DurableEventStore — a read-modify-write there is
|
|
# racy and a crash mid-rewrite poisons the canonical store.
|
|
path = tmp_path / "events.jsonl"
|
|
spool = JsonlSpool(path)
|
|
spool.append(_spool_record("spool-x"))
|
|
before = path.read_bytes()
|
|
spool.mark_consumed("spool-x")
|
|
assert path.read_bytes() == before # byte-identical: no rewrite, ever
|
|
assert spool.unconsumed() == [] # consumption recorded in the sidecar
|
|
assert (tmp_path / "events.jsonl.consumed").read_text().splitlines() == ["spool-x"]
|
|
|
|
|
|
def test_concurrent_append_between_read_and_mark_is_not_lost(tmp_path, monkeypatch):
|
|
# A DurableEventStore.append landing while mark_consumed runs must
|
|
# survive. The old implementation read the journal, then rewrote it —
|
|
# the append landed in between and was silently DESTROYED.
|
|
import builtins
|
|
|
|
path = tmp_path / "events.jsonl"
|
|
spool = JsonlSpool(path)
|
|
spool.append(_spool_record("spool-x"))
|
|
real_open = builtins.open
|
|
state = {"fired": False}
|
|
|
|
def racing_open(file, mode="r", *a, **k):
|
|
# The concurrent O_APPEND lands while mark_consumed is mid-flight
|
|
# (fires on the first write/append-mode open inside the mark).
|
|
if not state["fired"] and any(m in mode for m in ("a", "w", "x")):
|
|
state["fired"] = True
|
|
with real_open(path, "a", encoding="utf-8") as fh:
|
|
fh.write(json.dumps({"event_id": "e-race", "event_type": "task.failed"}) + "\n")
|
|
return real_open(file, mode, *a, **k)
|
|
|
|
monkeypatch.setattr(builtins, "open", racing_open)
|
|
spool.mark_consumed("spool-x")
|
|
rows = [json.loads(line) for line in path.read_text().splitlines()]
|
|
assert any(r.get("event_id") == "e-race" for r in rows) # NOT destroyed
|
|
assert spool.unconsumed() == []
|
|
|
|
|
|
def test_reconcile_routes_tier2_line_through_the_tier_gate():
|
|
# #122 gate: a crafted tier=2 spool line must stay journal-only on
|
|
# replay — the SAME one tier gate as the live path (no _write bypass).
|
|
forge = FakeForge()
|
|
journal = FakeJournal()
|
|
filer = FailureAutoFiler(forge, journal, wait=False)
|
|
journal.append(_spool_record("spool-t2") | {"tier": 2})
|
|
report = filer.reconcile()
|
|
assert forge.searched == [] # zero client calls: journal-only on replay
|
|
assert forge.created == []
|
|
assert forge.commented == []
|
|
assert report["skipped"] == 1
|
|
assert journal.unconsumed() == [] # resolved: consumed, never written
|
|
|
|
|
|
def test_reconcile_mixed_group_tier2_lines_dropped_tier1_filed():
|
|
# A group mixing tiers keeps filing the T0/T1 lines; only the T2
|
|
# lines are gate-dropped (journal-only).
|
|
forge = FakeForge()
|
|
journal = FakeJournal()
|
|
filer = FailureAutoFiler(forge, journal, batch_size=1, wait=False)
|
|
journal.append(_spool_record("spool-t1") | {"tier": 1})
|
|
journal.append(_spool_record("spool-t2") | {"tier": 2})
|
|
report = filer.reconcile()
|
|
assert len(forge.created) == 1
|
|
assert report["filed"] == 1
|
|
assert report["skipped"] == 1
|
|
assert journal.unconsumed() == []
|
|
|
|
|
|
def test_reconcile_and_worker_single_flight_per_signature():
|
|
# #122 gate: the reconcile thread and the worker racing on the SAME
|
|
# signature must not duplicate the issue — single-flight lets exactly
|
|
# one write land per signature at a time.
|
|
import threading
|
|
|
|
forge = FakeForge()
|
|
entries = {"n": 0}
|
|
release = threading.Event()
|
|
|
|
def slow_search(query):
|
|
entries["n"] += 1
|
|
if entries["n"] == 1:
|
|
release.wait(timeout=5) # hold the first writer inside the window
|
|
# Like a real forge: issues created earlier ARE found by later
|
|
# searches, and search results carry the issue number.
|
|
return [
|
|
{"number": n, "title": i["title"]} for n, i in enumerate(forge.created, start=100) if query in i["title"]
|
|
]
|
|
|
|
forge.search_issues = slow_search # type: ignore[method-assign]
|
|
journal = FakeJournal()
|
|
journal.append(_spool_record("spool-race"))
|
|
filer = FailureAutoFiler(forge, journal, batch_size=1, wait=False)
|
|
filer.submit("s", "E: x", tier=1) # WORKER path (submit), same signature
|
|
deadline = time.monotonic() + 5
|
|
while entries["n"] < 1 and time.monotonic() < deadline:
|
|
time.sleep(0.01) # worker is inside the write window
|
|
reconcile_report = filer.reconcile() # races the worker on the same sig
|
|
release.set()
|
|
filer.drain(timeout=5)
|
|
assert len(forge.created) == 1 # NEVER a duplicate issue
|
|
assert forge.commented == []
|
|
assert len(forge.created) + len(forge.commented) == 1
|
|
# The racing replay stayed spooled (not falsely consumed): the
|
|
# in-flight writer owned the signature; a later replay dedups.
|
|
assert reconcile_report["filed"] == 0
|
|
filer.reconcile()
|
|
assert len(forge.created) == 1 # dedup re-applies: comment, not issue 2
|
|
assert len(forge.commented) == 1
|
|
assert journal.unconsumed() == []
|
|
|
|
|
|
def test_reconcile_empty_spool_is_a_noop():
|
|
forge = FakeForge()
|
|
filer = FailureAutoFiler(forge, FakeJournal(), wait=False)
|
|
filer.reconcile()
|
|
assert forge.searched == []
|
|
assert forge.created == []
|
|
|
|
|
|
def test_reconcile_files_and_consumes_multiple_spooled_signatures():
|
|
forge = FakeForge()
|
|
journal = FakeJournal()
|
|
filer = FailureAutoFiler(forge, journal, batch_size=1, wait=False)
|
|
forge.fail_with = ConnectionError("down")
|
|
filer.file("memory_selector", "RuntimeError: boom", tier=1)
|
|
filer.file("warm_router", "TimeoutError: slow", tier=1)
|
|
assert len(journal.unconsumed()) == 2
|
|
forge.fail_with = None
|
|
filer.reconcile()
|
|
assert len(forge.created) == 2
|
|
assert journal.unconsumed() == []
|
|
|
|
|
|
def test_null_adapter_reconcile_consumes_spool():
|
|
# NullAdapter never raises; a search returns [] and create_issue is a
|
|
# no-op dict — the filer treats that as "nothing was filed" and keeps
|
|
# the line spooled rather than falsely consuming it.
|
|
journal = FakeJournal()
|
|
filer = FailureAutoFiler(NullAdapter(), journal, batch_size=1, wait=False)
|
|
journal.append(
|
|
{
|
|
"kind": "failure_auto_file_spool",
|
|
"spool_id": "spool-1",
|
|
"timestamp": "t",
|
|
"signature": ["memory_selector", "RuntimeError"],
|
|
"tier": 1,
|
|
"count": 1,
|
|
"error": "RuntimeError: boom",
|
|
}
|
|
)
|
|
filer.reconcile()
|
|
assert journal.unconsumed() != [] # not filed → stays spooled
|
|
|
|
|
|
def test_reconcile_never_raises_when_journal_read_fails():
|
|
class ExplodingJournal(FakeJournal):
|
|
def unconsumed(self):
|
|
raise OSError("disk gone")
|
|
|
|
filer = FailureAutoFiler(FakeForge(), ExplodingJournal(), wait=False)
|
|
filer.reconcile() # must not raise
|
|
|
|
|
|
def test_reconcile_malformed_line_is_skipped_without_crash():
|
|
# #122 gate: assert against the REAL fake state (the old version ended
|
|
# in forge_created_not_called(), which returned True unconditionally).
|
|
journal = FakeJournal()
|
|
journal.lines.append({"kind": "failure_auto_file_spool", "spool_id": "bad", "tier": 1})
|
|
forge = FakeForge()
|
|
filer = FailureAutoFiler(forge, journal, wait=False)
|
|
report = filer.reconcile()
|
|
assert forge.created == [] # nothing was written to the forge
|
|
assert forge.commented == []
|
|
assert forge.searched == []
|
|
assert report["replayed"] == 1
|
|
assert report["failed"] == 1
|
|
assert journal.unconsumed() == [{"kind": "failure_auto_file_spool", "spool_id": "bad", "tier": 1}]
|