From 4569f8f6439521bd80541855ac9e2797739e920e Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Tue, 7 Jul 2026 04:09:37 +0000 Subject: [PATCH] Align SFTP, FTP, and SMB protocols with WebDAV behavior - Implement parallel metadata fetching in SFTP and FTP directory listings for accurate file sizes and modification times. - Ensure reliable file uploads in SFTP by awaiting Discord upload completion on file close. - Integrate global FSCache (gfc) across all protocols to ensure cross-protocol consistency. - Optimize WebDAV efficiency by utilizing cached Folder and Resource objects. - Centralize cache invalidation on write operations (create, delete, rename, mkdir). Co-authored-by: VulcanoSoftware <113239901+VulcanoSoftware@users.noreply.github.com> --- dcfs/app/ftp/path_io.py | 107 ++++++++++---- dcfs/app/sftp/handler.py | 270 +++++++++++++++++++++--------------- dcfs/app/smb/backend.py | 110 ++++++++++----- dcfs/app/webdav/__init__.py | 16 ++- dcfs/app/webdav/folder.py | 22 ++- 5 files changed, 334 insertions(+), 191 deletions(-) diff --git a/dcfs/app/ftp/path_io.py b/dcfs/app/ftp/path_io.py index 1e31d41..8bd0273 100644 --- a/dcfs/app/ftp/path_io.py +++ b/dcfs/app/ftp/path_io.py @@ -7,6 +7,7 @@ import aioftp +from dcfs.app.fs_cache import gfc from dcfs.app.utils import normalize_global_path, split_global_path from dcfs.config import get_config from dcfs.core import Clients, Ops @@ -21,10 +22,12 @@ def __init__(self, clients: Clients): super().__init__() self.clients = clients - def _get_ops(self, path: pathlib.PurePosixPath) -> tuple[Optional[Ops], str]: + def _get_ops( + self, path: pathlib.PurePosixPath + ) -> tuple[Optional[Ops], str, Optional[str]]: path_str = str(path) if not path_str: - return None, "/" + return None, "/", None try: path_str = normalize_global_path(path_str) @@ -32,22 +35,26 @@ def _get_ops(self, path: pathlib.PurePosixPath) -> tuple[Optional[Ops], str]: logger.debug(f"Failed to normalize global path: {path_str}") if path_str in (".", "/"): - return None, "/" + return None, "/", None try: client_name, sub_path = split_global_path(path_str) if client_name in self.clients: - return Ops(self.clients[client_name]), "/" + sub_path.lstrip("/") + return ( + Ops(self.clients[client_name]), + "/" + sub_path.lstrip("/"), + client_name, + ) raise FileNotFoundError(f"No such client: {client_name}") except FileNotFoundError: raise except Exception: logger.debug(f"Failed to split global path: {path_str}") - return None, path_str + return None, path_str, None async def exists(self, path: pathlib.PurePosixPath) -> bool: try: - ops, sub_path = self._get_ops(path) + ops, sub_path, _ = self._get_ops(path) except FileNotFoundError: return False @@ -73,7 +80,7 @@ async def exists(self, path: pathlib.PurePosixPath) -> bool: async def is_dir(self, path: pathlib.PurePosixPath) -> bool: try: - ops, sub_path = self._get_ops(path) + ops, sub_path, _ = self._get_ops(path) except FileNotFoundError: return False @@ -91,7 +98,7 @@ async def is_dir(self, path: pathlib.PurePosixPath) -> bool: async def is_file(self, path: pathlib.PurePosixPath) -> bool: try: - ops, sub_path = self._get_ops(path) + ops, sub_path, _ = self._get_ops(path) except FileNotFoundError: return False @@ -111,30 +118,38 @@ async def mkdir( parents: bool = False, exist_ok: bool = False, ) -> None: - ops, sub_path = self._get_ops(path) - if ops is None: + ops, sub_path, client_name = self._get_ops(path) + if ops is None or client_name is None: raise PermissionError("Cannot create directory in root") try: await ops.mkdir(sub_path, parents) + if client_name in gfc: + gfc[client_name].reset(os.path.dirname(sub_path)) except Exception: if not exist_ok: raise async def rmdir(self, path: pathlib.PurePosixPath) -> None: - ops, sub_path = self._get_ops(path) - if ops is None: + ops, sub_path, client_name = self._get_ops(path) + if ops is None or client_name is None: raise PermissionError("Cannot remove client directory") await ops.rm_dir(sub_path, recursive=False) + if client_name in gfc: + gfc[client_name].reset(os.path.dirname(sub_path)) async def unlink(self, path: pathlib.PurePosixPath) -> None: - ops, sub_path = self._get_ops(path) - if ops is None: + ops, sub_path, client_name = self._get_ops(path) + if ops is None or client_name is None: raise PermissionError("Cannot unlink client directory") await ops.rm_file(sub_path) + if client_name in gfc: + gfc[client_name].reset(os.path.dirname(sub_path)) - async def list(self, path: pathlib.PurePosixPath) -> AsyncIterator[pathlib.PurePosixPath]: + async def list( + self, path: pathlib.PurePosixPath + ) -> AsyncIterator[pathlib.PurePosixPath]: try: - ops, sub_path = self._get_ops(path) + ops, sub_path, _ = self._get_ops(path) except FileNotFoundError: return @@ -147,15 +162,30 @@ async def list(self, path: pathlib.PurePosixPath) -> AsyncIterator[pathlib.PureP if sub_path == "/": directory = ops._client.dir_api.root else: - directory = ops.cd(sub_path) + try: + directory = ops.cd(sub_path) + except FileOrDirectoryDoesNotExist: + return + + # Pre-fetch file descriptors in parallel to populate the repository cache. + # This makes subsequent sequential stat calls from the FTP server much faster. + files = directory.find_files() + if files: + await asyncio.gather( + *[ + ops.desc(sub_path.rstrip("/") + "/" + f.name, validate=False) + for f in files + ], + return_exceptions=True, + ) for d in directory.find_dirs(): yield path / d.name - for f in directory.find_files(): + for f in files: yield path / f.name async def stat(self, path: pathlib.PurePosixPath) -> os.stat_result: - ops, sub_path = self._get_ops(path) + ops, sub_path, _ = self._get_ops(path) mode = 0 size = 0 @@ -179,10 +209,10 @@ async def stat(self, path: pathlib.PurePosixPath) -> os.stat_result: return os.stat_result((mode, 0, 0, 0, 0, 0, size, mtime, mtime, mtime)) async def _open(self, path: pathlib.PurePosixPath, mode: str) -> "DCFSFileIO": # type: ignore[override] - ops, sub_path = self._get_ops(path) - if ops is None: + ops, sub_path, client_name = self._get_ops(path) + if ops is None or client_name is None: raise PermissionError("Cannot open root or client directory") - return DCFSFileIO(ops, sub_path, mode) + return DCFSFileIO(ops, sub_path, mode, client_name) async def read(self, file: "DCFSFileIO", block_size: int) -> bytes: # type: ignore[override] return await file.read(block_size) @@ -197,12 +227,20 @@ async def seek(self, file: "DCFSFileIO", offset: int, whence: int = os.SEEK_SET) async def close(self, file: "DCFSFileIO") -> None: # type: ignore[override] await file.close() - async def rename(self, source: pathlib.PurePosixPath, destination: pathlib.PurePosixPath) -> None: - ops_src, sub_src = self._get_ops(source) - ops_dst, sub_dst = self._get_ops(destination) - - if ops_src is None or ops_dst is None or ops_src._client != ops_dst._client: - raise PermissionError("Cannot rename across clients") + async def rename( + self, source: pathlib.PurePosixPath, destination: pathlib.PurePosixPath + ) -> None: + ops_src, sub_src, client_src = self._get_ops(source) + ops_dst, sub_dst, client_dst = self._get_ops(destination) + + if ( + ops_src is None + or ops_dst is None + or ops_src._client != ops_dst._client + or client_src is None + or client_dst is None + ): + raise PermissionError("Cannot rename across clients") # Check if source is dir or file try: @@ -211,12 +249,18 @@ async def rename(self, source: pathlib.PurePosixPath, destination: pathlib.PureP except FileOrDirectoryDoesNotExist: await ops_src.mv_file(sub_src, sub_dst) + if client_src in gfc: + gfc[client_src].reset(os.path.dirname(sub_src)) + if client_dst in gfc: + gfc[client_dst].reset(os.path.dirname(sub_dst)) + class DCFSFileIO: - def __init__(self, ops: Ops, path: str, mode: str): + def __init__(self, ops: Ops, path: str, mode: str, client_name: str): self.ops = ops self.path = path self.mode = mode + self.client_name = client_name self.buffer = bytearray() self.pos = 0 self.closed = False @@ -284,6 +328,9 @@ async def close(self) -> None: for attempt in range(max_retries): try: await self.ops.upload_from_bytes(data, self.path) + if self.client_name in gfc: + gfc[self.client_name].reset(self.path) + gfc[self.client_name].reset(os.path.dirname(self.path)) return except Exception as ex: last_ex = ex @@ -315,7 +362,7 @@ async def seek(self, offset: int, whence: int = os.SEEK_SET) -> int: self.buffer = bytearray() self.pos = offset return self.pos - elif whence == os.SEEK_CUR: + if whence == os.SEEK_CUR: # We don't support moving CUR except by 0 for now to keep it simple, # but usually it's used with 0 to get current pos. self.pos += offset diff --git a/dcfs/app/sftp/handler.py b/dcfs/app/sftp/handler.py index 7dc5385..b12a2b1 100644 --- a/dcfs/app/sftp/handler.py +++ b/dcfs/app/sftp/handler.py @@ -1,13 +1,14 @@ import asyncio +import inspect import logging import os import stat import time -from typing import Any, AsyncIterator, List, Optional +from typing import Any, AsyncIterator, Optional import asyncssh -import inspect +from dcfs.app.fs_cache import gfc from dcfs.app.utils import normalize_global_path, split_global_path from dcfs.config import get_config from dcfs.core import Clients, Ops @@ -22,45 +23,52 @@ def __init__(self, clients: Clients, chan: Any): self.clients = clients super().__init__(chan) - def _get_ops(self, path_bytes: bytes) -> tuple[Optional[Ops], str]: + def _get_ops(self, path_bytes: bytes) -> tuple[Optional[Ops], str, Optional[str]]: path = path_bytes.decode("utf-8", errors="replace") if not path: - return None, "/" + return None, "/", None try: path = normalize_global_path(path) except Exception: logger.debug(f"Failed to normalize global path: {path}") - return None, "/" + return None, "/", None if path in (".", "/"): - return None, "/" + return None, "/", None # Handle empty client name edge case if path.startswith("//"): - return None, "/" + return None, "/", None try: client_name, sub_path = split_global_path(path) if client_name in self.clients: - return Ops(self.clients[client_name]), "/" + sub_path.lstrip("/") - else: - raise asyncssh.SFTPNoSuchFile(f"No such client: {client_name}") + return ( + Ops(self.clients[client_name]), + "/" + sub_path.lstrip("/"), + client_name, + ) + raise asyncssh.SFTPNoSuchFile(f"No such client: {client_name}") except asyncssh.SFTPNoSuchFile: raise except Exception: logger.debug(f"Failed to split global path: {path}") - return None, "/" + return None, "/", None # Fallback to root - return None, "/" + return None, "/", None async def scandir(self, path: bytes) -> AsyncIterator[asyncssh.SFTPName]: # type: ignore[override] - ops, sub_path = self._get_ops(path) + ops, sub_path, _ = self._get_ops(path) # Standard . and .. entries - yield asyncssh.SFTPName(b".", attrs=asyncssh.SFTPAttrs(permissions=stat.S_IFDIR | 0o755)) - yield asyncssh.SFTPName(b"..", attrs=asyncssh.SFTPAttrs(permissions=stat.S_IFDIR | 0o755)) + yield asyncssh.SFTPName( + b".", attrs=asyncssh.SFTPAttrs(permissions=stat.S_IFDIR | 0o755) + ) + yield asyncssh.SFTPName( + b"..", attrs=asyncssh.SFTPAttrs(permissions=stat.S_IFDIR | 0o755) + ) if ops is None: if sub_path == "/": @@ -77,7 +85,9 @@ async def scandir(self, path: bytes) -> AsyncIterator[asyncssh.SFTPName]: # typ try: directory = ops.cd(sub_path) except FileOrDirectoryDoesNotExist: - raise asyncssh.SFTPNoSuchFile(f"No such directory: {path.decode('utf-8')}") + raise asyncssh.SFTPNoSuchFile( + f"No such directory: {path.decode('utf-8')}" + ) for d in directory.find_dirs(): mtime = int(d.modified_at_timestamp / 1000) @@ -87,17 +97,37 @@ async def scandir(self, path: bytes) -> AsyncIterator[asyncssh.SFTPName]: # typ permissions=stat.S_IFDIR | 0o755, mtime=mtime, atime=mtime ), ) - for f in directory.find_files(): - # For files we might not want to fetch full desc just for listing - # as it involves network calls to Discord for each file. - # We provide the basic mode to help clients like FileZilla. - yield asyncssh.SFTPName( - f.name.encode("utf-8"), - attrs=asyncssh.SFTPAttrs(permissions=stat.S_IFREG | 0o644), + + # Fetch file sizes and mtimes in parallel + files = directory.find_files() + if files: + + async def _get_file_attrs(fr): + try: + fd = await ops.desc( + sub_path.rstrip("/") + "/" + fr.name, validate=False + ) + fv = fd.get_latest_version() + size = await ops._client.fc_repo.content_length(fv) + mtime = int(fv.updated_at_timestamp / 1000) + return asyncssh.SFTPAttrs( + permissions=stat.S_IFREG | 0o644, + size=size, + mtime=mtime, + atime=mtime, + ) + except Exception: + return asyncssh.SFTPAttrs(permissions=stat.S_IFREG | 0o644) + + attrs_list = await asyncio.gather( + *[_get_file_attrs(f) for f in files] ) + for f, attrs in zip(files, attrs_list): + yield asyncssh.SFTPName(f.name.encode("utf-8"), attrs=attrs) + async def stat(self, path: bytes) -> asyncssh.SFTPAttrs: # type: ignore[override] - ops, sub_path = self._get_ops(path) + ops, sub_path, _ = self._get_ops(path) mode = 0 size = 0 @@ -118,18 +148,34 @@ async def stat(self, path: bytes) -> asyncssh.SFTPAttrs: # type: ignore[overrid mode = stat.S_IFREG | 0o644 mtime = int(fv.updated_at_timestamp / 1000) except FileOrDirectoryDoesNotExist: - raise asyncssh.SFTPNoSuchFile(f"No such file: {path.decode('utf-8')}") + raise asyncssh.SFTPNoSuchFile( + f"No such file: {path.decode('utf-8')}" + ) - return asyncssh.SFTPAttrs(permissions=mode, size=size, mtime=mtime, atime=mtime) + return asyncssh.SFTPAttrs( + permissions=mode, size=size, mtime=mtime, atime=mtime + ) async def open(self, path: bytes, flags: int, attrs: asyncssh.SFTPAttrs) -> 'DCFSSFTPFileBase': # type: ignore[override] - ops, sub_path = self._get_ops(path) - if ops is None: - raise asyncssh.SFTPPermissionDenied("Cannot open root or client directory") + ops, sub_path, client_name = self._get_ops(path) + if ops is None or client_name is None: + raise asyncssh.SFTPPermissionDenied( + "Cannot open root or client directory" + ) + + if flags & asyncssh.FXF_CREAT: + try: + await ops.touch(sub_path) + if client_name in gfc: + gfc[client_name].reset(os.path.dirname(sub_path)) + except FileOrDirectoryDoesNotExist: + # Parent might not exist, that's fine if we are not creating + # or if touch fails it might be caught later + pass mode = "" - if flags & asyncssh.FXF_WRITE: - mode = "w" + if flags & (asyncssh.FXF_WRITE | asyncssh.FXF_APPEND | asyncssh.FXF_CREAT): + mode = "w" else: mode = "r" @@ -143,8 +189,8 @@ async def open(self, path: bytes, flags: int, attrs: asyncssh.SFTPAttrs) -> 'DCF expected_size = None if mode == "w" and expected_size is not None: - return DCFSSFTPStreamingFile(ops, sub_path, expected_size) - return DCFSSFTPBufferedFile(ops, sub_path, mode) + return DCFSSFTPStreamingFile(ops, sub_path, expected_size, client_name) + return DCFSSFTPBufferedFile(ops, sub_path, mode, client_name) async def read(self, file_obj: object, offset: int, size: int) -> bytes: # type: ignore[override] if isinstance(file_obj, DCFSSFTPFileBase): @@ -182,29 +228,47 @@ async def fstat(self, file_obj: object) -> asyncssh.SFTPAttrs: # type: ignore[o return result async def mkdir(self, path: bytes, attrs: asyncssh.SFTPAttrs) -> None: # type: ignore[override] - ops, sub_path = self._get_ops(path) - if ops is None: - raise asyncssh.SFTPPermissionDenied("Cannot create directory in root") + ops, sub_path, client_name = self._get_ops(path) + if ops is None or client_name is None: + raise asyncssh.SFTPPermissionDenied( + "Cannot create directory in root" + ) await ops.mkdir(sub_path, parents=False) + if client_name in gfc: + gfc[client_name].reset(os.path.dirname(sub_path)) async def rmdir(self, path: bytes) -> None: # type: ignore[override] - ops, sub_path = self._get_ops(path) - if ops is None: - raise asyncssh.SFTPPermissionDenied("Cannot remove client directory") + ops, sub_path, client_name = self._get_ops(path) + if ops is None or client_name is None: + raise asyncssh.SFTPPermissionDenied( + "Cannot remove client directory" + ) await ops.rm_dir(sub_path, recursive=False) + if client_name in gfc: + gfc[client_name].reset(os.path.dirname(sub_path)) async def remove(self, path: bytes) -> None: # type: ignore[override] - ops, sub_path = self._get_ops(path) - if ops is None: - raise asyncssh.SFTPPermissionDenied("Cannot remove client directory") + ops, sub_path, client_name = self._get_ops(path) + if ops is None or client_name is None: + raise asyncssh.SFTPPermissionDenied( + "Cannot remove client directory" + ) await ops.rm_file(sub_path) + if client_name in gfc: + gfc[client_name].reset(os.path.dirname(sub_path)) async def rename(self, oldpath: bytes, newpath: bytes) -> None: # type: ignore[override] - ops_src, sub_src = self._get_ops(oldpath) - ops_dst, sub_dst = self._get_ops(newpath) - - if ops_src is None or ops_dst is None or ops_src._client != ops_dst._client: - raise asyncssh.SFTPPermissionDenied("Cannot rename across clients") + ops_src, sub_src, client_src = self._get_ops(oldpath) + ops_dst, sub_dst, client_dst = self._get_ops(newpath) + + if ( + ops_src is None + or ops_dst is None + or ops_src._client != ops_dst._client + or client_src is None + or client_dst is None + ): + raise asyncssh.SFTPPermissionDenied("Cannot rename across clients") try: ops_src.cd(sub_src) @@ -212,6 +276,11 @@ async def rename(self, oldpath: bytes, newpath: bytes) -> None: # type: ignore[ except FileOrDirectoryDoesNotExist: await ops_src.mv_file(sub_src, sub_dst) + if client_src in gfc: + gfc[client_src].reset(os.path.dirname(sub_src)) + if client_dst in gfc: + gfc[client_dst].reset(os.path.dirname(sub_dst)) + async def realpath(self, path: bytes) -> bytes: # type: ignore[override] try: decoded = path.decode("utf-8") @@ -253,13 +322,13 @@ async def fstat(self) -> asyncssh.SFTPAttrs: class DCFSSFTPBufferedFile(DCFSSFTPFileBase): - def __init__(self, ops: Ops, path: str, mode: str): + def __init__(self, ops: Ops, path: str, mode: str, client_name: str): self.ops = ops self.path = path self.mode = mode + self.client_name = client_name self.buffer = bytearray() self.closed = False - self._upload_task: Optional[asyncio.Task[None]] = None async def read(self, offset: int, size: int) -> bytes: if "r" not in self.mode: @@ -303,55 +372,38 @@ async def close(self) -> None: self.closed = True if "w" in self.mode and self.buffer: - # Fire-and-forget: upload happens in the background so the - # SFTP client does not hang waiting for the Discord upload. data = bytes(self.buffer) self.buffer.clear() - self._upload_task = asyncio.create_task( - self._do_upload(data) - ) - self._upload_task.add_done_callback(self._on_upload_done) - async def _do_upload(self, data: bytes) -> None: - """Upload the buffered data with exponential backoff retry on transient errors.""" - last_ex: Optional[Exception] = None - cfg = get_config().dcfs.download - for attempt in range(cfg.upload_max_retries): - try: - await self.ops.upload_from_bytes(data, self.path) - logger.info(f"Background upload completed for {self.path}") - return - except Exception as ex: - last_ex = ex - if not _is_transient(ex): - logger.error( - f"Permanent upload failure for {self.path}: {ex}" + last_ex: Optional[Exception] = None + cfg = get_config().dcfs.download + for attempt in range(cfg.upload_max_retries): + try: + await self.ops.upload_from_bytes(data, self.path) + if self.client_name in gfc: + gfc[self.client_name].reset(self.path) + gfc[self.client_name].reset(os.path.dirname(self.path)) + logger.info(f"Upload completed for {self.path}") + return + except Exception as ex: + last_ex = ex + if not _is_transient(ex): + logger.error(f"Permanent upload failure for {self.path}: {ex}") + raise + + delay = cfg.upload_base_retry_delay * (2**attempt) + logger.warning( + f"Transient upload failure for {self.path} ({ex}). " + f"Retry {attempt + 1}/{cfg.upload_max_retries} in {delay:.1f}s..." ) - return # Don't raise — fire-and-forget, just log - - delay = cfg.upload_base_retry_delay * (2**attempt) - logger.warning( - f"Transient upload failure for {self.path} ({ex}). " - f"Retry {attempt + 1}/{cfg.upload_max_retries} in {delay:.1f}s..." - ) - await asyncio.sleep(delay) - - logger.error( - f"Background upload failed for {self.path} after " - f"{cfg.upload_max_retries} retries: {last_ex}", - exc_info=last_ex, - ) + await asyncio.sleep(delay) - def _on_upload_done(self, task: asyncio.Task[None]) -> None: - try: - exc = task.exception() - if exc is not None: - logger.error( - f"Background upload failed for {self.path}: {exc}", - exc_info=exc, - ) - except asyncio.CancelledError: - logger.warning(f"Background upload cancelled for {self.path}") + logger.error( + f"Upload failed for {self.path} after " + f"{cfg.upload_max_retries} retries: {last_ex}", + exc_info=last_ex, + ) + raise last_ex # type: ignore[misc] async def fstat(self) -> asyncssh.SFTPAttrs: if "w" in self.mode: @@ -376,21 +428,19 @@ async def fstat(self) -> asyncssh.SFTPAttrs: class DCFSSFTPStreamingFile(DCFSSFTPFileBase): - def __init__(self, ops: Ops, path: str, expected_size: int): + def __init__(self, ops: Ops, path: str, expected_size: int, client_name: str): self.ops = ops self.path = path self.expected_size = expected_size + self.client_name = client_name self.closed = False self._queue: asyncio.Queue[Optional[bytes]] = asyncio.Queue(maxsize=32) self._pending: dict[int, bytes] = {} self._next_offset = 0 # Start the upload task immediately so it consumes from the queue - # concurrently with writes. The task is not awaited on close() — - # instead it runs to completion in the background so the SFTP - # client does not hang waiting for the Discord upload to finish. + # concurrently with writes. self._upload_task = asyncio.create_task(self._run_upload()) - self._upload_task.add_done_callback(self._on_upload_done) async def _stream(self) -> AsyncIterator[bytes]: while True: @@ -417,6 +467,9 @@ async def _run_upload(self) -> None: await self.ops.upload_from_stream( self._stream(), self.expected_size, self.path ) + if self.client_name in gfc: + gfc[self.client_name].reset(self.path) + gfc[self.client_name].reset(os.path.dirname(self.path)) return except Exception as ex: last_ex = ex @@ -424,7 +477,7 @@ async def _run_upload(self) -> None: logger.error( f"Permanent streaming upload failure for {self.path}: {ex}" ) - return # Don't raise — fire-and-forget, just log + raise # If data has already been consumed from the queue we cannot # retry because the queue items are gone. @@ -435,7 +488,7 @@ async def _run_upload(self) -> None: f"Streaming upload failed after {self._next_offset} bytes " f"for {self.path}: {ex}. Cannot retry — stream consumed." ) - return + raise delay = cfg.upload_base_retry_delay * (2**attempt) logger.warning( @@ -449,17 +502,7 @@ async def _run_upload(self) -> None: f"{cfg.upload_max_retries} retries: {last_ex}", exc_info=last_ex, ) - - def _on_upload_done(self, task: asyncio.Task[None]) -> None: - try: - exc = task.exception() - if exc is not None: - logger.error( - f"Background upload failed for {self.path}: {exc}", - exc_info=exc, - ) - except asyncio.CancelledError: - logger.warning(f"Background upload cancelled for {self.path}") + raise last_ex # type: ignore[misc] async def read(self, offset: int, size: int) -> bytes: raise asyncssh.SFTPOpUnsupported("Streaming file handle does not support reads") @@ -504,10 +547,9 @@ async def close(self) -> None: f"Upload incomplete: received {self._next_offset} of {self.expected_size} bytes" ) - # Signal end-of-stream to the upload task and return immediately. - # The Discord upload continues in the background — we do NOT await - # the upload task here so the SFTP client doesn't hang at 100%. + # Signal end-of-stream to the upload task and await it. await self._queue.put(None) + await self._upload_task async def fstat(self) -> asyncssh.SFTPAttrs: now = int(time.time()) diff --git a/dcfs/app/smb/backend.py b/dcfs/app/smb/backend.py index f1b64eb..0e778b8 100644 --- a/dcfs/app/smb/backend.py +++ b/dcfs/app/smb/backend.py @@ -4,6 +4,7 @@ import time from typing import Optional +from dcfs.app.fs_cache import gfc from dcfs.app.utils import normalize_global_path, split_global_path from dcfs.config import get_config from dcfs.core import Clients, Ops @@ -20,9 +21,9 @@ def __init__(self, clients: Clients, loop: asyncio.AbstractEventLoop): self.clients = clients self.loop = loop - def _get_ops(self, path: str) -> tuple[Optional[Ops], str]: + def _get_ops(self, path: str) -> tuple[Optional[Ops], str, Optional[str]]: if not path: - return None, "/" + return None, "/", None try: path = normalize_global_path(path) @@ -31,22 +32,26 @@ def _get_ops(self, path: str) -> tuple[Optional[Ops], str]: logger.debug(f"Failed to normalize global path: {path}") if path in (".", "/"): - return None, "/" + return None, "/", None try: client_name, sub_path = split_global_path(path) if client_name in self.clients: - return Ops(self.clients[client_name]), "/" + sub_path.lstrip("/") + return ( + Ops(self.clients[client_name]), + "/" + sub_path.lstrip("/"), + client_name, + ) raise OSError(f"No such client: {client_name}") except OSError: raise except Exception: - logger.debug(f"Failed to split global path: {path}") - return None, path + logger.debug(f"Failed to split global path: {path}") + return None, path, None def listPath(self, path: str, filter: str = "*"): try: - ops, sub_path = self._get_ops(path) + ops, sub_path, _ = self._get_ops(path) except OSError: return [] @@ -115,7 +120,7 @@ def __init__(self, name, is_dir, size, mtime): return SMBStat(name, is_dir, size, mtime) def getFileStat(self, path: str): - ops, sub_path = self._get_ops(path) + ops, sub_path, _ = self._get_ops(path) if ops is None or sub_path == "/": return self._make_stat(os.path.basename(path) or "/", is_dir=True) @@ -148,48 +153,80 @@ async def _file_size_for_ref(ops: Ops, sub_path: str, file_name: str) -> int: return 0 def openFile(self, path: str, mode: str): - ops, sub_path = self._get_ops(path) - if ops is None: - raise OSError("Cannot open root") - return DCFSSMBFile(ops, sub_path, mode, self.loop) + ops, sub_path, client_name = self._get_ops(path) + if ops is None or client_name is None: + raise OSError("Cannot open root") + return DCFSSMBFile(ops, sub_path, mode, self.loop, client_name) def mkdir(self, path: str): - ops, sub_path = self._get_ops(path) - if ops: - future = asyncio.run_coroutine_threadsafe(ops.mkdir(sub_path, parents=False), self.loop) + ops, sub_path, client_name = self._get_ops(path) + if ops and client_name: + future = asyncio.run_coroutine_threadsafe( + ops.mkdir(sub_path, parents=False), self.loop + ) future.result() + if client_name in gfc: + gfc[client_name].reset(os.path.dirname(sub_path)) def rmdir(self, path: str): - ops, sub_path = self._get_ops(path) - if ops: - future = asyncio.run_coroutine_threadsafe(ops.rm_dir(sub_path, recursive=False), self.loop) - future.result() + ops, sub_path, client_name = self._get_ops(path) + if ops and client_name: + future = asyncio.run_coroutine_threadsafe( + ops.rm_dir(sub_path, recursive=False), self.loop + ) + future.result() + if client_name in gfc: + gfc[client_name].reset(os.path.dirname(sub_path)) def deleteFile(self, path: str): - ops, sub_path = self._get_ops(path) - if ops: - future = asyncio.run_coroutine_threadsafe(ops.rm_file(sub_path), self.loop) - future.result() + ops, sub_path, client_name = self._get_ops(path) + if ops and client_name: + future = asyncio.run_coroutine_threadsafe( + ops.rm_file(sub_path), self.loop + ) + future.result() + if client_name in gfc: + gfc[client_name].reset(os.path.dirname(sub_path)) def rename(self, oldpath: str, newpath: str): - ops_src, sub_src = self._get_ops(oldpath) - ops_dst, sub_dst = self._get_ops(newpath) - if ops_src and ops_dst and ops_src._client == ops_dst._client: - async def _rename(): - try: - await ops_src.mv_file(sub_src, sub_dst) - except Exception: - await ops_src.mv_dir(sub_src, sub_dst) - - future = asyncio.run_coroutine_threadsafe(_rename(), self.loop) - future.result() + ops_src, sub_src, client_src = self._get_ops(oldpath) + ops_dst, sub_dst, client_dst = self._get_ops(newpath) + if ( + ops_src + and ops_dst + and ops_src._client == ops_dst._client + and client_src + and client_dst + ): + + async def _rename(): + try: + await ops_src.mv_file(sub_src, sub_dst) + except Exception: + await ops_src.mv_dir(sub_src, sub_dst) + + future = asyncio.run_coroutine_threadsafe(_rename(), self.loop) + future.result() + + if client_src in gfc: + gfc[client_src].reset(os.path.dirname(sub_src)) + if client_dst in gfc: + gfc[client_dst].reset(os.path.dirname(sub_dst)) class DCFSSMBFile: - def __init__(self, ops: Ops, path: str, mode: str, loop: asyncio.AbstractEventLoop): + def __init__( + self, + ops: Ops, + path: str, + mode: str, + loop: asyncio.AbstractEventLoop, + client_name: str, + ): self.ops = ops self.path = path self.mode = mode self.loop = loop + self.client_name = client_name self.pos = 0 self.buffer = bytearray() @@ -238,6 +275,9 @@ def close(self): self.ops.upload_from_bytes(data, self.path), self.loop ) future.result() # Wait for Discord upload to finish + if self.client_name in gfc: + gfc[self.client_name].reset(self.path) + gfc[self.client_name].reset(os.path.dirname(self.path)) return except Exception as ex: last_ex = ex diff --git a/dcfs/app/webdav/__init__.py b/dcfs/app/webdav/__init__.py index cc96d65..0027e48 100644 --- a/dcfs/app/webdav/__init__.py +++ b/dcfs/app/webdav/__init__.py @@ -11,15 +11,21 @@ async def _get_member(path: str, clients: Clients) -> Optional[Member]: if path == "" or path == "/": - folders = { - client_name: Folder("/", clients[client_name]) - for client_name in clients.keys() - } + folders = {} + for client_name, client in clients.items(): + cache = gfc[client_name] + if not (folder := cache.get("/")): + folder = Folder("/", client) + cache.set("/", folder) + folders[client_name] = folder return RootFolder(folders) client_name, sub_path = split_global_path(path) + cache = gfc[client_name] - root = Folder("/", clients[client_name]) + if not (root := cache.get("/")): + root = Folder("/", clients[client_name]) + cache.set("/", root) if res := await root.member(sub_path.lstrip("/")): return res diff --git a/dcfs/app/webdav/folder.py b/dcfs/app/webdav/folder.py index 25e7174..51d193a 100644 --- a/dcfs/app/webdav/folder.py +++ b/dcfs/app/webdav/folder.py @@ -35,15 +35,23 @@ async def member(self, path: str): if path_parts[0] == "": return self - if fr := self.__sub_files.get(path_parts[0]): - return Resource(self._sub_path(path_parts[0]), self.__client, fr=fr) + name = path_parts[0] + relative_sub_path = self._sub_path(name) + + if fr := self.__sub_files.get(name): + if not (res := self.fs_cache.get(relative_sub_path)): + res = Resource(relative_sub_path, self.__client, fr=fr) + self.fs_cache.set(relative_sub_path, res) + return res + + if name in self.__sub_folders: + if not (sub_folder := self.fs_cache.get(relative_sub_path)): + sub_folder = Folder(f"{relative_sub_path}/", self.__client) + self.fs_cache.set(relative_sub_path, sub_folder) - if path_parts[0] in self.__sub_folders: if len(path_parts) > 1: - return await Folder( - f"{self._sub_path(path_parts[0])}/", self.__client - ).member(path_parts[1]) - return Folder(f"{self._sub_path(path_parts[0])}/", self.__client) + return await sub_folder.member(path_parts[1]) + return sub_folder return None