fix: tolerate transient SQLite write contention

openhands/automation/db.py

@@ -15,6 +15,7 @@
 from typing import Any
 
 from fastapi import Request
+from sqlalchemy import event
 from sqlalchemy.engine import URL
 from sqlalchemy.ext.asyncio import (
     AsyncEngine,
@@ -28,6 +29,7 @@
 
 logger = logging.getLogger("automation.db")
 SUPPORTED_DB_SSL_MODES = {"prefer", "require", "disable"}
+SQLITE_BUSY_TIMEOUT_SECONDS = 30
 
 
 def _normalize_db_ssl_mode(db_ssl_mode: str | None) -> str | None:
@@ -163,10 +165,24 @@ def _create_sqlite_engine(db_url: str) -> EngineResult:
     engine = create_async_engine(
         db_url,
         # SQLite-specific settings
-        connect_args={"check_same_thread": False},
+        connect_args={
+            "check_same_thread": False,
+            "timeout": SQLITE_BUSY_TIMEOUT_SECONDS,
+        },
         # No pooling for SQLite - it handles this internally
         pool_pre_ping=True,
     )
+
+    @event.listens_for(engine.sync_engine, "connect")
+    def configure_sqlite_connection(dbapi_connection: Any, _: Any) -> None:
+        cursor = dbapi_connection.cursor()
+        try:
+            cursor.execute("PRAGMA journal_mode=WAL")
+            cursor.execute("PRAGMA foreign_keys=ON")
+            cursor.execute(f"PRAGMA busy_timeout={SQLITE_BUSY_TIMEOUT_SECONDS * 1000}")
+        finally:
+            cursor.close()
+
     logger.info("Created SQLite engine: %s", db_url.split("?")[0])
     return EngineResult(engine=engine, is_sqlite=True)
 

tests/test_db.py

@@ -1,11 +1,14 @@
 """Tests for database module."""
 
+import asyncio
 import os
+import sqlite3
 import tempfile
 from pathlib import Path
 from typing import Any
 
 import pytest
+from sqlalchemy import text
 
 from openhands.automation import db as db_module
 from openhands.automation.config import ServiceSettings
@@ -97,6 +100,56 @@ def test_absolute_path(self):
         result = _create_sqlite_engine("sqlite+aiosqlite:////data/automations.db")
         assert result.is_sqlite is True
 
+    @pytest.mark.asyncio
+    async def test_configures_file_database_for_concurrent_service_tasks(
+        self, tmp_path: Path
+    ):
+        result = _create_sqlite_engine(f"sqlite+aiosqlite:///{tmp_path / 'test.db'}")
+        try:
+            async with result.engine.connect() as connection:
+                assert (
+                    await connection.execute(text("PRAGMA journal_mode"))
+                ).scalar() == "wal"
+                assert (
+                    await connection.execute(text("PRAGMA foreign_keys"))
+                ).scalar() == 1
+                assert (
+                    await connection.execute(text("PRAGMA busy_timeout"))
+                ).scalar() == 30_000
+        finally:
+            await result.dispose()
+
+    @pytest.mark.asyncio
+    async def test_waits_for_a_short_lived_write_lock(self, tmp_path: Path):
+        path = tmp_path / "contention.db"
+        result = _create_sqlite_engine(f"sqlite+aiosqlite:///{path}")
+        try:
+            async with result.engine.begin() as connection:
+                await connection.execute(text("CREATE TABLE writes (value INTEGER)"))
+
+            lock = sqlite3.connect(path)
+            lock.execute("BEGIN IMMEDIATE")
+            lock.execute("INSERT INTO writes VALUES (1)")
+
+            async def write_from_service_connection() -> None:
+                async with result.engine.begin() as connection:
+                    await connection.execute(text("INSERT INTO writes VALUES (2)"))
+
+            pending_write = asyncio.create_task(write_from_service_connection())
+            await asyncio.sleep(0.1)
+            assert not pending_write.done()
+            lock.commit()
+            lock.close()
+            await asyncio.wait_for(pending_write, timeout=2)
+
+            async with result.engine.connect() as connection:
+                count = (
+                    await connection.execute(text("SELECT count(*) FROM writes"))
+                ).scalar()
+            assert count == 2
+        finally:
+            await result.dispose()
+
 
 class TestEngineResult:
     """Tests for EngineResult dataclass."""