feat(file_storage): sync worker API, S3 multipart server puts, BIGINT sizes (#708)

CHANGELOG.md

@@ -19,6 +19,13 @@ 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).
+- **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).
@@ -43,6 +50,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).

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:

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` |
@@ -34,11 +34,11 @@ A module's first migration sets `branch_labels = ("<module_name>",)`. 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.
-- `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 `<label>@head` resolves to, so treat it as a deliberate, reviewed change rather than a cleanup.
 

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/`.

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
+        )

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
 }

modules/file_storage/README.md

@@ -27,6 +27,24 @@ 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. A file path's size is
+known, so parts grow to fit S3's 10,000-part limit; an unknown-size stream uses
+`multipart_part_size_bytes` as is (raise it for streams beyond ~156 GiB). 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.
+
 ## Usage
 
 From another module, inject the service via its dependency:

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,
     )