perf(sftp): make downloads substantially faster

dcfs/app/sftp/__init__.py

@@ -2,6 +2,7 @@
 import os
 
 import asyncssh
+from asyncssh.encryption import get_encryption_algs
 
 from dcfs.config import DATA_DIR, Config
 from dcfs.core import Clients
@@ -10,6 +11,24 @@
 
 logger = logging.getLogger(__name__)
 
+# AES-GCM runs on the CPU's AES instructions and encrypts roughly five times
+# faster than asyncssh's ChaCha20-Poly1305, which is the default first choice.
+# Every other algorithm keeps its original relative order so no client loses
+# compatibility.
+PREFERRED_ENCRYPTION_ALGS = (
+    "aes128-gcm@openssh.com",
+    "aes256-gcm@openssh.com",
+    "aes128-ctr",
+    "aes256-ctr",
+)
+
+
+def _encryption_algs() -> list[str]:
+    supported = [alg.decode("ascii") for alg in get_encryption_algs()]
+    preferred = [alg for alg in PREFERRED_ENCRYPTION_ALGS if alg in supported]
+    return preferred + [alg for alg in supported if alg not in preferred]
+
+
 class DCFSSFTPServer(asyncssh.SSHServer):
     def __init__(self, clients: Clients, config: Config):
         self.clients = clients
@@ -58,7 +77,11 @@ def sftp_factory(channel):
         config.dcfs.sftp.host,
         config.dcfs.sftp.port,
         server_host_keys=[host_key],
-        sftp_factory=sftp_factory
+        sftp_factory=sftp_factory,
+        encryption_algs=_encryption_algs(),
+        # File content is already compressed and/or encrypted, so compression
+        # only burns CPU that the transfer needs.
+        compression_algs=["none"],
     )
 
 async def run_sftp_server(server: asyncssh.SSHListener, host: str, port: int):

dcfs/app/sftp/handler.py

@@ -4,6 +4,7 @@
 import os
 import stat
 import time
+from collections import deque
 from typing import Any, AsyncGenerator, AsyncIterator, Optional, cast
 
 import asyncssh
@@ -333,10 +334,13 @@ def __init__(self, ops: Ops, path: str, mode: str, client_name: str):
         self.buffer = bytearray()
         self.closed = False
 
-        # Read streaming state
+        # Read streaming state. Buffered data is kept as the chunks handed
+        # over by the download stream, tagged with their absolute file
+        # offset, so serving a read never copies or shifts the whole buffer.
         self._read_stream: Optional[AsyncIterator[bytes]] = None
         self._buf_offset = 0
-        self._read_buf = bytearray()
+        self._buf_len = 0
+        self._chunks: deque[tuple[int, bytes]] = deque()
         self._read_lock = asyncio.Lock()
         self._cached_attrs: Optional[asyncssh.SFTPAttrs] = None
         self._highest_offset = 0
@@ -380,8 +384,59 @@ async def _stop_prefetch(self) -> None:
         self._prefetch_queue = None
         self._prefetch_eof = False
 
+    def _buf_end(self) -> int:
+        return self._buf_offset + self._buf_len
+
+    def _append_chunk(self, chunk: bytes) -> None:
+        if not isinstance(chunk, bytes):
+            chunk = bytes(chunk)
+        self._chunks.append((self._buf_end(), chunk))
+        self._buf_len += len(chunk)
+
+    def _extract(self, offset: int, size: int) -> bytes:
+        """Copy ``size`` bytes starting at ``offset`` out of the buffer."""
+        if size <= 0 or offset < self._buf_offset or offset >= self._buf_end():
+            return b""
+
+        out: Optional[bytearray] = None
+        pos = offset
+        remaining = size
+        for start, chunk in self._chunks:
+            end = start + len(chunk)
+            if end <= pos:
+                continue
+            if start > pos:
+                break
+            rel = pos - start
+            piece = chunk[rel : rel + remaining]
+            if out is None and len(piece) == remaining:
+                # Whole request satisfied by a single chunk: no extra copy.
+                return piece
+            if out is None:
+                out = bytearray()
+            out += piece
+            remaining -= len(piece)
+            pos += len(piece)
+            if remaining == 0:
+                break
+
+        return bytes(out) if out else b""
+
+    def _prune(self) -> None:
+        """Drop chunks that fall behind the backward-retain window."""
+        keep_from = self._highest_offset - self.MAX_BACKWARD_RETAIN
+        while self._chunks:
+            start, chunk = self._chunks[0]
+            end = start + len(chunk)
+            if end > keep_from:
+                break
+            self._chunks.popleft()
+            self._buf_len -= len(chunk)
+            self._buf_offset = end
+
     async def _start_prefetch(self, offset: int) -> None:
-        self._read_buf = bytearray()
+        self._chunks.clear()
+        self._buf_len = 0
         self._buf_offset = offset
         self._highest_offset = offset
         self._read_stream = await self.ops.download(
@@ -397,66 +452,46 @@ async def _start_prefetch(self, offset: int) -> None:
             self._run_prefetch(self._read_stream, self._prefetch_queue)
         )
 
+    async def _fill_until(self, target_end: int) -> None:
+        while self._buf_end() < target_end and not self._prefetch_eof:
+            if self._prefetch_queue is None:
+                break
+            item = await self._prefetch_queue.get()
+            if item is None:
+                self._prefetch_eof = True
+                break
+            if isinstance(item, Exception):
+                self._prefetch_eof = True
+                raise item
+            self._append_chunk(item)
+
     async def read(self, offset: int, size: int) -> bytes:
         if "r" not in self.mode:
             raise asyncssh.SFTPPermissionDenied("File not open for reading")
 
         async with self._read_lock:
