fix(bg): sweep completion-dedup map by delivery lifecycle, not channel (#4633)
BG_TASK_COMPLETE_EVENTS_SEEN gained a session_id -> set[process_id] entry the first time a bg task completed for a session and was never deleted anywhere, so it grew unbounded for the server lifetime. The entry is created in _process_one for EVERY completion, whether or not any SSE channel/tab exists, so pruning it only when the SessionChannel is reaped would miss the dominant headless case (task fires, tab closed or never opened — no channel to collect). Instead the reaper now sweeps the map by DELIVERY: once a completion is drained (its session_id removed from PENDING_BG_TASK_COMPLETIONS) the short _move_to_finished dedup window is closed and the entry is swept, every tick, regardless of any channel. The registry's per-process_id _completion_consumed gate remains the primary idempotency backstop, so sweeping a delivered session's set can never resurrect an already-delivered completion. Session deletion also prunes the entry (new forget_bg_task_completion_dedup), covering a session deleted while a completion is still pending (which the delivery-gated sweep deliberately retains). Refs #4633 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
c008e4443a
commit
15470a9a35
3 files changed
+172
No files matched your search
@@ -377,6 +377,32 @@ def _reaper_loop() -> None:
|
||||
for sid in collected:
|
||||
_LAST_EMIT_TS.pop(sid, None)
|
||||
logger.debug("SessionChannel reaper collected: %s", collected)
|
||||
# Sweep the per-session completion-dedup map by DELIVERY lifecycle,
|
||||
# not channel collection. ``BG_TASK_COMPLETE_EVENTS_SEEN`` gains a
|
||||
# ``session_id -> set[process_id]`` entry the first time a bg task
|
||||
# completes for a session — in ``_process_one``, whether or not any
|
||||
# tab/SSE channel ever existed — and is otherwise never deleted, so it
|
||||
# grows unbounded. Coupling the prune to channel collection (an
|
||||
# earlier version of this fix) missed the dominant case: a headless
|
||||
# completion (task fires, tab closed or never opened) has no channel
|
||||
# to collect. Instead, once a completion has been drained (its
|
||||
# ``session_id`` removed from ``PENDING_BG_TASK_COMPLETIONS``), the
|
||||
# short ``_move_to_finished`` dedup window is closed and the entry is
|
||||
# pure leak — so sweep every delivered (not-pending) session here,
|
||||
# every tick. The registry's own per-``process_id``
|
||||
# ``_completion_consumed`` gate remains the primary idempotency
|
||||
# backstop, so sweeping a delivered session's set can never resurrect
|
||||
# an already-delivered completion (even in the tiny window between
|
||||
# this module's ``SEEN.add`` and ``PENDING.add`` in ``_process_one``).
|
||||
from api import config as _cfg
|
||||
|
||||
with _cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
|
||||
for sid in [
|
||||
s
|
||||
for s in _cfg.BG_TASK_COMPLETE_EVENTS_SEEN
|
||||
if s not in _cfg.PENDING_BG_TASK_COMPLETIONS
|
||||
]:
|
||||
_cfg.BG_TASK_COMPLETE_EVENTS_SEEN.pop(sid, None)
|
||||
except Exception:
|
||||
logger.warning("SessionChannel reaper iteration failed", exc_info=True)
|
||||
# Wait but wake up promptly on stop.
|
||||
@@ -1328,6 +1354,21 @@ def unregister_process_session(session_key: str) -> None:
|
||||
_cfg.PROCESS_SESSION_INDEX.pop(str(session_key), None)
|
||||
|
||||
|
||||
def forget_bg_task_completion_dedup(session_id: str) -> None:
|
||||
"""Drop a session's ``BG_TASK_COMPLETE_EVENTS_SEEN`` entry.
|
||||
|
||||
Called on session deletion so a session deleted while a completion is still
|
||||
pending (undelivered) — which the reaper's delivery-gated sweep deliberately
|
||||
keeps — can't leak its dedup set forever. Safe for unknown ids (no-op).
|
||||
"""
|
||||
if not session_id:
|
||||
return
|
||||
from api import config as _cfg
|
||||
|
||||
with _cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
|
||||
_cfg.BG_TASK_COMPLETE_EVENTS_SEEN.pop(str(session_id), None)
|
||||
|
||||
|
||||
def start_drain_thread() -> bool:
|
||||
"""Start the background drain thread idempotently. Returns True on first start."""
|
||||
global _DRAIN_THREAD
|
||||
|
||||
@@ -13849,6 +13849,15 @@ def handle_post(handler, parsed) -> bool:
|
||||
# Lock entries in SESSION_AGENT_LOCKS forever.
|
||||
with SESSION_AGENT_LOCKS_LOCK:
|
||||
SESSION_AGENT_LOCKS.pop(sid, None)
|
||||
# Prune the completion-dedup entry too. The reaper sweeps it once the
|
||||
# completion is delivered (drained from PENDING); a session deleted
|
||||
# while a completion is still pending would otherwise keep its entry.
|
||||
try:
|
||||
from api.background_process import forget_bg_task_completion_dedup
|
||||
|
||||
forget_bg_task_completion_dedup(sid)
|
||||
except Exception:
|
||||
logger.debug("Failed to prune bg-task dedup entry for deleted session %s", sid)
|
||||
try:
|
||||
from api.terminal import close_terminal
|
||||
close_terminal(sid)
|
||||
|
||||
@@ -0,0 +1,122 @@
|
||||
"""Regression test for the ``BG_TASK_COMPLETE_EVENTS_SEEN`` memory leak (#4633).
|
||||
|
||||
The per-session completion-dedup map gained one ``session_id -> set[process_id]``
|
||||
entry the first time a background task completed for a session and was never
|
||||
removed anywhere in the repo, so it grew unbounded for the server lifetime.
|
||||
|
||||
The dedup entry is created in ``_process_one`` for EVERY completion, whether or
|
||||
not any SSE channel/tab exists — so coupling the prune to channel collection
|
||||
would miss the dominant headless case (task fires, tab closed or never opened).
|
||||
Instead the reaper sweeps the map by DELIVERY lifecycle: once a completion has
|
||||
been drained (its ``session_id`` removed from ``PENDING_BG_TASK_COMPLETIONS``),
|
||||
the short dedup window is closed and the entry is swept. Session deletion prunes
|
||||
it too, covering a session deleted while a completion is still pending.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
|
||||
|
||||
def test_reaper_source_sweeps_dedup_map_by_delivery():
|
||||
"""The reaper cleanup must sweep BG_TASK_COMPLETE_EVENTS_SEEN under its lock,
|
||||
gated on PENDING_BG_TASK_COMPLETIONS (delivery), not channel collection."""
|
||||
src = (REPO_ROOT / "api" / "background_process.py").read_text(encoding="utf-8")
|
||||
reaper = src[src.index("def _reaper_loop("):]
|
||||
reaper = reaper[: reaper.index("\ndef ", 1)]
|
||||
assert "BG_TASK_COMPLETE_EVENTS_SEEN_LOCK" in reaper, (
|
||||
"reaper must take the dedup-map lock"
|
||||
)
|
||||
assert "BG_TASK_COMPLETE_EVENTS_SEEN.pop(" in reaper, (
|
||||
"reaper must pop dedup entries"
|
||||
)
|
||||
assert "PENDING_BG_TASK_COMPLETIONS" in reaper, (
|
||||
"the sweep must be gated on delivery (PENDING_BG_TASK_COMPLETIONS), "
|
||||
"not on channel collection — otherwise headless completions leak"
|
||||
)
|
||||
|
||||
|
||||
def _run_reaper_once(bp, cfg):
|
||||
"""Start the real reaper, wait for one sweep pass, stop it."""
|
||||
bp.start_session_channel_reaper()
|
||||
time.sleep(0.4) # first pass runs immediately (interval sleep is at loop tail)
|
||||
bp.stop_session_channel_reaper()
|
||||
|
||||
|
||||
def test_reaper_sweeps_delivered_dedup_entry_without_any_channel():
|
||||
"""The dominant headless case: a completed+drained task with NO SSE channel
|
||||
ever — its dedup entry must still be swept (this is what the channel-coupled
|
||||
version missed)."""
|
||||
from api import background_process as bp
|
||||
from api import config as cfg
|
||||
|
||||
sid = "sess-headless-4633"
|
||||
with cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
|
||||
cfg.BG_TASK_COMPLETE_EVENTS_SEEN[sid] = {"proc-1"}
|
||||
cfg.PENDING_BG_TASK_COMPLETIONS.discard(sid) # delivered/drained
|
||||
# No SessionChannel for sid — nothing for channel collection to reap.
|
||||
try:
|
||||
with bp.SESSION_CHANNELS_LOCK:
|
||||
assert sid not in bp.SESSION_CHANNELS
|
||||
_run_reaper_once(bp, cfg)
|
||||
with cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
|
||||
assert sid not in cfg.BG_TASK_COMPLETE_EVENTS_SEEN, (
|
||||
"reaper did not sweep the delivered headless dedup entry"
|
||||
)
|
||||
finally:
|
||||
bp.stop_session_channel_reaper()
|
||||
with cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
|
||||
cfg.BG_TASK_COMPLETE_EVENTS_SEEN.pop(sid, None)
|
||||
|
||||
|
||||
def test_reaper_retains_dedup_entry_while_completion_pending():
|
||||
"""An UNDELIVERED completion (still in PENDING) must keep its dedup entry —
|
||||
the _move_to_finished dedup window is still open."""
|
||||
from api import background_process as bp
|
||||
from api import config as cfg
|
||||
|
||||
sid = "sess-pending-4633"
|
||||
with cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
|
||||
cfg.BG_TASK_COMPLETE_EVENTS_SEEN[sid] = {"proc-2"}
|
||||
cfg.PENDING_BG_TASK_COMPLETIONS.add(sid) # undelivered
|
||||
try:
|
||||
_run_reaper_once(bp, cfg)
|
||||
with cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
|
||||
assert sid in cfg.BG_TASK_COMPLETE_EVENTS_SEEN, (
|
||||
"reaper swept a still-pending (undelivered) dedup entry"
|
||||
)
|
||||
finally:
|
||||
bp.stop_session_channel_reaper()
|
||||
cfg.PENDING_BG_TASK_COMPLETIONS.discard(sid)
|
||||
with cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
|
||||
cfg.BG_TASK_COMPLETE_EVENTS_SEEN.pop(sid, None)
|
||||
|
||||
|
||||
def test_session_delete_prunes_dedup_entry():
|
||||
"""Deleting a session prunes its dedup entry even if the completion never
|
||||
delivered (so a delete-while-pending session can't leak forever)."""
|
||||
from api import background_process as bp
|
||||
from api import config as cfg
|
||||
|
||||
sid = "sess-delete-4633"
|
||||
with cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
|
||||
cfg.BG_TASK_COMPLETE_EVENTS_SEEN[sid] = {"proc-3"}
|
||||
cfg.PENDING_BG_TASK_COMPLETIONS.add(sid) # undelivered, but being deleted
|
||||
try:
|
||||
bp.forget_bg_task_completion_dedup(sid)
|
||||
with cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
|
||||
assert sid not in cfg.BG_TASK_COMPLETE_EVENTS_SEEN
|
||||
finally:
|
||||
cfg.PENDING_BG_TASK_COMPLETIONS.discard(sid)
|
||||
with cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
|
||||
cfg.BG_TASK_COMPLETE_EVENTS_SEEN.pop(sid, None)
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover - manual invocation
|
||||
raise SystemExit(pytest.main([__file__, "-v"]))
|
||||
Reference in new issue
Block a user