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/preview.py

@@ -2,7 +2,6 @@ import asyncio
 import gc
 import io
 import mimetypes
-import os
 import struct
 import sys
 import threading
@@ -20,10 +19,8 @@ import msgspec
 import av
 import fitz  # PyMuPDF
 import numpy as np
-import pillow_heif
 import pyvips
 from blake3 import blake3
-from PIL import Image
 from sanic import Blueprint, empty, raw, redirect
 from sanic.exceptions import NotFound
 from sanic.log import logger
@@ -32,8 +29,6 @@ from cista import auth, config
 from cista.preview_worker import PreviewRequest, PreviewResponse
 from cista.util.filename import sanitize
 
-pillow_heif.register_heif_opener()
-
 bp = Blueprint("preview", url_prefix="/preview")
 
 
@@ -85,7 +80,6 @@ _active_procs: set[asyncio.subprocess.Process] = set()
 _preview_pool = None
 _preview_pool_lock = asyncio.Lock()
 AVIF_FAST_EFFORT = 0
-FORCE_PIL = os.environ.get("CISTA_PIL") == "1"
 WORKER_CHECKSUM_BYTES = 32
 WORKER_MAX_JSON_BYTES = 1_000_000
 
@@ -135,7 +129,11 @@ class _PreviewWorker:
 
         resp = msgspec.json.decode(meta_raw, type=PreviewResponse)
         if not resp.ok:
-            raise PreviewError(resp.error or "preview worker error")
+            raise PreviewError(
+                resp.error or "preview worker error",
+                stderr=resp.stderr,
+                backend=resp.backend,
+            )
         return payload or None, resp
 
     async def kill(self) -> None:
@@ -207,7 +205,7 @@ class _PreviewWorkerPool:
         except WorkerChecksumError:
             replace = True
             logger.error("Preview checksum mismatch for %s", filepath.name)
