Uploads by PUT range requests rather than WS, remove dead code WS handlers (#9)

Increases block size to 16 MiB and adjusts progress display to work smoothly with that. Also removes download WS that was already unused. Provides faster upload speed than over WS.

cista/api.py

@@ -1,5 +1,4 @@
 import asyncio
-import typing
 from pathlib import PurePosixPath
 from secrets import token_bytes
 
@@ -9,7 +8,7 @@ from sanic.exceptions import BadRequest
 
 from cista import __version__, auth, config, sso, watching
 from cista.fileio import FileServer
-from cista.protocol import ControlTypes, FileRange, StatusMsg
+from cista.protocol import ControlTypes, StatusMsg
 from cista.util.apphelpers import asend, websocket_wrapper
 
 bp = Blueprint("api", url_prefix="/api")
@@ -26,65 +25,6 @@ async def stop_fileserver(app):
     await fileserver.stop()
 
 
-@bp.websocket("upload")
-@websocket_wrapper
-async def upload(req, ws):
-    alink = fileserver.alink
-    while True:
-        req = None
-        text = await ws.recv()
-        if not isinstance(text, str):
-            raise ValueError(
-                f"Expected JSON control, got binary len(data) = {len(text)}",
-            )
-        req = msgspec.json.decode(text, type=FileRange)
-        pos = req.start
-        while True:
-            data = await ws.recv()
-            if not isinstance(data, bytes):
-                break
-            if len(data) > req.end - pos:
-                raise ValueError(
-                    f"Expected up to {req.end - pos} bytes, got {len(data)} bytes"
-                )
-            sentsize = await alink(("upload", req.name, pos, data, req.size))
-            pos += typing.cast(int, sentsize)
-            if pos >= req.end:
-                break
-        if pos != req.end:
-            d = f"{len(data)} bytes" if isinstance(data, bytes) else data
-            raise ValueError(f"Expected {req.end - pos} more bytes, got {d}")
-        # Signal the watcher about the uploaded file and its parent directories
-        path = PurePosixPath(req.name)
-        watching.notify_change(path, *path.parents)
-        # Report success
-        res = StatusMsg(status="ack", req=req)
-        await asend(ws, res)
-
-
-@bp.websocket("download")
-@websocket_wrapper
-async def download(req, ws):
-    alink = fileserver.alink
-    while True:
-        req = None
-        text = await ws.recv()
-        if not isinstance(text, str):
-            raise ValueError(
-                f"Expected JSON control, got binary len(data) = {len(text)}",
-            )
-        req = msgspec.json.decode(text, type=FileRange)
-        pos = req.start
-        while pos < req.end:
-            end = min(req.end, pos + (1 << 20))
-            data = typing.cast(bytes, await alink(("download", req.name, pos, end)))
-            await asend(ws, data)
-            pos += len(data)
-        # Report success
-        res = StatusMsg(status="ack", req=req)
-        await asend(ws, res)
-
-
 @bp.websocket("control")
 @websocket_wrapper
 async def control(req, ws):

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/<name:path>")
+    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):

cista/fileio.py

@@ -1,9 +1,8 @@
-import asyncio
 import os
+import threading
 
 from cista import config
 from cista.util import filename
-from cista.util.asynclink import AsyncLink
 from cista.util.lrucache import LRUCache
 
 
@@ -62,38 +61,32 @@ class File:
 
 class FileServer:
     async def start(self):
-        self.alink = AsyncLink()
-        self.worker = asyncio.get_event_loop().run_in_executor(
-            None,
-            self.worker_thread,
-            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()
-        await self.worker
+        self.cache.close()
 
-    def worker_thread(self, slink):
+    @staticmethod
+    def _stat_size(path):
         try:
-            for req in slink:
-                with req as (command, *args):
-                    if command == "upload":
-                        req.set_result(self.upload(*args))
-                    elif command == "download":
-                        req.set_result(self.download(*args))
-                    else:
-                        raise NotImplementedError(f"Unhandled {command=} {args}")
-        finally:
-            self.cache.close()
+            return os.stat(path).st_size
+        except FileNotFoundError:
+            return None
 
-    def upload(self, name, pos, data, file_size):
+    def upload_info(self, name, pos, data, file_size):
         name = filename.sanitize(name)
-        f = self.cache[name]
-        f.write(pos, data, file_size=file_size)
-        return len(data)
-
-    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:
+            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,
+        }

cista/protocol.py

@@ -12,7 +12,6 @@ from cista.util import filename
 
 ## Control commands
 
-
 class ControlBase(msgspec.Struct, tag_field="op", tag=str.lower):
     def __call__(self):
         raise NotImplementedError
@@ -118,19 +117,9 @@ class Cp(ControlBase):
 ControlTypes = MkDir | Rename | Rm | Mv | Cp
 
 
-## File uploads and downloads
-
-
-class FileRange(msgspec.Struct):
-    name: str
-    size: int
-    start: int
-    end: int
-
-
 class StatusMsg(msgspec.Struct):
     status: str
-    req: FileRange
+    req: Any
 
 
 class ErrorMsg(msgspec.Struct):