Coverage for src/local_deep_research/web/auth/connection_cleanup.py: 96%
126 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1"""
2Automatic cleanup of idle database connections.
4Periodically closes database connections for users who have no active sessions
5and no active research, preventing resource leaks when users close their browser
6without logging out.
8Also periodically disposes all QueuePool engines to release accumulated WAL/SHM
9file handles. See ADR-0004 for why this is necessary with SQLCipher + WAL mode.
10"""
12import os
13import time
14from pathlib import Path
16from apscheduler.schedulers.background import BackgroundScheduler
17from loguru import logger
19from ...database.session_passwords import session_password_store
20from ...database.thread_local_session import (
21 cleanup_dead_threads,
22 clear_user_credentials,
23)
24from ...web.research_state import get_usernames_with_active_research
26# ---------------------------------------------------------------------------
27# File Descriptor Monitoring
28# ---------------------------------------------------------------------------
29# WHY: After days of idle operation in Docker, the app crashed with
30# OSError: [Errno 24] Too many open files
31# This monitoring logs the FD count every 5 minutes so we can correlate
32# FD growth with specific events and find leaks.
33#
34# WHAT IT LOGS:
35# - open_fds: total open file descriptors for the process
36# - pool_engines: number of per-user QueuePool engines
37# - pool_checked_out: connections currently checked out from QueuePool
38# - protected_users: users with active sessions
39#
40# HOW TO USE: grep "Resource monitor" in container logs. If open_fds
41# grows steadily over hours, something is leaking.
42# ---------------------------------------------------------------------------
44# Dispose all pool engines every 30 minutes to release WAL/SHM handles.
45# SQLCipher + WAL mode leaks handles when connections close out of order
46# (which QueuePool's pool_recycle causes). Periodic dispose() closes ALL
47# pooled connections at once, resetting the handle state cleanly.
48# The next DB operation transparently reopens a fresh connection.
49_DISPOSE_INTERVAL_SECONDS = 1800
50_last_dispose_time = 0.0
53def _pop_per_user_locks(username: str) -> None:
54 """Release safe-to-remove per-user lock-cache entries for ``username``.
56 The library-init, backup, queue-processor, and library-RAG modules each
57 maintain per-user lock registries. Plain locks cannot be safely removed
58 because a caller may already hold a reference before acquisition, so their
59 compatibility cleanup hooks retain stable identity. The library-RAG cache
60 uses tracked acquisition and can safely evict idle entries.
62 The research-start gate, queue admission view, library-init lock and backup
63 lock are bounded by the user population.
65 Lazy-imported here to keep this module's import graph shallow:
66 ``connection_cleanup`` runs at startup and shouldn't pull in the
67 queue / backup / library-init / library-RAG modules eagerly.
68 """
69 try:
70 from ...database.library_init import pop_user_init_lock
72 pop_user_init_lock(username)
73 except Exception:
74 # Surface at WARNING to match the sibling scheduler-unregister
75 # error handler in this same module (line ~111). A failure
76 # here means the lock-dict entry will accumulate on every
77 # subsequent close cycle for this user; we want it visible.
78 logger.warning(f"Failed to pop _user_init_locks for {username}")
80 try:
81 from ...database.backup.backup_service import pop_user_lock
83 pop_user_lock(username)
84 except Exception:
85 logger.warning(f"Failed to pop _user_locks for {username}")
87 try:
88 from ...web.queue.processor_v2 import queue_processor
90 queue_processor.pop_user_critical_lock(username)
91 except Exception:
92 logger.warning(f"Failed to pop _user_critical_locks for {username}")
94 # NOTE: the per-user research-start gate (_user_research_start_gates in
95 # web/research_state.py) is deliberately NOT popped here — it may be held
96 # across a multi-second rekey and removing a held gate would let a
97 # concurrent same-user check_and_start_research create a second gate
98 # instance and bypass the exclusion. See the docstring above.
100 try:
101 from ...research_library.services.library_rag_service import (
102 pop_faiss_locks_for_user,
103 )
105 pop_faiss_locks_for_user(username)
106 except Exception:
107 logger.warning(f"Failed to pop _faiss_write_locks for {username}")
110def _disconnect_all_user_sockets(username: str) -> None:
111 """Best-effort disconnect of ALL of ``username``'s live sockets.
113 Called from the idle-connection sweep, where the user has no active
114 session at all — so every one of their still-open sockets (authorised
115 once at handshake and never re-checked) should be severed. Lazy-imported
116 to keep this module's startup import graph shallow and to tolerate
117 non-web contexts (the socket server may not exist).
119 Retargeted from the deleted Flask ``SocketIOService`` (#5535) onto the
120 ASGI socket layer. Behaviour is equivalent: ``disconnect_user`` severs
121 every sid authenticated as this user, and the resulting ``disconnect``
122 handler drops their subscriptions. It schedules onto the main loop and
123 returns False rather than raising when no loop is running, which is
124 the non-web case main handled by catching ValueError.
125 """
126 try:
127 from ...web.services.socketio_asgi import disconnect_user
129 disconnect_user(username)
130 except Exception:
131 logger.warning(f"Failed to disconnect sockets for idle user {username}")
134def _count_open_fds() -> int:
135 """Count open file descriptors for the current process."""
136 proc_fd = Path("/proc/self/fd")
137 if proc_fd.is_dir():
138 try:
139 return len(list(proc_fd.iterdir()))
140 except OSError:
141 pass
142 import resource
144 soft_limit = resource.getrlimit(resource.RLIMIT_NOFILE)[0]
145 count = 0
146 for fd in range(soft_limit):
147 try:
148 os.fstat(fd)
149 count += 1
150 except OSError:
151 pass
152 return count
155def cleanup_idle_connections(session_manager, db_manager):
156 """Close db connections for users with no active sessions and no active research."""
157 # 1. Purge expired sessions first
158 session_manager.cleanup_expired_sessions()
160 # 2. Get protected usernames (active sessions OR active research)
161 active_usernames = session_manager.get_active_usernames()
162 researching_usernames = get_usernames_with_active_research()
163 protected = active_usernames | researching_usernames
165 # 3. Get usernames with open connections
166 connected_usernames = db_manager.get_connected_usernames()
168 # 4. Find idle candidates
169 candidates = connected_usernames - protected
171 # 5. Double-check before closing (narrows race window)
172 closed = 0
173 for username in candidates:
174 if session_manager.has_active_sessions_for(username):
175 logger.debug(
176 f"Skipped {username} (active session appeared since snapshot)"
177 )
178 continue # User logged in since snapshot
179 if username in get_usernames_with_active_research():
180 logger.debug(
181 f"Skipped {username} (active research appeared since snapshot)"
182 )
183 continue # Research started since snapshot
184 # Unregister news scheduler jobs (matches logout pattern in routes.py)
185 try:
186 from ...scheduler.background import (
187 get_background_job_scheduler,
188 )
190 sched = get_background_job_scheduler()
191 if sched.is_running:
192 sched.unregister_user(username)
193 except Exception:
194 logger.warning(
195 f"Failed to unregister scheduler for {username}",
196 )
197 try:
198 db_manager.close_user_database(username)
199 session_password_store.clear_all_for_user(username)
200 closed += 1
201 logger.debug(f"Closed idle connection for {username}")
202 except Exception:
203 logger.warning(f"Connection cleanup failed for {username}")
204 # Drop cached plaintext credentials on pooled worker threads, for the
205 # same reason logout does. This sweep is the teardown path for the
206 # MAJORITY of users — most close the tab rather than clicking logout —
207 # so omitting it left the SQLCipher master key in a process-global
208 # dict indefinitely after the server had already decided the user was
209 # gone and closed their database. Outside the try above for the same
210 # reason _pop_per_user_locks is: independent of engine teardown, and
211 # it matters most on the path where close raises.
212 clear_user_credentials(username)
213 # This user has no active session at all (that's why we're closing
214 # their DB), so tear down every one of their still-open sockets — a
215 # socket authorised at handshake is never re-checked and would
216 # otherwise keep receiving the user's events after the session lapsed.
217 _disconnect_all_user_sockets(username)
218 # Run lock-cache cleanup regardless of whether close succeeded.
219 # Stable plain-lock registries keep their identities; tracked caches
220 # can evict idle entries. This remains independent of engine teardown.
221 _pop_per_user_locks(username)
223 if closed:
224 logger.info(f"Connection cleanup: closed {closed} idle connection(s)")
225 logger.debug(
226 f"Connection cleanup: evaluated {len(candidates)} candidate(s), "
227 f"closed {closed}, protected {len(protected)} active user(s)"
228 )
230 # Sweep dead-thread sessions and credentials — safety net when neither
231 # HTTP requests nor the queue processor are triggering sweeps.
232 cleanup_dead_threads()
234 # --- Periodic pool dispose to release WAL/SHM handles ---
235 # SQLCipher + WAL mode accumulates file handles when QueuePool recycles
236 # connections out of open-order (ADR-0004). Periodically calling
237 # dispose() on all engines closes ALL pooled connections, releasing any
238 # leaked handles. The pool is transparently recreated on the next DB
239 # operation.
240 #
241 # Safe to run against engines with checked-out connections: SA 2.0
242 # `QueuePool.dispose` only drains idle queue entries and
243 # `Engine.dispose` calls `pool.recreate()`; a thread holding a
244 # checked-out connection keeps using it until return. SA docs are
245 # explicit — "Connections that are still checked out will not be
246 # closed". The post-login bulk write (_perform_post_login_tasks in
247 # web/auth/routes.py) is additionally protected by being a single
248 # atomic transaction, so any interruption (dispose, crash, OOM)
249 # rolls back cleanly without leaving partial state.
250 #
251 # Do not add a `checkedout() > 0` skip guard here without first
252 # reproducing a real torn-write against the actual SA source path:
253 # see PR #3487 discussion — the speculative skip introduces an
254 # unbounded "skip forever" risk on busy engines in exchange for
255 # preventing a failure mode that SA 2.0 does not produce.
256 global _last_dispose_time
257 now = time.monotonic()
258 if now - _last_dispose_time >= _DISPOSE_INTERVAL_SECONDS:
259 _last_dispose_time = now
260 disposed = 0
261 with db_manager._connections_lock:
262 for username, engine in list(db_manager.connections.items()):
263 try:
264 db_manager._checkpoint_wal(engine, f"for {username}")
265 engine.dispose()
266 disposed += 1
267 except Exception as exc:
268 # Surface the failure. Pre-fix this was logger.debug,
269 # which hid the symptom — if WAL checkpoint or pool
270 # dispose repeatedly fails (disk pressure, lock
271 # starvation, etc.) the WAL file silently grows on
272 # disk and pooled connections leak. The 30-min
273 # periodic-dispose workaround for ADR-0004's WAL/SHM
274 # handle leak depends on this loop succeeding.
275 #
276 # Only the exception's TYPE NAME is logged, matching
277 # the codebase's `_report_silent_exception` pattern
278 # (utilities/log_utils.py:146-194). The exception
279 # value itself can carry sensitive locals (DB paths,
280 # query fragments, etc.) and our sensitive-logging
281 # hook flags any `f"...{exc}"` interpolation.
282 exc_type = type(exc).__name__
283 logger.warning(
284 f"Error disposing engine for {username}: {exc_type}"
285 )
286 if disposed:
287 logger.info(
288 f"Pool dispose: reset {disposed} engine(s) to release "
289 f"WAL/SHM handles"
290 )
292 # --- FD monitoring ---
293 try:
294 fd_count = _count_open_fds()
295 pool_engine_count = len(db_manager.connections)
296 pool_checked_out = 0
297 with db_manager._connections_lock:
298 for engine in db_manager.connections.values():
299 try:
300 pool_checked_out += engine.pool.checkedout()
301 except Exception: # noqa: silent-exception
302 pass
303 logger.debug(
304 f"Resource monitor: open_fds={fd_count}, "
305 f"pool_engines={pool_engine_count}, "
306 f"pool_checked_out={pool_checked_out}, "
307 f"protected_users={len(protected)}"
308 )
309 if fd_count > 800:
310 logger.warning(
311 f"High FD count ({fd_count}) — approaching system limit. "
312 f"Check for resource leaks."
313 )
314 except Exception:
315 logger.debug("FD monitoring failed") # noqa: silent-exception
318def start_connection_cleanup_scheduler(
319 session_manager, db_manager, interval_seconds=300
320):
321 """Start APScheduler job for periodic connection cleanup.
323 Args:
324 session_manager: The SessionManager singleton.
325 db_manager: The DatabaseManager singleton.
326 interval_seconds: How often to run cleanup (default: 5 minutes).
328 Returns:
329 The BackgroundScheduler instance (for shutdown registration).
330 """
331 scheduler = BackgroundScheduler()
332 scheduler.add_job(
333 cleanup_idle_connections,
334 "interval",
335 seconds=interval_seconds,
336 args=[session_manager, db_manager],
337 id="cleanup_idle_connections",
338 jitter=30,
339 )
340 scheduler.start()
341 logger.info(
342 f"Connection cleanup scheduler started "
343 f"(interval={interval_seconds}s, jitter=30s)"
344 )
345 return scheduler