feat: optimize chunked uploads and concurrent downloads

domain/adapters/providers/local.py

@@ -310,6 +310,7 @@ class LocalAdapter:
             "is_dir": is_dir,
             "size": st.st_size,
             "mtime": int(st.st_mtime),
+            "mtime_ns": st.st_mtime_ns,
             "mode": stat.S_IMODE(st.st_mode),
             "type": "dir" if is_dir else "file",
             "path": str(fp),

domain/adapters/providers/s3.py

@@ -1,11 +1,14 @@
 import asyncio
+import logging
 import mimetypes
+import re
 from datetime import datetime
 from typing import List, Dict, Tuple, AsyncIterator
 from urllib.parse import quote
 
 import aioboto3
 from botocore.exceptions import ClientError
+from botocore.config import Config
 from fastapi import HTTPException
 from fastapi.responses import StreamingResponse
 from models import StorageAdapter
@@ -23,6 +26,10 @@ class S3Adapter:
         self.region_name = cfg.get("region_name")
         self.endpoint_url = cfg.get("endpoint_url")
         self.root = cfg.get("root", "").strip("/")
+        self.multipart_part_size = max(5, min(64, int(cfg.get("multipart_part_size_mb") or 8))) * 1024 * 1024
+        self.multipart_concurrency = max(1, min(16, int(cfg.get("multipart_concurrency") or 4)))
+        self.download_segment_size = max(1, min(64, int(cfg.get("download_segment_size_mb") or 8))) * 1024 * 1024
+        self.download_concurrency = max(1, min(16, int(cfg.get("download_concurrency") or 4)))
 
         if not all([self.bucket_name, self.aws_access_key_id, self.aws_secret_access_key]):
             raise ValueError(
@@ -48,7 +55,10 @@ class S3Adapter:
         return rel_path
 
     def _get_client(self):
-        return self.session.client("s3", endpoint_url=self.endpoint_url)
+        return self.session.client(
+            "s3", endpoint_url=self.endpoint_url,
+            config=Config(max_pool_connections=max(10, self.multipart_concurrency, self.download_concurrency)),
+        )
 
     async def list_dir(self, root: str, rel: str, page_num: int = 1, page_size: int = 50, sort_by: str = "name", sort_order: str = "asc") -> Tuple[List[Dict], int]:
         prefix = self._get_s3_key(rel)
@@ -128,62 +138,80 @@ class S3Adapter:
 
     async def write_file_stream(self, root: str, rel: str, data_iter: AsyncIterator[bytes]):
         key = self._get_s3_key(rel)
-        MIN_PART_SIZE = 5 * 1024 * 1024
-        
+
         async with self._get_client() as s3:
             mpu = await s3.create_multipart_upload(Bucket=self.bucket_name, Key=key)
             upload_id = mpu['UploadId']
-            
+
             parts = []
+            pending = set()
             part_number = 1
             total_size = 0
             buffer = bytearray()
-            
+
+            async def upload_part(number: int, body: bytes):
+                result = await s3.upload_part(
+                    Bucket=self.bucket_name, Key=key, PartNumber=number,
+                    UploadId=upload_id, Body=body,
+                )
+                parts.append({'PartNumber': number, 'ETag': result['ETag']})
+
+            async def schedule_part(body: bytes):
+                nonlocal part_number, total_size
+                if part_number > 10000:
+                    raise ValueError("Too many S3 parts; increase multipart_part_size_mb")
+                # Backpressure bounds the number of retained part buffers.
+                if len(pending) >= self.multipart_concurrency:
+                    done, _ = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
+                    for task in done:
+                        task.result()
+                    pending.difference_update(done)
+                pending.add(asyncio.create_task(upload_part(part_number, body)))
+                total_size += len(body)
+                part_number += 1
+
             try:
                 async for chunk in data_iter:
                     if not chunk:
                         continue
-                    buffer.extend(chunk)
-                    
-                    while len(buffer) >= MIN_PART_SIZE:
-                        part_data = buffer[:MIN_PART_SIZE]
-                        del buffer[:MIN_PART_SIZE]
-                        
-                        part = await s3.upload_part(
-                            Bucket=self.bucket_name,
-                            Key=key,
-                            PartNumber=part_number,
-                            UploadId=upload_id,
-                            Body=part_data
-                        )
-                        
-                        parts.append({'PartNumber': part_number, 'ETag': part['ETag']})
-                        total_size += len(part_data)
-                        part_number += 1
+                    view = memoryview(chunk)
+                    offset = 0
+                    while offset < len(view):
+                        take = min(self.multipart_part_size - len(buffer), len(view) - offset)
+                        buffer.extend(view[offset:offset + take])
+                        offset += take
+                        if len(buffer) == self.multipart_part_size:
+                            part_data = bytes(buffer)
+                            buffer.clear()
+                            await schedule_part(part_data)
 
                 if buffer:
-                    part = await s3.upload_part(
-                        Bucket=self.bucket_name,
-                        Key=key,
-                        PartNumber=part_number,
-                        UploadId=upload_id,
-                        Body=bytes(buffer)
-                    )
-                    parts.append({'PartNumber': part_number, 'ETag': part['ETag']})
-                    total_size += len(buffer)
-                
+                    await schedule_part(bytes(buffer))
+                if pending:
+                    await asyncio.gather(*pending)
+                if not parts:
+                    await s3.abort_multipart_upload(Bucket=self.bucket_name, Key=key, UploadId=upload_id)
+                    await s3.put_object(Bucket=self.bucket_name, Key=key, Body=b"")
+                    return 0
+
                 await s3.complete_multipart_upload(
                     Bucket=self.bucket_name,
                     Key=key,
                     UploadId=upload_id,
-                    MultipartUpload={'Parts': parts}
-                )
-            except Exception as e:
-                await s3.abort_multipart_upload(
-                    Bucket=self.bucket_name,
-                    Key=key,
-                    UploadId=upload_id
+                    MultipartUpload={'Parts': sorted(parts, key=lambda part: part['PartNumber'])}
                 )
+            except BaseException as e:
+                for task in pending:
+                    task.cancel()
+                await asyncio.gather(*pending, return_exceptions=True)
+                try:
+                    await s3.abort_multipart_upload(
+                        Bucket=self.bucket_name, Key=key, UploadId=upload_id,
+                    )
+                except Exception:
+                    logging.getLogger(__name__).warning("Failed to abort S3 multipart upload", exc_info=True)
+                if not isinstance(e, Exception):
+                    raise
                 raise IOError(f"S3 stream upload failed: {e}") from e
 
         return total_size
@@ -254,6 +282,7 @@ class S3Adapter:
                     "is_dir": False,
                     "size": head["ContentLength"],
                     "mtime": int(head["LastModified"].timestamp()),
+                    "etag": head.get("ETag"),
                     "type": "file",
                 }
             except ClientError as e:
@@ -297,31 +326,74 @@ class S3Adapter:
             }
 
             if range_header:
-                range_val = range_header.strip().partition("=")[2]
-                s, _, e = range_val.partition("-")
-                try:
-                    start = int(s) if s else 0
-                    end = int(e) if e else file_size - 1
-                    if start >= file_size or end >= file_size or start > end:
-                        raise HTTPException(
-                            status_code=416, detail="Requested Range Not Satisfiable")
-                    status = 206
-                    headers["Content-Length"] = str(end - start + 1)
-                    headers["Content-Range"] = f"bytes {start}-{end}/{file_size}"
-                except ValueError:
-                    raise HTTPException(
-                        status_code=400, detail="Invalid Range header")
+                match = re.fullmatch(r"bytes=(\d*)-(\d*)", range_header.strip())
+                if not match or not any(match.groups()):
+                    raise HTTPException(400, detail="Invalid Range header")
+                first, last = match.groups()
+                if first:
+                    start = int(first)
+                    end = min(int(last), end) if last else end
+                else:
+                    suffix = int(last)
+                    start = max(0, file_size - suffix) if suffix else file_size
+                if start > end:
+                    raise HTTPException(416, detail="Requested Range Not Satisfiable",
+                                        headers={"Content-Range": f"bytes */{file_size}"})
+                status = 206
+                headers["Content-Length"] = str(end - start + 1)
+                headers["Content-Range"] = f"bytes {start}-{end}/{file_size}"
 
-            range_arg = f"bytes={start}-{end}"
+            segment_size = self.download_segment_size
+            segment_concurrency = self.download_concurrency
+            etag = head.get("ETag")
 
             async def iterator():
-                try:
-                    resp = await s3.get_object(Bucket=self.bucket_name, Key=key, Range=range_arg)
-                    body = resp["Body"]
-                    while chunk := await body.read(65536):
-                        yield chunk
-                except Exception as e:
-                    raise
+                # Reuse one async S3 client while fetching a bounded window of
+                # ordered ranges.  This keeps memory bounded and lets object
+                # stores serve large downloads over several connections.
+                async with self._get_client() as download_s3:
+                    async def fetch(seg_start: int, seg_end: int) -> bytes:
+                        options = {"IfMatch": etag} if etag else {}
+                        response = await download_s3.get_object(
+                            Bucket=self.bucket_name,
+                            Key=key,
+                            Range=f"bytes={seg_start}-{seg_end}",
+                            **options,
+                        )
+                        body = response["Body"]
+                        expected_size = seg_end - seg_start + 1
+                        try:
+                            if (response.get("ContentRange") != f"bytes {seg_start}-{seg_end}/{file_size}"
+                                    or response.get("ContentLength") != expected_size):
+                                raise IOError("S3 returned an invalid range response")
+                            data = bytearray()
+                            while chunk := await body.read(min(1024 * 1024, expected_size + 1 - len(data))):
+                                data.extend(chunk)
+                                if len(data) > expected_size:
+                                    raise IOError("S3 range exceeded requested size")
+                            if len(data) != expected_size:
+                                raise IOError("Incomplete S3 range response")
+                            return bytes(data)
+                        finally:
+                            body.close()
+
+                    pending: list[asyncio.Task[bytes]] = []
+                    next_start = start
+                    try:
+                        while next_start <= end or pending:
+                            while next_start <= end and len(pending) < segment_concurrency:
+                                next_end = min(next_start + segment_size - 1, end)
+                                pending.append(asyncio.create_task(fetch(next_start, next_end)))
+                                next_start = next_end + 1
+                            if not pending:
+                                break
+                            data = await pending[0]
+                            pending.pop(0)
+                            yield data
+                    finally:
+                        for task in pending:
+                            task.cancel()
+                        await asyncio.gather(*pending, return_exceptions=True)
 
         return StreamingResponse(iterator(), status_code=status, headers=headers, media_type=content_type)
 
