Implement PUT chunk uploads, 16MiB chunk size for faster transfers with resilient retries and smoother progress

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 import Blueprint, Sanic, empty, json, raw, redirect
-from sanic.exceptions import Forbidden, NotFound
+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,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]
+        with self.cache_lock:
-        f.write(pos, data, file_size=file_size)
+            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]
+        with self.cache_lock:
-        return f[start:end]
+            f = self.cache[name]
+            lock = self.file_locks.setdefault(name, threading.Lock())
+        with lock:
+            return f[start:end]

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) {
+  const now = Date.now()
-    // Reset
+  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 *= .95
-    store.uprogress.statbytes *= .9
+    store.uprogress.statdur *= .95
-    store.uprogress.statdur *= .9
   }
 }, 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)
   }
-
+  refreshProgress(Date.now())
-  // 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
 
   // 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 enqueuePendingUploads = () => {
-  const ws = await WSCreate()
+  while (upqueue.length) {
-  while (blockQueue.length) {
+    const file = upqueue.shift()!
-    const upload = blockQueue[0]!
+    const blocks = getUploadBlocks(file)
-    const f = upload.file
+    blockQueue.push({ file, blocks, nextIndex: 0, completed: 0, runId: uploadRunId })
-    const block = upload.blocks[upload.blockIndex]!
+  }
+}
 
-    const control = { name: f.cloudName, size: f.file.size, start: block.start, end: block.end }
+const uploadUrlForFile = (cloudName: string) => {
-    const data = f.file.slice(block.start, block.end)
+  const normalized = cloudName.replace(/^\/+/, '')
+  const encoded = normalized.split('/').map(encodeURIComponent).join('/')
+  return `/files/${encoded}`
+}
 
-    // Note: files may get modified during I/O
+const uploadBlock = async (upload: BlockUpload, block: {start: number, end: number}) => {
-    // @ts-ignore FIXME proper WebSocket class, avoid attaching functions to WebSocket object
+  const body = upload.file.file.slice(block.start, block.end)
-    ws.sendMsg(control)
+  const range = `bytes ${block.start}-${block.end - 1}/${upload.file.file.size}`
-    // @ts-ignore
+  const fallbackReq = {
-    await ws.sendData(data)
+    name: upload.file.cloudName,
+    size: upload.file.file.size,
+    start: block.start,
+    end: block.end,
+  }
+  let attempt = 0
 
-    // Move to next block
+  while (true) {
-    upload.blockIndex++
+    attempt += 1
-    if (upload.blockIndex >= upload.blocks.length) {
+    if (upload.runId !== uploadRunId) throw new Error('Upload cancelled')
-      // File upload complete
+    try {
-      blockQueue.shift()
+      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) {
+const startInflightBlock = (name: string, block: {start: number, end: number}) => {
-      workerRunning = true
+  inflightBlocks.set(inflightKey(name, block.start), {
-      worker()
+    name,
-    } else {
+    start: block.start,
-      workerRunning = false
+    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)
 }