feat(file_storage): direct, resumable multipart uploads (#454)

CHANGELOG.md

@@ -18,6 +18,13 @@ All notable changes to this project are documented in this file. The format is b
   previously served, and anonymous, in production.
 
 ### Added
+- Resumable multipart uploads for `file_storage` (#454): `POST /api/file-storage/uploads`
+  sessions with presigned S3 part URLs (or relayed parts on the filesystem backend),
+  4-way parallel parts with retry and `localStorage` resume in the Files page,
+  `s3_public_endpoint_url`, CSP `connect-src` for the S3 origin, and an hourly
+  `file_storage.reap_stale_uploads` janitor (with `background_tasks`). Sessions count
+  against the quota and the assembled size is re-verified at completion. Migration
+  `a659221c464b` adds `file_storage_upload_session`.
 - background_tasks Celery config (#447, #448, #450): a module can ship a light
   `task_config.py` (`TASK_ROUTES`, `QUEUES`, `BEAT_SCHEDULE`) read without
   importing `tasks.py`; `task_routes_override` re-routes tasks at runtime and the
@@ -140,6 +147,9 @@ All notable changes to this project are documented in this file. The format is b
   production-mode containers pass `UsersSettings` boot validation.
 
 ### Changed
+- `StoredFile.checksum_sha256` / `StoredFileOut.checksum_sha256` are now nullable (#454):
+  files uploaded through the multipart flow have no checksum until
+  `file_storage.compute_checksum` runs, and `ETag`s fall back to id + size.
 - **Dashboard stats are admin-only on single-tenant installs too** (#464).
   `dashboard.view` was mapped to the ordinary `user` role when
   `multi_tenant` was off, so any self-registered account could read user

docs/modules/file_storage.md

@@ -108,6 +108,43 @@ Redirected (presigned) downloads are served by the object store, which handles
 `Range` itself. Always redirecting large downloads to a presigned URL, as an
 opt-in setting, is not implemented.
 
+### Direct, resumable uploads
+
+Files above `direct_upload_threshold_bytes` (default 64 MiB) upload in parts through a *session*, so a dropped connection or a reload loses at most one part. The Files page does this automatically; API clients drive it themselves. All routes need `file_storage.upload` and act only on the caller's own sessions (anyone else's id is a 404).
+
+| Route | Purpose |
+|---|---|
+| `POST /api/file-storage/uploads` `{filename, content_type, total_size, public}` | Open a session. Checks the content-type allow-list, `max_direct_upload_bytes` (default 10 GiB) and the quota (stored bytes plus the declared size of every still-open session). Returns `part_size` (16 MiB, grown so the file needs at most 10,000 parts) and `total_parts`. |
+| `GET /uploads/{id}` | The session plus the parts the backend holds — the source of truth for resuming. |
+| `POST /uploads/{id}/part-urls` `{parts: [n]}` | Where to PUT each part: presigned S3 URLs (`direct: true`, bound to the part's length) or the app's own part route. Only part numbers `1..total_parts`. |
+| `PUT /uploads/{id}/parts/{n}` | Filesystem backend only: the part body, relayed through the app (capped at that part's exact size). |
+| `POST /uploads/{id}/complete` `{parts: [{n, etag}]}` | Assemble, verify, create the `StoredFile`, publish `FileUploaded`. The assembled size must equal `total_size`; the maximum and the quota are re-checked against the stored object, and a violation deletes it. Retrying a completed session returns the same file. |
+| `DELETE /uploads/{id}` | Abort and discard the parts. |
+
+Sessions expire after `upload_session_ttl_seconds` (24 h). With `background_tasks` installed, the hourly beat task `file_storage.reap_stale_uploads` aborts expired sessions and, on S3, orphaned multipart uploads older than twice the TTL that have this module's key shape and no active session (a bucket shared with other applications is left alone); without it, `POST /uploads` aborts the caller's own expired sessions. Completed and aborted session rows are deleted `upload_session_retention_days` (default 7) after their expiry, by the same task or, without `background_tasks`, by the caller's next `POST /uploads` (own rows only). `file_storage.compute_checksum` fills `checksum_sha256` after completion when `background_tasks` is installed.
+
+In the browser, parts go four at a time with per-part retry and backoff. The session id is kept in `localStorage` keyed by file name, size and last-modified time; after a reload, select the same file again and the page offers **Resume upload** or **Start over**. (Browsers cannot keep a file handle across a reload, so the file must be picked again.)
+
+Known limitations:
+
+- Two simultaneous `PUT`s of the same part on the filesystem backend can leave the part's data and ETag from different writers; `complete` then reports a mismatch and the client re-sends the part.
+- The resume key is file name, size and last-modified time, so two tabs uploading an identical file share one session.
+- A part shorter than expected is stored and refused with 400; it stays until the client re-sends the part (the resend overwrites it), and `complete` flags it as mismatched.
+
+#### S3 setup
+
+Parts are PUT by the browser straight to the bucket, so:
+
+- **CORS** on the bucket must allow `PUT` from the app's origin and expose `ETag`:
+  ```json
+  [{"AllowedOrigins": ["https://app.example.com"], "AllowedMethods": ["PUT"],
+    "AllowedHeaders": ["*"], "ExposeHeaders": ["ETag"], "MaxAgeSeconds": 3000}]
+  ```
+- `s3_public_endpoint_url` is the endpoint browsers reach, when it differs from the one the server uses (`s3_endpoint_url`, e.g. an in-cluster MinIO hostname). Presigned URLs are signed for it, and the redirect for presigned downloads uses it too. Blank falls back to `s3_endpoint_url`.
+- The endpoint's origin is added to the CSP `connect-src` automatically (for AWS defaults, the bucket's `*.s3.<region>.amazonaws.com` hosts).
+
+New settings: `direct_upload_threshold_bytes`, `max_direct_upload_bytes`, `multipart_part_size_bytes`, `upload_session_ttl_seconds`, `upload_session_retention_days`, `s3_public_endpoint_url`. A backend opts in with `supports_direct_upload = True` and the `DirectUploadBackend` methods in `file_storage.contracts.service`; without it, `POST /uploads` answers 501 and the page never uses sessions.
+
 ### View
 
 | Method + path | Inertia component | Permission |
@@ -153,7 +190,7 @@ from file_storage.contracts import (
 | `content_type` | `str(128)` | sniffed / declared MIME type |
 | `size_bytes` | `int` | |
 | `backend` | `str(32)` | `"filesystem"` or `"s3"` — recorded at upload time |
-| `checksum_sha256` | `str(64)` | computed during stream-upload |
+| `checksum_sha256` | `str(64) \| null` | computed during stream-upload; **null** for a direct (multipart) upload until `file_storage.compute_checksum` runs (never, without `background_tasks`) |
 | `public` | `bool` | default `false`; opt-in anonymous serving |
 | `extra_metadata` | `dict` | per-backend extras |
 | audit + soft-delete | from `AuditMixin` + `SoftDeleteMixin` | |

host/migrations/versions/a659221c464b_file_storage_upload_sessions.py

@@ -0,0 +1,96 @@
+"""file_storage: resumable upload sessions, nullable checksum (#454)
+
+Multipart uploads bypass the app, so nothing hashes them on the way in: the
+checksum becomes nullable and is filled in later by a background task. The
+downgrade backfills nulls with an empty string before restoring NOT NULL.
+
+Revision ID: a659221c464b
+Revises: e5a8c1f27b94
+Create Date: 2026-10-15 10:18:19.289035
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "a659221c464b"
+down_revision: str | None = "e5a8c1f27b94"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "file_storage_upload_session",
+        sa.Column("tenant_id", sa.String(length=50), nullable=False),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("(CURRENT_TIMESTAMP)"),
+            nullable=False,
+        ),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("created_by", sa.String(length=255), nullable=True),
+        sa.Column("updated_by", sa.String(length=255), nullable=True),
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("key", sa.String(length=512), nullable=False),
+        sa.Column("filename", sa.String(length=255), nullable=False),
+        sa.Column("content_type", sa.String(length=128), nullable=False),
+        sa.Column("total_size", sa.BigInteger(), nullable=False),
+        sa.Column("part_size", sa.BigInteger(), nullable=False),
+        sa.Column("backend_upload_id", sa.String(length=255), nullable=False),
+        sa.Column("status", sa.String(length=16), nullable=False),
+        sa.Column("public", sa.Boolean(), server_default=sa.false(), nullable=False),
+        sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("file_id", sa.Uuid(), nullable=True),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_file_storage_upload_session")),
+    )
+    op.create_index(
+        "ix_file_storage_upload_session_created_by",
+        "file_storage_upload_session",
+        ["created_by"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_file_storage_upload_session_status_expires",
+        "file_storage_upload_session",
+        ["status", "expires_at"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_file_storage_upload_session_tenant_id"),
+        "file_storage_upload_session",
+        ["tenant_id"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_file_storage_upload_session_tenant_key",
+        "file_storage_upload_session",
+        ["tenant_id", "key"],
+        unique=True,
+    )
+    with op.batch_alter_table("file_storage_stored_file") as batch:
+        batch.alter_column("checksum_sha256", existing_type=sa.String(length=64), nullable=True)
+
+
+def downgrade() -> None:
+    op.execute(
+        "UPDATE file_storage_stored_file SET checksum_sha256 = '' WHERE checksum_sha256 IS NULL"
+    )
+    with op.batch_alter_table("file_storage_stored_file") as batch:
+        batch.alter_column("checksum_sha256", existing_type=sa.String(length=64), nullable=False)
+    op.drop_index(
+        "ix_file_storage_upload_session_tenant_key", table_name="file_storage_upload_session"
+    )
+    op.drop_index(
+        op.f("ix_file_storage_upload_session_tenant_id"), table_name="file_storage_upload_session"
+    )
+    op.drop_index(
+        "ix_file_storage_upload_session_status_expires", table_name="file_storage_upload_session"
+    )
+    op.drop_index(
+        "ix_file_storage_upload_session_created_by", table_name="file_storage_upload_session"
+    )
+    op.drop_table("file_storage_upload_session")

modules/file_storage/file_storage/backends/filesystem.py

@@ -11,6 +11,7 @@
 
 from file_storage import constants
 from file_storage.backends import register_backend
+from file_storage.backends.filesystem_multipart import FilesystemMultipartMixin
 from file_storage.contracts.service import (
     NotSupportedError,
     StorageBackendError,
@@ -19,7 +20,7 @@
 from file_storage.settings import FileStorageSettings
 
 
-class FilesystemBackend:
+class FilesystemBackend(FilesystemMultipartMixin):
     """Stores objects on the local filesystem under ``root``.
 
     Keys are sharded by their first two characters to keep any single

modules/file_storage/file_storage/backends/filesystem_multipart.py

@@ -0,0 +1,147 @@
+"""Multipart-upload half of the filesystem backend (#454).
+
+Parts live under ``<root>/.multipart/<upload_id>/<n>`` with the part's MD5 in a
+``<n>.etag`` sidecar, so :meth:`list_parts` never re-reads gigabytes to answer.
+:meth:`complete_multipart` concatenates them into the final object through a
+temp file and an atomic rename, like :meth:`FilesystemBackend.put`.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import hashlib
+import re
+import shutil
+import uuid
+from collections.abc import AsyncIterator
+from pathlib import Path
+
+import aiofiles
+
+from file_storage import constants
+from file_storage.contracts.service import (
+    NotSupportedError,
+    ObjectInfo,
+    Part,
+    StorageBackendError,
+    StorageNotFoundError,
+)
+
+_UPLOAD_ID = re.compile(r"^[0-9a-f]{32}$")
+_ETAG_SUFFIX = ".etag"
+
+
+class FilesystemMultipartMixin:
+    """Mixed into :class:`~file_storage.backends.filesystem.FilesystemBackend`."""
+
+    supports_direct_upload = True
+    root: Path
+
+    def _resolve(self, key: str) -> Path:  # provided by the concrete backend
+        raise NotImplementedError
+
+    def _parts_dir(self, upload_id: str) -> Path:
+        # Upload ids are minted here, but they come back in a URL and a DB row:
+        # refuse anything that is not exactly what we issued.
+        if not _UPLOAD_ID.match(upload_id):
+            raise StorageBackendError(f"Invalid upload id: {upload_id!r}")
+        return self.root / constants.FS_MULTIPART_DIR / upload_id
+
+    async def create_multipart(self, key: str, content_type: str) -> str:
+        self._resolve(key)  # validates the key before we hand out an id
+        upload_id = uuid.uuid4().hex
+        self._parts_dir(upload_id).mkdir(parents=True, exist_ok=True)
+        return upload_id
+
+    async def presign_part(
+        self, key: str, upload_id: str, part_no: int, ttl: int, size: int | None = None
+    ) -> str:
+        raise NotSupportedError("Filesystem parts are uploaded through the app.")
+
+    async def presigned_put_url(self, key: str, ttl: int, content_type: str) -> str:
+        raise NotSupportedError("Filesystem backend cannot mint presigned URLs.")
+
+    async def write_part(
+        self,
+        key: str,
+        upload_id: str,
+        part_no: int,
+        stream: AsyncIterator[bytes],
+        *,
+        max_bytes: int,
+    ) -> Part:
+        directory = self._parts_dir(upload_id)
+        if not directory.is_dir():
+            raise StorageNotFoundError(upload_id)
+        target = directory / str(part_no)
+        tmp = directory / f"{part_no}.{uuid.uuid4().hex}.tmp"
+        md5 = hashlib.md5(usedforsecurity=False)
+        size = 0
+        cleanup = [tmp]
+        try:
+            async with aiofiles.open(tmp, "wb") as fh:
+                async for chunk in stream:
+                    size += len(chunk)
+                    if size > max_bytes:
+                        raise StorageBackendError(f"Part {part_no} exceeds {max_bytes} bytes.")
+                    md5.update(chunk)
+                    await fh.write(chunk)
+            etag = md5.hexdigest()
+            # Etag first: a part is "present" only once its data file is, and the
+            # data file is renamed last.
+            # The sidecar goes through its own temp name too, so a concurrent
+            # writer of the same part never leaves a half-written etag.
+            etag_tmp = directory / f"{part_no}.{uuid.uuid4().hex}.etag.tmp"
+            cleanup.append(etag_tmp)
+            etag_tmp.write_text(etag)
+            etag_tmp.replace(directory / f"{part_no}{_ETAG_SUFFIX}")
+            tmp.replace(target)
+        except BaseException:
+            for leftover in cleanup:
+                with contextlib.suppress(OSError):
+                    leftover.unlink(missing_ok=True)
+            raise
+        return Part(no=part_no, etag=etag, size=size)
+
+    async def list_parts(self, key: str, upload_id: str) -> list[Part]:
+        directory = self._parts_dir(upload_id)
+        if not directory.is_dir():
+            raise StorageNotFoundError(upload_id)
+        parts = []
+        for path in directory.iterdir():
+            if not path.name.isdigit():
+                continue  # ``.etag`` sidecars and in-flight ``.tmp`` files
+            etag_file = path.with_name(path.name + _ETAG_SUFFIX)
+            etag = etag_file.read_text() if etag_file.exists() else ""
+            parts.append(Part(no=int(path.name), etag=etag, size=path.stat().st_size))
+        return sorted(parts, key=lambda p: p.no)
+
+    async def complete_multipart(self, key: str, upload_id: str, parts: list[Part]) -> None:
+        directory = self._parts_dir(upload_id)
+        path = self._resolve(key)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        tmp = path.with_name(f"{path.name}.{uuid.uuid4().hex}.tmp")
+        try:
+            async with aiofiles.open(tmp, "wb") as out:
+                for part in sorted(parts, key=lambda p: p.no):
+                    source = directory / str(part.no)
+                    if not source.is_file():
+                        raise StorageBackendError(f"Part {part.no} is missing.")
+                    async with aiofiles.open(source, "rb") as fh:
+                        while chunk := await fh.read(constants.DEFAULT_CHUNK_SIZE):
+                            await out.write(chunk)
+            tmp.replace(path)
+        except BaseException:
+            with contextlib.suppress(OSError):
+                tmp.unlink(missing_ok=True)
+            raise
+        shutil.rmtree(directory, ignore_errors=True)
+
+    async def abort_multipart(self, key: str, upload_id: str) -> None:
+        shutil.rmtree(self._parts_dir(upload_id), ignore_errors=True)
+
+    async def head(self, key: str) -> ObjectInfo:
+        path = self._resolve(key)
+        if not path.is_file():
+            raise StorageNotFoundError(key)
+        return ObjectInfo(size=path.stat().st_size)

modules/file_storage/file_storage/backends/s3.py

@@ -13,6 +13,7 @@
 
 from file_storage import constants
 from file_storage.backends import register_backend
+from file_storage.backends.s3_multipart import S3MultipartMixin
 from file_storage.contracts.service import (
     ConfigurationError,
     StorageBackendError,
@@ -26,7 +27,7 @@
 )
 
 
-class S3Backend:
+class S3Backend(S3MultipartMixin):
     """S3-compatible object storage.
 
     Each operation opens a fresh ``aioboto3`` client context (cheap; the
@@ -38,7 +39,14 @@ class S3Backend:
     backend_id = constants.BackendId.S3
     supports_presigned_url = True
 
-    def __init__(self, *, bucket: str, region: str, client_kwargs: dict[str, Any]) -> None:
+    def __init__(
+        self,
+        *,
+        bucket: str,
+        region: str,
+        client_kwargs: dict[str, Any],
+        public_endpoint_url: str = "",
+    ) -> None:
         try:
             import aioboto3
         except ImportError as exc:
@@ -48,6 +56,7 @@ def __init__(self, *, bucket: str, region: str, client_kwargs: dict[str, Any]) -
         self.bucket = bucket
         self.region = region
         self.client_kwargs = client_kwargs
+        self.public_endpoint_url = public_endpoint_url
 
     def _client(self):
         return self._session.client("s3", **self.client_kwargs)
@@ -132,7 +141,7 @@ async def exists(self, key: str) -> bool:
 
     async def presigned_get_url(self, key: str, ttl_seconds: int) -> str:
         try:
-            async with self._client() as client:
+            async with self._presign_client() as client:
                 return await client.generate_presigned_url(
                     "get_object",
                     Params={"Bucket": self.bucket, "Key": key},
@@ -200,4 +209,5 @@ def _build(settings: FileStorageSettings) -> S3Backend:
         bucket=settings.s3_bucket,
         region=settings.s3_region,
         client_kwargs=client_kwargs,
+        public_endpoint_url=settings.s3_public_endpoint_url,
     )