fix(db): prevent runtime lock convoys

.env.example

@@ -276,6 +276,8 @@ DB_POOL_MIN=5
 DB_POOL_MAX=auto
 DB_POOL_ACQUIRE_TIMEOUT=10
 DB_POOL_HEALTH_CHECK=true
+DB_IDLE_IN_TRANSACTION_TIMEOUT_MS=60000
+DB_LOCK_TIMEOUT_MS=10000
 DB_POOL_AUTO_CAP=true
 DB_POOL_AUTO_DEFAULT_MAX=50
 DB_POOL_RESERVE_FOR_OTHER_CLIENTS=20

backend_api_python/app/services/billing_service.py

@@ -89,7 +89,6 @@ def get_user_vip_status(self, user_id: int) -> Tuple[bool, Optional[datetime]]:
             with get_db_connection() as db:
                 cur = db.cursor()
                 # Ensure lifetime membership monthly credits are granted (best-effort, silent on failure).
-                self._ensure_membership_schema_best_effort(cur)
                 self._grant_lifetime_monthly_credits_best_effort(cur, user_id)
                 try:
                     db.commit()
@@ -124,7 +123,6 @@ def get_membership_plans(self, include_inactive: bool = False) -> Dict[str, Any]
         try:
             with get_db_connection() as db:
                 cur = db.cursor()
-                self._ensure_plan_schema(cur)
                 self._seed_default_plans(cur)
                 sql = """
                     SELECT code, name, description, price_usd, duration_days,
@@ -187,7 +185,6 @@ def save_membership_plans(self, items) -> Tuple[bool, str, Dict[str, Any]]:
         try:
             with get_db_connection() as db:
                 cur = db.cursor()
-                self._ensure_plan_schema(cur)
                 for row in normalized:
                     cur.execute(
                         """
@@ -232,7 +229,6 @@ def delete_membership_plan(self, code: str) -> Tuple[bool, str, Dict[str, Any]]:
         try:
             with get_db_connection() as db:
                 cur = db.cursor()
-                self._ensure_plan_schema(cur)
                 cur.execute("SELECT code, is_active FROM qd_billing_plans WHERE code=? FOR UPDATE", (code,))
                 plan = cur.fetchone() or {}
                 if not plan:
@@ -273,32 +269,6 @@ def _serialize_plan(row: Dict[str, Any]) -> Dict[str, Any]:
             "stripe_price_id": row.get("stripe_price_id") or "",
         }
 
-    def _ensure_plan_schema(self, cur) -> None:
-        cur.execute(
-            """
-            CREATE TABLE IF NOT EXISTS qd_billing_plans (
-              code VARCHAR(64) PRIMARY KEY,
-              name VARCHAR(120) NOT NULL,
-              description TEXT NOT NULL DEFAULT '',
-              price_usd DECIMAL(12,2) NOT NULL DEFAULT 0,
-              duration_days INTEGER NOT NULL DEFAULT 0,
-              credits_once INTEGER NOT NULL DEFAULT 0,
-              credits_monthly INTEGER NOT NULL DEFAULT 0,
-              strategy_limit INTEGER NOT NULL DEFAULT 10,
-              referral_eligible BOOLEAN NOT NULL DEFAULT FALSE,
-              is_lifetime BOOLEAN NOT NULL DEFAULT FALSE,
-              is_active BOOLEAN NOT NULL DEFAULT TRUE,
-              is_popular BOOLEAN NOT NULL DEFAULT FALSE,
-              sort_order INTEGER NOT NULL DEFAULT 0,
-              stripe_price_id VARCHAR(255) NOT NULL DEFAULT '',
-              created_at TIMESTAMP DEFAULT NOW(),
-              updated_at TIMESTAMP DEFAULT NOW()
-            )
-            """
-        )
-        cur.execute("ALTER TABLE qd_billing_plans ADD COLUMN IF NOT EXISTS strategy_limit INTEGER NOT NULL DEFAULT 10")
-        cur.execute("ALTER TABLE qd_billing_plans ADD COLUMN IF NOT EXISTS referral_eligible BOOLEAN NOT NULL DEFAULT FALSE")
-
     def _seed_default_plans(self, cur) -> None:
         cur.execute("SELECT COUNT(*) AS count FROM qd_billing_plans")
         if int((cur.fetchone() or {}).get("count") or 0) > 0:
