aboutsummaryrefslogtreecommitdiffstats
path: root/idatui
diff options
context:
space:
mode:
authorblasty <blasty@local>2026-07-09 11:55:43 +0200
committerblasty <blasty@local>2026-07-09 11:55:43 +0200
commit21d2cbbfcdbb712b907573b784bc9d98d5c5e648 (patch)
tree8d2c94975c30e13501f0c8c8785230cae235ae98 /idatui
parentdomain/paging layer: FunctionIndex, block-cached DisasmModel, decompile, resolve (diff)
downloadida-tui-21d2cbbfcdbb712b907573b784bc9d98d5c5e648.tar.gz
ida-tui-21d2cbbfcdbb712b907573b784bc9d98d5c5e648.tar.xz
ida-tui-21d2cbbfcdbb712b907573b784bc9d98d5c5e648.zip
keep idle workers alive: bump_idle_ttl + KeepAlive heartbeat
Root cause: idalib workers run an idle watchdog (worker_lifecycle.py) that self-exits after idle_ttl_sec (default 600s) of no requests -- deliberate resource hygiene for a shared/ephemeral supervisor, wrong for an interactive TUI. CLI-opened startup binary always gets 600s (no flag). Fix (both verified against a 12s-TTL session): - IDAClient.bump_idle_ttl(1e9): re-open (idempotent) re-applies TTL, no upper cap -> effectively immortal. Primary fix, call once at startup. - IDAClient.keepalive()/KeepAlive: background server_health heartbeat resets the watchdog; covers adopted sessions too. Safety net. tests/test_keepalive.py: 4/4 (heartbeat + bump both keep a short-TTL worker alive past 2x TTL). docs updated with why + fix.
Diffstat (limited to 'idatui')
-rw-r--r--idatui/__init__.py2
-rw-r--r--idatui/client.py79
2 files changed, 81 insertions, 0 deletions
diff --git a/idatui/__init__.py b/idatui/__init__.py
index 8dd371c..24f3c0f 100644
--- a/idatui/__init__.py
+++ b/idatui/__init__.py
@@ -10,6 +10,7 @@ from .client import (
IDAToolError,
IDASessionError,
Session,
+ KeepAlive,
)
from .domain import (
Program,
@@ -42,4 +43,5 @@ __all__ = [
"IDAToolError",
"IDASessionError",
"Session",
+ "KeepAlive",
]
diff --git a/idatui/client.py b/idatui/client.py
index 44f5cc9..3e5edb4 100644
--- a/idatui/client.py
+++ b/idatui/client.py
@@ -522,6 +522,35 @@ class IDAClient:
def db(self) -> str | None:
return self._db
+ def _session_input_path(self, db: str) -> str:
+ for s in self.list_sessions():
+ if s.session_id == db:
+ if not s.input_path:
+ raise IDASessionError(f"session {db} has no input_path to re-open")
+ return s.input_path
+ raise IDASessionError(f"session {db} not found")
+
+ def bump_idle_ttl(self, idle_ttl_sec: int = 1_000_000_000,
+ path: str | None = None) -> None:
+ """Raise the worker's idle self-exit TTL so an interactive session never
+ gets reaped while the user is just reading.
+
+ Headless idalib workers self-exit after ``idle_ttl_sec`` (default 600s)
+ with no requests. ``idb_open`` on the already-open path is idempotent
+ (returns the same session) and re-applies the TTL, so this simply opens
+ the pinned session's path with a huge TTL. The default (~31 years) is
+ effectively 'never'.
+ """
+ p = path or self._session_input_path(self.resolve_db())
+ self.call("idb_open", input_path=p, idle_ttl_sec=int(idle_ttl_sec))
+
+ def keepalive(self, interval: float = 120.0) -> "KeepAlive":
+ """Return a (not-yet-started) heartbeat that touches the worker's idle
+ watchdog every ``interval`` seconds. Belt-and-suspenders next to
+ :meth:`bump_idle_ttl`; also covers adopted sessions we didn't open.
+ """
+ return KeepAlive(self, interval=interval)
+
# -- convenience ------------------------------------------------------- #
def health(self) -> dict:
return self.call("server_health")
@@ -542,6 +571,56 @@ class IDAClient:
raise IDAError(f"tool not found: {tool}")
+class KeepAlive:
+ """Background heartbeat that keeps an idalib worker from idling out.
+
+ Any forwarded request resets the worker's idle timer, so a periodic cheap
+ ``server_health`` is enough. Failures are swallowed (the next real call will
+ auto-recover); the heartbeat just keeps a chilling TUI's worker alive.
+ """
+
+ def __init__(self, client: "IDAClient", interval: float = 120.0):
+ if interval <= 0:
+ raise ValueError("interval must be > 0")
+ self._client = client
+ self.interval = interval
+ self._stop = threading.Event()
+ self._thread: threading.Thread | None = None
+ self.beats = 0
+ self.failures = 0
+
+ def start(self) -> "KeepAlive":
+ if self._thread is not None:
+ return self
+ self._stop.clear()
+ self._thread = threading.Thread(
+ target=self._run, daemon=True, name="idatui-keepalive"
+ )
+ self._thread.start()
+ return self
+
+ def stop(self) -> None:
+ self._stop.set()
+ t = self._thread
+ self._thread = None
+ if t is not None:
+ t.join(timeout=2.0)
+
+ def __enter__(self) -> "KeepAlive":
+ return self.start()
+
+ def __exit__(self, *exc) -> None:
+ self.stop()
+
+ def _run(self) -> None:
+ while not self._stop.wait(self.interval):
+ try:
+ self._client.health()
+ self.beats += 1
+ except IDAError:
+ self.failures += 1
+
+
def _is_stale_session(e: IDAToolError) -> bool:
msg = e.message.lower()
return any(marker in msg for marker in _STALE_SESSION_MARKERS)