perf(api): run handlers in threadpool, tune SQLite, precompute is_observer

Three related performance fixes for slow (>500ms) API responses.

1. Stop blocking the event loop. Route handlers were declared `async def`
   but ran synchronous SQLAlchemy queries (and synchronous Redis calls via
   the cache decorator) directly on the event loop, serializing requests.
   Convert all handlers to sync `def` so FastAPI runs them in its
   threadpool, and make the `@cached` decorator dual-mode (sync wrapper for
   sync handlers, async wrapper preserved for a future async/Postgres path).

2. Tune SQLite for concurrency. Enable WAL, busy_timeout and
   synchronous=NORMAL on every connection, and size the pool above the
   threadpool so handlers don't wait on connections. In-memory SQLite is
   guarded (no overflow-pool kwargs).

3. Precompute an indexed `nodes.is_observer` flag. The `observer=true`
   filter scanned ~68k+ event rows to find a handful of observers (the
   advertisements page calls it with limit=500). Replace the 5-way OR of
   subqueries with `WHERE nodes.is_observer = ?`. The collector sets the
   flag on first observation (in add_event_observer); the cleanup job
   clears it once a node's events are all pruned; a migration adds the
   column/index and backfills from the existing union.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

alembic/versions/20260610_1200_add_node_is_observer.py

@@ -0,0 +1,52 @@
+"""add is_observer flag to nodes
+
+Revision ID: 20260610_1200
+Revises: 20260604_1200
+Create Date: 2026-06-10 12:00:00.000000+00:00
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision: str = "20260610_1200"
+down_revision: Union[str, None] = "20260604_1200"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+# Backfill the flag from every event source, matching the union the API used to
+# evaluate at query time. Portable across SQLite and PostgreSQL.
+_BACKFILL_SQL = """
+UPDATE nodes SET is_observer = true WHERE id IN (
+    SELECT observer_node_id FROM advertisements WHERE observer_node_id IS NOT NULL
+    UNION SELECT observer_node_id FROM messages WHERE observer_node_id IS NOT NULL
+    UNION SELECT observer_node_id FROM telemetry WHERE observer_node_id IS NOT NULL
+    UNION SELECT observer_node_id FROM trace_paths WHERE observer_node_id IS NOT NULL
+    UNION SELECT observer_node_id FROM event_observers
+)
+"""
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("nodes", schema=None) as batch_op:
+        batch_op.add_column(
+            sa.Column(
+                "is_observer",
+                sa.Boolean(),
+                nullable=False,
+                server_default=sa.false(),
+            )
+        )
+        batch_op.create_index("ix_nodes_is_observer", ["is_observer"])
+
+    op.execute(_BACKFILL_SQL)
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("nodes", schema=None) as batch_op:
+        batch_op.drop_index("ix_nodes_is_observer")
+        batch_op.drop_column("is_observer")

src/meshcore_hub/api/app.py

@@ -186,12 +186,12 @@ def create_app(
 
     # Health check endpoints
     @app.get("/health", tags=["Health"])
-    async def health() -> dict:
+    def health() -> dict:
         """Basic health check."""
         return {"status": "healthy", "version": __version__}
 
     @app.get("/health/ready", tags=["Health"])
-    async def health_ready() -> dict:
+    def health_ready() -> dict:
         """Readiness check including database and optional Redis."""
         try:
             db = get_db_manager()

src/meshcore_hub/api/cache.py

@@ -1,6 +1,7 @@
 """Cache decorator for API endpoints."""
 
 import functools
+import inspect
 import json
 import logging
 from typing import Any, Callable, Optional
@@ -28,6 +29,53 @@ def _find_request(kwargs: dict[str, Any]) -> Request:
     raise TypeError("No Request parameter found in handler arguments")
 
 
+# Sentinel distinguishing "cache miss" from "cache hit holding a JSON null".
+_MISS = object()
+
+
+def _build_cache_key(
+    request: Request,
+    endpoint_name: str,
+    key_builder: Optional[Callable[[Request], str]],
+) -> str:
+    """Build the cache key for a request."""
+    if key_builder is not None:
+        return key_builder(request)
+    return f"{endpoint_name}:{sorted_query_string(request)}"
+
+
+def _lookup(cache: Any, cache_key: str, request: Request) -> Any:
+    """Return the cached value, or _MISS, and record the cache status."""
+    try:
+        cached_value = cache.get(cache_key)
+    except Exception as e:
+        logger.warning("Redis GET error for %s: %s", cache_key, e)
+        cached_value = None
+
+    if cached_value is not None:
+        logger.debug("Cache HIT: %s", cache_key)
+        request.state.cache_status = "HIT"
+        return json.loads(cached_value)
+
+    logger.debug("Cache MISS: %s", cache_key)
+    request.state.cache_status = "MISS"
+    return _MISS
+
+
+def _store(cache: Any, cache_key: str, result: Any, ttl: int) -> None:
+    """Serialize and store a handler result in the cache."""
+    try:
+        if hasattr(result, "model_dump"):
+            serialized = json.dumps(result.model_dump(mode="json"))
+        elif isinstance(result, dict):
+            serialized = json.dumps(result)
+        else:
+            serialized = json.dumps(result, default=str)
+        cache.set(cache_key, serialized, ttl)
+    except Exception as e:
+        logger.warning("Cache store error for %s: %s", cache_key, e)
+
+
 def cached(
     endpoint_name: str,
     ttl_setting: str = "redis_cache_ttl",
@@ -43,49 +91,49 @@ def cached(
     """
 
     def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
