perf(background_tasks): project task reads, separate exact count, index the default order

CHANGELOG.md

@@ -67,6 +67,8 @@ All notable changes to this project are documented in this file. The format is b
 
 ### Changed
 
+- Reduce background task polling and list reads to their required columns; count task history separately from paginated reads and index newest-first, nulls-last ordering (migration `c5d8e2f1a903`, 2026-11-05 performance review).
+- Scaffold new modules with `core/` and `web/` areas and mirrored nested Python tests; existing modules migrate individually. See `docs/module-layout.md`.
 - `OidcAuthProvider` and `KeycloakAuthProvider` gain a public `sign_in_from_claims(request, claims)`; the OIDC and Keycloak login callbacks call it instead of the private `_identity`/`_upsert_user_cache`/`_claims_to_user_context` helpers (#848). The OIDC variant returns `(UserContext, identity)` and raises the new `IdentityUnavailableError` when the user cache refuses the identity. No behaviour change.
 - Scaffold new modules with `core/` and `web/` areas and mirrored nested Python tests; existing modules migrate individually. See `docs/module-layout.md`. Moved modules log under their new paths (for example `mail.delivery.sender`), so logging config that names an old logger needs the new name.
 - `mail` follows the module layout: `mail.core.{constants,settings,state,data_compatibility}` and `mail.delivery.{sender,runtime}`. `mail.settings` and `mail.constants` still import, with a `DeprecationWarning`, until 0.0.46.

docs/database/migrations.md

@@ -71,7 +71,7 @@ The history is **not** a single linear chain. It has several roots and several h
 
 | Head | Owning module | Label Alembic reports |
 |---|---|---|
-| `e805a1b2c3d4` | users (browser session retirement) | `api_tokens`, `background_tasks`, `file_storage`, `users` |
+| `c5d8e2f1a903` | background_tasks (newest-first listing index) | `api_tokens`, `background_tasks`, `file_storage`, `users` |
 | `b8d4f2a6c901` | audit_log | `audit_log` |
 | `f1a9c3d7b2e4` | tenants | `tenants` |
 | `d263ea727663` | framework (`sm_lease`) | `leases` |

docs/modules/background_tasks.md

@@ -505,6 +505,24 @@ from background_tasks.contracts.schemas import (
 | `TaskExecutionDetail` | Full record incl. `args`, `kwargs`, `result`, `traceback`. |
 | `WorkerInfo` / `WorkerSnapshot` | Live worker probe response shape. |
 
+## Read performance
+
+Status polling selects the progress/status DTO fields plus owner, tenant and
+resource identifiers needed for authorization. It does not fetch or decode
+arguments, results or tracebacks. The admin list keeps `args` and `kwargs` for
+retry confirmation, but excludes results and tracebacks; the detail view still
+loads the complete execution.
+
+Pagination uses a separate exact count with the same filters, then clamps a
+stale nonempty page to the last page before fetching it. Counting still scales
+with matching history; page reads can stop at the requested limit. Migration
+`c5d8e2f1a903` on `background_tasks@head` adds
+`ix_background_tasks_task_execution_queued_order` on
+`((queued_at IS NULL), queued_at DESC)`. The list orders by those expressions,
+which preserve newest-first, nulls-last ordering on SQLite and PostgreSQL.
+Expression indexes must be maintained by hand when reviewing autogenerated
+migrations on SQLite. Apply `make migrate` before relying on this index.
+
 ## Models
 
 `TaskExecution` (table `background_tasks_task_execution`)

docs/perf/2026-11-05-background-task-reads.md

@@ -0,0 +1,87 @@
+# Background task read performance
+
+Measured on 2026-11-05 for the local review of `main@4920a13a`.
+
+## Method
+
+Local SQLite 3.53.1, Python 3.12, synchronous SQLAlchemy sessions against a
+throwaway file database. No app server, worker, browser or external service.
+This measures SQL execution, ORM materialization and DTO validation, excluding
+HTTP handling and network latency. These are SQLite results, not PostgreSQL
+estimates.
+
+The list dataset contains 100,000 successful executions, each with one integer
+argument, one integer keyword argument, a JSON result containing 1,000 ASCII
+characters, and a 180-character traceback. Timestamps increase by one second;
+every twentieth timestamp is null. Existing model indexes remain present.
+Baseline runs omit the new ordering index; revised runs include it.
+
+The baseline list runs the original full-entity select with `count(*) OVER ()`,
+`ORDER BY queued_at DESC NULLS LAST` and `LIMIT 20`. The revised list runs an
+exact `count(*)` followed by a DTO-column-only select with
+`ORDER BY queued_at IS NULL, queued_at DESC LIMIT 20`. Both validate all returned
+list DTOs. Assertions check equal first-page IDs, 20 rows and a 100,000 total.
+Timings are medians of five runs after one warmup, with a fresh session each run.
+
+For polling, one row's result is replaced with `{"data": "x" * 1_000_000}`
+(1,000,012 bytes encoded as JSON). Both selects use the endpoint's
+`celery_task_id` predicate, descending queued timestamp and `LIMIT 1`. The revised
+select loads DTO fields plus authorization fields, with other columns set to
+raise on deferred access. An assertion checks identical status DTOs. Timings
+use five runs after warmup; memory uses the median peak from 15 fresh
+`tracemalloc` runs, collecting garbage before each run. Memory measures Python
+allocations during the complete read and DTO conversion, not native SQLite
+buffers or HTTP response size.
+
+## Results
+
+| Measurement | Before | After |
+|---|---:|---:|
+| First list page, including exact total | 1,418.00 ms | 3.54 ms |
+| Poll with 1 MB result | 3.10 ms | 0.48 ms |
+| Peak traced Python memory per poll | 2,024,547 bytes | 23,428 bytes |
+| Selected raw values per poll, UTF-8 bytes | 1,000,309 | 49 |
+
+The raw-value byte sum encodes each non-null SQLite driver value as text and
+adds its byte length. It excludes driver/protocol framing and is not a wire-byte
+measurement. The status response itself is unchanged. The entire 1,000,012-byte
+JSON result, as well as arguments and traceback, is excluded from the new poll.
+
+`EXPLAIN QUERY PLAN` before:
+
+```text
+CO-ROUTINE (subquery-2)
+SCAN background_tasks_task_execution
+SCAN (subquery-2)
+USE TEMP B-TREE FOR ORDER BY
+```
+
+After, count:
+
+```text
+SCAN background_tasks_task_execution USING COVERING INDEX ix_background_tasks_task_execution_queued_order
+```
+
+After, page:
+
+```text
+SCAN background_tasks_task_execution USING INDEX ix_background_tasks_task_execution_queued_order
+```
+
+## Limits and PostgreSQL follow-up
+
+PostgreSQL `EXPLAIN (ANALYZE, BUFFERS)` was not run for this review: no
+PostgreSQL instance was available. The migration's
+PostgreSQL DDL is checked offline; execution and planner behavior there remain
+unverified. No million-row run was made. SQLite's large improvement includes
+avoiding its wide-row window spool and sort; do not extrapolate its ratio to
+PostgreSQL.
+
+On a disposable PostgreSQL database with 100,000–1,000,000 representative
+executions, compare the original full-row window query with the separate count
+and projected page, using `EXPLAIN (ANALYZE, BUFFERS)` for each. The page order
+must be `queued_at IS NULL, queued_at DESC` to match the portable expression
+index in migration `c5d8e2f1a903`. Test filtered and deep pages too. Exact counts
+still visit matching history, and offset pagination still visits skipped index
+entries. Count and page are separate statements, so concurrent writes can make
+them reflect different snapshots under PostgreSQL's default READ COMMITTED.

host/migrations/versions/c5d8e2f1a903_background_tasks_queued_order.py

@@ -0,0 +1,25 @@
+"""Index the default newest-first task listing, with NULL timestamps last.
+
+Revision ID: c5d8e2f1a903
+Revises: e805a1b2c3d4
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "c5d8e2f1a903"
+down_revision = "e805a1b2c3d4"
+branch_labels = None
+depends_on = None
+
+_TABLE = "background_tasks_task_execution"
+_INDEX = "ix_background_tasks_task_execution_queued_order"
+
+
+def upgrade() -> None:
+    # Hand-written: SQLite autogenerate cannot reflect expression indexes.
+    op.create_index(_INDEX, _TABLE, [sa.text("(queued_at IS NULL)"), sa.text("queued_at DESC")])
+
+
+def downgrade() -> None:
+    op.drop_index(_INDEX, table_name=_TABLE)

host/tests/test_migration_graph.py

@@ -43,6 +43,7 @@
     "1f5985daa10b": "background_tasks",  # retained log records (#537)
     "a625f0b7c912": "users",  # refresh-token families (#625)
     "a626b1c8d023": "users",  # hashed bearer tokens (#626)
+    "c5d8e2f1a903": "background_tasks",  # newest-first listing index
     "e731a1b2c3d4": "users",  # unique lower(email) (#731)
     "e805a1b2c3d4": "users",  # browser session retirement
     "51b2058d8afd": "api_tokens",
@@ -71,7 +72,7 @@
 # head revision -> module owning the tables it creates or alters.
 _HEADS = {
     "b8d4f2a6c901": "audit_log",
-    "e805a1b2c3d4": "users",  # also carries ``api_tokens``, ``background_tasks``, ``file_storage``
+    "c5d8e2f1a903": "background_tasks",  # also carries ``api_tokens``, ``file_storage``, ``users``
     "f1a9c3d7b2e4": "tenants",
     "f6a3b9d2e8c4": "keycloak",
     "c9d2e4f6a810": "oidc",
@@ -120,6 +121,7 @@
     "f5b1c3e8a207": "api_tokens",
     "e731a1b2c3d4": "api_tokens",
     "e805a1b2c3d4": "api_tokens",
+    "c5d8e2f1a903": "api_tokens",
     "a626b1c8d023": "api_tokens",  # also inherits ``background_tasks``
     "51b2058d8afd": "api_tokens",  # ancestors of a labelled revision inherit its label too
 }

modules/background_tasks/background_tasks/endpoints/api_status.py

@@ -25,7 +25,7 @@
 
 from background_tasks.constants import MODULE_NAME, PERM_VIEW, STATUS_ROUTER_PREFIX
 from background_tasks.contracts.schemas import TaskStatusDTO
-from background_tasks.models import TaskExecution
+from background_tasks.models import TaskExecution, dto_columns, newest_first
 
 router = APIRouter(prefix=STATUS_ROUTER_PREFIX, tags=[f"{MODULE_NAME}-status"])
 
@@ -73,8 +73,13 @@ async def get_task_status(
     row = (
         await db.execute(
             select(TaskExecution)
+            .options(
+                dto_columns(
+                    TaskStatusDTO, "owner_user_id", "tenant_id", "resource_type", "resource_id"
+                )
+            )
             .where(col(TaskExecution.celery_task_id) == celery_task_id)
-            .order_by(TaskExecution.queued_at.desc())
+            .order_by(*newest_first(TaskExecution.queued_at))
             .limit(1)
         )
     ).scalar_one_or_none()

modules/background_tasks/background_tasks/endpoints/views.py

@@ -91,7 +91,7 @@ async def index(
     # from "nothing exists at all".
     has_executions = response.total > 0
     if not has_executions and not unfiltered_and_empty:
-        has_executions = (await service.list(page=1, per_page=1)).total > 0
+        has_executions = await service.has_any()
 
     async def strip_data() -> tuple[dict[str, int], list[str]]:
         """Everything above the table. Serial: one session, one connection."""

modules/background_tasks/background_tasks/models.py

@@ -9,8 +9,9 @@
 from simple_module_db.base import create_module_base
 from simple_module_db.mixins import AuditMixin
 from simple_module_db.types import AwareDateTime
-from sqlalchemy import JSON, CheckConstraint, Column, Index, String
-from sqlmodel import Field
+from sqlalchemy import JSON, CheckConstraint, Column, Index, String, column
+from sqlalchemy.orm import load_only
+from sqlmodel import Field, SQLModel
 
 from background_tasks.constants import (
     DEFAULT_QUEUE,
@@ -35,6 +36,16 @@
 Base = create_module_base(MODULE_NAME)
 
 
+def newest_first(queued_at: Any) -> tuple[Any, Any]:
+    """Portable ``queued_at DESC NULLS LAST``, as the two expressions to ``ORDER BY``.
+
+    Postgres sorts NULL first under plain ``DESC`` and SQLite last, so the null
+    flag leads. The model's expression index is built from the same function,
+    so the planner can serve every newest-first read from it.
+    """
+    return queued_at.is_(None).self_group(), queued_at.desc()
+
+
 class TaskExecution(Base, AuditMixin, table=True):  # ty: ignore[unsupported-base]
     """Persistent record of a single Celery task execution.
 
@@ -124,6 +135,11 @@ class TaskExecution(Base, AuditMixin, table=True):  # ty: ignore[unsupported-bas
     heartbeat_at: datetime | None = None
 
     __table_args__ = (
+        # Portable DESC NULLS LAST: NULL timestamps sort after dated tasks.
+        Index(
+            "ix_background_tasks_task_execution_queued_order",
+            *newest_first(column("queued_at")),
+        ),
         CheckConstraint(
             "(resource_type IS NULL) = (resource_id IS NULL)",
             name="task_resource_pair",
@@ -142,6 +158,14 @@ class TaskExecution(Base, AuditMixin, table=True):  # ty: ignore[unsupported-bas
     )
 
 
+def dto_columns(dto: type[SQLModel], *extra: str) -> Any:
+    """``load_only`` option loading just ``dto``'s fields (plus ``extra``); other columns raise."""
+    return load_only(
+        *(getattr(TaskExecution, field) for field in (*dto.model_fields, *extra)),
+        raiseload=True,
+    )
+
+
 class OutboxMessage(Base, table=True):  # ty: ignore[unsupported-base]
     """One durable event awaiting delivery to one durable handler (GH #459).
 

modules/background_tasks/background_tasks/service.py

@@ -3,14 +3,14 @@
 from __future__ import annotations
 
 import uuid
-from collections.abc import Sequence
 from datetime import UTC, datetime, timedelta
 from typing import TYPE_CHECKING
 
 from fastapi import HTTPException, status
 from simple_module_core.events import EventBus
 from simple_module_db import RequestSession
-from sqlalchemy import Row, func, select
+from simple_module_db.pagination import total_pages
+from sqlalchemy import func, select
 
 from background_tasks.constants import RETRYABLE_STATUSES, TaskStatus
 from background_tasks.contracts.schemas import (
@@ -20,7 +20,7 @@
     TaskExecutionListResponse,
 )
 from background_tasks.filters import execution_filters
-from background_tasks.models import TaskExecution
+from background_tasks.models import TaskExecution, dto_columns, newest_first
 from background_tasks.retry_service import RetryCoordinator
 
 if TYPE_CHECKING:
@@ -70,36 +70,22 @@ async def list(
             status=status, task_name=task_name, queue=queue, tenant_id=tenant_id
         )
 
-        # A window-function total lets us fetch the page and the count in a
-        # single round trip. SQLAlchemy's ``AsyncSession`` serialises calls on
-        # one connection, so ``asyncio.gather`` wouldn't help here.
-        async def fetch(p: int) -> tuple[Sequence[Row[tuple[TaskExecution, int]]], int]:
-            total_col = func.count().over().label("_total")
-            query = (
-                select(TaskExecution, total_col)
-                .where(*conditions)
-                .order_by(TaskExecution.queued_at.desc().nulls_last())
-                .offset((p - 1) * per_page)
-                .limit(per_page)
-            )
-            rows = (await self.db.execute(query)).all()
-            return rows, (int(rows[0][1]) if rows else 0)
-
-        rows, total = await fetch(page)
-        if not rows and page > 1:
-            # The window-function total only rides along on a row that's
-            # actually returned, so a page past the end (stale bookmark, a
-            # manually-edited ?page=) can't self-report the real total — it
-            # looks identical to "there are no rows at all". Count for real,
-            # clamp to the last valid page, and refetch so callers never see
-            # a false "nothing exists" for a query that has rows elsewhere.
-            count_query = select(func.count()).select_from(TaskExecution).where(*conditions)
-            real_total = int((await self.db.execute(count_query)).scalar_one())
-            if real_total:
-                page = max(1, -(-real_total // per_page))  # ceil division
-                rows, total = await fetch(page)
-
-        items = [TaskExecutionListItem.model_validate(row[0]) for row in rows]
+        count_query = select(func.count()).select_from(TaskExecution).where(*conditions)
+        total = int((await self.db.execute(count_query)).scalar_one())
+        if total:
+            page = min(page, total_pages(total, per_page))
+
+        query = (
+            select(TaskExecution)
+            .options(dto_columns(TaskExecutionListItem))
+            .where(*conditions)
+            # The expression index preserves DESC NULLS LAST on both DBs.
+            .order_by(*newest_first(TaskExecution.queued_at))
+            .offset((page - 1) * per_page)
+            .limit(per_page)
+        )
+        rows = (await self.db.execute(query)).scalars().all()
+        items = [TaskExecutionListItem.model_validate(row) for row in rows]
 
         return TaskExecutionListResponse(
             items=items,
@@ -112,6 +98,11 @@ async def fetch(p: int) -> tuple[Sequence[Row[tuple[TaskExecution, int]]], int]:
             tenant_id=tenant_id,
         )
 
+    async def has_any(self) -> bool:
+        """Whether any execution exists at all: an index-friendly probe, no COUNT."""
+        probe = select(TaskExecution.id).limit(1)
+        return (await self.db.execute(probe)).first() is not None
+
     async def status_counts(
         self,
         *,