From 76c928a24c86d4d85feccfd4197d08f6eae34cc3 Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Sat, 25 Apr 2026 01:08:21 +0000 Subject: [PATCH 1/2] Implement PUT chunk uploads, 16MiB chunk size for faster transfers with resilient retries and smoother progress --- cista/app.py | 93 +++++- cista/fileio.py | 42 ++- frontend/src/components/UploadButton.vue | 344 ++++++++++++++++------- 3 files changed, 370 insertions(+), 109 deletions(-) diff --git a/cista/app.py b/cista/app.py index a2e8897..3c64b13 100644 --- a/cista/app.py +++ b/cista/app.py @@ -1,6 +1,7 @@ import asyncio import datetime import mimetypes +import re import time from concurrent.futures import ThreadPoolExecutor from multiprocessing import cpu_count @@ -11,8 +12,8 @@ from wsgiref.handlers import format_date_time import sanic.helpers from blake3 import blake3 -from sanic import Blueprint, Sanic, empty, raw, redirect -from sanic.exceptions import Forbidden, NotFound +from sanic import Blueprint, Sanic, empty, json, raw, redirect +from sanic.exceptions import BadRequest, Forbidden, NotFound from sanic.log import logger from setproctitle import setproctitle from stream_zip import ZIP_AUTO, stream_zip @@ -20,7 +21,7 @@ from zstandard import ZstdCompressor from cista import auth, config, preview, session, sso, watching from cista.preview import shutdown_preview_workers, start_preview_workers -from cista.api import bp +from cista.api import bp, fileserver from cista.sanic_logging import configure_access_logging, configure_main_logging, format_access_log from cista.sanic_logging import logger as access_logger from cista.util.apphelpers import handle_sanic_exception @@ -127,6 +128,68 @@ def http_fileserver(app): """Verify access to file server routes.""" await auth.verify(request) + @bp.put("/files/") + async def upload_file_chunk(request, *args, **kwargs): + body = request.body + header = request.headers.get("content-range") + if header: + start, end, total = _parse_content_range(header, len(body)) + else: + start = 0 + end = len(body) + total = end + raw_name = kwargs.get("name") + if raw_name is None and args: + raw_name = args[0] + if not isinstance(raw_name, str) or not raw_name: + prefix = "/files/" + if not request.path.startswith(prefix): + raise BadRequest("Invalid upload path") + raw_name = request.path[len(prefix) :] + rel_name = unquote(raw_name) + upload_info = await asyncio.to_thread( + fileserver.upload_info, + rel_name, + start, + body, + total, + ) + extras = [] + chunk_len = end - start + whole_file = start == 0 and end == total + if not whole_file: + start_mib = _to_mib_int(start) + chunk_mib = _to_mib_int(chunk_len) + # Keep range logs compact for fixed-size upload blocks. + if chunk_mib == 16: + extras.append(f"{start_mib}MiB") + else: + extras.append(f"{start_mib}+{chunk_mib}MiB") + if upload_info.get("created"): + extras.append(f"created {_to_mib_int(total)}MiB") + size_before = upload_info.get("size_before") + size_after = upload_info.get("size_after") + if ( + size_before is not None + and size_after is not None + and size_before != size_after + ): + extras.append("resized") + request.ctx._log_extra = " ".join(extras) if extras else None + path = PurePosixPath(rel_name) + watching.notify_change(path, *path.parents) + return json( + { + "status": "ack", + "req": { + "name": rel_name, + "size": total, + "start": start, + "end": end, + }, + } + ) + bp.static( "/files/", config.config.path, @@ -138,6 +201,30 @@ def http_fileserver(app): www = {} +_CONTENT_RANGE_RE = re.compile(r"^bytes (\d+)-(\d+)/(\d+)$") + + +def _parse_content_range(header: str, body_len: int) -> tuple[int, int, int]: + m = _CONTENT_RANGE_RE.fullmatch(header.strip()) + if m is None: + raise BadRequest("Invalid Content-Range format") + start, end_inclusive, total = (int(v) for v in m.groups()) + if total <= 0: + raise BadRequest("Invalid Content-Range total size") + if start > end_inclusive: + raise BadRequest("Invalid Content-Range range") + if end_inclusive >= total: + raise BadRequest("Content-Range exceeds total size") + expected_len = end_inclusive - start + 1 + if expected_len != body_len: + raise BadRequest( + f"Content length mismatch for range: expected {expected_len}, got {body_len}" + ) + return start, end_inclusive + 1, total + + +def _to_mib_int(value_bytes: int) -> int: + return round(value_bytes / (1 << 20)) def _load_wwwroot(www): diff --git a/cista/fileio.py b/cista/fileio.py index b9052f8..663bb4e 100644 --- a/cista/fileio.py +++ b/cista/fileio.py @@ -1,5 +1,6 @@ import asyncio import os +import threading from cista import config from cista.util import filename @@ -69,6 +70,8 @@ class FileServer: self.alink.to_sync, ) self.cache = LRUCache(File, capacity=10, maxage=5.0) + self.cache_lock = threading.Lock() + self.file_locks: dict[str, threading.Lock] = {} async def stop(self): await self.alink.stop() @@ -80,6 +83,8 @@ class FileServer: with req as (command, *args): if command == "upload": req.set_result(self.upload(*args)) + elif command == "upload_info": + req.set_result(self.upload_info(*args)) elif command == "download": req.set_result(self.download(*args)) else: @@ -87,13 +92,42 @@ class FileServer: finally: self.cache.close() + @staticmethod + def _stat_size(path): + try: + return os.stat(path).st_size + except FileNotFoundError: + return None + def upload(self, name, pos, data, file_size): name = filename.sanitize(name) - f = self.cache[name] - f.write(pos, data, file_size=file_size) + with self.cache_lock: + f = self.cache[name] + lock = self.file_locks.setdefault(name, threading.Lock()) + with lock: + f.write(pos, data, file_size=file_size) return len(data) + def upload_info(self, name, pos, data, file_size): + name = filename.sanitize(name) + with self.cache_lock: + f = self.cache[name] + lock = self.file_locks.setdefault(name, threading.Lock()) + with lock: + size_before = self._stat_size(f.path) + f.write(pos, data, file_size=file_size) + size_after = self._stat_size(f.path) + return { + "written": len(data), + "created": size_before is None, + "size_before": size_before, + "size_after": size_after, + } + def download(self, name, start, end): name = filename.sanitize(name) - f = self.cache[name] - return f[start:end] + with self.cache_lock: + f = self.cache[name] + lock = self.file_locks.setdefault(name, threading.Lock()) + with lock: + return f[start:end] diff --git a/frontend/src/components/UploadButton.vue b/frontend/src/components/UploadButton.vue index cce59e3..795e3e2 100644 --- a/frontend/src/components/UploadButton.vue +++ b/frontend/src/components/UploadButton.vue @@ -8,12 +8,11 @@