Skip to content

Fix sqlite "database is locked" errors in scheduled tasks #18

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
Rikul merged 2 commits into from
Jul 2, 2026
Merged

Fix sqlite "database is locked" errors in scheduled tasks #18

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
93c7e9a
fix: enable WAL mode + busy_timeout for shared app.db to prevent lock…
claude Jul 1, 2026
939a09f
fix: address review feedback on WAL/connection handling
claude Jul 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
16 changes: 16 additions & 0 deletions app/config.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
from __future__ import annotations

import os
import sqlite3
import tomllib
from pathlib import Path

Expand All @@ -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.

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.
"""
return sqlite3.connect(db_path, timeout=timeout, isolation_level=isolation_level)


def load(path: Path | str = HOME_CONFIG_PATH) -> None:
global _config

Expand Down
117 changes: 63 additions & 54 deletions app/core/scheduled_tasks.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -19,45 +19,46 @@ 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:

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]:
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:
Expand All @@ -72,7 +73,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:
Expand All @@ -86,17 +87,19 @@ 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:
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:
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():
Expand All @@ -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 sqlite3.connect(APP_DB) 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 sqlite3.connect(APP_DB) 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)]

Expand Down Expand Up @@ -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 sqlite3.connect(APP_DB) 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)

Expand Down
7 changes: 4 additions & 3 deletions app/infra/conversations.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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,
Expand Down Expand Up @@ -79,7 +80,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)

@gemini-code-assist gemini-code-assist Bot Jul 1, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

Explicitly Enable WAL Mode During Schema Initialization

Since PRAGMA journal_mode=WAL is removed from get_db_connection to avoid connection-time overhead and potential race conditions, we should explicitly enable WAL mode once during schema initialization.

Recommendation:
Execute PRAGMA journal_mode=WAL at the start of _ensure_schema (e.g., inside the _fk_conn block or on mconn before starting the transaction):

            # 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 (
                    ...

try:
mconn.execute("PRAGMA foreign_keys = ON")
mconn.execute("BEGIN IMMEDIATE")
Expand Down
83 changes: 45 additions & 38 deletions app/infra/message_history.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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 sqlite3.connect(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)}")
Expand All @@ -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 sqlite3.connect(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 sqlite3.connect(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)]


Loading