@@ -342,13 +312,8 @@ def purchase_membership(
         try:
             with get_db_connection() as db:
                 cur = db.cursor()
-                self._ensure_membership_schema_best_effort(cur)
-                if record_membership_order:
-                    self._ensure_membership_orders_table_best_effort(cur)
-
                 fulfillment_ref = (fulfillment_ref or "").strip()
                 if fulfillment_ref:
-                    self._ensure_fulfillment_schema(cur)
                     cur.execute(
                         """
                         INSERT INTO qd_membership_fulfillments (provider_ref, user_id, plan, created_at)
@@ -525,49 +490,6 @@ def purchase_membership(
             logger.error(f"purchase_membership failed: {e}", exc_info=True)
             return False, f"error:{str(e)}", {}
 
-    def _ensure_membership_schema_best_effort(self, cur):
-        """Best-effort schema upgrade for membership fields on qd_users."""
-        try:
-            # vip_plan / vip_is_lifetime / vip_monthly_credits_last_grant
-            cur.execute("ALTER TABLE qd_users ADD COLUMN IF NOT EXISTS vip_plan VARCHAR(64) DEFAULT ''")
-            cur.execute("ALTER TABLE qd_users ALTER COLUMN vip_plan TYPE VARCHAR(64)")
-            cur.execute("ALTER TABLE qd_users ADD COLUMN IF NOT EXISTS vip_is_lifetime BOOLEAN DEFAULT FALSE")
-            cur.execute("ALTER TABLE qd_users ADD COLUMN IF NOT EXISTS vip_monthly_credits_last_grant TIMESTAMP")
-        except Exception:
-            # Ignore schema upgrade failures (e.g., insufficient privileges)
-            pass
-
-    def _ensure_membership_orders_table_best_effort(self, cur):
-        """Best-effort create membership orders table (legacy; optional row per mock checkout)."""
-        try:
-            cur.execute(
-                """
-                CREATE TABLE IF NOT EXISTS qd_membership_orders (
-                  id SERIAL PRIMARY KEY,
-                  user_id INTEGER NOT NULL,
-                  plan VARCHAR(64) NOT NULL,
-                  price_usd DECIMAL(10,2) DEFAULT 0,
-                  status VARCHAR(20) DEFAULT 'paid',
-                  created_at TIMESTAMP DEFAULT NOW(),
-                  paid_at TIMESTAMP
-                )
-                """
-            )
-        except Exception:
-            pass
-
-    def _ensure_fulfillment_schema(self, cur):
-        cur.execute(
-            """
-            CREATE TABLE IF NOT EXISTS qd_membership_fulfillments (
-              provider_ref VARCHAR(255) PRIMARY KEY,
-              user_id INTEGER NOT NULL,
-              plan VARCHAR(64) NOT NULL,
-              created_at TIMESTAMP DEFAULT NOW()
-            )
-            """
-        )
-
     def _add_credits_in_tx(self, cur, user_id: int, amount: int, action: str, remark: str, reference_id: str = ''):
         """Add credits within an existing DB transaction and write qd_credits_log."""
         try:
@@ -600,7 +522,6 @@ def _grant_lifetime_monthly_credits_best_effort(self, cur, user_id: int):
                 return
             # Reuse the caller's transaction. Opening another pooled connection
             # here can deadlock a small deployment when all connections are busy.
-            self._ensure_plan_schema(cur)
             self._seed_default_plans(cur)
             cur.execute(
                 "SELECT credits_monthly FROM qd_billing_plans WHERE code = ?",

backend_api_python/app/services/user_preferences.py

@@ -15,20 +15,7 @@
 
 
 def ensure_chart_templates_column() -> None:
-    """Add qd_users.chart_templates when upgrading existing databases."""
-    try:
-        with get_db_connection() as db:
-            cur = db.cursor()
-            cur.execute(
-                """
-                ALTER TABLE qd_users
-                ADD COLUMN IF NOT EXISTS chart_templates TEXT DEFAULT ''
-                """
-            )
-            db.commit()
-            cur.close()
-    except Exception as exc:
-        logger.warning("ensure chart_templates column skipped: %s", exc)
+    """Compatibility hook; canonical schema changes run in migrations."""
 
 
 def get_notification_settings(user_id: int) -> dict | None:

backend_api_python/app/services/user_service.py

@@ -92,36 +92,8 @@ def normalize_user_record(cls, user: Optional[Dict[str, Any]]) -> Optional[Dict[
         return normalized
 
     def ensure_password_changed_column(self) -> None:
-        """Add password_changed_at if missing (idempotent)."""
-        if UserService._password_changed_column_ready:
-            return
-        try:
-            with get_db_connection() as db:
-                cur = db.cursor()
-                cur.execute(
-                    """
-                    SELECT 1 FROM information_schema.columns
-                    WHERE table_name = 'qd_users' AND column_name = 'password_changed_at'
-                    """
-                )
-                if not cur.fetchone():
-                    cur.execute(
-                        "ALTER TABLE qd_users ADD COLUMN password_changed_at TIMESTAMP NULL"
-                    )
-                    # Existing accounts: assume they already passed initial setup.
-                    cur.execute(
-                        """
-                        UPDATE qd_users
-                        SET password_changed_at = COALESCE(updated_at, created_at, NOW())
-                        WHERE password_changed_at IS NULL
-                        """
-                    )
-                    db.commit()
-                    logger.info("Added password_changed_at column to qd_users (existing rows backfilled)")
-                cur.close()
-            UserService._password_changed_column_ready = True
-        except Exception as e:
-            logger.warning(f"ensure_password_changed_column failed: {e}")
+        """Compatibility hook; canonical schema changes run in migrations."""
+        UserService._password_changed_column_ready = True
 
     def mark_password_changed(self, user_id: int) -> None:
         """Record that the user has set a non-initial password."""
@@ -1245,27 +1217,28 @@ def ensure_admin_exists(self):
                 cur.execute("SELECT COUNT(*) as count FROM qd_users")
                 count = cur.fetchone()['count']
                 cur.close()
-
-                if count == 0:
-                    # Create admin using env credentials
-                    from app.config.settings import Config
-                    admin_user = os.getenv('ADMIN_USER', Config.ADMIN_USER)
-                    admin_password = os.getenv('ADMIN_PASSWORD', Config.ADMIN_PASSWORD)
-                    admin_email = os.getenv('ADMIN_EMAIL', 'admin@example.com')
-
-                    self.create_user({
-                        'username': admin_user,
-                        'password': admin_password,
-                        'email': admin_email,
-                        'nickname': 'Administrator',
-                        'role': 'admin',
-                        'status': 'active',
-                        'email_verified': True  # Admin email is pre-verified
-                    })
-                    logger.info(f"Created admin user: {admin_user} ({admin_email})")
-                else:
-                    self.sync_bootstrap_admin_credentials_from_env()
-                    self.sync_admin_email_from_config(overwrite_existing=False)
+
+            # Do not call helpers that acquire another pooled connection while
+            # the startup read transaction is still open.
+            if count == 0:
+                from app.config.settings import Config
+                admin_user = os.getenv('ADMIN_USER', Config.ADMIN_USER)
+                admin_password = os.getenv('ADMIN_PASSWORD', Config.ADMIN_PASSWORD)
+                admin_email = os.getenv('ADMIN_EMAIL', 'admin@example.com')
+
+                self.create_user({
+                    'username': admin_user,
+                    'password': admin_password,
+                    'email': admin_email,
+                    'nickname': 'Administrator',
+                    'role': 'admin',
+                    'status': 'active',
+                    'email_verified': True
+                })
+                logger.info(f"Created admin user: {admin_user} ({admin_email})")
+            else:
+                self.sync_bootstrap_admin_credentials_from_env()
+                self.sync_admin_email_from_config(overwrite_existing=False)
         except Exception as e:
             logger.error(f"ensure_admin_exists failed: {e}")
 

backend_api_python/app/utils/db_postgres.py

@@ -9,6 +9,8 @@
     DB_POOL_MAX               maxconn or "auto"             default auto
     DB_POOL_ACQUIRE_TIMEOUT   seconds to wait on exhaustion default 10
     DB_POOL_HEALTH_CHECK      "true" / "false"              default "true"
+    DB_IDLE_IN_TRANSACTION_TIMEOUT_MS server-side timeout    default 60000
+    DB_LOCK_TIMEOUT_MS        server-side lock wait timeout  default 10000
 """
 import os
 import time
@@ -81,6 +83,8 @@ def _env_bool(key: str, default: bool) -> bool:
 DB_POOL_HEALTH_CHECK = _env_bool("DB_POOL_HEALTH_CHECK", True)
 DB_POOL_AUTO_CAP = _env_bool("DB_POOL_AUTO_CAP", True)
 DB_POOL_RESERVE_FOR_OTHER_CLIENTS = _env_int("DB_POOL_RESERVE_FOR_OTHER_CLIENTS", 20)
+DB_IDLE_IN_TRANSACTION_TIMEOUT_MS = _env_int("DB_IDLE_IN_TRANSACTION_TIMEOUT_MS", 60_000)
+DB_LOCK_TIMEOUT_MS = _env_int("DB_LOCK_TIMEOUT_MS", 10_000)
 DB_APPLICATION_NAME = os.getenv("DB_APPLICATION_NAME", "quantdinger_api").strip() or "quantdinger_api"
 
 
@@ -172,7 +176,11 @@ def _get_connection_pool():
                 # "idle in transaction" state when no explicit commit/rollback
                 # followed).  keepalives keep dead sockets from lingering in
                 # the pool when the PG side or a NAT drops them.
-                options="-c timezone=UTC",
+                options=(
+                    "-c timezone=UTC "
+                    f"-c idle_in_transaction_session_timeout={DB_IDLE_IN_TRANSACTION_TIMEOUT_MS} "
+                    f"-c lock_timeout={DB_LOCK_TIMEOUT_MS}"
+                ),
                 keepalives=1,
                 keepalives_idle=30,
                 keepalives_interval=10,
@@ -317,6 +325,23 @@ def _pool_max_config_label() -> str:
     return str(DB_POOL_MAX_CONFIGURED) if DB_POOL_MAX_CONFIGURED is not None else "auto"
 
 
+def _rollback_open_transaction(conn) -> bool:
+    """Return a pooled connection to PostgreSQL's idle transaction state."""
+    if conn is None or getattr(conn, "closed", 0):
+        return False
+
+    try:
+        status_reader = getattr(conn, "get_transaction_status", None)
+        if callable(status_reader):
+            idle_status = psycopg2.extensions.TRANSACTION_STATUS_IDLE
+            if status_reader() == idle_status:
+                return True
+        conn.rollback()
+        return True
+    except Exception:
+        return False
+
+
 def _is_connection_healthy(conn) -> bool:
     """Quick health check: make sure the connection is not closed and can
     actually round-trip a trivial query.  Used only when DB_POOL_HEALTH_CHECK
@@ -327,14 +352,28 @@ def _is_connection_healthy(conn) -> bool:
     # psycopg2 sets .closed to nonzero when the connection is closed.
     if getattr(conn, "closed", 0):
         return False
+    if not _rollback_open_transaction(conn):
+        return False
+
+    probe_ok = False
+    cur = None
     try:
         cur = conn.cursor()
         cur.execute("SELECT 1")
         cur.fetchone()
-        cur.close()
-        return True
+        probe_ok = True
     except Exception:
-        return False
+        probe_ok = False
+    finally:
+        if cur is not None:
+            try:
+                cur.close()
+            except Exception:
+                probe_ok = False
+
+    # psycopg2 starts a transaction even for SELECT 1. Never let the health
+    # probe itself put a pooled connection into idle-in-transaction state.
+    return _rollback_open_transaction(conn) and probe_ok
 
 
 def _acquire_conn_with_wait(pg_pool):
@@ -413,7 +452,11 @@ def _acquire_conn_with_wait(pg_pool):
             backoff = min(backoff * 2, 0.5)
             continue
 
-        if DB_POOL_HEALTH_CHECK and not _is_connection_healthy(conn):
+        connection_ready = _rollback_open_transaction(conn)
+        if connection_ready and DB_POOL_HEALTH_CHECK:
+            connection_ready = _is_connection_healthy(conn)
+
+        if not connection_ready:
             # Drop the dead connection and let the pool create a new one on
             # next attempt.  putconn(close=True) asks the pool to discard it.
             try:
@@ -658,10 +701,14 @@ def close(self):
         """Return connection to pool.  Broken connections are discarded so
         we don't poison the pool with closed sockets.
         """
-        if self._pool and self._conn:
+        conn = self._conn
+        self._conn = None
+        if self._pool and conn:
             try:
-                broken = bool(getattr(self._conn, "closed", 0))
-                self._pool.putconn(self._conn, close=broken)
+                broken = bool(getattr(conn, "closed", 0))
+                if not broken and not _rollback_open_transaction(conn):
+                    broken = True
+                self._pool.putconn(conn, close=broken)
                 _observe_pool_stats(self._pool)
             except Exception as e:
                 logger.warning(f"Failed to return connection to pool: {e}")
@@ -772,6 +819,8 @@ def get_pg_connection():
     finally:
         if conn is not None:
             try:
+                if not broken and not _rollback_open_transaction(conn):
+                    broken = True
                 pg_pool.putconn(conn, close=broken)
                 _observe_pool_stats(pg_pool)
             except Exception:

backend_api_python/env.example

@@ -153,6 +153,8 @@ DB_POOL_MIN=5
 DB_POOL_MAX=auto
 DB_POOL_ACQUIRE_TIMEOUT=10
 DB_POOL_HEALTH_CHECK=true
+DB_IDLE_IN_TRANSACTION_TIMEOUT_MS=60000
+DB_LOCK_TIMEOUT_MS=10000
 DB_POOL_AUTO_CAP=true
 DB_POOL_AUTO_DEFAULT_MAX=50
 DB_POOL_RESERVE_FOR_OTHER_CLIENTS=20