-            buf_end = self._buf_offset + len(self._read_buf)
-
             can_reuse_stream = (
                 self._read_stream is not None
-                and self._buf_offset <= offset <= buf_end + self.MAX_FORWARD_SKIP
+                and self._buf_offset <= offset <= self._buf_end() + self.MAX_FORWARD_SKIP
             )
 
             if not can_reuse_stream:
                 await self._stop_prefetch()
                 await self._start_prefetch(offset)
 
             target_end = offset + size
-            while (self._buf_offset + len(self._read_buf) < target_end) and not self._prefetch_eof:
-                if self._prefetch_queue is None:
-                    break
-                item = await self._prefetch_queue.get()
-                if item is None:
-                    self._prefetch_eof = True
-                    break
-                if isinstance(item, Exception):
-                    self._prefetch_eof = True
-                    raise item
-                self._read_buf.extend(item)
+            await self._fill_until(target_end)
 
             # If stream reached EOF before reaching offset, restart stream at offset
-            if self._prefetch_eof and (self._buf_offset + len(self._read_buf) <= offset) and size > 0:
+            if self._prefetch_eof and self._buf_end() <= offset and size > 0:
                 await self._stop_prefetch()
                 await self._start_prefetch(offset)
-                while (self._buf_offset + len(self._read_buf) < target_end) and not self._prefetch_eof:
-                    if self._prefetch_queue is None:
-                        break
-                    item = await self._prefetch_queue.get()
-                    if item is None:
-                        self._prefetch_eof = True
-                        break
-                    if isinstance(item, Exception):
-                        self._prefetch_eof = True
-                        raise item
-                    self._read_buf.extend(item)
-
-            rel_offset = offset - self._buf_offset
-            if rel_offset >= 0 and rel_offset < len(self._read_buf):
-                data = bytes(self._read_buf[rel_offset : rel_offset + size])
-            else:
-                data = b""
+                await self._fill_until(target_end)
 
-            self._highest_offset = max(self._highest_offset, offset + len(data))
+            data = self._extract(offset, size)
 
-            # Prune buffer behind prune_target to keep memory bounded
-            prune_target = self._highest_offset - self.MAX_BACKWARD_RETAIN
-            if prune_target > self._buf_offset:
-                discard = min(prune_target - self._buf_offset, len(self._read_buf))
-                if discard > 0:
-                    self._read_buf = self._read_buf[discard:]
-                    self._buf_offset += discard
+            self._highest_offset = max(self._highest_offset, offset + len(data))
+            self._prune()
 
             return data
 

dcfs/core/api/message/__init__.py

@@ -29,6 +29,13 @@
 DELETE_BATCH_SIZE = 100
 
 DISCORD_MSG_LIMIT = 4000
+
+# Bounds on the number of concurrent CDN range requests used to fetch a single
+# message attachment, and how many chunks each of them may run ahead.
+MIN_PARALLEL_RANGE_REQUESTS = 4
+MAX_PARALLEL_RANGE_REQUESTS = 8
+PARALLEL_RANGE_TARGET_SIZE = 2 * 1024 * 1024
+PARALLEL_QUEUE_MAXSIZE = 32
 OVERFLOW_SENTINEL = "DCFS_OVERFLOW"
 OVERFLOW_FILENAME = "overflow.json"
 
@@ -175,10 +182,22 @@ def split_download_tasks(
     def _size(begin: int, end: int) -> int:
         return end - begin + 1
 
+    @staticmethod
+    def _parallel_split_count(size: int) -> int:
+        """Number of concurrent CDN range requests for a ``size`` byte range.
+
+        Scales with the range size so large parts use more connections, while
+        staying bounded to avoid per-request overhead outweighing the added
+        bandwidth.
+        """
+        scaled = size // PARALLEL_RANGE_TARGET_SIZE
+        return max(MIN_PARALLEL_RANGE_REQUESTS,
+                   min(MAX_PARALLEL_RANGE_REQUESTS, scaled))
+
     async def download_file_parallel(self, message_id: int, begin: int, end: int):
         # Split the range into concurrent sub-range downloads so we can
         # utilise CDN bandwidth better for large single-part files.
-        n = 4
+        n = self._parallel_split_count(self._size(begin, end))
         sub_ranges = list(self.split_download_tasks(begin, end, n))
 
         resps = await asyncio.gather(*[
@@ -195,7 +214,7 @@ async def download_file_parallel(self, message_id: int, begin: int, end: int):
         ])
 
         queues: list[asyncio.Queue[object]] = [
-            asyncio.Queue(maxsize=32) for _ in range(n)
+            asyncio.Queue(maxsize=PARALLEL_QUEUE_MAXSIZE) for _ in range(n)
         ]
 
         async def _producer(

dcfs/core/repository/impl/file_content/__init__.py

@@ -26,7 +26,7 @@ async def _empty_iterator() -> AsyncIterator[bytes]:
 
 
 class DCMsgFileContentRepository(IFileContentRepository):
-    def __init__(self, message_api: MessageApi, max_concurrent_parts: int = 3):
+    def __init__(self, message_api: MessageApi, max_concurrent_parts: int = 6):
         self._message_api = message_api
         self._download_semaphore = asyncio.Semaphore(max_concurrent_parts)
         self._upload_semaphore = asyncio.Semaphore(max_concurrent_parts)