feat: vendor russh-sftp with serde_bytes perf fix

CHANGELOG.md

@@ -5,6 +5,14 @@ All notable changes to bssh will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Added
+- Internal fork of `russh-sftp` as `crates/bssh-russh-sftp` with a `serde_bytes` performance fix for `SSH_FXP_WRITE` and `SSH_FXP_DATA` packets. The upstream serde derive routes `Vec<u8>` through `deserialize_seq` (byte-by-byte), accounting for ~42% of server CPU during 1 GiB SFTP uploads in `perf` profiling. Annotating the `data` fields with `#[serde(with = "serde_bytes")]` and implementing wire-compatible `serialize_bytes` on the SFTP `Serializer` routes through the existing bulk `deserialize_byte_buf`/`try_get_bytes` path. Measured impact on a CPU-bound host (Xeon Silver 4214): 1 GiB SFTP upload throughput improves from 74.8 MiB/s to 96.4 MiB/s (+29%), closing the gap to OpenSSH `sftp-server` from ~26% to ~5%.
+
+### Changed
+- Switched the top-level `russh-sftp` dependency from crates.io `russh-sftp = "2.1.1"` to `russh-sftp = { package = "bssh-russh-sftp", version = "2.1.1", path = "crates/bssh-russh-sftp" }`. All existing `use russh_sftp::...` imports continue to work unchanged.
+
 ## [2.1.1] - 2026-04-17
 
 ### Fixed

Cargo.toml

@@ -2,6 +2,7 @@
 members = [
     ".",
     "crates/bssh-russh",
+    "crates/bssh-russh-sftp",
 ]
 
 [package]
@@ -23,7 +24,8 @@ tokio = { version = "1.51.1", features = ["full"] }
 # - Development: uses local path (crates/bssh-russh)
 # - Publishing: uses crates.io version (path ignored)
 russh = { package = "bssh-russh", version = "0.60.1", path = "crates/bssh-russh" }
-russh-sftp = "2.1.1"
+# Use our internal russh-sftp fork with a serde_bytes perf fix
+russh-sftp = { package = "bssh-russh-sftp", version = "2.1.1", path = "crates/bssh-russh-sftp" }
 clap = { version = "4.6.0", features = ["derive", "env"] }
 anyhow = "1.0.102"
 thiserror = "2.0.18"

crates/bssh-russh-sftp/Cargo.toml

@@ -0,0 +1,35 @@
+[package]
+name = "bssh-russh-sftp"
+version = "2.1.1"
+authors = ["Jeongkyu Shin <inureyes@gmail.com>"]
+description = "Temporary fork of russh-sftp with a serde_bytes performance fix for SFTP Write/Data packets"
+documentation = "https://docs.rs/bssh-russh-sftp"
+edition = "2021"
+homepage = "https://github.com/lablup/bssh"
+keywords = ["russh", "sftp", "ssh2", "server", "client"]
+license = "Apache-2.0"
+readme = "README.md"
+repository = "https://github.com/lablup/bssh"
+
+[dependencies]
+tokio = { version = "1", default-features = false, features = [
+    "io-util",
+    "rt",
+    "sync",
+    "time",
+    "macros",
+] }
+tokio-util = "0.7"
+serde = { version = "1.0", features = ["derive"] }
+serde_bytes = "0.11"
+bitflags = { version = "2.9", features = ["serde"] }
+async-trait = { version = "0.1", optional = true }
+
+thiserror = "2.0"
+chrono = "0.4"
+bytes = "1.10"
+log = "0.4"
+flurry = "0.5"
+
+[features]
+async-trait = ["dep:async-trait"]

crates/bssh-russh-sftp/create-patch.sh

