diff --git a/CHANGELOG.md b/CHANGELOG.md index 8cf93ff3a..b0c255829 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/docs/database/migrations.md b/docs/database/migrations.md index 1cd95f353..8602d6aa9 100644 --- a/docs/database/migrations.md +++ b/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` | diff --git a/docs/modules/background_tasks.md b/docs/modules/background_tasks.md index d0e833e8a..bb489ba57 100644 --- a/docs/modules/background_tasks.md +++ b/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`) diff --git a/docs/perf/2026-11-05-background-task-reads.md b/docs/perf/2026-11-05-background-task-reads.md new file mode 100644 index 000000000..ef47f3496 --- /dev/null +++ b/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. diff --git a/host/migrations/versions/c5d8e2f1a903_background_tasks_queued_order.py b/host/migrations/versions/c5d8e2f1a903_background_tasks_queued_order.py new file mode 100644 index 000000000..e77712982 --- /dev/null +++ b/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) diff --git a/host/tests/test_migration_graph.py b/host/tests/test_migration_graph.py index 454001765..46da61036 100644 --- a/host/tests/test_migration_graph.py +++ b/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 } diff --git a/modules/background_tasks/background_tasks/endpoints/api_status.py b/modules/background_tasks/background_tasks/endpoints/api_status.py index c2b3da859..0afec8c9d 100644 --- a/modules/background_tasks/background_tasks/endpoints/api_status.py +++ b/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() diff --git a/modules/background_tasks/background_tasks/endpoints/views.py b/modules/background_tasks/background_tasks/endpoints/views.py index 716988261..6dc7a079c 100644 --- a/modules/background_tasks/background_tasks/endpoints/views.py +++ b/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.""" diff --git a/modules/background_tasks/background_tasks/models.py b/modules/background_tasks/background_tasks/models.py index 882262999..d30dc69b0 100644 --- a/modules/background_tasks/background_tasks/models.py +++ b/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). diff --git a/modules/background_tasks/background_tasks/service.py b/modules/background_tasks/background_tasks/service.py index a0946ed7f..2b3302910 100644 --- a/modules/background_tasks/background_tasks/service.py +++ b/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, *, diff --git a/modules/background_tasks/tests/test_background_tasks_order_index.py b/modules/background_tasks/tests/test_background_tasks_order_index.py new file mode 100644 index 000000000..f1d85af49 --- /dev/null +++ b/modules/background_tasks/tests/test_background_tasks_order_index.py @@ -0,0 +1,90 @@ +"""Default task order is index-backed, including NULL timestamps.""" + +from io import StringIO +from pathlib import Path + +import sqlalchemy as sa +from alembic.config import Config +from alembic.migration import MigrationContext +from alembic.operations import Operations +from alembic.script import ScriptDirectory +from background_tasks.models import TaskExecution +from sqlalchemy.dialects import postgresql +from sqlalchemy.schema import CreateIndex + +INDEX = "ix_background_tasks_task_execution_queued_order" +REVISION = "c5d8e2f1a903" + + +def _script() -> ScriptDirectory: + root = Path(__file__).resolve().parents[3] + return ScriptDirectory.from_config(Config(str(root / "host/alembic.ini"))) + + +def _model_index(): + return next(i for i in TaskExecution.__table__.indexes if i.name == INDEX) + + +def test_default_order_uses_index_without_sort(): + engine = sa.create_engine("sqlite://") + try: + TaskExecution.__table__.create(engine) + with engine.connect() as conn: + plan = conn.exec_driver_sql( + "EXPLAIN QUERY PLAN SELECT id FROM background_tasks_task_execution " + "ORDER BY queued_at IS NULL, queued_at DESC LIMIT 20" + ).all() + detail = " ".join(str(row) for row in plan) + assert INDEX in detail + assert "TEMP B-TREE" not in detail + finally: + engine.dispose() + + +def test_order_index_migration_roundtrip(): + script = _script() + revision = script.get_revision(REVISION) + # Pin membership, not tip: later background_tasks revisions must not break this. + assert REVISION in { + r.revision for r in script.walk_revisions(base="base", head="background_tasks@head") + } + engine = sa.create_engine("sqlite://") + try: + with engine.begin() as conn: + conn.exec_driver_sql( + "CREATE TABLE background_tasks_task_execution (id TEXT, queued_at DATETIME)" + ) + with Operations.context(MigrationContext.configure(conn)): + revision.module.upgrade() + definition = conn.exec_driver_sql( + "SELECT sql FROM sqlite_master WHERE name = ?", (INDEX,) + ).scalar_one() + model_index = _model_index() + assert definition == str(CreateIndex(model_index).compile(conn)).strip() + revision.module.downgrade() + assert ( + conn.exec_driver_sql( + "SELECT count(*) FROM sqlite_master WHERE name = ?", (INDEX,) + ).scalar_one() + == 0 + ) + revision.module.upgrade() + finally: + engine.dispose() + + +def test_order_index_postgres_ddl(): + revision = _script().get_revision(REVISION) + output = StringIO() + context = MigrationContext.configure( + dialect_name="postgresql", opts={"as_sql": True, "output_buffer": output} + ) + with Operations.context(context): + revision.module.upgrade() + expected = ( + f"CREATE INDEX {INDEX} ON background_tasks_task_execution " + "((queued_at IS NULL), queued_at DESC)" + ) + assert output.getvalue().strip() == expected + ";" + model_index = _model_index() + assert str(CreateIndex(model_index).compile(dialect=postgresql.dialect())) == expected diff --git a/modules/background_tasks/tests/test_background_tasks_read_projection.py b/modules/background_tasks/tests/test_background_tasks_read_projection.py new file mode 100644 index 000000000..dc201073e --- /dev/null +++ b/modules/background_tasks/tests/test_background_tasks_read_projection.py @@ -0,0 +1,107 @@ +"""Task reads must not decode discarded payloads or window-count wide history.""" + +from contextlib import contextmanager +from datetime import UTC, datetime, timedelta +from unittest.mock import MagicMock + +import pytest +from background_tasks.constants import TaskStatus +from background_tasks.models import TaskExecution +from background_tasks.service import BackgroundTaskService +from simple_module_core.events import EventBus +from sqlalchemy import event + + +@contextmanager +def _capture_task_selects(sync_engine): + statements = [] + + def capture(conn, cursor, statement, parameters, context, executemany): + if ( + statement.lstrip().upper().startswith("SELECT") + and "background_tasks_task_execution" in statement + ): + statements.append(statement) + + event.listen(sync_engine, "before_cursor_execute", capture) + try: + yield statements + finally: + event.remove(sync_engine, "before_cursor_execute", capture) + + +@pytest.fixture +def task_selects(engine): + with _capture_task_selects(engine.sync_engine) as statements: + yield statements + + +async def test_status_does_not_select_payload(seed_execution, authenticated_client, app): + row = await seed_execution(result={"data": "x" * 1_000_000}, traceback="secret") + with _capture_task_selects(app.state.sm.db.engine.sync_engine) as statements: + response = await authenticated_client.get( + f"/api/background_tasks/tasks/{row.celery_task_id}/status" + ) + assert response.status_code == 200 + assert len(statements) == 1 + projection = statements[0].split("FROM")[0] + for field in ("args", "kwargs", "result", "traceback"): + assert f"background_tasks_task_execution.{field}" not in projection + + +@pytest.mark.parametrize( + "page,expected_page,expected_args", + [(1, 1, [0, 1]), (2, 2, [2, 3]), (99, 3, [4])], +) +async def test_list_projects_payload_and_counts_separately( + db_session, task_selects, page, expected_page, expected_args +): + now = datetime.now(UTC) + rows = [ + TaskExecution( + task_name="demo.task", + status=TaskStatus.FAILED, + tenant_id="acme", + queued_at=now - timedelta(seconds=i) if i < 4 else None, + args=[i], + kwargs={"value": i}, + result={"data": "x" * 10000}, + traceback="secret", + ) + for i in range(5) + ] + db_session.add_all(rows) + db_session.add(TaskExecution(task_name="other", status=TaskStatus.SUCCESS)) + await db_session.flush() + db_session.expunge_all() + task_selects.clear() + service = BackgroundTaskService(db_session, MagicMock(), EventBus()) + result = await service.list( + status=TaskStatus.FAILED, + task_name="demo", + tenant_id="acme", + queue="default", + page=page, + per_page=2, + ) + assert (result.total, result.page, result.per_page) == (5, expected_page, 2) + assert [(item.args, item.kwargs) for item in result.items] == [ + ([i], {"value": i}) for i in expected_args + ] + assert len(task_selects) == 2 + count, page = task_selects + assert "count(" in count.lower() and "OVER" not in count + assert "ORDER BY" not in count + projection = page.split("FROM")[0] + assert "count(" not in projection.lower() + for field in ("result", "traceback"): + assert f"background_tasks_task_execution.{field}" not in projection + + +@pytest.mark.parametrize("page", [0, 99]) +async def test_empty_list_preserves_page_contract(db_session, page): + service = BackgroundTaskService(db_session, MagicMock(), EventBus()) + result = await service.list(page=page) + assert result.items == [] + assert result.total == 0 + assert result.page == max(1, page) diff --git a/modules/background_tasks/tests/test_bg_newest_first_status.py b/modules/background_tasks/tests/test_bg_newest_first_status.py new file mode 100644 index 000000000..466d4d48c --- /dev/null +++ b/modules/background_tasks/tests/test_bg_newest_first_status.py @@ -0,0 +1,53 @@ +"""Newest-first ordering is one clause, shared by the list and the status lookup.""" + +from __future__ import annotations + +from datetime import UTC, datetime + +from background_tasks.constants import TaskStatus +from background_tasks.models import TaskExecution +from background_tasks.service import BackgroundTaskService +from simple_module_core.tenancy import TenantRole + +URL = "/api/background_tasks/tasks/{}/status" + + +async def test_null_queued_at_does_not_win_the_status_lookup(app, seed_execution, tenant_client): + async with tenant_client(TenantRole.MEMBER) as me: + dated = await seed_execution( + status=TaskStatus.RUNNING, + tenant_id=me.tenant_id, + owner_user_id=me.user_id, + progress_message="dated", + queued_at=datetime(2026, 5, 1, 12, 0, tzinfo=UTC), + ) + async with app.state.sm.db.session_factory() as session: + session.add( + TaskExecution( + celery_task_id=dated.celery_task_id, + task_name="demo.task", + status=TaskStatus.PENDING, + args=[], + kwargs={}, + tenant_id=me.tenant_id, + owner_user_id=me.user_id, + progress_message="undated", + queued_at=None, + ) + ) + await session.commit() + + resp = await me.client.get(URL.format(dated.celery_task_id)) + + assert resp.status_code == 200, resp.text + assert resp.json()["progress_message"] == "dated" + + +async def test_has_any_probe(app, seed_execution): + async with app.state.sm.db.session_factory() as session: + service = BackgroundTaskService(session, None, None) + assert await service.has_any() is False + await seed_execution() + async with app.state.sm.db.session_factory() as session: + service = BackgroundTaskService(session, None, None) + assert await service.has_any() is True