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..95496b5a 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,17 +27,23 @@ 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. +`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. ## License diff --git a/crates/bssh-russh-sftp/create-patch.sh b/crates/bssh-russh-sftp/create-patch.sh index 5df155ea..16159c61 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,13 +16,14 @@ 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" +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 @@ -34,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))" @@ -49,22 +64,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), 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 ""