@@ -0,0 +1,48 @@
+#!/bin/bash
+# create-patch.sh
+# Creates a patch file from the current bssh-russh-sftp changes compared to upstream russh-sftp.
+#
+# Usage: ./create-patch.sh
+
+set -e
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+BSSH_ROOT="$SCRIPT_DIR/../.."
+UPSTREAM_DIR="$BSSH_ROOT/references/russh-sftp/src"
+CURRENT_DIR="$SCRIPT_DIR/src"
+PATCH_DIR="$SCRIPT_DIR/patches"
+PATCH_FILE="$PATCH_DIR/sftp-serde-bytes-perf.patch"
+
+GREEN='\033[0;32m'
+YELLOW='\033[1;33m'
+NC='\033[0m'
+
+log_info() { echo -e "${GREEN}[INFO]${NC} $1"; }
+log_warn() { echo -e "${YELLOW}[WARN]${NC} $1"; }
+
+if [ ! -d "$UPSTREAM_DIR" ]; then
+    echo "Error: Upstream russh-sftp not found at $UPSTREAM_DIR"
+    echo "Please ensure references/russh-sftp exists with the upstream source."
+    exit 1
+fi
+
+mkdir -p "$PATCH_DIR"
+
+log_info "Creating patch from differences..."
+
+/usr/bin/diff -urN "$UPSTREAM_DIR" "$CURRENT_DIR" \
+    | sed "s|$UPSTREAM_DIR|a/src|g" \
+    | sed "s|$CURRENT_DIR|b/src|g" \
+    > "$PATCH_FILE" || true
+
+if [ -s "$PATCH_FILE" ]; then
+    LINES=$(wc -l < "$PATCH_FILE" | tr -d ' ')
+    log_info "Patch created: $PATCH_FILE ($LINES lines)"
+
+    echo ""
+    echo "Patch summary:"
+    echo "=============="
+    grep -E "^@@|^\+\+\+|^---" "$PATCH_FILE" | head -20
+else
+    log_warn "No differences found - patch file is empty"
+fi

crates/bssh-russh-sftp/patches/sftp-serde-bytes-perf.patch