+        # Async handlers keep an async wrapper (runs on the event loop); sync
+        # handlers get a sync wrapper so FastAPI runs them in its threadpool,
+        # keeping blocking DB/Redis calls off the event loop.
+        if inspect.iscoroutinefunction(func):
+
+            @functools.wraps(func)
+            async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
+                request = _find_request(kwargs)
+                cache = getattr(request.app.state, "redis_cache", None)
+                if cache is None:
+                    return await func(*args, **kwargs)
+
+                ttl = getattr(request.app.state, ttl_setting, 30)
+                cache_key = _build_cache_key(request, endpoint_name, key_builder)
+
+                cached = _lookup(cache, cache_key, request)
+                if cached is not _MISS:
+                    return cached
+
+                result = await func(*args, **kwargs)
+                _store(cache, cache_key, result, ttl)
+                return result
+
+            return async_wrapper
+
         @functools.wraps(func)
-        async def wrapper(*args: Any, **kwargs: Any) -> Any:
+        def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
             request = _find_request(kwargs)
             cache = getattr(request.app.state, "redis_cache", None)
             if cache is None:
-                return await func(*args, **kwargs)
+                return func(*args, **kwargs)
 
             ttl = getattr(request.app.state, ttl_setting, 30)
+            cache_key = _build_cache_key(request, endpoint_name, key_builder)
 
-            if key_builder is not None:
-                cache_key = key_builder(request)
-            else:
-                cache_key = f"{endpoint_name}:{sorted_query_string(request)}"
-
-            try:
-                cached_value = cache.get(cache_key)
-            except Exception as e:
-                logger.warning("Redis GET error for %s: %s", cache_key, e)
-                cached_value = None
-
-            if cached_value is not None:
-                logger.debug("Cache HIT: %s", cache_key)
-                request.state.cache_status = "HIT"
-                return json.loads(cached_value)
-
-            logger.debug("Cache MISS: %s", cache_key)
-            request.state.cache_status = "MISS"
-
-            result = await func(*args, **kwargs)
-
-            try:
-                if hasattr(result, "model_dump"):
-                    serialized = json.dumps(result.model_dump(mode="json"))
-                elif isinstance(result, dict):
-                    serialized = json.dumps(result)
-                else:
-                    serialized = json.dumps(result, default=str)
-                cache.set(cache_key, serialized, ttl)
-            except Exception as e:
-                logger.warning("Cache store error for %s: %s", cache_key, e)
+            cached = _lookup(cache, cache_key, request)
+            if cached is not _MISS:
+                return cached
 
+            result = func(*args, **kwargs)
+            _store(cache, cache_key, result, ttl)
             return result
 
-        return wrapper
+        return sync_wrapper
 
     return decorator

src/meshcore_hub/api/routes/adoptions.py

@@ -18,7 +18,7 @@ router = APIRouter()
 
 
 @router.post("", response_model=AdoptedNodeRead, status_code=201)