-            raise PreviewError(filepath.name)
+            raise PreviewError(f"worker checksum mismatch for {filepath.name}")
         except PreviewError:
             raise
         except (
@@ -223,7 +221,9 @@ class _PreviewWorkerPool:
             logger.warning(
                 "Preview worker protocol failure for %s: %s", filepath.name, e
             )
-            raise PreviewError(filepath.name)
+            raise PreviewError(
+                f"worker protocol failure for {filepath.name}: {e}"
+            )
         finally:
             if replace:
                 await self._replace_worker(worker)
@@ -295,6 +295,17 @@ class PreviewTimeout(Exception):
 class PreviewError(Exception):
     """Raised when the preview subprocess exits with a non-zero status."""
 
+    def __init__(
+        self,
+        message: str,
+        *,
+        stderr: str | None = None,
+        backend: str | None = None,
+    ):
+        super().__init__(message)
+        self.stderr = stderr
+        self.backend = backend
+
 
 async def _run_preview_process(
     filepath, quality: int, maxsize: int, maxzoom: float
@@ -302,22 +313,10 @@ async def _run_preview_process(
     """Run preview request in a persistent worker process."""
     await start_preview_workers()
     if _preview_pool is None:
-        raise PreviewError(filepath.name)
+        raise PreviewError(f"preview worker pool unavailable for {filepath.name}")
     return await _preview_pool.run(filepath, quality, maxsize, maxzoom)
 
 
-# Map EXIF Orientation value to a corresponding PIL transpose
-EXIF_ORI = {
-    2: Image.Transpose.FLIP_LEFT_RIGHT,
-    3: Image.Transpose.ROTATE_180,
-    4: Image.Transpose.FLIP_TOP_BOTTOM,
-    5: Image.Transpose.TRANSPOSE,
-    6: Image.Transpose.ROTATE_270,
-    7: Image.Transpose.TRANSVERSE,
-    8: Image.Transpose.ROTATE_90,
-}
-
-
 DOC_PREVIEW_SUFFIXES = {".pdf", ".xps", ".epub", ".mobi"}
 
 
@@ -368,7 +367,15 @@ async def preview(req, path):
         )
     except PreviewTimeout:
         return empty(504)
-    except PreviewError:
+    except PreviewError as e:
+        if e.backend:
+            req.ctx._log_extra = e.backend
+        detail = str(e)
+        if detail == "preview worker error" and e.stderr:
+            captured = e.stderr.strip()
+            if captured:
+                detail = captured.splitlines()[0]
+        logger.error("%s preview: %s", filepath, detail)
         return empty(422)
     if preview_resp and preview_resp.backend:
         if preview_resp.timings:
@@ -403,28 +410,26 @@ async def preview(req, path):
 
 
 def dispatch(path, quality, maxsize, maxzoom):
+    backend = "unknown"
     try:
         if path.suffix.lower() in DOC_PREVIEW_SUFFIXES:
+            backend = "pdf"
             return process_pdf(path, quality=quality, maxsize=maxsize, maxzoom=maxzoom)
         mime_type, _ = mimetypes.guess_type(path.name)
         if mime_type and mime_type.startswith("video/"):
+            backend = "video"
             return process_video(path, quality=quality, maxsize=maxsize)
         if mime_type and mime_type.startswith("image/"):
+            backend = "pyvips"
             return process_image(path, quality=quality, maxsize=maxsize)
     except ValueError as e:
-        logger.warning(f"Cannot generate preview for {path}: {e}")
+        return None, PreviewResponse(ok=False, backend=backend, error=str(e))
     except Exception as e:
-        logger.exception(f"Error generating preview for {path}: {e}")
-    return None, PreviewResponse(ok=False)
+        return None, PreviewResponse(ok=False, backend=backend, error=str(e))
+    return None, PreviewResponse(ok=False, backend=backend, error="preview unsupported")
 
 
 def process_image(path, *, maxsize, quality):
-    return process_image_with_timing(path, maxsize=maxsize, quality=quality)
-
-
-def process_image_with_timing(path, *, maxsize, quality):
-    if FORCE_PIL:
-        return process_image_pillow(path, maxsize=maxsize, quality=quality)
     return process_image_pyvips(path, maxsize=maxsize, quality=quality)
 
 
@@ -451,45 +456,6 @@ def process_image_pyvips(path, *, maxsize, quality):
     )
 
 
-def process_image_pillow(path, *, maxsize, quality):
-    t_load = perf_counter()
-    with Image.open(path) as img:
-        # Force decode to include I/O in load timing
-        img.load()
-        t_proc = perf_counter()
-        # Resize
-        w, h = img.size
-        img.thumbnail((min(w, maxsize), min(h, maxsize)))
-        # Transpose pixels according to EXIF Orientation
-        orientation = img.getexif().get(274, 1)
-        if orientation in EXIF_ORI:
-            img = img.transpose(EXIF_ORI[orientation])
-        # Save as AVIF
-        imgdata = io.BytesIO()
-        t_save = perf_counter()
-        img.save(
-            imgdata,
-            format="avif",
-            quality=quality,
-            speed=10,
-            max_threads=1,
-            avif=1,
-        )
-
-    t_end = perf_counter()
-    ret = imgdata.getvalue()
-
-    load_ms = (t_proc - t_load) * 1000
-    proc_ms = (t_save - t_proc) * 1000
-    save_ms = (t_end - t_save) * 1000
-    return ret, PreviewResponse(
-        ok=True,
-        mime="image/avif",
-        backend="pillow",
-        timings=[round(load_ms, 1), round(proc_ms, 1), round(save_ms, 1)],
-    )
-
-
 def process_pdf(path, *, maxsize, maxzoom, quality, page_number=0):
     t_load_start = perf_counter()
     pdf = fitz.open(path)
@@ -501,19 +467,13 @@ def process_pdf(path, *, maxsize, maxzoom, quality, page_number=0):
     t_load_end = perf_counter()
 
     t_save_start = perf_counter()
-    if FORCE_PIL:
-        ret = pix.pil_tobytes(
-            format="avif", quality=quality, speed=10, max_threads=1, avif=1
-        )
-        backend = "pdf"
-    else:
-        img = pyvips.Image.new_from_memory(
-            pix.samples_mv, pix.width, pix.height, pix.n, "uchar"
-        )
-        ret = img.write_to_buffer(
-            ".avif", Q=quality, effort=AVIF_FAST_EFFORT, strip=True
-        )
-        backend = "pdf+pyvips"
+    img = pyvips.Image.new_from_memory(
+        pix.samples_mv, pix.width, pix.height, pix.n, "uchar"
+    )
+    ret = img.write_to_buffer(
+        ".avif", Q=quality, effort=AVIF_FAST_EFFORT, strip=True
+    )
+    backend = "pdf+pyvips"
     t_save_end = perf_counter()
 
     return ret, PreviewResponse(

cista/preview_worker.py

@@ -10,6 +10,8 @@ where packet = (uint32 json size)(uint32 payload size)(json)(binary payload).
 """
 
 import logging
+import contextlib
+import io
 import struct
 import sys
 from pathlib import Path
@@ -31,6 +33,7 @@ class PreviewResponse(msgspec.Struct, omit_defaults=True):
     backend: str | None = None
     timings: list[float] | None = None
     error: str | None = None
+    stderr: str | None = None
 
 
 _enc = msgspec.json.Encoder()
@@ -70,14 +73,34 @@ def _run_loop() -> None:
         line = sys.stdin.buffer.readline()
         if not line:
             return
+        stderr_capture = io.StringIO()
+        handler = logging.StreamHandler(stderr_capture)
+        root_logger = logging.getLogger()
+        root_logger.addHandler(handler)
         try:
-            req = _dec_req.decode(line)
-            result, resp = dispatch(
-                Path(req.path), req.quality, req.maxsize, req.maxzoom
-            )
+            with contextlib.redirect_stderr(stderr_capture):
+                req = _dec_req.decode(line)
+                result, resp = dispatch(
+                    Path(req.path), req.quality, req.maxsize, req.maxzoom
+                )
+            if not resp.ok:
+                captured = stderr_capture.getvalue().strip()
+                if captured:
+                    resp = PreviewResponse(
+                        ok=False,
+                        backend=resp.backend,
+                        error=resp.error,
+                        stderr=captured,
+                    )
             _write_response(resp, result or b"")
         except Exception as e:
-            _write_response(PreviewResponse(ok=False, error=str(e)), b"")
+            captured = stderr_capture.getvalue().strip()
+            _write_response(
+                PreviewResponse(ok=False, error=str(e), stderr=captured or None), b""
+            )
+        finally:
+            root_logger.removeHandler(handler)
+            handler.close()
 
 
 def main() -> None:

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

cista/sanic_logging.py

@@ -236,7 +236,8 @@ class _EmojiFormatter(logging.Formatter):
 
     def format(self, record: logging.LogRecord) -> str:
         emoji = _LEVEL_EMOJI.get(record.levelno, "▪️")
-        return f"{emoji}  {record.getMessage()}"
+        sep = "  " if record.levelno in (logging.INFO, logging.WARNING) else " "
+        return f"{emoji}{sep}{record.getMessage()}"
 
 
 def configure_main_logging() -> None:

frontend/src/components/UploadButton.vue

@@ -8,12 +8,11 @@
 </template>
 
 <script setup lang="ts">
-import { connect, uploadUrl } from '@/repositories/WS';
 import { useMainStore } from '@/stores/main'
 import { getDocuments } from '@/stores/documentStore'
 import { Doc } from '@/repositories/Document'
 import { collator } from '@/utils';
-import { onMounted, onUnmounted, reactive, ref } from 'vue'
+import { onMounted, onUnmounted, ref } from 'vue'
 import { useRouter } from 'vue-router'
 
 const router = useRouter()
@@ -29,6 +28,22 @@ type CloudFile = {
   cloudName: string
   cloudPos: number
 }
+
+type UploadRange = {
+  name: string
+  size: number
+  start: number
+  end: number
+}
+
+type InflightBlock = {
+  name: string
+  start: number
+  end: number
+  startedAt: number
+}
+
+const UPLOAD_BLOCK_SIZE = 16 << 20  // 16 MiB
 function pasteHandler(event: ClipboardEvent) {
   const items = Array.from(event.clipboardData?.items ?? [])
   const infiles = [] as File[]
@@ -46,7 +61,8 @@ function pasteHandler(event: ClipboardEvent) {
   if (infiles.length || dirs.length) {
     event.preventDefault()
     uploadFiles(infiles)
-    for (const entry of dirs) pasteDirectory(entry, `${props.path!.join('/')}/${entry.name}`)
+    const base = props.path!.join('/')
+    for (const entry of dirs) pasteDirectory(entry, `${base ? `${base}/` : ''}${entry.name}`)
   }
 }
 const pasteDirectory = async (entry: FileSystemDirectoryEntry, loc: string) => {
@@ -82,7 +98,7 @@ const uploadFiles = (infiles: File[]) => {
     if (!folderName && file.webkitRelativePath) folderName = relPath.split('/')[0] ?? ''
     files.push({
       file,
-      cloudName: loc + '/' + relPath,
+      cloudName: `${loc ? `${loc}/` : ''}${relPath}`,
       cloudPos: 0,
     })
   }
@@ -130,8 +146,12 @@ const uploadCloudFiles = (files: CloudFile[]) => {
 }
 
 const cancelUploads = () => {
+  uploadRunId += 1
   upqueue = []
   blockQueue = []
+  inflightBlocks.clear()
+  uploadedBytes.clear()
+  store.uprogress.status = 'idle'
   statReset()
 }
 
@@ -152,51 +172,98 @@ const uprogress_init = {
   status: 'idle',
 }
 store.uprogress = {...uprogress_init}
+// Track uploaded bytes for each file to handle out-of-order uploads
+const uploadedBytes = new Map<string, Set<number>>()
+const inflightBlocks = new Map<string, InflightBlock>()
+let smoothedBlockMs = 1500
+let lastProgressTick = Date.now()
+let lastVisualUploaded = 0
+
+const inflightKey = (name: string, start: number) => `${name}:${start}`
+
+const completedUploadedBytes = (name: string, size: number) => {
+  const uploaded = uploadedBytes.get(name)
+  if (!uploaded) return 0
+  const blockSize = UPLOAD_BLOCK_SIZE
+  let total = 0
+  for (let i = 0; i < size; i += blockSize) {
+    if (uploaded.has(i)) total += Math.min(blockSize, size - i)
+  }
+  return total
+}
+
+const simulatedInflightBytes = (name: string, now: number) => {
+  let total = 0
+  for (const block of inflightBlocks.values()) {
+    if (block.name !== name) continue
+    const size = block.end - block.start
+    const elapsed = Math.max(0, now - block.startedAt)
+    const fraction = Math.min(0.98, elapsed / Math.max(200, smoothedBlockMs))
+    total += size * fraction
+  }
+  return total
+}
+
+const refreshProgress = (now: number) => {
+  const name = store.uprogress.filename
+  const size = store.uprogress.filesize
+  if (!name || !size) {
+    lastProgressTick = now
+    return 0
+  }
+
+  const completed = completedUploadedBytes(name, size)
+  const estimated = simulatedInflightBytes(name, now)
+  const visualUploaded = Math.min(size, Math.round(completed + estimated))
+  const delta = Math.max(0, visualUploaded - lastVisualUploaded)
+  const dt = Math.max(1, now - lastProgressTick)
+
+  store.uprogress.filepos = visualUploaded
+  store.uprogress.xfer = store.uprogress.filestart + visualUploaded
+
+  if (delta > 0) {
+    store.uprogress.statbytes += delta
+    store.uprogress.statdur += dt
+    store.uprogress.tlast = now
+  }
+
+  lastVisualUploaded = visualUploaded
+  lastProgressTick = now
+  return delta
+}
+
 setInterval(() => {
-  if (Date.now() - store.uprogress.tlast > 3000) {
-    // Reset
+  const now = Date.now()
+  const delta = refreshProgress(now)
+  if (delta > 0) return
+  if (now - store.uprogress.tlast > 3000) {
     store.uprogress.statbytes = 0
     store.uprogress.statdur = 1
   } else {
-    // Running average by decay
-    store.uprogress.statbytes *= .9
-    store.uprogress.statdur *= .9
+    store.uprogress.statbytes *= .95
+    store.uprogress.statdur *= .95
   }
 }, 100)
-// Track uploaded bytes for each file to handle out-of-order uploads
-const uploadedBytes = new Map<string, Set<number>>()
 
-const statUpdate = ({name, size, start, end}: {name: string, size: number, start: number, end: number}) => {
+const statUpdate = ({name, size, start, end}: UploadRange) => {
   if (name !== store.uprogress.filename) return  // If stats have been reset
-  const now = Date.now()
 
   // Track which bytes have been uploaded (using start to end range)
   if (!uploadedBytes.has(name)) uploadedBytes.set(name, new Set())
   const uploaded = uploadedBytes.get(name)!
-  const blockSize = 1 << 20
+  const blockSize = UPLOAD_BLOCK_SIZE
 
   // Mark all bytes in this block as uploaded
   for (let i = start; i < end; i += blockSize) {
     uploaded.add(i)
   }
-
-  // Calculate total uploaded bytes for progress
-  let totalUploaded = 0
-  for (let i = 0; i < size; i += blockSize) {
-    if (uploaded.has(i)) totalUploaded += blockSize
-  }
-
-  store.uprogress.xfer = store.uprogress.filestart + totalUploaded
-  store.uprogress.filepos = totalUploaded
-  store.uprogress.statbytes += end - start
-  store.uprogress.statdur += now - store.uprogress.tlast
-  store.uprogress.tlast = now
+  refreshProgress(Date.now())
 
   // Check if file is fully uploaded by examining the block queue
   const currentUpload = blockQueue[0]
   if (!currentUpload) return
 
-  if (currentUpload.file.cloudName === name && currentUpload.blockIndex >= currentUpload.blocks.length) {
+  if (currentUpload.file.cloudName === name && currentUpload.completed >= currentUpload.blocks.length) {
     // All blocks for this file have been uploaded
     uploadedBytes.delete(name)  // Clean up tracking
     store.uprogress.filestart += size
@@ -210,11 +277,15 @@ const statNextFile = () => {
   store.uprogress.filepos = 0
   store.uprogress.filesize = f.file.size
   store.uprogress.filename = f.cloudName
+  lastVisualUploaded = 0
+  lastProgressTick = Date.now()
 }
 const statReset = () => {
   Object.assign(store.uprogress, uprogress_init)
   store.uprogress.t0 = Date.now()
   store.uprogress.tlast = store.uprogress.t0 + 1
+  lastVisualUploaded = 0
+  lastProgressTick = store.uprogress.t0
 }
 const statsAdd = (f: CloudFile[]) => {
   if (store.uprogress.files.length === 0) statReset()
@@ -224,10 +295,12 @@ const statsAdd = (f: CloudFile[]) => {
   statNextFile()
 }
 let upqueue = [] as CloudFile[]
+const MAX_PARALLEL_REQUESTS = 4
+const RETRY_DELAY_MS = 400
 
 // Helper function to get upload blocks for a file, prioritizing final 4 blocks if file >= 32 MiB
 const getUploadBlocks = (file: CloudFile): {start: number, end: number}[] => {
-  const BLOCK_SIZE = 1 << 20  // 1 MiB
+  const BLOCK_SIZE = UPLOAD_BLOCK_SIZE
   const MIN_SIZE_FOR_REORDER = 32 * BLOCK_SIZE  // 32 MiB = 33554432 bytes
   const FINAL_BLOCKS_COUNT = 2
 
@@ -261,95 +334,162 @@ const getUploadBlocks = (file: CloudFile): {start: number, end: number}[] => {
   return blocks
 }
 
-// TODO: Rewrite as WebSocket class
-const WSCreate = async () => await new Promise<WebSocket>(resolve => {
-  const ws = connect(uploadUrl, {
-    open(ev: Event) { resolve(ws) },
-    error(ev: Event) {
-      console.error('Upload socket error', ev)
-      store.error = 'Upload socket error'
-    },
-    message(ev: MessageEvent) {
-      const res = JSON.parse(ev!.data)
-      if ('error' in res) {
-        console.error('Upload socket error', res.error)
-        store.error = res.error.message
-        return
-      }
-      if (res.status === 'ack') {
-        statUpdate(res.req)
-      } else console.log('Unknown upload response', res)
-    },
-  })
-  // @ts-ignore
-  ws.sendMsg = (msg: any) => ws.send(JSON.stringify(msg))
-  // @ts-ignore
-  ws.sendData = async (data: any) => {
-    // Wait until the WS is ready to send another message
-    store.uprogress.status = "uploading"
-    await new Promise(resolve => {
-      const t = setInterval(() => {
-        if (ws.bufferedAmount > 1<<20) return
-        resolve(undefined)
-        clearInterval(t)
-      }, 1)
-    })
-    store.uprogress.status = "processing"
-    ws.send(data)
-  }
-})
-
 type BlockUpload = {
   file: CloudFile
   blocks: {start: number, end: number}[]
-  blockIndex: number
+  nextIndex: number
+  completed: number
+  runId: number
 }
 
 let blockQueue = [] as BlockUpload[]
+let workerRunning = false
+let uploadRunId = 0
 
-const worker = async () => {
-  const ws = await WSCreate()
-  while (blockQueue.length) {
-    const upload = blockQueue[0]!
-    const f = upload.file
-    const block = upload.blocks[upload.blockIndex]!
+const enqueuePendingUploads = () => {
+  while (upqueue.length) {
+    const file = upqueue.shift()!
+    const blocks = getUploadBlocks(file)
+    blockQueue.push({ file, blocks, nextIndex: 0, completed: 0, runId: uploadRunId })
+  }
+}
 
-    const control = { name: f.cloudName, size: f.file.size, start: block.start, end: block.end }
-    const data = f.file.slice(block.start, block.end)
+const uploadUrlForFile = (cloudName: string) => {
+  const normalized = cloudName.replace(/^\/+/, '')
+  const encoded = normalized.split('/').map(encodeURIComponent).join('/')
+  return `/files/${encoded}`
+}
 
-    // Note: files may get modified during I/O
-    // @ts-ignore FIXME proper WebSocket class, avoid attaching functions to WebSocket object
-    ws.sendMsg(control)
-    // @ts-ignore
-    await ws.sendData(data)
+const uploadBlock = async (upload: BlockUpload, block: {start: number, end: number}) => {
+  const body = upload.file.file.slice(block.start, block.end)
+  const range = `bytes ${block.start}-${block.end - 1}/${upload.file.file.size}`
+  const fallbackReq = {
+    name: upload.file.cloudName,
+    size: upload.file.file.size,
+    start: block.start,
+    end: block.end,
+  }
+  let attempt = 0
 
-    // Move to next block
-    upload.blockIndex++
-    if (upload.blockIndex >= upload.blocks.length) {
-      // File upload complete
-      blockQueue.shift()
+  while (true) {
+    attempt += 1
+    if (upload.runId !== uploadRunId) throw new Error('Upload cancelled')
+    try {
+      const res = await fetch(uploadUrlForFile(upload.file.cloudName), {
+        method: 'PUT',
+        headers: {
+          'Content-Type': 'application/octet-stream',
+          'Content-Range': range,
+        },
+        body,
+      })
+      if (!res.ok) {
+        const message = await res.text().catch(() => '')
+        const retryable = res.status >= 500 || res.status === 408 || res.status === 429
+        if (!retryable) throw new Error(message || `HTTP ${res.status}`)
+        await new Promise(resolve => setTimeout(resolve, RETRY_DELAY_MS))
+        continue
+      }
+      const payload = await res.json().catch(() => null)
+      return payload?.status === 'ack' && payload.req ? payload.req : fallbackReq
+    } catch (err: any) {
+      const message = err instanceof Error ? err.message : String(err)
+      if (message === 'Upload cancelled') throw err
+      if (upload.runId !== uploadRunId) throw new Error('Upload cancelled')
+      if (attempt % 10 === 0) {
+        console.warn(`Upload retry ${attempt} for ${upload.file.cloudName}: ${message}`)
+      }
+      await new Promise(resolve => setTimeout(resolve, RETRY_DELAY_MS))
     }
   }
-  if (blockQueue.length) startWorker()
-  store.uprogress.status = "idle"
-  workerRunning = false
 }
-let workerRunning: any = false
-const startWorker = () => {
-  if (workerRunning === false) workerRunning = setTimeout(() => {
-    // Convert new CloudFile entries to BlockUpload entries
-    while (upqueue.length) {
-      const file = upqueue.shift()!
-      const blocks = getUploadBlocks(file)
-      blockQueue.push({ file, blocks, blockIndex: 0 })
-    }
 
-    if (blockQueue.length) {
-      workerRunning = true
-      worker()
-    } else {
-      workerRunning = false
+const startInflightBlock = (name: string, block: {start: number, end: number}) => {
+  inflightBlocks.set(inflightKey(name, block.start), {
+    name,
+    start: block.start,
+    end: block.end,
+    startedAt: Date.now(),
+  })
+}
+
+const finishInflightBlock = (name: string, block: {start: number, end: number}) => {
+  const key = inflightKey(name, block.start)
+  const info = inflightBlocks.get(key)
+  if (!info) return
+  const elapsed = Math.max(1, Date.now() - info.startedAt)
+  smoothedBlockMs = smoothedBlockMs * 0.85 + elapsed * 0.15
+  inflightBlocks.delete(key)
+}
+
+const worker = async (runId: number) => {
+  try {
+    while (runId === uploadRunId) {
+      enqueuePendingUploads()
+      if (!blockQueue.length) break
+
+      const upload = blockQueue[0]!
+      const inflight = new Set<Promise<void>>()
+
+      while (runId === uploadRunId && upload.completed < upload.blocks.length) {
+        while (
+          runId === uploadRunId
+          && upload.nextIndex < upload.blocks.length
+          && inflight.size < MAX_PARALLEL_REQUESTS
+        ) {
+          const block = upload.blocks[upload.nextIndex++]!
+          store.uprogress.status = 'uploading'
+          startInflightBlock(upload.file.cloudName, block)
+          let task: Promise<void>
+          task = uploadBlock(upload, block)
+            .then(req => {
+              finishInflightBlock(upload.file.cloudName, block)
+              upload.completed += 1
+              statUpdate(req)
+            })
+            .catch(err => {
+              finishInflightBlock(upload.file.cloudName, block)
+              throw err
+            })
+            .finally(() => {
+              inflight.delete(task)
+            })
+          inflight.add(task)
+        }
+
+        if (!inflight.size) break
+        await Promise.race(inflight)
+      }
+
+      if (runId !== uploadRunId) return
+
+      if (upload.completed >= upload.blocks.length) {
+        blockQueue.shift()
+      } else {
+        break
+      }
     }
+  } catch (err: any) {
+    if (runId !== uploadRunId) return
+    console.error('Upload error', err)
+    store.error = err?.message || 'Upload failed'
+    uploadRunId += 1
+    upqueue = []
+    blockQueue = []
+    inflightBlocks.clear()
+  } finally {
+    store.uprogress.status = 'idle'
+    workerRunning = false
+    if (upqueue.length) startWorker()
+  }
+}
+
+const startWorker = () => {
+  if (workerRunning) return
+  workerRunning = true
+  const runId = uploadRunId
+  setTimeout(() => {
+    void worker(runId)
   }, 0)
 }
 

frontend/src/repositories/WS.ts

@@ -3,7 +3,6 @@ import { showAuthIframe, AuthCancelledError, isAuthIframeOpen } from 'paskia'
 import type { FileEntry, UpdateEntry, errorEvent } from "./Document"
 
 export const controlUrl = '/api/control'
-export const uploadUrl = '/api/upload'
 export const watchUrl = '/api/watch'
 
 let tree = [] as FileEntry[]