@@ -0,0 +1,62 @@
+diff -urN a/src/lib.rs src/lib.rs
+--- a/src/lib.rs	2026-04-21 17:00:59
++++ b/src/lib.rs	2026-04-21 17:05:30
+@@ -1,3 +1,6 @@
++// Lints tripped by vendored upstream source that we do not want to diverge from.
++#![allow(clippy::io_other_error)]
++
+ //! SFTP subsystem with client and server support for Russh and more!
+ //!
+ //! Crate can provide compatibility with anything that can provide the raw data
+diff -urN a/src/protocol/data.rs src/protocol/data.rs
+--- a/src/protocol/data.rs	2026-04-21 17:00:59
++++ b/src/protocol/data.rs	2026-04-21 17:00:36
+@@ -4,6 +4,7 @@
+ #[derive(Debug, Serialize, Deserialize)]
+ pub struct Data {
+     pub id: u32,
++    #[serde(with = "serde_bytes")]
+     pub data: Vec<u8>,
+ }
+
+diff -urN a/src/protocol/write.rs src/protocol/write.rs
+--- a/src/protocol/write.rs	2026-04-21 17:00:59
++++ b/src/protocol/write.rs	2026-04-21 17:00:36
+@@ -6,6 +6,7 @@
+     pub id: u32,
+     pub handle: String,
+     pub offset: u64,
++    #[serde(with = "serde_bytes")]
+     pub data: Vec<u8>,
+ }
+
+diff -urN a/src/ser.rs src/ser.rs
+--- a/src/ser.rs	2026-04-21 17:00:59
++++ b/src/ser.rs	2026-04-21 17:00:36
+@@ -103,8 +103,10 @@
+         Ok(())
+     }
+
+-    fn serialize_bytes(self, _v: &[u8]) -> Result<Self::Ok, Self::Error> {
+-        Err(Error::BadMessage("bytes not supported".to_owned()))
++    fn serialize_bytes(self, v: &[u8]) -> Result<Self::Ok, Self::Error> {
++        self.output.put_u32(v.len() as u32);
++        self.output.put_slice(v);
++        Ok(())
+     }
+
+     fn serialize_none(self) -> Result<Self::Ok, Self::Error> {
+diff -urN a/src/utils.rs src/utils.rs
+--- a/src/utils.rs	2026-04-21 17:00:59
++++ b/src/utils.rs	2026-04-21 17:04:11
+@@ -9,9 +9,7 @@
+     DateTime::<Utc>::from(time).timestamp() as u32
+ }
+
+-pub async fn read_packet<S: AsyncRead + Unpin>(
+-    stream: &mut S,
+-) -> Result<Bytes, Error> {
++pub async fn read_packet<S: AsyncRead + Unpin>(stream: &mut S) -> Result<Bytes, Error> {
+     let length = stream.read_u32().await?;
+
+     let mut buf = vec![0; length as usize];

crates/bssh-russh-sftp/src/buf.rs

@@ -0,0 +1,27 @@
+use bytes::Buf;
+
+use crate::error::Error;
+
+pub trait TryBuf: Buf {
+    fn try_get_bytes(&mut self) -> Result<Vec<u8>, Error>;
+    fn try_get_string(&mut self) -> Result<String, Error>;
+}
+
+impl<T: Buf> TryBuf for T {
+    fn try_get_bytes(&mut self) -> Result<Vec<u8>, Error> {
+        let len = self
+            .try_get_u32()
+            .map_err(|e| Error::UnexpectedBehavior(e.to_string()))? as usize;
+        if self.remaining() < len {
+            return Err(Error::BadMessage("no remaining for vec".to_owned()));
+        }
+
+        Ok(self.copy_to_bytes(len).to_vec())
+    }
+
+    fn try_get_string(&mut self) -> Result<String, Error> {
+        let bytes = self.try_get_bytes()?;
+        //String::from_utf8(bytes).map_err(|_| Error::BadMessage("unable to parse str".to_owned()))
+        Ok(String::from_utf8_lossy(&bytes).into())
+    }
+}

crates/bssh-russh-sftp/src/client/error.rs

@@ -0,0 +1,67 @@
+use std::io;
+use thiserror::Error;
+use tokio::sync::mpsc::error::SendError as MpscSendError;
+use tokio::sync::oneshot::error::RecvError as OneshotRecvError;
+use tokio::time::error::Elapsed as TimeElapsed;
+
+use crate::error;
+use crate::protocol::Status;
+
+/// Enum for client errors
+#[derive(Debug, Clone, Error)]
+pub enum Error {
+    /// Contains an error status packet
+    #[error("{}: {}", .0.status_code, .0.error_message)]
+    Status(Status),
+    /// Any errors related to I/O
+    #[error("I/O: {0}")]
+    IO(String),
+    /// Time limit for receiving response packet exceeded
+    #[error("Timeout")]
+    Timeout,
+    /// Occurs due to exceeding the limits set by the `limits@openssh.com` extension
+    #[error("Limit exceeded: {0}")]
+    Limited(String),
+    /// Occurs when an unexpected packet is sent
+    #[error("Unexpected packet")]
+    UnexpectedPacket,
+    /// Occurs when unexpected server behavior differs from the protocol specifition
+    #[error("{0}")]
+    UnexpectedBehavior(String),
+}
+
+impl From<Status> for Error {
+    fn from(status: Status) -> Self {
+        Self::Status(status)
+    }
+}
+
+impl From<io::Error> for Error {
+    fn from(error: io::Error) -> Self {
+        Self::IO(error.to_string())
+    }
+}
+
+impl<T> From<MpscSendError<T>> for Error {
+    fn from(err: MpscSendError<T>) -> Self {
+        Self::UnexpectedBehavior(format!("SendError: {}", err))
+    }
+}
+
+impl From<OneshotRecvError> for Error {
+    fn from(err: OneshotRecvError) -> Self {
+        Self::UnexpectedBehavior(format!("RecvError: {}", err))
+    }
+}
+
+impl From<TimeElapsed> for Error {
+    fn from(_: TimeElapsed) -> Self {
+        Self::Timeout
+    }
+}
+
+impl From<error::Error> for Error {
+    fn from(error: error::Error) -> Self {
+        Self::UnexpectedBehavior(error.to_string())
+    }
+}