From a66b229807f9505449323806a7965af3ad3f7350 Mon Sep 17 00:00:00 2001 From: Anto Subash Date: Fri, 30 Oct 2026 23:46:20 +0100 Subject: [PATCH 1/4] feat(file_storage): sync worker API (store_path/open_local/presigned_url), S3 multipart server puts, BIGINT sizes (#708) Claude-Session: https://claude.ai/code/session_01M9neheZZEe3sVpDi2S3zT4 --- CHANGELOG.md | 10 + CLAUDE.md | 2 +- docs/database/migrations.md | 2 +- docs/modules/file_storage.md | 62 +++++ ...c3e8a207_file_storage_size_bytes_bigint.py | 41 ++++ host/tests/test_migration_graph.py | 5 +- modules/file_storage/README.md | 16 ++ .../file_storage/file_storage/backends/s3.py | 29 +-- .../file_storage/backends/s3_put.py | 121 ++++++++++ .../file_storage/file_storage/constants.py | 1 + .../file_storage/contracts/__init__.py | 38 +++ .../file_storage/contracts/service.py | 2 +- modules/file_storage/file_storage/models.py | 3 +- modules/file_storage/file_storage/service.py | 4 +- modules/file_storage/file_storage/tasks.py | 12 +- modules/file_storage/file_storage/worker.py | 166 ++++++++++++++ .../file_storage/worker_support.py | 13 ++ modules/file_storage/tests/conftest.py | 1 + .../tests/file_storage_worker_helpers.py | 40 ++++ .../tests/test_file_storage_server_put.py | 208 +++++++++++++++++ .../tests/test_file_storage_worker.py | 217 ++++++++++++++++++ .../test_file_storage_worker_failures.py | 178 ++++++++++++++ 22 files changed, 1137 insertions(+), 34 deletions(-) create mode 100644 host/migrations/versions/f5b1c3e8a207_file_storage_size_bytes_bigint.py create mode 100644 modules/file_storage/file_storage/backends/s3_put.py create mode 100644 modules/file_storage/file_storage/worker.py create mode 100644 modules/file_storage/file_storage/worker_support.py create mode 100644 modules/file_storage/tests/file_storage_worker_helpers.py create mode 100644 modules/file_storage/tests/test_file_storage_server_put.py create mode 100644 modules/file_storage/tests/test_file_storage_worker.py create mode 100644 modules/file_storage/tests/test_file_storage_worker_failures.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 9579e4bcb..60e070d75 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,11 @@ All notable changes to this project are documented in this file. The format is b including delegated share management: a role implying `ACTION_SHARE` may list, grant and revoke shares on that resource. Default view/edit roles remain unchanged; no migration (#707). +- Tenant-scoped sync file worker contracts (`store_path`, `open_local`, and + `presigned_url`), reusing worker settings/pools, upload validation and hashing, + with commit compensation and `FileUploaded` publication after commit (#708). +- Bounded S3 server-side multipart uploads above the stored threshold, including + unknown-length streams, size/part bounds and abort on failure (#708). - Users auth emails support ordered host template directories through the restart-required DB setting `mail_template_dirs`, with per-file bundled fallback and optional autoescaped HTML alternatives (#712). @@ -43,6 +48,11 @@ All notable changes to this project are documented in this file. The format is b most 32 characters now fail at registration (#707). ### Fixed +- `file_storage_stored_file.size_bytes` is now `BIGINT` (migration `f5b1c3e8a207`, + which also gives `file_storage` its own branch label): on PostgreSQL it was a + 32-bit `INTEGER`, so a file of 2 GiB or more failed on insert (#708). +- A cancelled one-shot upload's metadata write now also deletes the uploaded + object (#708). - **Behaviour change:** bare UI borders now use the theme border colour in light and dark themes, including pseudo-elements. Use `border-current` to retain the previous foreground-coloured look; explicit border colour utilities still win (#714). diff --git a/CLAUDE.md b/CLAUDE.md index 265690d07..3a8147407 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -313,7 +313,7 @@ development and testing (`session_cookie_secure`). four roots (`77162e7b184b` initial schema, `e36ab09c4a92` tenants, `f733c7938fd1` sharing, `d263ea727663` the framework's `sm_lease`) and six heads (`audit_log`, `tenants`, `keycloak`, `sharing`, `leases`, and the users - trunk, which Alembic reports as `api_tokens` and, since #536, + trunk, which Alembic reports as `api_tokens`, `users`, `file_storage` and, since #536, `background_tasks`). Apply with `upgrade heads` (what `make migrate` does); plain `revision --autogenerate` refuses with "Multiple heads", so name the branch: diff --git a/docs/database/migrations.md b/docs/database/migrations.md index 096f18cfd..5765f8bab 100644 --- a/docs/database/migrations.md +++ b/docs/database/migrations.md @@ -21,7 +21,7 @@ The history is **not** a single linear chain. It has several roots and several h | Head | Owning module | Label Alembic reports | |---|---|---| -| `b31a70971001` | background_tasks (resource references) | `api_tokens`, `background_tasks`, `users` | +| `f5b1c3e8a207` | file_storage (`size_bytes` BIGINT) | `api_tokens`, `background_tasks`, `file_storage`, `users` | | `a7c2e91d4b58` | audit_log | `audit_log` | | `f1a9c3d7b2e4` | tenants | `tenants` | | `d263ea727663` | framework (`sm_lease`) | `leases` | diff --git a/docs/modules/file_storage.md b/docs/modules/file_storage.md index 3e20fb918..7249e742b 100644 --- a/docs/modules/file_storage.md +++ b/docs/modules/file_storage.md @@ -285,6 +285,68 @@ Any task that touches `StoredFile` outside a request must run under `tenant_context(row.tenant_id)` for per-tenant work, or `all_tenants()` for a deliberate platform-wide sweep. +## Synchronous worker API + +Install the `[tasks]` extra and configure the background-tasks worker normally. +`file_storage.contracts` lazily exports `store_path`, `open_local`, and +`presigned_url` for synchronous task bodies (#708): + +```python +from simple_module_db import tenant_context +from file_storage.contracts import open_local, presigned_url, store_path + +with tenant_context(tenant_id): + stored = store_path("output.tif", content_type="image/tiff") + with open_local(stored.id) as local_path: + process_raster(local_path) + url = presigned_url(stored.id, ttl_seconds=300) # S3-compatible backends +``` + +`store_path(source, *, filename=None, content_type="application/octet-stream", +public=False)` accepts a string/path-like path or binary stream and returns +`StoredFileOut`. Paths default to their basename; streams use their `name` +basename or require an explicit filename. Streams are read from their current +position and remain open. The API reuses MIME, size, quota and SHA-256 checks, +with an independent transaction. It returns after commit and publishes +`FileUploaded` on the worker event bus only after a successful commit. Flush or +commit failures attempt to delete the uploaded object; an uncertain commit +outcome is checked in a fresh session first. If that check cannot reach the DB, +the object is preserved for reconciliation. Cleanup failures are logged without +replacing the original error. Settings are hydrated from the DB +through the existing worker cache (normally refreshed within 30 seconds), and +calls reuse the worker thread's event loop and pooled engine. + +All three operations inherit the caller's tenant context. A strict multi-tenant +install refuses unscoped access; missing and cross-tenant IDs behave alike. +Reads reload metadata by ID and refuse a file recorded against another backend. +`open_local(file_id)` closes the metadata transaction before downloading, yields +a path inside a private temporary directory, and removes the directory on +normal exit, download failure or a consumer exception. Budget temporary disk +space for the entire downloaded object. `presigned_url(file_id, *, +ttl_seconds=None)` uses `s3_presign_ttl_seconds` unless overridden with a positive +TTL; unsupported backends (including filesystem) raise `NotSupportedError`. +These sync functions are for task bodies, not async handlers running on the +worker loop itself. + +S3 server-side `put` switches to multipart above +`direct_upload_threshold_bytes`, even for streams whose size is unknown. It +spools only the threshold prefix and one part at a time (each spills to disk +above 10 MiB). Part size comes from `multipart_part_size_bytes`, grown for known +sizes to fit 10,000 parts; unknown-size streams abort if they exceed that bound. +Declared nonzero sizes must match the bytes read. Multipart is also used above +S3's single-PUT limit, even if the threshold was configured higher. Stream, +part-upload and completion failures, including cancellation, attempt abort +without masking the failure. Worker uploads still honor `max_file_size_bytes` +(default 100 MiB): raise that existing DB limit for multi-GB outputs. +`StoredFile.size_bytes` is a `BIGINT` since migration `f5b1c3e8a207` (it was a +32-bit `INTEGER` on PostgreSQL, which refused files of 2 GiB or more), and the +S3 backend's `put` goes multipart, so objects beyond 5 GiB work end to end. + +A process killed mid-upload cannot run compensation. Its incomplete multipart +upload is an orphan like an abandoned browser one, and the upload reaper aborts +it once it is older than twice `upload_session_ttl_seconds` (48 hours by +default), which no real put runs for. + ## Settings DB-backed via `register_module_settings`; pydantic defaults seed at boot. Live edits under Files at `/admin/settings/`. diff --git a/host/migrations/versions/f5b1c3e8a207_file_storage_size_bytes_bigint.py b/host/migrations/versions/f5b1c3e8a207_file_storage_size_bytes_bigint.py new file mode 100644 index 000000000..171b7f654 --- /dev/null +++ b/host/migrations/versions/f5b1c3e8a207_file_storage_size_bytes_bigint.py @@ -0,0 +1,41 @@ +"""file_storage_stored_file.size_bytes: BIGINT (#708) + +``size_bytes`` was a 32-bit ``INTEGER`` on Postgres, so storing a file of +2 GiB or more failed on insert. Server-side puts now go multipart above +5 GiB (#708), so the column must hold any object size. SQLite's INTEGER is +already 64-bit; the batch alter is a no-op rebuild there. + +file_storage had no label of its own (its table came in with the initial +schema); this revision gives it one. + +Revision ID: f5b1c3e8a207 +Revises: b31a70971001 +Create Date: 2026-10-30 18:00:00.000000 +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "f5b1c3e8a207" +down_revision: str | None = "b31a70971001" +branch_labels: str | Sequence[str] | None = ("file_storage",) +depends_on: str | Sequence[str] | None = None + +_TABLE = "file_storage_stored_file" + + +def upgrade() -> None: + with op.batch_alter_table(_TABLE) as batch: + batch.alter_column( + "size_bytes", existing_type=sa.Integer(), type_=sa.BigInteger(), existing_nullable=False + ) + + +def downgrade() -> None: + # Fails on Postgres if a stored file is 2 GiB or larger; that is the point. + with op.batch_alter_table(_TABLE) as batch: + batch.alter_column( + "size_bytes", existing_type=sa.BigInteger(), type_=sa.Integer(), existing_nullable=False + ) diff --git a/host/tests/test_migration_graph.py b/host/tests/test_migration_graph.py index 5c0598042..9bac2ad1f 100644 --- a/host/tests/test_migration_graph.py +++ b/host/tests/test_migration_graph.py @@ -34,6 +34,7 @@ "a659221c464b": "file_storage", "e5a8c1f27b94": "file_storage", "c7f2d9a41e83": "file_storage", + "f5b1c3e8a207": "file_storage", # size_bytes BIGINT (#708) "e5f2a8c1d7b3": "core", # timezone-aware timestamps across modules "b5d3f08a6e17": "background_tasks", "e878882924e7": "background_tasks", @@ -58,7 +59,7 @@ # head revision -> module owning the tables it creates or alters. _HEADS = { "a7c2e91d4b58": "audit_log", - "b31a70971001": "background_tasks", # also carries ``api_tokens`` and ``users`` + "f5b1c3e8a207": "file_storage", # also carries ``api_tokens``, ``background_tasks``, ``users`` "f1a9c3d7b2e4": "tenants", "f6a3b9d2e8c4": "keycloak", "f733c7938fd1": "sharing", @@ -75,6 +76,7 @@ "bb906e1a1089": "background_tasks", "d263ea727663": "leases", "a625f0b7c912": "users", + "f5b1c3e8a207": "file_storage", } # Known mislabels: revisions that inherit a label from an ancestor owned by a @@ -99,6 +101,7 @@ "1f5985daa10b": "api_tokens", # also inherits ``background_tasks`` "a625f0b7c912": "api_tokens", # also inherits ``background_tasks`` "b31a70971001": "api_tokens", + "f5b1c3e8a207": "api_tokens", "a626b1c8d023": "api_tokens", # also inherits ``background_tasks`` "51b2058d8afd": "api_tokens", # ancestors of a labelled revision inherit its label too } diff --git a/modules/file_storage/README.md b/modules/file_storage/README.md index 19384b8f2..e51111862 100644 --- a/modules/file_storage/README.md +++ b/modules/file_storage/README.md @@ -27,6 +27,22 @@ The extras combine as `simple_module_file_storage[s3,tasks]`. Without - Pluggable backend selected by the `backend` setting (`filesystem` default | `s3`); third-party backends can register under any id. - Settings (`backend`, `s3_bucket`, `s3_region`, `s3_endpoint_url` for R2/MinIO/etc., `s3_access_key_id`, `s3_secret_access_key`, `s3_presign_ttl_seconds`) are configured from the DB settings store via the admin UI — they are not read from `SM_FILE_STORAGE_*` environment variables at runtime. +## Sync workers + +With the `[tasks]` extra installed, synchronous task bodies can import +`store_path`, `open_local`, and `presigned_url` from `file_storage.contracts`. +Run calls inside `simple_module_db.tenant_context(tenant_id)`. `store_path` +accepts a path or binary stream and returns committed `StoredFileOut` metadata; +`open_local(file_id)` yields a temporary local path and always cleans it up. +`presigned_url(file_id)` requires a backend with signing support. + +Uploads retain the existing MIME, quota, size and hashing rules. S3 writes above +`direct_upload_threshold_bytes` use bounded multipart, including unknown-size +streams. Raise the DB's `max_file_size_bytes` for large worker outputs and set an +S3 lifecycle rule to abort incomplete server uploads left by terminated jobs. +See `docs/modules/file_storage.md` in the repository for transaction and cleanup +semantics. + ## Usage From another module, inject the service via its dependency: diff --git a/modules/file_storage/file_storage/backends/s3.py b/modules/file_storage/file_storage/backends/s3.py index 119750ebe..8a7f15af5 100644 --- a/modules/file_storage/file_storage/backends/s3.py +++ b/modules/file_storage/file_storage/backends/s3.py @@ -9,7 +9,6 @@ from collections.abc import AsyncGenerator, AsyncIterator from datetime import datetime -from tempfile import SpooledTemporaryFile from typing import Any from file_storage import constants @@ -47,6 +46,8 @@ def __init__( region: str, client_kwargs: dict[str, Any], public_endpoint_url: str = "", + multipart_threshold_bytes: int = constants.DEFAULT_DIRECT_UPLOAD_THRESHOLD_BYTES, + multipart_part_size_bytes: int = constants.DEFAULT_PART_SIZE_BYTES, ) -> None: try: import aioboto3 @@ -58,6 +59,8 @@ def __init__( self.region = region self.client_kwargs = client_kwargs self.public_endpoint_url = public_endpoint_url + self.multipart_threshold_bytes = multipart_threshold_bytes + self.multipart_part_size_bytes = multipart_part_size_bytes def _client(self): return self._session.client("s3", **self.client_kwargs) @@ -70,25 +73,9 @@ async def put( content_type: str, size: int, ) -> None: - # Spool to memory up to SPOOL_MAX_SIZE_BYTES, then to a temp file. - # put_object needs a seekable file-like; multipart upload would be - # the next step for very large files (deferred to v2). - with SpooledTemporaryFile(max_size=constants.SPOOL_MAX_SIZE_BYTES) as buf: - async for chunk in stream: - buf.write(chunk) - length = buf.tell() - buf.seek(0) - try: - async with self._client() as client: - await client.put_object( - Bucket=self.bucket, - Key=key, - Body=buf, - ContentType=content_type, - ContentLength=length, - ) - except Exception as exc: - raise StorageBackendError(f"S3 put_object failed: {exc}") from exc + from file_storage.backends.s3_put import put_stream + + await put_stream(self, key, stream, content_type, size) async def get(self, key: str) -> AsyncIterator[bytes]: return await _open(self._stream(key, {})) @@ -233,4 +220,6 @@ def _build(settings: FileStorageSettings) -> S3Backend: region=settings.s3_region, client_kwargs=client_kwargs, public_endpoint_url=settings.s3_public_endpoint_url, + multipart_threshold_bytes=settings.direct_upload_threshold_bytes, + multipart_part_size_bytes=settings.multipart_part_size_bytes, ) diff --git a/modules/file_storage/file_storage/backends/s3_put.py b/modules/file_storage/file_storage/backends/s3_put.py new file mode 100644 index 000000000..330f6c184 --- /dev/null +++ b/modules/file_storage/file_storage/backends/s3_put.py @@ -0,0 +1,121 @@ +"""Bounded server-side S3 uploads, including streams of unknown size (#708).""" + +from __future__ import annotations + +import logging +from collections.abc import AsyncIterator +from tempfile import SpooledTemporaryFile +from typing import TYPE_CHECKING + +from file_storage import constants, upload_support +from file_storage.contracts.service import Part, StorageBackendError +from file_storage.upload_support import SizeMismatchError + +if TYPE_CHECKING: + from file_storage.backends.s3 import S3Backend + +logger = logging.getLogger(__name__) + + +async def put_stream( + backend: S3Backend, key: str, stream: AsyncIterator[bytes], content_type: str, size: int +) -> None: + """Buffer up to the threshold, then send one bounded part at a time.""" + if size < 0: + raise ValueError("size must be nonnegative (0 means unknown)") + threshold = min(backend.multipart_threshold_bytes, constants.MAX_S3_PART_SIZE_BYTES) + iterator = aiter(stream) + tail: memoryview | None = None + with SpooledTemporaryFile(max_size=constants.SPOOL_MAX_SIZE_BYTES) as prefix: + if size <= threshold: + async for chunk in iterator: + take = min(len(chunk), threshold + 1 - prefix.tell()) + prefix.write(memoryview(chunk)[:take]) + if prefix.tell() > threshold: + tail = memoryview(chunk)[take:] + break + else: + length = prefix.tell() + if size and length != size: + raise SizeMismatchError(size, length) + prefix.seek(0) + try: + async with backend._client() as client: + await client.put_object( + Bucket=backend.bucket, + Key=key, + Body=prefix, + ContentType=content_type, + ContentLength=length, + ) + except Exception as exc: + raise StorageBackendError(f"S3 put_object failed: {exc}") from exc + return + prefix.seek(0) + + async def body() -> AsyncIterator[bytes]: + while chunk := prefix.read(constants.DEFAULT_CHUNK_SIZE): + yield chunk + if tail is not None: + for offset in range(0, len(tail), constants.DEFAULT_CHUNK_SIZE): + yield bytes(tail[offset : offset + constants.DEFAULT_CHUNK_SIZE]) + async for chunk in iterator: + yield chunk + + await _multipart(backend, key, body(), content_type, size) + + +async def _multipart(backend: S3Backend, key, stream, content_type, size) -> None: + part_size = upload_support.choose_part_size(size, backend.multipart_part_size_bytes) + if part_size > constants.MAX_S3_PART_SIZE_BYTES: + raise ValueError("S3 part size exceeds 5 GiB") + upload_id = await backend.create_multipart(key, content_type) + parts: list[Part] = [] + seen = 0 + try: + with SpooledTemporaryFile(max_size=constants.SPOOL_MAX_SIZE_BYTES) as buf: + async for chunk in stream: + seen += len(chunk) + if size and seen > size: + raise SizeMismatchError(size, seen) + remaining = memoryview(chunk) + while remaining: + take = min(len(remaining), part_size - buf.tell()) + buf.write(remaining[:take]) + remaining = remaining[take:] + if buf.tell() == part_size: + parts.append(await _send(backend, key, upload_id, parts, buf)) + buf.seek(0) + buf.truncate() + if size and seen != size: + raise SizeMismatchError(size, seen) + if buf.tell() or not parts: + parts.append(await _send(backend, key, upload_id, parts, buf)) + await backend.complete_multipart(key, upload_id, parts) + except BaseException: + try: + await backend.abort_multipart(key, upload_id) + except BaseException: + logger.exception("file_storage.server_upload_abort_failed key=%s", key) + raise + + +async def _send(backend: S3Backend, key, upload_id, parts, buf) -> Part: + number = len(parts) + 1 + if number > constants.MAX_UPLOAD_PARTS: + raise ValueError("S3 upload exceeds maximum part count") + length = buf.tell() + buf.seek(0) + try: + async with backend._client() as client: + result = await client.upload_part( + Bucket=backend.bucket, + Key=key, + UploadId=upload_id, + PartNumber=number, + Body=buf, + ContentLength=length, + ) + except Exception as exc: + raise StorageBackendError(f"S3 upload_part failed: {exc}") from exc + return Part(no=number, etag=result["ETag"], size=length) diff --git a/modules/file_storage/file_storage/constants.py b/modules/file_storage/file_storage/constants.py index ca213ac4b..4072e2a7b 100644 --- a/modules/file_storage/file_storage/constants.py +++ b/modules/file_storage/file_storage/constants.py @@ -171,6 +171,7 @@ class I18nKey: DEFAULT_PART_SIZE_BYTES: Final = 16 * MIB # S3's hard ceiling; ``part_size`` is grown until the upload fits under it. MAX_UPLOAD_PARTS: Final = 10_000 +MAX_S3_PART_SIZE_BYTES: Final = 5 * 1024 * MIB # S3 rejects non-final parts below 5 MiB. MIN_PART_SIZE_BYTES: Final = 5 * MIB DEFAULT_UPLOAD_SESSION_TTL_SECONDS: Final = 24 * 3600 diff --git a/modules/file_storage/file_storage/contracts/__init__.py b/modules/file_storage/file_storage/contracts/__init__.py index 8a4c2c5a7..8844f81ea 100644 --- a/modules/file_storage/file_storage/contracts/__init__.py +++ b/modules/file_storage/file_storage/contracts/__init__.py @@ -1,5 +1,13 @@ """file_storage contracts — public interface for other modules.""" +from __future__ import annotations + +from contextlib import AbstractContextManager +from os import PathLike +from pathlib import Path +from typing import BinaryIO +from uuid import UUID + from file_storage.contracts.events import FileDeleted, FileUploaded from file_storage.contracts.schemas import StoredFileListOut, StoredFileOut from file_storage.contracts.service import ( @@ -28,4 +36,34 @@ "StorageNotFoundError", "StoredFileListOut", "StoredFileOut", + "open_local", + "presigned_url", + "store_path", ] + + +def store_path( + source: str | PathLike[str] | BinaryIO, + *, + filename: str | None = None, + content_type: str = "application/octet-stream", + public: bool = False, +) -> StoredFileOut: + """Store a path or stream in the caller's tenant, in an independent transaction.""" + from file_storage.worker import store_path as store + + return store(source, filename=filename, content_type=content_type, public=public) + + +def open_local(file_id: UUID) -> AbstractContextManager[Path]: + """Download to a private local path for the lifetime of this context.""" + from file_storage.worker import open_local as local + + return local(file_id) + + +def presigned_url(file_id: UUID, *, ttl_seconds: int | None = None) -> str: + """Return a scoped download URL; unsupported backends raise NotSupportedError.""" + from file_storage.worker import presigned_url as sign + + return sign(file_id, ttl_seconds=ttl_seconds) diff --git a/modules/file_storage/file_storage/contracts/service.py b/modules/file_storage/file_storage/contracts/service.py index 7f2958fbc..6d6536f57 100644 --- a/modules/file_storage/file_storage/contracts/service.py +++ b/modules/file_storage/file_storage/contracts/service.py @@ -63,7 +63,7 @@ async def put( content_type: str, size: int, ) -> None: - """Persist ``stream`` under ``key``. ``size`` is the total byte count.""" + """Persist ``stream`` under ``key``. ``size`` is the byte count, or 0 if unknown.""" async def get(self, key: str) -> AsyncIterator[bytes]: """Return an iterator over the object's bytes in bounded chunks. diff --git a/modules/file_storage/file_storage/models.py b/modules/file_storage/file_storage/models.py index 005892d28..ef7531f58 100644 --- a/modules/file_storage/file_storage/models.py +++ b/modules/file_storage/file_storage/models.py @@ -39,7 +39,8 @@ class StoredFile(Base, AuditMixin, SoftDeleteMixin, MultiTenantMixin, table=True key: str = Field(max_length=512) filename: str = Field(max_length=255) content_type: str = Field(max_length=128) - size_bytes: int = Field() + # BIGINT: server-side puts can exceed 2 GiB (#708). + size_bytes: int = Field(sa_column=sa.Column(sa.BigInteger(), nullable=False)) backend: str = Field(max_length=32) # Null for multipart uploads until ``file_storage.compute_checksum`` has # streamed the object (or forever, with no background_tasks): the bytes diff --git a/modules/file_storage/file_storage/service.py b/modules/file_storage/file_storage/service.py index 10bdd15b1..1d6669d63 100644 --- a/modules/file_storage/file_storage/service.py +++ b/modules/file_storage/file_storage/service.py @@ -139,10 +139,10 @@ async def _hashing_stream() -> AsyncIterator[bytes]: self.db.add(row) await self.db.flush() await self.db.refresh(row) - except Exception: + except BaseException: try: await self.backend.delete(key) - except Exception: + except BaseException: _logger.exception( "file_storage.cleanup_failed key=%s — original upload error follows", key, diff --git a/modules/file_storage/file_storage/tasks.py b/modules/file_storage/file_storage/tasks.py index 522f99a5e..fa982ae60 100644 --- a/modules/file_storage/file_storage/tasks.py +++ b/modules/file_storage/file_storage/tasks.py @@ -44,13 +44,9 @@ def _backend_and_settings(): - from background_tasks import settings_for + from file_storage.worker_support import backend_and_settings - from file_storage.backends import build_backend - from file_storage.settings import FileStorageSettings - - settings = settings_for(FileStorageSettings, constants.MODULE_NAME) - return build_backend(settings), settings + return backend_and_settings() @shared_task(name=constants.Task.REAP_STALE_UPLOADS) @@ -122,7 +118,9 @@ async def _abort_all(backend, targets: list[tuple[str, str]]) -> set[tuple[str, return failed -# Keys this module mints: ``///
/<32 hex>``. +# Keys this module mints: ``///
/<32 hex>``. A server-side +# put that went multipart and died is an orphan like a browser one: no real put +# runs for the 2 x session-TTL age the reaper waits, so it is safe to abort. _OWN_KEY = constants.OWN_KEY diff --git a/modules/file_storage/file_storage/worker.py b/modules/file_storage/file_storage/worker.py new file mode 100644 index 000000000..8d5227cf2 --- /dev/null +++ b/modules/file_storage/file_storage/worker.py @@ -0,0 +1,166 @@ +"""Synchronous, tenant-scoped file access for workers (#708). + +Uses the background-tasks calling thread's persistent loop and session factory; +imports of this module are deferred by the public contracts package. +""" + +from __future__ import annotations + +import logging +import os +from collections.abc import Generator +from contextlib import contextmanager +from functools import partial +from pathlib import Path +from tempfile import TemporaryDirectory +from typing import BinaryIO +from uuid import UUID + +from background_tasks.outbox_loop import run_on_worker_loop +from background_tasks.outbox_worker import worker_bus, worker_session_factory +from fastapi import UploadFile +from simple_module_db import independent_session +from simple_module_db.callbacks import register_on_commit +from starlette.datastructures import Headers + +from file_storage.backend_guard import require_backend +from file_storage.contracts import FileUploaded, NotSupportedError, StoredFileOut +from file_storage.downloads import presigned_download_url +from file_storage.models import StoredFile +from file_storage.scope import owning_tenant +from file_storage.service import FileStorageService +from file_storage.worker_support import backend_and_settings + +logger = logging.getLogger(__name__) + + +def store_path( + source: str | os.PathLike[str] | BinaryIO, + *, + filename: str | None = None, + content_type: str = "application/octet-stream", + public: bool = False, +) -> StoredFileOut: + """Store a path or the remaining bytes of a caller-owned stream. + + Anonymous streams require ``filename``. Caller-owned streams stay open. + The independent transaction commits before this function returns. + """ + backend, settings = backend_and_settings() + factory = worker_session_factory() + bus = worker_bus() + + async def store() -> StoredFileOut: + out = None + committed = False + + def mark_committed() -> None: + nonlocal committed + committed = True + + try: + async with independent_session(factory) as db: + owning_tenant(db) # Fail closed before even opening the source path. + service = FileStorageService(db, backend, settings) + with _source(source, filename) as (handle, name): + upload = UploadFile( + handle, filename=name, headers=Headers({"content-type": content_type}) + ) + out = await service.upload(upload, public=public) + register_on_commit(db, mark_committed) + register_on_commit( + db, + partial( + bus.publish, + FileUploaded( + file_id=out.id, + key=out.key, + backend=out.backend, + size_bytes=out.size_bytes, + uploaded_by=out.uploaded_by, + ), + ), + ) + return out + except BaseException: + if out is not None and not committed: + await _compensate(factory, backend, out) + raise + + return run_on_worker_loop(store) + + +async def _compensate(factory, backend, out: StoredFileOut) -> None: + # A lost COMMIT acknowledgement is ambiguous: never delete a durable file. + try: + async with factory() as db: + row = await db.get(StoredFile, out.id) + except BaseException: + logger.exception("file_storage.worker_commit_check_failed key=%s", out.key) + return + if row is None: + try: + await backend.delete(out.key) + except BaseException: + logger.exception("file_storage.worker_cleanup_failed key=%s", out.key) + + +@contextmanager +def _source(source, filename) -> Generator[tuple[BinaryIO, str]]: + if isinstance(source, (str, os.PathLike)): + path = Path(source) + with path.open("rb") as handle: + yield handle, filename or path.name + else: + name = filename or getattr(source, "name", None) + if not isinstance(name, (str, os.PathLike)) or not os.fspath(name): + raise ValueError("filename is required for anonymous streams") + yield source, filename or Path(name).name + + +def _reader(file_id: UUID): + backend, settings = backend_and_settings() + factory = worker_session_factory() + + async def metadata(): + async with factory() as db: + service = FileStorageService(db, backend, settings) + row = await service.get(file_id) + require_backend(backend, row) + return row + + # Metadata is detached and its transaction closed before any streaming. + return backend, settings, run_on_worker_loop(metadata) + + +@contextmanager +def open_local(file_id: UUID) -> Generator[Path]: + """Download into a private temporary directory, removed on every exit.""" + backend, _, row = _reader(file_id) + with TemporaryDirectory(prefix="sm-file-") as directory: + path = Path(directory) / ("file" + Path(row.filename).suffix) + + async def download() -> None: + body = await backend.get(row.key) + try: + with path.open("wb") as handle: + async for chunk in body: + handle.write(chunk) + finally: + close = getattr(body, "aclose", None) + if close is not None: + await close() + + run_on_worker_loop(download) + yield path + + +def presigned_url(file_id: UUID, *, ttl_seconds: int | None = None) -> str: + """Sign a scoped file download, or raise ``NotSupportedError``.""" + backend, settings, row = _reader(file_id) + if not backend.supports_presigned_url: + raise NotSupportedError("The configured backend cannot presign downloads") + ttl = settings.s3_presign_ttl_seconds if ttl_seconds is None else ttl_seconds + if ttl <= 0: + raise ValueError("ttl_seconds must be positive") + return run_on_worker_loop(lambda: presigned_download_url(backend, row, ttl, attachment=True)) diff --git a/modules/file_storage/file_storage/worker_support.py b/modules/file_storage/file_storage/worker_support.py new file mode 100644 index 000000000..e8b6b4577 --- /dev/null +++ b/modules/file_storage/file_storage/worker_support.py @@ -0,0 +1,13 @@ +"""Hydrated backend construction shared by sync worker APIs and tasks.""" + +from file_storage import constants +from file_storage.backends import build_backend +from file_storage.contracts.service import StorageBackend +from file_storage.settings import FileStorageSettings + + +def backend_and_settings() -> tuple[StorageBackend, FileStorageSettings]: + from background_tasks.worker_settings import settings_for + + settings = settings_for(FileStorageSettings, constants.MODULE_NAME) + return build_backend(settings), settings diff --git a/modules/file_storage/tests/conftest.py b/modules/file_storage/tests/conftest.py index a6387828a..53839ecd2 100644 --- a/modules/file_storage/tests/conftest.py +++ b/modules/file_storage/tests/conftest.py @@ -15,6 +15,7 @@ import pytest from file_storage.backends.filesystem import FilesystemBackend from file_storage_uploads_helpers import PART +from file_storage_worker_helpers import worker_db as worker_db from sqlalchemy import event from sqlalchemy.engine import Engine diff --git a/modules/file_storage/tests/file_storage_worker_helpers.py b/modules/file_storage/tests/file_storage_worker_helpers.py new file mode 100644 index 000000000..9d5361ce8 --- /dev/null +++ b/modules/file_storage/tests/file_storage_worker_helpers.py @@ -0,0 +1,40 @@ +"""A real worker database and filesystem, without a Celery broker.""" + +import pytest +from background_tasks import outbox_worker, sync_db +from background_tasks.worker_settings import clear_settings_cache +from file_storage.models import StoredFile +from simple_module_core.events import EventBus +from simple_module_db.stored_settings import Setting +from sqlalchemy import create_engine +from sqlalchemy.orm import Session + + +@pytest.fixture +def worker_db(tmp_path, monkeypatch): + outbox_worker.reset_worker_state() + sync_db.dispose_sync_engine() + clear_settings_cache() + monkeypatch.setattr(sync_db, "discover_audit_callback", lambda: None) + engine = create_engine(f"sqlite:///{tmp_path / 'worker.db'}") + StoredFile.metadata.create_all(engine) + Setting.metadata.create_all(engine) + root = tmp_path / "objects" + with Session(engine) as db: + db.add(Setting(key="file_storage.fs_root_path", value=str(root), value_type="string")) + db.commit() + sync_db.set_database_url(str(engine.url), tenant_strict=True) + bus = EventBus() + outbox_worker.use_bus(bus) + yield root, engine, bus + outbox_worker.reset_worker_state() + sync_db.dispose_sync_engine() + clear_settings_cache() + engine.dispose() + + +def set_setting(engine, name, value, value_type="integer"): + with Session(engine) as db: + db.add(Setting(key=f"file_storage.{name}", value=value, value_type=value_type)) + db.commit() + clear_settings_cache() diff --git a/modules/file_storage/tests/test_file_storage_server_put.py b/modules/file_storage/tests/test_file_storage_server_put.py new file mode 100644 index 000000000..8ee464830 --- /dev/null +++ b/modules/file_storage/tests/test_file_storage_server_put.py @@ -0,0 +1,208 @@ +"""Server-side multipart boundaries without a network service (#708).""" + +import asyncio +from contextlib import asynccontextmanager + +import pytest +from file_storage import constants +from file_storage.backends import build_backend +from file_storage.settings import FileStorageSettings +from file_storage.upload_support import SizeMismatchError + + +class S3Wire: + def __init__(self): + self.parts = [] + self.single = None + self.completed = None + self.aborted = False + self.failure = None + self.abort_failure = False + + async def put_object(self, **kw): + self.single = kw["Body"].read() + assert kw["ContentLength"] == len(self.single) + + async def create_multipart_upload(self, **kw): + return {"UploadId": "upload"} + + async def upload_part(self, **kw): + assert kw["PartNumber"] == len(self.parts) + 1 + body = kw["Body"] + self.parts.append(body if isinstance(body, bytes) else body.read()) + return {"ETag": str(kw["PartNumber"])} + + async def complete_multipart_upload(self, **kw): + if self.failure: + raise self.failure + self.completed = kw["MultipartUpload"]["Parts"] + + async def abort_multipart_upload(self, **kw): + self.aborted = True + if self.abort_failure: + raise RuntimeError("abort failed") + + +@pytest.fixture +def wire_backend(monkeypatch): + wire = S3Wire() + cfg = FileStorageSettings( + backend="s3", + s3_bucket="bucket", + s3_region="us-east-1", + direct_upload_threshold_bytes=6, + ) + backend = build_backend(cfg) + + @asynccontextmanager + async def client(): + yield wire + + monkeypatch.setattr(backend, "_client", client) + # Tiny, hand-verifiable parts after testing settings propagation separately. + backend.multipart_part_size_bytes = 5 + monkeypatch.setattr( + "file_storage.upload_support.choose_part_size", lambda size, preferred: preferred + ) + return wire, backend + + +async def chunks(*values): + for value in values: + yield value + + +@pytest.mark.parametrize( + "payload,multipart", + [(b"", False), (b"123456", False), (b"1234567", True), (b"1234567890", True)], +) +@pytest.mark.parametrize("known", [False, True]) +async def test_threshold_and_exact_final_part(wire_backend, payload, multipart, known): + wire, backend = wire_backend + await backend.put( + "key", + chunks(payload[:3], payload[3:]), + content_type="text/plain", + size=len(payload) if known else 0, + ) + if multipart: + assert b"".join(wire.parts) == payload + assert [len(p) for p in wire.parts] == ([5, 2] if len(payload) == 7 else [5, 5]) + assert wire.single is None + assert wire.completed == [{"PartNumber": 1, "ETag": "1"}, {"PartNumber": 2, "ETag": "2"}] + else: + assert wire.single == payload + assert wire.parts == [] + + +@pytest.mark.parametrize("failure", [ValueError("source failed"), asyncio.CancelledError()]) +async def test_stream_failure_aborts_and_preserves_error(wire_backend, failure): + wire, backend = wire_backend + wire.abort_failure = True + + async def broken(): + yield b"1234567" + raise failure + + with pytest.raises(type(failure)) as caught: + await backend.put("key", broken(), content_type="text/plain", size=0) + assert caught.value is failure + assert wire.aborted + assert wire.completed is None + + +async def test_completion_failure_aborts(wire_backend): + wire, backend = wire_backend + wire.failure = RuntimeError("complete failed") + wire.abort_failure = True + with pytest.raises(Exception, match="complete failed"): + await backend.put("key", chunks(b"1234567"), content_type="text/plain", size=0) + assert wire.aborted + + +@pytest.mark.parametrize("declared", [2, 8, 12]) +async def test_declared_size_mismatch_is_not_published(wire_backend, declared): + wire, backend = wire_backend + with pytest.raises(SizeMismatchError): + await backend.put("key", chunks(b"1234567890"), content_type="text/plain", size=declared) + assert wire.single is None and wire.completed is None + + +async def test_part_count_bound_aborts(wire_backend, monkeypatch): + wire, backend = wire_backend + monkeypatch.setattr(constants, "MAX_UPLOAD_PARTS", 1) + with pytest.raises(Exception, match="part"): + await backend.put("key", chunks(b"12345678901"), content_type="text/plain", size=0) + assert wire.aborted + assert len(wire.parts) == 1 + + +def test_factory_passes_stored_multipart_settings(): + backend = build_backend( + FileStorageSettings( + backend="s3", + s3_bucket="bucket", + s3_region="us-east-1", + direct_upload_threshold_bytes=123, + multipart_part_size_bytes=7 * constants.MIB, + ) + ) + assert backend.multipart_threshold_bytes == 123 + assert backend.multipart_part_size_bytes == 7 * constants.MIB + + +async def test_unknown_stream_uploads_before_source_is_exhausted(wire_backend): + wire, backend = wire_backend + + async def source(): + yield b"1234567" + assert wire.parts == [b"12345"] + yield b"890" + + await backend.put("key", source(), content_type="text/plain", size=0) + assert b"".join(wire.parts) == b"1234567890" + + +async def test_large_source_chunk_keeps_spools_bounded(wire_backend, monkeypatch): + from tempfile import SpooledTemporaryFile + + from file_storage.backends import s3_put + + maximums = [] + + class TrackedBuffer: + def __init__(self, **kwargs): + self.file = SpooledTemporaryFile(**kwargs) # noqa: SIM115 — closed by wrapper context + self.index = len(maximums) + maximums.append(0) + + def __getattr__(self, name): + return getattr(self.file, name) + + def __enter__(self): + return self + + def __exit__(self, *args): + self.file.close() + + def write(self, data): + self.file.write(data) + maximums[self.index] = max(maximums[self.index], self.file.tell()) + + monkeypatch.setattr(s3_put, "SpooledTemporaryFile", TrackedBuffer) + wire, backend = wire_backend + await backend.put("key", chunks(b"x" * 100), content_type="text/plain", size=0) + assert b"".join(wire.parts) == b"x" * 100 + assert maximums == [7, 5] + + +async def test_failed_part_upload_aborts(wire_backend, monkeypatch): + wire, backend = wire_backend + + async def fail(**kwargs): + raise RuntimeError("part failed") + + monkeypatch.setattr(wire, "upload_part", fail) + with pytest.raises(Exception, match="part failed"): + await backend.put("key", chunks(b"1234567"), content_type="text/plain", size=0) + assert wire.aborted and wire.completed is None diff --git a/modules/file_storage/tests/test_file_storage_worker.py b/modules/file_storage/tests/test_file_storage_worker.py new file mode 100644 index 000000000..4afd6c43a --- /dev/null +++ b/modules/file_storage/tests/test_file_storage_worker.py @@ -0,0 +1,217 @@ +"""Public synchronous worker file operations (#708).""" + +import hashlib +import stat +from io import BytesIO +from uuid import uuid4 + +import pytest +from background_tasks import outbox_worker +from file_storage import contracts +from file_storage.backend_guard import BackendUnavailableError +from file_storage.contracts import FileUploaded, NotSupportedError, StoredFileOut +from file_storage.models import StoredFile +from file_storage.service import ( + ContentTypeNotAllowedError, + FileTooLargeError, + StoredFileNotFoundError, +) +from file_storage.upload_support import QuotaExceededError +from file_storage_worker_helpers import set_setting +from simple_module_db import tenant_context +from simple_module_db.tenancy import TenantIsolationError +from sqlalchemy import select +from sqlalchemy.orm import Session + + +def test_contract_exports_worker_functions(): + for name in ("store_path", "open_local", "presigned_url"): + assert callable(getattr(contracts, name, None)), name + + +def test_path_roundtrip_and_after_commit_event(worker_db, tmp_path): + _root, engine, bus = worker_db + source = tmp_path / "input.tif" + source.write_bytes(b"raster") + received = [] + + async def uploaded(event): + # A separate connection sees only committed metadata. + with Session(engine) as db: + assert db.get(StoredFile, event.file_id) is not None + received.append(event) + + bus.subscribe(FileUploaded, uploaded) + with tenant_context("alpha"): + out = contracts.store_path(source, content_type="image/tiff", public=True) + assert isinstance(out, StoredFileOut) + assert (out.filename, out.content_type, out.size_bytes, out.public) == ( + "input.tif", + "image/tiff", + 6, + True, + ) + assert out.key.startswith("alpha/") + assert out.checksum_sha256 == hashlib.sha256(b"raster").hexdigest() + with contracts.open_local(out.id) as path: + assert path.read_bytes() == b"raster" + assert stat.S_IMODE(path.parent.stat().st_mode) == 0o700 + assert not path.parent.exists() + assert len(received) == 1 + assert (received[0].file_id, received[0].key, received[0].size_bytes) == (out.id, out.key, 6) + + +def test_stream_current_position_and_ownership(worker_db): + stream = BytesIO(b"skipcontents") + stream.seek(4) + with tenant_context("alpha"): + out = contracts.store_path(stream, filename="output.bin") + assert out.size_bytes == 8 + with contracts.open_local(out.id) as path: + assert path.read_bytes() == b"contents" + assert not stream.closed + assert stream.tell() == 12 + + +def test_anonymous_stream_needs_filename(worker_db): + stream = BytesIO(b"unchanged") + with tenant_context("alpha"), pytest.raises(ValueError, match="filename"): + contracts.store_path(stream) + assert stream.tell() == 0 and not stream.closed + + +def test_unscoped_upload_fails_before_opening_source(worker_db, tmp_path): + with pytest.raises(TenantIsolationError): + contracts.store_path(tmp_path / "does-not-exist") + assert not worker_db[0].exists() + + +@pytest.mark.parametrize("operation", ["open_local", "presigned_url"]) +def test_missing_and_cross_tenant_ids(worker_db, operation): + with tenant_context("alpha"): + out = contracts.store_path(BytesIO(b"private"), filename="a.bin") + for target in (out.id, uuid4()): + with tenant_context("beta"), pytest.raises(StoredFileNotFoundError): + if operation == "open_local": + with contracts.open_local(target): + pass + else: + contracts.presigned_url(target) + + +@pytest.mark.parametrize( + "name,value,content_type,error", + [ + ("max_file_size_bytes", "2", "text/plain", FileTooLargeError), + ("allowed_content_types", '["image/png"]', "text/plain", ContentTypeNotAllowedError), + ("quota_bytes", "2", "text/plain", QuotaExceededError), + ], +) +def test_upload_limits_leave_no_objects_or_rows(worker_db, name, value, content_type, error): + root, engine, _ = worker_db + set_setting(engine, name, value, "json" if name == "allowed_content_types" else "integer") + with tenant_context("alpha"), pytest.raises(error): + contracts.store_path(BytesIO(b"too large"), filename="a.txt", content_type=content_type) + assert not list(root.rglob("*.*")) + with Session(engine) as db: + assert list(db.scalars(select(StoredFile))) == [] + + +@pytest.mark.parametrize("phase", ["flush", "commit"]) +def test_database_failure_compensates(worker_db, monkeypatch, phase): + root, engine, bus = worker_db + received = [] + + async def uploaded(event): + received.append(event) + + bus.subscribe(FileUploaded, uploaded) + factory = outbox_worker.worker_session_factory() + + async def fail(*args, **kwargs): + raise RuntimeError("database failed") + + monkeypatch.setattr(factory.class_, phase, fail) + with tenant_context("alpha"), pytest.raises(RuntimeError, match="database failed"): + contracts.store_path(BytesIO(b"contents"), filename="a.bin") + assert not list(root.rglob("*.bin")) + with Session(engine) as db: + assert list(db.scalars(select(StoredFile))) == [] + assert received == [] + + +def test_local_context_cleans_on_body_failure(worker_db): + with tenant_context("alpha"): + out = contracts.store_path(BytesIO(b"contents"), filename="a.bin") + with pytest.raises(RuntimeError, match="consumer"), contracts.open_local(out.id) as path: + raise RuntimeError("consumer") + assert not path.parent.exists() + + +def test_filesystem_presign_is_unsupported(worker_db): + with tenant_context("alpha"): + out = contracts.store_path(BytesIO(), filename="empty") + with pytest.raises(NotSupportedError): + contracts.presigned_url(out.id) + + +@pytest.mark.parametrize("operation", ["open_local", "presigned_url"]) +def test_read_refuses_wrong_backend(worker_db, operation): + with tenant_context("alpha"): + out = contracts.store_path(BytesIO(b"contents"), filename="a.bin") + with Session(worker_db[1]) as db: + row = db.get(StoredFile, out.id) + row.backend = "other" + db.commit() + with tenant_context("alpha"), pytest.raises(BackendUnavailableError): + if operation == "open_local": + with contracts.open_local(out.id): + pass + else: + contracts.presigned_url(out.id) + + +def test_worker_context_does_not_leak_between_calls(worker_db): + with tenant_context("alpha"): + a = contracts.store_path(BytesIO(b"a"), filename="a.bin") + factory = outbox_worker.worker_session_factory() + with tenant_context("beta"): + b = contracts.store_path(BytesIO(b"b"), filename="b.bin") + assert factory is outbox_worker.worker_session_factory() + with pytest.raises(StoredFileNotFoundError), contracts.open_local(a.id): + pass + with pytest.raises(TenantIsolationError): + contracts.store_path(BytesIO(b"c"), filename="c.bin") + assert a.key.startswith("alpha/") and b.key.startswith("beta/") + + +def test_concurrent_callers_keep_tenants_and_pools_separate(worker_db): + from concurrent.futures import ThreadPoolExecutor + + def store(tenant): + with tenant_context(tenant): + out = contracts.store_path(BytesIO(tenant.encode()), filename="result.bin") + return out, id(outbox_worker.worker_session_factory()) + + with ThreadPoolExecutor(max_workers=2) as executor: + results = list(executor.map(store, ["alpha", "beta"])) + assert [out.key.split("/")[0] for out, _ in results] == ["alpha", "beta"] + assert len({pool for _, pool in results}) == 2 + + +def test_contract_import_does_not_load_worker_dependencies(): + import subprocess + import sys + + code = """ +import builtins +original = builtins.__import__ +def guarded(name, *args, **kwargs): + if name == "background_tasks" or name.startswith("background_tasks."): + raise AssertionError("contracts imported worker dependencies eagerly") + return original(name, *args, **kwargs) +builtins.__import__ = guarded +from file_storage.contracts import store_path, open_local, presigned_url +assert all(callable(fn) for fn in (store_path, open_local, presigned_url)) +""" + subprocess.run([sys.executable, "-c", code], check=True, timeout=20) diff --git a/modules/file_storage/tests/test_file_storage_worker_failures.py b/modules/file_storage/tests/test_file_storage_worker_failures.py new file mode 100644 index 000000000..5de01206c --- /dev/null +++ b/modules/file_storage/tests/test_file_storage_worker_failures.py @@ -0,0 +1,178 @@ +"""Worker download, event and compensation failure boundaries (#708).""" + +import asyncio +from concurrent.futures import CancelledError +from io import BytesIO + +import pytest +from background_tasks import outbox_worker +from file_storage import contracts +from file_storage.backends.filesystem import FilesystemBackend +from file_storage.models import StoredFile +from file_storage_worker_helpers import set_setting +from simple_module_db import tenant_context +from sqlalchemy import select +from sqlalchemy.orm import Session + + +def test_cancelled_flush_removes_uploaded_bytes(worker_db, monkeypatch): + factory = outbox_worker.worker_session_factory() + + async def fail(*args, **kwargs): + raise asyncio.CancelledError() + + monkeypatch.setattr(factory.class_, "flush", fail) + with tenant_context("alpha"), pytest.raises(CancelledError): + contracts.store_path(BytesIO(b"contents"), filename="a.bin") + assert not list(worker_db[0].rglob("*.bin")) + + +def test_failed_compensation_preserves_commit_error(worker_db, monkeypatch): + factory = outbox_worker.worker_session_factory() + + async def commit(*args, **kwargs): + raise ValueError("commit failed") + + async def delete(*args, **kwargs): + raise RuntimeError("cleanup failed") + + monkeypatch.setattr(factory.class_, "commit", commit) + monkeypatch.setattr(FilesystemBackend, "delete", delete) + with tenant_context("alpha"), pytest.raises(ValueError, match="commit failed"): + contracts.store_path(BytesIO(b"contents"), filename="a.bin") + with Session(worker_db[1]) as db: + assert list(db.scalars(select(StoredFile))) == [] + + +def test_download_failure_closes_stream_and_removes_temporary_directory(worker_db, monkeypatch): + from tempfile import TemporaryDirectory + + from file_storage import worker + + directories = [] + closed = [] + + def temporary(**kwargs): + directory = TemporaryDirectory(**kwargs) + directories.append(directory.name) + return directory + + async def get(self, key): + async def broken(): + try: + yield b"partial" + raise ValueError("download failed") + finally: + closed.append(True) + + return broken() + + with tenant_context("alpha"): + out = contracts.store_path(BytesIO(b"contents"), filename="a.bin") + monkeypatch.setattr(FilesystemBackend, "get", get) + monkeypatch.setattr(worker, "TemporaryDirectory", temporary) + with pytest.raises(ValueError, match="download failed"), contracts.open_local(out.id): + pytest.fail("partial download was yielded") + from pathlib import Path + + assert directories and not Path(directories[0]).exists() + assert closed == [True] + + +def test_presign_honors_settings_override_and_download_headers(worker_db, monkeypatch): + seen = [] + + async def sign(self, key, ttl_seconds, **kwargs): + seen.append((key, ttl_seconds, kwargs)) + return "https://bucket.example/signed" + + monkeypatch.setattr(FilesystemBackend, "supports_presigned_url", True) + monkeypatch.setattr(FilesystemBackend, "presigned_get_url", sign) + set_setting(worker_db[1], "s3_presign_ttl_seconds", "42") + with tenant_context("alpha"): + out = contracts.store_path( + BytesIO(b"contents"), filename="report.tif", content_type="image/tiff" + ) + assert contracts.presigned_url(out.id) == "https://bucket.example/signed" + assert contracts.presigned_url(out.id, ttl_seconds=12) == "https://bucket.example/signed" + with pytest.raises(ValueError, match="positive"): + contracts.presigned_url(out.id, ttl_seconds=0) + assert [ttl for _, ttl, _ in seen] == [42, 12] + assert all(key == out.key for key, _, _ in seen) + assert seen[0][2] == { + "content_type": "image/tiff", + "content_disposition": "attachment; filename=\"report.tif\"; filename*=UTF-8''report.tif", + } + + +def test_named_handle_infers_basename_and_stays_open(worker_db, tmp_path): + source = tmp_path / "named.bin" + source.write_bytes(b"named") + with source.open("rb") as handle, tenant_context("alpha"): + out = contracts.store_path(handle) + assert out.filename == "named.bin" + assert not handle.closed + + +@pytest.mark.parametrize("operation", ["open_local", "presigned_url"]) +def test_unscoped_reads_fail_closed(worker_db, operation): + from simple_module_db.tenancy import TenantIsolationError + + with tenant_context("alpha"): + out = contracts.store_path(BytesIO(b"private"), filename="a.bin") + with pytest.raises(TenantIsolationError): + if operation == "open_local": + with contracts.open_local(out.id): + pytest.fail("unscoped file was exposed") + else: + contracts.presigned_url(out.id) + + +def test_download_does_not_hold_metadata_connection(worker_db, monkeypatch): + factory = outbox_worker.worker_session_factory() + real = FilesystemBackend.get + checked_out = [] + + async def get(self, key): + checked_out.append(factory.kw["bind"].pool.checkedout()) + return await real(self, key) + + with tenant_context("alpha"): + out = contracts.store_path(BytesIO(b"contents"), filename="a.bin") + monkeypatch.setattr(FilesystemBackend, "get", get) + with contracts.open_local(out.id) as path: + assert path.read_bytes() == b"contents" + assert checked_out == [0] + + +def test_commit_acknowledgement_failure_keeps_committed_object(worker_db, monkeypatch): + factory = outbox_worker.worker_session_factory() + real_commit = factory.class_.commit + + async def commit_then_cancel(db): + await real_commit(db) + raise asyncio.CancelledError() + + monkeypatch.setattr(factory.class_, "commit", commit_then_cancel) + with tenant_context("alpha"), pytest.raises(CancelledError): + contracts.store_path(BytesIO(b"contents"), filename="a.bin") + with Session(worker_db[1]) as db: + row = db.scalars(select(StoredFile)).one() + assert (worker_db[0] / row.key[:2] / row.key).read_bytes() == b"contents" + + +def test_unverifiable_commit_outcome_preserves_object_and_original_error(worker_db, monkeypatch): + factory = outbox_worker.worker_session_factory() + + async def commit(*args): + raise ValueError("commit failed") + + async def get(*args, **kwargs): + raise RuntimeError("database unavailable") + + monkeypatch.setattr(factory.class_, "commit", commit) + monkeypatch.setattr(factory.class_, "get", get) + with tenant_context("alpha"), pytest.raises(ValueError, match="commit failed"): + contracts.store_path(BytesIO(b"contents"), filename="a.bin") + objects = list(worker_db[0].rglob("*.bin")) + assert len(objects) == 1 and objects[0].read_bytes() == b"contents" From d37ca40e8791e339487ab373461904d8771120a5 Mon Sep 17 00:00:00 2001 From: Anto Subash Date: Fri, 30 Oct 2026 23:55:32 +0100 Subject: [PATCH 2/4] fix: address code review findings (round 1, pass 1) Cap multipart part size at S3's 5 GiB; trashed rows count as committed in worker compensation; README matches the reaper. Claude-Session: https://claude.ai/code/session_01M9neheZZEe3sVpDi2S3zT4 --- CHANGELOG.md | 6 ++++-- modules/file_storage/README.md | 4 ++-- modules/file_storage/file_storage/settings.py | 2 ++ modules/file_storage/file_storage/worker.py | 3 ++- .../tests/test_file_storage_server_put.py | 13 +++++++++++++ 5 files changed, 23 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 60e070d75..ac784312d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,8 +22,10 @@ All notable changes to this project are documented in this file. The format is b - Tenant-scoped sync file worker contracts (`store_path`, `open_local`, and `presigned_url`), reusing worker settings/pools, upload validation and hashing, with commit compensation and `FileUploaded` publication after commit (#708). -- Bounded S3 server-side multipart uploads above the stored threshold, including - unknown-length streams, size/part bounds and abort on failure (#708). +- **Behaviour change:** S3 server-side uploads (including browser single-request + ones) above `direct_upload_threshold_bytes` now use bounded multipart, including + unknown-length streams, with size/part bounds and abort on failure; the S3 + credentials need `s3:AbortMultipartUpload` (#708). - Users auth emails support ordered host template directories through the restart-required DB setting `mail_template_dirs`, with per-file bundled fallback and optional autoescaped HTML alternatives (#712). diff --git a/modules/file_storage/README.md b/modules/file_storage/README.md index e51111862..d65d4936b 100644 --- a/modules/file_storage/README.md +++ b/modules/file_storage/README.md @@ -38,8 +38,8 @@ accepts a path or binary stream and returns committed `StoredFileOut` metadata; Uploads retain the existing MIME, quota, size and hashing rules. S3 writes above `direct_upload_threshold_bytes` use bounded multipart, including unknown-size -streams. Raise the DB's `max_file_size_bytes` for large worker outputs and set an -S3 lifecycle rule to abort incomplete server uploads left by terminated jobs. +streams. Raise the DB's `max_file_size_bytes` for large worker outputs. An +incomplete multipart upload left by a killed job is aborted by the upload reaper. See `docs/modules/file_storage.md` in the repository for transaction and cleanup semantics. diff --git a/modules/file_storage/file_storage/settings.py b/modules/file_storage/file_storage/settings.py index e16f8c79d..80d47164d 100644 --- a/modules/file_storage/file_storage/settings.py +++ b/modules/file_storage/file_storage/settings.py @@ -123,6 +123,8 @@ class FileStorageSettings(DbBackedSettings): multipart_part_size_bytes: int = Field( default=constants.DEFAULT_PART_SIZE_BYTES, ge=constants.MIN_PART_SIZE_BYTES, + # S3 refuses parts over 5 GiB; a larger value would fail every multipart put. + le=constants.MAX_S3_PART_SIZE_BYTES, description="Preferred part size; grown automatically to stay under 10,000 parts.", ) upload_session_ttl_seconds: int = Field( diff --git a/modules/file_storage/file_storage/worker.py b/modules/file_storage/file_storage/worker.py index 8d5227cf2..eff320d4b 100644 --- a/modules/file_storage/file_storage/worker.py +++ b/modules/file_storage/file_storage/worker.py @@ -94,7 +94,8 @@ async def _compensate(factory, backend, out: StoredFileOut) -> None: # A lost COMMIT acknowledgement is ambiguous: never delete a durable file. try: async with factory() as db: - row = await db.get(StoredFile, out.id) + # Trashed rows keep their object for restore: only a missing row means rollback. + row = await db.get(StoredFile, out.id, execution_options={"include_deleted": True}) except BaseException: logger.exception("file_storage.worker_commit_check_failed key=%s", out.key) return diff --git a/modules/file_storage/tests/test_file_storage_server_put.py b/modules/file_storage/tests/test_file_storage_server_put.py index 8ee464830..e0832d9a0 100644 --- a/modules/file_storage/tests/test_file_storage_server_put.py +++ b/modules/file_storage/tests/test_file_storage_server_put.py @@ -206,3 +206,16 @@ async def fail(**kwargs): with pytest.raises(Exception, match="part failed"): await backend.put("key", chunks(b"1234567"), content_type="text/plain", size=0) assert wire.aborted and wire.completed is None + + +def test_part_size_setting_is_capped_at_the_s3_limit(): + from pydantic import ValidationError + + with pytest.raises(ValidationError): + FileStorageSettings(multipart_part_size_bytes=constants.MAX_S3_PART_SIZE_BYTES + 1) + assert ( + FileStorageSettings( + multipart_part_size_bytes=constants.MAX_S3_PART_SIZE_BYTES + ).multipart_part_size_bytes + == constants.MAX_S3_PART_SIZE_BYTES + ) From 1ec0cee110e509830f5f1c584851cea5998eda56 Mon Sep 17 00:00:00 2001 From: Anto Subash Date: Fri, 30 Oct 2026 23:58:27 +0100 Subject: [PATCH 3/4] docs: file_storage branch label in the migrations guide Claude-Session: https://claude.ai/code/session_01M9neheZZEe3sVpDi2S3zT4 --- docs/database/migrations.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/database/migrations.md b/docs/database/migrations.md index 5765f8bab..6cdde5c57 100644 --- a/docs/database/migrations.md +++ b/docs/database/migrations.md @@ -34,7 +34,7 @@ A module's first migration sets `branch_labels = ("",)`. Alembic ** ### Known mislabels -The declared `branch_labels` are `audit_log`, `api_tokens`, `background_tasks`, `keycloak`, `tenants`, `sharing`, `leases` and `users` (added by `a625f0b7c912`). The graph carries these defects; they are documented rather than rewritten, because changing labels on revisions that installed databases already applied is not worth the risk: +The declared `branch_labels` are `audit_log`, `api_tokens`, `background_tasks`, `keycloak`, `tenants`, `sharing`, `leases`, `users` (added by `a625f0b7c912`) and `file_storage` (added by `f5b1c3e8a207`). The graph carries these defects; they are documented rather than rewritten, because changing labels on revisions that installed databases already applied is not worth the risk: - `51b2058d8afd_api_tokens_initial` sits in the middle of the trunk, so every revision above it (file_storage, background_tasks including the outbox, and the users revisions through `c8e4f1a2b6d9`) reports the `api_tokens` label. The `api_tokens` head is therefore really a users revision. - `keycloak` branches off a users revision, so a Keycloak-only install still builds the users schema. From 8bb8668aa00d82f7a9a7cd23d89f75fea1e53e7a Mon Sep 17 00:00:00 2001 From: Anto Subash Date: Sat, 31 Oct 2026 00:05:47 +0100 Subject: [PATCH 4/4] fix: address code review findings (round 1, pass 2) store_path declares a real file's size so S3 parts grow past the 10,000-part limit; presigned TTL checked before the query; abort asserted on mismatch. Claude-Session: https://claude.ai/code/session_01M9neheZZEe3sVpDi2S3zT4 --- docs/database/migrations.md | 2 +- modules/file_storage/README.md | 6 ++++-- modules/file_storage/file_storage/service.py | 3 ++- modules/file_storage/file_storage/worker.py | 21 ++++++++++++++++--- .../tests/test_file_storage_server_put.py | 1 + .../tests/test_file_storage_worker.py | 14 +++++++++++++ 6 files changed, 40 insertions(+), 7 deletions(-) diff --git a/docs/database/migrations.md b/docs/database/migrations.md index 6cdde5c57..c1f54896e 100644 --- a/docs/database/migrations.md +++ b/docs/database/migrations.md @@ -38,7 +38,7 @@ The declared `branch_labels` are `audit_log`, `api_tokens`, `background_tasks`, - `51b2058d8afd_api_tokens_initial` sits in the middle of the trunk, so every revision above it (file_storage, background_tasks including the outbox, and the users revisions through `c8e4f1a2b6d9`) reports the `api_tokens` label. The `api_tokens` head is therefore really a users revision. - `keycloak` branches off a users revision, so a Keycloak-only install still builds the users schema. -- `users`, `file_storage`, `background_tasks`, `settings`, `permissions` and `feature_flags` have no label of their own. +- `settings`, `permissions` and `feature_flags` have no label of their own. `users` and `file_storage` gained labels late (`a625f0b7c912`, `f5b1c3e8a207`), so their earlier revisions stay unlabelled and report `api_tokens`. Guidance: give an unlabelled module a `branch_labels` on its **next** revision. Adding a label to an existing revision file is technically possible, because `alembic_version` stores revision ids and never labels, but it also relabels every descendant and changes what `