@@ -329,6 +401,14 @@ class S3Adapter:
 ADAPTER_TYPE = "s3"
 
 CONFIG_SCHEMA = [
+    {"key": "download_segment_size_mb", "label": "下载分片大小 (MiB)",
+     "type": "number", "required": False, "default": 8},
+    {"key": "download_concurrency", "label": "下载并发数",
+     "type": "number", "required": False, "default": 4},
+    {"key": "multipart_part_size_mb", "label": "上传分片大小 (MiB)",
+     "type": "number", "required": False, "default": 8},
+    {"key": "multipart_concurrency", "label": "上传并发数",
+     "type": "number", "required": False, "default": 4},
     {"key": "bucket_name", "label": "Bucket 名称",
         "type": "string", "required": True},
     {"key": "access_key_id", "label": "Access Key ID",

domain/adapters/providers/webdav.py

@@ -1,11 +1,14 @@
 from typing import List, Dict, Optional, Tuple, AsyncIterator
+import asyncio
+from collections import deque
+import re
 import httpx
 from urllib.parse import urljoin, quote
 from urllib.parse import urlparse, unquote
 import xml.etree.ElementTree as ET
 from models import StorageAdapter
 import mimetypes
-import logging
+from starlette.background import BackgroundTask
 from fastapi import HTTPException
 from fastapi.responses import StreamingResponse, Response
 
@@ -22,6 +25,8 @@ class WebDAVAdapter:
         self.username = cfg.get("username")
         self.password = cfg.get("password")
         self.timeout = cfg.get("timeout", 15)
+        self.download_segment_size = max(1, min(64, int(cfg.get("download_segment_size_mb") or 8))) * 1024 * 1024
+        self.download_concurrency = max(1, min(16, int(cfg.get("download_concurrency") or 4)))
 
     def get_effective_root(self, sub_path: str | None) -> str:
         base_url = self.record.config.get("base_url", "").rstrip('/') + '/'
@@ -235,146 +240,149 @@ class WebDAVAdapter:
         url = self._build_url(rel)
         mime, _ = mimetypes.guess_type(rel)
         content_type = mime or "application/octet-stream"
-        logger = logging.getLogger(__name__)
-        timeout = self.timeout
-        auth = (self.username, self.password) if self.username else None
+        range_match = None
+        if range_header:
+            range_match = re.fullmatch(r"bytes=(\d*)-(\d*)", range_header.strip())
+            if not range_match or not any(range_match.groups()):
+                raise HTTPException(400, detail="Invalid Range header")
 
-        client_start = 0
-        client_end = None
-        status_code = 200
-        if range_header and range_header.startswith("bytes="):
-            status_code = 206
-            part = range_header.removeprefix("bytes=")
-            s, e = part.split("-", 1)
-            if s.strip():
-                client_start = int(s)
-            if e.strip():
-                client_end = int(e)
-
-        total_size = None
-        accept_ranges = False
-        async with httpx.AsyncClient(timeout=timeout, auth=auth, follow_redirects=True) as client:
-            try:
-                head_resp = await client.head(url)
-                if head_resp.status_code == 404:
+        client = self._client()
+        identity_headers = {"Accept-Encoding": "identity"}
+        try:
+            total_size = None
+            validator = None
+            accept_ranges = False
+            # Probe the real response: Accept-Ranges alone is not reliable.
+            async with client.stream("GET", url, headers={**identity_headers, "Range": "bytes=0-0"}) as probe:
+                if probe.status_code == 404:
                     raise HTTPException(404, detail="File not found")
-                if head_resp.status_code == 200:
-                    cl = head_resp.headers.get("Content-Length")
-                    if cl and cl.isdigit():
-                        total_size = int(cl)
-                    ar = head_resp.headers.get("Accept-Ranges", "").lower()
-                    accept_ranges = "bytes" in ar
-            except HTTPException:
-                raise
-            except Exception as e:
-                logger.debug("HEAD failed %s err=%s", url, e)
-            if total_size is None and (client_end is None):
-                try:
-                    probe_req = client.build_request("GET", url, headers={"Range": "bytes=0-0"})
-                    probe_resp = await client.send(probe_req, stream=True)
-                    if probe_resp.status_code in (200, 206):
-                        cr = probe_resp.headers.get("Content-Range")
-                        if cr and "/" in cr:
-                            try:
-                                total_size = int(cr.rsplit("/", 1)[1])
-                            except Exception:
-                                pass
-                    await probe_resp.aclose()
-                except Exception as e:
-                    logger.debug("Probe 0-0 failed %s err=%s", url, e)
+                if probe.status_code == 206:
+                    match = re.fullmatch(r"bytes 0-0/(\d+)", probe.headers.get("Content-Range", ""))
+                    if match and int(match[1]) > 0:
+                        total_size = int(match[1])
+                        accept_ranges = True
+                        etag = probe.headers.get("ETag")
+                        validator = etag if etag and not etag.startswith("W/") else probe.headers.get("Last-Modified")
+                elif probe.status_code >= 400 and probe.status_code != 416:
+                    raise HTTPException(probe.status_code, detail="Upstream download failed")
 
-        if total_size is not None and client_end is None:
-            client_end = total_size - 1
-        if client_end is not None and client_end < client_start:
-            raise HTTPException(416, detail="Requested Range Not Satisfiable")
-
-        # 若客户端未请求范围且上游不支持 Range，直接透传
-        if status_code == 200 and (range_header is None) and not accept_ranges:
-            async with httpx.AsyncClient(timeout=timeout, auth=auth, follow_redirects=True) as client:
-                req = client.build_request("GET", url)
-                resp = await client.send(req, stream=True)
-                if resp.status_code == 404:
-                    await resp.aclose()
-                    raise HTTPException(404, detail="File not found")
-                upstream_ct = resp.headers.get("Content-Type", content_type)
+            if not accept_ranges:
+                request_headers = dict(identity_headers)
+                if range_header:
+                    request_headers["Range"] = range_header
+                upstream = await client.send(client.build_request("GET", url, headers=request_headers), stream=True)
+                if upstream.status_code not in (200, 206):
+                    await upstream.aclose()
+                    raise HTTPException(upstream.status_code, detail="Upstream download failed")
+                headers = {"X-VFS-Remote-Status": str(upstream.status_code)}
+                for name in ("Content-Length", "Content-Range", "Accept-Ranges", "Content-Encoding"):
+                    if name in upstream.headers:
+                        headers[name] = upstream.headers[name]
 
                 async def passthrough():
                     try:
-                        async for chunk in resp.aiter_bytes():
-                            if chunk:
-                                yield chunk
+                        async for chunk in upstream.aiter_raw():
+                            yield chunk
                     finally:
-                        await resp.aclose()
-                return StreamingResponse(passthrough(), status_code=resp.status_code,
-                                         headers={"Accept-Ranges": "bytes",
-                                                  "X-VFS-Remote-Status": str(resp.status_code)},
-                                         media_type=upstream_ct)
+                        await upstream.aclose()
+                        await client.aclose()
 
-        SEGMENT_SIZE = 5 * 1024 * 1024
-        MAX_RETRY_PER_SEG = 3
-        FIRST_BYTE_MAX_RETRY = 3
+                return StreamingResponse(
+                    passthrough(), status_code=upstream.status_code, headers=headers,
+                    media_type=upstream.headers.get("Content-Type", content_type),
+                    background=BackgroundTask(client.aclose),
+                )
 
-        resp_headers = {
-            "Accept-Ranges": "bytes",
-            "Content-Type": content_type,
-            "X-VFS-Segmented": "1",
-        }
-        if status_code == 206 and total_size is not None:
-            resp_headers["Content-Range"] = f"bytes {client_start}-{client_end}/{total_size}"
+            start, end = 0, total_size - 1
+            if range_match:
+                first, last = range_match.groups()
+                if first:
+                    start = int(first)
+                    end = min(int(last), end) if last else end
+                else:
+                    suffix = int(last)
+                    if suffix <= 0:
+                        raise HTTPException(416, detail="Requested Range Not Satisfiable",
+                                            headers={"Content-Range": f"bytes */{total_size}"})
+                    start = max(0, total_size - suffix)
+                if start > end:
+                    raise HTTPException(416, detail="Requested Range Not Satisfiable",
+                                        headers={"Content-Range": f"bytes */{total_size}"})
 
-        async def segmented_body():
-            current = client_start
-            first_byte_sent = False
-            while True:
-                if client_end is not None and current > client_end:
-                    break
-                seg_start = current
-                seg_end = (min(seg_start + SEGMENT_SIZE - 1, client_end)
-                           if client_end is not None else seg_start + SEGMENT_SIZE - 1)
-                attempt = 0
-                ok = False
-                while attempt < MAX_RETRY_PER_SEG and not ok:
-                    attempt += 1
-                    headers_req = {"Range": f"bytes={seg_start}-{seg_end}"}
+            async def fetch_segment(seg_start: int, seg_end: int) -> bytes:
+                headers = {**identity_headers, "Range": f"bytes={seg_start}-{seg_end}"}
+                if validator:
+                    headers["If-Range"] = validator
+                for attempt in range(3):
                     try:
-                        async with httpx.AsyncClient(timeout=timeout, auth=auth, follow_redirects=True) as cseg:
-                            req = cseg.build_request("GET", url, headers=headers_req)
-                            rseg = await cseg.send(req, stream=True)
-                            if rseg.status_code in (200, 206):
-                                async for chunk in rseg.aiter_bytes():
-                                    if chunk:
-                                        first_byte_sent = True
-                                        yield chunk
-                                await rseg.aclose()
-                                ok = True
-                            elif rseg.status_code == 404:
-                                await rseg.aclose()
-                                if not first_byte_sent:
-                                    raise HTTPException(404, detail="File not found")
-                                return
-                            else:
-                                await rseg.aclose()
-                                logger.warning("Segment unexpected status %s %s-%s %s", rel, seg_start, seg_end, rseg.status_code)
-                        if not ok:
-                            continue
-                    except (httpx.ReadError, httpx.HTTPError, httpx.StreamError) as e:
-                        if not first_byte_sent and attempt >= FIRST_BYTE_MAX_RETRY:
-                            raise HTTPException(502, detail=f"Upstream error before first byte err={e}")
-                        logger.warning("Segment error %s %s-%s attempt=%d err=%s", rel, seg_start, seg_end, attempt, e)
-                    except Exception as e:
-                        if not first_byte_sent:
-                            raise
-                        logger.error("Segment unexpected %s %s-%s attempt=%d err=%s", rel, seg_start, seg_end, attempt, e)
-                if not ok:
-                    logger.error("Abort streaming %s at %s-%s", rel, seg_start, seg_end)
-                    break
-                current = seg_end + 1
-                if client_end is None:
-                    continue
-                if current > client_end:
-                    break
+                        async with client.stream("GET", url, headers=headers) as response:
+                            if response.status_code == 404:
+                                raise HTTPException(404, detail="File not found")
+                            if response.status_code == 429 or response.status_code >= 500:
+                                raise httpx.HTTPStatusError("Upstream temporarily unavailable",
+                                                            request=response.request, response=response)
+                            expected_range = f"bytes {seg_start}-{seg_end}/{total_size}"
+                            if (response.status_code != 206
+                                    or response.headers.get("Content-Range") != expected_range
+                                    or response.headers.get("Content-Encoding", "identity") != "identity"):
+                                raise HTTPException(502, detail="Upstream returned an invalid range response")
+                            data = bytearray()
+                            expected_size = seg_end - seg_start + 1
+                            async for chunk in response.aiter_raw():
+                                if len(data) + len(chunk) > expected_size:
+                                    raise HTTPException(502, detail="Upstream range exceeded requested size")
+                                data.extend(chunk)
+                            if len(data) != expected_size:
+                                raise httpx.ReadError("Incomplete range response")
+                            return bytes(data)
+                    except httpx.HTTPError as exc:
+                        if attempt == 2:
+                            raise HTTPException(502, detail="Upstream segment download failed") from exc
+                        await asyncio.sleep(0.25 * 2 ** attempt)
 
-        return StreamingResponse(segmented_body(), status_code=status_code, headers=resp_headers, media_type=content_type)
+            # Validate the first segment before sending response headers.
+            first_end = min(start + self.download_segment_size - 1, end)
+            first_segment = await fetch_segment(start, first_end)
+            headers = {"Accept-Ranges": "bytes", "Content-Length": str(end - start + 1),
+                       "X-VFS-Segmented": "1"}
+            if range_match:
+                headers["Content-Range"] = f"bytes {start}-{end}/{total_size}"
+
+            async def segmented_body():
+                pending = deque()
+                next_start = first_end + 1
+
+                def fill_window():
+                    nonlocal next_start
+                    while next_start <= end and len(pending) < self.download_concurrency:
+                        next_end = min(next_start + self.download_segment_size - 1, end)
+                        pending.append(asyncio.create_task(fetch_segment(next_start, next_end)))
+                        next_start = next_end + 1
+
+                try:
+                    fill_window()
+                    yield first_segment
+                    while pending:
+                        # Finished segments stay in order and occupy a window slot
+                        # until consumed, bounding memory even for a slow client.
+                        data = await pending[0]
+                        pending.popleft()
+                        yield data
+                        fill_window()
+                finally:
+                    for task in pending:
+                        task.cancel()
+                    await asyncio.gather(*pending, return_exceptions=True)
+                    await client.aclose()
+
+            return StreamingResponse(
+                segmented_body(), status_code=206 if range_match else 200,
+                headers=headers, media_type=content_type,
+                background=BackgroundTask(client.aclose),
+            )
+        except BaseException:
+            await client.aclose()
+            raise
 
     async def stat_file(self, root: str, rel: str, include_metadata: bool = False):
         url = self._build_url(rel)
@@ -481,6 +489,10 @@ class WebDAVAdapter:
 
 ADAPTER_TYPE = "webdav"
 CONFIG_SCHEMA = [
+    {"key": "download_segment_size_mb", "label": "下载分片大小 (MiB)",
+        "type": "number", "required": False, "default": 8},
+    {"key": "download_concurrency", "label": "下载并发数",
+        "type": "number", "required": False, "default": 4},
     {"key": "base_url", "label": "基础地址", "type": "string",
         "required": True, "placeholder": "https://example.com/dav/"},
     {"key": "username", "label": "用户名", "type": "string", "required": False},

domain/virtual_fs/api.py

@@ -1,6 +1,11 @@
 from typing import Annotated
+import hashlib
+import json
+import mimetypes
+from urllib.parse import quote
 
-from fastapi import APIRouter, Depends, File, Query, Request, UploadFile
+from fastapi import APIRouter, Depends, File, HTTPException, Query, Request, UploadFile
+from fastapi.responses import Response
 
 from api.response import success
 from domain.audit import AuditAction, audit
@@ -9,10 +14,99 @@ from domain.permission import require_path_permission
 from domain.permission.types import PathAction
 from .service import VirtualFSService
 from .types import MkdirRequest, MoveRequest
+from .chunk_uploads import ChunkUploadCreate, ChunkUploadService
 
 router = APIRouter(prefix="/api/fs", tags=["virtual-fs"])
 
 
+@router.post("/uploads")
+async def create_chunk_upload(
+    body: ChunkUploadCreate,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+):
+    return success(await ChunkUploadService.create(body, current_user.id))
+
+
+@router.put("/uploads/{upload_id}/parts/{part}")
+async def upload_chunk(
+    upload_id: str,
+    part: int,
+    request: Request,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+):
+    return success(await ChunkUploadService.put_part(upload_id, part, current_user.id, request.stream()))
+
+
+@router.get("/uploads/{upload_id}")
+async def chunk_upload_status(
+    upload_id: str,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+):
+    return success(await ChunkUploadService.status(upload_id, current_user.id))
+
+
+@router.post("/uploads/{upload_id}/complete")
+@audit(action=AuditAction.UPLOAD, description="Complete chunked upload")
+async def complete_chunk_upload(
+    upload_id: str,
+    request: Request,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    overwrite: bool | None = Query(None),
+):
+    return success(await ChunkUploadService.complete(upload_id, current_user.id, overwrite))
+
+
+@router.delete("/uploads/{upload_id}")
+async def abort_chunk_upload(
+    upload_id: str,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+):
+    return success(await ChunkUploadService.abort(upload_id, current_user.id))
+
+
+@router.get("/download/{full_path:path}")
+@audit(action=AuditAction.DOWNLOAD, description="Download original file")
+@require_path_permission(PathAction.READ, "full_path")
+async def download_original(
+    full_path: str,
+    request: Request,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+):
+    return await _download_original_response(full_path, request)
+
+
+@router.get("/download-public/{token}/{filename}")
+@audit(action=AuditAction.DOWNLOAD, description="Download original file through temporary link")
+async def download_original_public(token: str, filename: str, request: Request):
+    path = await VirtualFSService.verify_temp_link_token(token)
+    return await _download_original_response(path, request)
+
+
+async def _download_original_response(full_path: str, request: Request):
+    full_path = VirtualFSService._normalize_path(full_path)
+    try:
+        stat = await VirtualFSService.stat_file(full_path)
+    except FileNotFoundError:
+        raise HTTPException(404, detail="File not found")
+    if stat.get("is_dir"):
+        raise HTTPException(400, detail="Path is a directory")
+    # Pin the metadata version across browser range requests. Providers such as
+    # S3 additionally pin their native object version while fetching each range.
+    etag = '"' + hashlib.sha256(json.dumps(stat, sort_keys=True, default=str).encode()).hexdigest() + '"'
+    if request.headers.get("If-Match") not in (None, etag):
+        raise HTTPException(412, detail="File changed during download")
+    adapter, _, root, rel = await VirtualFSService.resolve_adapter_and_rel(full_path)
+    stream = getattr(adapter, "stream_file", None)
+    if callable(stream):
+        response = await stream(root, rel, request.headers.get("Range"))
+    else:
+        data = await VirtualFSService.read_file(full_path)
+        response = Response(data, media_type=mimetypes.guess_type(full_path)[0] or "application/octet-stream")
+    response.headers["ETag"] = etag
+    response.headers["Content-Disposition"] = "attachment; filename*=UTF-8''" + quote(full_path.rstrip("/").split("/")[-1], safe="")
+    return response
+
+
 @router.get("/file/{full_path:path}")
 @audit(action=AuditAction.DOWNLOAD, description="获取文件")
 @require_path_permission(PathAction.READ, "full_path")

domain/virtual_fs/chunk_uploads.py

@@ -0,0 +1,212 @@
+import asyncio
+from contextlib import asynccontextmanager
+import json
+import math
+from pathlib import Path
+import re
+import shutil
+import time
+import uuid
+
+import aiofiles
+import portalocker
+from fastapi import HTTPException
+from pydantic import BaseModel, Field
+
+from domain.permission import PermissionService
+from domain.permission.execution import ExecutionError, normalize_path
+from domain.permission.types import PathAction
+from .service import VirtualFSService
+
+
+class ChunkUploadCreate(BaseModel):
+    path: str
+    size: int = Field(ge=1, le=8 * 1024 * 1024 * 10000)
+    overwrite: bool = True
+
+
+class ChunkUploadService:
+    root = Path("data/tmp/web_uploads")
+    chunk_size = 8 * 1024 * 1024
+    lifetime = 24 * 60 * 60
+
+    @classmethod
+    @asynccontextmanager
+    async def _lock(cls, directory: Path, shared: bool = False):
+        # OS locks coordinate multiple server workers and release on crashes.
+        try:
+            file = open(directory / "lock", "a+b")
+        except FileNotFoundError:
+            raise HTTPException(404, detail="Upload session not found")
+        flags = portalocker.LOCK_SH if shared else portalocker.LOCK_EX
+        try:
+            for _ in range(300):
+                try:
+                    portalocker.lock(file, flags | portalocker.LOCK_NB)
+                    break
+                except portalocker.LockException:
+                    await asyncio.sleep(0.1)
+            else:
+                raise HTTPException(409, detail="Upload is busy")
+            if not (directory / "meta.json").exists():
+                raise HTTPException(404, detail="Upload session not found")
+            async with aiofiles.open(directory / "meta.json") as metadata:
+                if json.loads(await metadata.read()).get("aborted"):
+                    raise HTTPException(404, detail="Upload session not found")
+            yield
+        finally:
+            file.close()
+
+    @classmethod
+    async def create(cls, body: ChunkUploadCreate, user_id: int):
+        try:
+            path = normalize_path(body.path)
+        except ExecutionError:
+            raise HTTPException(400, detail="Invalid upload path")
+        if body.path.endswith("/") or path == "/":
+            raise HTTPException(400, detail="Path must be a file")
+        await PermissionService.require_path_permission(user_id, path, PathAction.WRITE)
+        _, _, _, rel = await VirtualFSService.resolve_adapter_and_rel(path)
+        if not rel:
+            raise HTTPException(400, detail="Path must be a file")
+        cls.root.mkdir(parents=True, exist_ok=True)
+        await asyncio.to_thread(cls._cleanup_expired)
+        upload_id = uuid.uuid4().hex
+        directory = cls.root / upload_id
+        directory.mkdir()
+        meta = {"path": path, "size": body.size, "overwrite": body.overwrite,
+                "user_id": user_id, "created": time.time(), "chunk_size": cls.chunk_size}
+        async with aiofiles.open(directory / "meta.tmp", "w") as file:
+            await file.write(json.dumps(meta))
+        (directory / "meta.tmp").replace(directory / "meta.json")
+        return {"upload_id": upload_id, "chunk_size": cls.chunk_size,
+                "parts": math.ceil(body.size / cls.chunk_size)}
+
+    @classmethod
+    def _cleanup_expired(cls):
+        for directory in cls.root.iterdir():
+            if not directory.is_dir() or not re.fullmatch(r"[a-f0-9]{32}", directory.name):
+                continue
+            try:
+                meta = json.loads((directory / "meta.json").read_text())
+                if time.time() - meta["created"] > cls.lifetime:
+                    with open(directory / "lock", "a+b") as file:
+                        try:
+                            portalocker.lock(file, portalocker.LOCK_EX | portalocker.LOCK_NB)
+                        except portalocker.LockException:
+                            continue
+                        shutil.rmtree(directory)
+            except (OSError, ValueError, KeyError):
+                if directory.exists() and time.time() - directory.stat().st_mtime > cls.lifetime:
+                    shutil.rmtree(directory, ignore_errors=True)
+                continue
+
+    @classmethod
+    async def _load(cls, upload_id: str, user_id: int, authorize=True):
+        if not re.fullmatch(r"[a-f0-9]{32}", upload_id):
+            raise HTTPException(404, detail="Upload session not found")
+        directory = cls.root / upload_id
+        try:
+            async with aiofiles.open(directory / "meta.json") as file:
+                meta = json.loads(await file.read())
+        except FileNotFoundError:
+            raise HTTPException(404, detail="Upload session not found")
+        if meta["user_id"] != user_id or meta.get("aborted"):
+            raise HTTPException(404, detail="Upload session not found")
+        if time.time() - meta["created"] > cls.lifetime:
+            raise HTTPException(410, detail="Upload session expired")
+        if authorize:
+            PermissionService.clear_cache(user_id)
+            await PermissionService.require_path_permission(user_id, meta["path"], PathAction.WRITE)
+        return directory, meta
+
+    @classmethod
+    async def put_part(cls, upload_id: str, part: int, user_id: int, chunks):
+        directory, meta = await cls._load(upload_id, user_id)
+        count = math.ceil(meta["size"] / meta["chunk_size"])
+        if part < 0 or part >= count:
+            raise HTTPException(400, detail="Invalid part number")
+        expected = min(meta["chunk_size"], meta["size"] - part * meta["chunk_size"])
+        temporary = directory / f"{part}-{uuid.uuid4().hex}.tmp"
+        async with cls._lock(directory, shared=True):
+            if (directory / "result.json").exists():
+                raise HTTPException(409, detail="Upload already completed")
+            try:
+                size = 0
+                async with aiofiles.open(temporary, "wb") as file:
+                    async for chunk in chunks:
+                        size += len(chunk)
+                        if size > expected:
+                            raise HTTPException(400, detail="Part exceeds expected size")
+                        await file.write(chunk)
+                if size != expected:
+                    raise HTTPException(400, detail="Incomplete upload part")
+                await asyncio.to_thread(temporary.replace, directory / f"part-{part}")
+            finally:
+                temporary.unlink(missing_ok=True)
+        return {"part": part, "size": size}
+
+    @classmethod
+    async def status(cls, upload_id: str, user_id: int):
+        directory, meta = await cls._load(upload_id, user_id)
+        async with cls._lock(directory, shared=True):
+            return {"upload_id": upload_id, "chunk_size": meta["chunk_size"],
+                    "parts": math.ceil(meta["size"] / meta["chunk_size"]),
+                    "uploaded": [int(path.name[5:]) for path in directory.glob("part-*")],
+                    "completed": (directory / "result.json").exists()}
+
+    @classmethod
+    async def complete(cls, upload_id: str, user_id: int, overwrite: bool | None = None):
+        directory, meta = await cls._load(upload_id, user_id)
+        async with cls._lock(directory):
+            result_path = directory / "result.json"
+            if result_path.exists():
+                async with aiofiles.open(result_path) as file:
+                    return json.loads(await file.read())
+            count = math.ceil(meta["size"] / meta["chunk_size"])
+            for part in range(count):
+                path = directory / f"part-{part}"
+                expected = min(meta["chunk_size"], meta["size"] - part * meta["chunk_size"])
+                if not path.exists() or path.stat().st_size != expected:
+                    raise HTTPException(400, detail=f"Missing upload part: {part}")
+
+            async def merged():
+                for part in range(count):
+                    async with aiofiles.open(directory / f"part-{part}", "rb") as file:
+                        while chunk := await file.read(1024 * 1024):
+                            yield chunk
+
+            allow_overwrite = meta["overwrite"] if overwrite is None else overwrite
+            if not allow_overwrite:
+                try:
+                    await VirtualFSService.stat_file(meta["path"])
+                except FileNotFoundError:
+                    pass
+                except HTTPException as exc:
+                    if exc.status_code != 404:
+                        raise
+                else:
+                    raise HTTPException(409, detail="Destination exists")
+            result = await VirtualFSService.write_file_stream(
+                meta["path"], merged(), overwrite=allow_overwrite,
+            )
+            async with aiofiles.open(directory / "result.tmp", "w") as file:
+                await file.write(json.dumps(result))
+            (directory / "result.tmp").replace(result_path)
+            for part in range(count):
+                (directory / f"part-{part}").unlink(missing_ok=True)
+            return result
+
+    @classmethod
+    async def abort(cls, upload_id: str, user_id: int):
+        directory, meta = await cls._load(upload_id, user_id, authorize=False)
+        async with cls._lock(directory):
+            # Leave the lock inode in place for workers already waiting on it.
+            for path in directory.iterdir():
+                if path.name not in ("lock", "meta.json"):
+                    path.unlink(missing_ok=True)
+            meta["aborted"] = True
+            async with aiofiles.open(directory / "meta.tmp", "w") as file:
+                await file.write(json.dumps(meta))
+            (directory / "meta.tmp").replace(directory / "meta.json")
+        return {"aborted": True}

main.py

@@ -115,6 +115,7 @@ app.add_middleware(
     allow_credentials=True,
     allow_methods=["*"],
     allow_headers=["*"],
+    expose_headers=["Content-Range", "Content-Length", "Accept-Ranges", "ETag", "Last-Modified"],
 )
 if __name__ == "__main__":
     import uvicorn

pyproject.toml

@@ -20,6 +20,7 @@ dependencies = [
     "pysocks>=1.7.1",
     "python-dotenv>=1.2.4",
     "python-multipart>=0.0.32",
+    "portalocker>=3.2.0,<4",
     "qdrant-client>=1.19.1",
     "setuptools>=84.0.0",
     "telethon>=1.45.0",

tests/test_adapter_transfers.py

@@ -0,0 +1,366 @@
+import asyncio
+from contextlib import asynccontextmanager
+from datetime import datetime
+from types import SimpleNamespace
+import unittest
+from unittest.mock import patch
+
+import httpx
+from fastapi import HTTPException
+
+from domain.adapters.providers.s3 import S3Adapter
+from domain.adapters.providers.webdav import WebDAVAdapter
+
+
+async def source(data):
+    yield b""
+    for offset in range(0, len(data), 7):
+        yield data[offset:offset + 7]
+
+
+async def consume(response):
+    return b"".join([chunk async for chunk in response.body_iterator])
+
+
+class MemoryBody:
+    def __init__(self, data):
+        self.data = data
+        self.offset = 0
+        self.closed = False
+
+    async def read(self, size):
+        data = self.data[self.offset:self.offset + size]
+        self.offset += len(data)
+        return data
+
+    def close(self):
+        self.closed = True
+
+
+class FakeS3:
+    def __init__(self, data=b"", fail_part=None):
+        self.data = data
+        self.fail_part = fail_part
+        self.parts = {}
+        self.completed = None
+        self.aborted = False
+        self.active = 0
+        self.maximum_active = 0
+        self.completion_order = []
+        self.bodies = []
+        self.get_options = []
+        self.clients_active = 0
+        self.started = asyncio.Event()
+        self.block = None
+        self.invalid_range = False
+
+    @asynccontextmanager
+    async def client(self):
+        self.clients_active += 1
+        try:
+            yield self
+        finally:
+            self.clients_active -= 1
+
+    async def create_multipart_upload(self, **kwargs):
+        return {"UploadId": "upload"}
+
+    async def upload_part(self, **kwargs):
+        number = kwargs["PartNumber"]
+        self.active += 1
+        self.maximum_active = max(self.maximum_active, self.active)
+        self.started.set()
+        try:
+            if self.block:
+                await self.block.wait()
+            await asyncio.sleep(0.005 if number % 2 else 0.001)
+            if number == self.fail_part:
+                raise IOError("part failed")
+            self.parts[number] = kwargs["Body"]
+            self.completion_order.append(number)
+            return {"ETag": f"part-{number}"}
+        finally:
+            self.active -= 1
+
+    async def complete_multipart_upload(self, **kwargs):
+        self.completed = kwargs["MultipartUpload"]["Parts"]
+        self.data = b"".join(self.parts[part["PartNumber"]] for part in self.completed)
+
+    async def abort_multipart_upload(self, **kwargs):
+        self.aborted = True
+
+    async def put_object(self, **kwargs):
+        self.data = kwargs["Body"]
+
+    async def head_object(self, **kwargs):
+        return {"ContentLength": len(self.data), "ETag": '"file-version"',
+                "ContentType": "application/octet-stream", "LastModified": datetime.now()}
+
+    async def get_object(self, **kwargs):
+        if not self.clients_active:
+            raise RuntimeError("client is closed")
+        self.get_options.append(kwargs)
+        start, end = map(int, kwargs["Range"][6:].split("-"))
+        self.active += 1
+        self.maximum_active = max(self.maximum_active, self.active)
+        self.started.set()
+        try:
+            if self.block:
+                await self.block.wait()
+            await asyncio.sleep(0.005 if start % 8 == 0 else 0.001)
+            body = MemoryBody(self.data[start:end + 1])
+            self.bodies.append(body)
+            return {"Body": body, "ContentLength": end - start + 1,
+                    "ContentRange": "invalid" if self.invalid_range else f"bytes {start}-{end}/{len(self.data)}"}
+        finally:
+            self.active -= 1
+
+
+class S3TransferTests(unittest.IsolatedAsyncioTestCase):
+    def make_adapter(self, fake):
+        adapter = S3Adapter(SimpleNamespace(config={"bucket_name": "bucket",
+                            "access_key_id": "test", "secret_access_key": "test"}))
+        adapter.multipart_part_size = 8
+        adapter.multipart_concurrency = 3
+        adapter.download_segment_size = 8
+        adapter.download_concurrency = 3
+        adapter._get_client = fake.client
+        return adapter
+
+    async def test_upload_concurrency_order_and_exact_content(self):
+        fake = FakeS3()
+        adapter = self.make_adapter(fake)
+        data = bytes(range(101))
+        size = await adapter.write_file_stream("", "file", source(data))
+        self.assertEqual(size, len(data))
+        self.assertEqual(fake.data, data)
+        self.assertEqual(fake.maximum_active, 3)
+        self.assertNotEqual(fake.completion_order, sorted(fake.completion_order))
+        self.assertEqual([p["PartNumber"] for p in fake.completed], list(range(1, 14)))
+        self.assertFalse(fake.aborted)
+
+    async def test_failed_upload_aborts_and_cancels_workers(self):
+        fake = FakeS3(fail_part=2)
+        adapter = self.make_adapter(fake)
+        with self.assertRaisesRegex(IOError, "part failed"):
+            await adapter.write_file_stream("", "file", source(bytes(range(100))))
+        self.assertTrue(fake.aborted)
+        self.assertIsNone(fake.completed)
+        self.assertEqual(fake.active, 0)
+
+    async def test_cancelled_upload_cleans_up(self):
+        fake = FakeS3()
+        fake.block = asyncio.Event()
+        adapter = self.make_adapter(fake)
+        task = asyncio.create_task(adapter.write_file_stream("", "file", source(bytes(range(100)))))
+        await fake.started.wait()
+        task.cancel()
+        with self.assertRaises(asyncio.CancelledError):
+            await task
+        self.assertTrue(fake.aborted)
+        self.assertEqual(fake.active, 0)
+        self.assertEqual(fake.clients_active, 0)
+
+    async def test_empty_upload_creates_empty_object(self):
+        fake = FakeS3(b"old content")
+        adapter = self.make_adapter(fake)
+        self.assertEqual(await adapter.write_file_stream("", "file", source(b"")), 0)
+        self.assertEqual(fake.data, b"")
+        self.assertTrue(fake.aborted)
+        self.assertIsNone(fake.completed)
+
+    async def test_download_concurrency_order_and_closed_bodies(self):
+        data = bytes(range(100))
+        fake = FakeS3(data)
+        response = await self.make_adapter(fake).stream_file("", "file", None)
+        self.assertEqual(fake.clients_active, 0)
+        self.assertEqual(await consume(response), data)
+        self.assertEqual(fake.maximum_active, 3)
+        self.assertTrue(all(body.closed for body in fake.bodies))
+        self.assertTrue(all(options["IfMatch"] == '"file-version"' for options in fake.get_options))
+        self.assertEqual(fake.clients_active, 0)
+
+    async def test_download_ranges(self):
+        data = bytes(range(100))
+        for header, expected in [("bytes=5-21", data[5:22]), ("bytes=-7", data[-7:]),
+                                 ("bytes=95-999", data[95:]), ("bytes=95-", data[95:])]:
+            with self.subTest(header=header):
+                response = await self.make_adapter(FakeS3(data)).stream_file("", "file", header)
+                self.assertEqual(response.status_code, 206)
+                self.assertEqual(int(response.headers["content-length"]), len(expected))
+                self.assertEqual(await consume(response), expected)
+
+    async def test_download_invalid_range_closes_body(self):
+        fake = FakeS3(b"123456789")
+        fake.invalid_range = True
+        response = await self.make_adapter(fake).stream_file("", "file", None)
+        with self.assertRaisesRegex(IOError, "invalid range"):
+            await consume(response)
+        self.assertTrue(all(body.closed for body in fake.bodies))
+        self.assertEqual(fake.clients_active, 0)
+
+    async def test_empty_download(self):
+        fake = FakeS3()
+        response = await self.make_adapter(fake).stream_file("", "file", None)
+        self.assertEqual(await consume(response), b"")
+        self.assertEqual(fake.get_options, [])
+
+    async def test_cancelled_download_closes_client_and_workers(self):
+        fake = FakeS3(bytes(range(100)))
+        fake.block = asyncio.Event()
+        response = await self.make_adapter(fake).stream_file("", "file", None)
+        task = asyncio.create_task(anext(response.body_iterator))
+        await fake.started.wait()
+        task.cancel()
+        with self.assertRaises(asyncio.CancelledError):
+            await task
+        self.assertEqual(fake.active, 0)
+        self.assertEqual(fake.clients_active, 0)
+
+
+class RawStream(httpx.AsyncByteStream):
+    def __init__(self, data, truncate=False):
+        self.data = data
+        self.truncate = truncate
+        self.closed = False
+
+    async def __aiter__(self):
+        yield self.data[:2]
+        if self.truncate:
+            raise httpx.ReadError("connection lost")
+        yield self.data[2:]
+
+    async def aclose(self):
+        self.closed = True
+
+
+class WebDAVTransferTests(unittest.IsolatedAsyncioTestCase):
+    def make_adapter(self, supports_range=True, failure=None):
+        self.data = bytes(range(100))
+        self.active = 0
+        self.maximum_active = 0
+        self.requests = []
+        self.streams = []
+        self.attempts = 0
+        self.started = asyncio.Event()
+        self.block = None
+
+        async def handler(request):
+            self.requests.append(request)
+            requested_range = request.headers.get("Range")
+            probe = requested_range == "bytes=0-0"
+            self.active += 1
+            self.maximum_active = max(self.maximum_active, self.active)
+            try:
+                if not probe:
+                    self.started.set()
+                    if self.block:
+                        await self.block.wait()
+                    await asyncio.sleep(0.005)
+                headers = {"ETag": '"version"'}
+                if supports_range and requested_range:
+                    start, end = map(int, requested_range[6:].split("-"))
+                    data = self.data[start:end + 1]
+                    headers["Content-Range"] = f"bytes {start}-{end}/{len(self.data)}"
+                    status = 206
+                else:
+                    data = self.data
+                    status = 200
+                truncate = False
+                if not probe:
+                    self.attempts += 1
+                    if failure == "ignore":
+                        data, status = self.data, 200
+                    elif failure == "range":
+                        headers["Content-Range"] = "bytes 0-0/100"
+                    elif failure == "oversize":
+                        data += b"extra"
+                    elif failure == "truncate" and self.attempts == 1:
+                        truncate = True
+                    elif failure == "missing":
+                        status = 404
+                headers["Content-Length"] = str(len(data))
+                stream = RawStream(data, truncate)
+                self.streams.append(stream)
+                return httpx.Response(status, headers=headers, stream=stream)
+            finally:
+                self.active -= 1
+
+        adapter = WebDAVAdapter(SimpleNamespace(config={"base_url": "https://dav.example/"}))
+        adapter.download_segment_size = 8
+        adapter.download_concurrency = 3
+        self.client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
+        adapter._client = lambda: self.client
+        self.addAsyncCleanup(self.client.aclose)
+        return adapter
+
+    async def test_download_concurrency_and_order(self):
+        adapter = self.make_adapter()
+        response = await adapter.stream_file("", "file", None)
+        self.assertFalse(self.client.is_closed)
+        self.assertEqual(await consume(response), self.data)
+        self.assertEqual(self.maximum_active, 3)
+        self.assertTrue(self.client.is_closed)
+        self.assertTrue(all(stream.closed for stream in self.streams))
+        self.assertTrue(all(r.headers["If-Range"] == '"version"' for r in self.requests[1:]))
+        self.assertEqual(response.headers["content-length"], "100")
+
+    async def test_ranges(self):
+        for header, start, end in [("bytes=5-21", 5, 21), ("bytes=-7", 93, 99),
+                                   ("bytes=95-999", 95, 99), ("bytes=95-", 95, 99)]:
+            with self.subTest(header=header):
+                adapter = self.make_adapter()
+                response = await adapter.stream_file("", "file", header)
+                self.assertEqual(await consume(response), self.data[start:end + 1])
+                self.assertEqual(response.status_code, 206)
+                self.assertEqual(response.headers["content-range"], f"bytes {start}-{end}/100")
+
+    async def test_fallback_without_range_support(self):
+        adapter = self.make_adapter(supports_range=False)
+        response = await adapter.stream_file("", "file", None)
+        self.assertEqual(await consume(response), self.data)
+        self.assertEqual(response.status_code, 200)
+        self.assertNotIn("accept-ranges", response.headers)
+        self.assertEqual(len(self.requests), 2)
+        self.assertTrue(self.client.is_closed)
+
+    async def test_invalid_range_responses_rejected_before_headers(self):
+        for failure in ("ignore", "range", "oversize"):
+            with self.subTest(failure=failure):
+                adapter = self.make_adapter(failure=failure)
+                with self.assertRaises(HTTPException) as caught:
+                    await adapter.stream_file("", "file", None)
+                self.assertEqual(caught.exception.status_code, 502)
+                self.assertTrue(self.client.is_closed)
+
+    async def test_retry_discards_partial_data(self):
+        adapter = self.make_adapter(failure="truncate")
+        with patch("domain.adapters.providers.webdav.asyncio.sleep", return_value=None):
+            response = await adapter.stream_file("", "file", None)
+        self.assertEqual(await consume(response), self.data)
+        self.assertEqual(self.attempts, 14)
+        self.assertTrue(all(stream.closed for stream in self.streams))
+
+    async def test_bad_client_ranges(self):
+        for header, expected in [("bytes=", 400), ("bytes=1-2,4-5", 400),
+                                 ("bytes=100-", 416), ("bytes=-0", 416), ("bytes=9-3", 416)]:
+            with self.subTest(header=header):
+                adapter = self.make_adapter()
+                with self.assertRaises(HTTPException) as caught:
+                    await adapter.stream_file("", "file", header)
+                self.assertEqual(caught.exception.status_code, expected)
+
+    async def test_close_download_cancels_prefetch(self):
+        adapter = self.make_adapter()
+        response = await adapter.stream_file("", "file", None)
+        self.started.clear()
+        self.block = asyncio.Event()
+        self.assertEqual(await anext(response.body_iterator), self.data[:8])
+        await self.started.wait()
+        await response.body_iterator.aclose()
+        self.assertEqual(self.active, 0)
+        self.assertTrue(self.client.is_closed)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_web_transfers.py

@@ -0,0 +1,219 @@
+import asyncio
+import json
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import AsyncMock, patch
+from types import SimpleNamespace
+
+from fastapi import FastAPI, HTTPException
+from fastapi.responses import Response
+from starlette.requests import Request
+import httpx
+
+from domain.adapters.providers.local import LocalAdapter
+from domain.auth import User, get_current_active_user
+from domain.virtual_fs.api import _download_original_response, router
+from domain.virtual_fs.chunk_uploads import ChunkUploadCreate, ChunkUploadService
+from domain.virtual_fs.service import VirtualFSService
+
+
+async def chunks(data):
+    for offset in range(0, len(data), 3):
+        yield data[offset:offset + 3]
+
+
+class ChunkUploadTests(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temporary.cleanup)
+        self.root = Path(self.temporary.name)
+        for target, value in [("root", self.root), ("chunk_size", 8)]:
+            patched = patch.object(ChunkUploadService, target, value)
+            patched.start()
+            self.addCleanup(patched.stop)
+        self.permission = patch("domain.virtual_fs.chunk_uploads.PermissionService.require_path_permission", new_callable=AsyncMock).start()
+        self.addCleanup(patch.stopall)
+        self.resolve = patch.object(VirtualFSService, "resolve_adapter_and_rel", new_callable=AsyncMock).start()
+        self.resolve.return_value = (None, None, "", "file")
+        self.written = b""
+
+        async def write(path, iterator, overwrite=True):
+            self.written = b"".join([data async for data in iterator])
+            return {"path": path, "size": len(self.written)}
+
+        self.write = patch.object(VirtualFSService, "write_file_stream", new=AsyncMock(side_effect=write)).start()
+
+    async def create(self, data=b"123456789abcdef", overwrite=True):
+        result = await ChunkUploadService.create(ChunkUploadCreate(path="/mount/file", size=len(data), overwrite=overwrite), 1)
+        return result["upload_id"]
+
+    async def test_parallel_parts_order_and_idempotent_complete(self):
+        data = b"123456789abcdef"
+        upload = await self.create(data)
+        await asyncio.gather(ChunkUploadService.put_part(upload, 1, 1, chunks(data[8:])),
+                             ChunkUploadService.put_part(upload, 0, 1, chunks(data[:8])))
+        status = await ChunkUploadService.status(upload, 1)
+        self.assertEqual(set(status["uploaded"]), {0, 1})
+        first, second = await asyncio.gather(ChunkUploadService.complete(upload, 1), ChunkUploadService.complete(upload, 1))
+        self.assertEqual(first, second)
+        self.assertEqual(self.written, data)
+        self.write.assert_awaited_once()
+        self.assertFalse(list((self.root / upload).glob("part-*")))
+        self.assertTrue((await ChunkUploadService.status(upload, 1))["completed"])
+
+    async def test_incomplete_and_oversized_parts_do_not_replace_good_part(self):
+        upload = await self.create()
+        await ChunkUploadService.put_part(upload, 0, 1, chunks(b"12345678"))
+        for data in (b"123", b"123456789"):
+            with self.assertRaises(HTTPException) as caught:
+                await ChunkUploadService.put_part(upload, 0, 1, chunks(data))
+            self.assertEqual(caught.exception.status_code, 400)
+        self.assertEqual((self.root / upload / "part-0").read_bytes(), b"12345678")
+        self.assertFalse(list((self.root / upload).glob("*.tmp")))
+
+    async def test_missing_parts_do_not_write_destination(self):
+        upload = await self.create()
+        with self.assertRaises(HTTPException) as caught:
+            await ChunkUploadService.complete(upload, 1)
+        self.assertEqual(caught.exception.status_code, 400)
+        self.write.assert_not_awaited()
+
+    async def test_session_owner_and_current_permissions_are_checked(self):
+        upload = await self.create()
+        with self.assertRaises(HTTPException) as caught:
+            await ChunkUploadService.status(upload, 2)
+        self.assertEqual(caught.exception.status_code, 404)
+        self.permission.side_effect = HTTPException(403, detail="permission revoked")
+        with self.assertRaises(HTTPException) as caught:
+            await ChunkUploadService.put_part(upload, 0, 1, chunks(b"12345678"))
+        self.assertEqual(caught.exception.status_code, 403)
+        self.write.assert_not_awaited()
+        await ChunkUploadService.abort(upload, 1)
+
+    async def test_completion_waits_for_inflight_part(self):
+        upload = await self.create(b"12345678")
+        started = asyncio.Event()
+        release = asyncio.Event()
+
+        async def delayed():
+            started.set()
+            await release.wait()
+            yield b"12345678"
+
+        part = asyncio.create_task(ChunkUploadService.put_part(upload, 0, 1, delayed()))
+        await started.wait()
+        completion = asyncio.create_task(ChunkUploadService.complete(upload, 1))
+        await asyncio.sleep(0.02)
+        self.assertFalse(completion.done())
+        release.set()
+        await part
+        await completion
+        self.assertEqual(self.written, b"12345678")
+
+    async def test_aborting_deletes_parts_and_rejects_waiting_operations(self):
+        upload = await self.create()
+        await ChunkUploadService.put_part(upload, 0, 1, chunks(b"12345678"))
+        await ChunkUploadService.abort(upload, 1)
+        self.assertFalse(list((self.root / upload).glob("part-*")))
+        with self.assertRaises(HTTPException) as caught:
+            await ChunkUploadService.status(upload, 1)
+        self.assertEqual(caught.exception.status_code, 404)
+
+    async def test_failed_completion_preserves_parts_for_retry(self):
+        upload = await self.create(b"12345678")
+        await ChunkUploadService.put_part(upload, 0, 1, chunks(b"12345678"))
+        self.write.side_effect = IOError("storage unavailable")
+        with self.assertRaises(IOError):
+            await ChunkUploadService.complete(upload, 1)
+        self.assertEqual((await ChunkUploadService.status(upload, 1))["uploaded"], [0])
+
+    async def test_expired_sessions_are_rejected_and_cleaned(self):
+        upload = await self.create()
+        path = self.root / upload / "meta.json"
+        meta = json.loads(path.read_text())
+        meta["created"] = 0
+        path.write_text(json.dumps(meta))
+        with self.assertRaises(HTTPException) as caught:
+            await ChunkUploadService.status(upload, 1)
+        self.assertEqual(caught.exception.status_code, 410)
+        await self.create()
+        self.assertFalse((self.root / upload).exists())
+
+    async def test_invalid_ids_and_paths(self):
+        with self.assertRaises(HTTPException):
+            await ChunkUploadService.status("../../secret", 1)
+        with self.assertRaises(HTTPException):
+            await ChunkUploadService.create(ChunkUploadCreate(path="/mount/../file", size=1), 1)
+
+    async def test_no_overwrite_checks_destination_at_completion(self):
+        upload = await self.create(b"12345678", overwrite=False)
+        await ChunkUploadService.put_part(upload, 0, 1, chunks(b"12345678"))
+        with patch.object(VirtualFSService, "stat_file", new=AsyncMock(return_value={"size": 4})):
+            with self.assertRaises(HTTPException) as caught:
+                await ChunkUploadService.complete(upload, 1)
+        self.assertEqual(caught.exception.status_code, 409)
+        self.write.assert_not_awaited()
+
+
+class OriginalDownloadTests(unittest.IsolatedAsyncioTestCase):
+    async def test_original_download_bypasses_preview_conversion_and_checks_version(self):
+        stat = {"size": 10, "mtime": 100, "is_dir": False}
+        adapter = type("Adapter", (), {"stream_file": AsyncMock(return_value=Response(b"RAW data"))})()
+        request = Request({"type": "http", "headers": []})
+        with patch.object(VirtualFSService, "stat_file", new=AsyncMock(return_value=stat)), patch.object(
+            VirtualFSService, "resolve_adapter_and_rel", new=AsyncMock(return_value=(adapter, None, "", "image.raw"))
+        ):
+            response = await _download_original_response("/mount/image.raw", request)
+            self.assertEqual(response.body, b"RAW data")
+            request = Request({"type": "http", "headers": [(b"if-match", response.headers["etag"].encode())]})
+            stat["size"] = 11
+            with self.assertRaises(HTTPException) as caught:
+                await _download_original_response("/mount/image.raw", request)
+            self.assertEqual(caught.exception.status_code, 412)
+
+
+class WebTransferEndpointTests(unittest.IsolatedAsyncioTestCase):
+    async def test_http_upload_and_original_range_download_with_local_storage(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            adapter = LocalAdapter(SimpleNamespace(config={"root": str(root)}, path="/mount"))
+            model = SimpleNamespace(path="/mount")
+            async def resolve(path):
+                return adapter, model, str(root), path.removeprefix("/mount/")
+
+            app = FastAPI()
+            app.include_router(router)
+            app.dependency_overrides[get_current_active_user] = lambda: User(id=1, username="tester")
+            with patch.object(ChunkUploadService, "root", root / "sessions"), patch.object(
+                ChunkUploadService, "chunk_size", 8
+            ), patch.object(VirtualFSService, "resolve_adapter_and_rel", new=AsyncMock(side_effect=resolve)), patch(
+                "domain.permission.service.PermissionService.require_path_permission", new=AsyncMock()
+            ), patch("domain.audit.service.AuditService.log", new=AsyncMock()), patch(
+                "domain.tasks.TaskService.trigger_tasks", new=AsyncMock()
+            ):
+                async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
+                    path = "/mount/original #?.raw"
+                    data = b"original RAW file data"
+                    response = await client.post("/api/fs/uploads", json={"path": path, "size": len(data), "overwrite": False})
+                    self.assertEqual(response.status_code, 200, response.text)
+                    upload = response.json()["data"]["upload_id"]
+                    for index in reversed(range(3)):
+                        response = await client.put(f"/api/fs/uploads/{upload}/parts/{index}", content=data[index * 8:(index + 1) * 8])
+                        self.assertEqual(response.status_code, 200, response.text)
+                    response = await client.post(f"/api/fs/uploads/{upload}/complete")
+                    self.assertEqual(response.status_code, 200, response.text)
+                    self.assertEqual((root / "original #?.raw").read_bytes(), data)
+                    url = "/api/fs/download/mount/original%20%23%3F.raw"
+                    response = await client.get(url, headers={"Range": "bytes=1-10"})
+                    self.assertEqual(response.status_code, 206, response.text)
+                    self.assertEqual(response.content, data[1:11])
+                    self.assertTrue(response.headers["Content-Disposition"].startswith("attachment"))
+                    etag = response.headers["ETag"]
+                    (root / "original #?.raw").write_bytes(b"x" * len(data))
+                    response = await client.get(url, headers={"Range": "bytes=0-1", "If-Match": etag})
+                    self.assertEqual(response.status_code, 412)
+
+
+if __name__ == "__main__":
+    unittest.main()

uv.lock

@@ -593,6 +593,7 @@ dependencies = [
     { name = "milvus-lite", marker = "sys_platform != 'win32'" },
     { name = "paramiko" },
     { name = "pillow" },
+    { name = "portalocker" },
     { name = "pydantic", extra = ["email"] },
     { name = "pyjwt" },
     { name = "pymilvus", extra = ["milvus-lite"] },
@@ -618,6 +619,7 @@ requires-dist = [
     { name = "milvus-lite", marker = "sys_platform != 'win32'", specifier = ">=3.2.1" },
     { name = "paramiko", specifier = ">=5.0.0" },
     { name = "pillow", specifier = ">=12.3.0" },
+    { name = "portalocker", specifier = ">=3.2.0,<4" },
     { name = "pydantic", extras = ["email"], specifier = ">=2.13.5,<3" },
     { name = "pyjwt", specifier = ">=2.15.1" },
     { name = "pymilvus", extras = ["milvus-lite"], specifier = ">=3.0.2" },

web/src/api/transfers.ts

@@ -0,0 +1,259 @@
+export const TRANSFER_CHUNK_SIZE = 8 * 1024 * 1024;
+export const MAX_BLOB_DOWNLOAD_SIZE = 128 * 1024 * 1024;
+
+export class TransferError extends Error {
+  readonly status?: number;
+
+  constructor(message: string, status?: number) {
+    super(message);
+    this.name = 'TransferError';
+    this.status = status;
+  }
+}
+
+export const encodeFilePath = (path: string) => path.replace(/^\/+/, '').split('/').map(encodeURIComponent).join('/');
+
+export function authHeaders(): Record<string, string> {
+  const token = localStorage.getItem('token');
+  return token ? { Authorization: `Bearer ${token}` } : {};
+}
+
+class RequestPool {
+  private active = 0;
+  private waiting: Array<() => void> = [];
+
+  async run<T>(work: () => Promise<T>, signal?: AbortSignal): Promise<T> {
+    signal?.throwIfAborted();
+    if (this.active >= 6) {
+      await new Promise<void>((resolve, reject) => {
+        const ready = () => {
+          signal?.removeEventListener('abort', aborted);
+          resolve();
+        };
+        const aborted = () => {
+          this.waiting = this.waiting.filter(item => item !== ready);
+          reject(signal?.reason ?? new DOMException('Aborted', 'AbortError'));
+        };
+        this.waiting.push(ready);
+        signal?.addEventListener('abort', aborted, { once: true });
+      });
+    } else {
+      this.active += 1;
+    }
+    try {
+      signal?.throwIfAborted();
+      return await work();
+    } finally {
+      const next = this.waiting.shift();
+      if (next) next();
+      else this.active -= 1;
+    }
+  }
+}
+
+export const transferRequests = new RequestPool();
+
+export async function transferJson<T>(url: string, options: RequestInit = {}): Promise<T> {
+  const response = await fetch(url, {
+    ...options, headers: { ...authHeaders(), 'Content-Type': 'application/json', ...options.headers },
+  });
+  const data = await response.json();
+  if (!response.ok || data.code !== 0) {
+    throw new TransferError(data.detail || data.msg || 'Transfer failed', response.status);
+  }
+  return data.data;
+}
+
+export async function retryTransfer<T>(work: () => Promise<T>, signal?: AbortSignal): Promise<T> {
+  for (let attempt = 0; ; attempt += 1) {
+    signal?.throwIfAborted();
+    try {
+      return await work();
+    } catch (error) {
+      signal?.throwIfAborted();
+      if (attempt >= 2 || (error instanceof TransferError && error.status !== undefined
+        && error.status < 500 && ![408, 429].includes(error.status))) throw error;
+      await new Promise<void>((resolve, reject) => {
+        const aborted = () => {
+          clearTimeout(timer);
+          reject(signal?.reason ?? new DOMException('Aborted', 'AbortError'));
+        };
+        const timer = setTimeout(() => {
+          signal?.removeEventListener('abort', aborted);
+          resolve();
+        }, 400 * 2 ** attempt);
+        signal?.addEventListener('abort', aborted, { once: true });
+      });
+    }
+  }
+}
+
+export async function runTransferWorkers(count: number, concurrency: number,
+  work: (index: number, signal: AbortSignal) => Promise<void>, signal?: AbortSignal) {
+  const controller = new AbortController();
+  const abort = () => controller.abort(signal?.reason);
+  if (signal?.aborted) abort();
+  else signal?.addEventListener('abort', abort, { once: true });
+  let next = 0;
+  let failure: unknown;
+  const workers = Array.from({ length: Math.min(count, concurrency) }, async () => {
+    while (next < count && !controller.signal.aborted) {
+      const index = next++;
+      try {
+        await work(index, controller.signal);
+      } catch (error) {
+        if (failure === undefined) failure = error;
+        controller.abort(error);
+      }
+    }
+  });
+  await Promise.all(workers);
+  signal?.removeEventListener('abort', abort);
+  if (failure !== undefined) throw failure;
+  signal?.throwIfAborted();
+}
+
+export function sendUpload(url: string, body: Blob | FormData,
+  onProgress?: (loaded: number, total: number) => void, signal?: AbortSignal,
+  method: 'PUT' | 'POST' = 'PUT'): Promise<{ path: string; size: number }> {
+  return transferRequests.run(() => new Promise((resolve, reject) => {
+    const xhr = new XMLHttpRequest();
+    const abort = () => xhr.abort();
+    const finish = (error?: Error, result?: { path: string; size: number }) => {
+      signal?.removeEventListener('abort', abort);
+      if (error) reject(error);
+      else resolve(result!);
+    };
+    xhr.open(method, url);
+    xhr.timeout = 10 * 60 * 1000;
+    Object.entries(authHeaders()).forEach(([key, value]) => xhr.setRequestHeader(key, value));
+    if (body instanceof Blob) xhr.setRequestHeader('Content-Type', body.type || 'application/octet-stream');
+    xhr.upload.onprogress = event => {
+      if (event.lengthComputable) onProgress?.(event.loaded, event.total);
+    };
+    xhr.onload = () => {
+      let data;
+      try { data = JSON.parse(xhr.responseText); } catch { /* Handled below. */ }
+      if (xhr.status >= 200 && xhr.status < 300 && data?.code === 0) finish(undefined, data.data);
+      else finish(new TransferError(data?.detail || data?.msg || 'Upload failed', xhr.status));
+    };
+    xhr.onerror = () => finish(new TransferError('Upload connection failed'));
+    xhr.ontimeout = () => finish(new TransferError('Upload timed out', 408));
+    xhr.onabort = () => finish(new DOMException('Aborted', 'AbortError'));
+    signal?.addEventListener('abort', abort, { once: true });
+    xhr.send(body);
+  }), signal);
+}
+
+export interface DownloadTarget {
+  write(data: { type: 'write'; position: number; data: Blob }): Promise<void>;
+  close(): Promise<void>;
+  abort(): Promise<void>;
+}
+
+interface SaveFileHandle {
+  createWritable(): Promise<DownloadTarget>;
+}
+
+export function selectDownloadTarget(name: string): Promise<SaveFileHandle> | undefined {
+  if (!window.isSecureContext) return undefined;
+  const picker = (window as Window & { showSaveFilePicker?:
+    (options: { suggestedName: string }) => Promise<SaveFileHandle> }).showSaveFilePicker;
+  return picker?.call(window, { suggestedName: name });
+}
+
+export function saveDownloadUrl(url: string, name: string) {
+  const link = document.createElement('a');
+  link.href = url;
+  link.download = name;
+  document.body.appendChild(link);
+  link.click();
+  link.remove();
+}
+
+export async function downloadRanges(url: string, size: number,
+  onProgress?: (loaded: number, total: number) => void, signal?: AbortSignal,
+  target?: DownloadTarget): Promise<Blob | null> {
+  const probe = await transferRequests.run(() => fetch(url, {
+    headers: { ...authHeaders(), Range: 'bytes=0-0' }, signal,
+  }), signal);
+  const range = probe.headers.get('Content-Range');
+  const supportsRanges = probe.status === 206 && range === `bytes 0-0/${size}`;
+  const etag = probe.headers.get('ETag');
+  await probe.body?.cancel();
+  if (!supportsRanges) {
+    if (!probe.ok) throw new TransferError('Download failed', probe.status);
+    if (target) {
+      await transferRequests.run(async () => {
+        const response = await fetch(url, { headers: authHeaders(), signal });
+        if (!response.ok) throw new TransferError('Download failed', response.status);
+        if (!response.body) throw new TransferError('Empty download response');
+        const reader = response.body.getReader();
+        let position = 0;
+        try {
+          while (true) {
+            const { done, value } = await reader.read();
+            if (done) break;
+            const data = new Blob([value]);
+            await target.write({ type: 'write', position, data });
+            position += data.size;
+            onProgress?.(position, size);
+          }
+        } finally {
+          await reader.cancel().catch(() => void 0);
+        }
+      }, signal);
+      return new Blob();
+    }
+    return null;
+  }
+  const parts: Blob[] = [];
+  const count = Math.ceil(size / TRANSFER_CHUNK_SIZE);
+  let completed = 0;
+  let writing = Promise.resolve();
+  await runTransferWorkers(count, 4, async (index, workerSignal) => {
+    const start = index * TRANSFER_CHUNK_SIZE;
+    const end = Math.min(size - 1, start + TRANSFER_CHUNK_SIZE - 1);
+    const blob = await retryTransfer(() => transferRequests.run(async () => {
+      const headers: Record<string, string> = { ...authHeaders(), Range: `bytes=${start}-${end}` };
+      if (etag) headers['If-Match'] = etag;
+      const response = await fetch(url, { headers, signal: AbortSignal.any([workerSignal, AbortSignal.timeout(120_000)]) });
+      if (!response.ok) throw new TransferError('Download failed', response.status);
+      if (response.status !== 206 || response.headers.get('Content-Range') !== `bytes ${start}-${end}/${size}`) {
+        await response.body?.cancel();
+        throw new TransferError('Invalid download range', 400);
+      }
+      if (!response.body) throw new TransferError('Empty download segment');
+      const reader = response.body.getReader();
+      const chunks: Uint8Array<ArrayBuffer>[] = [];
+      let received = 0;
+      try {
+        while (true) {
+          const { done, value } = await reader.read();
+          if (done) break;
+          received += value.byteLength;
+          if (received > end - start + 1) throw new TransferError('Download segment exceeds requested size', 400);
+          chunks.push(value);
+        }
+      } finally {
+        await reader.cancel().catch(() => void 0);
+      }
+      if (received !== end - start + 1) throw new TransferError('Incomplete download segment');
+      return new Blob(chunks);
+    }, workerSignal), workerSignal);
+    if (target) {
+      // Serialize writes to one file stream while network requests stay parallel.
+      const write = writing.then(() => target.write({ type: 'write', position: start, data: blob }));
+      writing = write;
+      await write;
+    } else {
+      parts[index] = blob;
+    }
+    completed += blob.size;
+    onProgress?.(completed, size);
+  }, signal).catch(async error => {
+    await writing.catch(() => void 0);
+    throw error;
+  });
+  return target ? new Blob() : new Blob(parts, { type: 'application/octet-stream' });
+}

web/src/api/vfs.ts

@@ -1,4 +1,10 @@
 import request, { API_BASE_URL } from './client';
+import { encodeFilePath, retryTransfer, runTransferWorkers, sendUpload, transferJson,
+  TRANSFER_CHUNK_SIZE } from './transfers';
+
+export interface UploadTransferState {
+  session?: { upload_id: string; chunk_size: number; parts: number };
+}
 
 export interface VfsEntry {
   name: string;
@@ -51,6 +57,62 @@ export interface SearchResponse {
 }
 
 export const vfsApi = {
+  abortUploadSession: (id: string) => request(`/fs/uploads/${id}`, { method: 'DELETE' }),
+  uploadOptimized: async (fullPath: string, file: File, overwrite = true,
+    onProgress?: (loaded: number, total: number) => void, signal?: AbortSignal,
+    state: UploadTransferState = {}) => {
+    if (file.size <= TRANSFER_CHUNK_SIZE) {
+      return sendUpload(`${API_BASE_URL}/fs/upload-raw/${encodeFilePath(fullPath)}?overwrite=${overwrite}`,
+        file, onProgress, signal);
+    }
+    const base = `${API_BASE_URL}/fs/uploads`;
+    let uploaded: number[] = [];
+    if (state.session) {
+      try {
+        const status = await transferJson<{ uploaded: number[]; completed: boolean }>(`${base}/${state.session.upload_id}`, { signal });
+        if (status.completed) {
+          const result = await transferJson<{ path: string; size: number }>(`${base}/${state.session.upload_id}/complete`, { method: 'POST', signal });
+          state.session = undefined;
+          return result;
+        }
+        uploaded = status.uploaded;
+      } catch (error) {
+        const status = (error as { status?: number }).status;
+        if (status !== 404 && status !== 410) throw error;
+        state.session = undefined;
+      }
+    }
+    if (!state.session) {
+      state.session = await transferJson(`${base}`, {
+        method: 'POST', body: JSON.stringify({ path: fullPath, size: file.size, overwrite }), signal,
+      });
+    }
+    const session = state.session!;
+    const done = new Set(uploaded);
+    const progress = Array.from({ length: session.parts }, (_, index) => done.has(index)
+      ? Math.min(session.chunk_size, file.size - index * session.chunk_size) : 0);
+    const report = () => onProgress?.(progress.reduce((sum, bytes) => sum + bytes, 0), file.size);
+    report();
+    await runTransferWorkers(session.parts, 4, async (index, workerSignal) => {
+      if (done.has(index)) return;
+      const start = index * session.chunk_size;
+      const chunk = file.slice(start, Math.min(file.size, start + session.chunk_size));
+      await retryTransfer(() => {
+        progress[index] = 0;
+        report();
+        return sendUpload(`${base}/${session.upload_id}/parts/${index}`, chunk, loaded => {
+          progress[index] = Math.min(loaded, chunk.size);
+          report();
+        }, workerSignal);
+      }, workerSignal);
+      progress[index] = chunk.size;
+      report();
+    }, signal);
+    const result = await retryTransfer(() => transferJson<{ path: string; size: number }>(
+      `${base}/${session.upload_id}/complete?overwrite=${overwrite}`, { method: 'POST', signal }), signal);
+    state.session = undefined;
+    return result;
+  },
   list: (path: string, page: number = 1, pageSize: number = 50, sortBy: string = 'name', sortOrder: string = 'asc', cursor?: string | null) => {
     const cleaned = path.replace(/\\/g, '/');
     const trimmed = cleaned === '/' ? '' : cleaned.replace(/^\/+/, '');
@@ -64,7 +126,7 @@ export const vfsApi = {
     return request<DirListing>(`/fs/${encodeURI(trimmed)}?${params}`);
   },
   readFile: async (path: string) => {
-    const enc = encodeURI(path.replace(/^\/+/, ''));
+    const enc = encodeFilePath(path);
     const resp = await request(`/fs/file/${enc}`, { rawResponse: true });
     return await (resp as Response).arrayBuffer();
   },
@@ -95,10 +157,10 @@ export const vfsApi = {
     const params = new URLSearchParams();
     if (options?.verbose) params.set('verbose', 'true');
     const query = params.toString();
-    return request(`/fs/stat/${encodeURI(path.replace(/^\/+/, ''))}${query ? `?${query}` : ''}`);
+    return request(`/fs/stat/${encodeFilePath(path)}${query ? `?${query}` : ''}`);
   },
   getTempLinkToken: (path: string, expiresIn: number = 3600) =>
-    request<{token: string, path: string, url: string}>(`/fs/temp-link/${encodeURI(path.replace(/^\/+/, ''))}?expires_in=${expiresIn}`),
+    request<{token: string, path: string, url: string}>(`/fs/temp-link/${encodeFilePath(path)}?expires_in=${expiresIn}`),
   getTempPublicUrl: (token: string) => `${API_BASE_URL}/fs/public/${token}`,
   uploadRaw: (fullPath: string, file: File, overwrite: boolean = true, onProgress?: (loaded: number, total: number) => void) => {
     const enc = encodeURI(fullPath.replace(/^\/+/, ''));

web/src/i18n/locales/en.json

@@ -861,5 +861,8 @@
   "Delete files and folders": "Delete files and folders",
   "Create share links": "Create share links",
   "permission.category.system": "System",
-  "permission.category.adapter": "Adapter"
+  "permission.category.adapter": "Adapter",
+  "Retry failed uploads": "Retry failed uploads",
+  "Upload cancelled": "Upload cancelled",
+  "Download completed": "Download completed"
 }

web/src/i18n/locales/zh.json

@@ -864,5 +864,8 @@
   "Delete files and folders": "删除文件和目录",
   "Create share links": "创建分享链接",
   "permission.category.system": "系统",
-  "permission.category.adapter": "存储适配器"
+  "permission.category.adapter": "存储适配器",
+  "Retry failed uploads": "重试失败上传",
+  "Upload cancelled": "上传已取消",
+  "Download completed": "下载完成"
 }

web/src/pages/FileExplorerPage/FileExplorerPage.tsx

@@ -519,6 +519,8 @@ const FileExplorerPage = memo(function FileExplorerPage() {
         conflict={uploader.conflict}
         onClose={uploader.closeModal}
         onStartUpload={uploader.startUpload}
+        onRetryFailed={uploader.retryFailed}
+        onCancelUpload={uploader.cancelUpload}
         onResolveConflict={uploader.confirmConflict}
       />
       <DropzoneOverlay visible={isDragging} />

web/src/pages/FileExplorerPage/components/Modals/UploadModal.tsx

@@ -1,6 +1,6 @@
 import React, { useEffect, useMemo } from 'react';
 import { Modal, Button, List, Progress, Typography, message, Flex, Tag, Space } from 'antd';
-import { CopyOutlined, CheckCircleFilled, CloseCircleFilled } from '@ant-design/icons';
+import { CopyOutlined, CheckCircleFilled, CloseCircleFilled, ReloadOutlined } from '@ant-design/icons';
 import type { ConflictDecision, UploadConflict, UploadFile } from '../../hooks/useUploader';
 import { useI18n } from '../../../../i18n';
 
@@ -14,6 +14,8 @@ interface UploadModalProps {
   conflict: UploadConflict | null;
   onClose: () => void;
   onStartUpload: () => void;
+  onRetryFailed: () => void;
+  onCancelUpload: () => void;
   onResolveConflict: (decision: ConflictDecision) => void;
 }
 
@@ -35,6 +37,8 @@ const UploadModal: React.FC<UploadModalProps> = ({
   conflict,
   onClose,
   onStartUpload,
+  onRetryFailed,
+  onCancelUpload,
   onResolveConflict,
 }) => {
   const { t } = useI18n();
@@ -120,6 +124,10 @@ const UploadModal: React.FC<UploadModalProps> = ({
       maskClosable={!isUploading}
       onCancel={onClose}
       footer={[
+        isUploading && <Button key="cancel" onClick={onCancelUpload}>{t('Cancel')}</Button>,
+        summary.failures > 0 && <Button key="retry" icon={<ReloadOutlined />} onClick={onRetryFailed} disabled={isUploading}>
+          {t('Retry failed uploads')}
+        </Button>,
         <Button key="close" onClick={onClose} disabled={!allFinished || isUploading}>
           {allFinished ? t('Close') : t('Done')}
         </Button>,

web/src/pages/FileExplorerPage/hooks/useFileActions.tsx

@@ -1,7 +1,12 @@
-import { useCallback } from 'react';
-import { message, Modal } from 'antd';
+import { useCallback, useEffect, useRef } from 'react';
+import { Button, message, Modal, Progress, Tooltip } from 'antd';
+import { CloseOutlined } from '@ant-design/icons';
 import { useI18n } from '../../../i18n';
 import { vfsApi, type VfsEntry } from '../../../api/client';
+import { API_BASE_URL } from '../../../api/client';
+import { downloadRanges, encodeFilePath, MAX_BLOB_DOWNLOAD_SIZE, saveDownloadUrl,
+  selectDownloadTarget, TRANSFER_CHUNK_SIZE } from '../../../api/transfers';
+import type { DownloadTarget } from '../../../api/transfers';
 
 interface FileActionsParams {
   path: string;
@@ -13,6 +18,11 @@ interface FileActionsParams {
 
 export function useFileActions({ path, refresh, clearSelection, onShare, onGetDirectLink }: FileActionsParams) {
   const { t } = useI18n();
+  const downloadsRef = useRef(new Map<string, AbortController>());
+  useEffect(() => {
+    const downloads = downloadsRef.current;
+    return () => { for (const controller of downloads.values()) controller.abort(); };
+  }, []);
   const normalizeFullPath = useCallback((name: string) => {
     const base = path === '/' ? '' : path;
     return `${base}/${name}`.replace(/\/{2,}/g, '/');
@@ -168,23 +178,68 @@ export function useFileActions({ path, refresh, clearSelection, onShare, onGetDi
     refresh();
   }, [normalizeDestination, normalizeFullPath, t, buildEntryDestination, refresh]);
 
-  const doDownload = useCallback((entry: VfsEntry) => {
+  const doDownload = useCallback(async (entry: VfsEntry) => {
     if (entry.is_dir) {
       message.warning(t('Downloading folders is not supported'));
       return;
     }
+    const fullPath = normalizeFullPath(entry.name);
+    if (downloadsRef.current.has(fullPath)) return;
+    const controller = new AbortController();
+    downloadsRef.current.set(fullPath, controller);
+    const key = `download:${fullPath}`;
+    let target: DownloadTarget | undefined;
+    let lastUpdate = 0;
+    const report = (loaded: number, total: number) => {
+      if (Date.now() - lastUpdate < 200 && loaded < total) return;
+      lastUpdate = Date.now();
+      message.open({ key, duration: 0, content: <div style={{ width: 280, maxWidth: '70vw', textAlign: 'left' }}>
+        <div style={{ display: 'flex', alignItems: 'center', gap: 8 }}>
+          <span style={{ flex: 1, minWidth: 0, overflow: 'hidden', textOverflow: 'ellipsis', whiteSpace: 'nowrap' }}>{entry.name}</span>
+          <Tooltip title={t('Cancel')}><Button type="text" size="small" icon={<CloseOutlined />}
+            aria-label={t('Cancel')} onClick={() => controller.abort()} /></Tooltip>
+        </div>
+        <Progress percent={total ? Math.min(99, Math.round(loaded / total * 100)) : 0} size="small" />
+      </div> });
+    };
     try {
-      const url = vfsApi.streamUrl((path === '/' ? '' : path) + '/' + entry.name);
-      const a = document.createElement('a');
-      a.href = url;
-      a.download = entry.name;
-      document.body.appendChild(a);
-      a.click();
-      document.body.removeChild(a);
-    } catch (e: any) {
-      message.error(e.message || t('Download failed'));
+      // The picker must be called while the click still has user activation.
+      const handle = entry.size > TRANSFER_CHUNK_SIZE ? await selectDownloadTarget(entry.name) : undefined;
+      controller.signal.throwIfAborted();
+      const optimized = entry.size > TRANSFER_CHUNK_SIZE
+        && (handle !== undefined || entry.size <= MAX_BLOB_DOWNLOAD_SIZE);
+      if (optimized) {
+        target = await handle?.createWritable();
+        report(0, entry.size);
+        const blob = await downloadRanges(`${API_BASE_URL}/fs/download/${encodeFilePath(fullPath)}`,
+          entry.size, report, controller.signal, target);
+        if (blob !== null) {
+          if (target) {
+            await target.close();
+            target = undefined;
+          } else {
+            const objectUrl = URL.createObjectURL(blob);
+            saveDownloadUrl(objectUrl, entry.name);
+            setTimeout(() => URL.revokeObjectURL(objectUrl), 60_000);
+          }
+          message.success({ key, content: t('Download completed') });
+          return;
+        }
+        await target?.abort();
+        target = undefined;
+      }
+      const link = await vfsApi.getTempLinkToken(fullPath);
+      controller.signal.throwIfAborted();
+      saveDownloadUrl(`${API_BASE_URL}/fs/download-public/${encodeURIComponent(link.token)}/${encodeURIComponent(entry.name)}`, entry.name);
+      message.destroy(key);
+    } catch (error) {
+      await target?.abort().catch(() => void 0);
+      if ((error as Error).name === 'AbortError' || controller.signal.aborted) message.destroy(key);
+      else message.error({ key, content: error instanceof Error ? error.message : t('Download failed') });
+    } finally {
+      downloadsRef.current.delete(fullPath);
     }
-  }, [path, t]);
+  }, [normalizeFullPath, t]);
 
   const doShare = useCallback((entries: VfsEntry[]) => {
     if (entries.length === 0) {

web/src/pages/FileExplorerPage/hooks/useUploader.ts

@@ -2,6 +2,7 @@ import type { ChangeEvent, RefObject } from 'react';
 import { useState, useCallback, useRef, useMemo, useEffect } from 'react';
 import { message } from 'antd';
 import { vfsApi } from '../../../api/client';
+import type { UploadTransferState } from '../../../api/vfs';
 import { useI18n } from '../../../i18n';
 
 type UploadStatus = 'pending' | 'waiting' | 'uploading' | 'success' | 'error' | 'skipped';
@@ -270,6 +271,18 @@ export function useUploader(path: string, onUploadComplete: () => void) {
   const createdDirsRef = useRef<Set<string>>(new Set());
   const filesRef = useRef<UploadFile[]>(files);
   const isUploadingRef = useRef(false);
+  const transferStatesRef = useRef(new Map<string, UploadTransferState>());
+  const batchControllerRef = useRef<AbortController | null>(null);
+
+  useEffect(() => {
+    const states = transferStatesRef.current;
+    return () => {
+      batchControllerRef.current?.abort();
+      for (const state of states.values()) {
+        if (state.session) void vfsApi.abortUploadSession(state.session.upload_id).catch(() => void 0);
+      }
+    };
+  }, []);
 
   const fileInputRef = useRef<HTMLInputElement | null>(null);
   const directoryInputRef = useRef<HTMLInputElement | null>(null);
@@ -282,11 +295,9 @@ export function useUploader(path: string, onUploadComplete: () => void) {
   }, []);
 
   const mutateFiles = useCallback((updater: (prev: UploadFile[]) => UploadFile[]) => {
-    setFiles((prev) => {
-      const next = updater(prev);
-      filesRef.current = next;
-      return next;
-    });
+    const next = updater(filesRef.current);
+    filesRef.current = next;
+    setFiles(next);
   }, []);
 
   const replaceFiles = useCallback((next: UploadFile[]) => {
@@ -320,6 +331,10 @@ export function useUploader(path: string, onUploadComplete: () => void) {
       return;
     }
     setIsModalVisible(false);
+    for (const state of transferStatesRef.current.values()) {
+      if (state.session) void vfsApi.abortUploadSession(state.session.upload_id).catch(() => void 0);
+    }
+    transferStatesRef.current.clear();
     replaceFiles([]);
     resetOverwriteDecisions();
     setConflict(null);
@@ -498,12 +513,17 @@ export function useUploader(path: string, onUploadComplete: () => void) {
 
     updateFile(task.id, { status: 'uploading', progress: 0, loadedBytes: 0 });
     try {
-      const uploadResult = await vfsApi.uploadRaw(task.targetPath, task.file, shouldOverwrite, (loaded, total) => {
+      const state = transferStatesRef.current.get(task.id) ?? {};
+      transferStatesRef.current.set(task.id, state);
+      let lastProgress = 0;
+      const uploadResult = await vfsApi.uploadOptimized(task.targetPath, task.file, shouldOverwrite, (loaded, total) => {
+        if (loaded < total && Date.now() - lastProgress < 100) return;
+        lastProgress = Date.now();
         mutateFiles((prev) => prev.map((f) => {
           if (f.id !== task.id) return f;
           const effectiveTotal = total > 0 ? total : f.size;
           const size = Math.max(f.size, effectiveTotal, loaded);
-          const percent = size > 0 ? Math.min(100, Math.round((loaded / size) * 100)) : 0;
+          const percent = size > 0 ? Math.min(99, Math.round((loaded / size) * 100)) : 0;
           return {
             ...f,
             size,
@@ -511,7 +531,8 @@ export function useUploader(path: string, onUploadComplete: () => void) {
             progress: percent,
           };
         }));
-      });
+      }, batchControllerRef.current?.signal, state);
+      transferStatesRef.current.delete(task.id);
 
       const actualPath = uploadResult?.path || task.targetPath;
       const finalSize = typeof uploadResult?.size === 'number' && uploadResult.size > 0
@@ -525,6 +546,10 @@ export function useUploader(path: string, onUploadComplete: () => void) {
         targetPath: actualPath,
       });
     } catch (err: unknown) {
+      if (batchControllerRef.current?.signal.aborted) {
+        updateFile(task.id, { status: 'error', error: t('Upload cancelled') });
+        return;
+      }
       const error = err instanceof Error ? err.message : t('Upload failed');
       updateFile(task.id, { status: 'error', error, progress: 0 });
       message.error(`${task.relativePath}: ${error}`);
@@ -537,6 +562,7 @@ export function useUploader(path: string, onUploadComplete: () => void) {
 
     const runWorker = async () => {
       while (nextIndex < preparedFiles.length) {
+        if (batchControllerRef.current?.signal.aborted) return;
         const current = preparedFiles[nextIndex];
         nextIndex += 1;
         await uploadPreparedFile(current.task, current.overwrite);
@@ -551,10 +577,12 @@ export function useUploader(path: string, onUploadComplete: () => void) {
     if (!filesRef.current.length) return;
 
     isUploadingRef.current = true;
+    batchControllerRef.current = new AbortController();
     setIsUploading(true);
     try {
       const preparedFiles: Array<{ task: UploadFile; overwrite: boolean }> = [];
       for (const task of filesRef.current) {
+        if (batchControllerRef.current.signal.aborted) break;
         if (task.status !== 'pending' && task.status !== 'waiting') {
           continue;
         }
@@ -566,12 +594,31 @@ export function useUploader(path: string, onUploadComplete: () => void) {
         }
       }
       await uploadPreparedFiles(preparedFiles);
+      if (batchControllerRef.current.signal.aborted) {
+        mutateFiles(previous => previous.map(task => ['pending', 'waiting', 'uploading'].includes(task.status)
+          ? { ...task, status: 'error', error: t('Upload cancelled') } : task));
+      }
       onUploadComplete();
     } finally {
       isUploadingRef.current = false;
+      batchControllerRef.current = null;
       setIsUploading(false);
     }
-  }, [onUploadComplete, processDirectoryTask, prepareFileTask, uploadPreparedFiles]);
+  }, [onUploadComplete, processDirectoryTask, prepareFileTask, uploadPreparedFiles, mutateFiles, t]);
+
+  const cancelUpload = useCallback(() => {
+    batchControllerRef.current?.abort();
+    conflictResolverRef.current?.('skip');
+    conflictResolverRef.current = null;
+    setConflict(null);
+  }, []);
+
+  const retryFailed = useCallback(() => {
+    if (isUploadingRef.current) return;
+    replaceFiles(filesRef.current.map(task => task.status === 'error'
+      ? { ...task, status: 'pending', error: undefined, loadedBytes: 0, progress: 0 } : task));
+    void startUpload();
+  }, [replaceFiles, startUpload]);
 
   const totalFileBytes = useMemo(
     () => files.reduce((acc, f) => acc + (f.type === 'file' ? f.size : 0), 0),
@@ -611,7 +658,7 @@ export function useUploader(path: string, onUploadComplete: () => void) {
     files,
     isModalVisible,
     isUploading,
-    totalProgress: Math.min(100, Math.max(0, totalProgress)),
+    totalProgress: Math.min(isUploading ? 99 : 100, Math.max(0, totalProgress)),
     totalFileBytes,
     uploadedFileBytes,
     conflict,
@@ -622,6 +669,8 @@ export function useUploader(path: string, onUploadComplete: () => void) {
     openFilePicker,
     openDirectoryPicker,
     closeModal,
+    retryFailed,
+    cancelUpload,
     handleFileInputChange,
     handleDirectoryInputChange,
     handleFileDrop,