Phase 2 backend reliability/data-integrity batch. Each byte-verified vs PR head, Codex SAFE, 20 own + 355 regression tests pass, disjoint files. - #5835 (ai-ag2026): stop leaking fd + reader thread + shell on client/shell exit (#4633 family) - #5843 (ai-ag2026): fsync atomic JSON writers (oauth/passkeys/discoverability) + thread-safe temp Co-authored-by: t <a@b> Co-authored-by: ai-ag2026 <ai-ag2026@users.noreply.github.com>
This commit is contained in:
8 files changed
+474
-17
No files matched your search
@@ -11,6 +11,10 @@
|
||||
|
||||
### Fixed
|
||||
|
||||
- **The embedded terminal no longer leaks a file descriptor, reader thread, and shell process on exit.** When the terminal client disconnected or the shell exited, the PTY fd, its reader thread, and the shell process were left dangling — accumulating over a long uptime (part of the #4633 long-uptime-crash family). Teardown now closes the fd once, stops the reader thread, and reaps the shell on both the client-exit and shell-exit paths. Thanks @ai-ag2026. (#5835, #4633)
|
||||
|
||||
- **Atomic JSON writers now `fsync` before rename, and the discoverability temp write is thread-safe.** The OAuth, passkeys, and session-discoverability stores wrote via temp-file → `os.replace` but without an `fsync`, so a crash right after the rename could still surface a partially-flushed file; they now flush → `fsync` → replace (matching the settings/config atomic-write hardening). The discoverability temp file also now uses a thread-specific name so concurrent writers can't collide. Thanks @ai-ag2026. (#5843)
|
||||
|
||||
- **The dashboard status poll now pauses while the browser tab is hidden.** The Insights/dashboard status poll kept firing on a backgrounded tab; it now skips unforced polls while `document.hidden` and does one catch-up fetch on re-show (sibling of the hidden-tab session-poll pause). Forced init/save/resume fetches are unaffected. Thanks @ai-ag2026. (#5840)
|
||||
|
||||
- **A slow or stuck client can no longer pin the session-events invalidation stream.** The session-events SSE stream now arms a write deadline after headers and before subscription, so a client that stops reading is torn down (and unsubscribed in `finally`) instead of holding the stream open indefinitely. A healthy slow-but-alive client is unaffected. Thanks @ai-ag2026. (#5841)
|
||||
|
||||
+4
-1
@@ -191,7 +191,10 @@ def _write_auth_json(data: dict[str, Any], auth_path: Path | None = None) -> Pat
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
tmp = path.with_name(f"{path.name}.tmp.{os.getpid()}.{uuid.uuid4().hex}")
|
||||
try:
|
||||
tmp.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
|
||||
with open(tmp, "w", encoding="utf-8") as f:
|
||||
f.write(json.dumps(data, indent=2, ensure_ascii=False) + "\n")
|
||||
f.flush()
|
||||
os.fsync(f.fileno()) # durable before rename: no truncated auth.json on power loss
|
||||
try:
|
||||
tmp.chmod(0o600)
|
||||
except OSError as exc:
|
||||
|
||||
@@ -72,6 +72,8 @@ def _atomic_write_json(path: Path, payload: Any) -> None:
|
||||
try:
|
||||
with os.fdopen(fd, "w", encoding="utf-8") as f:
|
||||
json.dump(payload, f, indent=2, sort_keys=True)
|
||||
f.flush()
|
||||
os.fsync(f.fileno()) # durable before rename: no zero-length passkey file on power loss
|
||||
os.chmod(tmp, 0o600)
|
||||
os.replace(tmp, path)
|
||||
except Exception:
|
||||
|
||||
@@ -13,6 +13,7 @@ from __future__ import annotations
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import threading
|
||||
import shutil
|
||||
import sqlite3
|
||||
from collections import Counter
|
||||
@@ -387,9 +388,27 @@ def audit_session_discoverability(
|
||||
|
||||
|
||||
def _atomic_write_json(path: Path, payload) -> None:
|
||||
tmp = path.with_suffix(path.suffix + f".tmp.{os.getpid()}")
|
||||
tmp.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
os.replace(tmp, path)
|
||||
# Temp name includes the thread id, not just the pid: WebUI runs a
|
||||
# ThreadingHTTPServer, so two request threads reconciling the same _index.json
|
||||
# would otherwise collide on one `<name>.tmp.<pid>` path and truncate each
|
||||
# other's temp before its os.replace. fsync before the rename so a power loss
|
||||
# can't leave a zero-length/garbage file where the durable JSON should be; a
|
||||
# finally drops the temp if anything fails. Mode matches the prior write_text
|
||||
# (umask default) since the plain open() respects the umask.
|
||||
text = json.dumps(payload, ensure_ascii=False, indent=2)
|
||||
tmp = path.with_name(f".{path.name}.{os.getpid()}.{threading.get_ident()}.tmp")
|
||||
try:
|
||||
with open(tmp, "w", encoding="utf-8") as f:
|
||||
f.write(text)
|
||||
f.flush()
|
||||
os.fsync(f.fileno())
|
||||
os.replace(tmp, path)
|
||||
except BaseException:
|
||||
try:
|
||||
os.unlink(tmp)
|
||||
except OSError:
|
||||
pass
|
||||
raise
|
||||
|
||||
|
||||
def _backup_file(path: Path, backup_dir: Path, backed_up: dict[Path, str]) -> str | None:
|
||||
@@ -482,7 +501,7 @@ def _materialize_sidecar_from_state_db(session_dir: Path, state_db_path: Path |
|
||||
payload = _state_db_row_to_sidecar(row)
|
||||
_backup_file(state_db_path, backup_dir, backed_up)
|
||||
session_dir.mkdir(parents=True, exist_ok=True)
|
||||
tmp = target.with_suffix(target.suffix + f".tmp.{os.getpid()}")
|
||||
tmp = target.with_suffix(target.suffix + f".tmp.{os.getpid()}.{threading.get_ident()}")
|
||||
tmp.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
try:
|
||||
os.link(str(tmp), str(target))
|
||||
|
||||
+105
-10
@@ -64,11 +64,23 @@ class TerminalSession:
|
||||
output: queue.Queue = field(default_factory=lambda: queue.Queue(maxsize=2000))
|
||||
closed: threading.Event = field(default_factory=threading.Event)
|
||||
reader: threading.Thread | None = None
|
||||
# Serializes fd-touching ops (os.write, resize ioctl) against os.close, so a
|
||||
# write can never land on a master_fd that was closed and whose number was
|
||||
# already recycled by a concurrent openpty — that would inject the user's
|
||||
# keystrokes into a foreign fd. Holders re-check ``closed`` under this lock
|
||||
# and bail if the terminal has been torn down.
|
||||
io_lock: threading.Lock = field(default_factory=threading.Lock)
|
||||
# Wall-clock of the last input written or output produced. Drives which
|
||||
# terminal the cap evicts first (least-recently-active): a shell abandoned
|
||||
# at its prompt has neither, so it sorts oldest and is evicted before an
|
||||
# actively used one.
|
||||
last_activity: float = field(default_factory=time.time)
|
||||
|
||||
def is_alive(self) -> bool:
|
||||
return not self.closed.is_set() and self.proc.poll() is None
|
||||
|
||||
def put_output(self, event: str, payload: dict) -> None:
|
||||
self.last_activity = time.time()
|
||||
try:
|
||||
self.output.put_nowait((event, payload))
|
||||
except queue.Full:
|
||||
@@ -85,6 +97,14 @@ class TerminalSession:
|
||||
|
||||
_TERMINALS: dict[str, TerminalSession] = {}
|
||||
_LOCK = threading.RLock()
|
||||
# Hard cap on concurrently live embedded terminals. Each holds a shell process,
|
||||
# a pty master fd, and a reader thread; a client that drops its output stream
|
||||
# without POSTing /api/terminal/close (tab close, crash, network drop) leaves
|
||||
# the shell running (no PDEATHSIG — see the note below), so without a ceiling
|
||||
# these accumulate over a long uptime toward fd/thread exhaustion (#4633). The
|
||||
# cap evicts the least-recently-active terminal to make room. Generous enough
|
||||
# that real interactive use never trips it.
|
||||
_MAX_TERMINALS = 32
|
||||
_spawn_queue: queue.Queue = queue.Queue()
|
||||
_spawn_supervisor_started = False
|
||||
_spawn_supervisor_lock = threading.Lock()
|
||||
@@ -278,15 +298,27 @@ def _reader_loop(term: TerminalSession) -> None:
|
||||
code = term.proc.poll()
|
||||
_reap_terminal_descendants(term.proc.pid)
|
||||
term.put_output("terminal_closed", {"exit_code": code})
|
||||
# The shell has exited (or its pty broke): retire the session so its
|
||||
# master fd and _TERMINALS entry are released. Previously only an
|
||||
# explicit close_terminal() / restart / atexit did this, so a shell that
|
||||
# exited on its own (user typed `exit`, process died) leaked its master
|
||||
# fd and dict entry for the rest of the WebUI's uptime. ``expected=term``
|
||||
# makes this a no-op if a restart already replaced the entry with a new
|
||||
# terminal for the same session id.
|
||||
close_terminal(term.session_id, expected=term)
|
||||
|
||||
|
||||
def _set_size(term: TerminalSession, rows: int, cols: int) -> None:
|
||||
term.rows = max(8, min(int(rows or term.rows or 24), 80))
|
||||
term.cols = max(20, min(int(cols or term.cols or 80), 240))
|
||||
try:
|
||||
fcntl.ioctl(term.master_fd, termios.TIOCSWINSZ, _winsize(term.rows, term.cols))
|
||||
except OSError:
|
||||
pass
|
||||
# The ioctl touches master_fd, so guard it against a concurrent close (and
|
||||
# the fd-number recycling that can follow) with the same io_lock as writes.
|
||||
with term.io_lock:
|
||||
if not term.closed.is_set():
|
||||
try:
|
||||
fcntl.ioctl(term.master_fd, termios.TIOCSWINSZ, _winsize(term.rows, term.cols))
|
||||
except OSError:
|
||||
pass
|
||||
try:
|
||||
if term.proc.poll() is None:
|
||||
os.killpg(term.proc.pid, signal.SIGWINCH)
|
||||
@@ -294,6 +326,43 @@ def _set_size(term: TerminalSession, rows: int, cols: int) -> None:
|
||||
pass
|
||||
|
||||
|
||||
def _enforce_terminal_cap(*, exclude_sid: str | None = None) -> None:
|
||||
"""Evict terminals until there is room under ``_MAX_TERMINALS``.
|
||||
|
||||
Picks a victim under the lock but closes it *outside* the lock (close_terminal
|
||||
may spend up to a couple of seconds killing/waiting on the shell). Prefers a
|
||||
dead-process terminal, else the least-recently-active one — an abandoned
|
||||
shell idle at its prompt sorts oldest and goes first. ``exclude_sid`` is the
|
||||
session about to reuse/replace its own entry, so it never evicts itself.
|
||||
"""
|
||||
if not _TERMINAL_SUPPORTED:
|
||||
return
|
||||
# Bounded loop: at most the current population; guards against a pathological
|
||||
# spin if close_terminal somehow can't remove an entry.
|
||||
for _ in range(_MAX_TERMINALS + 1):
|
||||
victim_sid = None
|
||||
victim_term = None
|
||||
with _LOCK:
|
||||
# Reuse/restart of an existing sid replaces in place — no growth.
|
||||
if exclude_sid in _TERMINALS:
|
||||
return
|
||||
if len(_TERMINALS) < _MAX_TERMINALS:
|
||||
return
|
||||
candidates = [
|
||||
(sid, term) for sid, term in _TERMINALS.items() if sid != exclude_sid
|
||||
]
|
||||
if not candidates:
|
||||
return
|
||||
dead = [(sid, term) for sid, term in candidates if not term.is_alive()]
|
||||
victim_sid, victim_term = (
|
||||
dead[0] if dead else min(candidates, key=lambda kv: kv[1].last_activity)
|
||||
)
|
||||
# ``expected=victim_term`` so that if this sid was restarted/replaced in
|
||||
# the gap between picking it and closing it, we don't tear down the new
|
||||
# (possibly active) terminal — symmetric with the reader-loop retire.
|
||||
close_terminal(victim_sid, expected=victim_term)
|
||||
|
||||
|
||||
def start_terminal(session_id: str, workspace: Path, rows: int = 24, cols: int = 80, restart: bool = False) -> TerminalSession:
|
||||
"""Start or return the embedded terminal for a WebUI session."""
|
||||
if not _TERMINAL_SUPPORTED:
|
||||
@@ -305,6 +374,12 @@ def start_terminal(session_id: str, workspace: Path, rows: int = 24, cols: int =
|
||||
if not Path(cwd).is_dir():
|
||||
raise ValueError("workspace is not a directory")
|
||||
|
||||
# Enforce the cap before spawning. Done outside the main lock below (the
|
||||
# eviction's process teardown must not run while _LOCK is held for the whole
|
||||
# spawn), and skipped for a same-sid reuse/restart, which replaces rather
|
||||
# than adds an entry.
|
||||
_enforce_terminal_cap(exclude_sid=sid)
|
||||
|
||||
with _LOCK:
|
||||
current = _TERMINALS.get(sid)
|
||||
if current and current.is_alive() and not restart and current.workspace == cwd:
|
||||
@@ -402,7 +477,14 @@ def write_terminal(session_id: str, data: str) -> None:
|
||||
term = get_terminal(session_id)
|
||||
if not term or not term.is_alive():
|
||||
raise KeyError("terminal not running")
|
||||
os.write(term.master_fd, str(data or "").encode("utf-8", errors="replace"))
|
||||
# Re-check ``closed`` under io_lock and write while holding it, so the fd
|
||||
# can't be closed (and its number recycled by another openpty) between the
|
||||
# check and the write — which would inject this input into a foreign fd.
|
||||
with term.io_lock:
|
||||
if term.closed.is_set():
|
||||
raise KeyError("terminal not running")
|
||||
os.write(term.master_fd, str(data or "").encode("utf-8", errors="replace"))
|
||||
term.last_activity = time.time()
|
||||
|
||||
|
||||
def resize_terminal(session_id: str, rows: int, cols: int) -> None:
|
||||
@@ -414,11 +496,21 @@ def resize_terminal(session_id: str, rows: int, cols: int) -> None:
|
||||
_set_size(term, rows, cols)
|
||||
|
||||
|
||||
def close_terminal(session_id: str) -> bool:
|
||||
def close_terminal(session_id: str, *, expected: TerminalSession | None = None) -> bool:
|
||||
"""Tear down the terminal for *session_id*: kill the shell, close the pty
|
||||
master fd, reap descendants, and drop the ``_TERMINALS`` entry.
|
||||
|
||||
``expected`` guards the retire-from-reader-loop path: only act if the live
|
||||
entry is still that exact terminal, so an old reader thread finishing after
|
||||
a restart cannot tear down the *new* terminal that replaced it (the old
|
||||
one's fd was already closed by the restart's own close_terminal call).
|
||||
"""
|
||||
if not _TERMINAL_SUPPORTED:
|
||||
return False
|
||||
sid = str(session_id or "")
|
||||
with _LOCK:
|
||||
if expected is not None and _TERMINALS.get(sid) is not expected:
|
||||
return False
|
||||
term = _TERMINALS.pop(sid, None)
|
||||
if not term:
|
||||
return False
|
||||
@@ -441,10 +533,13 @@ def close_terminal(session_id: str) -> bool:
|
||||
except (subprocess.TimeoutExpired, ProcessLookupError):
|
||||
pass
|
||||
finally:
|
||||
try:
|
||||
os.close(term.master_fd)
|
||||
except OSError:
|
||||
pass
|
||||
# ``closed`` is already set above, so a writer/resizer blocked on io_lock
|
||||
# will see it and bail rather than touch the fd we are about to close.
|
||||
with term.io_lock:
|
||||
try:
|
||||
os.close(term.master_fd)
|
||||
except OSError:
|
||||
pass
|
||||
_reap_terminal_descendants(term.proc.pid)
|
||||
return True
|
||||
|
||||
|
||||
@@ -0,0 +1,105 @@
|
||||
"""Regression tests — atomic JSON writers fsync and use collision-safe temps.
|
||||
|
||||
Three writers wrote to a temp file and os.replace'd it into place but never
|
||||
fsynced, so a power loss between the write and the physical flush could leave a
|
||||
zero-length/garbage file where durable JSON should be:
|
||||
- session_discoverability._atomic_write_json (also used a pid-only temp name,
|
||||
which collides between two request threads of the same ThreadingHTTPServer
|
||||
process writing the same _index.json),
|
||||
- passkeys._atomic_write_json (credentials),
|
||||
- oauth._write_auth_json (access/refresh tokens).
|
||||
|
||||
Each now fsyncs before the rename; session_discoverability additionally puts the
|
||||
thread id in the temp name and cleans it up on failure. These pin the fsync, the
|
||||
per-thread temp name, and that content/permissions are unchanged.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
def _spy_fsync(monkeypatch, module):
|
||||
calls = []
|
||||
real = os.fsync
|
||||
monkeypatch.setattr(module.os, "fsync", lambda fd: (calls.append(fd), real(fd))[1])
|
||||
return calls
|
||||
|
||||
|
||||
def test_session_discoverability_writer_fsyncs_and_roundtrips(tmp_path, monkeypatch):
|
||||
import api.session_discoverability as sd
|
||||
|
||||
calls = _spy_fsync(monkeypatch, sd)
|
||||
path = tmp_path / "_index.json"
|
||||
payload = [{"session_id": "s1", "title": "Grüße 🌍"}]
|
||||
|
||||
sd._atomic_write_json(path, payload)
|
||||
|
||||
assert calls, "atomic write did not fsync before rename"
|
||||
assert json.loads(path.read_text(encoding="utf-8")) == payload
|
||||
assert [p.name for p in tmp_path.iterdir()] == ["_index.json"] # no temp debris
|
||||
|
||||
|
||||
def test_session_discoverability_temp_name_is_thread_unique(monkeypatch):
|
||||
"""Two threads (same pid) must not collide on one temp path."""
|
||||
import api.session_discoverability as sd
|
||||
|
||||
names = set()
|
||||
real_replace = os.replace
|
||||
|
||||
def capture(src, dst):
|
||||
names.add(Path(src).name)
|
||||
return real_replace(src, dst)
|
||||
|
||||
monkeypatch.setattr(sd.os, "replace", capture)
|
||||
|
||||
import tempfile as _tf
|
||||
d = Path(_tf.mkdtemp())
|
||||
idents = iter([1111, 2222])
|
||||
monkeypatch.setattr(sd.threading, "get_ident", lambda: next(idents))
|
||||
sd._atomic_write_json(d / "_index.json", [{"a": 1}])
|
||||
sd._atomic_write_json(d / "_index.json", [{"a": 2}])
|
||||
|
||||
assert len(names) == 2, f"temp names collided across threads: {names}"
|
||||
|
||||
|
||||
def test_session_discoverability_failed_write_leaves_no_debris(tmp_path, monkeypatch):
|
||||
import api.session_discoverability as sd
|
||||
|
||||
original = tmp_path / "_index.json"
|
||||
original.write_text('["keep"]', encoding="utf-8")
|
||||
|
||||
def boom(src, dst):
|
||||
raise RuntimeError("simulated crash before rename")
|
||||
|
||||
monkeypatch.setattr(sd.os, "replace", boom)
|
||||
with pytest.raises(RuntimeError):
|
||||
sd._atomic_write_json(original, [{"new": True}])
|
||||
|
||||
assert original.read_text(encoding="utf-8") == '["keep"]' # original intact
|
||||
assert [p.name for p in tmp_path.iterdir()] == ["_index.json"] # temp cleaned up
|
||||
|
||||
|
||||
def test_passkeys_writer_fsyncs_and_keeps_0600(tmp_path, monkeypatch):
|
||||
import api.passkeys as passkeys
|
||||
|
||||
calls = _spy_fsync(monkeypatch, passkeys)
|
||||
path = tmp_path / "passkeys.json"
|
||||
passkeys._atomic_write_json(path, [{"id": "cred-1"}])
|
||||
|
||||
assert calls, "passkeys write did not fsync"
|
||||
assert json.loads(path.read_text(encoding="utf-8")) == [{"id": "cred-1"}]
|
||||
assert (os.stat(path).st_mode & 0o777) == 0o600 # owner-only preserved
|
||||
|
||||
|
||||
def test_oauth_writer_fsyncs_and_keeps_0600(tmp_path, monkeypatch):
|
||||
import api.oauth as oauth
|
||||
|
||||
calls = _spy_fsync(monkeypatch, oauth)
|
||||
path = tmp_path / "auth.json"
|
||||
oauth._write_auth_json({"access_token": "x"}, auth_path=path)
|
||||
|
||||
assert calls, "oauth auth.json write did not fsync"
|
||||
assert json.loads(path.read_text(encoding="utf-8")) == {"access_token": "x"}
|
||||
assert (os.stat(path).st_mode & 0o777) == 0o600 # owner-only preserved
|
||||
@@ -0,0 +1,222 @@
|
||||
"""Regression tests for the embedded-terminal resource leak (#4633).
|
||||
|
||||
Two leaks are closed here:
|
||||
|
||||
1. **fd + dict entry on shell exit.** When a shell exited on its own (user typed
|
||||
`exit`, the process died), `_reader_loop` broke out of its loop but only set
|
||||
`closed` and emitted `terminal_closed`; it never closed the pty `master_fd`
|
||||
nor removed the `_TERMINALS` entry. Only an explicit `close_terminal` /
|
||||
restart / atexit did that, so a self-exited shell leaked its fd and dict
|
||||
entry for the rest of the WebUI's uptime. The reader loop now retires the
|
||||
session (identity-guarded so a restart's replacement is never touched).
|
||||
|
||||
2. **Unbounded accumulation of abandoned terminals.** A client that drops its
|
||||
output stream without POSTing /api/terminal/close (tab close, crash, network
|
||||
drop) leaves the shell running (no PDEATHSIG, by design). Without a ceiling
|
||||
these pile up toward fd/thread exhaustion. `_MAX_TERMINALS` now caps the live
|
||||
population, evicting the least-recently-active terminal to make room.
|
||||
|
||||
These use fakes for the shell/pty so they run without spawning real processes;
|
||||
the real-shell path is covered by test_terminal_linux_lifecycle.
|
||||
"""
|
||||
import os
|
||||
|
||||
import pytest
|
||||
|
||||
if os.name != "posix":
|
||||
pytest.skip("terminal tests require POSIX terminal support", allow_module_level=True)
|
||||
|
||||
import api.terminal as terminal
|
||||
|
||||
|
||||
class _FakeProc:
|
||||
"""A shell process fake. ``alive`` controls poll()."""
|
||||
|
||||
def __init__(self, pid=424242, alive=True):
|
||||
self.pid = pid
|
||||
self._alive = alive
|
||||
self.wait_calls = []
|
||||
|
||||
def poll(self):
|
||||
return None if self._alive else 0
|
||||
|
||||
def wait(self, timeout=None):
|
||||
self.wait_calls.append(timeout)
|
||||
return 0
|
||||
|
||||
|
||||
def _make_registered_term(monkeypatch, sid, *, alive=True, last_activity=0.0, real_fd=True):
|
||||
"""Build a TerminalSession, register it in _TERMINALS, no reader thread."""
|
||||
if real_fd:
|
||||
r, w = os.pipe()
|
||||
os.close(w) # leave a real, closable fd as master_fd
|
||||
master_fd = r
|
||||
else:
|
||||
master_fd = -1
|
||||
term = terminal.TerminalSession(
|
||||
session_id=sid,
|
||||
workspace="/tmp",
|
||||
proc=_FakeProc(alive=alive),
|
||||
master_fd=master_fd,
|
||||
last_activity=last_activity,
|
||||
)
|
||||
with terminal._LOCK:
|
||||
terminal._TERMINALS[sid] = term
|
||||
return term
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _clean_terminals(monkeypatch):
|
||||
# Never kill real process groups in these unit tests.
|
||||
monkeypatch.setattr(terminal.os, "killpg", lambda *a, **k: None)
|
||||
yield
|
||||
with terminal._LOCK:
|
||||
sids = list(terminal._TERMINALS)
|
||||
for sid in sids:
|
||||
try:
|
||||
terminal.close_terminal(sid)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
# ── Leak 1: reader loop retires the session on shell exit ────────────────────
|
||||
|
||||
def test_reader_loop_retires_session_and_closes_fd(monkeypatch):
|
||||
sid = "leak-reader-exit"
|
||||
term = _make_registered_term(monkeypatch, sid, alive=False)
|
||||
fd = term.master_fd
|
||||
|
||||
# Run the reader loop directly: a dead proc makes it exit its first iteration.
|
||||
terminal._reader_loop(term)
|
||||
|
||||
with terminal._LOCK:
|
||||
assert sid not in terminal._TERMINALS, "entry not retired on shell exit"
|
||||
with pytest.raises(OSError):
|
||||
os.fstat(fd) # master_fd was closed — no leak
|
||||
|
||||
|
||||
def test_reader_loop_retire_is_identity_guarded_against_restart(monkeypatch):
|
||||
"""An old reader thread finishing after a restart must not tear down the
|
||||
new terminal that replaced its session id."""
|
||||
sid = "leak-restart-race"
|
||||
old = _make_registered_term(monkeypatch, sid, alive=False)
|
||||
# Simulate a restart: a NEW terminal now occupies the same sid.
|
||||
new = _make_registered_term(monkeypatch, sid, alive=True)
|
||||
assert terminal._TERMINALS[sid] is new
|
||||
|
||||
# The OLD reader loop finishes now.
|
||||
terminal._reader_loop(old)
|
||||
|
||||
# The new terminal must still be registered and its fd open.
|
||||
assert terminal._TERMINALS.get(sid) is new
|
||||
os.fstat(new.master_fd) # not closed
|
||||
|
||||
|
||||
# ── Leak 2: _MAX_TERMINALS cap evicts the least-recently-active terminal ──────
|
||||
|
||||
def test_cap_evicts_least_recently_active(monkeypatch):
|
||||
monkeypatch.setattr(terminal, "_MAX_TERMINALS", 3)
|
||||
# Fill to the cap with alive terminals of increasing activity.
|
||||
terms = {}
|
||||
for i in range(3):
|
||||
terms[i] = _make_registered_term(
|
||||
monkeypatch, f"cap-{i}", alive=True, last_activity=100.0 + i
|
||||
)
|
||||
# cap-0 is least-recently-active. Enforcing the cap for a NEW sid evicts it.
|
||||
terminal._enforce_terminal_cap(exclude_sid="cap-new")
|
||||
|
||||
with terminal._LOCK:
|
||||
live = set(terminal._TERMINALS)
|
||||
assert "cap-0" not in live, "least-recently-active terminal was not evicted"
|
||||
assert {"cap-1", "cap-2"} <= live
|
||||
assert len(live) < 3 # room was made for the new terminal
|
||||
|
||||
|
||||
def test_cap_prefers_dead_terminals_for_eviction(monkeypatch):
|
||||
monkeypatch.setattr(terminal, "_MAX_TERMINALS", 3)
|
||||
# A dead terminal that is NOT the least-recently-active must still go first.
|
||||
_make_registered_term(monkeypatch, "cap-dead", alive=False, last_activity=999.0)
|
||||
_make_registered_term(monkeypatch, "cap-live-a", alive=True, last_activity=1.0)
|
||||
_make_registered_term(monkeypatch, "cap-live-b", alive=True, last_activity=2.0)
|
||||
|
||||
terminal._enforce_terminal_cap(exclude_sid="cap-new")
|
||||
|
||||
with terminal._LOCK:
|
||||
live = set(terminal._TERMINALS)
|
||||
assert "cap-dead" not in live, "dead terminal not preferred for eviction"
|
||||
assert {"cap-live-a", "cap-live-b"} <= live
|
||||
|
||||
|
||||
def test_cap_reuse_of_existing_sid_evicts_nothing(monkeypatch):
|
||||
monkeypatch.setattr(terminal, "_MAX_TERMINALS", 2)
|
||||
_make_registered_term(monkeypatch, "keep-a", alive=True, last_activity=1.0)
|
||||
_make_registered_term(monkeypatch, "keep-b", alive=True, last_activity=2.0)
|
||||
|
||||
# Reusing an already-registered sid replaces in place — no growth, no evict.
|
||||
terminal._enforce_terminal_cap(exclude_sid="keep-a")
|
||||
|
||||
with terminal._LOCK:
|
||||
assert {"keep-a", "keep-b"} <= set(terminal._TERMINALS)
|
||||
|
||||
|
||||
# ── F1: writes/resizes are serialized against close (no fd-reuse injection) ───
|
||||
|
||||
def test_write_after_close_raises_and_never_touches_fd(monkeypatch):
|
||||
"""A write racing a teardown must not reach os.write once the terminal is
|
||||
closed — otherwise the recycled fd number could receive foreign input."""
|
||||
sid = "io-write-after-close"
|
||||
term = _make_registered_term(monkeypatch, sid, alive=True)
|
||||
|
||||
calls = []
|
||||
monkeypatch.setattr(terminal.os, "write", lambda fd, data: calls.append(fd))
|
||||
|
||||
# Simulate the teardown having marked the terminal closed.
|
||||
term.closed.set()
|
||||
with pytest.raises(KeyError):
|
||||
terminal.write_terminal(sid, "rm -rf /\n")
|
||||
assert calls == [], "write reached os.write after close — fd-reuse risk"
|
||||
|
||||
|
||||
def test_close_acquires_io_lock_before_closing_fd(monkeypatch):
|
||||
"""close_terminal must take io_lock around os.close so a concurrent writer
|
||||
holding io_lock finishes (or bails) first — proving the two are serialized."""
|
||||
import threading
|
||||
|
||||
sid = "io-close-serialized"
|
||||
term = _make_registered_term(monkeypatch, sid, alive=False)
|
||||
fd = term.master_fd
|
||||
|
||||
closed_fd = []
|
||||
monkeypatch.setattr(terminal.os, "close", lambda f: closed_fd.append(f))
|
||||
|
||||
# Hold io_lock, then kick off a close in another thread and confirm it blocks
|
||||
# on the fd-close until we release.
|
||||
with term.io_lock:
|
||||
t = threading.Thread(target=lambda: terminal.close_terminal(sid, expected=term))
|
||||
t.start()
|
||||
t.join(timeout=0.3)
|
||||
assert closed_fd == [], "close closed the fd without waiting for io_lock"
|
||||
t.join(timeout=2.0)
|
||||
assert closed_fd == [fd], "close did not close the fd after io_lock released"
|
||||
|
||||
|
||||
# ── F2: cap eviction is identity-guarded ─────────────────────────────────────
|
||||
|
||||
def test_cap_eviction_uses_expected_guard(monkeypatch):
|
||||
monkeypatch.setattr(terminal, "_MAX_TERMINALS", 1)
|
||||
victim = _make_registered_term(monkeypatch, "cap-victim", alive=True, last_activity=1.0)
|
||||
|
||||
seen = {}
|
||||
|
||||
real_close = terminal.close_terminal
|
||||
|
||||
def _spy_close(sid, *, expected=None):
|
||||
seen["sid"] = sid
|
||||
seen["expected"] = expected
|
||||
return real_close(sid, expected=expected)
|
||||
|
||||
monkeypatch.setattr(terminal, "close_terminal", _spy_close)
|
||||
terminal._enforce_terminal_cap(exclude_sid="cap-new")
|
||||
|
||||
assert seen.get("sid") == "cap-victim"
|
||||
assert seen.get("expected") is victim, "cap eviction must pass expected= for identity safety"
|
||||
@@ -366,9 +366,16 @@ def test_terminal_supervisor_continues_after_mixed_popen_failures(monkeypatch, t
|
||||
term_2 = start_terminal(sid_ok_2, tmp_path, restart=True)
|
||||
|
||||
try:
|
||||
# Both successful spawns produced distinct, registered terminals. We
|
||||
# assert on the returned objects rather than re-reading _TERMINALS[sid]:
|
||||
# these FakeProcs never hold the pty slave open, so the master EOFs
|
||||
# immediately and the reader loop now correctly retires the entry (the
|
||||
# fd-leak fix) — a real, alive shell keeps its pty open and stays
|
||||
# registered (covered by the real-shell lifecycle tests above).
|
||||
assert term_1 is not term_2
|
||||
assert terminal._TERMINALS[sid_ok_1].proc is term_1.proc
|
||||
assert terminal._TERMINALS[sid_ok_2].proc is term_2.proc
|
||||
assert term_1.proc is not term_2.proc
|
||||
assert isinstance(term_1, terminal.TerminalSession)
|
||||
assert isinstance(term_2, terminal.TerminalSession)
|
||||
assert terminal._spawn_supervisor_thread is not None
|
||||
assert terminal._spawn_supervisor_thread.is_alive()
|
||||
finally:
|
||||
|
||||
Reference in new issue
Block a user