From fcadbb84b9776bb6e8fb7a3abdc4305893571c84 Mon Sep 17 00:00:00 2001 From: AptS-1547 Date: Tue, 7 Jul 2026 15:37:45 +0800 Subject: [PATCH 1/2] feat(sftp): add connection pool to SFTP storage driver - Introduce `SftpConnectionPool` with a semaphore-based size limit (default 4) and idle connection reuse via `Mutex>` - Add `SftpConnectionLease` RAII guard that returns connections to the pool on drop when marked reusable, discards on error - Add `is_sftp_connection_reusable_after_error` to distinguish recoverable SFTP status errors from connection-loss errors - Replace per-operation `connect()` calls with `acquire_connection()` across all `StorageDriver` and `StreamUploadDriver` methods - Expose `debug_connection_pool_snapshot()` under `#[cfg(debug_assertions)]` for integration test assertions - Switch integration test container from `atmoz/sftp` to `lscr.io/linuxserver/openssh-server` with env-var configuration - Add pool-aware integration test assertions verifying connection reuse, idle count, and stream lease isolation - Add unit tests for pool size lower bound, timeout constant values, and connection reusability classification - Update developer docs (en + zh-CN) to reference the new container image Closed #390 --- developer-docs/en/testing.md | 2 +- developer-docs/zh-CN/testing.md | 2 +- src/storage/drivers/sftp.rs | 330 +++++++++++++++++++++++++++----- tests/test_sftp.rs | 92 ++++++++- 4 files changed, 372 insertions(+), 54 deletions(-) diff --git a/developer-docs/en/testing.md b/developer-docs/en/testing.md index f41c35c20..3161fd7ca 100644 --- a/developer-docs/en/testing.md +++ b/developer-docs/en/testing.md @@ -131,7 +131,7 @@ The SFTP driver has a dedicated integration test: cargo test --test test_sftp ``` -This test starts an `atmoz/sftp` container through `testcontainers` by default and runs a real upload, download, range read, delete, and host-key fingerprint confirmation flow. It requires a local Docker / container runtime. +This test starts an `lscr.io/linuxserver/openssh-server` container through `testcontainers` by default and runs a real upload, download, range read, delete, and host-key fingerprint confirmation flow. It requires a local Docker / container runtime. If the current environment cannot run Docker, disable it explicitly: diff --git a/developer-docs/zh-CN/testing.md b/developer-docs/zh-CN/testing.md index 176756d7e..8e9b4e491 100644 --- a/developer-docs/zh-CN/testing.md +++ b/developer-docs/zh-CN/testing.md @@ -131,7 +131,7 @@ SFTP 驱动有单独的集成测试： cargo test --test test_sftp ``` -这个测试默认会通过 `testcontainers` 启动 `atmoz/sftp` 容器，完成一次真实上传、下载、range 读取、删除和主机密钥指纹确认流程。它需要本机 Docker / 容器运行时可用。 +这个测试默认会通过 `testcontainers` 启动 `lscr.io/linuxserver/openssh-server` 容器，完成一次真实上传、下载、range 读取、删除和主机密钥指纹确认流程。它需要本机 Docker / 容器运行时可用。 如果当前环境不能跑 Docker，可以显式关闭： diff --git a/src/storage/drivers/sftp.rs b/src/storage/drivers/sftp.rs index 2b7a18395..7cf19626c 100644 --- a/src/storage/drivers/sftp.rs +++ b/src/storage/drivers/sftp.rs @@ -7,10 +7,12 @@ use russh_sftp::client::{Config as SftpClientConfig, SftpSession, error::Error a use russh_sftp::protocol::StatusCode; use std::io::SeekFrom; use std::pin::Pin; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use std::task::{Context, Poll}; -use std::time::Duration; +use std::time::{Duration, Instant}; use tokio::io::{AsyncRead, AsyncReadExt, AsyncSeekExt, AsyncWriteExt, ReadBuf}; +use tokio::sync::{OwnedSemaphorePermit, Semaphore}; use crate::entities::storage_policy; use crate::errors::{AsterError, Result}; @@ -23,6 +25,10 @@ use crate::types::parse_storage_policy_options; const DEFAULT_SFTP_PORT: u16 = 22; const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); const IO_TIMEOUT: Duration = Duration::from_secs(30); +const SSH_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(10); +const POOL_ACQUIRE_TIMEOUT: Duration = Duration::from_secs(30); +const POOLED_CONNECTION_IDLE_TTL: Duration = Duration::from_secs(60); +const DEFAULT_POOL_SIZE: usize = 4; #[derive(Debug, Clone)] struct SftpEndpoint { @@ -30,13 +36,14 @@ struct SftpEndpoint { port: u16, } -#[derive(Debug, Clone)] +#[derive(Clone)] pub struct SftpDriver { endpoint: SftpEndpoint, username: String, password: String, base_path: String, host_key_fingerprint: Option, + pool: Arc, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -105,9 +112,36 @@ struct SftpConnection { sftp: SftpSession, } +struct IdleSftpConnection { + connection: SftpConnection, + returned_at: Instant, +} + +struct SftpConnectionPool { + semaphore: Arc, + idle: Mutex>, + max_idle: usize, + created_connections: AtomicUsize, +} + +struct SftpConnectionLease { + connection: Option, + pool: Arc, + _permit: OwnedSemaphorePermit, + reusable: bool, +} + struct SftpFileReader { - _connection: SftpConnection, file: russh_sftp::client::fs::File, + connection: SftpConnectionLease, +} + +#[cfg(debug_assertions)] +#[doc(hidden)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct SftpConnectionPoolSnapshot { + pub idle_connections: usize, + pub created_connections: usize, } impl AsyncRead for SftpFileReader { @@ -116,7 +150,131 @@ impl AsyncRead for SftpFileReader { cx: &mut Context<'_>, buf: &mut ReadBuf<'_>, ) -> Poll> { - Pin::new(&mut self.file).poll_read(cx, buf) + let result = Pin::new(&mut self.file).poll_read(cx, buf); + if matches!(result, Poll::Ready(Err(_))) { + self.connection.discard(); + } + result + } +} + +impl SftpConnectionPool { + fn new(max_size: usize) -> Self { + let max_size = max_size.max(1); + Self { + semaphore: Arc::new(Semaphore::new(max_size)), + idle: Mutex::new(Vec::with_capacity(max_size)), + max_idle: max_size, + created_connections: AtomicUsize::new(0), + } + } + + async fn acquire(self: &Arc, driver: &SftpDriver) -> Result { + let permit = timeout_io( + "acquire SFTP connection lease", + POOL_ACQUIRE_TIMEOUT, + self.semaphore.clone().acquire_owned(), + ) + .await? + .map_err(|error| { + storage_driver_error( + StorageErrorKind::Transient, + format!("acquire SFTP connection lease failed: {error}"), + ) + })?; + + let connection = if let Some(connection) = self.take_idle_connection() { + connection + } else { + let connection = driver.connect_new_connection().await?; + self.created_connections.fetch_add(1, Ordering::Relaxed); + connection + }; + + Ok(SftpConnectionLease { + connection: Some(connection), + pool: Arc::clone(self), + _permit: permit, + reusable: false, + }) + } + + fn take_idle_connection(&self) -> Option { + let mut idle = match self.idle.lock() { + Ok(guard) => guard, + Err(error) => { + tracing::warn!("failed to lock SFTP connection pool: {error}"); + return None; + } + }; + + while let Some(connection) = idle.pop() { + if connection.returned_at.elapsed() <= POOLED_CONNECTION_IDLE_TTL { + return Some(connection.connection); + } + } + None + } + + fn return_connection(&self, connection: SftpConnection) { + let mut idle = match self.idle.lock() { + Ok(guard) => guard, + Err(error) => { + tracing::warn!("failed to return SFTP connection to pool: {error}"); + return; + } + }; + + if idle.len() < self.max_idle { + idle.push(IdleSftpConnection { + connection, + returned_at: Instant::now(), + }); + } + } + + #[cfg(debug_assertions)] + fn snapshot(&self) -> SftpConnectionPoolSnapshot { + let idle_connections = self.idle.lock().map(|idle| idle.len()).unwrap_or(0); + SftpConnectionPoolSnapshot { + idle_connections, + created_connections: self.created_connections.load(Ordering::Relaxed), + } + } +} + +impl SftpConnectionLease { + fn sftp(&self) -> &SftpSession { + &self + .connection + .as_ref() + .expect("SFTP connection lease must hold a connection") + .sftp + } + + fn mark_reusable(&mut self) { + self.reusable = true; + } + + fn discard(&mut self) { + self.reusable = false; + } + + fn map_sftp_error(&mut self, context: &'static str, error: SftpError) -> AsterError { + if is_sftp_connection_reusable_after_error(&error) { + self.mark_reusable(); + } + map_sftp_error(context, error) + } +} + +impl Drop for SftpConnectionLease { + fn drop(&mut self) { + if self.reusable + && let Some(connection) = self.connection.take() + { + self.pool.return_connection(connection); + } } } @@ -158,6 +316,7 @@ impl SftpDriver { base_path: normalize_remote_base_path(&policy.base_path)?, host_key_fingerprint: parse_storage_policy_options(policy.options.as_ref()) .sftp_host_key_fingerprint, + pool: Arc::new(SftpConnectionPool::new(DEFAULT_POOL_SIZE)), }) } @@ -181,10 +340,20 @@ impl SftpDriver { }) } - async fn connect(&self) -> Result { + #[cfg(debug_assertions)] + #[doc(hidden)] + pub fn debug_connection_pool_snapshot(&self) -> SftpConnectionPoolSnapshot { + self.pool.snapshot() + } + + async fn acquire_connection(&self) -> Result { + self.pool.acquire(self).await + } + + async fn connect_new_connection(&self) -> Result { let mut config = russh::client::Config::default(); config.inactivity_timeout = Some(IO_TIMEOUT); - config.keepalive_interval = Some(Duration::from_secs(10)); + config.keepalive_interval = Some(SSH_KEEPALIVE_INTERVAL); config.nodelay = true; let address = (self.endpoint.host.clone(), self.endpoint.port); @@ -256,21 +425,19 @@ impl SftpDriver { async fn open_reader(&self, path: &str, offset: u64) -> Result { let remote_path = self.full_path(path)?; - let connection = self.connect().await?; + let mut connection = self.acquire_connection().await?; let mut file = connection - .sftp + .sftp() .open(remote_path) .await - .map_err(|error| map_sftp_error("SFTP open failed", error))?; + .map_err(|error| connection.map_sftp_error("SFTP open failed", error))?; if offset > 0 { file.seek(SeekFrom::Start(offset)) .await .map_err(|error| map_io_error("SFTP seek failed", error))?; } - Ok(SftpFileReader { - _connection: connection, - file, - }) + connection.mark_reusable(); + Ok(SftpFileReader { file, connection }) } } @@ -278,13 +445,13 @@ impl SftpDriver { impl StorageDriver for SftpDriver { async fn put(&self, path: &str, data: &[u8]) -> Result { let remote_path = self.full_path(path)?; - let connection = self.connect().await?; - ensure_remote_parent_dir(&connection.sftp, &remote_path).await?; + let mut connection = self.acquire_connection().await?; + ensure_remote_parent_dir(connection.sftp(), &remote_path).await?; let mut file = connection - .sftp + .sftp() .create(remote_path) .await - .map_err(|error| map_sftp_error("SFTP create failed", error))?; + .map_err(|error| connection.map_sftp_error("SFTP create failed", error))?; file.write_all(data) .await .map_err(|error| map_io_error("SFTP write failed", error))?; @@ -294,17 +461,20 @@ impl StorageDriver for SftpDriver { file.shutdown() .await .map_err(|error| map_io_error("SFTP close failed", error))?; + connection.mark_reusable(); Ok(path.to_string()) } async fn get(&self, path: &str) -> Result> { let remote_path = self.full_path(path)?; - let connection = self.connect().await?; - connection - .sftp + let mut connection = self.acquire_connection().await?; + let data = connection + .sftp() .read(remote_path) .await - .map_err(|error| map_sftp_error("SFTP read failed", error)) + .map_err(|error| connection.map_sftp_error("SFTP read failed", error))?; + connection.mark_reusable(); + Ok(data) } async fn get_stream(&self, path: &str) -> Result> { @@ -334,32 +504,41 @@ impl StorageDriver for SftpDriver { async fn delete(&self, path: &str) -> Result<()> { let remote_path = self.full_path(path)?; - let connection = self.connect().await?; + let mut connection = self.acquire_connection().await?; connection - .sftp + .sftp() .remove_file(remote_path) .await - .map_err(|error| map_sftp_error("SFTP delete failed", error)) + .map_err(|error| connection.map_sftp_error("SFTP delete failed", error))?; + connection.mark_reusable(); + Ok(()) } async fn exists(&self, path: &str) -> Result { let remote_path = self.full_path(path)?; - let connection = self.connect().await?; - match connection.sftp.metadata(remote_path).await { - Ok(_) => Ok(true), - Err(error) if is_sftp_not_found(&error) => Ok(false), - Err(error) => Err(map_sftp_error("SFTP stat failed", error)), + let mut connection = self.acquire_connection().await?; + match connection.sftp().metadata(remote_path).await { + Ok(_) => { + connection.mark_reusable(); + Ok(true) + } + Err(error) if is_sftp_not_found(&error) => { + connection.mark_reusable(); + Ok(false) + } + Err(error) => Err(connection.map_sftp_error("SFTP stat failed", error)), } } async fn metadata(&self, path: &str) -> Result { let remote_path = self.full_path(path)?; - let connection = self.connect().await?; + let mut connection = self.acquire_connection().await?; let stat = connection - .sftp + .sftp() .metadata(remote_path) .await - .map_err(|error| map_sftp_error("SFTP stat failed", error))?; + .map_err(|error| connection.map_sftp_error("SFTP stat failed", error))?; + connection.mark_reusable(); Ok(BlobMetadata { size: stat.size.unwrap_or(0), content_type: None, @@ -369,18 +548,18 @@ impl StorageDriver for SftpDriver { async fn copy_object(&self, src_path: &str, dest_path: &str) -> Result { let src_remote_path = self.full_path(src_path)?; let dest_remote_path = self.full_path(dest_path)?; - let connection = self.connect().await?; - ensure_remote_parent_dir(&connection.sftp, &dest_remote_path).await?; + let mut connection = self.acquire_connection().await?; + ensure_remote_parent_dir(connection.sftp(), &dest_remote_path).await?; let mut src = connection - .sftp + .sftp() .open(src_remote_path) .await - .map_err(|error| map_sftp_error("SFTP source open failed", error))?; + .map_err(|error| connection.map_sftp_error("SFTP source open failed", error))?; let mut dest = connection - .sftp + .sftp() .create(dest_remote_path) .await - .map_err(|error| map_sftp_error("SFTP destination create failed", error))?; + .map_err(|error| connection.map_sftp_error("SFTP destination create failed", error))?; tokio::io::copy(&mut src, &mut dest) .await .map_err(|error| map_io_error("SFTP copy failed", error))?; @@ -390,6 +569,7 @@ impl StorageDriver for SftpDriver { dest.shutdown() .await .map_err(|error| map_io_error("SFTP copy close failed", error))?; + connection.mark_reusable(); Ok(dest_path.to_string()) } @@ -407,13 +587,13 @@ impl StreamUploadDriver for SftpDriver { _size: i64, ) -> Result { let remote_path = self.full_path(storage_path)?; - let connection = self.connect().await?; - ensure_remote_parent_dir(&connection.sftp, &remote_path).await?; + let mut connection = self.acquire_connection().await?; + ensure_remote_parent_dir(connection.sftp(), &remote_path).await?; let mut remote_file = connection - .sftp + .sftp() .create(remote_path) .await - .map_err(|error| map_sftp_error("SFTP create failed", error))?; + .map_err(|error| connection.map_sftp_error("SFTP create failed", error))?; tokio::io::copy(&mut reader, &mut remote_file) .await .map_err(|error| map_io_error("SFTP stream upload failed", error))?; @@ -425,6 +605,7 @@ impl StreamUploadDriver for SftpDriver { .shutdown() .await .map_err(|error| map_io_error("SFTP stream close failed", error))?; + connection.mark_reusable(); Ok(storage_path.to_string()) } @@ -755,6 +936,15 @@ fn classify_sftp_error(error: &SftpError) -> StorageErrorKind { } } +fn is_sftp_connection_reusable_after_error(error: &SftpError) -> bool { + matches!( + error, + SftpError::Status(status) + if status.status_code != StatusCode::NoConnection + && status.status_code != StatusCode::ConnectionLost + ) +} + fn classify_io_error(error: &std::io::Error) -> StorageErrorKind { match error.kind() { std::io::ErrorKind::NotFound => StorageErrorKind::NotFound, @@ -807,9 +997,11 @@ fn is_sftp_not_found(error: &SftpError) -> bool { #[cfg(test)] mod tests { use super::{ - classify_sftp_error, host_key_fingerprint_matches, is_valid_host_key_fingerprint, - join_remote_path, normalize_host_key_fingerprint, normalize_remote_base_path, - parse_sftp_endpoint, sanitize_relative_storage_path, + CONNECT_TIMEOUT, DEFAULT_POOL_SIZE, IO_TIMEOUT, POOL_ACQUIRE_TIMEOUT, + POOLED_CONNECTION_IDLE_TTL, SSH_KEEPALIVE_INTERVAL, SftpConnectionPool, + classify_sftp_error, host_key_fingerprint_matches, is_sftp_connection_reusable_after_error, + is_valid_host_key_fingerprint, join_remote_path, normalize_host_key_fingerprint, + normalize_remote_base_path, parse_sftp_endpoint, sanitize_relative_storage_path, }; use crate::storage::error::StorageErrorKind; use crate::storage::{StorageDriver, StreamUploadDriver}; @@ -932,6 +1124,54 @@ mod tests { assert!(!is_valid_host_key_fingerprint("SHA256:abc def")); } + #[test] + fn sftp_pool_defaults_match_storage_timeout_boundaries() { + assert_eq!(DEFAULT_POOL_SIZE, 4); + assert_eq!(CONNECT_TIMEOUT, std::time::Duration::from_secs(10)); + assert_eq!(IO_TIMEOUT, std::time::Duration::from_secs(30)); + assert_eq!(SSH_KEEPALIVE_INTERVAL, std::time::Duration::from_secs(10)); + assert_eq!(POOL_ACQUIRE_TIMEOUT, std::time::Duration::from_secs(30)); + assert_eq!( + POOLED_CONNECTION_IDLE_TTL, + std::time::Duration::from_secs(60) + ); + } + + #[test] + fn sftp_pool_size_has_lower_bound() { + let pool = SftpConnectionPool::new(0); + assert_eq!(pool.max_idle, 1); + assert_eq!(pool.semaphore.available_permits(), 1); + } + + #[test] + fn sftp_status_errors_keep_connection_reusable_unless_connection_is_lost() { + let status = |status_code, error_message: &str| { + SftpError::Status(Status { + id: 1, + status_code, + error_message: error_message.to_string(), + language_tag: String::new(), + }) + }; + + assert!(is_sftp_connection_reusable_after_error(&status( + StatusCode::NoSuchFile, + "missing" + ))); + assert!(is_sftp_connection_reusable_after_error(&status( + StatusCode::PermissionDenied, + "denied" + ))); + assert!(!is_sftp_connection_reusable_after_error(&status( + StatusCode::ConnectionLost, + "lost" + ))); + assert!(!is_sftp_connection_reusable_after_error( + &SftpError::Timeout + )); + } + fn env_policy() -> Option { let endpoint = std::env::var("ASTER_SFTP_TEST_ENDPOINT").ok()?; let username = std::env::var("ASTER_SFTP_TEST_USERNAME").ok()?; diff --git a/tests/test_sftp.rs b/tests/test_sftp.rs index 040e075b1..090fc06b4 100644 --- a/tests/test_sftp.rs +++ b/tests/test_sftp.rs @@ -7,11 +7,12 @@ use aster_drive::storage::{StorageDriver, StorageErrorKind, StreamUploadDriver}; use testcontainers::{GenericImage, ImageExt, core::IntoContainerPort, runners::AsyncRunner}; use tokio::io::AsyncReadExt as _; -const SFTP_IMAGE: &str = "atmoz/sftp"; -const SFTP_TAG: &str = "alpine"; -const SFTP_PORT: u16 = 22; +const SFTP_IMAGE: &str = "lscr.io/linuxserver/openssh-server"; +const SFTP_TAG: &str = "latest"; +const SFTP_PORT: u16 = 2222; const SFTP_USERNAME: &str = "aster"; const SFTP_PASSWORD: &str = "asterpass"; +const SFTP_PROBE_TIMEOUT: Duration = Duration::from_secs(15); fn sftp_policy( endpoint: &str, @@ -68,7 +69,7 @@ async fn wait_for_sftp_host_key_fingerprint(driver: &SftpDriver) -> String { let mut last_error = None; let fingerprint = tokio::time::timeout(Duration::from_secs(45), async { loop { - match tokio::time::timeout(Duration::from_secs(5), driver.exists("readiness/probe.txt")) + match tokio::time::timeout(SFTP_PROBE_TIMEOUT, driver.exists("readiness/probe.txt")) .await { Ok(Ok(_)) => last_error = Some("untrusted host key was accepted".to_string()), @@ -101,7 +102,7 @@ async fn wait_for_sftp(driver: &SftpDriver) { let ready = tokio::time::timeout(Duration::from_secs(45), async { loop { match tokio::time::timeout( - Duration::from_secs(5), + SFTP_PROBE_TIMEOUT, driver.put("readiness/probe.txt", b"ready"), ) .await @@ -137,7 +138,13 @@ async fn test_sftp_driver_upload_download_round_trip() { let container = GenericImage::new(SFTP_IMAGE, SFTP_TAG) .with_exposed_port(IntoContainerPort::tcp(SFTP_PORT)) - .with_cmd(vec![format!("{SFTP_USERNAME}:{SFTP_PASSWORD}:::upload")]) + .with_env_var("PUID", "1000") + .with_env_var("PGID", "1000") + .with_env_var("TZ", "UTC") + .with_env_var("USER_NAME", SFTP_USERNAME) + .with_env_var("USER_PASSWORD", SFTP_PASSWORD) + .with_env_var("PASSWORD_ACCESS", "true") + .with_env_var("SUDO_ACCESS", "false") .start() .await .expect("failed to start sftp container"); @@ -147,7 +154,7 @@ async fn test_sftp_driver_upload_download_round_trip() { .await .expect("resolve mapped sftp port"); let endpoint = format!("sftp://127.0.0.1:{port}"); - let base_path = format!("/upload/asterdrive-itest-{}", uuid::Uuid::new_v4()); + let base_path = format!("asterdrive-itest-{}", uuid::Uuid::new_v4()); let untrusted_driver = SftpDriver::new(&sftp_policy(&endpoint, &base_path, None)).expect("create SftpDriver"); let host_key_fingerprint = wait_for_sftp_host_key_fingerprint(&untrusted_driver).await; @@ -164,10 +171,34 @@ async fn test_sftp_driver_upload_download_round_trip() { let data = b"hello sftp world"; driver.put("docs/hello.txt", data).await.unwrap(); + + #[cfg(debug_assertions)] + { + let baseline = driver.debug_connection_pool_snapshot(); + assert_eq!( + baseline.idle_connections, 1, + "successful sequential SFTP operation should return one reusable connection" + ); + assert!(driver.exists("docs/hello.txt").await.unwrap()); + assert_eq!(driver.get("docs/hello.txt").await.unwrap(), data); + assert_eq!( + driver.metadata("docs/hello.txt").await.unwrap().size, + u64::try_from(data.len()).unwrap() + ); + let after_sequential = driver.debug_connection_pool_snapshot(); + assert_eq!( + after_sequential.created_connections, baseline.created_connections, + "sequential SFTP operations should reuse the authenticated connection" + ); + assert_eq!(after_sequential.idle_connections, 1); + } + assert!(driver.exists("docs/hello.txt").await.unwrap()); assert!(!driver.exists("docs/missing.txt").await.unwrap()); assert_eq!(driver.get("docs/hello.txt").await.unwrap(), data); + #[cfg(debug_assertions)] + let before_missing_metadata = driver.debug_connection_pool_snapshot(); let missing_meta = driver .metadata("docs/missing.txt") .await @@ -176,6 +207,15 @@ async fn test_sftp_driver_upload_download_round_trip() { missing_meta.storage_error_kind(), Some(StorageErrorKind::NotFound) ); + #[cfg(debug_assertions)] + { + let after_missing_metadata = driver.debug_connection_pool_snapshot(); + assert_eq!( + after_missing_metadata.created_connections, before_missing_metadata.created_connections, + "not-found SFTP status should not force the pooled connection to reconnect" + ); + assert_eq!(after_missing_metadata.idle_connections, 1); + } let meta = driver.metadata("docs/hello.txt").await.unwrap(); assert_eq!(meta.size, u64::try_from(data.len()).unwrap()); @@ -184,6 +224,44 @@ async fn test_sftp_driver_upload_download_round_trip() { driver.put(unicode_path, b"encoded path").await.unwrap(); assert_eq!(driver.get(unicode_path).await.unwrap(), b"encoded path"); + #[cfg(debug_assertions)] + { + let before_stream = driver.debug_connection_pool_snapshot(); + assert_eq!(before_stream.idle_connections, 1); + let mut held_stream = driver.get_stream("docs/hello.txt").await.unwrap(); + let after_stream_open = driver.debug_connection_pool_snapshot(); + assert_eq!( + after_stream_open.created_connections, before_stream.created_connections, + "opening a stream should lease the existing idle connection" + ); + assert_eq!( + after_stream_open.idle_connections, 0, + "streaming reader must hold its connection until drop" + ); + + assert_eq!( + driver.metadata("docs/hello.txt").await.unwrap().size, + u64::try_from(data.len()).unwrap() + ); + let while_stream_held = driver.debug_connection_pool_snapshot(); + assert_eq!( + while_stream_held.created_connections, + before_stream.created_connections + 1, + "metadata while a stream is open should use another connection instead of sharing the stream lease" + ); + + let mut held_body = Vec::new(); + held_stream.read_to_end(&mut held_body).await.unwrap(); + assert_eq!(held_body, data); + drop(held_stream); + + let after_stream_drop = driver.debug_connection_pool_snapshot(); + assert_eq!( + after_stream_drop.idle_connections, 2, + "dropping the streaming reader should return its connection lease" + ); + } + let mut full_stream = driver.get_stream("docs/hello.txt").await.unwrap(); let mut full_body = Vec::new(); full_stream.read_to_end(&mut full_body).await.unwrap(); From 1d5a934104e3873bf5a2a969582ff9711298c747 Mon Sep 17 00:00:00 2001 From: AptS-1547 Date: Tue, 7 Jul 2026 16:21:22 +0800 Subject: [PATCH 2/2] fix(sftp): improve connection reuse safety by allowlisting safe status codes Change connection reuse logic from denylisting connection errors to allowlisting only known-safe status codes (NoSuchFile, PermissionDenied). This prevents potential connection corruption from unexpected error types. Changes: - Refactor `is_sftp_connection_reusable_after_error()` to use allowlist approach - Only reuse connections for NoSuchFile and PermissionDenied status codes - Add test coverage for BadMessage, Failure, OpUnsupported, and NoConnection codes - Rename test to reflect new allowlist-based behavior - Pin SFTP test container image to specific version (10.2_p1-r0-ls229) for stability --- src/storage/drivers/sftp.rs | 24 +++++++++++++++++++++--- tests/test_sftp.rs | 2 +- 2 files changed, 22 insertions(+), 4 deletions(-) diff --git a/src/storage/drivers/sftp.rs b/src/storage/drivers/sftp.rs index 7cf19626c..6b4eae423 100644 --- a/src/storage/drivers/sftp.rs +++ b/src/storage/drivers/sftp.rs @@ -940,8 +940,10 @@ fn is_sftp_connection_reusable_after_error(error: &SftpError) -> bool { matches!( error, SftpError::Status(status) - if status.status_code != StatusCode::NoConnection - && status.status_code != StatusCode::ConnectionLost + if matches!( + status.status_code, + StatusCode::NoSuchFile | StatusCode::PermissionDenied + ) ) } @@ -1145,7 +1147,7 @@ mod tests { } #[test] - fn sftp_status_errors_keep_connection_reusable_unless_connection_is_lost() { + fn sftp_status_errors_reuse_connection_only_for_known_safe_statuses() { let status = |status_code, error_message: &str| { SftpError::Status(Status { id: 1, @@ -1163,6 +1165,22 @@ mod tests { StatusCode::PermissionDenied, "denied" ))); + assert!(!is_sftp_connection_reusable_after_error(&status( + StatusCode::BadMessage, + "bad packet" + ))); + assert!(!is_sftp_connection_reusable_after_error(&status( + StatusCode::Failure, + "generic failure" + ))); + assert!(!is_sftp_connection_reusable_after_error(&status( + StatusCode::OpUnsupported, + "unsupported operation" + ))); + assert!(!is_sftp_connection_reusable_after_error(&status( + StatusCode::NoConnection, + "no connection" + ))); assert!(!is_sftp_connection_reusable_after_error(&status( StatusCode::ConnectionLost, "lost" diff --git a/tests/test_sftp.rs b/tests/test_sftp.rs index 090fc06b4..8bde92ef2 100644 --- a/tests/test_sftp.rs +++ b/tests/test_sftp.rs @@ -8,7 +8,7 @@ use testcontainers::{GenericImage, ImageExt, core::IntoContainerPort, runners::A use tokio::io::AsyncReadExt as _; const SFTP_IMAGE: &str = "lscr.io/linuxserver/openssh-server"; -const SFTP_TAG: &str = "latest"; +const SFTP_TAG: &str = "10.2_p1-r0-ls229"; const SFTP_PORT: u16 = 2222; const SFTP_USERNAME: &str = "aster"; const SFTP_PASSWORD: &str = "asterpass";