From 91e07d61b65b03facc22eb0f3d5512200f5a1e46 Mon Sep 17 00:00:00 2001
From: Leo Vasanko <leovasanko@git>
Date: Sat, 25 Apr 2026 03:55:52 +0100
Subject: [PATCH] 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.

Reviewed-on: https://git.zi.fi/Vasanko/cista-storage/pulls/9
---
 cista/api.py                             |  62 +---
 cista/app.py                             |  93 +++++-
 cista/fileio.py                          |  53 ++--
 cista/protocol.py                        |  13 +-
 frontend/src/components/UploadButton.vue | 344 ++++++++++++++++-------
 frontend/src/repositories/WS.ts          |   1 -
 6 files changed, 357 insertions(+), 209 deletions(-)

diff --git a/cista/api.py b/cista/api.py
index 93aaa37..aa764b9 100644
--- a/cista/api.py
+++ b/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):
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/<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):
diff --git a/cista/fileio.py b/cista/fileio.py
index b9052f8..8b340da 100644
--- a/cista/fileio.py
+++ b/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,
+        }
diff --git a/cista/protocol.py b/cista/protocol.py
index c04147a..220e887 100644
--- a/cista/protocol.py
+++ b/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):
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 @@
 </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)
 }
 
diff --git a/frontend/src/repositories/WS.ts b/frontend/src/repositories/WS.ts
index 3c8d37d..69e7da8 100644
--- a/frontend/src/repositories/WS.ts
+++ b/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[]