fix: harden sqlite lock retry handling

Author: zouyonghe

astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py

@@ -521,10 +521,25 @@ async def _save_to_history(
 BLOCKED = {"dGZid2h2d3IuY2xvdWQuc2VhbG9zLmlv", "a291cmljaGF0"}
 decoded_blocked = [base64.b64decode(b).decode("utf-8") for b in BLOCKED]
 
+PROVIDER_STATS_SQLITE_LOCK_RETRY_ATTEMPTS = 3
+PROVIDER_STATS_SQLITE_LOCK_RETRY_BASE_DELAY = 0.2
+
+
+def _iter_exception_messages(exc: BaseException):
+    yield str(exc)
+    orig = getattr(exc, "orig", None)
+    if orig is not None and orig is not exc:
+        yield str(orig)
+        yield from (str(arg) for arg in getattr(orig, "args", ()) if arg)
+    yield from (str(arg) for arg in getattr(exc, "args", ()) if arg)
+
 
 def _is_sqlite_database_locked_error(exc: Exception) -> bool:
-    return (
-        isinstance(exc, OperationalError) and "database is locked" in str(exc).lower()
+    if not isinstance(exc, OperationalError):
+        return False
+    return any(
+        "database" in message.lower() and "locked" in message.lower()
+        for message in _iter_exception_messages(exc)
     )
 
 
@@ -557,7 +572,7 @@ async def _record_internal_agent_stats(
     else:
         status = "completed"
 
-    for attempt in range(3):
+    for attempt in range(PROVIDER_STATS_SQLITE_LOCK_RETRY_ATTEMPTS):
         try:
             await db_helper.insert_provider_stat(
                 umo=event.unified_msg_origin,
@@ -569,9 +584,19 @@ async def _record_internal_agent_stats(
                 agent_type="internal",
             )
             return
-        except Exception as e:
-            if _is_sqlite_database_locked_error(e) and attempt < 2:
-                await asyncio.sleep(0.2 * (2**attempt))
+        except OperationalError as e:
+            if (
+                _is_sqlite_database_locked_error(e)
+                and attempt < PROVIDER_STATS_SQLITE_LOCK_RETRY_ATTEMPTS - 1
+            ):
+                await asyncio.sleep(
+                    PROVIDER_STATS_SQLITE_LOCK_RETRY_BASE_DELAY * (2**attempt)
+                )
                 continue
             logger.warning("Persist provider stats failed: %s", e, exc_info=True)
             return
+        except asyncio.CancelledError:
+            raise
+        except Exception as e:
+            logger.warning("Persist provider stats failed: %s", e, exc_info=True)
+            return

tests/unit/test_provider_stats.py

@@ -112,3 +112,97 @@ async def no_sleep(delay: float) -> None:
     )
 
     assert attempts == 2
+
+
+@pytest.mark.asyncio
+async def test_record_internal_agent_stats_retries_database_table_lock(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    attempts = 0
+
+    class LockedOnceDb:
+        async def insert_provider_stat(self, **kwargs):
+            nonlocal attempts
+            attempts += 1
+            if attempts == 1:
+                raise OperationalError(
+                    "insert into provider_stats",
+                    {},
+                    Exception("database table is locked"),
+                )
+            return SimpleNamespace(**kwargs)
+
+    monkeypatch.setattr(internal, "db_helper", LockedOnceDb())
+
+    async def no_sleep(delay: float) -> None:
+        return None
+
+    monkeypatch.setattr(internal.asyncio, "sleep", no_sleep)
+
+    event = SimpleNamespace(unified_msg_origin="webchat:FriendMessage:session-42")
+    provider = SimpleNamespace(
+        provider_config={"id": "provider-1"},
+        meta=lambda: SimpleNamespace(id="provider-1", type="openai"),
+        get_model=lambda: "gpt-4.1",
+    )
+    agent_runner = SimpleNamespace(
+        provider=provider,
+        stats=AgentStats(),
+        was_aborted=lambda: False,
+    )
+
+    await internal._record_internal_agent_stats(
+        event,
+        ProviderRequest(conversation=SimpleNamespace(cid="conv-123")),
+        agent_runner,
+        SimpleNamespace(role="assistant"),
+    )
+
+    assert attempts == 2
+
+
+@pytest.mark.asyncio
+async def test_record_internal_agent_stats_does_not_retry_other_operational_errors(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    attempts = 0
+    warnings = []
+
+    class FailingDb:
+        async def insert_provider_stat(self, **kwargs):
+            nonlocal attempts
+            attempts += 1
+            raise OperationalError(
+                "insert into provider_stats",
+                {},
+                Exception("no such table: provider_stats"),
+            )
+
+    monkeypatch.setattr(internal, "db_helper", FailingDb())
+    monkeypatch.setattr(
+        internal.logger,
+        "warning",
+        lambda *args, **kwargs: warnings.append((args, kwargs)),
+    )
+
+    event = SimpleNamespace(unified_msg_origin="webchat:FriendMessage:session-42")
+    provider = SimpleNamespace(
+        provider_config={"id": "provider-1"},
+        meta=lambda: SimpleNamespace(id="provider-1", type="openai"),
+        get_model=lambda: "gpt-4.1",
+    )
+    agent_runner = SimpleNamespace(
+        provider=provider,
+        stats=AgentStats(),
+        was_aborted=lambda: False,
+    )
+
+    await internal._record_internal_agent_stats(
+        event,
+        ProviderRequest(conversation=SimpleNamespace(cid="conv-123")),
+        agent_runner,
+        SimpleNamespace(role="assistant"),
+    )
+
+    assert attempts == 1
+    assert len(warnings) == 1