feat(background_tasks): stamp executions with tenant, retry as original tenant (#371) [stack 8/11]

docs/framework/multi-tenancy.md

@@ -164,9 +164,16 @@ bound may name one. Beat tasks have no request: wrap cross-tenant work in
 The worker never builds the app, so `background_tasks.sync_db` attaches the
 same listeners to its own session class and reads `multi_tenant` from the host
 settings (`scripts/run_worker.py`): task bodies get the same fail-closed rules
-as request code. A process that talks to the DB some other way must do the
+as request code. It also passes `default_tenant`, so an unbound insert in a
+task body (an `all_tenants()` block) lands in that tenant exactly as it would
+in the web process. A process that talks to the DB some other way must do the
 same — `attach_session_listeners(MySession)` plus
-`bind_engine_policy(engine, EngineTenancy(tenant_strict=...))`.
+`bind_engine_policy(engine, EngineTenancy(tenant_strict=..., default_tenant_id=...))`.
+
+The worker-side signals (prerun, success, failure, retry, revoked) stamp the
+`TaskExecution` row with the tenant on the message header themselves — the
+publish signal may never have written the row — and a signal whose message
+carries no tenant never blanks a row already stamped.
 
 ## Resolution
 

host/migrations/versions/b5d3f08a6e17_background_tasks_execution_tenant_id.py

@@ -0,0 +1,43 @@
+"""background_tasks_task_execution: add nullable tenant_id
+
+NULL means a beat / platform publish (no tenant bound), which is also what
+every pre-existing row is. The table is deliberately not ``MultiTenantMixin``:
+those rows have no tenant and strict mode would raise on their insert. See
+GH #371.
+
+background_tasks has no migration chain of its own (its tables came in with
+the host's initial schema), so this extends the mainline head.
+
+Revision ID: b5d3f08a6e17
+Revises: d4e8a1b6c392
+Create Date: 2026-10-01 13:00:00.000000
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "b5d3f08a6e17"
+down_revision: str | None = "d4e8a1b6c392"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+_TABLE = "background_tasks_task_execution"
+_SINGLE = "ix_background_tasks_task_execution_tenant_id"
+_COMPOSITE = "ix_background_tasks_task_execution_tenant_status_queued"
+
+
+def upgrade() -> None:
+    with op.batch_alter_table(_TABLE) as batch:
+        batch.add_column(sa.Column("tenant_id", sa.String(length=50), nullable=True))
+        batch.create_index(_SINGLE, ["tenant_id"], unique=False)
+        batch.create_index(_COMPOSITE, ["tenant_id", "status", "queued_at"], unique=False)
+
+
+def downgrade() -> None:
+    with op.batch_alter_table(_TABLE) as batch:
+        batch.drop_index(_COMPOSITE)
+        batch.drop_index(_SINGLE)
+        batch.drop_column("tenant_id")

modules/background_tasks/background_tasks/_signal_bus.py

@@ -0,0 +1,61 @@
+"""Bridge from Celery's sync signal thread to the app's async event bus.
+
+Split out of :mod:`.signals` to keep it under the 300-line cap; ``signals``
+re-exports :func:`bind_event_bus` / :func:`unbind_event_bus`.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any
+
+from simple_module_core.events import EventBus
+
+logger = logging.getLogger("background_tasks.signals")
+
+
+_bus: EventBus | None = None
+_loop: asyncio.AbstractEventLoop | None = None
+
+
+def bind_event_bus(bus: EventBus, loop: asyncio.AbstractEventLoop) -> None:
+    """Bind an event bus + its running loop so signals can publish events.
+
+    Signals fire on the Celery sync thread; `run_coroutine_threadsafe`
+    bridges back to ``loop`` so handlers run on the API event loop
+    regardless of which thread triggered the signal.
+    """
+    global _bus, _loop
+    _bus = bus
+    _loop = loop
+
+
+def unbind_event_bus() -> None:
+    """Drop the bound bus — called from ``on_shutdown`` so tests stay isolated."""
+    global _bus, _loop
+    _bus = None
+    _loop = None
+
+
+def publish_from_signal(event: Any) -> None:
+    """Dispatch ``event`` onto the bound bus without blocking the signal thread."""
+    if _bus is None or _loop is None:
+        return
+    try:
+        future = asyncio.run_coroutine_threadsafe(_bus.publish(event), _loop)
+    except RuntimeError:
+        # Loop has stopped (shutdown race). The DB row is already written.
+        logger.debug("Event bus loop is not running; skipping %s", type(event).__name__)
+        return
+    # Surface subscriber exceptions — run_coroutine_threadsafe otherwise only
+    # logs them when the Future is GC'd, which happens far from the failure.
+    future.add_done_callback(_log_publish_failure)
+
+
+def _log_publish_failure(future: asyncio.Future[Any]) -> None:
+    if future.cancelled():
+        return
+    exc = future.exception()
+    if exc is not None:
+        logger.error("Event publish raised: %s", exc, exc_info=exc)

modules/background_tasks/background_tasks/constants.py

@@ -102,3 +102,7 @@ class TaskStatus(StrEnum):
 # Harmless round-trip task; only wired into CI smoke tests and local dev —
 # not scheduled, not invoked by other modules.
 DEMO_ECHO_TASK = "background_tasks.demo_echo"
+
+TENANT_ID_MAX_LENGTH = 50
+# Filter value meaning "executions with no tenant" (beat / platform publishes).
+PLATFORM_TENANT_FILTER = "__platform__"

modules/background_tasks/background_tasks/contracts/schemas.py

@@ -28,6 +28,7 @@ class TaskExecutionListItem(SQLModel):
     task_name: str
     status: TaskStatus
     queue: str
+    tenant_id: str | None = None
     args: list[Any] = []
     kwargs: dict[str, Any] = {}
     retries: int
@@ -49,6 +50,7 @@ class TaskExecutionDetail(SQLModel):
     task_name: str
     status: TaskStatus
     queue: str
+    tenant_id: str | None = None
     args: list[Any] = []
     kwargs: dict[str, Any] = {}
     result: dict[str, Any] | None = None

modules/background_tasks/background_tasks/endpoints/api_admin.py

@@ -37,6 +37,7 @@ async def list_executions(
     q: str | None = Query(default=None),
     task_name: str | None = Query(default=None, deprecated=True),
     queue: str | None = Query(default=None),
+    tenant_id: str | None = Query(default=None),
     page: int = Query(default=1, ge=1),
     per_page: int = Query(default=20, ge=1, le=200),
     service: BackgroundTaskService = Depends(get_background_task_service),
@@ -50,7 +51,12 @@ async def list_executions(
     existing API callers keep working, and ``q`` wins when both are given.
     """
     return await service.list(
-        status=status, task_name=q or task_name, queue=queue, page=page, per_page=per_page
+        status=status,
+        task_name=q or task_name,
+        queue=queue,
+        tenant_id=tenant_id,
+        page=page,
+        per_page=per_page,
     )
 
 
@@ -74,15 +80,18 @@ async def retry_failed_executions(
     status: TaskStatus | None = Query(default=None),
     task_name: str | None = Query(default=None, alias="q"),
     queue: str | None = Query(default=None),
+    tenant_id: str | None = Query(default=None),
     service: BackgroundTaskService = Depends(get_background_task_service),
 ) -> RetryFailedResult:
     """Re-enqueue every failed or stuck execution the current filter can see.
 
-    Takes the same three filters as the listing — status, search, queue — so
+    Takes the same filters as the listing — status, search, queue, tenant — so
     the sweep covers exactly the rows the operator is looking at and nothing
     else. Capped per call; the response says how many eligible rows are left.
     """
-    return await service.retry_failed(status=status, task_name=task_name, queue=queue)
+    return await service.retry_failed(
+        status=status, task_name=task_name, queue=queue, tenant_id=tenant_id
+    )
 
 
 @router.post(

modules/background_tasks/background_tasks/endpoints/views.py

@@ -12,6 +12,7 @@
 
 from background_tasks.constants import (
     PERM_VIEW,
+    PLATFORM_TENANT_FILTER,
     TaskStatus,
 )
 from background_tasks.deps import get_background_task_service
@@ -53,29 +54,36 @@ async def index(
     status: TaskStatus | None = Query(default=None),
     task_name: str = Query(default="", alias="q"),
     queue: str = Query(default=""),
+    tenant_id: str = Query(default="", alias="tenant"),
     page: int = Query(default=1),
     service: BackgroundTaskService = Depends(get_background_task_service),
 ) -> InertiaResponse:
     response = await service.list(
         status=status,
         task_name=task_name or None,
         queue=queue or None,
+        tenant_id=tenant_id or None,
         page=page,
         per_page=PER_PAGE,
     )
     # A filtered-empty list says nothing about the fleet, so don't pay to poll
     # it: the screen already knows to blame the filter.
-    unfiltered_and_empty = response.total == 0 and not task_name and not queue and status is None
+    unfiltered_and_empty = (
+        response.total == 0 and not task_name and not queue and not tenant_id and status is None
+    )
 
     async def strip_data() -> tuple[dict[str, int], list[str]]:
         """Everything above the table. Serial: one session, one connection."""
-        counts = await service.status_counts(task_name=task_name or None, queue=queue or None)
+        counts = await service.status_counts(
+            task_name=task_name or None, queue=queue or None, tenant_id=tenant_id or None
+        )
         # The strip's third tile is a windowed throughput reading rather than a
         # status total, so it needs its own count — see `success_count_since`.
         counts["success_24h"] = await service.success_count_since(
             hours=SUCCESS_WINDOW_HOURS,
             task_name=task_name or None,
             queue=queue or None,
+            tenant_id=tenant_id or None,
         )
         # Deliberately unfiltered: the dropdown is how an operator leaves the
         # queue they are in, so it keeps offering the ones they are not in.
@@ -97,6 +105,8 @@ async def strip_data() -> tuple[dict[str, int], list[str]]:
                 "success_24h": counts["success_24h"],
             },
             "queues": queues,
+            "tenant_ids": await service.tenant_ids(),
+            "platform_tenant_value": PLATFORM_TENANT_FILTER,
             "worker_presence": presence,
             "pagination": {
                 "page": response.page,
@@ -107,6 +117,7 @@ async def strip_data() -> tuple[dict[str, int], list[str]]:
                 "status": status.value if status else "",
                 "task_name": task_name,
                 "queue": queue,
+                "tenant": tenant_id,
             },
         },
     )

modules/background_tasks/background_tasks/filters.py

@@ -11,7 +11,7 @@
 from simple_module_db import LIKE_ESCAPE_CHAR, like_contains_pattern
 from sqlalchemy import ColumnElement, select
 
-from background_tasks.constants import RETRYABLE_STATUSES, TaskStatus
+from background_tasks.constants import PLATFORM_TENANT_FILTER, RETRYABLE_STATUSES, TaskStatus
 from background_tasks.models import TaskExecution
 
 Conditions = list[ColumnElement[bool]]
@@ -22,9 +22,19 @@ def execution_filters(
     status: TaskStatus | None = None,
     task_name: str | None = None,
     queue: str | None = None,
+    tenant_id: str | None = None,
 ) -> Conditions:
-    """The three axes the executions screen filters on."""
+    """The axes the executions screen filters on.
+
+    *tenant_id* is a tenant id, or :data:`PLATFORM_TENANT_FILTER` for
+    executions published with no tenant. The screen is platform-wide; tenant is
+    a filter, not a scope.
+    """
     conditions: Conditions = []
+    if tenant_id == PLATFORM_TENANT_FILTER:
+        conditions.append(TaskExecution.tenant_id.is_(None))
+    elif tenant_id:
+        conditions.append(TaskExecution.tenant_id == tenant_id)
     if status is not None:
         conditions.append(TaskExecution.status == status)
     if task_name:
@@ -41,6 +51,7 @@ def bulk_retry_conditions(
     status: TaskStatus | None = None,
     task_name: str | None = None,
     queue: str | None = None,
+    tenant_id: str | None = None,
 ) -> Conditions | None:
     """What the bulk sweep may touch, or ``None`` when it may touch nothing.
 
@@ -64,6 +75,6 @@ def bulk_retry_conditions(
     already_retried = select(child.c.id).where(child.c.retried_from_id == TaskExecution.id).exists()
     return [
         TaskExecution.status.in_(sorted(wanted)),
-        *execution_filters(task_name=task_name, queue=queue),
+        *execution_filters(task_name=task_name, queue=queue, tenant_id=tenant_id),
         ~already_retried,
     ]

modules/background_tasks/background_tasks/locales/en.json

@@ -26,7 +26,11 @@
     "stuck": "Stuck",
     "queue_label": "Queue:",
     "queue_all": "all",
-    "queue_aria": "Filter by queue"
+    "queue_aria": "Filter by queue",
+    "tenant_label": "Tenant:",
+    "tenant_all": "all",
+    "tenant_platform": "Platform",
+    "tenant_aria": "Filter by tenant"
   },
   "status": {
     "queued": "queued",
@@ -44,7 +48,8 @@
     "queued_at": "Queued",
     "duration": "Duration",
     "actions": "Actions",
-    "retry": "Retry"
+    "retry": "Retry",
+    "tenant": "Tenant"
   },
   "detail": {
     "head_title": "Task detail",
@@ -69,7 +74,8 @@
     "traceback": "Traceback",
     "copy": "Copy",
     "copied": "Copied",
-    "no_traceback": "No traceback recorded."
+    "no_traceback": "No traceback recorded.",
+    "tenant": "Tenant"
   },
   "retry_dialog": {
     "title": "Retry {name}?",

modules/background_tasks/background_tasks/models.py

@@ -15,6 +15,7 @@
     DEFAULT_QUEUE,
     MODULE_NAME,
     TABLE_TASK_EXECUTION,
+    TENANT_ID_MAX_LENGTH,
     TaskStatus,
 )
 
@@ -48,6 +49,12 @@ class TaskExecution(Base, AuditMixin, table=True):  # ty: ignore[unsupported-bas
     )
     queue: str = Field(default=DEFAULT_QUEUE, max_length=64)
 
+    # The tenant the task was published for; NULL for beat and other platform
+    # publishes. Deliberately not MultiTenantMixin: those rows have no tenant
+    # and strict mode would raise on their insert. The admin screens are
+    # platform-wide and filter on this column.
+    tenant_id: str | None = Field(default=None, index=True, max_length=TENANT_ID_MAX_LENGTH)
+
     args: list[Any] = Field(default_factory=list, sa_column=Column(JSON))
     kwargs: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON))
     result: dict[str, Any] | None = Field(default=None, sa_column=Column(JSON))
@@ -79,4 +86,10 @@ class TaskExecution(Base, AuditMixin, table=True):  # ty: ignore[unsupported-bas
             "status",
             "queued_at",
         ),
+        Index(
+            "ix_background_tasks_task_execution_tenant_status_queued",
+            "tenant_id",
+            "status",
+            "queued_at",
+        ),
     )

modules/background_tasks/background_tasks/module.py

@@ -161,8 +161,9 @@ async def on_startup(self, app: FastAPI) -> None:
         # SQLite default and silently drop ``TaskExecution`` rows.
         host = app.state.sm.settings
         multi = bool(getattr(host, "multi_tenant", False))
-        set_database_url(host.database_url, tenant_strict=multi)
-        set_default_tenant(None if multi else getattr(host, "default_tenant", "") or None)
+        default_tenant = None if multi else getattr(host, "default_tenant", "") or None
+        set_database_url(host.database_url, tenant_strict=multi, default_tenant=default_tenant)
+        set_default_tenant(default_tenant)
         # build_celery imports `signals` for side effects and runs
         # `autodiscover_tasks` across every installed module.
         services.celery = build_celery(services.settings)