COR-891 Fix SQLite flow initialization contention (#7821)

* COR-891 Fix SQLite flow initialization contention

* fix: update vulnerable auth dependencies (COR-891)

* COR-891 Fix SQLite flow initialization contention

lib/crewai/pyproject.toml

@@ -30,6 +30,7 @@ dependencies = [
     # Authentication and Security
     "python-dotenv>=1.2.2,<2",
     "pyjwt>=2.15.0,<3",
+    "oauthlib>=4.0.0,<5",
     # Configuration and Utils
     "click>=8.1.7,<9",
     "appdirs~=1.4.4",

lib/crewai/src/crewai/flow/persistence/sqlite.py

@@ -8,6 +8,7 @@ import json
 import os
 from pathlib import Path
 import sqlite3
+from time import monotonic, sleep
 from typing import TYPE_CHECKING, Any
 
 from crewai_core.lock_store import lock as store_lock
@@ -22,6 +23,10 @@ if TYPE_CHECKING:
     from crewai.flow.async_feedback.types import PendingFeedbackContext
 
 
+_SQLITE_BUSY_TIMEOUT_SECONDS = 30.0
+_SCHEMA_RETRY_DELAY_SECONDS = 0.05
+
+
 def _json_default(obj: Any) -> Any:
     """Fallback serializer for non-primitive types in JSON dumps."""
     if isinstance(obj, BaseModel):
@@ -85,9 +90,21 @@ class SQLiteFlowPersistence(FlowPersistence):
 
     def init_db(self) -> None:
         """Create the necessary tables if they don't exist."""
+        deadline = monotonic() + _SQLITE_BUSY_TIMEOUT_SECONDS
+        while True:
+            try:
+                self._create_schema()
+                return
+            except sqlite3.OperationalError as exc:  # noqa: PERF203
+                if "locked" not in str(exc).lower() or monotonic() >= deadline:
+                    raise
+                sleep(_SCHEMA_RETRY_DELAY_SECONDS)
+
+    def _create_schema(self) -> None:
         with (
-            store_lock(self._lock_name),
-            closing(sqlite3.connect(self.db_path, timeout=30)) as conn,
+            closing(
+                sqlite3.connect(self.db_path, timeout=_SQLITE_BUSY_TIMEOUT_SECONDS)
+            ) as conn,
             conn,
         ):
             conn.execute("PRAGMA journal_mode=WAL")
@@ -190,7 +207,9 @@ class SQLiteFlowPersistence(FlowPersistence):
 
         with (
             store_lock(self._lock_name),
-            closing(sqlite3.connect(self.db_path, timeout=30)) as conn,
+            closing(
+                sqlite3.connect(self.db_path, timeout=_SQLITE_BUSY_TIMEOUT_SECONDS)
+            ) as conn,
             conn,
         ):
             self._save_state_sql(conn, flow_uuid, method_name, state_dict)
@@ -204,7 +223,12 @@ class SQLiteFlowPersistence(FlowPersistence):
         Returns:
             The most recent state as a dictionary, or None if no state exists
         """
-        with closing(sqlite3.connect(self.db_path, timeout=30)) as conn, conn:
+        with (
+            closing(
+                sqlite3.connect(self.db_path, timeout=_SQLITE_BUSY_TIMEOUT_SECONDS)
+            ) as conn,
+            conn,
+        ):
             cursor = conn.execute(
                 """
             SELECT state_json
@@ -242,7 +266,9 @@ class SQLiteFlowPersistence(FlowPersistence):
 
         with (
             store_lock(self._lock_name),
-            closing(sqlite3.connect(self.db_path, timeout=30)) as conn,
+            closing(
+                sqlite3.connect(self.db_path, timeout=_SQLITE_BUSY_TIMEOUT_SECONDS)
+            ) as conn,
             conn,
         ):
             self._save_state_sql(conn, flow_uuid, context.method_name, state_dict)
@@ -280,7 +306,12 @@ class SQLiteFlowPersistence(FlowPersistence):
         # Import here to avoid circular imports
         from crewai.flow.async_feedback.types import PendingFeedbackContext
 
-        with closing(sqlite3.connect(self.db_path, timeout=30)) as conn, conn:
+        with (
+            closing(
+                sqlite3.connect(self.db_path, timeout=_SQLITE_BUSY_TIMEOUT_SECONDS)
+            ) as conn,
+            conn,
+        ):
             cursor = conn.execute(
                 """
             SELECT state_json, context_json
@@ -306,7 +337,9 @@ class SQLiteFlowPersistence(FlowPersistence):
         """
         with (
             store_lock(self._lock_name),
-            closing(sqlite3.connect(self.db_path, timeout=30)) as conn,
+            closing(
+                sqlite3.connect(self.db_path, timeout=_SQLITE_BUSY_TIMEOUT_SECONDS)
+            ) as conn,
             conn,
         ):
             conn.execute(

lib/crewai/tests/test_sqlite_connection_lifecycle.py

@@ -2,17 +2,27 @@
 
 from __future__ import annotations
 
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import closing
 import json
 import os
-import sqlite3
 from pathlib import Path
+import sqlite3
+import subprocess
+import sys
+from threading import Barrier
 
 import pytest
 from crewai.flow.async_feedback.types import PendingFeedbackContext
+from crewai.flow.persistence import sqlite as sqlite_module
 from crewai.flow.persistence.sqlite import SQLiteFlowPersistence
 from crewai.state.provider.sqlite_provider import SqliteProvider
 
 
+_CONCURRENT_START_TIMEOUT_SECONDS = 10
+_CHILD_PROCESS_TIMEOUT_SECONDS = 60
+
+
 def _track_connections(
     monkeypatch: pytest.MonkeyPatch,
 ) -> list[sqlite3.Connection]:
@@ -111,6 +121,79 @@ def test_flow_persistence_failed_write_rolls_back_and_closes(
     assert persistence.load_state("flow-1") is None
 
 
+@pytest.mark.parametrize("existing_db", [False, True])
+def test_parallel_flow_initialization_does_not_use_store_lock(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, existing_db: bool
+) -> None:
+    db_path = tmp_path / "flows.db"
+    if existing_db:
+        persistence = SQLiteFlowPersistence(str(db_path))
+        persistence.save_state("prior-flow", "start", {"step": 1})
+        persistence.save_pending_feedback("prior-flow", _make_context(), {"step": 2})
+
+    def reject_store_lock(*args: object, **kwargs: object) -> None:
+        raise AssertionError("initialization must not acquire a shared store lock")
+
+    monkeypatch.setattr(sqlite_module, "store_lock", reject_store_lock)
+    barrier = Barrier(16)
+
+    def initialize(_: int) -> SQLiteFlowPersistence:
+        barrier.wait(timeout=_CONCURRENT_START_TIMEOUT_SECONDS)
+        return SQLiteFlowPersistence(str(db_path))
+
+    with ThreadPoolExecutor(max_workers=16) as pool:
+        persistence_instances = list(pool.map(initialize, range(16)))
+
+    assert len(persistence_instances) == 16
+    with closing(sqlite3.connect(db_path)) as conn:
+        assert conn.execute("PRAGMA journal_mode").fetchone() == ("wal",)
+        assert {
+            row[0]
+            for row in conn.execute(
+                "SELECT name FROM sqlite_master WHERE type = 'table'"
+            )
+        } >= {"flow_states", "pending_feedback"}
+    if existing_db:
+        assert persistence_instances[0].load_state("prior-flow") == {"step": 2}
+        pending = persistence_instances[0].load_pending_feedback("prior-flow")
+        assert pending is not None and pending[0] == {"step": 2}
+
+
+@pytest.mark.parametrize("existing_db", [False, True])
+def test_parallel_processes_initialize_flow_db(
+    tmp_path: Path, existing_db: bool
+) -> None:
+    db_path = tmp_path / "flows.db"
+    if existing_db:
+        persistence = SQLiteFlowPersistence(str(db_path))
+        persistence.save_state("prior-flow", "start", {"step": 1})
+
+    barrier = Barrier(2)
+    script = (
+        "import sys; from crewai.flow.persistence.sqlite import SQLiteFlowPersistence; "
+        "SQLiteFlowPersistence(sys.argv[1])"
+    )
+
+    def initialize(_: int) -> None:
+        barrier.wait(timeout=_CONCURRENT_START_TIMEOUT_SECONDS)
+        subprocess.run(
+            [sys.executable, "-c", script, str(db_path)],
+            check=True,
+            timeout=_CHILD_PROCESS_TIMEOUT_SECONDS,
+        )
+
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        list(pool.map(initialize, range(2)))
+
+    with closing(sqlite3.connect(db_path)) as conn:
+        assert conn.execute("PRAGMA journal_mode").fetchone() == ("wal",)
+    persistence = SQLiteFlowPersistence(str(db_path))
+    if existing_db:
+        assert persistence.load_state("prior-flow") == {"step": 1}
+    persistence.save_state("new-flow", "start", {"step": 2})
+    assert persistence.load_state("new-flow") == {"step": 2}
+
+
 def test_sqlite_provider_closes_every_connection(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:

pyproject.toml

@@ -172,7 +172,7 @@ info = "Commits must follow Conventional Commits 1.0.0."
 [tool.uv]
 exclude-newer = "3 days"
 # These security fixes are newer than the global supply-chain cutoff.
-exclude-newer-package = { msgpack = "2026-06-20T00:00:00Z", pydantic-settings = "2026-06-20T00:00:00Z", langsmith = "2026-06-20T00:00:00Z" }
+exclude-newer-package = { msgpack = "2026-06-20T00:00:00Z", pydantic-settings = "2026-06-20T00:00:00Z", langsmith = "2026-06-20T00:00:00Z", oauthlib = "2026-09-29T00:00:00Z" }
 
 # composio-core pins rich<14 but textual requires rich>=14.
 # onnxruntime 1.24+ dropped Python 3.10 wheels; cap it so qdrant[fastembed] resolves on 3.10.

uv.lock

@@ -23,6 +23,7 @@ exclude-newer-span = "P3D"
 [options.exclude-newer-package]
 msgpack = "2026-06-20T00:00:00Z"
 langsmith = "2026-06-20T00:00:00Z"
+oauthlib = "2026-09-29T00:00:00Z"
 pydantic-settings = "2026-06-20T00:00:00Z"
 
 [manifest]
@@ -1448,6 +1449,7 @@ dependencies = [
     { name = "jsonref" },
     { name = "lancedb" },
     { name = "mcp" },
+    { name = "oauthlib" },
     { name = "openai" },
     { name = "openpyxl" },
     { name = "opentelemetry-api" },
@@ -1565,6 +1567,7 @@ requires-dist = [
     { name = "litellm", marker = "extra == 'litellm'", specifier = ">=1.88.6,<2" },
     { name = "mcp", specifier = "~=1.28.1" },
     { name = "mem0ai", marker = "extra == 'mem0'", specifier = ">=2.0.0,<3" },
+    { name = "oauthlib", specifier = ">=4.0.0,<5" },
     { name = "openai", specifier = ">=2.30.0,<3" },
     { name = "openpyxl", specifier = "~=3.1.5" },
     { name = "openpyxl", marker = "extra == 'openpyxl'", specifier = "~=3.1.5" },