Public module APIs: in_delivery, forget_resource, render_error_page, worker_tenancy (#682 #683 #684 #685)

CHANGELOG.md

@@ -14,6 +14,16 @@ All notable changes to this project are documented in this file. The format is b
 
 ## [Unreleased]
 
+### Added
+- Public durable delivery context (`in_delivery()` and `IN_DELIVERY`), inherited
+  by handler child tasks and reset after delivery (#682).
+- Transactional, tenant-scoped `sharing.contracts.forget_resource` for resource
+  owners purging shared resources (#683).
+- Public `simple_module_hosting.render_error_page` with standard JSON/Inertia
+  negotiation, status and header preservation (#684).
+- Public `background_tasks.sync_db.worker_tenancy` and `WorkerTenancy` snapshot,
+  distinguishing an unconfigured worker from single-tenant operation (#685).
+
 ### Fixed
 - File storage: cancellation during upload-session lookup prevents transport from
   starting; clearing completed upload rows releases their retained files (#695, #588).

docs/framework/events.md

@@ -177,6 +177,14 @@ await bus.publish_durable(OrderPlaced(order_id=order.id, ...), db)
   publisher).
 - A durable handler may itself `publish_durable(..., ctx.db)`; the new event
   commits with the handler's delivery.
+- Public `simple_module_core.durable_events.in_delivery()` reports whether the
+  current context is running a durable handler (#682). The outbox transport sets
+  the public `IN_DELIVERY: ContextVar[bool]` around the handler and resets it in
+  `finally`, including after failure or cancellation. Child tasks inherit `True`
+  when created inside a handler; unrelated tasks remain `False`. A child retains
+  its inherited value even if it outlives the handler. Use this to skip awaiting
+  an eager outbox drain from delivery code, which would wait on itself. Modules
+  no longer need stack inspection or pins on private dispatcher functions.
 - The handler `name=` (default `f"{handler.__module__}.{handler.__qualname__}"`)
   keys delivery state, so keep it stable across deploys.
 - Events must be `@dataclass` classes serialisable to JSON; `UUID`, `datetime`,

docs/module-authoring.md

@@ -99,6 +99,26 @@ existing service. Translate messages at the request boundary. Endpoint-level
 `HTTPException` remains appropriate for pure request validation. Inertia forms
 may catch domain errors to report field errors or redirect instead.
 
+For a custom exception handler that needs the standard error response, import
+the public `render_error_page` from `simple_module_hosting` (#684):
+
+```python
+from simple_module_hosting import render_error_page
+
+
+async def handle_module_error(request, exc):
+    return await render_error_page(request, exc.status_code, detail=str(exc))
+```
+
+Its stable signature is
+`render_error_page(request, status_code, detail, headers=None, *, required_permission=None)`.
+It uses the framework's content negotiation: API clients receive
+`{"detail": detail}`, while browser navigations and Inertia visits receive the
+standard `Error` page with shared layout props. Both retain the status and
+supplied headers (for example, `Retry-After`). `required_permission` optionally
+names a missing permission on a 403 page. Rendering failures fall back to JSON.
+Modules need no imports from private hosting modules.
+
 ### Minimal `pyproject.toml`
 
 ```toml

docs/modules/background_tasks.md

@@ -202,6 +202,18 @@ A publisher that never imported a task can still enqueue it by name: `enqueue_af
 
 ## Worker tuning
 
+Task bodies can inspect the host's configured tenancy with the public
+`background_tasks.sync_db.worker_tenancy() -> WorkerTenancy | None` (#685).
+`WorkerTenancy` is an immutable SQLModel snapshot with `multi_tenant: bool` and
+`default_tenant: str | None`. It returns `None` until `set_database_url` pins a
+URL, including when a URL is only available from the environment. A configured
+single-tenant worker returns `multi_tenant=False`; a strict worker returns
+`multi_tenant=True` and `default_tenant=None` (the default is ignored in strict
+mode). In single-tenant mode, a null default means the framework's default
+tenant id. Reconfiguration updates subsequent snapshots; `set_database_url(None)`
+or `dispose_sync_engine()` returns the accessor to `None`. Treat unknown tenancy
+as unconfigured rather than assuming single-tenant operation.
+
 Each field defaults to `None`, which keeps Celery's own default; all require a worker restart.
 
 | Field | Celery key |

docs/modules/sharing.md

@@ -83,6 +83,36 @@ row.
 
 Audit entries link to `/admin/sharing/?share={id}` (`AuditLink.for_model`).
 
+## Cleanup
+
+Durable `UserDeleted` cleanup removes shares addressed to the deleted user
+across tenants; durable `TenantDeleted` cleanup removes the deleted tenant's
+shares. Neither detects deletion of an arbitrary shared resource.
+
+Resource owners must call the public
+`sharing.contracts.forget_resource(db, resource_type, resource_id)` when purging
+a resource (#683). Pass one canonical text id (UUID hex, decimal integer, or the
+resource's natural key), or an iterable of those ids for a batch. The function
+deletes every matching share, including expired shares, under the caller's
+bound tenant using the caller's transaction. Other tenants and resource types
+are unaffected. Repeated calls and empty batches are safe.
+
+```python
+from sharing.contracts import forget_resource
+
+with tenant_context(tenant_id):
+    await forget_resource(db, NOTE_TYPE, note.id.hex)
+    await db.delete(note)
+    await db.flush()
+```
+
+This is a trusted cleanup API for the owning module: it performs no authorization
+or resource-existence check and does not commit. Bind the resource's tenant;
+strict tenancy fails closed without one. Declare `Sharing` in `depends_on` when
+using this contract. Clean up in the same transaction as the purge so a rollback
+restores both the resource and its shares. Existing orphan rows are not swept
+automatically.
+
 ## Migration
 
 `f733c7938fd1` creates the table. It is its own root and head

framework/core/simple_module_core/durable_events.py

@@ -21,6 +21,7 @@
 
 import logging
 from collections.abc import Awaitable, Callable, Sequence
+from contextvars import ContextVar
 from dataclasses import dataclass
 from functools import cache
 from typing import Any, Protocol, runtime_checkable
@@ -33,6 +34,14 @@
 # subscribe time so an over-long name fails at boot, not on a publisher's flush.
 MAX_NAME_LENGTH = 255
 
+# Set by delivery transports only while invoking a durable handler.
+IN_DELIVERY: ContextVar[bool] = ContextVar("durable_event_in_delivery", default=False)
+
+
+def in_delivery() -> bool:
+    """Whether this context runs in a durable handler (including child tasks)."""
+    return IN_DELIVERY.get()
+
 
 class DurableDeliveryUnavailable(RuntimeError):  # noqa: N818 - name fixed by the design
     """``publish_durable`` was called but no durable transport is installed.
@@ -177,6 +186,7 @@ def __len__(self) -> int:
 
 
 __all__ = [
+    "IN_DELIVERY",
     "DurableContext",
     "DurableDeliveryUnavailable",
     "DurableHandler",
@@ -186,5 +196,6 @@ def __len__(self) -> int:
     "deserialize_event",
     "event_key",
     "handler_name",
+    "in_delivery",
     "serialize_event",
 ]

framework/core/simple_module_core/events.py

@@ -11,6 +11,7 @@
 from pyee.asyncio import AsyncIOEventEmitter
 
 from simple_module_core.durable_events import (
+    IN_DELIVERY,
     DurableContext,
     DurableDeliveryUnavailable,
     DurableHandler,
@@ -160,4 +161,9 @@ async def publish_durable_or_inline(self, event: Event, db: Any) -> None:
             subscription = self.durable.lookup(key, name)
             if subscription is None:
                 continue
-            await subscription.handler(event, ctx)
+            # Inline delivery is still a durable delivery for ``in_delivery()`` (#682).
+            token = IN_DELIVERY.set(True)
+            try:
+                await subscription.handler(event, ctx)
+            finally:
+                IN_DELIVERY.reset(token)

framework/core/tests/test_core_inline_delivery_context.py

@@ -0,0 +1,19 @@
+"""Inline durable dispatch marks its handlers as in-delivery (#682)."""
+
+from simple_module_core.durable_events import in_delivery
+from simple_module_core.events import EventBus
+from simple_module_core.tenant_events import TenantDeleted
+
+
+async def test_inline_durable_handler_sees_in_delivery():
+    bus = EventBus()
+    seen: list[bool] = []
+
+    async def handler(event, ctx):
+        seen.append(in_delivery())
+
+    bus.subscribe_durable(TenantDeleted, handler, name="probe")
+    assert in_delivery() is False
+    await bus.publish_durable_or_inline(TenantDeleted("id", "slug", "Name"), object())
+    assert seen == [True]
+    assert in_delivery() is False

framework/hosting/simple_module_hosting/__init__.py

@@ -2,6 +2,7 @@
 
 from simple_module_hosting._preapp_config import merge_host_settings
 from simple_module_hosting.app_builder import create_app
+from simple_module_hosting.errors import render_error_page
 from simple_module_hosting.logging import correlation_id, setup_logging
 from simple_module_hosting.settings import Settings
 from simple_module_hosting.shared_props import (
@@ -16,5 +17,6 @@
     "create_app",
     "merge_host_settings",
     "register_inertia_shared_provider",
+    "render_error_page",
     "setup_logging",
 ]

framework/hosting/simple_module_hosting/errors.py

@@ -0,0 +1,28 @@
+"""Public error rendering for module exception handlers (#684)."""
+
+from collections.abc import Mapping
+
+from starlette.requests import Request
+from starlette.responses import JSONResponse, Response
+
+from simple_module_hosting._error_negotiation import _wants_json
+from simple_module_hosting._error_page import render_error_page as _render_error_page
+
+
+async def render_error_page(
+    request: Request,
+    status_code: int,
+    detail: str,
+    headers: Mapping[str, str] | None = None,
+    *,
+    required_permission: str | None = None,
+) -> Response:
+    """Return standard JSON for API clients or the Inertia Error page for browsers.
+
+    Uses the framework's content negotiation, shared layout props and rendering
+    fallback. Status and supplied headers survive either response format.
+    ``required_permission`` optionally identifies the permission a 403 lacked.
+    """
+    if _wants_json(request):
+        return JSONResponse(status_code=status_code, content={"detail": detail}, headers=headers)
+    return await _render_error_page(request, status_code, detail, headers, required_permission)

framework/hosting/tests/test_hosting_public_error_page.py

@@ -0,0 +1,36 @@
+"""Modules can use the public renderer with standard content negotiation (#684)."""
+
+import pytest
+from fastapi import Request
+
+
+@pytest.mark.parametrize(
+    "headers, json_response",
+    [
+        ({"Accept": "application/json"}, True),
+        ({"Accept": "text/html"}, False),
+        ({"Accept": "application/json", "X-Inertia": "true"}, False),
+    ],
+)
+async def test_public_renderer_negotiates_and_preserves_status_and_headers(
+    app, authenticated_client, headers, json_response
+):
+    from simple_module_hosting import render_error_page
+
+    async def custom_error(request: Request):
+        return await render_error_page(
+            request, 429, detail="Try later", headers={"Retry-After": "30"}
+        )
+
+    app.add_api_route("/custom-error", custom_error)
+    response = await authenticated_client.get("/custom-error", headers=headers)
+    assert response.status_code == 429
+    assert response.headers["Retry-After"] == "30"
+    if json_response:
+        assert response.json() == {"detail": "Try later"}
+    elif "X-Inertia" in headers:
+        assert response.json()["component"] == "Error"
+        assert response.json()["props"]["message"] == "Try later"
+    else:
+        assert "data-page" in response.text
+        assert "Try later" in response.text

modules/background_tasks/background_tasks/outbox_dispatch.py

@@ -41,7 +41,7 @@
 from datetime import datetime, timedelta
 from typing import Any
 
-from simple_module_core.durable_events import DurableContext, deserialize_event
+from simple_module_core.durable_events import IN_DELIVERY, DurableContext, deserialize_event
 from simple_module_core.events import EventBus
 from simple_module_db import is_valid_tenant_id
 from simple_module_db.callbacks import discard_on_commit_callbacks, run_on_commit_callbacks
@@ -166,7 +166,11 @@ async def deliver_one(
                     async with deadline:
                         with handler_guard(session):
                             ctx = DurableContext(db=session, attempt=row.attempts)
-                            await sub.handler(event, ctx)
+                            token = IN_DELIVERY.set(True)
+                            try:
+                                await sub.handler(event, ctx)
+                            finally:
+                                IN_DELIVERY.reset(token)
                 except TimeoutError as exc:
                     if not deadline.expired():
                         raise  # the handler's own timeout, not ours

modules/background_tasks/background_tasks/sync_db.py

@@ -28,6 +28,7 @@
 from sqlalchemy import create_engine
 from sqlalchemy.engine import Engine
 from sqlalchemy.orm import Session, sessionmaker
+from sqlmodel import SQLModel
 
 logger = logging.getLogger(__name__)
 
@@ -43,6 +44,20 @@
 _default_tenant: str | None = None
 
 
+class WorkerTenancy(SQLModel, frozen=True):
+    """Snapshot of the host tenancy configured for this worker (#685)."""
+
+    multi_tenant: bool
+    default_tenant: str | None
+
+
+def worker_tenancy() -> WorkerTenancy | None:
+    """Return configured tenancy, or ``None`` when no database URL is pinned."""
+    if _url_override is None:
+        return None
+    return WorkerTenancy(multi_tenant=_tenant_strict, default_tenant=_default_tenant)
+
+
 class WorkerSession(Session):
     """This engine's own session class, so the entity listeners attached to it
     never double up with a ``DatabaseState``'s (each state has its own

modules/background_tasks/tests/test_background_tasks_delivery_context.py

@@ -0,0 +1,72 @@
+"""Public durable delivery context follows tasks, not process-global state (#682)."""
+
+import asyncio
+from dataclasses import dataclass
+
+import pytest
+from background_tasks.outbox_claim import OutboxPolicy
+from background_tasks.outbox_dispatch import dispatch
+from simple_module_core import durable_events
+from simple_module_core.events import Event
+
+pytestmark = pytest.mark.usefixtures("_stub_celery")
+
+
+@dataclass
+class ContextProbe(Event):
+    pass
+
+
+@pytest.mark.parametrize("outcome", ["success", "failure", "cancel"])
+async def test_delivery_context_is_inherited_isolated_and_reset(app, outcome):
+    in_delivery = durable_events.in_delivery
+    assert not in_delivery()
+    entered = asyncio.Event()
+    release = asyncio.Event()
+    seen = []
+
+    async def child():
+        seen.append(in_delivery())
+
+    async def handler(event, ctx):
+        seen.append(in_delivery())
+        await asyncio.create_task(child())
+        entered.set()
+        await release.wait()
+        if outcome == "failure":
+            raise ValueError("delivery failed")
+
+    bus = app.state.sm.event_bus
+    bus.subscribe_durable(ContextProbe, handler, name="context-probe")
+    factory = app.state.sm.db.session_factory
+    async with factory() as db:
+        await bus.publish_durable(ContextProbe(), db)
+        await db.commit()
+
+    async def run():
+        try:
+            return await dispatch(factory, bus, OutboxPolicy(max_attempts=8, lease_seconds=300))
+        finally:
+            seen.append(in_delivery())
+
+    task = asyncio.create_task(run())
+    try:
+        await asyncio.wait_for(entered.wait(), timeout=5)
+        assert not in_delivery()  # unrelated task while the handler is still running
+        if outcome == "cancel":
+            task.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await task
+        else:
+            release.set()
+            result = await asyncio.wait_for(task, timeout=5)
+            assert (result.delivered, result.retrying) == (
+                (1, 0) if outcome == "success" else (0, 1)
+            )
+        assert seen == [True, True, False]
+        assert not in_delivery()
+    finally:
+        release.set()
+        if not task.done():
+            task.cancel()
+            await asyncio.gather(task, return_exceptions=True)