From 840fb646a3936a46de77519d29ee3f81b8e9de5a Mon Sep 17 00:00:00 2001 From: Jeongkyu Shin Date: Mon, 24 Aug 2026 14:22:15 +0900 Subject: [PATCH 1/2] update: sync bssh-russh-sftp with upstream russh-sftp 2.4.0 Brings in every upstream change released since the fork's 2.3.0 base. Upstream `src/` is identical between 2.4.0 and current master, so this covers all of it. The behavior fix is `FileAttributes::default()`: it used to return *dummy* attributes (every field `Some(0)`) while `empty()` returned omitted ones, which made `Default` the opposite of what a caller would assume. Upstream now derives `Default` as the omitted form, makes `empty()` an alias for it, and moves the zero-filled placeholder to an explicit `FileAttributes::dummy()`, which `protocol::File::new` now calls so its wire output is unchanged. bssh is unaffected: `src/server/sftp.rs` builds every `FileAttributes` and `protocol::File` with fully specified struct literals and never uses `Default`, verified by inspection and by an SFTP round trip against both servers. Also vendored: `File::close()` plus documentation that dropping a handle closes it without awaiting the reply, and client support for the `expand-path@openssh.com` extension (`ExpandPathExtension`, `features.expand_path`, `RawSftpSession::expand_path`, `SftpSession::expand_path`). Fixes a fork maintenance hazard found while doing the sync. `src/server/mod.rs` carried the request read-ahead intake queue and sequential-write coalescing from issue #227, roughly 750 changed lines, with no patch file. Since `sync-upstream.sh` deletes `src/**/*.rs` before copying upstream over it and then re-applies only `patches/*.patch`, the next sync would have silently deleted that work; `create-patch.sh` would not have noticed because it only ever diffed `client/fs/file.rs`. That change is now captured in `patches/server-readahead-write-coalescing.patch`, `create-patch.sh` regenerates a patch per entry in a `PATCH_TARGETS` list and warns about any other vendored file that drifts from upstream without an entry, and the README documents both the second patch and the deletion hazard. Both patch files were verified to apply cleanly onto a pristine upstream 2.4.0 checkout, and `create-patch.sh` reproduces both byte-identically. Validation: `cargo check --workspace --all-targets`, `cargo clippy --workspace --all-targets --all-features -- -D warnings`, and `cargo fmt --all -- --check` clean. Full workspace test suite green (bssh lib 1407 passed, bssh-russh-sftp 8 passed) except the three `integration_test` cases that CI already skips and that fail on a stale localhost entry in the developer's `~/.ssh/known_hosts`. End-to-end SFTP re-verified after the sync: round trip against bssh-server, and 300 KB plus 3 MB round trips against OpenSSH 10.3p1, all with matching SHA-256. Refs #227 --- Cargo.lock | 2 +- Cargo.toml | 7 +- crates/bssh-russh-sftp/Cargo.toml | 6 +- crates/bssh-russh-sftp/README.md | 24 +- crates/bssh-russh-sftp/create-patch.sh | 71 +- .../patches/pipelined-file-io.patch | 12 +- .../server-readahead-write-coalescing.patch | 873 ++++++++++++++++++ crates/bssh-russh-sftp/src/client/fs/file.rs | 15 +- .../bssh-russh-sftp/src/client/rawsession.rs | 16 +- crates/bssh-russh-sftp/src/client/session.rs | 18 +- crates/bssh-russh-sftp/src/extensions.rs | 8 + crates/bssh-russh-sftp/src/protocol/file.rs | 2 +- .../src/protocol/file_attrs.rs | 21 +- 13 files changed, 1012 insertions(+), 63 deletions(-) create mode 100644 crates/bssh-russh-sftp/patches/server-readahead-write-coalescing.patch diff --git a/Cargo.lock b/Cargo.lock index da49d369..bb020e87 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -451,7 +451,7 @@ dependencies = [ [[package]] name = "bssh-russh-sftp" -version = "2.3.0" +version = "2.4.0" dependencies = [ "async-trait", "bitflags 2.13.1", diff --git a/Cargo.toml b/Cargo.toml index 541f7f3e..c2ed407e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -34,9 +34,10 @@ tokio = { version = "1.52.3", features = ["full"] } # handler in src/ssh/tokio_client/connection.rs refuses certificates because # bssh verifies no CA signatures. russh = "0.63.1" -# Use our internal russh-sftp fork tracking upstream 2.3.0 -# (adds pipelined File I/O; serde_bytes perf fix is now upstreamed) -russh-sftp = { package = "bssh-russh-sftp", version = "2.3.0", path = "crates/bssh-russh-sftp" } +# Use our internal russh-sftp fork tracking upstream 2.4.0 +# (adds pipelined File I/O and the server read-ahead / write-coalescing loop; +# the serde_bytes perf fix is now upstreamed) +russh-sftp = { package = "bssh-russh-sftp", version = "2.4.0", path = "crates/bssh-russh-sftp" } clap = { version = "4.6.1", features = ["derive", "env"] } anyhow = "1.0.102" thiserror = "2.0.18" diff --git a/crates/bssh-russh-sftp/Cargo.toml b/crates/bssh-russh-sftp/Cargo.toml index 394df061..fb396ef7 100644 --- a/crates/bssh-russh-sftp/Cargo.toml +++ b/crates/bssh-russh-sftp/Cargo.toml @@ -1,8 +1,8 @@ [package] name = "bssh-russh-sftp" -version = "2.3.0" +version = "2.4.0" authors = ["Jeongkyu Shin "] -description = "Temporary fork of russh-sftp 2.3.0 adding pipelined SFTP File I/O (write_all_pipelined / read_to_writer_pipelined). These helpers hide per-request RTT for fast bulk transfers and are the only value-add over upstream russh-sftp." +description = "Temporary fork of russh-sftp 2.4.0 adding pipelined SFTP File I/O (write_all_pipelined / read_to_writer_pipelined). These helpers hide per-request RTT for fast bulk transfers and are the only value-add over upstream russh-sftp." documentation = "https://docs.rs/bssh-russh-sftp" edition = "2021" homepage = "https://github.com/lablup/bssh" @@ -11,7 +11,7 @@ license = "Apache-2.0" readme = "README.md" repository = "https://github.com/lablup/bssh" -# Dependency versions mirror upstream russh-sftp 2.3.0 (AspectUnk/russh-sftp). +# Dependency versions mirror upstream russh-sftp 2.4.0 (AspectUnk/russh-sftp). # Update via ./sync-upstream.sh; the only fork addition is the `futures` dep, # needed by the forward-ported pipelined helpers in src/client/fs/file.rs. [dependencies] diff --git a/crates/bssh-russh-sftp/README.md b/crates/bssh-russh-sftp/README.md index 2d97b713..e7f72269 100644 --- a/crates/bssh-russh-sftp/README.md +++ b/crates/bssh-russh-sftp/README.md @@ -1,17 +1,25 @@ # bssh-russh-sftp -**Temporary fork of [russh-sftp](https://crates.io/crates/russh-sftp) (tracking upstream `2.3.0`) adding pipelined SFTP file I/O.** +**Temporary fork of [russh-sftp](https://crates.io/crates/russh-sftp) (tracking upstream `2.4.0`) adding pipelined SFTP file I/O and a read-ahead server loop.** This crate exists so bssh can ship faster bulk SFTP transfers independently, while keeping the public crate name usable through Cargo's `package = "bssh-russh-sftp"` dependency alias. ## The Value-Add -The fork adds two helpers to `client::fs::File` that keep many SFTP requests in flight at once, hiding per-request round-trip latency (mirroring how OpenSSH's `sftp` client keeps ~64 requests outstanding): +### Client: pipelined file I/O (`src/client/fs/file.rs`) -- `File::write_all_pipelined(reader, max_inflight)` — streams a reader to the remote file with up to `max_inflight` concurrent `SSH_FXP_WRITE`s. -- `File::read_to_writer_pipelined(writer, max_inflight)` — streams the remote file to a writer with up to `max_inflight` concurrent `SSH_FXP_READ`s, reassembling chunks in offset order so the output matches a sequential read. +Two helpers on `client::fs::File` keep many SFTP requests in flight at once, hiding per-request round-trip latency (mirroring how OpenSSH's `sftp` client keeps ~64 requests outstanding): -These are the only additions over upstream. They live in `src/client/fs/file.rs` and are re-applied on each sync from `patches/pipelined-file-io.patch`. +- `File::write_all_pipelined(reader, max_inflight)` streams a reader to the remote file with up to `max_inflight` concurrent `SSH_FXP_WRITE`s. +- `File::read_to_writer_pipelined(writer, max_inflight)` streams the remote file to a writer with up to `max_inflight` concurrent `SSH_FXP_READ`s, reassembling chunks in offset order so the output matches a sequential read. + +Re-applied on sync from `patches/pipelined-file-io.patch`. + +### Server: request read-ahead and write coalescing (`src/server/mod.rs`) + +The serial request loop is replaced by a byte-bounded intake queue plus a processor, adding two `server::Config` knobs: `max_buffered_request_bytes` (default 8 MiB) and `max_write_coalesce_len` (default 256 KiB). Read-ahead keeps the transport decrypting requests while the handler is blocked on file I/O, and consecutive `SSH_FXP_WRITE`s to the same handle at sequential offsets are merged into one handler call while each request id still gets its own status reply. The unbounded-in-count, bounded-in-bytes intake is deliberate: stalling intake can deadlock against the russh session loop waiting on channel window (see issue lablup/bssh#227, paramiko's unbounded READ prefetch). + +Re-applied on sync from `patches/server-readahead-write-coalescing.patch`. > The `serde_bytes` packet-serialization performance fix that originally motivated this fork was upstreamed in russh-sftp 2.1.2; it is kept for reference under `patches/historical/`. @@ -19,18 +27,20 @@ These are the only additions over upstream. They live in `src/client/fs/file.rs` ```toml [dependencies] -russh-sftp = { package = "bssh-russh-sftp", version = "2.3.0" } +russh-sftp = { package = "bssh-russh-sftp", version = "2.4.0" } ``` ## Sync with Upstream ```bash cd crates/bssh-russh-sftp -./sync-upstream.sh 2.3.0 # omit the version to use upstream's default branch +./sync-upstream.sh 2.4.0 # omit the version to use upstream's default branch ``` `sync-upstream.sh` copies upstream `src` verbatim and re-applies every patch directly under `patches/` (anything under `patches/historical/` is excluded). Patches already merged upstream are detected via reverse-apply and skipped. +Because the sync deletes `src/**/*.rs` before copying upstream over it, **a fork change with no patch file is silently lost**. Regenerate the patches with `./create-patch.sh ` after editing vendored code; it diffs every file listed in its `PATCH_TARGETS` and warns about any other file that drifts from upstream without an entry. + ## License Apache-2.0 (same as russh-sftp) diff --git a/crates/bssh-russh-sftp/create-patch.sh b/crates/bssh-russh-sftp/create-patch.sh index 5df155ea..e29adaf6 100755 --- a/crates/bssh-russh-sftp/create-patch.sh +++ b/crates/bssh-russh-sftp/create-patch.sh @@ -1,13 +1,13 @@ #!/bin/bash # create-patch.sh -# Regenerates patches/pipelined-file-io.patch by diffing the current vendored -# source against a fresh checkout of upstream russh-sftp. +# Regenerates every file in patches/ by diffing the current vendored source +# against a fresh checkout of upstream russh-sftp. # # Self-contained: clones upstream into a temp dir (no manually-maintained # references/ directory needed), so it always diffs against the exact version. # # Usage: ./create-patch.sh [version] -# version: optional, e.g. "2.3.0" (default: upstream's default branch, since +# version: optional, e.g. "2.4.0" (default: upstream's default branch, since # russh-sftp does not publish git tags) set -e @@ -16,7 +16,6 @@ SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" UPSTREAM_URL="https://github.com/AspectUnk/russh-sftp.git" TEMP_DIR="/tmp/russh-sftp-createpatch-$$" PATCH_DIR="$SCRIPT_DIR/patches" -PATCH_FILE="$PATCH_DIR/pipelined-file-io.patch" GREEN='\033[0;32m' YELLOW='\033[1;33m' @@ -49,22 +48,50 @@ UPSTREAM_SRC="$TEMP_DIR/src" CURRENT_SRC="$SCRIPT_DIR/src" mkdir -p "$PATCH_DIR" -# The only fork change is the pipelined File I/O in client/fs/file.rs -# (write_all_pipelined / read_to_writer_pipelined). Emit a -p1 patch. -diff -u \ - --label a/src/client/fs/file.rs \ - --label b/src/client/fs/file.rs \ - "$UPSTREAM_SRC/client/fs/file.rs" \ - "$CURRENT_SRC/client/fs/file.rs" \ - > "$PATCH_FILE" || true +# Every fork change, one patch per file. Keep this list in sync with the fork: +# a file that drifts from upstream without an entry here is silently deleted by +# sync-upstream.sh, which wipes src/ before copying upstream over it. +# client/fs/file.rs - pipelined File I/O (write_all_pipelined / +# read_to_writer_pipelined) +# server/mod.rs - request read-ahead intake queue and sequential-write +# coalescing (issue lablup/bssh#227) +PATCH_TARGETS=( + "client/fs/file.rs:pipelined-file-io.patch" + "server/mod.rs:server-readahead-write-coalescing.patch" +) -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 +# Guard against exactly the failure this list exists to prevent: any src file +# that differs from upstream but has no patch entry. +UNTRACKED=0 +while IFS= read -r REL; do + for TARGET in "${PATCH_TARGETS[@]}"; do + [ "${TARGET%%:*}" = "$REL" ] && continue 2 + done + log_warn "src/$REL differs from upstream but has no PATCH_TARGETS entry; sync-upstream.sh would discard it" + UNTRACKED=1 +done < <(cd "$UPSTREAM_SRC" && find . -name '*.rs' -type f | sed 's|^\./||' | while read -r F; do + if [ ! -f "$CURRENT_SRC/$F" ] || ! diff -q "$UPSTREAM_SRC/$F" "$CURRENT_SRC/$F" > /dev/null 2>&1; then + echo "$F" + fi +done) + +for TARGET in "${PATCH_TARGETS[@]}"; do + REL="${TARGET%%:*}" + OUT="$PATCH_DIR/${TARGET##*:}" + + diff -u \ + --label "a/src/$REL" \ + --label "b/src/$REL" \ + "$UPSTREAM_SRC/$REL" \ + "$CURRENT_SRC/$REL" \ + > "$OUT" || true + + if [ -s "$OUT" ]; then + LINES=$(wc -l < "$OUT" | tr -d ' ') + log_info "Patch created: $OUT ($LINES lines)" + else + log_warn "No differences in src/$REL - $OUT is empty (already upstream?)" + fi +done + +[ "$UNTRACKED" -eq 0 ] || log_warn "One or more fork changes are untracked; add them to PATCH_TARGETS before syncing." diff --git a/crates/bssh-russh-sftp/patches/pipelined-file-io.patch b/crates/bssh-russh-sftp/patches/pipelined-file-io.patch index f263d299..a21057fd 100644 --- a/crates/bssh-russh-sftp/patches/pipelined-file-io.patch +++ b/crates/bssh-russh-sftp/patches/pipelined-file-io.patch @@ -1,10 +1,9 @@ --- a/src/client/fs/file.rs +++ b/src/client/fs/file.rs -@@ -91,6 +91,205 @@ - +@@ -96,6 +96,205 @@ self.session.fsync(self.handle.as_str()).await.map(|_| ()) } -+ + + /// Streams `reader` to this remote file with up to `max_inflight` concurrent + /// SFTP `WRITE` requests in flight. Each request carries up to the negotiated + /// `write_len` (or the per-handle packet ceiling when no limit is advertised). @@ -203,6 +202,7 @@ + self.pos = next_to_write; + Ok(total) + } - } - - fn check_write_result( ++ + /// Closes the file waiting for all pending writes and the close itself + /// to be confirmed by the remote party. + /// Equivalent to [`shutdown`](tokio::io::AsyncWriteExt::shutdown) diff --git a/crates/bssh-russh-sftp/patches/server-readahead-write-coalescing.patch b/crates/bssh-russh-sftp/patches/server-readahead-write-coalescing.patch new file mode 100644 index 00000000..3fac1c94 --- /dev/null +++ b/crates/bssh-russh-sftp/patches/server-readahead-write-coalescing.patch @@ -0,0 +1,873 @@ +--- a/src/server/mod.rs ++++ b/src/server/mod.rs +@@ -1,8 +1,12 @@ + mod handler; + mod reply; + ++use std::sync::atomic::{AtomicUsize, Ordering}; ++use std::sync::Arc; ++ + use bytes::Bytes; + use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt}; ++use tokio::sync::mpsc; + + pub use self::handler::Handler; + pub use self::reply::StatusReply; +@@ -35,12 +39,40 @@ + pub struct Config { + /// Maximum allowed size of SFTP packets sent by clients. Default: 256 KiB. + pub max_client_packet_len: u32, ++ ++ /// Maximum number of request bytes buffered ahead of the one currently ++ /// being processed. Read-ahead lets the transport keep delivering (and ++ /// decrypting) requests while the handler is blocked on file I/O, and it ++ /// feeds the sequential-write coalescer. ++ /// ++ /// The intake queue is unbounded in request count and bounded in bytes, ++ /// and the reader never stops draining the stream while under this ++ /// budget. This matters for deadlock avoidance: if request intake ever ++ /// stalls while the response side is waiting for channel window, the ++ /// russh session loop can block on delivering channel data and stop ++ /// processing the very `SSH_MSG_CHANNEL_WINDOW_ADJUST` that would free ++ /// the response side (issue lablup/bssh#227, paramiko's unbounded READ ++ /// prefetch). Well-behaved clients stay far under this budget: a READ ++ /// request is ~50 bytes, so the default admits hundreds of thousands of ++ /// outstanding reads. A client that exceeds it is flooding and its ++ /// session is terminated. Default: 8 MiB. ++ pub max_buffered_request_bytes: usize, ++ ++ /// Maximum number of bytes merged into a single coalesced `SSH_FXP_WRITE` ++ /// handler call. Consecutive queued WRITE requests targeting the same ++ /// handle at strictly sequential offsets are merged into one handler ++ /// invocation (one seek + one write instead of one per request), and every ++ /// merged request id still receives its own status reply. Set to 0 to ++ /// disable coalescing. Default: 256 KiB. ++ pub max_write_coalesce_len: usize, + } + + impl Default for Config { + fn default() -> Self { + Self { + max_client_packet_len: 262144, ++ max_buffered_request_bytes: 8 * 1024 * 1024, ++ max_write_coalesce_len: 262144, + } + } + } +@@ -76,23 +108,227 @@ + } + } + +-async fn process_handler(stream: &mut S, handler: &mut H, cfg: &Config) -> Result<(), Error> ++/// A client packet after framing and decoding, as seen by the processor loop. ++enum Queued { ++ /// A well-formed request. ++ Request(Packet), ++ /// A frame that could not be decoded; answered with `SSH_FXP_STATUS` ++ /// `BadMessage` (id 0), matching the previous serial-loop behavior. ++ Malformed, ++ /// The reader failed to obtain a frame (I/O error or EOF). ++ ReadError(Error), ++} ++ ++fn decode(item: Result) -> Queued { ++ match item { ++ Ok(mut bytes) => match Packet::try_from(&mut bytes) { ++ Ok(packet) => Queued::Request(packet), ++ Err(_) => Queued::Malformed, ++ }, ++ Err(err) => Queued::ReadError(err), ++ } ++} ++ ++/// Encode and send one response packet without flushing. Flushing is deferred ++/// to the moment the request queue runs empty so a burst of pipelined ++/// requests is answered with one flush instead of one per request. ++async fn send_response(writer: &mut W, response: Packet) -> Result<(), Error> + where ++ W: AsyncWrite + Unpin, ++{ ++ let bytes = Bytes::try_from(response)?; ++ writer.write_all(&bytes).await?; ++ Ok(()) ++} ++ ++/// Drive one SFTP session over `stream` until EOF. ++/// ++/// Architecture: a reader task frames client packets and feeds a queue that ++/// is unbounded in request count and bounded in bytes ++/// (`Config::max_buffered_request_bytes`), so the transport keeps delivering ++/// requests while the handler is busy with file I/O. The reader never ++/// applies per-request backpressure: stalling intake while a response write ++/// waits for channel window lets the russh session loop block on channel ++/// data delivery, which stops WINDOW_ADJUST processing and deadlocks the ++/// session (issue lablup/bssh#227). The processor loop consumes the queue ++/// strictly in order: requests are handled one at a time against ++/// `&mut handler` and responses are written in request order, so response ++/// ordering and error semantics are identical to the previous serial ++/// read -> process -> write -> flush loop. Two optimizations apply on top: ++/// ++/// - **Deferred flush**: responses are flushed only when the queue is ++/// momentarily empty (or the session ends) instead of after every request. ++/// - **Sequential write coalescing**: consecutive queued `SSH_FXP_WRITE` ++/// requests for the same handle at strictly contiguous offsets are merged ++/// into a single handler call (bounded by ++/// `Config::max_write_coalesce_len`). Every merged request id receives its ++/// own status reply carrying the outcome of the merged write; on failure ++/// all merged ids receive the same error, which is the conservative ++/// superset of what a partially-failed serial sequence would report. ++async fn process_stream(stream: S, handler: &mut H, cfg: &Config) ++where ++ S: AsyncRead + AsyncWrite + Unpin + Send + 'static, + H: Handler + Send, +- S: AsyncRead + AsyncWrite + Unpin, + { +- let mut bytes = read_packet(stream, cfg.max_client_packet_len).await?; ++ let (mut read_half, mut write_half) = tokio::io::split(stream); + +- let response = match Packet::try_from(&mut bytes) { +- Ok(request) => process_request(request, handler).await, +- Err(_) => Packet::error(0, StatusCode::BadMessage), ++ let (tx, mut rx) = mpsc::unbounded_channel::>(); ++ let queued_bytes = Arc::new(AtomicUsize::new(0)); ++ let max_packet_len = cfg.max_client_packet_len; ++ let budget = cfg.max_buffered_request_bytes; ++ let reader_queued = Arc::clone(&queued_bytes); ++ let reader = tokio::spawn(async move { ++ loop { ++ let item = read_packet(&mut read_half, max_packet_len).await; ++ // Stop on EOF; keep reading after other errors to preserve the ++ // previous loop's behavior (it warned and retried). ++ let stop = matches!(item, Err(Error::UnexpectedEof)); ++ let len = item.as_ref().map_or(0, Bytes::len); ++ if reader_queued.fetch_add(len, Ordering::Relaxed) + len > budget { ++ // A client this far ahead of the processor is flooding, not ++ // pipelining; terminate instead of stalling intake (which ++ // could deadlock the whole session, see the doc above). ++ let _ = tx.send(Err(Error::UnexpectedBehavior(format!( ++ "request backlog exceeded {budget} buffered bytes" ++ )))); ++ break; ++ } ++ if tx.send(item).is_err() || stop { ++ break; ++ } ++ } ++ }); ++ ++ // Return a dequeued item's framed length to the byte budget. ++ let release = |item: &Result| { ++ if let Ok(bytes) = item { ++ queued_bytes.fetch_sub(bytes.len(), Ordering::Relaxed); ++ } + }; + +- let packet = Bytes::try_from(response)?; +- stream.write_all(&packet).await?; +- stream.flush().await?; ++ // Holds a packet dequeued by the coalescer that did not merge into the ++ // current write; it must be processed next to preserve ordering. ++ let mut pending: Option = None; + +- Ok(()) ++ 'session: loop { ++ let queued = match pending.take() { ++ Some(queued) => queued, ++ None => match rx.try_recv() { ++ Ok(item) => { ++ release(&item); ++ decode(item) ++ } ++ Err(mpsc::error::TryRecvError::Empty) => { ++ // No request ready: flush buffered responses before ++ // blocking so the client is never left waiting on ++ // replies we already produced. ++ if let Err(err) = write_half.flush().await { ++ warn!("sftp: flush failed: {err}"); ++ break 'session; ++ } ++ match rx.recv().await { ++ Some(item) => { ++ release(&item); ++ decode(item) ++ } ++ None => break 'session, ++ } ++ } ++ Err(mpsc::error::TryRecvError::Disconnected) => break 'session, ++ }, ++ }; ++ ++ match queued { ++ Queued::ReadError(Error::UnexpectedEof) => break 'session, ++ Queued::ReadError(err @ Error::UnexpectedBehavior(_)) => { ++ // Reader-side budget overflow: the session is being flooded. ++ warn!("sftp: terminating session: {err}"); ++ break 'session; ++ } ++ Queued::ReadError(err) => { ++ warn!("{}", err); ++ } ++ Queued::Malformed => { ++ if let Err(err) = ++ send_response(&mut write_half, Packet::error(0, StatusCode::BadMessage)).await ++ { ++ warn!("{}", err); ++ } ++ } ++ Queued::Request(Packet::Write(mut write)) => { ++ // Coalesce strictly sequential queued writes to the same ++ // handle into one handler call. ++ let mut ids = vec![write.id]; ++ while write.data.len() < cfg.max_write_coalesce_len { ++ let Ok(item) = rx.try_recv() else { ++ // Empty or disconnected: nothing more to merge now. ++ // A disconnect is surfaced by the next dequeue. ++ break; ++ }; ++ release(&item); ++ match decode(item) { ++ Queued::Request(Packet::Write(next)) ++ if next.handle == write.handle ++ && write.offset.checked_add(write.data.len() as u64) ++ == Some(next.offset) ++ && write.data.len() + next.data.len() ++ <= cfg.max_write_coalesce_len => ++ { ++ ids.push(next.id); ++ write.data.extend_from_slice(&next.data); ++ } ++ other => { ++ pending = Some(other); ++ break; ++ } ++ } ++ } ++ ++ let reply: StatusReply = match handler ++ .write(write.id, write.handle, write.offset, write.data) ++ .await ++ { ++ Ok(status) => StatusReply { ++ status_code: status.status_code, ++ error_message: Some(status.error_message), ++ language_tag: Some(status.language_tag), ++ }, ++ Err(err) => err.into(), ++ }; ++ ++ for id in ids { ++ let response = Packet::Status(Status { ++ id, ++ status_code: reply.status_code, ++ error_message: reply ++ .error_message ++ .clone() ++ .unwrap_or_else(|| reply.status_code.to_string()), ++ language_tag: reply ++ .language_tag ++ .clone() ++ .unwrap_or_else(|| "en-US".to_string()), ++ }); ++ if let Err(err) = send_response(&mut write_half, response).await { ++ warn!("{}", err); ++ } ++ } ++ } ++ Queued::Request(request) => { ++ let response = process_request(request, handler).await; ++ if let Err(err) = send_response(&mut write_half, response).await { ++ warn!("{}", err); ++ } ++ } ++ } ++ } ++ ++ if let Err(err) = write_half.flush().await { ++ debug!("sftp: final flush failed: {err}"); ++ } ++ reader.abort(); ++ ++ debug!("sftp stream ended"); + } + + /// Run processing stream as SFTP +@@ -105,20 +341,570 @@ + } + + /// Run processing stream as SFTP with custom configuration +-pub async fn run_with_config(mut stream: S, mut handler: H, cfg: Config) ++pub async fn run_with_config(stream: S, mut handler: H, cfg: Config) + where + S: AsyncRead + AsyncWrite + Unpin + Send + 'static, + H: Handler + Send + 'static, + { + tokio::spawn(async move { +- loop { +- match process_handler(&mut stream, &mut handler, &cfg).await { +- Err(Error::UnexpectedEof) => break, +- Err(err) => warn!("{}", err), +- Ok(_) => (), ++ process_stream(stream, &mut handler, &cfg).await; ++ }); ++} ++ ++#[cfg(test)] ++mod tests { ++ use std::collections::HashMap; ++ use std::sync::{Arc, Mutex}; ++ use std::time::Duration; ++ ++ use bytes::{Buf, BytesMut}; ++ use tokio::io::AsyncReadExt; ++ ++ use super::*; ++ use crate::protocol::{Handle, OpenFlags, Write}; ++ ++ /// Call log entry for the mock handler's write method. ++ #[derive(Debug, Clone, PartialEq)] ++ struct WriteCall { ++ handle: String, ++ offset: u64, ++ len: usize, ++ } ++ ++ /// State shared between the test body and the mock handler. ++ #[derive(Debug, Default)] ++ struct Shared { ++ /// Sparse file image keyed by handle. ++ files: HashMap>, ++ /// Every write call the handler received, in order. ++ write_calls: Vec, ++ /// Offsets whose writes must fail with `StatusCode::Failure`. ++ fail_offsets: Vec, ++ } ++ ++ /// Mock handler. `open` sleeps 50 ms before replying, which gives the ++ /// reader task ample time to enqueue every already-sent request. Tests ++ /// exploit this: sending OPEN followed by a burst of WRITEs makes the ++ /// queue state during write processing deterministic, so coalescing ++ /// expectations can be exact instead of timing-tolerant. ++ #[derive(Debug, Clone, Default)] ++ struct MockHandler { ++ shared: Arc>, ++ } ++ ++ impl Handler for MockHandler { ++ type Error = StatusCode; ++ ++ fn unimplemented(&self) -> Self::Error { ++ StatusCode::OpUnsupported ++ } ++ ++ fn open( ++ &mut self, ++ id: u32, ++ filename: String, ++ _pflags: OpenFlags, ++ _attrs: crate::protocol::FileAttributes, ++ ) -> impl std::future::Future> + Send { ++ let shared = Arc::clone(&self.shared); ++ async move { ++ // Let the reader task queue all pipelined requests sent ++ // after this OPEN before the processor resumes. ++ tokio::time::sleep(Duration::from_millis(50)).await; ++ shared ++ .lock() ++ .unwrap() ++ .files ++ .insert(filename.clone(), Vec::new()); ++ Ok(Handle { ++ id, ++ handle: filename, ++ }) + } + } + +- debug!("sftp stream ended"); +- }); ++ fn write( ++ &mut self, ++ id: u32, ++ handle: String, ++ offset: u64, ++ data: Vec, ++ ) -> impl std::future::Future> + Send { ++ let shared = Arc::clone(&self.shared); ++ async move { ++ let mut shared = shared.lock().unwrap(); ++ if shared.fail_offsets.contains(&offset) { ++ return Err(StatusCode::Failure); ++ } ++ shared.write_calls.push(WriteCall { ++ handle: handle.clone(), ++ offset, ++ len: data.len(), ++ }); ++ let file = shared.files.entry(handle).or_default(); ++ let end = offset as usize + data.len(); ++ if file.len() < end { ++ file.resize(end, 0); ++ } ++ file[offset as usize..end].copy_from_slice(&data); ++ Ok(Status { ++ id, ++ status_code: StatusCode::Ok, ++ error_message: String::new(), ++ language_tag: "en".to_string(), ++ }) ++ } ++ } ++ ++ async fn close(&mut self, id: u32, _handle: String) -> Result { ++ Ok(Status { ++ id, ++ status_code: StatusCode::Ok, ++ error_message: String::new(), ++ language_tag: "en".to_string(), ++ }) ++ } ++ ++ async fn read( ++ &mut self, ++ id: u32, ++ _handle: String, ++ _offset: u64, ++ len: u32, ++ ) -> Result { ++ Ok(crate::protocol::Data { ++ id, ++ data: vec![0u8; len as usize], ++ }) ++ } ++ } ++ ++ fn encode(packet: Packet) -> Bytes { ++ Bytes::try_from(packet).expect("packet must encode") ++ } ++ ++ /// OPEN request used as a queue-priming barrier (see [`MockHandler`]). ++ fn open_packet(id: u32, filename: &str) -> Bytes { ++ encode(Packet::Open(crate::protocol::Open { ++ id, ++ filename: filename.to_string(), ++ pflags: OpenFlags::WRITE, ++ attrs: crate::protocol::FileAttributes::default(), ++ })) ++ } ++ ++ fn write_packet(id: u32, handle: &str, offset: u64, data: Vec) -> Bytes { ++ encode(Packet::Write(Write { ++ id, ++ handle: handle.to_string(), ++ offset, ++ data, ++ })) ++ } ++ ++ fn read_request_packet(id: u32, handle: &str, offset: u64, len: u32) -> Bytes { ++ encode(Packet::Read(crate::protocol::Read { ++ id, ++ handle: handle.to_string(), ++ offset, ++ len, ++ })) ++ } ++ ++ /// Feed `requests` into a session and collect one decoded response per ++ /// request. ++ async fn run_session( ++ requests: Vec, ++ expected_responses: usize, ++ shared: Arc>, ++ cfg: Config, ++ ) -> Vec { ++ let (client, server) = tokio::io::duplex(1 << 20); ++ ++ let session = tokio::spawn(async move { ++ let mut handler = MockHandler { shared }; ++ process_stream(server, &mut handler, &cfg).await; ++ }); ++ ++ let (mut client_rd, mut client_wr) = tokio::io::split(client); ++ for request in &requests { ++ client_wr.write_all(request).await.unwrap(); ++ } ++ client_wr.shutdown().await.unwrap(); ++ ++ let mut responses = Vec::new(); ++ let mut buf = BytesMut::new(); ++ while responses.len() < expected_responses { ++ let mut chunk = [0u8; 4096]; ++ let n = tokio::time::timeout(Duration::from_secs(5), client_rd.read(&mut chunk)) ++ .await ++ .expect("timed out waiting for responses") ++ .expect("read must succeed"); ++ assert!(n > 0, "stream closed before all responses arrived"); ++ buf.extend_from_slice(&chunk[..n]); ++ ++ loop { ++ if buf.len() < 4 { ++ break; ++ } ++ let length = u32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]) as usize; ++ if buf.len() < 4 + length { ++ break; ++ } ++ buf.advance(4); ++ let mut frame = buf.split_to(length).freeze(); ++ responses.push(Packet::try_from(&mut frame).expect("response must decode")); ++ } ++ } ++ ++ session.await.unwrap(); ++ responses ++ } ++ ++ fn status_of(packet: &Packet) -> (u32, StatusCode) { ++ match packet { ++ Packet::Status(status) => (status.id, status.status_code), ++ other => panic!("expected status packet, got {other:?}"), ++ } ++ } ++ ++ fn response_id(packet: &Packet) -> u32 { ++ match packet { ++ Packet::Status(status) => status.id, ++ Packet::Handle(handle) => handle.id, ++ other => panic!("unexpected response packet: {other:?}"), ++ } ++ } ++ ++ #[tokio::test] ++ async fn sequential_writes_coalesce_into_one_handler_call() { ++ let shared = Arc::new(Mutex::new(Shared::default())); ++ let payload: Vec = (0..128u32).flat_map(|i| i.to_be_bytes()).collect(); ++ let chunk = payload.len() / 4; ++ ++ // OPEN primes the queue: while its handler sleeps, the reader ++ // enqueues all four WRITEs, so they coalesce into one handler call. ++ let mut requests = vec![open_packet(10, "h")]; ++ for (index, part) in payload.chunks(chunk).enumerate() { ++ requests.push(write_packet( ++ index as u32 + 1, ++ "h", ++ (index * chunk) as u64, ++ part.to_vec(), ++ )); ++ } ++ requests.push(encode(Packet::Close(crate::protocol::Close { ++ id: 99, ++ handle: "h".to_string(), ++ }))); ++ ++ let responses = run_session(requests, 6, Arc::clone(&shared), Config::default()).await; ++ ++ // Every request id gets a reply, in request order. ++ let ids: Vec = responses.iter().map(response_id).collect(); ++ assert_eq!(ids, vec![10, 1, 2, 3, 4, 99]); ++ for response in &responses[1..] { ++ assert_eq!(status_of(response).1, StatusCode::Ok); ++ } ++ ++ let shared = shared.lock().unwrap(); ++ // Coalescing must not change the file image. ++ assert_eq!(shared.files.get("h"), Some(&payload)); ++ // All four queued sequential writes merge into a single call. ++ assert_eq!( ++ shared.write_calls, ++ vec![WriteCall { ++ handle: "h".into(), ++ offset: 0, ++ len: payload.len() ++ }] ++ ); ++ } ++ ++ #[tokio::test] ++ async fn non_contiguous_writes_are_not_merged() { ++ let shared = Arc::new(Mutex::new(Shared::default())); ++ // Two queued writes with a hole between them must stay two calls. ++ let requests = vec![ ++ open_packet(10, "h"), ++ write_packet(1, "h", 0, vec![0xAA; 16]), ++ write_packet(2, "h", 64, vec![0xBB; 16]), ++ ]; ++ ++ let responses = run_session(requests, 3, Arc::clone(&shared), Config::default()).await; ++ assert_eq!( ++ responses[1..].iter().map(status_of).collect::>(), ++ vec![(1, StatusCode::Ok), (2, StatusCode::Ok)] ++ ); ++ ++ let shared = shared.lock().unwrap(); ++ assert_eq!( ++ shared.write_calls, ++ vec![ ++ WriteCall { ++ handle: "h".into(), ++ offset: 0, ++ len: 16 ++ }, ++ WriteCall { ++ handle: "h".into(), ++ offset: 64, ++ len: 16 ++ }, ++ ] ++ ); ++ } ++ ++ #[tokio::test] ++ async fn different_handles_are_not_merged() { ++ let shared = Arc::new(Mutex::new(Shared::default())); ++ let requests = vec![ ++ open_packet(10, "a"), ++ write_packet(1, "a", 0, vec![0xAA; 16]), ++ write_packet(2, "b", 16, vec![0xBB; 16]), ++ ]; ++ ++ let responses = run_session(requests, 3, Arc::clone(&shared), Config::default()).await; ++ assert_eq!( ++ responses[1..].iter().map(status_of).collect::>(), ++ vec![(1, StatusCode::Ok), (2, StatusCode::Ok)] ++ ); ++ ++ let shared = shared.lock().unwrap(); ++ assert_eq!(shared.write_calls.len(), 2); ++ assert_eq!(shared.write_calls[0].handle, "a"); ++ assert_eq!(shared.write_calls[1].handle, "b"); ++ } ++ ++ #[tokio::test] ++ async fn coalesce_respects_byte_budget() { ++ let shared = Arc::new(Mutex::new(Shared::default())); ++ let cfg = Config { ++ max_write_coalesce_len: 32, ++ ..Config::default() ++ }; ++ // Three queued sequential 16-byte writes with a 32-byte budget merge ++ // as [32, 16]. ++ let requests = vec![ ++ open_packet(10, "h"), ++ write_packet(1, "h", 0, vec![1; 16]), ++ write_packet(2, "h", 16, vec![2; 16]), ++ write_packet(3, "h", 32, vec![3; 16]), ++ ]; ++ ++ let responses = run_session(requests, 4, Arc::clone(&shared), cfg).await; ++ for (index, response) in responses[1..].iter().enumerate() { ++ assert_eq!(status_of(response), (index as u32 + 1, StatusCode::Ok)); ++ } ++ ++ let shared = shared.lock().unwrap(); ++ assert_eq!( ++ shared.write_calls, ++ vec![ ++ WriteCall { ++ handle: "h".into(), ++ offset: 0, ++ len: 32 ++ }, ++ WriteCall { ++ handle: "h".into(), ++ offset: 32, ++ len: 16 ++ }, ++ ] ++ ); ++ let mut expected = vec![1u8; 16]; ++ expected.extend_from_slice(&[2; 16]); ++ expected.extend_from_slice(&[3; 16]); ++ assert_eq!(shared.files.get("h"), Some(&expected)); ++ } ++ ++ #[tokio::test] ++ async fn merged_write_failure_reports_error_to_every_merged_id() { ++ let shared = Arc::new(Mutex::new(Shared { ++ fail_offsets: vec![0], ++ ..Shared::default() ++ })); ++ let requests = vec![ ++ open_packet(10, "h"), ++ write_packet(1, "h", 0, vec![1; 16]), ++ write_packet(2, "h", 16, vec![2; 16]), ++ ]; ++ ++ let responses = run_session(requests, 3, Arc::clone(&shared), Config::default()).await; ++ for (index, response) in responses[1..].iter().enumerate() { ++ let (id, code) = status_of(response); ++ assert_eq!(id, index as u32 + 1); ++ assert_eq!( ++ code, ++ StatusCode::Failure, ++ "every merged id must observe the write failure" ++ ); ++ } ++ ++ let shared = shared.lock().unwrap(); ++ assert!(shared.write_calls.is_empty()); ++ } ++ ++ #[tokio::test] ++ async fn coalescing_disabled_with_zero_budget() { ++ let shared = Arc::new(Mutex::new(Shared::default())); ++ let cfg = Config { ++ max_write_coalesce_len: 0, ++ ..Config::default() ++ }; ++ let requests = vec![ ++ open_packet(10, "h"), ++ write_packet(1, "h", 0, vec![1; 16]), ++ write_packet(2, "h", 16, vec![2; 16]), ++ ]; ++ ++ let responses = run_session(requests, 3, Arc::clone(&shared), cfg).await; ++ assert_eq!(responses.len(), 3); ++ ++ let shared = shared.lock().unwrap(); ++ assert_eq!(shared.write_calls.len(), 2, "no merging with zero budget"); ++ } ++ ++ /// Regression test for the paramiko prefetch deadlock (lablup/bssh#227). ++ /// ++ /// The client sends a burst of READ requests and reads no responses until ++ /// every request has been written, over a transport with a tiny buffer so ++ /// the server's response writes block almost immediately. The old ++ /// count-bounded intake queue stopped draining the stream once full, ++ /// which left the client's send side blocked too: a mutual stall, and at ++ /// the russh layer the same coupling froze WINDOW_ADJUST processing. The ++ /// byte-budgeted intake must keep draining, letting the client finish ++ /// sending and then collect every response. ++ #[tokio::test] ++ async fn pipelined_read_burst_drains_while_response_path_blocked() { ++ const READS: usize = 300; ++ const READ_LEN: u32 = 1024; ++ ++ let shared = Arc::new(Mutex::new(Shared::default())); ++ let cfg = Config::default(); ++ // Tiny per-direction buffer: response writes block after ~4 frames, ++ // and the request burst does not fit in the transport either. ++ let (client, server) = tokio::io::duplex(4096); ++ ++ let session = tokio::spawn(async move { ++ let mut handler = MockHandler { shared }; ++ process_stream(server, &mut handler, &cfg).await; ++ }); ++ ++ let (mut client_rd, mut client_wr) = tokio::io::split(client); ++ ++ let send_all = async move { ++ client_wr.write_all(&open_packet(0, "h")).await.unwrap(); ++ for i in 0..READS { ++ client_wr ++ .write_all(&read_request_packet( ++ i as u32 + 1, ++ "h", ++ u64::from(READ_LEN) * i as u64, ++ READ_LEN, ++ )) ++ .await ++ .unwrap(); ++ } ++ client_wr.shutdown().await.unwrap(); ++ }; ++ let send_all = tokio::time::timeout(Duration::from_secs(10), send_all); ++ ++ let recv_all = async move { ++ let mut responses = 0usize; ++ let mut buf = BytesMut::new(); ++ loop { ++ let mut chunk = [0u8; 4096]; ++ let n = client_rd.read(&mut chunk).await.unwrap(); ++ if n == 0 { ++ break; ++ } ++ buf.extend_from_slice(&chunk[..n]); ++ loop { ++ if buf.len() < 4 { ++ break; ++ } ++ let length = u32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]) as usize; ++ if buf.len() < 4 + length { ++ break; ++ } ++ buf.advance(4); ++ let mut frame = buf.split_to(length).freeze(); ++ Packet::try_from(&mut frame).expect("response must decode"); ++ responses += 1; ++ } ++ } ++ responses ++ }; ++ let recv_all = tokio::time::timeout(Duration::from_secs(10), recv_all); ++ ++ // The send side must complete even though nothing reads responses ++ // concurrently; only then does the receive side start. ++ send_all.await.expect("request burst must not deadlock"); ++ let responses = recv_all.await.expect("responses must not deadlock"); ++ assert_eq!(responses, READS + 1, "one HANDLE plus one DATA per READ"); ++ ++ session.await.unwrap(); ++ } ++ ++ /// A client that floods requests far beyond `max_buffered_request_bytes` ++ /// is terminated instead of being allowed to grow the queue without ++ /// bound (and instead of stalling intake, which is the deadlock shape ++ /// covered by the test above). ++ #[tokio::test] ++ async fn request_flood_beyond_budget_terminates_session() { ++ const READS: usize = 100; ++ ++ let shared = Arc::new(Mutex::new(Shared::default())); ++ let cfg = Config { ++ // Far below the ~4 KiB the burst below queues while the OPEN ++ // handler sleeps. ++ max_buffered_request_bytes: 512, ++ ..Config::default() ++ }; ++ let (client, server) = tokio::io::duplex(1 << 20); ++ ++ let session = tokio::spawn(async move { ++ let mut handler = MockHandler { shared }; ++ process_stream(server, &mut handler, &cfg).await; ++ }); ++ ++ let (mut client_rd, mut client_wr) = tokio::io::split(client); ++ // OPEN's 50 ms handler sleep keeps the processor busy while the ++ // reader accounts the whole burst against the byte budget. ++ client_wr.write_all(&open_packet(0, "h")).await.unwrap(); ++ for i in 0..READS { ++ client_wr ++ .write_all(&read_request_packet(i as u32 + 1, "h", 0, 64)) ++ .await ++ .unwrap(); ++ } ++ ++ // The server must close the stream (EOF) rather than answer the ++ // whole flood or hang. ++ let drain = async move { ++ let mut total = 0usize; ++ let mut chunk = [0u8; 4096]; ++ loop { ++ let n = client_rd.read(&mut chunk).await.unwrap(); ++ if n == 0 { ++ break; ++ } ++ total += n; ++ } ++ total ++ }; ++ let answered_bytes = tokio::time::timeout(Duration::from_secs(10), drain) ++ .await ++ .expect("flooded session must terminate, not hang"); ++ // A full flood's worth of DATA replies would be ~100 * 64 bytes of ++ // payload plus framing; termination must cut this short. ++ assert!( ++ answered_bytes < READS * 64, ++ "expected early termination, got {answered_bytes} response bytes" ++ ); ++ ++ session.await.unwrap(); ++ } + } diff --git a/crates/bssh-russh-sftp/src/client/fs/file.rs b/crates/bssh-russh-sftp/src/client/fs/file.rs index cac51598..7a42a555 100644 --- a/crates/bssh-russh-sftp/src/client/fs/file.rs +++ b/crates/bssh-russh-sftp/src/client/fs/file.rs @@ -7,7 +7,7 @@ use std::{ task::{ready, Context, Poll}, }; use tokio::{ - io::{AsyncRead, AsyncSeek, AsyncWrite, ReadBuf}, + io::{AsyncRead, AsyncSeek, AsyncWrite, AsyncWriteExt, ReadBuf}, sync::oneshot, }; @@ -34,9 +34,13 @@ struct FileState { /// Provides high-level methods for interaction with a remote file. /// -/// In order to properly close the handle, [`shutdown`] on a file should be called. +/// In order to properly close the handle, [`File::close`] or +/// [`shutdown`](tokio::io::AsyncWriteExt::shutdown) on a file should be called. /// Also implement [`AsyncSeek`] and other async i/o implementations. /// +/// On drop the handle is closed as well, but the reply is not awaited, so +/// pending write errors and the close status are silently discarded +/// /// # Weakness /// Using [`SeekFrom::End`] is costly and time-consuming because we need to /// request the actual file size from the remote server. @@ -290,6 +294,13 @@ impl File { self.pos = next_to_write; Ok(total) } + + /// Closes the file waiting for all pending writes and the close itself + /// to be confirmed by the remote party. + /// Equivalent to [`shutdown`](tokio::io::AsyncWriteExt::shutdown) + pub async fn close(mut self) -> io::Result<()> { + self.shutdown().await + } } fn check_write_result( diff --git a/crates/bssh-russh-sftp/src/client/rawsession.rs b/crates/bssh-russh-sftp/src/client/rawsession.rs index 1ea5a40a..6a5d31c9 100644 --- a/crates/bssh-russh-sftp/src/client/rawsession.rs +++ b/crates/bssh-russh-sftp/src/client/rawsession.rs @@ -17,7 +17,8 @@ use crate::{ client::{run, Config}, de, extensions::{ - self, FsyncExtension, HardlinkExtension, LimitsExtension, Statvfs, StatvfsExtension, + self, ExpandPathExtension, FsyncExtension, HardlinkExtension, LimitsExtension, Statvfs, + StatvfsExtension, }, protocol::{ Attrs, Close, Data, Extended, ExtendedReply, FSetStat, FileAttributes, Fstat, Handle, Init, @@ -740,6 +741,19 @@ impl RawSftpSession { _ => Err(Error::UnexpectedPacket), } } + + /// Expands `~`/`~user` and canonicalizes the path. + /// Replies in the same format as [`RawSftpSession::realpath`] + pub async fn expand_path>(&self, path: P) -> SftpResult { + let result = self + .extended( + extensions::EXPAND_PATH, + ExpandPathExtension { path: path.into() }.try_into()?, + ) + .await?; + + into_with_status!(result, Name) + } } impl Drop for RawSftpSession { diff --git a/crates/bssh-russh-sftp/src/client/session.rs b/crates/bssh-russh-sftp/src/client/session.rs index d393d6ec..d2172c2d 100644 --- a/crates/bssh-russh-sftp/src/client/session.rs +++ b/crates/bssh-russh-sftp/src/client/session.rs @@ -18,6 +18,7 @@ pub(crate) struct Features { pub hardlink: bool, pub fsync: bool, pub statvfs: bool, + pub expand_path: bool, pub limits: Option, pub max_concurrent_writes: usize, pub max_packet_len: u32, @@ -55,6 +56,7 @@ impl SftpSession { hardlink: has_extension(extensions::HARDLINK, "1"), fsync: has_extension(extensions::FSYNC, "1"), statvfs: has_extension(extensions::STATVFS, "2"), + expand_path: has_extension(extensions::EXPAND_PATH, "1"), limits: None, max_concurrent_writes, max_packet_len, @@ -265,7 +267,7 @@ impl SftpSession { } /// Performs a statvfs on the remote file system path. - /// Returns [`Ok(None)`] if the remote SFTP server does not support `statvfs@openssh.com` extension v2. + /// Returns `Ok(None)` if the remote SFTP server does not support `statvfs@openssh.com` extension v2. pub async fn fs_info>(&self, path: P) -> SftpResult> { if !self.features.statvfs { return Ok(None); @@ -273,4 +275,18 @@ impl SftpSession { self.session.statvfs(path).await.map(Some) } + + /// Expands a `~`/`~user`-prefixed or relative path and returns its canonicalized absolute form. + /// Returns `Ok(None)` if the remote SFTP server does not support `expand-path@openssh.com` extension v1. + pub async fn expand_path>(&self, path: P) -> SftpResult> { + if !self.features.expand_path { + return Ok(None); + } + + let name = self.session.expand_path(path).await?; + match name.files.first() { + Some(file) => Ok(Some(file.filename.to_owned())), + None => Err(Error::UnexpectedBehavior("no file".to_owned())), + } + } } diff --git a/crates/bssh-russh-sftp/src/extensions.rs b/crates/bssh-russh-sftp/src/extensions.rs index 76db05c9..857f51ac 100644 --- a/crates/bssh-russh-sftp/src/extensions.rs +++ b/crates/bssh-russh-sftp/src/extensions.rs @@ -4,6 +4,7 @@ pub const LIMITS: &str = "limits@openssh.com"; pub const HARDLINK: &str = "hardlink@openssh.com"; pub const FSYNC: &str = "fsync@openssh.com"; pub const STATVFS: &str = "statvfs@openssh.com"; +pub const EXPAND_PATH: &str = "expand-path@openssh.com"; macro_rules! impl_try_into_bytes { ($struct:ty) => { @@ -47,6 +48,13 @@ pub struct StatvfsExtension { impl_try_into_bytes!(StatvfsExtension); +#[derive(Debug, Serialize, Deserialize)] +pub struct ExpandPathExtension { + pub path: String, +} + +impl_try_into_bytes!(ExpandPathExtension); + #[derive(Debug, Serialize, Deserialize)] pub struct Statvfs { /// The file system block size diff --git a/crates/bssh-russh-sftp/src/protocol/file.rs b/crates/bssh-russh-sftp/src/protocol/file.rs index 5f2eccd5..88d49dc0 100644 --- a/crates/bssh-russh-sftp/src/protocol/file.rs +++ b/crates/bssh-russh-sftp/src/protocol/file.rs @@ -16,7 +16,7 @@ impl File { Self { filename: filename.into(), longname: "".to_string(), - attrs: FileAttributes::default(), + attrs: FileAttributes::dummy(), } } diff --git a/crates/bssh-russh-sftp/src/protocol/file_attrs.rs b/crates/bssh-russh-sftp/src/protocol/file_attrs.rs index 660ba3a4..6038ee9d 100644 --- a/crates/bssh-russh-sftp/src/protocol/file_attrs.rs +++ b/crates/bssh-russh-sftp/src/protocol/file_attrs.rs @@ -189,7 +189,7 @@ impl From for FilePermissions { /// clients that can be displayed in longname. Can be omitted. /// /// The `flags` field is omitted because it is set by itself depending on the fields -#[derive(Debug, Clone)] +#[derive(Debug, Default, Clone)] pub struct FileAttributes { pub size: Option, pub uid: Option, @@ -279,24 +279,13 @@ impl FileAttributes { } } - /// Creates a structure with omitted attributes + /// Creates a structure with omitted attributes. Same as [`Default`] pub fn empty() -> Self { - Self { - size: None, - uid: None, - user: None, - gid: None, - group: None, - permissions: None, - atime: None, - mtime: None, - } + Self::default() } -} -/// For packets which require dummy attributes -impl Default for FileAttributes { - fn default() -> Self { + /// For packets which require dummy attributes + pub fn dummy() -> Self { Self { size: Some(0), uid: Some(0), From b87089dea375461bde80d55e08ec67d27c203986 Mon Sep 17 00:00:00 2001 From: Jeongkyu Shin Date: Mon, 24 Aug 2026 14:34:00 +0900 Subject: [PATCH 2/2] fix: stop the bssh-russh-sftp sync scripts from wiping fork changes Running `./sync-upstream.sh 2.4.0` for real, instead of trusting it, surfaced two defects that between them made the documented sync workflow destructive. Version resolution never worked. russh-sftp publishes no git tags at all, so `git checkout v2.4.0 || git checkout 2.4.0` failed both ways and `set -e` aborted the script; the command in the README could not have run. Upstream marks releases with a `bump to ` commit, so both scripts now resolve a version argument to that commit. An unresolvable version is a hard error that lists the available release commits rather than a silent fall back to the default branch, which would vendor unreleased code while stamping Cargo.toml with the requested version. Resolution runs before anything is copied, so a bad version leaves the tree untouched. Patch-state detection was inverted-safe in name only. Apple's bundled `patch` (2.0-12u11) silently auto-corrects direction: with no tty it answers "yes" to `Unreversed (or previously applied) patch detected! Ignore -R?` and exits 0 for a forward patch, a reverse patch, an applied patch and an unapplied one alike. The obsolete-detection probe was `patch -p1 -R --dry-run`, so it succeeded unconditionally and every fork patch was classified "already present in upstream" and skipped. The observable result: a sync wiped both fork changes and the fork's test count silently dropped from 8 to 0, because the tests live inside the patched files. Detection now uses `git apply --check`, which never prompts and distinguishes the three real states (applies, already present, conflicts). The post-sync guard added alongside the 2.4.0 sync was subject to the same flaw, since it skipped anything the broken probe had marked obsolete and used the same `patch -R` probe itself. It now reverse-checks with `git apply` too, and the script additionally runs the fork tests, so a lost change fails the sync instead of passing quietly. Verified end to end rather than by inspection: `./sync-upstream.sh 2.4.0` now applies both patches, reports both present, passes 8 fork tests, and leaves `git diff` on the vendored tree completely empty, so the script reproduces the committed source byte for byte. `./create-patch.sh 2.4.0` resolves to the same commit and regenerates both patches byte-identically. Both scripts exit 1 on an unresolvable version without modifying the tree. Refs #227 --- crates/bssh-russh-sftp/README.md | 6 +- crates/bssh-russh-sftp/create-patch.sh | 26 +++++-- crates/bssh-russh-sftp/sync-upstream.sh | 92 +++++++++++++++++++------ 3 files changed, 98 insertions(+), 26 deletions(-) diff --git a/crates/bssh-russh-sftp/README.md b/crates/bssh-russh-sftp/README.md index e7f72269..95496b5a 100644 --- a/crates/bssh-russh-sftp/README.md +++ b/crates/bssh-russh-sftp/README.md @@ -37,7 +37,11 @@ cd crates/bssh-russh-sftp ./sync-upstream.sh 2.4.0 # omit the version to use upstream's default branch ``` -`sync-upstream.sh` copies upstream `src` verbatim and re-applies every patch directly under `patches/` (anything under `patches/historical/` is excluded). Patches already merged upstream are detected via reverse-apply and skipped. +`sync-upstream.sh` copies upstream `src` verbatim and re-applies every patch directly under `patches/` (anything under `patches/historical/` is excluded), then verifies each patch is present in the result, builds, and runs the fork tests. + +Upstream publishes **no git tags**, and marks releases with a `bump to ` commit instead, so both scripts resolve a version argument to that commit. An unresolvable version is a hard error listing the available release commits: falling back to the default branch would vendor unreleased code while stamping `Cargo.toml` with the requested version. Resolution happens before anything is copied, so a bad version leaves the tree untouched. + +Patch state is detected with `git apply --check`, not `patch --dry-run`. Apple's bundled `patch` silently auto-corrects direction and exits 0 whether a patch applies, is reversed, or is already applied, so its exit status cannot distinguish "already upstream" from "not applied yet". Because the sync deletes `src/**/*.rs` before copying upstream over it, **a fork change with no patch file is silently lost**. Regenerate the patches with `./create-patch.sh ` after editing vendored code; it diffs every file listed in its `PATCH_TARGETS` and warns about any other file that drifts from upstream without an entry. diff --git a/crates/bssh-russh-sftp/create-patch.sh b/crates/bssh-russh-sftp/create-patch.sh index e29adaf6..16159c61 100755 --- a/crates/bssh-russh-sftp/create-patch.sh +++ b/crates/bssh-russh-sftp/create-patch.sh @@ -17,11 +17,13 @@ UPSTREAM_URL="https://github.com/AspectUnk/russh-sftp.git" TEMP_DIR="/tmp/russh-sftp-createpatch-$$" PATCH_DIR="$SCRIPT_DIR/patches" +RED='\033[0;31m' 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"; } +log_error() { echo -e "${RED}[ERROR]${NC} $1" >&2; } cleanup() { [ -d "$TEMP_DIR" ] && rm -rf "$TEMP_DIR"; } trap cleanup EXIT @@ -33,14 +35,28 @@ git clone --quiet "$UPSTREAM_URL" "$TEMP_DIR" cd "$TEMP_DIR" if [ -z "$VERSION" ]; then - VERSION=$(git describe --tags --abbrev=0 2>/dev/null || echo "master") + VERSION="master" fi + +# Same resolution as sync-upstream.sh: russh-sftp publishes no git tags, and +# marks releases with a "bump to " commit. A patch must be generated +# against the exact base the vendored tree was synced from, so an unresolvable +# version is an error rather than a silent fall back to the default branch. if [ "$VERSION" != "master" ]; then - # russh-sftp publishes no git tags, so a version string may not be a ref. - if ! { git checkout --quiet "v$VERSION" 2>/dev/null || git checkout --quiet "$VERSION" 2>/dev/null; }; then - log_warn "No git ref '$VERSION' (russh-sftp publishes no tags); diffing against the default branch." - VERSION="master" + if git rev-parse --verify -q "v$VERSION^{commit}" > /dev/null; then + REF="v$VERSION" + elif git rev-parse --verify -q "$VERSION^{commit}" > /dev/null; then + REF="$VERSION" + else + REF=$(git log --format='%H' --grep="^bump to $VERSION\$" -1) + if [ -z "$REF" ]; then + log_error "Cannot resolve upstream version '$VERSION': no tag, no ref, and no 'bump to $VERSION' commit." + log_error "Available release commits:" + git log --oneline --grep='^bump to' | head -10 >&2 + exit 1 + fi fi + git checkout --quiet "$REF" fi log_info "Diffing against upstream $VERSION ($(git rev-parse --short HEAD))" diff --git a/crates/bssh-russh-sftp/sync-upstream.sh b/crates/bssh-russh-sftp/sync-upstream.sh index 565c1ee5..388ce5be 100755 --- a/crates/bssh-russh-sftp/sync-upstream.sh +++ b/crates/bssh-russh-sftp/sync-upstream.sh @@ -38,14 +38,33 @@ git clone "$UPSTREAM_URL" "$TEMP_DIR" cd "$TEMP_DIR" if [ -z "$VERSION" ]; then - VERSION=$(git describe --tags --abbrev=0 2>/dev/null || echo "master") - log_info "Using latest tag: $VERSION" -elif [ "$VERSION" != "master" ]; then + VERSION="master" + log_info "No version given; using upstream's default branch" +else log_info "Using specified version: $VERSION" fi +# russh-sftp publishes no git tags at all, so a version string is not a ref. +# Releases are marked by a "bump to " commit, which is the only +# reliable way to land on released code. Never fall back to the default branch +# for an explicit version: that would vendor unreleased commits while stamping +# Cargo.toml with the version the caller asked for. if [ "$VERSION" != "master" ]; then - git checkout "v$VERSION" 2>/dev/null || git checkout "$VERSION" + if git rev-parse --verify -q "v$VERSION^{commit}" > /dev/null; then + REF="v$VERSION" + elif git rev-parse --verify -q "$VERSION^{commit}" > /dev/null; then + REF="$VERSION" + else + REF=$(git log --format='%H' --grep="^bump to $VERSION\$" -1) + if [ -z "$REF" ]; then + log_error "Cannot resolve upstream version '$VERSION': no tag, no ref, and no 'bump to $VERSION' commit." + log_error "Available release commits:" + git log --oneline --grep='^bump to' | head -10 >&2 + exit 1 + fi + log_info "No tag for $VERSION (upstream publishes none); using its 'bump to' commit" + fi + git checkout --quiet "$REF" fi COMMIT_HASH=$(git rev-parse --short HEAD) @@ -76,8 +95,18 @@ fi # Apply every *.patch directly under patches/ (patches/historical/ is excluded: # those are forward-ports already merged upstream, kept only for reference). -# If a patch reverse-applies cleanly the change is already upstream, so we skip -# it and flag it as obsolete. +# +# Detection uses `git apply --check`, not `patch --dry-run`. Apple's bundled +# `patch` (2.0-12u11) silently auto-corrects direction: with no tty it answers +# "yes" to `Unreversed (or previously applied) patch detected! Ignore -R?` and +# exits 0 for a forward patch, a reverse patch, an applied patch and an +# unapplied one alike. Its exit code therefore carries no information, and the +# previous reverse-apply probe classified every fork patch as "already +# upstream" and skipped it, wiping the fork changes on every sync. +# `git apply --check` never prompts and returns a meaningful status: +# forward ok -> not applied yet, apply it +# forward no, reverse ok -> already present upstream, obsolete +# both no -> genuine conflict, stop log_info "Applying patches..." shopt -s nullglob @@ -93,24 +122,39 @@ OBSOLETE_PATCHES=() for PATCH_FILE in "${PATCH_FILES[@]}"; do PATCH_NAME=$(basename "$PATCH_FILE") - if patch -p1 -R --dry-run --silent < "$PATCH_FILE" > /dev/null 2>&1; then - log_info "Skipping $PATCH_NAME — already present in upstream (consider moving to patches/historical/)" + if git apply --check -p1 "$PATCH_FILE" > /dev/null 2>&1; then + git apply -p1 "$PATCH_FILE" + log_info "Applied $PATCH_NAME" + elif git apply --reverse --check -p1 "$PATCH_FILE" > /dev/null 2>&1; then + log_info "Skipping $PATCH_NAME: already present in upstream (consider moving to patches/historical/)" OBSOLETE_PATCHES+=("$PATCH_NAME") - continue + else + log_error "Failed to apply $PATCH_NAME: it neither applies nor is already present." + log_error "Upstream moved under the patch. Rebase it by hand, then regenerate with ./create-patch.sh" + log_error "Patch file: $PATCH_FILE" + git apply --check -p1 "$PATCH_FILE" || true + exit 1 fi +done - if patch -p1 --dry-run --silent < "$PATCH_FILE" > /dev/null 2>&1; then - patch -p1 --silent < "$PATCH_FILE" - log_info "Applied $PATCH_NAME" +# The sync wiped src/ before copying upstream over it, so every fork change +# must be back. Reverse-applying each non-obsolete patch proves its hunks are +# present in the vendored tree. A build check alone cannot catch a lost change: +# the fork's own tests live inside the patched files, so losing a patch loses +# its tests too and everything still compiles and passes. +log_info "Verifying fork changes survived the sync..." +for PATCH_FILE in "${PATCH_FILES[@]}"; do + PATCH_NAME=$(basename "$PATCH_FILE") + + for OBSOLETE in ${OBSOLETE_PATCHES[@]+"${OBSOLETE_PATCHES[@]}"}; do + [ "$OBSOLETE" = "$PATCH_NAME" ] && continue 2 + done + + if git apply --reverse --check -p1 "$PATCH_FILE" > /dev/null 2>&1; then + log_info "Present: $PATCH_NAME" else - log_warn "$PATCH_NAME may not apply cleanly, attempting with fuzz..." - if patch -p1 --fuzz=3 < "$PATCH_FILE"; then - log_warn "$PATCH_NAME applied with fuzz - please verify manually" - else - log_error "Failed to apply $PATCH_NAME. Manual intervention required." - log_error "Patch file: $PATCH_FILE" - exit 1 - fi + log_error "$PATCH_NAME is not present in the synced tree; the fork change was lost." + exit 1 fi done @@ -123,6 +167,14 @@ else exit 1 fi +log_info "Running fork tests..." +if cargo test -p bssh-russh-sftp --quiet; then + log_info "Fork tests passed" +else + log_error "Fork tests failed" + exit 1 +fi + log_info "Sync complete!" log_info "Upstream version: $VERSION ($COMMIT_HASH)" log_info ""