Skip to content

fix(gateway): quiesce the thread pool before closing state.db at shutdown #101118

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kshitijk4poor merged 2 commits into from
Sep 2, 2026
Merged

fix(gateway): quiesce the thread pool before closing state.db at shutdown #101118

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
49dcb59
fix(gateway): quiesce the thread pool before closing state.db at shut…
JoaoMarcos44 Sep 2, 2026
600dbec
fix(gateway): fail closed when a shutdown worker survives the quiesce…
JoaoMarcos44 Sep 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
190 changes: 138 additions & 52 deletions gateway/run.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3083,6 +3083,13 @@ def _platform_has_bot_credential(platform: "Platform", platform_config: "Platfor
logger = logging.getLogger(__name__)


# Ceiling for the shutdown quiesce of the gateway-owned thread pool. Drain has
# already waited for the agents, so what is left here is short blocking work
# (a transcript append, a routing save); anything slower is a stuck worker we
# must not wait on, and the caller clamps this to the watchdog leash anyway.
_EXECUTOR_QUIESCE_TIMEOUT = 2.0


_OWN_POLICY_OPEN_ENV = {
Platform.WECOM: ("WECOM_DM_POLICY", "WECOM_GROUP_POLICY", "WECOM_ALLOW_ALL_USERS"),
Platform.WEIXIN: ("WEIXIN_DM_POLICY", "WEIXIN_GROUP_POLICY", "WEIXIN_ALLOW_ALL_USERS"),
Expand Down Expand Up @@ -16578,58 +16585,115 @@ def _phase_elapsed() -> float:
except Exception as _e:
logger.debug("shutdown_cached_clients error: %s", _e)

# Close SQLite session DBs so the WAL write lock is released.
# Without this, --replace and similar restart flows leave the
# old gateway's connection holding the WAL lock until Python
# actually exits — causing 'database is locked' errors when
# the new gateway tries to open the same file.
# ``self`` holds the DB at ``_session_db`` (an AsyncSessionDB facade);
# unwrap to the sync handle. ``session_store`` holds it at ``_db``.
_self_db = getattr(self, "_session_db", None)
_self_db = getattr(_self_db, "_db", _self_db)
for _db in (_self_db, getattr(getattr(self, "session_store", None), "_db", None)):
if _db is None or not hasattr(_db, "close"):
continue
# Quiesce the gateway thread pool BEFORE the session databases
# are closed. This used to run *after* the close block below,
# which left two holes:
#
# (a) `_executor_closing` was still False during the close, so
# any coroutine reaching `_run_in_executor_with_context`
# minted a brand-new pool and ran more blocking DB work
# against handles that had just been closed;
# (b) cancelling `self._background_tasks` above does not stop a
# `run_in_executor` future that already started — the task
# dies, the worker thread keeps writing.
#
# Either way a write lands after `SessionDB.close()`, which has
# already checkpointed the WAL and let SQLite unlink the sidecar.
# The late write silently reopens the handle (#94736) and mints a
# fresh WAL generation behind that checkpoint, so teardown
# checkpoints the same file a second time from a connection the
# shutdown log never accounts for — the close-time page-write
# damage in #101093 and the split WAL generation in #101064.
#
# The wait is bounded and clamped to what is left of the shutdown
# watchdog leash (minus a second for the close itself), so a stuck
# worker can never cost us the post-close cleanup window (#82161).
_exec_quiesce_budget = max(
0.0,
min(
_EXECUTOR_QUIESCE_TIMEOUT,
resolve_shutdown_watchdog_delay(timeout)
- _phase_elapsed()
- 1.0,
),
)
_exec_live = GatewayRunner._shutdown_executor(
self, drain_timeout=_exec_quiesce_budget
)
if _exec_live:
# A live worker can still be mid-write against a SessionDB
# handle. Checkpointing/closing it now is exactly the
# sequence that produced the wrong-page-number corruption in
# #101093, so the close path below is skipped entirely
# rather than raced — the handle is left open for SQLite to
# recover from its own WAL on the next open, which is a
# transient "database is locked" on an immediate --replace
# at worst, not a corrupt file.
logger.warning(
"Shutdown phase: %d executor worker(s) still running after "
"a %.2fs quiesce — skipping the SessionDB close/checkpoint "
"to avoid racing a live write (#101093); handles are left "
"open for SQLite to recover on next open",
_exec_live,
_exec_quiesce_budget,
)
else:
logger.info(
"Shutdown phase: executor quiesced at +%.2fs",
_phase_elapsed(),
)

# Close SQLite session DBs so the WAL write lock is released.
# Without this, --replace and similar restart flows leave the
# old gateway's connection holding the WAL lock until Python
# actually exits — causing 'database is locked' errors when
# the new gateway tries to open the same file.
# ``self`` holds the DB at ``_session_db`` (an AsyncSessionDB facade);
# unwrap to the sync handle. ``session_store`` holds it at ``_db``.
_self_db = getattr(self, "_session_db", None)
_self_db = getattr(_self_db, "_db", _self_db)
for _db in (_self_db, getattr(getattr(self, "session_store", None), "_db", None)):
if _db is None or not hasattr(_db, "close"):
continue
try:
_db.close()
except Exception as _e:
logger.debug("SessionDB close error: %s", _e)
# A multiplexed session_store caches one SessionDB per profile
# path (#88532); reading ``_db`` above only resolved the handle
# for the shutdown task's own (root) scope. Sweep the rest so
# secondary profiles' WAL locks are released before --replace
# brings a new gateway up on the same files.
_sweep = getattr(
getattr(self, "session_store", None), "close_all_db_handles", None
)
if _sweep is not None:
try:
_sweep()
except Exception as _e:
logger.debug("SessionDB handle sweep error: %s", _e)
# Same sweep for the runner's own per-profile session_search
# handles (slash commands resolve them under profile scopes).
try:
_db.close()
GatewayRunner.close_all_session_db_handles(self)
except Exception as _e:
logger.debug("SessionDB close error: %s", _e)
# A multiplexed session_store caches one SessionDB per profile
# path (#88532); reading ``_db`` above only resolved the handle
# for the shutdown task's own (root) scope. Sweep the rest so
# secondary profiles' WAL locks are released before --replace
# brings a new gateway up on the same files.
_sweep = getattr(
getattr(self, "session_store", None), "close_all_db_handles", None
)
if _sweep is not None:
logger.debug("Runner SessionDB handle sweep error: %s", _e)
# Final sweep: close any shared SessionDB instances still held by
# the process-wide registry (in-process tools, cron, mirror, etc.
# that opened via get_shared_session_db but weren't released by
# the sweeps above). This is the safety net that guarantees no
# WAL write lock survives past gateway shutdown (#90837).
try:
_sweep()
from hermes_state import close_shared_session_dbs
closed = close_shared_session_dbs()
if closed:
logger.debug("Closed %d shared SessionDB instance(s) at shutdown", closed)
except Exception as _e:
logger.debug("SessionDB handle sweep error: %s", _e)
# Same sweep for the runner's own per-profile session_search
# handles (slash commands resolve them under profile scopes).
try:
GatewayRunner.close_all_session_db_handles(self)
except Exception as _e:
logger.debug("Runner SessionDB handle sweep error: %s", _e)
# Final sweep: close any shared SessionDB instances still held by
# the process-wide registry (in-process tools, cron, mirror, etc.
# that opened via get_shared_session_db but weren't released by
# the sweeps above). This is the safety net that guarantees no
# WAL write lock survives past gateway shutdown (#90837).
try:
from hermes_state import close_shared_session_dbs
closed = close_shared_session_dbs()
if closed:
logger.debug("Closed %d shared SessionDB instance(s) at shutdown", closed)
except Exception as _e:
logger.debug("Shared SessionDB close error: %s", _e)
GatewayRunner._shutdown_executor(self)
logger.info(
"Shutdown phase: SessionDB close done at +%.2fs",
_phase_elapsed(),
)
logger.debug("Shared SessionDB close error: %s", _e)
logger.info(
"Shutdown phase: SessionDB close done at +%.2fs",
_phase_elapsed(),
)

from gateway.status import remove_pid_file, release_gateway_runtime_lock
remove_pid_file()
Expand Down Expand Up @@ -26966,25 +27030,47 @@ def _get_executor(self) -> concurrent.futures.ThreadPoolExecutor:
self._executor = executor
return executor

def _shutdown_executor(self) -> None:
"""Stop the gateway-owned executor without touching the loop default."""
def _shutdown_executor(self, drain_timeout: float = 0.0) -> int:
"""Stop the gateway-owned executor without touching the loop default.

Returns the number of worker threads still running when this returns.
With the default ``drain_timeout`` of 0 this is the historical
fire-and-forget teardown; shutdown passes a bounded budget so blocking
DB work cannot outlive ``SessionDB.close()`` (see ``_stop_impl``).

``cancel_futures`` only drops work that has not started yet, and a
cancelled ``run_in_executor`` awaitable does not stop the thread behind
it, so the running futures have to be waited on explicitly.
"""
lock = getattr(self, "_executor_lock", None)
if lock is None:
return
return 0

with lock:
self._executor_closing = True
executor = getattr(self, "_executor", None)
self._executor = None

if executor is None:
return
return 0

try:
executor.shutdown(wait=False, cancel_futures=True)
except TypeError:
executor.shutdown(wait=False)

# ThreadPoolExecutor.shutdown() has no timeout, so join the worker
# threads directly. `_threads` is absent on the doubles some tests
# pass in, which just means no wait.
workers = list(getattr(executor, "_threads", None) or ())
deadline = time.monotonic() + max(float(drain_timeout or 0.0), 0.0)
for worker in workers:
remaining = deadline - time.monotonic()
if remaining <= 0:
break
worker.join(remaining)
return sum(1 for worker in workers if worker.is_alive())

def _decide_image_input_mode(
self,
*,
Expand Down
Loading
Loading