diff --git a/CHANGELOG.md b/CHANGELOG.md index 73f38d2ea..d93b2063c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,11 +10,32 @@ All notable changes to this project are documented in this file. The format is b > Versions not listed below still have no entry; consult the commit log. > > 0.0.28 through 0.0.37 were released before the release workflow cut this -> file (#572), so their changes are still listed under `[Unreleased]` and will -> land in the next dated section (#584). +> file (#572), so their changes are listed under `[0.0.38]` (#584). ## [Unreleased] +### Added +- `EventBus.publish_durable_or_inline(event, db)`: enqueue durably when a transport is + installed, otherwise run the durable handlers inline in `db`'s transaction (#583). + +### Fixed +- Tenants: deleting a tenant now tears down its data. `TenantDeleted` (now defined in + `simple_module_core.tenant_events`, still importable from `tenants.contracts.events`) is + published durably, and file_storage (files, upload sessions, then backend objects after + commit), api_tokens, sharing, feature_flags and the framework's tenant-scoped settings + overrides (including branding) purge that tenant's rows; audit history is kept. On multi-tenant + installs, public files of a deleted tenant answer 404 at once (the existence check is cached + 60 s per worker and evicted on every worker by `TENANT_DELETED_CHANNEL`). Without `background_tasks` the handlers run inline in + the deleting transaction. Rows orphaned by earlier deletions are not purged retroactively (#583). +- File storage: a file whose `backend` is not the active backend is no longer read or deleted + through the wrong store: downloads, public reads and thumbnails answer 409 with a translated + message, and a delete soft-deletes the row but leaves the object alone (logged). Switching + backends still needs a manual migration; see `docs/modules/file_storage.md` (#589). +- Background tasks: `enqueue_after_commit` is documented as at-most-once (a broker outage after + commit loses the task; use `publish_durable` for must-run work). file_storage gains an hourly + `file_storage.reconcile_checksums` beat task that re-enqueues checksums for files left without + one for 30 minutes, up to 500 per run (#590). + ## [0.0.38] — 2026-10-23 ### Fixed diff --git a/docs/framework/multi-tenancy.md b/docs/framework/multi-tenancy.md index 1fb17edfd..67f673a37 100644 --- a/docs/framework/multi-tenancy.md +++ b/docs/framework/multi-tenancy.md @@ -345,6 +345,48 @@ filter. A platform admin's actions are attributed to their **active** organisation when one is active — the write itself is scoped to it — and to the platform only when nothing is bound. See [audit_log](/modules/audit_log#multi-tenancy). +## Deleting tenants + +Tenant deletion publishes `TenantDeleted` through +`bus.publish_durable_or_inline` in the same transaction that removes the +tenant, memberships and invitations. With the `background_tasks` outbox +transport installed, delivery is durable and eventual (retried until it +succeeds). Without it, the durable handlers run inline in the deleting +transaction, so their row purges commit or roll back with the tenant row; +backend objects are then deleted after commit by an `on_commit` callback, +which is best-effort: a failure there is logged with the object keys, and +the hourly `file_storage.reap_objects` task removes the orphaned objects once +`background_tasks` is enabled. Either way, public file downloads check the +tenant directory (cached for 60 s per worker, forgotten on `TenantDeleted`) +and return 404 when the tenant no longer exists. + +Every owner of tenant data must register +`bus.subscribe_durable(TenantDeleted, handler)` in `register_event_handlers`, +including when `app=None` in the worker. Import the event from +`simple_module_core.tenant_events`; no dependency on the `tenants` plugin is +needed. Delete only `event.tenant_id`'s rows, using `tenant_context` for +`MultiTenantMixin` tables, and hard-delete soft-deleted rows too. Explicit +scope tables need matching scope and tenant predicates. Handlers must tolerate +repeated deliveries and must never commit `ctx.db`; the outbox owns that +transaction. Backend object deletion belongs after commit and must also be +idempotent. File storage writes a second durable cleanup event containing the +backend keys in the same transaction that hard-deletes its rows. Only after +that transaction commits does the second handler remove originals, thumbnail +variants and unfinished multipart uploads; backend failures retry from the +persisted manifest rather than losing the keys. + +The framework registers cleanup of tenant-scoped `Setting` rows in both host +and worker, even if the settings UI plugin is disabled. This includes branding +overrides; branding's uploaded files belong to file storage cleanup. System +and user settings and other tenants' rows are preserved. Feature flags also +purges its tenant-scoped persisted overrides; system overrides survive. + +Audit entries are deliberately retained for historical accountability. Their +original `tenant_id` remains the historical tenant marker even after the live +tenant row disappears; readers must not require a live tenant join to display +these entries. Deletion does not reassign audit entries to another tenant or +to the platform, and audit retention policy still applies independently. + ## Testing The `simple_module_test` plugin ships `tenant_client` (needs the `users` and diff --git a/docs/modules/background_tasks.md b/docs/modules/background_tasks.md index 1b8add88b..2d464a985 100644 --- a/docs/modules/background_tasks.md +++ b/docs/modules/background_tasks.md @@ -54,6 +54,20 @@ async def place(self, data: OrderCreate) -> OrderOut: - In a request (`get_db`), the publish runs from the session's `on_commit` hook — after the commit, before the response leaves. - Any other session — a task body's sync session, a CLI script's own `async with session.begin()` — hangs it off SQLAlchemy's `after_commit` event instead, and drops it when the outermost transaction ends without committing. Publishing then blocks the committing thread for the length of one kombu publish. - The task runs as the tenant bound when you called `enqueue_after_commit`, not whatever is bound at commit time. + +**Delivery is at-most-once.** The publish intent lives only in memory. A broker +failure after commit is logged as `background_tasks.after_commit.publish_failed` +without retry; a process exit between commit and publish also loses the job. +For must-run work, use the event bus's `publish_durable(event, db)` with a durable +handler (see durable events below), which persists outbox intent in the same +transaction. This helper itself provides no execution or redelivery guarantee. + +File storage recovers missing checksum jobs with the hourly +`file_storage.reconcile_checksums` beat task: it re-enqueues up to 500 completed, +non-deleted files whose checksum is null and whose creation was more than 30 +minutes ago, across tenants. Failed publishes remain eligible on subsequent +runs. Run beat and a worker for this recovery; duplicate checksum jobs are safe. + - A publish that fails after the commit is logged, not raised: the transaction is already durable. The admin retry buttons follow the same rule — the new `pending` row commits first, then the task is sent (GH #452). If the send itself then fails, the new row is flipped to `failed` so the retry buttons can pick it up again; a task with no row is neither. diff --git a/docs/modules/file_storage.md b/docs/modules/file_storage.md index 1e3615b8a..8f3869cdc 100644 --- a/docs/modules/file_storage.md +++ b/docs/modules/file_storage.md @@ -386,3 +386,33 @@ class MyModule(ModuleBase): ## Locales Top-level keys in `file_storage/locales/en.json`: `browse`, `table`, `actions`, `delete_dialog`, `toasts`, `errors`. The `errors` namespace is keyed by error *code* (`not_found`, `too_large`, `bad_type`, `backend_error`, `bad_image`) so the UI can render a deterministic message per `StorageError` subclass. + +### Changing storage providers or locations + +Storage uses one active backend configuration; it does not retain or resolve +historical configurations. A stored file records the provider ID (`filesystem` +or `s3`), not the filesystem root, bucket, endpoint, or credentials. When its +provider ID differs from the active provider, downloads, public reads and +thumbnails fail with HTTP 409 and a translated message that the file is stored +on a backend that is no longer configured. Checksum jobs also refuse the read. +Deleting such a file still soft-deletes its row, but skips the original object +and thumbnail deletion and logs a warning. Retain the old store for manual +cleanup; the object reaper only handles the active provider. + +Changes within the same provider (for example a different filesystem root or +S3 bucket) cannot be detected by this check. Do not switch locations while +existing rows still depend on the old location. Upload sessions do not record +a backend ID: finish or abort all active multipart sessions before any backend +or location change, including sessions belonging to other users and tenants. + +For a manual migration, pause uploads and background storage jobs, drain active +multipart sessions, and back up the database and both stores. Copy originals +and any thumbnail variants to the destination under exactly the same object +keys, including tenant prefixes; verify sizes and checksums before switching. +When changing provider type, update the affected `StoredFile.backend` values to +the destination provider ID in a controlled database maintenance transaction +only after verifying the copies. Configure the destination, restart application +and worker processes, and verify authenticated/public downloads, ranged reads, +thumbnails and deletion before resuming writes. Keep the source and backups +until verification is complete, then remove source objects explicitly. There +is no automatic migration command or historical-backend registry. diff --git a/framework/core/simple_module_core/events.py b/framework/core/simple_module_core/events.py index a906c2e06..2d6b2b092 100644 --- a/framework/core/simple_module_core/events.py +++ b/framework/core/simple_module_core/events.py @@ -11,6 +11,7 @@ from pyee.asyncio import AsyncIOEventEmitter from simple_module_core.durable_events import ( + DurableContext, DurableDeliveryUnavailable, DurableHandler, DurableRegistry, @@ -141,3 +142,22 @@ async def publish_durable(self, event: Event, db: Any) -> None: if not names: return await transport.enqueue(event, names, db) + + async def publish_durable_or_inline(self, event: Event, db: Any) -> None: + """Enqueue with a transport, otherwise run durable handlers in *db*. + + Inline handlers run sequentially in subscription order on the caller's + transaction with attempt=1. They must flush rather than commit; errors + propagate so the caller can roll back. Ordinary subscribers are not run. + External effects must be deferred to the caller's post-commit hooks. + """ + if self.durable_transport is not None: + await self.publish_durable(event, db) + return + ctx = DurableContext(db=db, attempt=1) + key = self._event_key(type(event)) + for name in self.durable.handler_names(type(event)): + subscription = self.durable.lookup(key, name) + if subscription is None: + continue + await subscription.handler(event, ctx) diff --git a/framework/core/simple_module_core/tenant_events.py b/framework/core/simple_module_core/tenant_events.py new file mode 100644 index 000000000..fc3f435dd --- /dev/null +++ b/framework/core/simple_module_core/tenant_events.py @@ -0,0 +1,20 @@ +"""Tenant lifecycle contracts shared without a dependency on a tenant provider.""" + +from dataclasses import dataclass + +from simple_module_core.events import Event + +TENANT_DELETED_CHANNEL = "tenants.deleted" + + +@dataclass +class TenantDeleted(Event): + """The tenant registry row is gone; owners must durably tear down its data. + + Delivery is at least once. Retention stores may keep historical rows, + identified by their original tenant id, but must never expose live resources. + """ + + tenant_id: str + slug: str + name: str diff --git a/framework/core/tests/test_events_inline.py b/framework/core/tests/test_events_inline.py new file mode 100644 index 000000000..97a83811a --- /dev/null +++ b/framework/core/tests/test_events_inline.py @@ -0,0 +1,38 @@ +"""Explicit inline fallback preserves transaction ownership and error propagation.""" + +from unittest.mock import AsyncMock + +import pytest +from simple_module_core.durable_events import DurableContext +from simple_module_core.events import EventBus +from simple_module_core.tenant_events import TenantDeleted + + +@pytest.mark.parametrize("transport", [False, True]) +async def test_durable_or_inline_dispatch(transport): + bus = EventBus() + db = object() + event = TenantDeleted("id", "slug", "Name") + handlers = [AsyncMock(), AsyncMock()] + for index, handler in enumerate(handlers): + bus.subscribe_durable(TenantDeleted, handler, name=str(index)) + ordinary = AsyncMock() + bus.subscribe(TenantDeleted, ordinary) + if transport: + bus.durable_transport = AsyncMock() + await bus.publish_durable_or_inline(event, db) + for handler in handlers: + if transport: + handler.assert_not_awaited() + else: + handler.assert_awaited_once_with(event, DurableContext(db=db, attempt=1)) + ordinary.assert_not_awaited() + if transport: + bus.durable_transport.enqueue.assert_awaited_once_with(event, ["0", "1"], db) + + +async def test_inline_failure_propagates(): + bus = EventBus() + bus.subscribe_durable(TenantDeleted, AsyncMock(side_effect=ValueError("failed")), name="fail") + with pytest.raises(ValueError, match="failed"): + await bus.publish_durable_or_inline(TenantDeleted("id", "slug", "Name"), object()) diff --git a/framework/db/simple_module_db/tenant_cleanup.py b/framework/db/simple_module_db/tenant_cleanup.py new file mode 100644 index 000000000..c6a36139b --- /dev/null +++ b/framework/db/simple_module_db/tenant_cleanup.py @@ -0,0 +1,22 @@ +"""Durable teardown of framework-owned tenant settings, including branding.""" + +from simple_module_core.durable_events import DurableContext +from simple_module_core.events import EventBus +from simple_module_core.tenant_events import TenantDeleted +from sqlmodel import col, delete + +from simple_module_db.stored_settings import SCOPE_TENANT, Setting + + +def register_tenant_cleanup(bus: EventBus) -> None: + """Register in both the host and workers, even without the settings plugin.""" + bus.subscribe_durable(TenantDeleted, delete_tenant_settings) + + +async def delete_tenant_settings(event: TenantDeleted, ctx: DurableContext) -> None: + """Hard-delete only this tenant's explicit overrides; safe on replay.""" + await ctx.db.execute( + delete(Setting).where( + col(Setting.scope) == SCOPE_TENANT, col(Setting.scope_id) == event.tenant_id + ) + ) diff --git a/framework/db/tests/test_db_tenant_cleanup.py b/framework/db/tests/test_db_tenant_cleanup.py new file mode 100644 index 000000000..7b4364bb1 --- /dev/null +++ b/framework/db/tests/test_db_tenant_cleanup.py @@ -0,0 +1,58 @@ +"""Framework tenant settings teardown works independently of the UI plugin.""" + +from unittest.mock import AsyncMock, patch + +from simple_module_core.durable_events import DurableContext +from simple_module_core.events import EventBus +from simple_module_core.tenant_events import TenantDeleted +from simple_module_db.stored_settings import SCOPE_SYSTEM, SCOPE_TENANT, SCOPE_USER, Setting +from simple_module_db.tenant_cleanup import register_tenant_cleanup +from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine +from sqlmodel import select + +_TENANT = "deleted-tenant" +_OTHER = "retained-tenant" +_KEY = "branding.logo_file_id" + + +async def test_settings_cleanup_is_durable_scoped_and_idempotent(): + bus = EventBus() + register_tenant_cleanup(bus) + names = bus.durable.handler_names(TenantDeleted) + assert len(names) == 1 + from simple_module_core.durable_events import event_key + + subscription = bus.durable.lookup(event_key(TenantDeleted), names[0]) + assert subscription is not None + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + try: + async with engine.begin() as conn: + await conn.run_sync(Setting.metadata.create_all, tables=[Setting.__table__]) + async with AsyncSession(engine) as session: + session.add_all( + [ + Setting(scope=scope, scope_id=scope_id, key=_KEY, value="file-id") + for scope, scope_id in ( + (SCOPE_TENANT, _TENANT), + (SCOPE_TENANT, _OTHER), + (SCOPE_USER, _TENANT), + (SCOPE_SYSTEM, ""), + ) + ] + ) + await session.commit() + event = TenantDeleted(tenant_id=_TENANT, slug=_TENANT, name=_TENANT) + with patch.object(session, "commit", new_callable=AsyncMock) as commit: + await subscription.handler(event, DurableContext(db=session, attempt=1)) + await subscription.handler(event, DurableContext(db=session, attempt=2)) + commit.assert_not_awaited() + remaining = (await session.execute(select(Setting))).scalars().all() + assert {(row.scope, row.scope_id) for row in remaining} == { + (SCOPE_TENANT, _OTHER), + (SCOPE_USER, _TENANT), + (SCOPE_SYSTEM, ""), + } + await session.rollback() + assert len((await session.execute(select(Setting))).scalars().all()) == 4 + finally: + await engine.dispose() diff --git a/framework/hosting/simple_module_hosting/app_builder.py b/framework/hosting/simple_module_hosting/app_builder.py index 1dc08ea3f..59947a809 100644 --- a/framework/hosting/simple_module_hosting/app_builder.py +++ b/framework/hosting/simple_module_hosting/app_builder.py @@ -20,6 +20,7 @@ from simple_module_core.resources import ResourceTypeRegistry from simple_module_core.services import DiagnosticsState, Services from simple_module_core.setup_steps import SetupRegistry +from simple_module_db.tenant_cleanup import register_tenant_cleanup from simple_module_hosting._app_boot import ( build_fastapi_app, @@ -128,6 +129,7 @@ def create_app(settings: Settings | None = None) -> FastAPI: perm_registry = PermissionRegistry() ff_registry = FeatureFlagRegistry() event_bus = EventBus() + register_tenant_cleanup(event_bus) invalidation_bus = InvalidationBus() health_registry = HealthRegistry() public_route_registry = PublicRouteRegistry() diff --git a/modules/api_tokens/api_tokens/module.py b/modules/api_tokens/api_tokens/module.py index a0d8daf40..32a09bbe6 100644 --- a/modules/api_tokens/api_tokens/module.py +++ b/modules/api_tokens/api_tokens/module.py @@ -17,6 +17,7 @@ if TYPE_CHECKING: from fastapi import APIRouter, FastAPI from simple_module_core.audit_links import AuditLinkRegistry + from simple_module_core.events import EventBus from simple_module_core.permissions import PermissionRegistry @@ -35,6 +36,13 @@ class ApiTokensModule(ModuleBase): optional_depends_on=[c.MODULE_USERS, c.MODULE_SETTINGS], ) + def register_event_handlers(self, bus: EventBus, app: FastAPI | None = None) -> None: + from simple_module_core.tenant_events import TenantDeleted + + from api_tokens.tenant_cleanup import purge_tenant_data + + bus.subscribe_durable(TenantDeleted, purge_tenant_data) + def register_settings(self, app: FastAPI) -> None: from api_tokens.services import ApiTokensServices from api_tokens.settings import ApiTokensSettings diff --git a/modules/api_tokens/api_tokens/tenant_cleanup.py b/modules/api_tokens/api_tokens/tenant_cleanup.py new file mode 100644 index 000000000..cbcd3adb2 --- /dev/null +++ b/modules/api_tokens/api_tokens/tenant_cleanup.py @@ -0,0 +1,14 @@ +"""Durable teardown of tenant-owned api_tokens rows.""" + +from simple_module_core.durable_events import DurableContext +from simple_module_core.tenant_events import TenantDeleted +from simple_module_db.tenancy import tenant_context +from sqlmodel import delete + +from api_tokens.models import ApiToken + + +async def purge_tenant_data(event: TenantDeleted, ctx: DurableContext) -> None: + """Hard-delete tenant rows atomically with outbox acknowledgement.""" + with tenant_context(event.tenant_id): + await ctx.db.execute(delete(ApiToken).where(ApiToken.tenant_id == event.tenant_id)) diff --git a/modules/api_tokens/tests/test_api_tokens_tenant_cleanup.py b/modules/api_tokens/tests/test_api_tokens_tenant_cleanup.py new file mode 100644 index 000000000..ec3a1d677 --- /dev/null +++ b/modules/api_tokens/tests/test_api_tokens_tenant_cleanup.py @@ -0,0 +1,44 @@ +"""Tenant deletion purges only the deleted tenant through a durable handler.""" + +from datetime import UTC, datetime, timedelta + +from api_tokens.models import ApiToken +from api_tokens.module import ApiTokensModule +from simple_module_core.durable_events import DurableContext, event_key +from simple_module_core.events import EventBus +from simple_module_core.tenant_events import TenantDeleted +from simple_module_db.tenancy import all_tenants, tenant_context +from sqlmodel import select + + +async def test_durable_tenant_cleanup_is_scoped_idempotent_and_transactional(db_session): + bus = EventBus() + ApiTokensModule().register_event_handlers(bus) + names = bus.durable.handler_names(TenantDeleted) + assert len(names) == 1 + subscription = bus.durable.lookup(event_key(TenantDeleted), names[0]) + assert subscription is not None + for tenant_id in ("deleted-tenant", "other-tenant"): + with tenant_context(tenant_id): + db_session.add( + ApiToken( + name="token", + prefix="test", + token_hash=tenant_id, + expires_at=datetime.now(UTC) + timedelta(days=1), + ) + ) + await db_session.flush() + await db_session.commit() + event = TenantDeleted(tenant_id="deleted-tenant", slug="deleted", name="Deleted") + ctx = DurableContext(db=db_session, attempt=1) + # A worker rollback must restore rows: the handler must never commit its session. + await subscription.handler(event, ctx) + await db_session.rollback() + with all_tenants(): + assert len((await db_session.execute(select(ApiToken))).scalars().all()) == 2 + await subscription.handler(event, ctx) + await subscription.handler(event, DurableContext(db=db_session, attempt=2)) + with all_tenants(): + remaining = (await db_session.execute(select(ApiToken))).scalars().all() + assert [row.tenant_id for row in remaining] == ["other-tenant"] diff --git a/modules/background_tasks/background_tasks/after_commit.py b/modules/background_tasks/background_tasks/after_commit.py index 9922d1eff..f91633a46 100644 --- a/modules/background_tasks/background_tasks/after_commit.py +++ b/modules/background_tasks/background_tasks/after_commit.py @@ -59,6 +59,11 @@ def __getattr__(self, name: str) -> Any: def enqueue_after_commit(db: AsyncSession | Session, task: Any, *args: Any, **kwargs: Any) -> str: """Send *task* once *db*'s transaction commits; never if it rolls back. + Delivery is at-most-once: publish intent exists only in memory. Broker + failure after commit is logged without retry; a process exit between commit + and publish also loses the job. For must-run work use the event bus's + ``publish_durable(event, db)`` transactional outbox with a durable handler. + *task* is a Celery task (``send_receipt``), a signature (``send_receipt.s().set(queue="mail")``, for routing options), or a registered task name. ``*args`` / ``**kwargs`` are the task's own @@ -89,7 +94,7 @@ def after_commit( request. Left off for a single publish so Celery's eager mode — which runs the task body inside the publish — stays on the loop's thread. One failed publish is logged and does not stop the rest: the rows are already - committed, and an unsent ``pending`` row is visible and retryable. + committed. Only callers with persisted pending work can reconcile failures. """ batch = list(publishes) if isinstance(db, RequestSession) and SESSION_START_KEY in db.info: diff --git a/modules/background_tasks/background_tasks/outbox_worker.py b/modules/background_tasks/background_tasks/outbox_worker.py index 49e5a6a7b..49598160f 100644 --- a/modules/background_tasks/background_tasks/outbox_worker.py +++ b/modules/background_tasks/background_tasks/outbox_worker.py @@ -30,6 +30,7 @@ from simple_module_db import init_db from simple_module_db.listeners import attach_session_listeners from simple_module_db.query_filter import bind_engine_policy +from simple_module_db.tenant_cleanup import register_tenant_cleanup from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker from background_tasks import sync_db @@ -118,6 +119,7 @@ def build_worker_bus() -> EventBus: from simple_module_core.discovery import discover_modules, topological_sort bus = EventBus() + register_tenant_cleanup(bus) bus.durable_transport = WorkerOutboxTransport() # Lenient on purpose: the worker delivers whatever is installed, and a # module set the web process refuses is reported there, not here. diff --git a/modules/feature_flags/feature_flags/module.py b/modules/feature_flags/feature_flags/module.py index 2bc990849..8ac84ee83 100644 --- a/modules/feature_flags/feature_flags/module.py +++ b/modules/feature_flags/feature_flags/module.py @@ -29,6 +29,7 @@ ) if TYPE_CHECKING: + from simple_module_core.events import EventBus from sqlalchemy.ext.asyncio import AsyncSession _logger = logging.getLogger(__name__) @@ -67,6 +68,13 @@ class FeatureFlagsModule(ModuleBase): i18n_audience="admin", ) + def register_event_handlers(self, bus: EventBus, app: FastAPI | None = None) -> None: + from simple_module_core.tenant_events import TenantDeleted + + from feature_flags.tenant_cleanup import purge_tenant_data + + bus.subscribe_durable(TenantDeleted, purge_tenant_data) + def register_routes(self, api_router: APIRouter, view_router: APIRouter) -> None: from feature_flags.endpoints.api import router as api from feature_flags.endpoints.views import router as views diff --git a/modules/feature_flags/feature_flags/tenant_cleanup.py b/modules/feature_flags/feature_flags/tenant_cleanup.py new file mode 100644 index 000000000..ea1320cf2 --- /dev/null +++ b/modules/feature_flags/feature_flags/tenant_cleanup.py @@ -0,0 +1,18 @@ +"""Durable cleanup of tenant-scoped feature flag overrides.""" + +from simple_module_core.durable_events import DurableContext +from simple_module_core.tenant_events import TenantDeleted +from sqlmodel import col, delete + +from feature_flags.constants import SCOPE_TENANT +from feature_flags.models import FeatureFlagOverride + + +async def purge_tenant_data(event: TenantDeleted, ctx: DurableContext) -> None: + """Keep system/other tenants' overrides; acknowledge with the worker transaction.""" + await ctx.db.execute( + delete(FeatureFlagOverride).where( + col(FeatureFlagOverride.scope) == SCOPE_TENANT, + col(FeatureFlagOverride.scope_id) == event.tenant_id, + ) + ) diff --git a/modules/feature_flags/tests/test_feature_flags_tenant_cleanup.py b/modules/feature_flags/tests/test_feature_flags_tenant_cleanup.py new file mode 100644 index 000000000..595459cac --- /dev/null +++ b/modules/feature_flags/tests/test_feature_flags_tenant_cleanup.py @@ -0,0 +1,34 @@ +"""Tenant deletion purges tenant overrides without touching other scopes.""" + +from feature_flags.constants import SCOPE_SYSTEM, SCOPE_TENANT +from feature_flags.models import FeatureFlagOverride +from feature_flags.module import FeatureFlagsModule +from simple_module_core.durable_events import DurableContext, event_key +from simple_module_core.events import EventBus +from simple_module_core.tenant_events import TenantDeleted +from sqlmodel import select + + +async def test_durable_tenant_cleanup_is_scoped_idempotent_and_transactional(db_session): + bus = EventBus() + FeatureFlagsModule().register_event_handlers(bus) + names = bus.durable.handler_names(TenantDeleted) + assert len(names) == 1 + subscription = bus.durable.lookup(event_key(TenantDeleted), names[0]) + assert subscription is not None + scopes = [(SCOPE_TENANT, "deleted"), (SCOPE_TENANT, "other"), (SCOPE_SYSTEM, "deleted")] + for scope, scope_id in scopes: + db_session.add(FeatureFlagOverride(scope=scope, scope_id=scope_id, name="test.flag")) + await db_session.commit() + event = TenantDeleted(tenant_id="deleted", slug="deleted", name="Deleted") + ctx = DurableContext(db=db_session, attempt=1) + await subscription.handler(event, ctx) + await db_session.rollback() + assert len((await db_session.execute(select(FeatureFlagOverride))).scalars().all()) == 3 + await subscription.handler(event, ctx) + await subscription.handler(event, DurableContext(db=db_session, attempt=2)) + rows = (await db_session.execute(select(FeatureFlagOverride))).scalars().all() + assert {(row.scope, row.scope_id) for row in rows} == { + (SCOPE_TENANT, "other"), + (SCOPE_SYSTEM, "deleted"), + } diff --git a/modules/file_storage/file_storage/backend_guard.py b/modules/file_storage/file_storage/backend_guard.py new file mode 100644 index 000000000..33ef0a49c --- /dev/null +++ b/modules/file_storage/file_storage/backend_guard.py @@ -0,0 +1,28 @@ +"""Refuse object access through a provider that does not own the row.""" + +from file_storage.contracts.service import StorageBackend +from file_storage.models import StoredFile + + +class BackendUnavailableError(Exception): + """The file's recorded provider is no longer configured.""" + + +def backend_mismatch(backend: StorageBackend, row: StoredFile) -> bool: + """True when both identities are known and differ.""" + stored_backend = getattr(row, "backend", None) + active_backend = getattr(backend, "backend_id", None) + # Minimal adapters may omit identity; only reject a known provider mismatch. + return ( + isinstance(stored_backend, str) + and isinstance(active_backend, str) + and stored_backend != active_backend + ) + + +def require_backend(backend: StorageBackend, row: StoredFile) -> None: + """Check before any IO, presigning, or conditional response.""" + if backend_mismatch(backend, row): + raise BackendUnavailableError( + "This file is stored on a backend that is no longer configured." + ) diff --git a/modules/file_storage/file_storage/constants.py b/modules/file_storage/file_storage/constants.py index f3690e190..ca213ac4b 100644 --- a/modules/file_storage/file_storage/constants.py +++ b/modules/file_storage/file_storage/constants.py @@ -25,6 +25,7 @@ _MODULE_BACKGROUND_TASKS: Final = "BackgroundTasks" # ── Configuration ──────────────────────────────────────────────────── + ENV_PREFIX: Final = "SM_FILE_STORAGE_" # ── Routing ────────────────────────────────────────────────────────── @@ -99,6 +100,8 @@ class FeatureFlag: class ErrorCode: """Machine-readable codes returned in 4xx response bodies.""" + BACKEND_UNAVAILABLE: Final = "file_storage.backend_unavailable" + TOO_LARGE: Final = "file_storage.too_large" BAD_TYPE: Final = "file_storage.bad_type" NOT_FOUND: Final = "file_storage.not_found" @@ -125,12 +128,15 @@ class Task: REAP_OBJECTS: Final = "file_storage.reap_objects" REAP_STALE_UPLOADS: Final = "file_storage.reap_stale_uploads" + RECONCILE_CHECKSUMS: Final = "file_storage.reconcile_checksums" COMPUTE_CHECKSUM: Final = "file_storage.compute_checksum" class I18nKey: """Translator keys (full dotted paths). Mirror locales/en.json shape.""" + ERR_BACKEND_UNAVAILABLE: Final = "file_storage.errors.backend_unavailable" + ERR_NOT_FOUND: Final = "file_storage.errors.not_found" ERR_TOO_LARGE: Final = "file_storage.errors.too_large" ERR_BAD_TYPE: Final = "file_storage.errors.bad_type" @@ -251,3 +257,8 @@ class I18nKey: # Keys this module writes: ``{tenant}/YYYY/MM/DD/{uuid hex}[.ext]``. Reapers only # ever touch objects shaped like this, never foreign ones in a shared store. OWN_KEY: Final = re.compile(r"^[^/]+/\d{4}/\d{2}/\d{2}/[0-9a-f]{32}(\.[^/]*)?$") + +CHECKSUM_RECONCILE_MIN_AGE_MINUTES: Final = 30 +CHECKSUM_RECONCILE_BATCH_SIZE: Final = 500 +CHECKSUM_RECONCILE_INTERVAL_SECONDS: Final = 3600.0 +BEAT_RECONCILE_CHECKSUMS: Final = "file-storage-reconcile-checksums" diff --git a/modules/file_storage/file_storage/deletes.py b/modules/file_storage/file_storage/deletes.py index 2270d0694..4e737216f 100644 --- a/modules/file_storage/file_storage/deletes.py +++ b/modules/file_storage/file_storage/deletes.py @@ -16,6 +16,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from file_storage import queries, thumbnails +from file_storage.backend_guard import backend_mismatch from file_storage.contracts.service import StorageNotFoundError from file_storage.models import StoredFile from file_storage.scope import platform_scope @@ -111,6 +112,14 @@ async def delete( async def drop_object(self, row: StoredFile) -> None: """Delete ``row``'s backend object and its thumbnails; absent ones are fine.""" + if backend_mismatch(self.backend, row): + _logger.warning( + "file_storage.backend_mismatch key=%s stored=%s active=%s — object retained", + row.key, + row.backend, + getattr(self.backend, "backend_id", None), + ) + return try: # Acceptably absent — eg. a previous delete partially succeeded. with contextlib.suppress(StorageNotFoundError): diff --git a/modules/file_storage/file_storage/downloads.py b/modules/file_storage/file_storage/downloads.py index c12218d19..152116952 100644 --- a/modules/file_storage/file_storage/downloads.py +++ b/modules/file_storage/file_storage/downloads.py @@ -12,6 +12,8 @@ from typing import TYPE_CHECKING from urllib.parse import quote +from file_storage.backend_guard import require_backend + if TYPE_CHECKING: from file_storage.contracts.service import StorageBackend from file_storage.models import StoredFile @@ -52,6 +54,7 @@ async def presigned_download_url( Without the overrides the backend answers with the object's stored metadata, and the browser saves the file under its opaque key (GH #489). """ + require_backend(backend, row) return await backend.presigned_get_url( row.key, ttl_seconds, diff --git a/modules/file_storage/file_storage/endpoints/public.py b/modules/file_storage/file_storage/endpoints/public.py index 55f0f6f66..0528c7246 100644 --- a/modules/file_storage/file_storage/endpoints/public.py +++ b/modules/file_storage/file_storage/endpoints/public.py @@ -10,19 +10,34 @@ import uuid from fastapi import APIRouter, Depends, Query, Request, Response +from simple_module_db.query_filter import strict_configured from simple_module_hosting.i18n_deps import TranslatorDep from file_storage import constants from file_storage.deps import get_file_storage_service +from file_storage.scope import PLATFORM_TENANT_ID from file_storage.service import FileStorageService, StoredFileNotFoundError from file_storage.serving import not_found, public_file_response, thumbnail_response +from file_storage.tenant_cache import cached_tenant_exists router = APIRouter() -async def _public_row(service: FileStorageService, file_id: uuid.UUID, t: TranslatorDep): +async def _public_row( + service: FileStorageService, file_id: uuid.UUID, request: Request, t: TranslatorDep +): try: - return await service.get_public(file_id) + row = await service.get_public(file_id) + # Only a multi-tenant install keys files to registry rows: platform files + # (branding logo/favicon) and a single-tenant install's fallback tenant + # (``default`` or ``default_tenant``) have none by design. + if row.tenant_id != PLATFORM_TENANT_ID and strict_configured(service.db.sync_session): + # Return the read connection before tenant_exists opens its own session; + # the helper keeps the row readable and leaves sessions with writes alone. + await service._release_connection(row) + if await cached_tenant_exists(request.app, row.tenant_id) is False: + raise StoredFileNotFoundError(str(file_id)) + return row except StoredFileNotFoundError as exc: raise not_found(t) from exc @@ -36,7 +51,7 @@ async def public_thumbnail( w: int | None = Query(default=None, description="Width in px; clamped and snapped."), service: FileStorageService = Depends(get_file_storage_service), ) -> Response: - row = await _public_row(service, file_id, t) + row = await _public_row(service, file_id, request, t) return await thumbnail_response( service, row, @@ -50,7 +65,7 @@ async def public_thumbnail( async def _serve_public( service: FileStorageService, file_id: uuid.UUID, request: Request, t: TranslatorDep ) -> Response: - row = await _public_row(service, file_id, t) + row = await _public_row(service, file_id, request, t) return await public_file_response(service, row, request, t) diff --git a/modules/file_storage/file_storage/errors.py b/modules/file_storage/file_storage/errors.py new file mode 100644 index 000000000..cffb70972 --- /dev/null +++ b/modules/file_storage/file_storage/errors.py @@ -0,0 +1,25 @@ +"""HTTP mapping for unavailable stored-file providers.""" + +from fastapi import FastAPI, Request, status +from fastapi.responses import JSONResponse +from simple_module_hosting.i18n_deps import get_translator + +from file_storage import constants +from file_storage.backend_guard import BackendUnavailableError + + +async def _backend_unavailable(request: Request, exc: Exception) -> JSONResponse: + translator = await get_translator(request) + return JSONResponse( + status_code=status.HTTP_409_CONFLICT, + content={ + "detail": { + "code": constants.ErrorCode.BACKEND_UNAVAILABLE, + "message": translator.t(constants.I18nKey.ERR_BACKEND_UNAVAILABLE), + } + }, + ) + + +def install_exception_handlers(app: FastAPI) -> None: + app.add_exception_handler(BackendUnavailableError, _backend_unavailable) diff --git a/modules/file_storage/file_storage/locales/de.json b/modules/file_storage/file_storage/locales/de.json index 20a545b01..4694cd599 100644 --- a/modules/file_storage/file_storage/locales/de.json +++ b/modules/file_storage/file_storage/locales/de.json @@ -72,7 +72,8 @@ "bad_part": "Dieser Upload-Teil ist für diese Datei ungültig", "parts_mismatch": "Die hochgeladenen Teile stimmen nicht mit den gesendeten Daten überein. Setzen Sie den Upload fort, um den Rest zu senden", "size_mismatch": "Die zusammengesetzte Datei hat die Größe {actual}, erwartet wurde {expected}. Laden Sie sie erneut hoch", - "upload_session_closed": "Dieser Upload ist nicht mehr aktiv. Starten Sie ihn erneut" + "upload_session_closed": "Dieser Upload ist nicht mehr aktiv. Starten Sie ihn erneut", + "backend_unavailable": "Diese Datei ist auf einem Backend gespeichert, das nicht mehr konfiguriert ist." }, "filters": { "search_placeholder": "Dateinamen suchen…", diff --git a/modules/file_storage/file_storage/locales/en.json b/modules/file_storage/file_storage/locales/en.json index 1f3dfd228..a62ce5573 100644 --- a/modules/file_storage/file_storage/locales/en.json +++ b/modules/file_storage/file_storage/locales/en.json @@ -72,7 +72,8 @@ "bad_part": "That upload part is not valid for this file", "parts_mismatch": "The uploaded parts do not match what was sent; resume the upload to send the rest", "size_mismatch": "The assembled file is {actual}, but {expected} was expected; upload it again", - "upload_session_closed": "This upload is no longer active; start it again" + "upload_session_closed": "This upload is no longer active; start it again", + "backend_unavailable": "This file is stored on a backend that is no longer configured." }, "filters": { "search_placeholder": "Search filenames…", diff --git a/modules/file_storage/file_storage/models.py b/modules/file_storage/file_storage/models.py index 71a57feb0..005892d28 100644 --- a/modules/file_storage/file_storage/models.py +++ b/modules/file_storage/file_storage/models.py @@ -21,8 +21,9 @@ class StoredFile(Base, AuditMixin, SoftDeleteMixin, MultiTenantMixin, table=True """A file persisted to a configured storage backend. The ``backend`` column records which provider holds the bytes — important - if the active backend is changed after ingest, since old rows still need - to be located on their original provider until they're migrated. + if the active backend is changed after ingest: object access is refused + until the bytes and row are manually migrated. This records provider type, + not a historical bucket/root configuration. Tenant-scoped (#383): every read, update and delete is filtered to the bound tenant, so a file id from another tenant simply does not resolve. diff --git a/modules/file_storage/file_storage/module.py b/modules/file_storage/file_storage/module.py index b1d5264cd..9072ee903 100644 --- a/modules/file_storage/file_storage/module.py +++ b/modules/file_storage/file_storage/module.py @@ -12,7 +12,9 @@ from simple_module_core.body_limits import BodyLimitRegistry from simple_module_core.cors import CorsPolicyRegistry from simple_module_core.csp import CspSourceRegistry +from simple_module_core.events import EventBus from simple_module_core.feature_flags import FeatureFlagDefinition, FeatureFlagRegistry +from simple_module_core.invalidation import InvalidationBus from simple_module_core.menu import MenuItem, MenuRegistry, MenuSection from simple_module_core.module import ModuleBase, ModuleMeta from simple_module_core.module_settings import register_module_settings @@ -55,6 +57,21 @@ def register_settings(self, app: FastAPI) -> None: lambda s: FileStorageServices(settings=s), ) + def register_event_handlers(self, bus: EventBus, app: FastAPI | None = None) -> None: + from file_storage.tenant_deletion import register_tenant_deletion + + register_tenant_deletion(bus, app) + + def register_invalidations(self, bus: InvalidationBus, app: FastAPI) -> None: + from file_storage.tenant_cache import register_tenant_cache + + register_tenant_cache(bus, app) + + def register_exception_handlers(self, app: FastAPI) -> None: + from file_storage.errors import install_exception_handlers + + install_exception_handlers(app) + def register_routes(self, api_router: APIRouter, view_router: APIRouter) -> None: from file_storage.endpoints.api import router as api from file_storage.endpoints.public import router as public diff --git a/modules/file_storage/file_storage/ranges.py b/modules/file_storage/file_storage/ranges.py index 274429493..5b844053d 100644 --- a/modules/file_storage/file_storage/ranges.py +++ b/modules/file_storage/file_storage/ranges.py @@ -17,6 +17,7 @@ from fastapi import Request, Response, status from fastapi.responses import StreamingResponse +from file_storage.backend_guard import require_backend from file_storage.contracts.service import StorageBackend from file_storage.models import StoredFile @@ -95,6 +96,7 @@ async def file_body_response( response actually served, so a ranged request never opens the full object. Raises :class:`StorageNotFoundError` if the object is gone. """ + require_backend(backend, row) size = row.size_bytes get_range = getattr(backend, "get_range", None) parsed = parse_range(_range_header(request, etag), size) if get_range else None diff --git a/modules/file_storage/file_storage/service.py b/modules/file_storage/file_storage/service.py index 086fab052..5ac0809ff 100644 --- a/modules/file_storage/file_storage/service.py +++ b/modules/file_storage/file_storage/service.py @@ -20,6 +20,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from file_storage import constants, queries +from file_storage.backend_guard import require_backend from file_storage.contracts.schemas import StoredFileOut from file_storage.deletes import FileStorageDeletes from file_storage.downloads import ( @@ -190,6 +191,7 @@ async def download( works without touching this method. """ row = await self.get(file_id, platform=platform, owner=owner) + require_backend(self.backend, row) if self.backend.supports_presigned_url: url = await presigned_download_url( self.backend, row, self.settings.s3_presign_ttl_seconds, attachment=True diff --git a/modules/file_storage/file_storage/serving.py b/modules/file_storage/file_storage/serving.py index 2b020698f..29ed02ffa 100644 --- a/modules/file_storage/file_storage/serving.py +++ b/modules/file_storage/file_storage/serving.py @@ -7,6 +7,7 @@ from fastapi.responses import RedirectResponse from file_storage import constants, thumbnails +from file_storage.backend_guard import require_backend from file_storage.contracts.service import StorageNotFoundError from file_storage.downloads import content_disposition, presigned_download_url from file_storage.media_types import is_active_content @@ -60,6 +61,7 @@ async def thumbnail_response( request: Request | None = None, ) -> Response: """Serve ``row``'s resized variant; 404 for non-images, 422 if undecodable.""" + require_backend(service.backend, row) snapped = thumbnails.snap_width(width) etag = etag_for(row, f"-w{snapped}") headers = _headers(etag, cache_control) @@ -94,6 +96,7 @@ async def public_file_response( and is streamed even on presigning backends so these headers always apply; everything else may redirect to the backend's own URL. """ + require_backend(service.backend, row) max_age = constants.PUBLIC_MAX_AGE_SECONDS etag = etag_for(row) active = is_active_content(row.content_type) diff --git a/modules/file_storage/file_storage/tasks.py b/modules/file_storage/file_storage/tasks.py index e6b842908..e91093593 100644 --- a/modules/file_storage/file_storage/tasks.py +++ b/modules/file_storage/file_storage/tasks.py @@ -25,6 +25,10 @@ logger = logging.getLogger(__name__) BEAT_SCHEDULE = { + constants.BEAT_RECONCILE_CHECKSUMS: { + "task": constants.Task.RECONCILE_CHECKSUMS, + "schedule": constants.CHECKSUM_RECONCILE_INTERVAL_SECONDS, + }, "file-storage-reap-objects": { "task": constants.Task.REAP_OBJECTS, "schedule": 3600.0, @@ -147,6 +151,7 @@ def compute_checksum(file_id: str, tenant_id: str) -> str | None: from background_tasks.sync_db import sync_session from simple_module_db import tenant_context + from file_storage.backend_guard import require_backend from file_storage.checksum import hash_object from file_storage.contracts.service import StorageNotFoundError from file_storage.models import StoredFile @@ -160,6 +165,7 @@ def compute_checksum(file_id: str, tenant_id: str) -> str | None: row = db.get(StoredFile, target) if row is None or row.checksum_sha256: return None + require_backend(backend, row) key = row.key try: digest = asyncio.run(hash_object(backend, key, on_chunk=heartbeat)) @@ -189,3 +195,39 @@ def reap_objects() -> int: grace=timedelta(seconds=2 * settings.upload_session_ttl_seconds), ) ) + + +@shared_task(name=constants.Task.RECONCILE_CHECKSUMS) +def reconcile_checksums() -> int: + """Retry missing checksums after the grace period; return bounded enqueue attempts. + + StoredFile rows represent completed uploads. Missing digests remain eligible + after publish failures or worker exits; duplicate checksum jobs are safe. + """ + from background_tasks import enqueue_after_commit + from background_tasks.sync_db import sync_session + from simple_module_db import all_tenants, tenant_context + from sqlalchemy import func + from sqlmodel import col, select + + from file_storage.models import StoredFile + + backend, _ = _backend_and_settings() + cutoff = datetime.now(UTC) - timedelta(minutes=constants.CHECKSUM_RECONCILE_MIN_AGE_MINUTES) + with sync_session() as db, all_tenants(): + targets = list( + db.execute( + select(StoredFile.id, StoredFile.tenant_id) + .where( + col(StoredFile.backend) == backend.backend_id, + col(StoredFile.checksum_sha256).is_(None), + col(StoredFile.created_at) < cutoff, + ) + .order_by(func.random()) + .limit(constants.CHECKSUM_RECONCILE_BATCH_SIZE) + ).all() + ) + for file_id, tenant_id in targets: + with tenant_context(tenant_id), sync_session() as db: + enqueue_after_commit(db, compute_checksum, str(file_id), tenant_id) + return len(targets) diff --git a/modules/file_storage/file_storage/tenant_cache.py b/modules/file_storage/file_storage/tenant_cache.py new file mode 100644 index 000000000..673d92fcf --- /dev/null +++ b/modules/file_storage/file_storage/tenant_cache.py @@ -0,0 +1,53 @@ +"""Bound public-file tenant lookups; deletion events accelerate TTL eviction.""" + +from collections import OrderedDict +from time import monotonic + +from fastapi import FastAPI +from simple_module_core.invalidation import Invalidation, InvalidationBus +from simple_module_core.tenancy import tenant_exists +from simple_module_core.tenant_events import TENANT_DELETED_CHANNEL + +TENANT_EXISTS_TTL_SECONDS = 60 +TENANT_EXISTS_CACHE_SIZE = 1024 + + +class TenantExistenceCache: + def __init__(self) -> None: + self.entries: OrderedDict[str, tuple[float, bool | None]] = OrderedDict() + self.generation = 0 + + def forget(self, tenant_id: str | None) -> None: + self.generation += 1 + if tenant_id is None: + self.entries.clear() + else: + self.entries.pop(tenant_id, None) + + +async def cached_tenant_exists(app: FastAPI, tenant_id: str) -> bool | None: + cache = getattr(app.state, "file_storage_tenant_cache", None) + if cache is None: + cache = app.state.file_storage_tenant_cache = TenantExistenceCache() + now = monotonic() + entry = cache.entries.get(tenant_id) + if entry is not None and now - entry[0] < TENANT_EXISTS_TTL_SECONDS: + return entry[1] + generation = cache.generation + answer = await tenant_exists(app, tenant_id) + # An in-flight lookup must not restore an entry evicted by deletion. + if generation == cache.generation: + cache.entries[tenant_id] = (now, answer) + cache.entries.move_to_end(tenant_id) + while len(cache.entries) > TENANT_EXISTS_CACHE_SIZE: + cache.entries.popitem(last=False) + return answer + + +def register_tenant_cache(bus: InvalidationBus, app: FastAPI) -> None: + app.state.file_storage_tenant_cache = TenantExistenceCache() + + async def forget(event: Invalidation) -> None: + app.state.file_storage_tenant_cache.forget(event.key) + + bus.subscribe(TENANT_DELETED_CHANNEL, forget) diff --git a/modules/file_storage/file_storage/tenant_deletion.py b/modules/file_storage/file_storage/tenant_deletion.py new file mode 100644 index 000000000..ab9682af8 --- /dev/null +++ b/modules/file_storage/file_storage/tenant_deletion.py @@ -0,0 +1,149 @@ +"""Durable tenant teardown, with backend work following the row-deletion commit.""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from simple_module_core.durable_events import DurableContext +from simple_module_core.events import Event, EventBus +from simple_module_core.tenant_events import TenantDeleted +from simple_module_db import tenant_context +from sqlmodel import col, delete, select + +from file_storage import constants +from file_storage.backends import build_backend +from file_storage.contracts.service import StorageBackend, StorageNotFoundError +from file_storage.models import StoredFile, UploadSession +from file_storage.settings import FileStorageSettings +from file_storage.thumbnails import variant_key + +if TYPE_CHECKING: + from fastapi import FastAPI + +_logger = logging.getLogger(__name__) + +CLEANUP_BATCH_SIZE = 500 + +_PURGE_HANDLER = "file_storage.tenant_deletion.purge_rows" +_OBJECT_HANDLER = "file_storage.tenant_deletion.purge_objects" + + +@dataclass +class TenantStorageCleanupRequested(Event): + """A persisted cleanup manifest, surviving deletion of the source rows.""" + + files: list[tuple[str, str]] + uploads: list[tuple[str, str | None]] + + +def register_tenant_deletion(bus: EventBus, app: FastAPI | None) -> None: + """Register on both web and worker buses, including workers without an app.""" + + async def purge_rows(event: TenantDeleted, ctx: DurableContext) -> None: + with tenant_context(event.tenant_id): + for model, columns in ( + (StoredFile, (StoredFile.backend, StoredFile.key)), + ( + UploadSession, + (UploadSession.key, UploadSession.backend_upload_id, UploadSession.status), + ), + ): + result = await ctx.db.execute( + select(*columns) + .where(col(model.tenant_id) == event.tenant_id) + .execution_options(include_deleted=True) + ) + projected = result.all() + for start in range(0, len(projected), CLEANUP_BATCH_SIZE): + rows = projected[start : start + CLEANUP_BATCH_SIZE] + manifest = TenantStorageCleanupRequested( + files=[(row[0], row[1]) for row in rows] if model is StoredFile else [], + uploads=[ + (row[0], row[1]) + for row in rows + if row[2] == constants.UploadStatus.ACTIVE + ] + if model is UploadSession + else [], + ) + if not manifest.files and not manifest.uploads: + continue + on_commit = getattr(ctx.db, "on_commit", None) + if bus.durable_transport is not None: + await bus.publish_durable(manifest, ctx.db) + elif on_commit is not None: + # Bind each chunk so callbacks cannot reuse the last manifest. + on_commit( + lambda manifest=manifest: bus.publish_durable_or_inline( + manifest, ctx.db + ) + ) + else: + _logger.warning( + "file_storage.tenant_cleanup_deferred tenant=%s objects=%d", + event.tenant_id, + len(manifest.files), + ) + for model in (UploadSession, StoredFile): + await ctx.db.execute( + delete(model) + .where(col(model.tenant_id) == event.tenant_id) + .execution_options(include_deleted=True) + ) + + async def purge_objects(event: TenantStorageCleanupRequested, ctx: DurableContext) -> None: + # Run only after the row transaction commits. With a transport, + # failures retain the manifest in the outbox for idempotent retries. + settings = _settings(app) + backends: dict[str, StorageBackend] = {} + failures: list[Exception] = [] + + async def attempt(operation) -> None: + try: + await operation + except StorageNotFoundError: + pass + except Exception as exc: + failures.append(exc) + + for backend_id, key in event.files: + if backend_id != settings.backend: + # Only the active provider is configured; building another from + # these settings would raise on every retry and wedge the rest. + _logger.warning( + "file_storage.backend_mismatch key=%s stored=%s active=%s — object retained", + key, + backend_id, + settings.backend, + ) + continue + backend = backends.get(backend_id) + if backend is None: + backend = build_backend(settings.model_copy(update={"backend": backend_id})) + backends[backend_id] = backend + for target in [key, *(variant_key(key, w) for w in constants.THUMBNAIL_WIDTHS)]: + await attempt(backend.delete(target)) + if event.uploads: + backend = build_backend(settings) + for key, upload_id in event.uploads: + if upload_id is None: + continue + # Attempt the delete even if abort fails; retry the entire + # idempotent manifest on failure, bounded by outbox max attempts. + await attempt(backend.abort_multipart(key, upload_id)) + await attempt(backend.delete(key)) + if failures: + raise ExceptionGroup("Tenant storage cleanup failed", failures) + + bus.subscribe_durable(TenantDeleted, purge_rows, name=_PURGE_HANDLER) + bus.subscribe_durable(TenantStorageCleanupRequested, purge_objects, name=_OBJECT_HANDLER) + + +def _settings(app: FastAPI | None) -> FileStorageSettings: + if app is not None: + return app.state.file_storage.settings + from background_tasks import settings_for + + return settings_for(FileStorageSettings, constants.MODULE_NAME) diff --git a/modules/file_storage/file_storage/visibility.py b/modules/file_storage/file_storage/visibility.py index 3388edeef..6975aa9d3 100644 --- a/modules/file_storage/file_storage/visibility.py +++ b/modules/file_storage/file_storage/visibility.py @@ -14,6 +14,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from file_storage import thumbnails +from file_storage.backend_guard import require_backend from file_storage.models import StoredFile if TYPE_CHECKING: @@ -74,6 +75,7 @@ async def thumbnail(self, row: StoredFile, width: int) -> bytes: is handed back first so a burst of cold (possibly anonymous) requests cannot pin the whole pool on work that needs no database. """ + require_backend(self.backend, row) key, content_type = row.key, row.content_type await self._release_connection(row) return await thumbnails.get_or_create( diff --git a/modules/file_storage/tests/test_file_storage_backend_mismatch.py b/modules/file_storage/tests/test_file_storage_backend_mismatch.py new file mode 100644 index 000000000..62f64699f --- /dev/null +++ b/modules/file_storage/tests/test_file_storage_backend_mismatch.py @@ -0,0 +1,100 @@ +"""Rows from a previous provider must never use the active provider (#589).""" + +from io import BytesIO +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +from fastapi import UploadFile +from file_storage import constants, thumbnails +from file_storage.backends.filesystem import FilesystemBackend +from file_storage.service import FileStorageService +from file_storage.settings import FileStorageSettings + + +@pytest.mark.parametrize( + ("backend", "row"), + [ + (SimpleNamespace(), SimpleNamespace(backend="filesystem")), + (SimpleNamespace(backend_id="filesystem"), SimpleNamespace()), + ( + SimpleNamespace(backend_id="filesystem"), + SimpleNamespace(backend="filesystem"), + ), + ], +) +def test_guard_allows_missing_or_matching_identity(backend, row): + from file_storage.backend_guard import require_backend + + require_backend(backend, row) + + +@pytest.fixture +async def mismatched(db_session, tmp_path): + backend = FilesystemBackend(root=tmp_path) + service = FileStorageService(db_session, backend, FileStorageSettings()) + out = await service.upload(UploadFile(filename="old.png", file=BytesIO(b"original"))) + row = await service.get(out.id) + row.backend = constants.BackendId.S3 + await db_session.flush() + return service, row + + +@pytest.mark.parametrize("operation", ["stream", "redirect", "metadata", "thumbnail"]) +async def test_old_provider_reads_are_refused(mismatched, operation, monkeypatch): + from file_storage.backend_guard import BackendUnavailableError + + service, row = mismatched + service.backend.get = AsyncMock(side_effect=AssertionError("wrong store")) + service.backend.presigned_get_url = AsyncMock(side_effect=AssertionError("wrong store")) + monkeypatch.setattr( + thumbnails, "get_or_create", AsyncMock(side_effect=AssertionError("wrong store")) + ) + if operation == "redirect": + service.backend.supports_presigned_url = True + with pytest.raises(BackendUnavailableError): + if operation == "thumbnail": + await service.thumbnail(row, 128) + else: + await service.download(row.id, open_body=operation != "metadata") + + +@pytest.mark.parametrize("bulk", [False, True]) +async def test_old_provider_delete_keeps_objects(mismatched, bulk, caplog): + service, row = mismatched + service.backend.delete = AsyncMock(side_effect=AssertionError("wrong store")) + if bulk: + await service.delete_many([row.id]) + else: + await service.delete(row.id) + assert row.is_deleted + service.backend.delete.assert_not_awaited() + assert "backend_mismatch" in caplog.text + + +@pytest.mark.parametrize( + "path", ["files/{id}/download", "files/{id}/thumbnail", "public/{id}", "public/{id}/thumbnail"] +) +async def test_old_provider_http_conflict(authenticated_client, storage, path): + response = await authenticated_client.post( + f"{constants.ROUTE_PREFIX_API}/upload", + files={"file": ("a.png", b"image", "image/png")}, + data={"public": "true"}, + ) + assert response.status_code == 201, response.text + uploaded = response.json() + file_id = uploaded["id"] + suffix = "-w256" if path.endswith("thumbnail") else "" + etag = f'"{uploaded["checksum_sha256"]}{suffix}"' + # Change provider identity while retaining the same objects to expose any + # accidental access, including a cached/conditional response. + storage.backend.backend_id = constants.BackendId.S3 + response = await authenticated_client.get( + f"{constants.ROUTE_PREFIX_API}/{path.format(id=file_id)}", + headers={"Range": "bytes=0-1", "If-None-Match": etag}, + ) + assert response.status_code == 409, response.text + assert response.json()["detail"] == { + "code": "file_storage.backend_unavailable", + "message": "This file is stored on a backend that is no longer configured.", + } diff --git a/modules/file_storage/tests/test_file_storage_checksum_reconciliation.py b/modules/file_storage/tests/test_file_storage_checksum_reconciliation.py new file mode 100644 index 000000000..dd49d528a --- /dev/null +++ b/modules/file_storage/tests/test_file_storage_checksum_reconciliation.py @@ -0,0 +1,127 @@ +"""Recover checksum jobs lost after the upload transaction committed (#590).""" + +from datetime import UTC, datetime, timedelta +from unittest.mock import Mock + +import pytest +from background_tasks import sync_db +from file_storage import constants, tasks +from file_storage.models import Base, StoredFile +from simple_module_db import tenant_context +from simple_module_db.tenancy import current_tenant_id +from sqlalchemy import create_engine + + +@pytest.fixture +def checksum_worker(tmp_path, monkeypatch): + url = f"sqlite:///{tmp_path / 'checksums.db'}" + engine = create_engine(url) + Base.metadata.create_all(engine) + engine.dispose() + monkeypatch.setattr(sync_db, "discover_audit_callback", lambda: None) + sync_db.set_database_url(url, tenant_strict=True) + monkeypatch.setattr( + tasks, "_backend_and_settings", lambda: (Mock(backend_id="filesystem"), None) + ) + yield + sync_db.dispose_sync_engine() + + +def _file(tenant, *, age=60, checksum=None, deleted=False): + return StoredFile( + tenant_id=tenant, + key=f"{tenant}/{age}/{deleted}/{checksum}", + filename="file.bin", + content_type="application/octet-stream", + size_bytes=1, + backend=constants.BackendId.FILESYSTEM, + created_at=datetime.now(UTC) - timedelta(minutes=age), + checksum_sha256=checksum, + is_deleted=deleted, + ) + + +def test_reconciler_is_registered_hourly(): + assert tasks.BEAT_SCHEDULE["file-storage-reconcile-checksums"] == { + "task": "file_storage.reconcile_checksums", + "schedule": 3600.0, + } + assert tasks.reconcile_checksums.name == "file_storage.reconcile_checksums" + + +def test_recovers_lost_jobs_across_tenants_and_skips_ineligible_files( + checksum_worker, + monkeypatch, +): + rows = [_file("tenant-a"), _file("tenant-b")] + with sync_db.sync_session() as db: + for row in [ + *rows, + _file("tenant-a", age=0), + _file("tenant-a", checksum="a" * 64), + _file("tenant-a", deleted=True), + ]: + with tenant_context(row.tenant_id): + db.add(row) + db.flush() + + published = [] + + def publish(*, args, **kwargs): + published.append((args, current_tenant_id.get())) + + monkeypatch.setattr(tasks.compute_checksum, "apply_async", publish) + assert tasks.reconcile_checksums() == 2 + assert set(published) == {((str(row.id), row.tenant_id), row.tenant_id) for row in rows} + + +def test_batch_is_bounded_and_failed_publishes_are_retried(checksum_worker, monkeypatch): + with sync_db.sync_session() as db, tenant_context("tenant-a"): + db.add(_file("tenant-a", age=60)) + monkeypatch.setattr(constants, "CHECKSUM_RECONCILE_BATCH_SIZE", 1) + publish = Mock(side_effect=OSError("broker unavailable")) + monkeypatch.setattr(tasks.compute_checksum, "apply_async", publish) + + assert tasks.reconcile_checksums() == 1 + assert publish.call_count == 1 + lost_args = publish.call_args.kwargs["args"] + publish.reset_mock(side_effect=True) + assert tasks.reconcile_checksums() == 1 + assert publish.call_args.kwargs["args"] == lost_args + + +def test_other_backend_rows_do_not_starve_active_backend(checksum_worker, monkeypatch): + monkeypatch.setattr(constants, "CHECKSUM_RECONCILE_BATCH_SIZE", 2) + with sync_db.sync_session() as db, tenant_context("tenant-a"): + for age in range(100, 103): + row = _file("tenant-a", age=age) + row.backend = constants.BackendId.S3 + db.add(row) + active = _file("tenant-a") + db.add(active) + publish = Mock() + monkeypatch.setattr(tasks.compute_checksum, "apply_async", publish) + assert tasks.reconcile_checksums() == 1 + assert publish.call_args.kwargs["args"] == (str(active.id), active.tenant_id) + + +def test_candidate_order_is_random(checksum_worker, monkeypatch): + from sqlalchemy import event + + with sync_db.sync_session() as db, tenant_context("tenant-a"): + db.add_all([_file("tenant-a", age=60), _file("tenant-a", age=90)]) + statements = [] + with sync_db.sync_session() as db: + engine = db.get_bind() + + def capture(conn, cursor, statement, parameters, context, executemany): + statements.append(statement) + + event.listen(engine, "before_cursor_execute", capture) + monkeypatch.setattr(constants, "CHECKSUM_RECONCILE_BATCH_SIZE", 1) + monkeypatch.setattr(tasks.compute_checksum, "apply_async", Mock()) + try: + assert tasks.reconcile_checksums() == 1 + finally: + event.remove(engine, "before_cursor_execute", capture) + assert any("ORDER BY random()" in statement for statement in statements) diff --git a/modules/file_storage/tests/test_file_storage_cleanup_chunks.py b/modules/file_storage/tests/test_file_storage_cleanup_chunks.py new file mode 100644 index 000000000..353baacef --- /dev/null +++ b/modules/file_storage/tests/test_file_storage_cleanup_chunks.py @@ -0,0 +1,55 @@ +"""Tenant cleanup manifests bound both file and upload payloads.""" + +from datetime import UTC, datetime +from unittest.mock import AsyncMock + +from file_storage import constants +from file_storage.models import StoredFile, UploadSession +from file_storage.tenant_deletion import register_tenant_deletion +from simple_module_core.events import EventBus +from simple_module_core.tenant_events import TenantDeleted +from simple_module_db import tenant_context +from simple_module_db.transaction import finalize_session + + +async def test_cleanup_chunks_preserve_every_entry(db_session, monkeypatch): + monkeypatch.setattr("file_storage.tenant_deletion.CLEANUP_BATCH_SIZE", 2) + bus = EventBus() + register_tenant_deletion(bus, None) + with tenant_context("chunked"): + for i in range(5): + db_session.add( + StoredFile( + key=f"file-{i}", + filename="file", + content_type="text/plain", + size_bytes=1, + backend=constants.BackendId.FILESYSTEM, + ) + ) + db_session.add( + UploadSession( + key=f"upload-{i}", + filename="upload", + content_type="text/plain", + total_size=1, + part_size=1, + backend_upload_id=f"id-{i}", + expires_at=datetime.now(UTC), + ) + ) + await finalize_session(db_session) + await bus.publish_durable_or_inline( + TenantDeleted("chunked", "chunked", "Chunked"), db_session + ) + publish = AsyncMock() + monkeypatch.setattr(bus, "publish_durable_or_inline", publish) + assert publish.await_count == 0 + await finalize_session(db_session) + manifests = [call.args[0] for call in publish.await_args_list] + assert len(manifests) == 6 + assert all(0 < len(m.files) + len(m.uploads) <= 2 for m in manifests) + assert {key for m in manifests for _, key in m.files} == {f"file-{i}" for i in range(5)} + assert {entry for m in manifests for entry in m.uploads} == { + (f"upload-{i}", f"id-{i}") for i in range(5) + } diff --git a/modules/file_storage/tests/test_file_storage_deleted_tenant_public.py b/modules/file_storage/tests/test_file_storage_deleted_tenant_public.py new file mode 100644 index 000000000..0cab5eb1e --- /dev/null +++ b/modules/file_storage/tests/test_file_storage_deleted_tenant_public.py @@ -0,0 +1,95 @@ +"""Public originals and thumbnails deny orphan files before teardown delivery.""" + +import io + +import pytest +from file_storage import constants +from PIL import Image +from simple_module_core.tenant_events import TENANT_DELETED_CHANNEL +from sqlalchemy import delete +from tenants.models import Invitation, Membership, Tenant + + +@pytest.mark.parametrize("directory", [True, False], ids=["with-directory", "without-directory"]) +async def test_file_storage_public_checks_deleted_tenant( + app, authenticated_client, client, admin_tenant_id, directory, monkeypatch +): + content = io.BytesIO() + Image.new("RGB", (20, 20)).save(content, format="PNG") + response = await authenticated_client.post( + f"{constants.ROUTE_PREFIX_API}{constants.PATH_UPLOAD}", + files={"file": ("image.png", content.getvalue(), "image/png")}, + data={"public": "true"}, + ) + assert response.status_code == 201, response.text + file = response.json() + urls = [ + file["public_url"], + f"{constants.ROUTE_PREFIX_API}/public/{file['id']}", + f"{constants.ROUTE_PREFIX_API}/public/{file['id']}/thumbnail?w=64", + ] + from file_storage.service import FileStorageService + + original_get_public = FileStorageService.get_public + original_lookup = app.state.tenant_exists + request_sessions = [] + + async def get_public(service, file_id): + row = await original_get_public(service, file_id) + assert service.db.in_transaction() + request_sessions.append(service.db) + return row + + async def lookup(tenant_id): + assert not request_sessions[-1].in_transaction() + return await original_lookup(tenant_id) + + monkeypatch.setattr(FileStorageService, "get_public", get_public) + monkeypatch.setattr(app.state, "tenant_exists", lookup) + monkeypatch.setattr("file_storage.tenant_cache.TENANT_EXISTS_TTL_SECONDS", 0) + for url in urls: + assert (await client.get(url)).status_code == 200 + assert not request_sessions[-1].in_transaction() + + # Bypass the deletion publisher to leave the rows/bytes in place: the + # immediate guard must work even while the outbox worker is unavailable. + async with app.state.sm.db.session_factory() as db: + for model in (Membership, Invitation): + await db.execute(delete(model).where(model.tenant_id == admin_tenant_id)) + await db.execute(delete(Tenant).where(Tenant.id == admin_tenant_id)) + await db.commit() + + # A missed event is bounded by the TTL; expire the previously cached answer. + monkeypatch.setattr("file_storage.tenant_cache.TENANT_EXISTS_TTL_SECONDS", 0) + if not directory: + # Hosts without a tenant registry retain the single-tenant contract. + monkeypatch.delattr(app.state, "tenant_exists") + for url in urls: + response = await client.get(url, headers={"If-None-Match": '"cached"'}) + assert response.status_code == (404 if directory else 200), response.text + + +async def test_tenant_existence_cache_hit_invalidation_and_ttl(monkeypatch): + from unittest.mock import AsyncMock + + from fastapi import FastAPI + from file_storage.tenant_cache import cached_tenant_exists + from simple_module_core.invalidation import InvalidationBus + + app = FastAPI() + bus = InvalidationBus() + from file_storage.module import FileStorageModule + + FileStorageModule().register_invalidations(bus, app) + lookup = AsyncMock(return_value=True) + app.state.tenant_exists = lookup + assert await cached_tenant_exists(app, "tenant") is True + assert await cached_tenant_exists(app, "tenant") is True + assert lookup.await_count == 1 + lookup.return_value = False + await bus.publish(TENANT_DELETED_CHANNEL, key="tenant") + assert await cached_tenant_exists(app, "tenant") is False + assert lookup.await_count == 2 + monkeypatch.setattr("file_storage.tenant_cache.TENANT_EXISTS_TTL_SECONDS", 0) + assert await cached_tenant_exists(app, "tenant") is False + assert lookup.await_count == 3 diff --git a/modules/file_storage/tests/test_file_storage_public.py b/modules/file_storage/tests/test_file_storage_public.py index fcaa57c30..25bc3160c 100644 --- a/modules/file_storage/tests/test_file_storage_public.py +++ b/modules/file_storage/tests/test_file_storage_public.py @@ -107,9 +107,12 @@ async def test_cross_tenant_public_file_served_but_private_not(app, client): by id (it is public by its owner's choice); a private one never does.""" from file_storage.models import StoredFile from simple_module_db import tenant_context + from tenants.models import Tenant ids = {} async with app.state.sm.db.session_factory() as session: + session.add(Tenant(id="acme", slug="acme", name="Acme")) + await session.flush() with tenant_context("acme"): for name, public in (("pub", True), ("priv", False)): row = StoredFile( diff --git a/modules/file_storage/tests/test_file_storage_single_tenant_public.py b/modules/file_storage/tests/test_file_storage_single_tenant_public.py new file mode 100644 index 000000000..293ab8d8a --- /dev/null +++ b/modules/file_storage/tests/test_file_storage_single_tenant_public.py @@ -0,0 +1,32 @@ +"""A single-tenant install keeps serving public files with the tenants plugin installed.""" + +import io + +import pytest +from file_storage import constants +from PIL import Image +from simple_module_hosting.settings import Settings + + +@pytest.fixture +def settings(settings: Settings) -> Settings: + return settings.model_copy(update={"multi_tenant": False}) + + +async def test_file_storage_public_serves_fallback_tenant_files(authenticated_client, client): + content = io.BytesIO() + Image.new("RGB", (20, 20)).save(content, format="PNG") + response = await authenticated_client.post( + f"{constants.ROUTE_PREFIX_API}{constants.PATH_UPLOAD}", + files={"file": ("image.png", content.getvalue(), "image/png")}, + data={"public": "true"}, + ) + assert response.status_code == 201, response.text + file = response.json() + # The fallback tenant has no registry row; that must not 404 the file. + for url in ( + file["public_url"], + f"{constants.ROUTE_PREFIX_API}/public/{file['id']}", + f"{constants.ROUTE_PREFIX_API}/public/{file['id']}/thumbnail?w=64", + ): + assert (await client.get(url)).status_code == 200, url diff --git a/modules/file_storage/tests/test_file_storage_tenant_deletion.py b/modules/file_storage/tests/test_file_storage_tenant_deletion.py new file mode 100644 index 000000000..d54549495 --- /dev/null +++ b/modules/file_storage/tests/test_file_storage_tenant_deletion.py @@ -0,0 +1,257 @@ +"""Tenant teardown preserves a retryable manifest after deleting source rows.""" + +from datetime import UTC, datetime, timedelta +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +from background_tasks.constants import OutboxStatus +from background_tasks.models import OutboxMessage +from background_tasks.outbox_claim import OutboxPolicy +from background_tasks.outbox_dispatch import dispatch +from background_tasks.outbox_transport import WorkerOutboxTransport +from file_storage import constants +from file_storage.backends.filesystem import FilesystemBackend +from file_storage.models import StoredFile, UploadSession +from file_storage.tenant_deletion import register_tenant_deletion +from file_storage.thumbnails import variant_key +from simple_module_core.events import EventBus +from simple_module_core.invalidation import InvalidationBus +from simple_module_core.tenant_events import TenantDeleted +from simple_module_db import all_tenants, tenant_context +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker +from sqlmodel import select + + +async def test_tenant_cleanup_commits_manifest_and_retries(app, tmp_path, monkeypatch): + bus = EventBus() + transport = WorkerOutboxTransport() + monkeypatch.setattr(transport, "schedule", lambda db, ids: None) + bus.durable_transport = transport + settings = app.state.file_storage.settings.model_copy(update={"fs_root_path": str(tmp_path)}) + backend = FilesystemBackend(tmp_path) + monkeypatch.setattr("file_storage.tenant_deletion.build_backend", lambda settings: backend) + register_tenant_deletion( + bus, + SimpleNamespace( + state=SimpleNamespace( + file_storage=SimpleNamespace(settings=settings), + sm=SimpleNamespace(invalidation=InvalidationBus()), + ) + ), + ) + tenant_id = "deleted-storage-tenant" + key = f"{tenant_id}/original.txt" + variant = variant_key(key, constants.THUMBNAIL_WIDTHS[0]) + + async def body(): + yield b"bytes" + + for target in (key, variant): + await backend.put(target, body(), content_type="text/plain", size=5) + upload_id = await backend.create_multipart(f"{tenant_id}/partial", "text/plain") + factory = async_sessionmaker( + app.state.sm.db.engine, class_=AsyncSession, expire_on_commit=False + ) + async with factory() as db: + assert not hasattr(db, "on_commit") + with tenant_context(tenant_id): + async with factory() as db: + db.add( + StoredFile( + tenant_id=tenant_id, + key=key, + filename="original.txt", + content_type="text/plain", + size_bytes=5, + backend=constants.BackendId.FILESYSTEM, + public=True, + is_deleted=True, + ) + ) + db.add( + UploadSession( + tenant_id=tenant_id, + key=f"{tenant_id}/partial", + filename="partial", + content_type="text/plain", + total_size=5, + part_size=5, + backend_upload_id=upload_id, + expires_at=datetime.now(UTC) + timedelta(hours=1), + ) + ) + await bus.publish_durable(TenantDeleted(tenant_id, "deleted", "Deleted"), db) + await db.commit() + with tenant_context("retained-storage-tenant"): + async with factory() as db: + db.add( + StoredFile( + tenant_id="retained-storage-tenant", + key="retained-storage-tenant/keep.txt", + filename="keep.txt", + content_type="text/plain", + size_bytes=5, + backend=constants.BackendId.FILESYSTEM, + ) + ) + await db.commit() + policy = OutboxPolicy(max_attempts=8, lease_seconds=300) + assert (await dispatch(factory, bus, policy, limit=1)).delivered == 1 + assert await backend.exists(key) + with all_tenants(): + async with factory() as db: + remaining = ( + (await db.execute(select(StoredFile).execution_options(include_deleted=True))) + .scalars() + .all() + ) + assert [row.tenant_id for row in remaining] == ["retained-storage-tenant"] + assert not (await db.execute(select(UploadSession))).scalars().all() + original_delete = backend.delete + monkeypatch.setattr(backend, "delete", AsyncMock(side_effect=RuntimeError("offline"))) + assert (await dispatch(factory, bus, policy)).retrying == 2 + async with factory() as db: + rows = (await db.execute(select(OutboxMessage))).scalars().all() + assert sum(row.status == OutboxStatus.PENDING for row in rows) == 2 + monkeypatch.setattr(backend, "delete", original_delete) + assert ( + await dispatch(factory, bus, policy, now=datetime.now(UTC) + timedelta(minutes=1)) + ).delivered == 2 + assert not await backend.exists(key) + assert not await backend.exists(variant) + assert not backend._parts_dir(upload_id).exists() + assert (await dispatch(factory, bus, policy)).delivered == 0 + + +@pytest.mark.parametrize("rollback", [False, True]) +async def test_inline_cleanup_defers_objects_until_commit( + db_session, tmp_path, monkeypatch, rollback +): + from simple_module_db.transaction import finalize_session, rollback_session + from tenants.models import Tenant + from tenants.service import TenantService + + bus = EventBus() + backend = FilesystemBackend(tmp_path) + from file_storage.settings import FileStorageSettings + + settings = FileStorageSettings() + monkeypatch.setattr("file_storage.tenant_deletion.build_backend", lambda settings: backend) + register_tenant_deletion( + bus, + SimpleNamespace( + state=SimpleNamespace( + file_storage=SimpleNamespace(settings=settings), + sm=SimpleNamespace(invalidation=InvalidationBus()), + ) + ), + ) + tenant_id = "inline-storage" + key = f"{tenant_id}/original.txt" + variant = variant_key(key, constants.THUMBNAIL_WIDTHS[0]) + + async def body(): + yield b"bytes" + + for target in (key, variant): + await backend.put(target, body(), content_type="text/plain", size=5) + upload_key = f"{tenant_id}/partial" + upload_id = await backend.create_multipart(upload_key, "text/plain") + db_session.add(Tenant(id=tenant_id, slug=tenant_id, name=tenant_id)) + with tenant_context(tenant_id): + db_session.add( + StoredFile( + key=key, + filename="original.txt", + content_type="text/plain", + size_bytes=5, + backend=constants.BackendId.FILESYSTEM, + is_deleted=True, + ) + ) + db_session.add( + UploadSession( + key=upload_key, + filename="partial", + content_type="text/plain", + total_size=5, + part_size=5, + backend_upload_id=upload_id, + expires_at=datetime.now(UTC) + timedelta(hours=1), + ) + ) + await finalize_session(db_session) + await TenantService(db_session, bus=bus).delete_tenant(tenant_id) + for model in (StoredFile, UploadSession): + assert ( + not ( + await db_session.execute(select(model).execution_options(include_deleted=True)) + ) + .scalars() + .all() + ) + assert await backend.exists(key) + assert await backend.exists(variant) + assert backend._parts_dir(upload_id).exists() + if rollback: + await rollback_session(db_session) + else: + await finalize_session(db_session) + assert await backend.exists(key) == rollback + assert await backend.exists(variant) == rollback + assert backend._parts_dir(upload_id).exists() == rollback + for model in (StoredFile, UploadSession): + rows = ( + (await db_session.execute(select(model).execution_options(include_deleted=True))) + .scalars() + .all() + ) + assert len(rows) == int(rollback) + + +async def test_cleanup_skips_upload_without_backend_id(db_session, monkeypatch): + from file_storage.tenant_deletion import TenantStorageCleanupRequested + + bus = EventBus() + backend = AsyncMock() + monkeypatch.setattr("file_storage.tenant_deletion.build_backend", lambda settings: backend) + register_tenant_deletion(bus, None) + monkeypatch.setattr("file_storage.tenant_deletion._settings", lambda app: SimpleNamespace()) + await bus.publish_durable_or_inline( + TenantStorageCleanupRequested(files=[], uploads=[("pending", None), ("active", "id")]), + db_session, + ) + backend.abort_multipart.assert_awaited_once_with("active", "id") + backend.delete.assert_awaited_once_with("active") + + +async def test_cleanup_retries_transient_abort_after_all_attempts(monkeypatch): + from file_storage.tenant_deletion import TenantStorageCleanupRequested + + bus = EventBus() + backend = AsyncMock() + backend.abort_multipart.side_effect = [RuntimeError("offline"), None, None, None] + monkeypatch.setattr("file_storage.tenant_deletion.build_backend", lambda settings: backend) + monkeypatch.setattr( + "file_storage.tenant_deletion._settings", + lambda app: SimpleNamespace(backend="filesystem", model_copy=lambda **kw: None), + ) + register_tenant_deletion(bus, None) + db_session = SimpleNamespace() + manifest = TenantStorageCleanupRequested( + files=[("filesystem", "original")], uploads=[("partial", "id"), ("other", "id2")] + ) + with pytest.raises(ExceptionGroup, match="Tenant storage cleanup failed"): + await bus.publish_durable_or_inline(manifest, db_session) + expected = [ + "original", + *(variant_key("original", w) for w in constants.THUMBNAIL_WIDTHS), + "partial", + "other", + ] + assert [call.args[0] for call in backend.delete.await_args_list] == expected + assert backend.abort_multipart.await_count == 2 + await bus.publish_durable_or_inline(manifest, db_session) + assert backend.abort_multipart.await_count == 4 + assert backend.delete.await_count == 2 * len(expected) diff --git a/modules/file_storage/tests/test_file_storage_upload_tasks.py b/modules/file_storage/tests/test_file_storage_upload_tasks.py index 55f17dd7b..6580f40e6 100644 --- a/modules/file_storage/tests/test_file_storage_upload_tasks.py +++ b/modules/file_storage/tests/test_file_storage_upload_tasks.py @@ -212,3 +212,28 @@ def test_the_reaper_purges_finished_rows_past_the_retention_window(worker): with sync_db.sync_session() as db, all_tenants(): left = {s.backend_upload_id: s.status for s in db.execute(select(UploadSession)).scalars()} assert left == {"c": "aborted", rows[3].backend_upload_id: "aborted"} + + +def test_checksum_refuses_previous_provider(worker, monkeypatch): + from file_storage.backend_guard import BackendUnavailableError + + file_id = uuid.uuid4() + with sync_db.sync_session() as db, tenant_context(TENANT): + db.add( + StoredFile( + id=file_id, + tenant_id=TENANT, + key="old", + filename="old", + content_type="text/plain", + size_bytes=1, + backend=constants.BackendId.S3, + ) + ) + + async def wrong_store(key): + pytest.fail("read from active provider") + + monkeypatch.setattr(worker, "get", wrong_store) + with pytest.raises(BackendUnavailableError): + tasks.compute_checksum(str(file_id), TENANT) diff --git a/modules/file_storage/tests/test_service.py b/modules/file_storage/tests/test_service.py index 51553ff4a..1ec706352 100644 --- a/modules/file_storage/tests/test_service.py +++ b/modules/file_storage/tests/test_service.py @@ -169,7 +169,8 @@ async def test_download_dispatches_on_backend_capability(tmp_path, db_session: A # Swap in a presigning backend; download should now redirect. class _PresigningBackend: - backend_id = "presign-test" + # Keep ownership the same while changing the download capability. + backend_id = backend.backend_id supports_presigned_url = True async def put(self, key, stream, **kwargs): ... diff --git a/modules/sharing/sharing/module.py b/modules/sharing/sharing/module.py index 3577ebf17..c1c8c4e94 100644 --- a/modules/sharing/sharing/module.py +++ b/modules/sharing/sharing/module.py @@ -17,6 +17,7 @@ if TYPE_CHECKING: from fastapi import APIRouter, FastAPI from simple_module_core.audit_links import AuditLinkRegistry + from simple_module_core.events import EventBus from simple_module_core.permissions import PermissionRegistry @@ -27,6 +28,13 @@ class SharingModule(ModuleBase): view_prefix=c.VIEW_PREFIX, ) + def register_event_handlers(self, bus: EventBus, app: FastAPI | None = None) -> None: + from simple_module_core.tenant_events import TenantDeleted + + from sharing.tenant_cleanup import purge_tenant_data + + bus.subscribe_durable(TenantDeleted, purge_tenant_data) + def register_settings(self, app: FastAPI) -> None: from sharing.services import SharingServices from sharing.settings import SharingSettings diff --git a/modules/sharing/sharing/tenant_cleanup.py b/modules/sharing/sharing/tenant_cleanup.py new file mode 100644 index 000000000..05bfa763a --- /dev/null +++ b/modules/sharing/sharing/tenant_cleanup.py @@ -0,0 +1,16 @@ +"""Durable teardown of tenant-owned sharing rows.""" + +from simple_module_core.durable_events import DurableContext +from simple_module_core.tenant_events import TenantDeleted +from simple_module_db.tenancy import tenant_context +from sqlmodel import delete + +from sharing.models import ResourceShare + + +async def purge_tenant_data(event: TenantDeleted, ctx: DurableContext) -> None: + """Hard-delete tenant rows atomically with outbox acknowledgement.""" + with tenant_context(event.tenant_id): + await ctx.db.execute( + delete(ResourceShare).where(ResourceShare.tenant_id == event.tenant_id) + ) diff --git a/modules/sharing/tests/test_sharing_tenant_cleanup.py b/modules/sharing/tests/test_sharing_tenant_cleanup.py new file mode 100644 index 000000000..8af2ed545 --- /dev/null +++ b/modules/sharing/tests/test_sharing_tenant_cleanup.py @@ -0,0 +1,43 @@ +"""Tenant deletion purges only the deleted tenant through a durable handler.""" + +from sharing.models import ResourceShare +from sharing.module import SharingModule +from simple_module_core.durable_events import DurableContext, event_key +from simple_module_core.events import EventBus +from simple_module_core.tenant_events import TenantDeleted +from simple_module_db.tenancy import all_tenants, tenant_context +from sqlmodel import select + + +async def test_durable_tenant_cleanup_is_scoped_idempotent_and_transactional(db_session): + bus = EventBus() + SharingModule().register_event_handlers(bus) + names = bus.durable.handler_names(TenantDeleted) + assert len(names) == 1 + subscription = bus.durable.lookup(event_key(TenantDeleted), names[0]) + assert subscription is not None + for tenant_id in ("deleted-tenant", "other-tenant"): + with tenant_context(tenant_id): + db_session.add( + ResourceShare( + resource_type="test.note", + resource_id="note", + subject_id="user", + role="view", + granted_by="owner", + ) + ) + await db_session.flush() + await db_session.commit() + event = TenantDeleted(tenant_id="deleted-tenant", slug="deleted", name="Deleted") + ctx = DurableContext(db=db_session, attempt=1) + # A worker rollback must restore rows: the handler must never commit its session. + await subscription.handler(event, ctx) + await db_session.rollback() + with all_tenants(): + assert len((await db_session.execute(select(ResourceShare))).scalars().all()) == 2 + await subscription.handler(event, ctx) + await subscription.handler(event, DurableContext(db=db_session, attempt=2)) + with all_tenants(): + remaining = (await db_session.execute(select(ResourceShare))).scalars().all() + assert [row.tenant_id for row in remaining] == ["other-tenant"] diff --git a/modules/tenants/tenants/contracts/events.py b/modules/tenants/tenants/contracts/events.py index 5d9a333f8..2d3ae6eb6 100644 --- a/modules/tenants/tenants/contracts/events.py +++ b/modules/tenants/tenants/contracts/events.py @@ -12,6 +12,7 @@ from dataclasses import dataclass from simple_module_core.events import Event +from simple_module_core.tenant_events import TenantDeleted as TenantDeleted @dataclass @@ -29,20 +30,6 @@ class TenantStatusChanged(Event): previous: str -@dataclass -class TenantDeleted(Event): - """The tenant and its memberships/invitations are gone. - - Modules keeping per-tenant state of their own (rows, external - subscriptions) clean it up here. Carries the slug and name because the - tenant row no longer exists to look them up. - """ - - tenant_id: str - slug: str - name: str - - @dataclass class MembershipAdded(Event): tenant_id: str diff --git a/modules/tenants/tenants/deps.py b/modules/tenants/tenants/deps.py index 1a9d16332..c66c23218 100644 --- a/modules/tenants/tenants/deps.py +++ b/modules/tenants/tenants/deps.py @@ -22,6 +22,7 @@ async def get_tenant_service(request: Request, db: AsyncSession = Depends(get_db db, bus=app.state.sm.event_bus, invalidate=make_invalidator(app), + invalidation_bus=app.state.sm.invalidation, entitlements=app.state.tenants.entitlements, ) diff --git a/modules/tenants/tenants/service.py b/modules/tenants/tenants/service.py index 0ed35869d..007e6700d 100644 --- a/modules/tenants/tenants/service.py +++ b/modules/tenants/tenants/service.py @@ -10,12 +10,15 @@ from __future__ import annotations +import logging import re import secrets from collections.abc import Awaitable, Callable, Sequence from typing import Any from simple_module_core.events import Event, EventBus +from simple_module_core.invalidation import InvalidationBus +from simple_module_core.tenant_events import TENANT_DELETED_CHANNEL from simple_module_db import PLATFORM_TENANT_ID from sqlalchemy import func, select from sqlalchemy.exc import IntegrityError @@ -31,12 +34,15 @@ from tenants.contracts.events import ( MembershipAdded, TenantCreated, + TenantDeleted, TenantStatusChanged, ) from tenants.contracts.schemas import TenantCreate from tenants.errors import TenantError from tenants.models import Membership, Tenant +logger = logging.getLogger(__name__) + Invalidate = Callable[[str | None], Awaitable[None]] @@ -57,10 +63,12 @@ def __init__( bus: EventBus | None = None, invalidate: Invalidate = _no_invalidate, entitlements: EntitlementProvider | None = None, + invalidation_bus: InvalidationBus | None = None, ) -> None: self.db = db self.bus = bus self.invalidate = invalidate + self.invalidation_bus = invalidation_bus self.entitlements = entitlements or UnlimitedEntitlements() self.pending: list[Callable[[], Awaitable[Any]]] = [] @@ -143,7 +151,20 @@ async def set_status(self, tenant_id: str, status: TenantStatus) -> Tenant: async def delete_tenant(self, tenant_id: str) -> None: tenant = await self.lock(tenant_id) - self._after_commit(*await delete_tenant_rows(self.db, tenant), invalidate=[None]) + events = await delete_tenant_rows(self.db, tenant) + deleted = next(event for event in events if isinstance(event, TenantDeleted)) + if self.bus is not None: + await self.bus.publish_durable_or_inline(deleted, self.db) + else: + logger.warning( + "Tenant-owned data cleanup was skipped tenant=%s: no event bus", tenant_id + ) + + async def invalidate_deleted() -> None: + if self.invalidation_bus is not None: + await self.invalidation_bus.publish(TENANT_DELETED_CHANNEL, key=tenant_id) + + self._after_commit(*events, invalidate=[None], callback=invalidate_deleted) async def list_all(self, *, search: str = "", limit: int = 50, offset: int = 0) -> list[Tenant]: stmt = select(Tenant).order_by(Tenant.created_at.desc()).limit(limit).offset(offset) diff --git a/modules/tenants/tests/tenants_deletion_helpers.py b/modules/tenants/tests/tenants_deletion_helpers.py new file mode 100644 index 000000000..c2264fe64 --- /dev/null +++ b/modules/tenants/tests/tenants_deletion_helpers.py @@ -0,0 +1,15 @@ +"""Outbox writer with dispatch deliberately paused for deletion tests.""" + +from background_tasks.outbox_transport import _OutboxWriter +from simple_module_core.events import EventBus + + +class PausedTransport(_OutboxWriter): + def schedule(self, db, ids) -> None: + pass + + +def deletion_bus() -> EventBus: + bus = EventBus() + bus.durable_transport = PausedTransport() + return bus diff --git a/modules/tenants/tests/test_tenant_deleted_event.py b/modules/tenants/tests/test_tenant_deleted_event.py index cbaff683f..66d3e5a7e 100644 --- a/modules/tenants/tests/test_tenant_deleted_event.py +++ b/modules/tenants/tests/test_tenant_deleted_event.py @@ -12,6 +12,7 @@ from tenants.contracts.schemas import TenantCreate from tenants.models import Membership, Tenant from tenants.service import TenantService +from tenants_deletion_helpers import deletion_bus async def _tenant_with_owner(db_session, bus: EventBus) -> tuple[TenantService, Tenant]: @@ -22,7 +23,7 @@ async def _tenant_with_owner(db_session, bus: EventBus) -> tuple[TenantService, async def test_delete_publishes_after_commit(db_session): - bus = EventBus() + bus = deletion_bus() seen: list[TenantDeleted] = [] async def handler(event: TenantDeleted) -> None: @@ -43,7 +44,7 @@ async def handler(event: TenantDeleted) -> None: async def test_rolled_back_delete_publishes_nothing(db_session): - bus = EventBus() + bus = deletion_bus() seen: list[TenantDeleted] = [] async def handler(event: TenantDeleted) -> None: @@ -82,7 +83,7 @@ async def invalidate(key: str | None) -> None: async def test_delete_removes_every_member_after_commit(db_session): - bus = EventBus() + bus = deletion_bus() removed: list[MembershipRemoved] = [] async def handler(event: MembershipRemoved) -> None: @@ -107,7 +108,7 @@ async def handler(event: MembershipRemoved) -> None: async def test_rolled_back_delete_removes_no_member(db_session): - bus = EventBus() + bus = deletion_bus() removed: list[MembershipRemoved] = [] async def handler(event: MembershipRemoved) -> None: @@ -124,3 +125,28 @@ async def handler(event: MembershipRemoved) -> None: assert resolver.cached_memberships("u2") is not None members = await db_session.scalar(select(func.count()).select_from(Membership)) assert members == 2 + + +@pytest.mark.parametrize("rollback", [False, True]) +async def test_delete_invalidates_tenant_channel_after_commit(db_session, app, rollback): + from simple_module_core.tenant_events import TENANT_DELETED_CHANNEL + from starlette.requests import Request + from tenants.deps import get_tenant_service + + bus = app.state.sm.invalidation + seen = [] + bus.subscribe(TENANT_DELETED_CHANNEL, seen.append) + tenant = Tenant(id="invalidate-delete", slug="invalidate-delete", name="Delete") + db_session.add(tenant) + await finalize_session(db_session) + service = await get_tenant_service(Request({"type": "http", "app": app}), db_session) + await service.delete_tenant(tenant.id) + assert seen == [] + if rollback: + await rollback_session(db_session) + assert seen == [] + else: + await finalize_session(db_session) + assert len(seen) == 1 + assert seen[0].key == tenant.id + assert await db_session.get(Tenant, tenant.id) is None diff --git a/modules/tenants/tests/test_tenants_durable_deletion.py b/modules/tenants/tests/test_tenants_durable_deletion.py new file mode 100644 index 000000000..9af42bff8 --- /dev/null +++ b/modules/tenants/tests/test_tenants_durable_deletion.py @@ -0,0 +1,52 @@ +"""Deleting the tenant and enqueueing its teardown are one transaction.""" + +import pytest +from background_tasks.models import OutboxMessage +from simple_module_core.durable_events import event_key +from simple_module_core.events import EventBus +from simple_module_core.tenant_events import TenantDeleted +from simple_module_db.transaction import finalize_session, rollback_session +from sqlalchemy import select +from tenants.contracts.events import TenantDeleted as ProviderTenantDeleted +from tenants.models import Tenant +from tenants.service import TenantService +from tenants_deletion_helpers import deletion_bus + + +async def _cleanup(event, ctx): + pass + + +async def test_tenants_delete_enqueues_before_commit(db_session): + bus = deletion_bus() + bus.subscribe_durable(TenantDeleted, _cleanup, name="test.cleanup") + tenant = Tenant(id="durable", slug="durable", name="Durable") + db_session.add(tenant) + await finalize_session(db_session) + + await TenantService(db_session, bus=bus).delete_tenant(tenant.id) + rows = (await db_session.execute(select(OutboxMessage))).scalars().all() + assert len(rows) == 1 + assert rows[0].payload == {"tenant_id": "durable", "slug": "durable", "name": "Durable"} + assert rows[0].event_key == event_key(TenantDeleted) + assert rows[0].handler_name == "test.cleanup" + assert ProviderTenantDeleted is TenantDeleted + + await rollback_session(db_session) + assert (await db_session.execute(select(OutboxMessage))).scalars().all() == [] + assert await db_session.get(Tenant, "durable") is not None + + +@pytest.mark.parametrize("bus", [None, EventBus()], ids=["no-bus", "no-transport"]) +async def test_tenants_delete_without_durable_transport(db_session, bus, caplog): + tenant = Tenant(id="durable", slug="durable", name="Durable") + db_session.add(tenant) + await finalize_session(db_session) + await TenantService(db_session, bus=bus).delete_tenant(tenant.id) + assert await db_session.get(Tenant, "durable") is None + await rollback_session(db_session) + assert await db_session.get(Tenant, "durable") is not None + + await TenantService(db_session, bus=bus).delete_tenant("durable") + await finalize_session(db_session) + assert await db_session.get(Tenant, "durable") is None diff --git a/modules/tenants/tests/test_tenants_inline_deletion.py b/modules/tenants/tests/test_tenants_inline_deletion.py new file mode 100644 index 000000000..b12b3d5d3 --- /dev/null +++ b/modules/tenants/tests/test_tenants_inline_deletion.py @@ -0,0 +1,64 @@ +"""Optional modules purge tenant data in the request without background tasks.""" + +from datetime import UTC, datetime, timedelta + +import pytest +from api_tokens.models import ApiToken +from api_tokens.module import ApiTokensModule +from feature_flags.models import FeatureFlagOverride +from feature_flags.module import FeatureFlagsModule +from sharing.models import ResourceShare +from sharing.module import SharingModule +from simple_module_core.events import EventBus +from simple_module_db import all_tenants, tenant_context +from simple_module_db.stored_settings import SCOPE_TENANT, Setting +from simple_module_db.tenant_cleanup import register_tenant_cleanup +from simple_module_db.transaction import finalize_session, rollback_session +from sqlmodel import select +from tenants.models import Tenant +from tenants.service import TenantService + + +@pytest.mark.parametrize("rollback", [False, True]) +async def test_inline_deletion_purges_rows_atomically(db_session, rollback): + bus = EventBus() + for module in (ApiTokensModule(), SharingModule(), FeatureFlagsModule()): + module.register_event_handlers(bus) + register_tenant_cleanup(bus) + for tenant_id in ("deleted", "retained"): + db_session.add(Tenant(id=tenant_id, slug=tenant_id, name=tenant_id)) + with tenant_context(tenant_id): + db_session.add( + ApiToken( + name="token", + prefix="test", + token_hash=tenant_id, + expires_at=datetime.now(UTC) + timedelta(days=1), + ) + ) + db_session.add( + ResourceShare( + resource_type="test.note", + resource_id="note", + subject_id="user", + role="view", + granted_by="owner", + ) + ) + await db_session.flush() + db_session.add(FeatureFlagOverride(scope=SCOPE_TENANT, scope_id=tenant_id, name="flag")) + db_session.add(Setting(scope=SCOPE_TENANT, scope_id=tenant_id, key="key", value="value")) + await finalize_session(db_session) + await TenantService(db_session, bus=bus).delete_tenant("deleted") + with all_tenants(): + for model in (ApiToken, ResourceShare, FeatureFlagOverride, Setting): + assert len((await db_session.execute(select(model))).scalars().all()) == 1 + if rollback: + await rollback_session(db_session) + else: + await finalize_session(db_session) + with all_tenants(): + for model in (Tenant, ApiToken, ResourceShare, FeatureFlagOverride, Setting): + assert len((await db_session.execute(select(model))).scalars().all()) == ( + 2 if rollback else 1 + ) diff --git a/modules/tenants/tests/test_tenants_teardown_integration.py b/modules/tenants/tests/test_tenants_teardown_integration.py new file mode 100644 index 000000000..8649df4b3 --- /dev/null +++ b/modules/tenants/tests/test_tenants_teardown_integration.py @@ -0,0 +1,118 @@ +"""A real tenant deletion delivers every installed owner's durable teardown.""" + +from datetime import UTC, datetime, timedelta + +from api_tokens.models import ApiToken +from audit_log.models import AuditEntry +from background_tasks.outbox_claim import OutboxPolicy +from background_tasks.outbox_dispatch import dispatch +from feature_flags.models import FeatureFlagOverride +from file_storage.backends.filesystem import FilesystemBackend +from file_storage.models import StoredFile +from simple_module_db import all_tenants, tenant_context +from simple_module_db.stored_settings import SCOPE_TENANT, Setting +from sqlmodel import select +from tenants.service import TenantService + + +async def test_tenants_delete_removes_owned_data_but_retains_audit( + app, tenant_client, client, tmp_path, monkeypatch +): + app.state.background_tasks.settings.task_always_eager = True + backend = FilesystemBackend(tmp_path) + app.state.file_storage.backend = backend + monkeypatch.setattr("file_storage.tenant_deletion.build_backend", lambda settings: backend) + async with tenant_client() as owner: + response = await owner.client.post( + "/api/file-storage/upload", + files={"file": ("public.txt", b"tenant bytes", "text/plain")}, + data={"public": "true"}, + ) + assert response.status_code == 201, response.text + public_url = response.json()["public_url"] + assert (await client.get(public_url)).status_code == 200 + factory = app.state.sm.db.session_factory + with tenant_context(owner.tenant_id): + async with factory() as db: + db.add( + ApiToken( + name="tenant credential", + prefix="test", + token_hash="a" * 64, + expires_at=datetime.now(UTC) + timedelta(days=1), + ) + ) + db.add( + Setting( + scope=SCOPE_TENANT, + scope_id=owner.tenant_id, + key="branding.logo_file_id", + value="old-file", + ) + ) + db.add( + FeatureFlagOverride( + scope=SCOPE_TENANT, + scope_id=owner.tenant_id, + name="test.tenant-feature", + enabled=True, + ) + ) + audit = AuditEntry( + entity_type="Tenant", + entity_id=owner.tenant_id, + action="created", + tenant_id=owner.tenant_id, + ) + db.add(audit) + await db.commit() + audit_id = audit.id + async with factory() as db: + await TenantService(db, bus=app.state.sm.event_bus).delete_tenant(owner.tenant_id) + await db.commit() + transport = app.state.sm.event_bus.durable_transport + await transport.drain() + await dispatch( + factory, app.state.sm.event_bus, OutboxPolicy(max_attempts=8, lease_seconds=300) + ) + assert (await client.get(public_url)).status_code == 404 + with all_tenants(): + async with factory() as db: + for model in (StoredFile, ApiToken): + assert ( + not ( + await db.execute( + select(model) + .where(model.tenant_id == owner.tenant_id) + .execution_options(include_deleted=True) + ) + ) + .scalars() + .all() + ) + assert ( + not ( + await db.execute( + select(Setting).where( + Setting.scope == SCOPE_TENANT, Setting.scope_id == owner.tenant_id + ) + ) + ) + .scalars() + .all() + ) + assert ( + not ( + await db.execute( + select(FeatureFlagOverride).where( + FeatureFlagOverride.scope == SCOPE_TENANT, + FeatureFlagOverride.scope_id == owner.tenant_id, + ) + ) + ) + .scalars() + .all() + ) + retained = await db.get(AuditEntry, audit_id) + assert retained is not None and retained.tenant_id == owner.tenant_id + assert not any(path.is_file() for path in tmp_path.rglob("*")) diff --git a/packages/i18n/src/generated-resources.ts b/packages/i18n/src/generated-resources.ts index cf120b1cd..76028a661 100644 --- a/packages/i18n/src/generated-resources.ts +++ b/packages/i18n/src/generated-resources.ts @@ -535,6 +535,7 @@ export default { 'file_storage.delete_dialog.title_one': '', 'file_storage.delete_dialog.title_other': '', 'file_storage.errors.backend_error': '', + 'file_storage.errors.backend_unavailable': '', 'file_storage.errors.bad_image': '', 'file_storage.errors.bad_part': '', 'file_storage.errors.bad_type': '', diff --git a/packages/i18n/src/keys.generated.ts b/packages/i18n/src/keys.generated.ts index f88db1108..3071422f6 100644 --- a/packages/i18n/src/keys.generated.ts +++ b/packages/i18n/src/keys.generated.ts @@ -680,6 +680,7 @@ export const keys = { }, errors: { backend_error: 'file_storage.errors.backend_error', + backend_unavailable: 'file_storage.errors.backend_unavailable', bad_image: 'file_storage.errors.bad_image', bad_part: 'file_storage.errors.bad_part', bad_type: 'file_storage.errors.bad_type',