-async def adopt_node(
+def adopt_node(
     adopt_request: NodeAdoptRequest,
     caller_info: RequireOperatorOrAdmin,
     session: DbSession,
@@ -74,7 +74,7 @@ async def adopt_node(
 
 
 @router.delete("/{public_key}", status_code=204)
-async def release_node(
+def release_node(
     public_key: str,
     caller_info: RequireOperatorOrAdmin,
     request: Request,

src/meshcore_hub/api/routes/advertisements.py

@@ -46,7 +46,7 @@ def _get_tag_description(node: Optional[Node]) -> Optional[str]:
 
 @router.get("", response_model=AdvertisementList)
 @cached("advertisements")
-async def list_advertisements(
+def list_advertisements(
     _: RequireRead,
     session: DbSession,
     request: Request,
@@ -243,7 +243,7 @@ async def list_advertisements(
 
 
 @router.get("/{advertisement_id}", response_model=AdvertisementRead)
-async def get_advertisement(
+def get_advertisement(
     _: RequireRead,
     session: DbSession,
     advertisement_id: str,

src/meshcore_hub/api/routes/channels.py

@@ -45,7 +45,7 @@ def _channel_to_read(channel: Channel, include_key: bool = False) -> ChannelRead
 
 @router.get("", response_model=ChannelList)
 @cached("channels", key_builder=_channels_key_builder)
-async def list_channels(
+def list_channels(
     _: RequireRead,
     session: DbSession,
     request: Request,
@@ -71,7 +71,7 @@ async def list_channels(
 
 
 @router.post("", response_model=ChannelRead, status_code=201)
-async def create_channel(
+def create_channel(
     __: RequireAdmin,
     session: DbSession,
     body: ChannelCreate,
@@ -110,7 +110,7 @@ async def create_channel(
 
 
 @router.put("/{channel_id}", response_model=ChannelRead)
-async def update_channel(
+def update_channel(
     __: RequireAdmin,
     session: DbSession,
     channel_id: str,
@@ -149,7 +149,7 @@ async def update_channel(
 
 
 @router.delete("/{channel_id}", status_code=204)
-async def delete_channel(
+def delete_channel(
     __: RequireAdmin,
     session: DbSession,
     channel_id: str,

src/meshcore_hub/api/routes/dashboard.py

@@ -65,7 +65,7 @@ def _flood_only_filter(
     ttl_setting="redis_cache_ttl_dashboard",
     key_builder=_dashboard_stats_key_builder,
 )
-async def get_stats(
+def get_stats(
     _: RequireRead,
     session: DbSession,
     request: Request,
@@ -323,7 +323,7 @@ async def get_stats(
 
 @router.get("/activity", response_model=DailyActivity)
 @cached("dashboard/activity", ttl_setting="redis_cache_ttl_dashboard")
-async def get_activity(
+def get_activity(
     _: RequireRead,
     session: DbSession,
     request: Request,
@@ -383,7 +383,7 @@ async def get_activity(
     ttl_setting="redis_cache_ttl_dashboard",
     key_builder=_dashboard_msg_activity_key_builder,
 )
-async def get_message_activity(
+def get_message_activity(
     _: RequireRead,
     session: DbSession,
     request: Request,
@@ -443,7 +443,7 @@ async def get_message_activity(
 
 @router.get("/node-count", response_model=NodeCountHistory)
 @cached("dashboard/node-count", ttl_setting="redis_cache_ttl_dashboard")
-async def get_node_count_history(
+def get_node_count_history(
     _: RequireRead,
     session: DbSession,
     request: Request,

src/meshcore_hub/api/routes/messages.py

@@ -41,7 +41,7 @@ def _get_tag_name(node: Optional[Node]) -> Optional[str]:
 
 @router.get("", response_model=MessageList)
 @cached("messages", key_builder=_messages_key_builder)
-async def list_messages(
+def list_messages(
     _: RequireRead,
     session: DbSession,
     request: Request,
@@ -233,7 +233,7 @@ async def list_messages(
 
 
 @router.get("/{message_id}", response_model=MessageRead)
-async def get_message(
+def get_message(
     _: RequireRead,
     session: DbSession,
     request: Request,

src/meshcore_hub/api/routes/node_tags.py

@@ -52,7 +52,7 @@ def _check_tag_access(
 
 
 @router.get("/nodes/{public_key}/tags", response_model=list[NodeTagRead])
-async def list_node_tags(
+def list_node_tags(
     _: RequireRead,
     session: DbSession,
     public_key: str,
@@ -68,7 +68,7 @@ async def list_node_tags(
 
 
 @router.post("/nodes/{public_key}/tags", response_model=NodeTagRead, status_code=201)
-async def create_node_tag(
+def create_node_tag(
     caller_info: RequireOperatorOrAdmin,
     session: DbSession,
     request: Request,
@@ -115,7 +115,7 @@ async def create_node_tag(
 
 
 @router.put("/nodes/{public_key}/tags/{key}", response_model=NodeTagRead)
-async def update_node_tag(
+def update_node_tag(
     caller_info: RequireOperatorOrAdmin,
     session: DbSession,
     request: Request,
@@ -171,7 +171,7 @@ async def update_node_tag(
 
 
 @router.delete("/nodes/{public_key}/tags/{key}", status_code=204)
-async def delete_node_tag(
+def delete_node_tag(
     caller_info: RequireOperatorOrAdmin,
     session: DbSession,
     request: Request,

src/meshcore_hub/api/routes/nodes.py

@@ -10,13 +10,8 @@ from meshcore_hub.api.auth import RequireRead
 from meshcore_hub.api.cache import cached
 from meshcore_hub.api.dependencies import DbSession
 from meshcore_hub.common.models import (
-    Advertisement,
-    EventObserver,
-    Message,
     Node,
     NodeTag,
-    Telemetry,
-    TracePath,
     UserProfileNode,
 )
 from meshcore_hub.common.schemas.nodes import AdoptedByUser, NodeList, NodeRead
@@ -49,7 +44,7 @@ VALID_NODE_SORT_COLUMNS = {"name", "public_key", "last_seen"}
 
 @router.get("", response_model=NodeList)
 @cached("nodes")
-async def list_nodes(
+def list_nodes(
     _: RequireRead,
     session: DbSession,
     request: Request,
@@ -149,56 +144,9 @@ async def list_nodes(
         )
 
     if observer is not None:
-        if observer:
-            query = query.where(
-                or_(
-                    Node.id.in_(
-                        select(Advertisement.observer_node_id).where(
-                            Advertisement.observer_node_id.is_not(None)
-                        )
-                    ),
-                    Node.id.in_(
-                        select(Message.observer_node_id).where(
-                            Message.observer_node_id.is_not(None)
-                        )
-                    ),
-                    Node.id.in_(
-                        select(Telemetry.observer_node_id).where(
-                            Telemetry.observer_node_id.is_not(None)
-                        )
-                    ),
-                    Node.id.in_(
-                        select(TracePath.observer_node_id).where(
-                            TracePath.observer_node_id.is_not(None)
-                        )
-                    ),
-                    Node.id.in_(select(EventObserver.observer_node_id)),
-                )
-            )
-        else:
-            query = query.where(
-                ~Node.id.in_(
-                    select(Advertisement.observer_node_id).where(
-                        Advertisement.observer_node_id.is_not(None)
-                    )
-                ),
-                ~Node.id.in_(
-                    select(Message.observer_node_id).where(
-                        Message.observer_node_id.is_not(None)
-                    )
-                ),
-                ~Node.id.in_(
-                    select(Telemetry.observer_node_id).where(
-                        Telemetry.observer_node_id.is_not(None)
-                    )
-                ),
-                ~Node.id.in_(
-                    select(TracePath.observer_node_id).where(
-                        TracePath.observer_node_id.is_not(None)
-                    )
-                ),
-                ~Node.id.in_(select(EventObserver.observer_node_id)),
-            )
+        # Uses the precomputed, indexed nodes.is_observer flag (maintained by the
+        # collector + cleanup job) instead of scanning the event tables.
+        query = query.where(Node.is_observer.is_(bool(observer)))
 
     # Get total count
     count_query = select(func.count()).select_from(query.subquery())
@@ -245,7 +193,7 @@ async def list_nodes(
 
 
 @router.get("/prefix/{prefix}", response_model=NodeRead)
-async def get_node_by_prefix(
+def get_node_by_prefix(
     _: RequireRead,
     session: DbSession,
     prefix: str = Path(description="Public key prefix to search for"),
@@ -275,7 +223,7 @@ async def get_node_by_prefix(
 
 
 @router.get("/{public_key}", response_model=NodeRead)
-async def get_node(
+def get_node(
     _: RequireRead,
     session: DbSession,
     public_key: str = Path(description="Full 64-character public key"),

src/meshcore_hub/api/routes/telemetry.py

@@ -17,7 +17,7 @@ router = APIRouter()
 
 
 @router.get("", response_model=TelemetryList)
-async def list_telemetry(
+def list_telemetry(
     _: RequireRead,
     session: DbSession,
     node_public_key: Optional[str] = Query(None, description="Filter by node"),
@@ -91,7 +91,7 @@ async def list_telemetry(
 
 
 @router.get("/{telemetry_id}", response_model=TelemetryRead)
-async def get_telemetry(
+def get_telemetry(
     _: RequireRead,
     session: DbSession,
     telemetry_id: str,

src/meshcore_hub/api/routes/trace_paths.py

@@ -17,7 +17,7 @@ router = APIRouter()
 
 
 @router.get("", response_model=TracePathList)
-async def list_trace_paths(
+def list_trace_paths(
     _: RequireRead,
     session: DbSession,
     observed_by: Optional[str] = Query(
@@ -91,7 +91,7 @@ async def list_trace_paths(
 
 
 @router.get("/{trace_path_id}", response_model=TracePathRead)
-async def get_trace_path(
+def get_trace_path(
     _: RequireRead,
     session: DbSession,
     trace_path_id: str,

src/meshcore_hub/api/routes/user_profiles.py

@@ -53,7 +53,7 @@ def _build_adopted_nodes(profile: UserProfile) -> list[AdoptedNodeRead]:
 
 @router.get("/profiles", response_model=UserProfileList)
 @cached("profiles")
-async def list_profiles(
+def list_profiles(
     _: RequireRead,
     session: DbSession,
     request: Request,
@@ -116,7 +116,7 @@ async def list_profiles(
 
 
 @router.get("/profile/me", response_model=UserProfileWithNodes)
-async def get_my_profile(
+def get_my_profile(
     request: Request,
     session: DbSession,
 ) -> UserProfileWithNodes:
@@ -148,7 +148,7 @@ async def get_my_profile(
 
 
 @router.get("/profile/{profile_id}")
-async def get_profile(
+def get_profile(
     profile_id: str,
     request: Request,
     session: DbSession,
@@ -201,7 +201,7 @@ async def get_profile(
 
 
 @router.put("/profile/{profile_id}", response_model=UserProfileRead)
-async def update_profile(
+def update_profile(
     profile_id: str,
     profile_update: UserProfileUpdate,
     caller_id: RequireUserOwner,

src/meshcore_hub/collector/cleanup.py

@@ -7,7 +7,7 @@ based on configured retention policies.
 import logging
 from datetime import datetime, timedelta, timezone
 
-from sqlalchemy import delete, func, select
+from sqlalchemy import CompoundSelect, delete, func, select, union, update
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from meshcore_hub.common.models import (
@@ -35,6 +35,7 @@ class CleanupStats:
         self.trace_paths_deleted: int = 0
         self.event_logs_deleted: int = 0
         self.nodes_deleted: int = 0
+        self.observers_cleared: int = 0
         self.total_deleted: int = 0
 
     def __repr__(self) -> str:
@@ -45,7 +46,8 @@ class CleanupStats:
             f"telemetry={self.telemetry_deleted}, "
             f"trace_paths={self.trace_paths_deleted}, "
             f"event_logs={self.event_logs_deleted}, "
-            f"nodes={self.nodes_deleted})"
+            f"nodes={self.nodes_deleted}, "
+            f"observers_cleared={self.observers_cleared})"
         )
 
 
@@ -107,6 +109,9 @@ async def cleanup_old_data(
         + stats.event_logs_deleted
     )
 
+    # Clear the is_observer flag for nodes whose events were all pruned above.
+    stats.observers_cleared = await recompute_observer_flags(db, dry_run)
+
     if not dry_run:
         await db.commit()
         logger.info("Cleanup completed: %s", stats)
@@ -116,6 +121,67 @@ async def cleanup_old_data(
     return stats
 
 
+def _observer_node_id_union() -> CompoundSelect:
+    """Union of observer_node_id across every event source (excluding NULLs)."""
+    return union(
+        select(Advertisement.observer_node_id).where(
+            Advertisement.observer_node_id.is_not(None)
+        ),
+        select(Message.observer_node_id).where(Message.observer_node_id.is_not(None)),
+        select(Telemetry.observer_node_id).where(
+            Telemetry.observer_node_id.is_not(None)
+        ),
+        select(TracePath.observer_node_id).where(
+            TracePath.observer_node_id.is_not(None)
+        ),
+        select(EventObserver.observer_node_id),
+    )
+
+
+async def recompute_observer_flags(db: AsyncSession, dry_run: bool = False) -> int:
+    """Clear nodes.is_observer for nodes that no longer have any events.
+
+    The collector sets the flag when a node observes an event; this clears it
+    again once retention has pruned all of a node's events, so stale observers
+    drop off the API's ``observer=true`` listing.
+
+    Args:
+        db: Database session
+        dry_run: If True, only count nodes without modifying them
+
+    Returns:
+        Number of nodes whose is_observer flag was (or would be) cleared.
+    """
+    observer_union = _observer_node_id_union().subquery()
+    stale = Node.id.not_in(select(observer_union.c.observer_node_id))
+
+    if dry_run:
+        count_stmt = (
+            select(func.count())
+            .select_from(Node)
+            .where(Node.is_observer.is_(True), stale)
+        )
+        result = await db.execute(count_stmt)
+        count = result.scalar() or 0
+    else:
+        upd = (
+            update(Node)
+            .where(Node.is_observer.is_(True), stale)
+            .values(is_observer=False)
+        )
+        result = await db.execute(upd)
+        count = result.rowcount or 0  # type: ignore[attr-defined]
+
+    if count > 0:
+        logger.info(
+            "is_observer: %s %d stale observer node(s)",
+            "would clear" if dry_run else "cleared",
+            count,
+        )
+
+    return count
+
+
 async def _cleanup_table(
     db: AsyncSession,
     model: type,

src/meshcore_hub/common/database.py

@@ -25,24 +25,42 @@ def create_database_engine(
         SQLAlchemy Engine instance
     """
     connect_args = {}
+    engine_kwargs: dict[str, Any] = {}
 
     # SQLite-specific configuration
     if database_url.startswith("sqlite"):
         connect_args["check_same_thread"] = False
 
+    # Size the pool above the default Starlette threadpool (~40 threads) so
+    # concurrent request handlers don't block waiting for a connection. Applies
+    # to file-based SQLite and networked backends (e.g. a future Postgres).
+    # In-memory SQLite uses a non-overflow pool, so skip these args there.
+    is_memory_sqlite = database_url in ("sqlite://", "sqlite:///:memory:")
+    if not is_memory_sqlite:
+        engine_kwargs["pool_size"] = 20
+        engine_kwargs["max_overflow"] = 30
+
     engine = create_engine(
         database_url,
         echo=echo,
         connect_args=connect_args,
         pool_pre_ping=True,
+        **engine_kwargs,
     )
 
-    # Enable foreign keys for SQLite
+    # Apply SQLite pragmas on every new connection
     if database_url.startswith("sqlite"):
 
         @event.listens_for(engine, "connect")
         def set_sqlite_pragma(dbapi_connection, connection_record):  # type: ignore
             cursor = dbapi_connection.cursor()
+            # WAL lets readers run concurrently with a single writer (the
+            # collector), and busy_timeout waits instead of immediately raising
+            # "database is locked" under contention. synchronous=NORMAL is safe
+            # under WAL and faster.
+            cursor.execute("PRAGMA journal_mode=WAL")
+            cursor.execute("PRAGMA busy_timeout=5000")
+            cursor.execute("PRAGMA synchronous=NORMAL")
             cursor.execute("PRAGMA foreign_keys=ON")
             cursor.close()
 
@@ -127,7 +145,8 @@ class DatabaseManager:
         async_url = self.database_url.replace("sqlite://", "sqlite+aiosqlite://")
         self._async_engine = create_async_engine(async_url, echo=self._echo)
 
-        # Enable foreign keys for async SQLite engine
+        # Apply the same SQLite pragmas as the sync engine (see
+        # create_database_engine) for the async engine's connections.
         if self.database_url.startswith("sqlite"):
 
             @event.listens_for(self._async_engine.sync_engine, "connect")
@@ -135,6 +154,9 @@ class DatabaseManager:
                 dbapi_connection: object, connection_record: object
             ) -> None:
                 cursor = dbapi_connection.cursor()  # type: ignore[attr-defined]
+                cursor.execute("PRAGMA journal_mode=WAL")
+                cursor.execute("PRAGMA busy_timeout=5000")
+                cursor.execute("PRAGMA synchronous=NORMAL")
                 cursor.execute("PRAGMA foreign_keys=ON")
                 cursor.close()
 

src/meshcore_hub/common/models/event_observer.py

@@ -12,6 +12,7 @@ from sqlalchemy import (
     Index,
     String,
     UniqueConstraint,
+    update,
 )
 from sqlalchemy.dialects.sqlite import insert as sqlite_insert
 from sqlalchemy.orm import Mapped, Session, mapped_column, relationship
@@ -138,4 +139,16 @@ def add_event_observer(
     )
     result = session.execute(stmt)
     rowcount = getattr(result, "rowcount", 0)
+
+    # Mark the observing node as an observer. Guarded on is_observer == False so
+    # this only writes on the first observation; the indexed flag lets the API's
+    # observer filter avoid scanning the (large) event tables.
+    from meshcore_hub.common.models.node import Node
+
+    session.execute(
+        update(Node)
+        .where(Node.id == observer_node_id, Node.is_observer.is_(False))
+        .values(is_observer=True)
+    )
+
     return bool(rowcount and rowcount > 0)

src/meshcore_hub/common/models/node.py

@@ -3,7 +3,7 @@
 from datetime import datetime
 from typing import TYPE_CHECKING, Optional
 
-from sqlalchemy import DateTime, Float, Index, Integer, String
+from sqlalchemy import Boolean, DateTime, Float, Index, Integer, String
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from meshcore_hub.common.models.base import Base, TimestampMixin, UUIDMixin, utc_now
@@ -26,6 +26,7 @@ class Node(Base, UUIDMixin, TimestampMixin):
         last_seen: Timestamp of most recent activity
         lat: GPS latitude coordinate (if available)
         lon: GPS longitude coordinate (if available)
+        is_observer: True if this node has observed at least one event
         created_at: Record creation timestamp
         updated_at: Record update timestamp
     """
@@ -74,6 +75,12 @@ class Node(Base, UUIDMixin, TimestampMixin):
         Float,
         nullable=True,
     )
+    is_observer: Mapped[bool] = mapped_column(
+        Boolean,
+        default=False,
+        server_default="0",
+        nullable=False,
+    )
 
     # Relationships
     tags: Mapped[list["NodeTag"]] = relationship(
@@ -92,6 +99,7 @@ class Node(Base, UUIDMixin, TimestampMixin):
     __table_args__ = (
         Index("ix_nodes_last_seen", "last_seen"),
         Index("ix_nodes_adv_type", "adv_type"),
+        Index("ix_nodes_is_observer", "is_observer"),
     )
 
     def __repr__(self) -> str:

src/meshcore_hub/common/schemas/nodes.py

@@ -116,6 +116,9 @@ class NodeRead(BaseModel):
     )
     lat: Optional[float] = Field(default=None, description="GPS latitude coordinate")
     lon: Optional[float] = Field(default=None, description="GPS longitude coordinate")
+    is_observer: bool = Field(
+        default=False, description="Whether this node has observed at least one event"
+    )
     created_at: datetime = Field(..., description="Record creation timestamp")
     updated_at: datetime = Field(..., description="Record update timestamp")
     tags: list[NodeTagRead] = Field(default_factory=list, description="Node tags")

tests/test_api/test_nodes.py

@@ -171,27 +171,10 @@ class TestListNodesFilters:
     def test_filter_by_observer_true(
         self, client_no_auth, api_db_session, receiver_node
     ):
-        """Test filtering nodes that have observed events."""
-        from datetime import datetime, timezone
-
-        from meshcore_hub.common.models import Advertisement, Message
-
-        # This node has observed an ad and a message
-        advert = Advertisement(
-            public_key="obsflt1obsflt1obsflt1obsflt1ob",
-            name="ObservedAd",
-            adv_type="CLIENT",
-            received_at=datetime.now(timezone.utc),
-            observer_node_id=receiver_node.id,
-        )
-        msg = Message(
-            message_type="channel",
-            channel_idx=1,
-            text="Observed msg",
-            received_at=datetime.now(timezone.utc),
-            observer_node_id=receiver_node.id,
-        )
-        api_db_session.add_all([advert, msg])
+        """Test filtering nodes by the precomputed is_observer flag."""
+        # The collector sets this flag when a node observes an event.
+        receiver_node.is_observer = True
+        api_db_session.add(receiver_node)
         api_db_session.commit()
 
         response = client_no_auth.get("/api/v1/nodes?observer=true")

tests/test_collector/test_cleanup.py

@@ -417,3 +417,52 @@ async def test_cleanup_orphaned_node_relations_dry_run(
         await async_db_session.scalar(select(func.count()).select_from(UserProfileNode))
         == 1
     )
+
+
+@pytest.mark.asyncio
+async def test_cleanup_clears_stale_observer_flags(
+    async_db_session: AsyncSession,
+) -> None:
+    """A node whose only events are pruned gets is_observer cleared; an active
+    observer keeps its flag."""
+    old_date = datetime.now(timezone.utc) - timedelta(days=60)
+    recent_date = datetime.now(timezone.utc) - timedelta(days=5)
+
+    stale = Node(public_key="s" * 64, name="Stale Observer", is_observer=True)
+    active = Node(public_key="b" * 64, name="Active Observer", is_observer=True)
+    async_db_session.add_all([stale, active])
+    await async_db_session.flush()
+
+    # Stale observer's only event is old and will be pruned
+    async_db_session.add(
+        Advertisement(
+            public_key=stale.public_key,
+            observer_node_id=stale.id,
+            created_at=old_date,
+            updated_at=old_date,
+        )
+    )
+    # Active observer has a recent event that survives cleanup
+    async_db_session.add(
+        Advertisement(
+            public_key=active.public_key,
+            observer_node_id=active.id,
+            created_at=recent_date,
+            updated_at=recent_date,
+        )
+    )
+    await async_db_session.commit()
+
+    stats = await cleanup_old_data(async_db_session, retention_days=30, dry_run=False)
+
+    assert stats.observers_cleared == 1
+
+    await async_db_session.rollback()  # Refresh from DB
+    stale_flag = await async_db_session.scalar(
+        select(Node.is_observer).where(Node.id == stale.id)
+    )
+    active_flag = await async_db_session.scalar(
+        select(Node.is_observer).where(Node.id == active.id)
+    )
+    assert stale_flag is False
+    assert active_flag is True

tests/test_common/test_database.py

@@ -0,0 +1,36 @@
+"""Tests for database engine configuration."""
+
+from pathlib import Path
+
+from sqlalchemy import text
+
+from meshcore_hub.common.database import create_database_engine
+
+
+class TestSqlitePragmas:
+    """Verify concurrency-related SQLite pragmas are applied on connect."""
+
+    def test_wal_and_busy_timeout_enabled(self, tmp_path: Path) -> None:
+        """File-based SQLite engines should run in WAL mode with a busy timeout."""
+        db_path = tmp_path / "pragma.db"
+        engine = create_database_engine(f"sqlite:///{db_path}")
+        try:
+            with engine.connect() as conn:
+                journal_mode = conn.execute(text("PRAGMA journal_mode")).scalar()
+                busy_timeout = conn.execute(text("PRAGMA busy_timeout")).scalar()
+                foreign_keys = conn.execute(text("PRAGMA foreign_keys")).scalar()
+
+            assert str(journal_mode).lower() == "wal"
+            assert busy_timeout is not None and int(busy_timeout) >= 5000
+            assert foreign_keys is not None and int(foreign_keys) == 1
+        finally:
+            engine.dispose()
+
+    def test_in_memory_engine_builds(self) -> None:
+        """In-memory SQLite must still build (no overflow-pool kwargs)."""
+        engine = create_database_engine("sqlite:///:memory:")
+        try:
+            with engine.connect() as conn:
+                assert conn.execute(text("SELECT 1")).scalar() == 1
+        finally:
+            engine.dispose()

tests/test_common/test_models.py

@@ -260,3 +260,21 @@ class TestEventObserverModel:
         observer = db_session.execute(select(EventObserver)).scalar_one()
         assert observer.path_len is None
         assert observer.snr is None
+
+    def test_add_event_observer_sets_is_observer_flag(self, db_session) -> None:
+        """Observing an event marks the observer node with is_observer=True."""
+        node = Node(public_key="c" * 64, name="Observer3")
+        db_session.add(node)
+        db_session.commit()
+        assert node.is_observer is False
+
+        add_event_observer(
+            session=db_session,
+            event_type="advertisement",
+            event_hash="deadbeef",
+            observer_node_id=node.id,
+        )
+        db_session.commit()
+
+        db_session.refresh(node)
+        assert node.is_observer is True