fix: prevent SharedPreferences deadlocks

[Soulter]

Summary

• replace the blocking SharedPreferences sync bridge with a startup-preloaded cache and an ordered persistence queue on the main event loop
• keep legacy synchronous APIs non-blocking and mark the public sync provider/tool APIs as deprecated
• add async provider selection and LLM tool toggle APIs, then migrate core, built-in, pipeline, and dashboard call sites
• flush pending preference writes during shutdown and keep the cache consistent when plugin KV data is cleared
• update the plugin development documentation and add deadlock, persistence, preload, and FIFO ordering regression tests

Root cause

The deprecated synchronous SharedPreferences methods submitted database coroutines to a secondary event loop and blocked on .result(). The secondary loop used the same SQLAlchemy async connection pool as the main loop. When the pool was exhausted, the main loop could block waiting for the synchronous result while also being the only loop able to release the required connection, creating a circular wait.

The new implementation serves synchronous reads from memory and makes synchronous writes update the cache immediately before scheduling ordered persistence. Async writes await persistence completion and surface database failures.

Fixes #9633.

Validation

• uv run ruff format .
• uv run ruff check .
• focused regression suite: 252 passed
• full tests/ suite: 2128 passed, 2 deselected
  • the two deselected tests are pre-existing dashboard caplog capture failures where the expected warning is emitted to stdout

Summary by Sourcery

Replace the blocking SharedPreferences sync bridge with an event-loop-bound cached store and ordered write queue, and introduce async provider selection and tool toggle APIs used across core, pipeline, plugins, and dashboard.

Bug Fixes:

• Eliminate SharedPreferences deadlocks by serving sync reads from an in-memory cache and decoupling sync writes from direct database blocking on shared async pools.

Enhancements:

• Add initialization, flush, and close lifecycle hooks for SharedPreferences to preload preferences, ensure FIFO persistence, and flush writes on shutdown and restart.
• Introduce async provider selection APIs for chat, TTS, and STT and migrate call sites to use them while keeping deprecated synchronous variants for backward compatibility.
• Add async activation/deactivation APIs for LLM tools that persist their state via SharedPreferences, and wire these into StarTools and dashboard tool toggling.
• Relax preference DB queries to support optional scope filtering and use SharedPreferences to clear plugin KV data rather than direct DB access.

Documentation:

• Update plugin development documentation to recommend the new async provider APIs and clarify deprecation status of synchronous provider methods.

Tests:

• Add a dedicated SharedPreferences test suite covering cache preload, non-blocking sync access, FIFO write ordering, and connection pool exhaustion behavior.
• Extend tests for provider selection, tool manager behavior, dashboard endpoints, plugin cleanup, and group chat wiring to cover the new async APIs and preference-backed behaviors.

[sourcery-ai]

Hey - I've found 4 issues

Prompt for AI Agents

Please address the comments from this code review:

## Individual Comments

### Comment 1
<location path="astrbot/core/utils/shared_preferences.py" line_range="125-134" />
<code_context>
+    async def _drain_write_queue(self) -> None:
</code_context>
<issue_to_address>
**issue (bug_risk):** Queued writes can get stuck if they are enqueued while the writer task is still running but the queue has just become empty.

Because `_drain_write_queue` uses `queue.get_nowait()` in a loop and `_schedule_write` only starts a writer when `self._writer_task is None or self._writer_task.done()`, there’s a race where the queue becomes empty, the task is still marked running, and a new item is enqueued before `_writer_task` completes. In that case no new writer task is scheduled and the new item is never processed. To avoid stuck writes, consider making `_drain_write_queue` a long‑lived consumer using `await queue.get()` in a loop, exiting only on an explicit shutdown/sentinel, or alternately removing `_writer_task` and spawning a fresh task per flush.
</issue_to_address>

### Comment 2
<location path="astrbot/core/provider/manager.py" line_range="218-227" />
<code_context>

