From 93c7e9a085e000e4609dcdaa8312972d56961fb2 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 1 Jul 2026 19:25:28 +0000 Subject: [PATCH 1/2] fix: enable WAL mode + busy_timeout for shared app.db to prevent lock errors Scheduled tasks, message history, and conversations all write to the same sqlite file from concurrent asyncio tasks/channels. The default rollback journal mode plus a short busy timeout meant contention (e.g. a scheduled task writing task_outputs while the web channel writes messages) eventually exceeded the timeout and raised "database is locked". Centralize connection creation in config.get_db_connection(), which turns on WAL (so readers don't block the writer) and raises busy_timeout to 30s. --- app/config.py | 16 ++++++++++++++++ app/core/scheduled_tasks.py | 18 +++++++++--------- app/infra/conversations.py | 6 +++--- app/infra/message_history.py | 8 ++++---- 4 files changed, 32 insertions(+), 16 deletions(-) diff --git a/app/config.py b/app/config.py index ed9f4d8..ee9342f 100644 --- a/app/config.py +++ b/app/config.py @@ -1,6 +1,7 @@ from __future__ import annotations import os +import sqlite3 import tomllib from pathlib import Path @@ -12,6 +13,21 @@ APP_DB = PROJECT_HOME / "app.db" +def get_db_connection(db_path: Path = APP_DB, *, timeout: float = 30.0, + isolation_level: str | None = "") -> sqlite3.Connection: + """Open a connection to a shared SQLite db (e.g. APP_DB) with settings safe + for concurrent access from multiple asyncio tasks/channels. + + WAL mode lets readers proceed without blocking the writer, and the longer + busy_timeout makes writers retry instead of immediately raising + "database is locked" when another connection briefly holds the write lock. + """ + conn = sqlite3.connect(db_path, timeout=timeout, isolation_level=isolation_level) + conn.execute("PRAGMA journal_mode=WAL") + conn.execute(f"PRAGMA busy_timeout={int(timeout * 1000)}") + return conn + + def load(path: Path | str = HOME_CONFIG_PATH) -> None: global _config diff --git a/app/core/scheduled_tasks.py b/app/core/scheduled_tasks.py index ff7c4f5..7f9bce1 100755 --- a/app/core/scheduled_tasks.py +++ b/app/core/scheduled_tasks.py @@ -2,7 +2,7 @@ from datetime import datetime, timedelta from ..infra.app_logging import log -from ..config import APP_DB +from ..config import APP_DB, get_db_connection import sqlite3 import asyncio from .helper_agent import HelperAgent @@ -19,7 +19,7 @@ def __init__(self, mqs: dict = None, channels: dict = None): self._init_tasks_db() def _init_tasks_db(self): - with sqlite3.connect(APP_DB) as conn: + with get_db_connection() as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS tasks ( @@ -57,7 +57,7 @@ def load_tasks(self) -> list[dict]: query = """SELECT name, prompt, enabled, repeat, interval_mins, last_run, next_run, delivery_channel, run_count, created_at FROM tasks""" - conn = sqlite3.connect(APP_DB) + conn = get_db_connection() try: rows = conn.execute(query).fetchall() finally: @@ -72,7 +72,7 @@ def load_tasks(self) -> list[dict]: def add_task(self, name: str, prompt: str, next_run: str, interval_mins: int = 1, repeat: int = 0, delivery_channel: str = "telegram", enabled: int = 1): now = datetime.now().isoformat() - conn = sqlite3.connect(APP_DB) + conn = get_db_connection() try: try: with conn: @@ -86,7 +86,7 @@ def add_task(self, name: str, prompt: str, next_run: str, interval_mins: int = 1 conn.close() def remove_task(self, name: str): - with sqlite3.connect(APP_DB) as conn: + with get_db_connection() as conn: conn.execute("DELETE FROM tasks WHERE name = ?", (name,)) conn.commit() conn.close() @@ -96,7 +96,7 @@ def update_task(self, name: str, **fields): return set_clause = ", ".join(f"{col} = ?" for col in fields) values = list(fields.values()) + [name] - conn = sqlite3.connect(APP_DB) + conn = get_db_connection() try: with conn: if not conn.execute("SELECT name FROM tasks WHERE name = ?", (name,)).fetchone(): @@ -107,7 +107,7 @@ def update_task(self, name: str, **fields): def save_output(self, name: str, prompt: str, output: str, status: str = "success", duration_secs: float = None): - with sqlite3.connect(APP_DB) as conn: + with get_db_connection() as conn: conn.execute(""" INSERT INTO task_outputs (name, prompt, output, status, duration_secs, timestamp) VALUES (?, ?, ?, ?, ?, ?) @@ -116,7 +116,7 @@ def save_output(self, name: str, prompt: str, output: str, conn.close() def get_output(self, name: str, num_entries: int = 5) -> list[dict]: - with sqlite3.connect(APP_DB) as conn: + with get_db_connection() as conn: rows = conn.execute(""" SELECT prompt, output, status, duration_secs, timestamp FROM task_outputs WHERE name = ? @@ -158,7 +158,7 @@ def _after_run(self, task: dict, now: datetime): intervals_passed = int(elapsed_secs // interval.total_seconds()) next_due = original_next + (intervals_passed + 1) * interval next_run = next_due.isoformat() - with sqlite3.connect(APP_DB) as conn: + with get_db_connection() as conn: conn.execute("""UPDATE tasks SET last_run = ?, next_run = ?, run_count = run_count + 1 WHERE name = ?""", (now.isoformat(), next_run, name)) conn.commit() diff --git a/app/infra/conversations.py b/app/infra/conversations.py index 38cc5c8..e6618bd 100644 --- a/app/infra/conversations.py +++ b/app/infra/conversations.py @@ -7,14 +7,14 @@ from datetime import datetime from pathlib import Path -from ..config import APP_DB, PROJECT_HOME +from ..config import APP_DB, PROJECT_HOME, get_db_connection from .app_logging import log @contextlib.contextmanager def _fk_conn(db_path: Path): """sqlite3 connection with FK constraints enforced.""" - conn = sqlite3.connect(db_path) + conn = get_db_connection(db_path) try: conn.execute("PRAGMA foreign_keys = ON") with conn: @@ -79,7 +79,7 @@ def _ensure_schema(self) -> None: # Phase 2: Migration — BEGIN IMMEDIATE for concurrency safety. # The WHERE conversation_id IS NULL guard makes this re-entrant. - mconn = sqlite3.connect(self.db_path, isolation_level=None) + mconn = get_db_connection(self.db_path, isolation_level=None) try: mconn.execute("PRAGMA foreign_keys = ON") mconn.execute("BEGIN IMMEDIATE") diff --git a/app/infra/message_history.py b/app/infra/message_history.py index 321af9d..7194cc9 100755 --- a/app/infra/message_history.py +++ b/app/infra/message_history.py @@ -4,7 +4,7 @@ from datetime import datetime from pathlib import Path from .app_logging import log -from ..config import APP_DB +from ..config import APP_DB, get_db_connection def _est_tokens(content: str) -> int: return max(1, len(content) // 4) @@ -18,7 +18,7 @@ def __init__(self, channel_type: str, db_path: Path = APP_DB): def _ensure_db(self): try: self.db_path.parent.mkdir(parents=True, exist_ok=True) - with sqlite3.connect(self.db_path) as conn: + with get_db_connection(self.db_path) as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -53,7 +53,7 @@ def _ensure_db(self): def add_message(self, role: str, content: str, conversation_id: int = None): timestamp = datetime.now().isoformat() est = _est_tokens(content) - with sqlite3.connect(self.db_path) as conn: + with get_db_connection(self.db_path) as conn: conn.execute( "INSERT INTO messages (channel, role, content, timestamp, est_tokens, conversation_id) " "VALUES (?, ?, ?, ?, ?, ?)", @@ -64,7 +64,7 @@ def add_message(self, role: str, content: str, conversation_id: int = None): log.info(f"Added message to history: role={role}, est_tokens={est}, content={content[:30]}...") def get_history(self, limit: int = 100) -> list[dict]: - with sqlite3.connect(self.db_path) as conn: + with get_db_connection(self.db_path) as conn: rows = conn.execute("""SELECT role, content FROM messages WHERE channel = ? ORDER BY id DESC LIMIT ?""", (self.channel, limit)).fetchall() From 939a09fe2f0afc11fd35621a359994c8dd890095 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 2 Jul 2026 18:29:09 +0000 Subject: [PATCH 2/2] fix: address review feedback on WAL/connection handling - get_db_connection() no longer sets PRAGMA journal_mode=WAL or busy_timeout on every connection: WAL is persisted in the db file so it only needs enabling once per store's schema-init, and busy_timeout is already set by sqlite3.connect's timeout param. - Replaced `with conn: ... ; conn.close()` patterns in scheduled_tasks.py and message_history.py with try/finally so an exception during the transaction can't skip closing the connection. --- app/config.py | 12 ++-- app/core/scheduled_tasks.py | 109 +++++++++++++++++++---------------- app/infra/conversations.py | 1 + app/infra/message_history.py | 81 ++++++++++++++------------ 4 files changed, 110 insertions(+), 93 deletions(-) diff --git a/app/config.py b/app/config.py index ee9342f..0f26206 100644 --- a/app/config.py +++ b/app/config.py @@ -18,14 +18,14 @@ def get_db_connection(db_path: Path = APP_DB, *, timeout: float = 30.0, """Open a connection to a shared SQLite db (e.g. APP_DB) with settings safe for concurrent access from multiple asyncio tasks/channels. - WAL mode lets readers proceed without blocking the writer, and the longer - busy_timeout makes writers retry instead of immediately raising + The timeout is passed straight to sqlite3.connect, which sets the + connection's busy timeout so writers retry instead of immediately raising "database is locked" when another connection briefly holds the write lock. + WAL mode (readers don't block the writer) is persisted in the db file + itself, so it only needs to be enabled once by each store's schema-init + step rather than on every connection. """ - conn = sqlite3.connect(db_path, timeout=timeout, isolation_level=isolation_level) - conn.execute("PRAGMA journal_mode=WAL") - conn.execute(f"PRAGMA busy_timeout={int(timeout * 1000)}") - return conn + return sqlite3.connect(db_path, timeout=timeout, isolation_level=isolation_level) def load(path: Path | str = HOME_CONFIG_PATH) -> None: diff --git a/app/core/scheduled_tasks.py b/app/core/scheduled_tasks.py index 7f9bce1..020e4be 100755 --- a/app/core/scheduled_tasks.py +++ b/app/core/scheduled_tasks.py @@ -19,38 +19,39 @@ def __init__(self, mqs: dict = None, channels: dict = None): self._init_tasks_db() def _init_tasks_db(self): - with get_db_connection() as conn: - - conn.execute(""" - CREATE TABLE IF NOT EXISTS tasks ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - name TEXT NOT NULL UNIQUE, - prompt TEXT NOT NULL, - enabled INTEGER NOT NULL DEFAULT 1, - repeat INTEGER NOT NULL DEFAULT 0, - interval_mins INTEGER NOT NULL DEFAULT 1, - last_run TEXT, - next_run TEXT NOT NULL, - delivery_channel TEXT NOT NULL DEFAULT 'telegram', - run_count INTEGER NOT NULL DEFAULT 0, - created_at TEXT NOT NULL - ) - """) - - conn.execute(""" - CREATE TABLE IF NOT EXISTS task_outputs ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - name TEXT NOT NULL, - prompt TEXT NOT NULL, - output TEXT NOT NULL, - status TEXT NOT NULL DEFAULT 'success', - duration_secs REAL, - timestamp TEXT NOT NULL - ) - """) - - conn.commit() - conn.close() + conn = get_db_connection() + try: + conn.execute("PRAGMA journal_mode=WAL") + with conn: + conn.execute(""" + CREATE TABLE IF NOT EXISTS tasks ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT NOT NULL UNIQUE, + prompt TEXT NOT NULL, + enabled INTEGER NOT NULL DEFAULT 1, + repeat INTEGER NOT NULL DEFAULT 0, + interval_mins INTEGER NOT NULL DEFAULT 1, + last_run TEXT, + next_run TEXT NOT NULL, + delivery_channel TEXT NOT NULL DEFAULT 'telegram', + run_count INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL + ) + """) + + conn.execute(""" + CREATE TABLE IF NOT EXISTS task_outputs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT NOT NULL, + prompt TEXT NOT NULL, + output TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'success', + duration_secs REAL, + timestamp TEXT NOT NULL + ) + """) + finally: + conn.close() def load_tasks(self) -> list[dict]: @@ -86,10 +87,12 @@ def add_task(self, name: str, prompt: str, next_run: str, interval_mins: int = 1 conn.close() def remove_task(self, name: str): - with get_db_connection() as conn: - conn.execute("DELETE FROM tasks WHERE name = ?", (name,)) - conn.commit() - conn.close() + conn = get_db_connection() + try: + with conn: + conn.execute("DELETE FROM tasks WHERE name = ?", (name,)) + finally: + conn.close() def update_task(self, name: str, **fields): if not fields: @@ -107,22 +110,26 @@ def update_task(self, name: str, **fields): def save_output(self, name: str, prompt: str, output: str, status: str = "success", duration_secs: float = None): - with get_db_connection() as conn: - conn.execute(""" - INSERT INTO task_outputs (name, prompt, output, status, duration_secs, timestamp) - VALUES (?, ?, ?, ?, ?, ?) - """, (name, prompt, output, status, duration_secs, datetime.now().isoformat())) - conn.commit() - conn.close() + conn = get_db_connection() + try: + with conn: + conn.execute(""" + INSERT INTO task_outputs (name, prompt, output, status, duration_secs, timestamp) + VALUES (?, ?, ?, ?, ?, ?) + """, (name, prompt, output, status, duration_secs, datetime.now().isoformat())) + finally: + conn.close() def get_output(self, name: str, num_entries: int = 5) -> list[dict]: - with get_db_connection() as conn: + conn = get_db_connection() + try: rows = conn.execute(""" SELECT prompt, output, status, duration_secs, timestamp FROM task_outputs WHERE name = ? ORDER BY id DESC LIMIT ? """, (name, num_entries)).fetchall() - conn.close() + finally: + conn.close() return [{"prompt": p, "output": o, "status": s, "duration_secs": d, "timestamp": t} for p, o, s, d, t in reversed(rows)] @@ -158,11 +165,13 @@ def _after_run(self, task: dict, now: datetime): intervals_passed = int(elapsed_secs // interval.total_seconds()) next_due = original_next + (intervals_passed + 1) * interval next_run = next_due.isoformat() - with get_db_connection() as conn: - conn.execute("""UPDATE tasks SET last_run = ?, next_run = ?, run_count = run_count + 1 - WHERE name = ?""", (now.isoformat(), next_run, name)) - conn.commit() - conn.close() + conn = get_db_connection() + try: + with conn: + conn.execute("""UPDATE tasks SET last_run = ?, next_run = ?, run_count = run_count + 1 + WHERE name = ?""", (now.isoformat(), next_run, name)) + finally: + conn.close() else: self.remove_task(name) diff --git a/app/infra/conversations.py b/app/infra/conversations.py index e6618bd..3e0a9c1 100644 --- a/app/infra/conversations.py +++ b/app/infra/conversations.py @@ -35,6 +35,7 @@ def _ensure_schema(self) -> None: # Phase 1: DDL — idempotent CREATE TABLE/INDEX and optional ALTER TABLE with _fk_conn(self.db_path) as conn: + conn.execute("PRAGMA journal_mode=WAL") conn.execute(""" CREATE TABLE IF NOT EXISTS conversations ( id INTEGER PRIMARY KEY AUTOINCREMENT, diff --git a/app/infra/message_history.py b/app/infra/message_history.py index 7194cc9..0c63715 100755 --- a/app/infra/message_history.py +++ b/app/infra/message_history.py @@ -18,33 +18,36 @@ def __init__(self, channel_type: str, db_path: Path = APP_DB): def _ensure_db(self): try: self.db_path.parent.mkdir(parents=True, exist_ok=True) - with get_db_connection(self.db_path) as conn: - conn.execute(""" - CREATE TABLE IF NOT EXISTS messages ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - channel text NOT NULL, - role TEXT NOT NULL, - content TEXT NOT NULL, - timestamp TEXT NOT NULL, - est_tokens INTEGER - ) - """) - conn.execute(""" - CREATE INDEX IF NOT EXISTS idx_messages_channel ON messages(channel) - """) - # Add conversation_id if missing (ConversationStore adds it too, - # but MessageHistory must be self-consistent for standalone use) - cols = {row[1] for row in conn.execute("PRAGMA table_info(messages)")} - if "conversation_id" not in cols: - conn.execute( - "ALTER TABLE messages ADD COLUMN conversation_id INTEGER" - ) - conn.execute(""" - CREATE INDEX IF NOT EXISTS idx_messages_conversation_id - ON messages(conversation_id) - """) - conn.commit() - conn.close() + conn = get_db_connection(self.db_path) + try: + conn.execute("PRAGMA journal_mode=WAL") + with conn: + conn.execute(""" + CREATE TABLE IF NOT EXISTS messages ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + channel text NOT NULL, + role TEXT NOT NULL, + content TEXT NOT NULL, + timestamp TEXT NOT NULL, + est_tokens INTEGER + ) + """) + conn.execute(""" + CREATE INDEX IF NOT EXISTS idx_messages_channel ON messages(channel) + """) + # Add conversation_id if missing (ConversationStore adds it too, + # but MessageHistory must be self-consistent for standalone use) + cols = {row[1] for row in conn.execute("PRAGMA table_info(messages)")} + if "conversation_id" not in cols: + conn.execute( + "ALTER TABLE messages ADD COLUMN conversation_id INTEGER" + ) + conn.execute(""" + CREATE INDEX IF NOT EXISTS idx_messages_conversation_id + ON messages(conversation_id) + """) + finally: + conn.close() except sqlite3.Error as e: log.error(f"Error creating message history database: {str(e)}") @@ -53,22 +56,26 @@ def _ensure_db(self): def add_message(self, role: str, content: str, conversation_id: int = None): timestamp = datetime.now().isoformat() est = _est_tokens(content) - with get_db_connection(self.db_path) as conn: - conn.execute( - "INSERT INTO messages (channel, role, content, timestamp, est_tokens, conversation_id) " - "VALUES (?, ?, ?, ?, ?, ?)", - (self.channel, role, content, timestamp, est, conversation_id), - ) - conn.commit() - conn.close() + conn = get_db_connection(self.db_path) + try: + with conn: + conn.execute( + "INSERT INTO messages (channel, role, content, timestamp, est_tokens, conversation_id) " + "VALUES (?, ?, ?, ?, ?, ?)", + (self.channel, role, content, timestamp, est, conversation_id), + ) + finally: + conn.close() log.info(f"Added message to history: role={role}, est_tokens={est}, content={content[:30]}...") def get_history(self, limit: int = 100) -> list[dict]: - with get_db_connection(self.db_path) as conn: + conn = get_db_connection(self.db_path) + try: rows = conn.execute("""SELECT role, content FROM messages WHERE channel = ? ORDER BY id DESC LIMIT ?""", (self.channel, limit)).fetchall() - conn.close() + finally: + conn.close() return [{"role": row[0], "content": row[1]} for row in reversed(rows)]