Fix solver process reaping and SQLite deadlocks

Author: AuuCoder

app/control/account/backends/local.py

@@ -3,12 +3,12 @@
 import asyncio
 import json
 import sqlite3
-import threading
 from contextlib import closing, contextmanager
 from pathlib import Path
 from typing import Any
 
 from app.platform.runtime.clock import now_ms
+from utils.sqlite_runtime import sqlite_connection_guard
 from ..commands import AccountPatch, AccountUpsert, BulkReplacePoolCommand, ListAccountsQuery
 from ..enums import AccountStatus
 from ..models import (
@@ -38,10 +38,7 @@ class LocalAccountRepository:
     def __init__(self, db_path: Path) -> None:
         self._path = Path(db_path)
         self._lock = asyncio.Lock()
-        # Python's bundled SQLite can deadlock inside WAL open/close when many
-        # worker threads churn short-lived connections concurrently.  Keep the
-        # complete connection lifecycle serial for this repository instance.
-        self._connect_lock = threading.RLock()
+        self._connect_lock = sqlite_connection_guard
 
     # ------------------------------------------------------------------
     # Internal helpers

app/control/account/backends/local_test.py

@@ -44,14 +44,17 @@ def connect(*_args, **_kwargs):
 
     async def test_connection_lifecycles_are_serialized(self) -> None:
         with tempfile.TemporaryDirectory() as temp_dir:
-            repo = LocalAccountRepository(Path(temp_dir) / "accounts.db")
+            repos = [
+                LocalAccountRepository(Path(temp_dir) / "accounts.db"),
+                LocalAccountRepository(Path(temp_dir) / "other-accounts.db"),
+            ]
             active = 0
             max_active = 0
             guard = threading.Lock()
 
             def hold_connection(_index: int) -> None:
                 nonlocal active, max_active
-                with repo._connection():
+                with repos[_index % len(repos)]._connection():
                     with guard:
                         active += 1
                         max_active = max(max_active, active)

app/platform/storage/media_cache.py

@@ -12,6 +12,7 @@
 
 from app.platform.config.snapshot import get_config
 from app.platform.logging.logger import logger
+from utils.sqlite_runtime import sqlite_connection_guard
 
 from .media_paths import (
     image_files_dir,
@@ -55,7 +56,7 @@ def reconcile(self, media_type: MediaType) -> None:
         max_bytes = self._limit_bytes(media_type)
         if max_bytes <= 0:
             return
-        with self._guard(media_type), closing(self._connect()) as conn:
+        with self._guard(media_type), sqlite_connection_guard, closing(self._connect()) as conn:
             rows = []
             for path in self._iter_files(media_type):
                 try:
@@ -127,7 +128,7 @@ def _save(
             self._write_if_missing(path, raw)
             return path
 
-        with self._guard(media_type), closing(self._connect()) as conn:
+        with self._guard(media_type), sqlite_connection_guard, closing(self._connect()) as conn:
             if path.exists():
                 self._upsert_existing_row(conn, media_type, path)
             else:
@@ -440,7 +441,7 @@ def _delete_index_row_if_present(self, media_type: MediaType, name: str) -> None
         db_path = local_media_cache_db_path()
         if not db_path.exists():
             return
-        with closing(self._connect()) as conn:
+        with sqlite_connection_guard, closing(self._connect()) as conn:
             conn.execute(
                 f"DELETE FROM {_TABLE} WHERE media_type = ? AND name = ?",
                 (media_type, name),
@@ -451,7 +452,7 @@ def _delete_index_rows_if_present(self, media_type: MediaType) -> None:
         db_path = local_media_cache_db_path()
         if not db_path.exists():
             return
-        with closing(self._connect()) as conn:
+        with sqlite_connection_guard, closing(self._connect()) as conn:
             conn.execute(f"DELETE FROM {_TABLE} WHERE media_type = ?", (media_type,))
             conn.commit()
 

deploy/docker-compose.captcha-solver.yml

@@ -5,6 +5,7 @@ services:
       dockerfile: Dockerfile
     image: chatgpt2api-captcha-solver:local
     container_name: chatgpt2api-captcha-solver
+    init: true
     restart: unless-stopped
     ports:
       - "127.0.0.1:${CAPTCHA_SOLVER_PORT:-8877}:8877"

services/backup_service.py

@@ -21,6 +21,7 @@
 from services.config import BASE_DIR, CONFIG_FILE, DATA_DIR, config, load_backup_state, save_backup_state
 from services.image_storage_service import IMAGE_INDEX_FILE
 from services.image_tags_service import TAGS_FILE
+from utils.sqlite_runtime import sqlite_connection_guard
 
 
 def _utc_now() -> datetime:
@@ -145,7 +146,7 @@ def _sqlite_backup_bytes(source: Path) -> bytes | None:
         with tempfile.NamedTemporaryFile(prefix="chatgpt2api-grok-", suffix=".db", delete=False) as handle:
             temp_path = Path(handle.name)
         source_uri = f"file:{quote(source.resolve().as_posix(), safe='/')}?mode=ro"
-        with closing(sqlite3.connect(source_uri, uri=True)) as source_db, closing(
+        with sqlite_connection_guard, closing(sqlite3.connect(source_uri, uri=True)) as source_db, closing(
             sqlite3.connect(temp_path)
         ) as backup_db:
             source_db.backup(backup_db)

services/openai_agent_identity_store.py

@@ -12,6 +12,7 @@
 from cryptography.fernet import Fernet, InvalidToken
 
 from services.config import DATA_DIR
+from utils.sqlite_runtime import sqlite_connection_guard
 
 
 class AgentIdentityStore:
@@ -82,7 +83,7 @@ def _connect(self) -> sqlite3.Connection:
         return connection
 
     def _initialize(self) -> None:
-        with self._lock, closing(self._connect()) as connection:
+        with self._lock, sqlite_connection_guard, closing(self._connect()) as connection:
             connection.executescript(
                 """
                 CREATE TABLE IF NOT EXISTS agent_identities (
@@ -123,7 +124,7 @@ def save(self, auth_json: dict[str, Any]) -> dict[str, Any]:
         if not account_id or not runtime_id or not private_key:
             raise ValueError("Agent Identity is incomplete")
         now = datetime.now(timezone.utc).isoformat()
-        with self._lock, closing(self._connect()) as connection:
+        with self._lock, sqlite_connection_guard, closing(self._connect()) as connection:
             connection.execute("BEGIN IMMEDIATE")
             connection.execute(
                 """
@@ -172,15 +173,15 @@ def get(self, account_id: str) -> dict[str, Any] | None:
         key = str(account_id or "").strip()
         if not key:
             return None
-        with self._lock, closing(self._connect()) as connection:
+        with self._lock, sqlite_connection_guard, closing(self._connect()) as connection:
             row = connection.execute(
                 "SELECT * FROM agent_identities WHERE account_id = ?", (key,)
             ).fetchone()
         return self._auth_json(row, self._decrypt(str(row["private_key_ciphertext"]))) if row else None
 
     def list_auth(self, account_ids: list[str] | None = None) -> list[dict[str, Any]]:
         targets = {str(item or "").strip() for item in (account_ids or []) if str(item or "").strip()}
-        with self._lock, closing(self._connect()) as connection:
+        with self._lock, sqlite_connection_guard, closing(self._connect()) as connection:
             rows = connection.execute(
                 "SELECT * FROM agent_identities ORDER BY updated_at DESC"
             ).fetchall()
@@ -191,7 +192,7 @@ def list_auth(self, account_ids: list[str] | None = None) -> list[dict[str, Any]
         ]
 
     def summary(self) -> list[dict[str, str]]:
-        with self._lock, closing(self._connect()) as connection:
+        with self._lock, sqlite_connection_guard, closing(self._connect()) as connection:
             rows = connection.execute(
                 """
                 SELECT account_id, user_id, email, plan_type, agent_runtime_id,

services/openai_survival_service.py

@@ -17,6 +17,7 @@
 from services.config import DATA_DIR
 from services.json_file import read_json_object, write_json_file
 from utils.log import logger
+from utils.sqlite_runtime import sqlite_connection_guard
 
 
 DEFAULT_CONFIG = {
@@ -106,7 +107,7 @@ def _normalize_config(raw: object) -> dict[str, Any]:
 
     def _initialize_lease_db(self) -> None:
         self._data_dir.mkdir(parents=True, exist_ok=True)
-        with closing(sqlite3.connect(self._lease_path, timeout=15)) as connection:
+        with sqlite_connection_guard, closing(sqlite3.connect(self._lease_path, timeout=15)) as connection:
             connection.execute("PRAGMA busy_timeout=15000")
             connection.execute(
                 """
@@ -124,7 +125,7 @@ def _initialize_lease_db(self) -> None:
 
     def _claim_run(self, ttl_seconds: int = 3600) -> bool:
         now = time.time()
-        with closing(sqlite3.connect(self._lease_path, timeout=15)) as connection:
+        with sqlite_connection_guard, closing(sqlite3.connect(self._lease_path, timeout=15)) as connection:
             connection.execute("PRAGMA busy_timeout=15000")
             connection.execute("BEGIN IMMEDIATE")
             connection.execute(
@@ -143,15 +144,15 @@ def _claim_run(self, ttl_seconds: int = 3600) -> bool:
         return True
 
     def _release_run(self) -> None:
-        with closing(sqlite3.connect(self._lease_path, timeout=15)) as connection:
+        with sqlite_connection_guard, closing(sqlite3.connect(self._lease_path, timeout=15)) as connection:
             connection.execute(
                 "DELETE FROM scheduler_lease WHERE name = 'openai_survival' AND owner = ?",
                 (self._owner,),
             )
             connection.commit()
 
     def _renew_run(self, ttl_seconds: int = 3600) -> bool:
-        with closing(sqlite3.connect(self._lease_path, timeout=15)) as connection:
+        with sqlite_connection_guard, closing(sqlite3.connect(self._lease_path, timeout=15)) as connection:
             cursor = connection.execute(
                 """
                 UPDATE scheduler_lease SET expires_at = ?

services/register/mail_provider.py

@@ -24,6 +24,7 @@
 from services.config import DATA_DIR
 from services.json_file import read_json_file, write_json_file
 from services.proxy_service import proxy_settings
+from utils.sqlite_runtime import sqlite_connection_guard
 
 DDG_ALIASES_FILE = DATA_DIR / "ddg_aliases.json"
 _ddg_aliases_lock = Lock()
@@ -185,7 +186,7 @@ def _load_outlook_token_state() -> dict[str, dict[str, Any]]:
 
     The legacy JSON file is imported once into SQLite on first access.
     """
-    with closing(_outlook_state_connection()) as connection:
+    with sqlite_connection_guard, closing(_outlook_state_connection()) as connection:
         _ensure_outlook_state_migrated(connection)
         rows = connection.execute(
             "SELECT email, state, reason, updated_at, lease_token FROM mailbox_states"
@@ -202,7 +203,7 @@ def _load_outlook_token_state() -> dict[str, dict[str, Any]]:
 
 
 def _save_outlook_token_state(state: dict[str, dict[str, Any]]) -> None:
-    with closing(_outlook_state_connection()) as connection:
+    with sqlite_connection_guard, closing(_outlook_state_connection()) as connection:
         _ensure_outlook_state_migrated(connection)
         connection.execute("BEGIN IMMEDIATE")
         connection.execute("DELETE FROM mailbox_states")
@@ -230,7 +231,7 @@ def _claim_outlook_credential(pool: list[dict[str, str]]) -> tuple[dict[str, str
     """Atomically reserve one available credential across worker processes."""
     now = datetime.now(timezone.utc)
     lease_token = secrets.token_urlsafe(24)
-    with closing(_outlook_state_connection()) as connection:
+    with sqlite_connection_guard, closing(_outlook_state_connection()) as connection:
         _ensure_outlook_state_migrated(connection)
         connection.execute("BEGIN IMMEDIATE")
         rows = connection.execute(
@@ -320,7 +321,7 @@ def _set_outlook_token_state(
     target = str(address or "").strip().lower()
     if not target:
         return False
-    with closing(_outlook_state_connection()) as connection:
+    with sqlite_connection_guard, closing(_outlook_state_connection()) as connection:
         _ensure_outlook_state_migrated(connection)
         connection.execute("BEGIN IMMEDIATE")
         if lease_token:
@@ -363,7 +364,7 @@ def _release_outlook_token_state(address: str, *, lease_token: str = "") -> bool
     target = str(address or "").strip().lower()
     if not target:
         return False
-    with closing(_outlook_state_connection()) as connection:
+    with sqlite_connection_guard, closing(_outlook_state_connection()) as connection:
         _ensure_outlook_state_migrated(connection)
         connection.execute("BEGIN IMMEDIATE")
         if lease_token:
@@ -390,7 +391,7 @@ def clear_outlook_token_states(addresses: list[str] | set[str], states: set[str]
     targets.discard("")
     if not targets:
         return 0
-    with closing(_outlook_state_connection()) as connection:
+    with sqlite_connection_guard, closing(_outlook_state_connection()) as connection:
         _ensure_outlook_state_migrated(connection)
         connection.execute("BEGIN IMMEDIATE")
         placeholders = ",".join("?" for _ in targets)
@@ -422,7 +423,7 @@ def reset_outlook_token_pool_state(scope: str = "all") -> int:
         target_states = OUTLOOK_BUSY_STATES
     else:
         target_states = set()
-    with closing(_outlook_state_connection()) as connection:
+    with sqlite_connection_guard, closing(_outlook_state_connection()) as connection:
         _ensure_outlook_state_migrated(connection)
         connection.execute("BEGIN IMMEDIATE")
         if target_states:

utils/sqlite_runtime.py

@@ -0,0 +1,8 @@
+from __future__ import annotations
+
+import threading
+
+
+# SQLite keeps a process-wide Unix file-descriptor cache. Serializing complete
+# connection lifecycles avoids an open/close mutex deadlock under thread load.
+sqlite_connection_guard = threading.RLock()