-    def get_using_provider(
-        self, provider_type: ProviderType, umo=None
+    def _resolve_using_provider(
+        self,
+        provider_type: ProviderType,
</code_context>
<issue_to_address>
**nitpick:** The helper method’s docstring advertises a ValueError that is never raised, which may mislead callers.

The `_resolve_using_provider` docstring claims it may raise `ValueError` for unsupported `provider_type`, but the implementation only returns `None` for disabled/unavailable providers. This contract mismatch can cause callers to handle exceptions that never occur instead of the `None` case.

Please either add explicit validation that raises on unsupported `ProviderType` values, or update the docstring to reflect the current behavior (no exceptions, `None` as the failure mode), especially since both sync and async `get_using_provider` variants now rely on this helper.
</issue_to_address>

### Comment 3
<location path="tests/unit/test_shared_preferences.py" line_range="28-31" />
<code_context>
+async def test_sync_put_updates_cache_and_persists_without_blocking(preferences):
+    store, database = preferences
+
+    started = time.monotonic()
+    store.put("theme", "dark", scope="global", scope_id="global")
+
+    assert time.monotonic() - started < 0.1
+    assert store.get("theme", scope="global", scope_id="global") == "dark"
+
</code_context>
<issue_to_address>
**suggestion (testing):** The strict timing assertion in `test_sync_put_updates_cache_and_persists_without_blocking` may be flaky across environments.

This test encodes a 100ms latency guarantee for `store.put`, which is likely to be flaky under CI load or on slower systems. Since the goal is to verify non-blocking behavior rather than exact timing, consider either mocking the persistence path so the test only asserts that `put` returns without awaiting, or relaxing the timing constraint (e.g., relative to a known blocking operation) to avoid spurious failures unrelated to the deadlock fix.

Suggested implementation:

```python
import time

import pytest
import pytest_asyncio

```

```python
@pytest.mark.asyncio
async def test_sync_put_updates_cache_and_persists_without_blocking(preferences):
    store, database = preferences

    # This test verifies that `put` returns promptly and updates the in-memory cache
    # without asserting a very tight latency bound that could be flaky under CI load.
    started = time.monotonic()
    store.put("theme", "dark", scope="global", scope_id="global")
    elapsed = time.monotonic() - started

    # Allow a more relaxed upper bound to avoid spurious failures while still
    # catching pathological blocking behavior.
    assert elapsed < 0.5
    assert store.get("theme", scope="global", scope_id="global") == "dark"

```

Depending on how `SharedPreferences` and the underlying persistence layer are implemented, you may want to further align this test with the non-blocking guarantee by:
1. Introducing a mock or spy for the persistence method (e.g., the database write call) and asserting that it is invoked asynchronously after `put` returns, rather than relying on wall-clock timing.
2. If you adopt that mocking approach, the `elapsed` assertion could be removed entirely, and the test would instead assert that `put` does not await the persistence coroutine (for example, by making the mocked persistence artificially slow and confirming that `put` returns before it completes).
These adjustments will require access to the actual `SharedPreferences` implementation and the persistence API to determine the best hook point for mocking.
</issue_to_address>

### Comment 4
<location path="astrbot/core/utils/shared_preferences.py" line_range="46" />
<code_context>
+        self._cache_initialized = False
+        self._initializing = False
+        self._initialize_lock = asyncio.Lock()
+        self._loop: asyncio.AbstractEventLoop | None = None
+        self._write_queue: asyncio.Queue[_WriteOperation] | None = None
+        self._writer_task: asyncio.Task[None] | None = None
</code_context>
<issue_to_address>
**issue (complexity):** Consider simplifying SharedPreferences to a single event loop with one writer task and queue, removing multi-loop rebinding logic, pending write handling, and extra state flags.

The main complexity comes from supporting rebinding across multiple event loops and coordinating `_pending_writes` vs the live `_write_queue`. If you don’t actually need multi-loop ownership, you can simplify to a single-loop, single-queue model and drop most of the state flags without losing any behaviour.

### 1. Bind to a single event loop and remove rebinding / `_pending_writes`

You can bind once in `initialize()` and enforce a stricter contract:

```python
class SharedPreferences:
    def __init__(self, db_helper: BaseDatabase, json_storage_path=None) -> None:
        ...
        self._cache: dict[tuple[str, str, str], Any] = {}
        self._cache_lock = threading.RLock()
        self._cache_initialized = False

        self._loop: asyncio.AbstractEventLoop | None = None
        self._write_queue: asyncio.Queue[_WriteOperation] | None = None
        self._writer_task: asyncio.Task[None] | None = None
```

```python
    async def initialize(self) -> None:
        loop = asyncio.get_running_loop()
        async with self._initialize_lock:
            if self._loop is not None and self._loop is not loop:
                raise RuntimeError("SharedPreferences must be used on a single event loop")

            if self._cache_initialized:
                return

            self._loop = loop
            self._write_queue = asyncio.Queue()
            self._writer_task = loop.create_task(
                self._writer(), name="shared_preferences_writer"
            )

            preferences = await self.db_helper.get_preferences()
            with self._cache_lock:
                self._cache = {
                    (item.scope, item.scope_id, item.key): deepcopy(item.value["val"])
                    for item in preferences
                }
                self._cache_initialized = True
```

Key changes:

- No `_initializing`, `_pending_writes`, or queue migration from an old loop.
- If someone calls from a different loop, you fail fast with a clear error instead of trying to rebind.

### 2. Single writer task that owns both cache and DB mutations

Instead of `_apply_cache_operation` + `_schedule_write` + `_drain_write_queue`, you can make the writer the single source of truth:

```python
    def _submit_write(self, operation: _WriteOperation) -> None:
        # Caller must have ensured initialize() was called
        if self._write_queue is None:
            raise RuntimeError("SharedPreferences not initialized")

        self._write_queue.put_nowait(operation)
```

```python
    async def _writer(self) -> None:
        assert self._write_queue is not None
        while True:
            action, scope, scope_id, key, value, completion = await self._write_queue.get()
            try:
                # Update cache first
                with self._cache_lock:
                    if action == "put" and key is not None:
                        self._cache[(scope, scope_id, key)] = deepcopy(value)
                    elif action == "remove" and key is not None:
                        self._cache.pop((scope, scope_id, key), None)
                    elif action == "clear":
                        keys = [
                            k for k in self._cache
                            if k[0] == scope and k[1] == scope_id
                        ]
                        for k in keys:
                            self._cache.pop(k, None)

                # Then persist
                if action == "put" and key is not None:
                    await self.db_helper.insert_preference_or_update(
                        scope, scope_id, key, {"val": value}
                    )
                elif action == "remove" and key is not None:
                    await self.db_helper.remove_preference(scope, scope_id, key)
                elif action == "clear":
                    await self.db_helper.clear_preferences(scope, scope_id)
                else:
                    raise ValueError(f"Unknown preference write operation: {action}")

                if completion is not None and not completion.done():
                    completion.set_result(None)
            except Exception as exc:
                logger.error(
                    "Failed to persist shared preference operation %s for %s/%s: %s",
                    action, scope, scope_id, exc, exc_info=True,
                )
                if completion is not None and not completion.done():
                    completion.set_exception(exc)
            finally:
                self._write_queue.task_done()
```

Effects:

- All mutations (cache + DB) are linearized through one queue and one task.
- `_apply_cache_operation` and `_schedule_write` can be removed entirely.
- You no longer need `_initializing` or to replay `_pending_writes` into both cache and queue.

### 3. Simplify `flush()` around the single writer/queue

With the writer always running after initialization, `flush()` becomes simpler:

```python
    async def flush(self) -> None:
        if self._write_queue is None or self._loop is None:
            return

        loop = asyncio.get_running_loop()
        if loop is not self._loop:
            raise RuntimeError("flush() must be called on the owning event loop")

        await self._write_queue.join()
```

No need to:

- Check `loop.is_running()` vs a stopped loop.
- Call `initialize()` from here or handle rebinding.
- Await `_writer_task` each time; the writer runs for the lifetime of the store.

### 4. Deprecated sync APIs: keep behaviour simple and consistent

If the deprecated sync APIs still need to hit the DB (and be consistent with async ones), you can reuse the async pipeline via the main loop instead of direct cache writes:

```python
    @deprecated(...)
    def put(self, key, value, scope: str | None = None, scope_id: str | None = None) -> None:
        if self._loop is None:
            raise RuntimeError("SharedPreferences must be initialized before sync calls")
        op_scope = scope or "unknown"
        op_scope_id = scope_id or "unknown"
        fut = asyncio.run_coroutine_threadsafe(
            self.put_async(op_scope, op_scope_id, key, value),
            self._loop,
        )
        fut.result()
```

Same pattern for `get`, `remove`, `clear`. That avoids:

- Divergent semantics between sync and async paths.
- Additional cache-only state that depends on whether `initialize()`/`flush()` ran.

---

All of the above keeps your new features (async access, in-memory cache, batching writes) but removes the multi-loop and pending-write state machine, which is where most of the incidental complexity comes from.
</issue_to_address>

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}

[sourcery-ai]

issue (bug_risk): Queued writes can get stuck if they are enqueued while the writer task is still running but the queue has just become empty.

Because _drain_write_queue uses queue.get_nowait() in a loop and _schedule_write only starts a writer when self._writer_task is None or self._writer_task.done(), there’s a race where the queue becomes empty, the task is still marked running, and a new item is enqueued before _writer_task completes. In that case no new writer task is scheduled and the new item is never processed. To avoid stuck writes, consider making _drain_write_queue a long‑lived consumer using await queue.get() in a loop, exiting only on an explicit shutdown/sentinel, or alternately removing _writer_task and spawning a fresh task per flush.

[sourcery-ai]

nitpick: The helper method’s docstring advertises a ValueError that is never raised, which may mislead callers.

The _resolve_using_provider docstring claims it may raise ValueError for unsupported provider_type, but the implementation only returns None for disabled/unavailable providers. This contract mismatch can cause callers to handle exceptions that never occur instead of the None case.

Please either add explicit validation that raises on unsupported ProviderType values, or update the docstring to reflect the current behavior (no exceptions, None as the failure mode), especially since both sync and async get_using_provider variants now rely on this helper.

[sourcery-ai]

suggestion (testing): The strict timing assertion in test_sync_put_updates_cache_and_persists_without_blocking may be flaky across environments.

This test encodes a 100ms latency guarantee for store.put, which is likely to be flaky under CI load or on slower systems. Since the goal is to verify non-blocking behavior rather than exact timing, consider either mocking the persistence path so the test only asserts that put returns without awaiting, or relaxing the timing constraint (e.g., relative to a known blocking operation) to avoid spurious failures unrelated to the deadlock fix.

Suggested implementation:

import time

import pytest
import pytest_asyncio

@pytest.mark.asyncio
async def test_sync_put_updates_cache_and_persists_without_blocking(preferences):
    store, database = preferences

    # This test verifies that `put` returns promptly and updates the in-memory cache
    # without asserting a very tight latency bound that could be flaky under CI load.
    started = time.monotonic()
    store.put("theme", "dark", scope="global", scope_id="global")
    elapsed = time.monotonic() - started

    # Allow a more relaxed upper bound to avoid spurious failures while still
    # catching pathological blocking behavior.
    assert elapsed < 0.5
    assert store.get("theme", scope="global", scope_id="global") == "dark"

Depending on how SharedPreferences and the underlying persistence layer are implemented, you may want to further align this test with the non-blocking guarantee by:

1. Introducing a mock or spy for the persistence method (e.g., the database write call) and asserting that it is invoked asynchronously after put returns, rather than relying on wall-clock timing.
2. If you adopt that mocking approach, the elapsed assertion could be removed entirely, and the test would instead assert that put does not await the persistence coroutine (for example, by making the mocked persistence artificially slow and confirming that put returns before it completes).
   These adjustments will require access to the actual SharedPreferences implementation and the persistence API to determine the best hook point for mocking.

[sourcery-ai]

issue (complexity): Consider simplifying SharedPreferences to a single event loop with one writer task and queue, removing multi-loop rebinding logic, pending write handling, and extra state flags.

The main complexity comes from supporting rebinding across multiple event loops and coordinating _pending_writes vs the live _write_queue. If you don’t actually need multi-loop ownership, you can simplify to a single-loop, single-queue model and drop most of the state flags without losing any behaviour.

1. Bind to a single event loop and remove rebinding / _pending_writes

You can bind once in initialize() and enforce a stricter contract:

class SharedPreferences:
    def __init__(self, db_helper: BaseDatabase, json_storage_path=None) -> None:
        ...
        self._cache: dict[tuple[str, str, str], Any] = {}
        self._cache_lock = threading.RLock()
        self._cache_initialized = False

        self._loop: asyncio.AbstractEventLoop | None = None
        self._write_queue: asyncio.Queue[_WriteOperation] | None = None
        self._writer_task: asyncio.Task[None] | None = None

    async def initialize(self) -> None:
        loop = asyncio.get_running_loop()
        async with self._initialize_lock:
            if self._loop is not None and self._loop is not loop:
                raise RuntimeError("SharedPreferences must be used on a single event loop")

            if self._cache_initialized:
                return

            self._loop = loop
            self._write_queue = asyncio.Queue()
            self._writer_task = loop.create_task(
                self._writer(), name="shared_preferences_writer"
            )

            preferences = await self.db_helper.get_preferences()
            with self._cache_lock:
                self._cache = {
                    (item.scope, item.scope_id, item.key): deepcopy(item.value["val"])
                    for item in preferences
                }
                self._cache_initialized = True

Key changes:

• No _initializing, _pending_writes, or queue migration from an old loop.
• If someone calls from a different loop, you fail fast with a clear error instead of trying to rebind.

2. Single writer task that owns both cache and DB mutations

Instead of _apply_cache_operation + _schedule_write + _drain_write_queue, you can make the writer the single source of truth:

    def _submit_write(self, operation: _WriteOperation) -> None:
        # Caller must have ensured initialize() was called
        if self._write_queue is None:
            raise RuntimeError("SharedPreferences not initialized")

        self._write_queue.put_nowait(operation)

    async def _writer(self) -> None:
        assert self._write_queue is not None
        while True:
            action, scope, scope_id, key, value, completion = await self._write_queue.get()
            try:
                # Update cache first
                with self._cache_lock:
                    if action == "put" and key is not None:
                        self._cache[(scope, scope_id, key)] = deepcopy(value)
                    elif action == "remove" and key is not None:
                        self._cache.pop((scope, scope_id, key), None)
                    elif action == "clear":
                        keys = [
                            k for k in self._cache
                            if k[0] == scope and k[1] == scope_id
                        ]
                        for k in keys:
                            self._cache.pop(k, None)

                # Then persist
                if action == "put" and key is not None:
                    await self.db_helper.insert_preference_or_update(
                        scope, scope_id, key, {"val": value}
                    )
                elif action == "remove" and key is not None:
                    await self.db_helper.remove_preference(scope, scope_id, key)
                elif action == "clear":
                    await self.db_helper.clear_preferences(scope, scope_id)
                else:
                    raise ValueError(f"Unknown preference write operation: {action}")

                if completion is not None and not completion.done():
                    completion.set_result(None)
            except Exception as exc:
                logger.error(
                    "Failed to persist shared preference operation %s for %s/%s: %s",
                    action, scope, scope_id, exc, exc_info=True,
                )
                if completion is not None and not completion.done():
                    completion.set_exception(exc)
            finally:
                self._write_queue.task_done()

Effects:

• All mutations (cache + DB) are linearized through one queue and one task.
• _apply_cache_operation and _schedule_write can be removed entirely.
• You no longer need _initializing or to replay _pending_writes into both cache and queue.

3. Simplify flush() around the single writer/queue

With the writer always running after initialization, flush() becomes simpler:

    async def flush(self) -> None:
        if self._write_queue is None or self._loop is None:
            return

        loop = asyncio.get_running_loop()
        if loop is not self._loop:
            raise RuntimeError("flush() must be called on the owning event loop")

        await self._write_queue.join()

No need to:

• Check loop.is_running() vs a stopped loop.
• Call initialize() from here or handle rebinding.
• Await _writer_task each time; the writer runs for the lifetime of the store.

4. Deprecated sync APIs: keep behaviour simple and consistent

If the deprecated sync APIs still need to hit the DB (and be consistent with async ones), you can reuse the async pipeline via the main loop instead of direct cache writes:

    @deprecated(...)
    def put(self, key, value, scope: str | None = None, scope_id: str | None = None) -> None:
        if self._loop is None:
            raise RuntimeError("SharedPreferences must be initialized before sync calls")
        op_scope = scope or "unknown"
        op_scope_id = scope_id or "unknown"
        fut = asyncio.run_coroutine_threadsafe(
            self.put_async(op_scope, op_scope_id, key, value),
            self._loop,
        )
        fut.result()

Same pattern for get, remove, clear. That avoids:

• Divergent semantics between sync and async paths.
• Additional cache-only state that depends on whether initialize()/flush() ran.

---

All of the above keeps your new features (async access, in-memory cache, batching writes) but removes the multi-loop and pending-write state machine, which is where most of the incidental complexity comes from.

[cloudflare-workers-and-pages]

Deploying with    Cloudflare Workers

The latest updates on your project. Learn more about integrating Git with Workers.

Status	Name	Latest Commit	Preview URL	Updated (UTC)
✅ Deployment successful! View logs	astrbot-docs	0222df3	Commit Preview URL Branch Preview URL	Aug 12 2026, 04:47 PM