From cab989d636597eefdd438be5e80aff8060c38ebc Mon Sep 17 00:00:00 2001 From: shiyu Date: Sat, 3 Oct 2026 12:25:50 +0800 Subject: [PATCH] feat: optimize chunked uploads and concurrent downloads --- domain/adapters/providers/local.py | 1 + domain/adapters/providers/s3.py | 202 +++++++--- domain/adapters/providers/webdav.py | 270 +++++++------ domain/virtual_fs/api.py | 96 ++++- domain/virtual_fs/chunk_uploads.py | 212 ++++++++++ main.py | 1 + pyproject.toml | 1 + tests/test_adapter_transfers.py | 366 ++++++++++++++++++ tests/test_web_transfers.py | 219 +++++++++++ uv.lock | 2 + web/src/api/transfers.ts | 259 +++++++++++++ web/src/api/vfs.ts | 68 +++- web/src/i18n/locales/en.json | 5 +- web/src/i18n/locales/zh.json | 5 +- .../FileExplorerPage/FileExplorerPage.tsx | 2 + .../components/Modals/UploadModal.tsx | 10 +- .../FileExplorerPage/hooks/useFileActions.tsx | 81 +++- .../FileExplorerPage/hooks/useUploader.ts | 69 +++- 18 files changed, 1649 insertions(+), 220 deletions(-) create mode 100644 domain/virtual_fs/chunk_uploads.py create mode 100644 tests/test_adapter_transfers.py create mode 100644 tests/test_web_transfers.py create mode 100644 web/src/api/transfers.ts diff --git a/domain/adapters/providers/local.py b/domain/adapters/providers/local.py index c2b1cce..ca323f0 100644 --- a/domain/adapters/providers/local.py +++ b/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), diff --git a/domain/adapters/providers/s3.py b/domain/adapters/providers/s3.py index 10efcb8..c18cc13 100644 --- a/domain/adapters/providers/s3.py +++ b/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", diff --git a/domain/adapters/providers/webdav.py b/domain/adapters/providers/webdav.py index 001ea54..87d1135 100644 --- a/domain/adapters/providers/webdav.py +++ b/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}, diff --git a/domain/virtual_fs/api.py b/domain/virtual_fs/api.py index caf984b..820c263 100644 --- a/domain/virtual_fs/api.py +++ b/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") diff --git a/domain/virtual_fs/chunk_uploads.py b/domain/virtual_fs/chunk_uploads.py new file mode 100644 index 0000000..ac0caf9 --- /dev/null +++ b/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} diff --git a/main.py b/main.py index 706a5a1..88c9bae 100644 --- a/main.py +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index 026e347..89cefba 100644 --- a/pyproject.toml +++ b/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", diff --git a/tests/test_adapter_transfers.py b/tests/test_adapter_transfers.py new file mode 100644 index 0000000..c0bd20f --- /dev/null +++ b/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() diff --git a/tests/test_web_transfers.py b/tests/test_web_transfers.py new file mode 100644 index 0000000..4d3efd0 --- /dev/null +++ b/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() diff --git a/uv.lock b/uv.lock index 3520382..46af78d 100644 --- a/uv.lock +++ b/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" }, diff --git a/web/src/api/transfers.ts b/web/src/api/transfers.ts new file mode 100644 index 0000000..910baf9 --- /dev/null +++ b/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 { + const token = localStorage.getItem('token'); + return token ? { Authorization: `Bearer ${token}` } : {}; +} + +class RequestPool { + private active = 0; + private waiting: Array<() => void> = []; + + async run(work: () => Promise, signal?: AbortSignal): Promise { + signal?.throwIfAborted(); + if (this.active >= 6) { + await new Promise((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(url: string, options: RequestInit = {}): Promise { + 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(work: () => Promise, signal?: AbortSignal): Promise { + 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((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, 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; + close(): Promise; + abort(): Promise; +} + +interface SaveFileHandle { + createWritable(): Promise; +} + +export function selectDownloadTarget(name: string): Promise | undefined { + if (!window.isSecureContext) return undefined; + const picker = (window as Window & { showSaveFilePicker?: + (options: { suggestedName: string }) => Promise }).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 { + 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 = { ...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[] = []; + 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' }); +} diff --git a/web/src/api/vfs.ts b/web/src/api/vfs.ts index 2f3c1e0..3c8ebcf 100644 --- a/web/src/api/vfs.ts +++ b/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(`/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(/^\/+/, '')); diff --git a/web/src/i18n/locales/en.json b/web/src/i18n/locales/en.json index ac7d85c..17b3362 100644 --- a/web/src/i18n/locales/en.json +++ b/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" } diff --git a/web/src/i18n/locales/zh.json b/web/src/i18n/locales/zh.json index 58e5601..a308e11 100644 --- a/web/src/i18n/locales/zh.json +++ b/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": "下载完成" } diff --git a/web/src/pages/FileExplorerPage/FileExplorerPage.tsx b/web/src/pages/FileExplorerPage/FileExplorerPage.tsx index 671294f..9e899f5 100644 --- a/web/src/pages/FileExplorerPage/FileExplorerPage.tsx +++ b/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} /> diff --git a/web/src/pages/FileExplorerPage/components/Modals/UploadModal.tsx b/web/src/pages/FileExplorerPage/components/Modals/UploadModal.tsx index eab9671..50c731b 100644 --- a/web/src/pages/FileExplorerPage/components/Modals/UploadModal.tsx +++ b/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 = ({ conflict, onClose, onStartUpload, + onRetryFailed, + onCancelUpload, onResolveConflict, }) => { const { t } = useI18n(); @@ -120,6 +124,10 @@ const UploadModal: React.FC = ({ maskClosable={!isUploading} onCancel={onClose} footer={[ + isUploading && , + summary.failures > 0 && , , diff --git a/web/src/pages/FileExplorerPage/hooks/useFileActions.tsx b/web/src/pages/FileExplorerPage/hooks/useFileActions.tsx index c8a88c3..bce310c 100644 --- a/web/src/pages/FileExplorerPage/hooks/useFileActions.tsx +++ b/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()); + 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:
+
+ {entry.name} +
+ +
}); + }; 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) { diff --git a/web/src/pages/FileExplorerPage/hooks/useUploader.ts b/web/src/pages/FileExplorerPage/hooks/useUploader.ts index e57e5df..5c5fe24 100644 --- a/web/src/pages/FileExplorerPage/hooks/useUploader.ts +++ b/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>(new Set()); const filesRef = useRef(files); const isUploadingRef = useRef(false); + const transferStatesRef = useRef(new Map()); + const batchControllerRef = useRef(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(null); const directoryInputRef = useRef(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,