Merge remote-tracking branch 'upstream/main' into test/heal-chaos-restart-recovery

# Conflicts:
#	.config/e2e-nightly-selection.txt

crates/e2e_test/src/cluster_multidrive_pool_test.rs

@@ -27,8 +27,10 @@
 //! Readiness is established by the harness's `start()` handshake (TCP reachability
 //! plus an S3 `ListBuckets` poll) — there are no fixed sleeps.
 //!
-//! Out of scope for this block (tracked separately): network fault injection
-//! (toxiproxy / socket proxy) and 5GiB large-object budgets.
+//! The volume-proxy smoke below also proves that the socket-level fault proxy
+//! can be installed before startup without changing the client-facing node URL.
+//! A full lock-plane partition matrix and 5GiB large-object budget remain
+//! tracked separately.
 
 use crate::common::{ClusterTopology, RustFSTestClusterEnvironment};
 
@@ -76,6 +78,28 @@ async fn cluster_multidrive_single_pool_smoke() -> TestResult {
     Ok(())
 }
 
+/// 4 nodes x 4 drives, single pool: exercise the maximum local erasure layout
+/// supported by the cluster harness. This remains in the nightly lane because
+/// it starts four real server processes and sixteen data directories.
+#[tokio::test]
+async fn cluster_four_node_four_drive_single_pool_smoke() -> TestResult {
+    crate::common::init_logging();
+
+    let mut cluster = RustFSTestClusterEnvironment::with_topology(ClusterTopology::single_pool_multidrive(4, 4)).await?;
+
+    let volumes = cluster.rustfs_volumes_arg();
+    assert_eq!(volumes.split(' ').count(), 16, "expected 16 explicit endpoints, got: {volumes}");
+    assert!(!volumes.contains('{'), "single-pool layout must not use ellipses: {volumes}");
+    assert!(cluster.nodes.iter().all(|node| node.data_dirs.len() == 4));
+
+    cluster.start().await?;
+    cluster.create_test_bucket(BUCKET).await?;
+
+    let payload = vec![0x3Cu8; 1024 * 1024];
+    put_get_roundtrip(&cluster, "multidrive-4/object", &payload).await?;
+    Ok(())
+}
+
 /// Two single-node pools, 2 drives each: the multi-pool layout boots and
 /// round-trips. Every pool is a distinct erasure pool (`pool_idx` 0 and 1).
 #[tokio::test]
@@ -103,3 +127,27 @@ async fn cluster_two_pool_smoke() -> TestResult {
     put_get_roundtrip(&cluster, "twopool/object", &payload).await?;
     Ok(())
 }
+
+/// A real cluster smoke for the volume FaultProxy wiring. The proxy target is
+/// not listening yet when it is created; cluster startup must still converge
+/// once the target node starts, and peer disk/RPC traffic must traverse it.
+#[tokio::test]
+async fn cluster_volume_fault_proxy_pass_smoke() -> TestResult {
+    crate::common::init_logging();
+
+    let mut cluster = RustFSTestClusterEnvironment::with_topology(ClusterTopology::single_pool_multidrive(2, 2)).await?;
+    let proxy = cluster.start_volume_proxy_for_node(0).await?;
+    let proxied = proxy.local_addr().to_string();
+    assert!(cluster.rustfs_volumes_arg().contains(&proxied));
+
+    let result: TestResult = async {
+        cluster.start().await?;
+        cluster.create_test_bucket(BUCKET).await?;
+        let payload = vec![0x6Du8; 256 * 1024];
+        put_get_roundtrip(&cluster, "volume-proxy/object", &payload).await
+    }
+    .await;
+
+    proxy.shutdown().await;
+    result
+}

crates/e2e_test/src/common.rs

@@ -34,6 +34,7 @@ use serde_json;
 use std::ffi::OsStr;
 use std::fs as stdfs;
 use std::io::ErrorKind;
+use std::net::SocketAddr;
 use std::path::{Path, PathBuf};
 use std::process::{Child, Command, Stdio};
 use std::sync::Once;
@@ -1214,6 +1215,9 @@ pub struct RustFSTestClusterEnvironment {
     pub node_extra_env: Vec<Vec<(String, String)>>,
     pub node_capture_log_paths: Vec<Option<String>>,
     pub topology: ClusterTopology,
+    /// Optional socket proxies used for the corresponding node's volume
+    /// endpoints. Proxies must be installed before [`Self::start`].
+    volume_proxy_addresses: Vec<Option<SocketAddr>>,
 }
 
 impl RustFSTestClusterEnvironment {
@@ -1305,6 +1309,7 @@ impl RustFSTestClusterEnvironment {
             extra_env.push(("RUSTFS_UNSAFE_BYPASS_DISK_CHECK".to_string(), "true".to_string()));
         }
 
+        let node_count = topology.node_count;
         Ok(Self {
             nodes,
             temp_dir,
@@ -1314,6 +1319,7 @@ impl RustFSTestClusterEnvironment {
             node_extra_env: vec![Vec::new(); topology.node_count],
             node_capture_log_paths: vec![None; topology.node_count],
             topology,
+            volume_proxy_addresses: vec![None; node_count],
         })
     }
 
@@ -1381,6 +1387,34 @@ impl RustFSTestClusterEnvironment {
         self.build_volumes_arg()
     }
 
+    /// Start a socket proxy for one node's volume endpoints and route all
+    /// subsequent `RUSTFS_VOLUMES` references for that node through it.
+    ///
+    /// Call this before [`Self::start`], then use the returned proxy's
+    /// [`crate::fault_proxy::FaultProxy::set_mode`] to inject latency,
+    /// blackhole, or one-way partition faults. The node's own listen address
+    /// remains direct, so S3 clients can still reach it while peer disk/RPC
+    /// traffic is steered through the proxy.
+    pub async fn start_volume_proxy_for_node(
+        &mut self,
+        node_idx: usize,
+    ) -> Result<crate::fault_proxy::FaultProxy, Box<dyn std::error::Error + Send + Sync>> {
+        self.ensure_node_index(node_idx)?;
+        if self.volume_proxy_addresses[node_idx].is_some() {
+            return Err(format!("a volume proxy is already configured for node {node_idx}").into());
+        }
+        let target = self.nodes[node_idx].address.parse::<SocketAddr>()?;
+        let proxy = crate::fault_proxy::FaultProxy::start(target).await?;
+        self.volume_proxy_addresses[node_idx] = Some(proxy.local_addr());
+        Ok(proxy)
+    }
+
+    fn volume_address(&self, node_idx: usize) -> String {
+        self.volume_proxy_addresses[node_idx]
+            .map(|address| address.to_string())
+            .unwrap_or_else(|| self.nodes[node_idx].address.clone())
+    }
+
     fn build_volumes_arg(&self) -> String {
         let pools = self.topology.normalized_pools();
 
@@ -1389,7 +1423,11 @@ impl RustFSTestClusterEnvironment {
             return self
                 .nodes
                 .iter()
-                .flat_map(|n| n.data_dirs.iter().map(move |dir| format!("http://{}{}", n.address, dir)))
+                .enumerate()
+                .flat_map(|(node_idx, n)| {
+                    let address = self.volume_address(node_idx);
+                    n.data_dirs.iter().map(move |dir| format!("http://{}{}", address, dir))
+                })
                 .collect::<Vec<_>>()
                 .join(" ");
         }
@@ -1400,13 +1438,19 @@ impl RustFSTestClusterEnvironment {
         pools
             .iter()
             .map(|nodes| {
-                let node = &self.nodes[nodes[0]];
+                let node_idx = nodes[0];
+                let node = &self.nodes[node_idx];
                 let base = node
                     .data_dirs
                     .first()
                     .and_then(|d| d.rsplit_once('/').map(|(parent, _)| parent))
                     .unwrap_or(&node.data_dir);
-                format!("http://{}{}/drive{{0...{}}}", node.address, base, self.topology.drives_per_node - 1)
+                format!(
+                    "http://{}{}/drive{{0...{}}}",
+                    self.volume_address(node_idx),
+                    base,
+                    self.topology.drives_per_node - 1
+                )
             })
             .collect::<Vec<_>>()
             .join(" ")
@@ -2000,7 +2044,7 @@ mod tests {
         }
         let multidrive = topology.drives_per_node > 1;
 
-        let nodes = (0..topology.node_count)
+        let nodes: Vec<ClusterNode> = (0..topology.node_count)
             .map(|i| {
                 let address = format!("127.0.0.1:{}", 9000 + i);
                 let data_dirs: Vec<String> = if multidrive {
@@ -2021,6 +2065,7 @@ mod tests {
             })
             .collect();
 
+        let node_count = nodes.len();
         RustFSTestClusterEnvironment {
             nodes,
             temp_dir,
@@ -2030,6 +2075,7 @@ mod tests {
             node_extra_env: vec![Vec::new(); topology.node_count],
             node_capture_log_paths: vec![None; topology.node_count],
             topology,
+            volume_proxy_addresses: vec![None; node_count],
         }
     }
 
@@ -2114,6 +2160,25 @@ mod tests {
         assert!(ClusterTopology::single_pool_multidrive(1, 1).validate().is_ok());
     }
 
+    #[tokio::test]
+    async fn volume_proxy_rewrites_cluster_volume_endpoint() {
+        let mut env = RustFSTestClusterEnvironment::new(1)
+            .await
+            .expect("cluster environment should allocate a node");
+        let direct = env.nodes[0].address.clone();
+        let proxy = env
+            .start_volume_proxy_for_node(0)
+            .await
+            .expect("volume proxy should bind before the target server starts");
+        let proxied = proxy.local_addr().to_string();
+        let volumes = env.rustfs_volumes_arg();
+
+        assert!(volumes.contains(&proxied), "volumes must use the proxy address: {volumes}");
+        assert!(!volumes.contains(&direct), "volumes must not retain the direct address: {volumes}");
+
+        proxy.shutdown().await;
+    }
+
     #[test]
     fn cluster_node_env_supports_per_node_overrides() {
         let mut env = fake_cluster(ClusterTopology::single_pool(4));

crates/e2e_test/src/reliant/mod.rs

@@ -21,5 +21,6 @@ mod head_tls_bodyless_test;
 mod lifecycle;
 mod lock;
 mod node_interact_test;
+mod s3_select_compression;
 mod sql;
 mod tiering;

crates/e2e_test/src/reliant/s3_select_compression.rs

@@ -0,0 +1,351 @@
+#![cfg(test)]
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::common::{RustFSTestEnvironment, init_logging};
+use async_compression::tokio::write::BzEncoder;
+use aws_sdk_s3::{
+    Client,
+    error::ProvideErrorMetadata,
+    operation::select_object_content::{SelectObjectContentOutput, builders::SelectObjectContentFluentBuilder},
+    types::{
+        CompressionType, CsvInput, CsvOutput, ExpressionType, FileHeaderInfo, InputSerialization, JsonInput, JsonOutput,
+        JsonType, OutputSerialization, SelectObjectContentEventStream,
+    },
+};
+use aws_smithy_types::event_stream::RawMessage;
+use bytes::Bytes;
+use flate2::{Compression, write::GzEncoder};
+use std::{error::Error, io::Cursor, time::Duration};
+use tokio::io::AsyncWriteExt;
+
+const BUCKET: &str = "s3-select-compression";
+const SELECT_RESPONSE_TIMEOUT: Duration = Duration::from_secs(30);
+
+type TestResult<T> = Result<T, Box<dyn Error + Send + Sync>>;
+
+async fn create_test_environment(extra_env: &[(&str, &str)]) -> TestResult<(RustFSTestEnvironment, Client)> {
+    init_logging();
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server_with_env(vec![], extra_env).await?;
+    let client = env.create_s3_client();
+    client.create_bucket().bucket(BUCKET).send().await?;
+    Ok((env, client))
+}
+
+async fn put_object(client: &Client, key: &str, body: &[u8]) -> TestResult<()> {
+    client
+        .put_object()
+        .bucket(BUCKET)
+        .key(key)
+        .body(Bytes::copy_from_slice(body).into())
+        .send()
+        .await?;
+    Ok(())
+}
+
+fn gzip(input: &[u8]) -> TestResult<Vec<u8>> {
+    let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
+    std::io::Write::write_all(&mut encoder, input)?;
+    Ok(encoder.finish()?)
+}
+
+async fn bzip2(input: &[u8]) -> TestResult<Vec<u8>> {
+    let mut encoder = BzEncoder::new(Cursor::new(Vec::new()));
+    encoder.write_all(input).await?;
+    encoder.shutdown().await?;
+    Ok(encoder.into_inner().into_inner())
+}
+
+fn csv_select_request(
+    client: &Client,
+    key: &str,
+    compression: CompressionType,
+    expression: &str,
+) -> SelectObjectContentFluentBuilder {
+    client
+        .select_object_content()
+        .bucket(BUCKET)
+        .key(key)
+        .expression(expression)
+        .expression_type(ExpressionType::Sql)
+        .input_serialization(
+            InputSerialization::builder()
+                .compression_type(compression)
+                .csv(CsvInput::builder().file_header_info(FileHeaderInfo::Use).build())
+                .build(),
+        )
+        .output_serialization(OutputSerialization::builder().csv(CsvOutput::builder().build()).build())
+}
+
+fn json_select_request(
+    client: &Client,
+    key: &str,
+    compression: CompressionType,
+    json_type: JsonType,
+) -> SelectObjectContentFluentBuilder {
+    client
+        .select_object_content()
+        .bucket(BUCKET)
+        .key(key)
+        .expression("SELECT name FROM S3Object")
+        .expression_type(ExpressionType::Sql)
+        .input_serialization(
+            InputSerialization::builder()
+                .compression_type(compression)
+                .json(JsonInput::builder().set_type(Some(json_type)).build())
+                .build(),
+        )
+        .output_serialization(OutputSerialization::builder().json(JsonOutput::builder().build()).build())
+}
+
+async fn collect_success(
+    mut response: SelectObjectContentOutput,
+    compressed_bytes: usize,
+    processed_bytes: usize,
+) -> TestResult<Vec<u8>> {
+    tokio::time::timeout(SELECT_RESPONSE_TIMEOUT, async move {
+        let mut records = Vec::new();
+        let mut stats = None;
+        let mut saw_end = false;
+
+        while let Some(event) = response.payload.recv().await? {
+            assert!(!saw_end, "Select emitted an event after End");
+            match event {
+                SelectObjectContentEventStream::Records(event) => {
+                    assert!(stats.is_none(), "Select emitted Records after Stats");
+                    if let Some(payload) = event.payload {
+                        records.extend_from_slice(payload.as_ref());
+                    }
+                }
+                SelectObjectContentEventStream::Stats(event) => {
+                    assert!(stats.is_none(), "Select emitted more than one Stats event");
+                    stats = event.details;
+                }
+                SelectObjectContentEventStream::End(_) => {
+                    assert!(stats.is_some(), "Select emitted End before Stats");
+                    saw_end = true;
+                }
+                _ => assert!(stats.is_none(), "Select emitted a non-terminal event after Stats"),
+            }
+        }
+
+        let stats = stats.ok_or("Select response ended without a Stats event")?;
+        assert_eq!(stats.bytes_scanned(), Some(i64::try_from(compressed_bytes)?));
+        assert_eq!(stats.bytes_processed(), Some(i64::try_from(processed_bytes)?));
+        assert_eq!(stats.bytes_returned(), Some(i64::try_from(records.len())?));
+        assert!(saw_end, "Select response ended without an End event");
+        Ok::<_, Box<dyn Error + Send + Sync>>(records)
+    })
+    .await
+    .map_err(|_| -> Box<dyn Error + Send + Sync> { "Select response timed out".into() })?
+}
+
+async fn assert_truncated_stream_failure(mut response: SelectObjectContentOutput) -> TestResult<()> {
+    tokio::time::timeout(SELECT_RESPONSE_TIMEOUT, async move {
+        loop {
+            match response.payload.recv().await {
+                Err(error) => {
+                    // S3 Select request-level errors use `error` frames, which this SDK version exposes as raw response errors.
+                    if let Some(code) = error.code() {
+                        assert_eq!(code, "TruncatedInput", "unexpected modeled event-stream error: {error:?}");
+                    } else if let aws_sdk_s3::error::SdkError::ResponseError(context) = &error
+                        && let RawMessage::Decoded(message) = context.raw()
+                    {
+                        let header = |name: &str| {
+                            message
+                                .headers()
+                                .iter()
+                                .find(|header| header.name().as_str() == name)
+                                .and_then(|header| header.value().as_string().ok())
+                                .map(|value| value.as_str())
+                        };
+                        assert_eq!(header(":message-type"), Some("error"));
+                        assert_eq!(header(":error-code"), Some("TruncatedInput"));
+                    } else {
+                        panic!("unexpected event-stream error: {error:?}");
+                    }
+                    return Ok(());
+                }
+                Ok(Some(SelectObjectContentEventStream::Stats(_))) | Ok(Some(SelectObjectContentEventStream::End(_))) => {
+                    return Err("truncated compressed input reached a success terminal event".into());
+                }
+                Ok(Some(_)) => {}
+                Ok(None) => return Err("truncated compressed input ended without an error event".into()),
+            }
+        }
+    })
+    .await
+    .map_err(|_| -> Box<dyn Error + Send + Sync> { "truncated Select response timed out".into() })?
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn test_select_object_content_compressed_csv_and_json() -> TestResult<()> {
+    const CSV: &[u8] = b"name,age\nAlice,30\nBob,25\n";
+    const JSON_LINES: &[u8] = b"{\"name\":\"Alice\"}\n{\"name\":\"Bob\"}\n";
+    const JSON_DOCUMENT: &[u8] = br#"[{"name":"Alice"},{"name":"Bob"}]"#;
+
+    let (_env, client) = create_test_environment(&[]).await?;
+
+    let gzip_csv = gzip(CSV)?;
+    put_object(&client, "records.csv.gz", &gzip_csv).await?;
+    let gzip_csv_records = collect_success(
+        csv_select_request(&client, "records.csv.gz", CompressionType::Gzip, "SELECT * FROM S3Object")
+            .send()
+            .await?,
+        gzip_csv.len(),
+        CSV.len(),
+    )
+    .await?;
+    assert_eq!(gzip_csv_records, b"Alice,30\nBob,25\n");
+
+    let bzip_csv = bzip2(CSV).await?;
+    put_object(&client, "records.csv.bz2", &bzip_csv).await?;
+    let bzip_csv_records = collect_success(
+        csv_select_request(&client, "records.csv.bz2", CompressionType::Bzip2, "SELECT * FROM S3Object")
+            .send()
+            .await?,
+        bzip_csv.len(),
+        CSV.len(),
+    )
+    .await?;
+    assert_eq!(bzip_csv_records, gzip_csv_records);
+
+    let gzip_json_lines = gzip(JSON_LINES)?;
+    put_object(&client, "json-lines", &gzip_json_lines).await?;
+    let gzip_json_records = collect_success(
+        json_select_request(&client, "json-lines", CompressionType::Gzip, JsonType::Lines)
+            .send()
+            .await?,
+        gzip_json_lines.len(),
+        JSON_LINES.len(),
+    )
+    .await?;
+    assert_eq!(gzip_json_records, JSON_LINES);
+
+    let bzip_json_lines = bzip2(JSON_LINES).await?;
+    put_object(&client, "records.jsonl.bz2", &bzip_json_lines).await?;
+    let bzip_json_records = collect_success(
+        json_select_request(&client, "records.jsonl.bz2", CompressionType::Bzip2, JsonType::Lines)
+            .send()
+            .await?,
+        bzip_json_lines.len(),
+        JSON_LINES.len(),
+    )
+    .await?;
+    assert_eq!(bzip_json_records, gzip_json_records);
+
+    let gzip_json_document = gzip(JSON_DOCUMENT)?;
+    put_object(&client, "document.json.gz", &gzip_json_document).await?;
+    let document_records = collect_success(
+        json_select_request(&client, "document.json.gz", CompressionType::Gzip, JsonType::Document)
+            .send()
+            .await?,
+        gzip_json_document.len(),
+        JSON_DOCUMENT.len(),
+    )
+    .await?;
+    assert_eq!(document_records, JSON_LINES);
+
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn test_select_object_content_invalid_compressed_stream_fails() -> TestResult<()> {
+    const CSV: &[u8] = b"name\nAlice\n";
+
+    let (_env, client) = create_test_environment(&[]).await?;
+
+    put_object(&client, "invalid.csv.gz", CSV).await?;
+    let invalid = csv_select_request(&client, "invalid.csv.gz", CompressionType::Gzip, "SELECT * FROM S3Object")
+        .send()
+        .await
+        .expect_err("invalid GZIP header must fail before streaming");
+    assert_eq!(
+        invalid.as_service_error().and_then(ProvideErrorMetadata::code),
+        Some("InvalidCompressionFormat")
+    );
+
+    put_object(&client, "empty.csv.gz", b"").await?;
+    let empty = csv_select_request(&client, "empty.csv.gz", CompressionType::Gzip, "SELECT * FROM S3Object")
+        .send()
+        .await
+        .expect_err("empty GZIP input must fail as truncated");
+    assert_eq!(empty.as_service_error().and_then(ProvideErrorMetadata::code), Some("TruncatedInput"));
+
+    let mut truncated = bzip2(CSV).await?;
+    truncated.pop();
+    put_object(&client, "truncated.csv.bz2", &truncated).await?;
+    let truncated = csv_select_request(&client, "truncated.csv.bz2", CompressionType::Bzip2, "SELECT * FROM S3Object")
+        .send()
+        .await?;
+    assert_truncated_stream_failure(truncated).await?;
+
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn test_select_object_content_compressed_disconnect_releases_query() -> TestResult<()> {
+    const OBJECT: &str = "disconnect.csv.gz";
+    const ROWS: usize = 16 * 1024;
+    const RELEASE_ATTEMPTS: usize = 20;
+    const RELEASE_BACKOFF: Duration = Duration::from_millis(25);
+
+    let (_env, client) = create_test_environment(&[("RUSTFS_S3SELECT_MAX_CONCURRENT_QUERIES", "1")]).await?;
+    let row = format!("{}\n", "x".repeat(1023));
+    let mut body = Vec::with_capacity("value\n".len() + ROWS * row.len());
+    body.extend_from_slice(b"value\n");
+    for _ in 0..ROWS {
+        body.extend_from_slice(row.as_bytes());
+    }
+    let compressed = gzip(&body)?;
+    put_object(&client, OBJECT, &compressed).await?;
+
+    let first = csv_select_request(&client, OBJECT, CompressionType::Gzip, "SELECT * FROM S3Object")
+        .send()
+        .await?;
+    let saturated = csv_select_request(&client, OBJECT, CompressionType::Gzip, "SELECT * FROM S3Object")
+        .send()
+        .await
+        .expect_err("the unread compressed response should retain the only query permit");
+    assert_eq!(saturated.as_service_error().and_then(ProvideErrorMetadata::code), Some("SlowDown"));
+
+    drop(first);
+    let second = tokio::time::timeout(Duration::from_secs(5), async {
+        for attempt in 0..RELEASE_ATTEMPTS {
+            match csv_select_request(&client, OBJECT, CompressionType::Gzip, "SELECT * FROM S3Object")
+                .send()
+                .await
+            {
+                Ok(response) => return Ok::<_, Box<dyn Error + Send + Sync>>(response),
+                Err(error)
+                    if error.as_service_error().and_then(ProvideErrorMetadata::code) == Some("SlowDown")
+                        && attempt + 1 < RELEASE_ATTEMPTS =>
+                {
+                    tokio::time::sleep(RELEASE_BACKOFF).await;
+                }
+                Err(error) if error.as_service_error().and_then(ProvideErrorMetadata::code) == Some("SlowDown") => {
+                    return Err("disconnected compressed Select retained its query permit".into());
+                }
+                Err(error) => return Err(format!("unexpected Select error after disconnect: {error}").into()),
+            }
+        }
+        Err("query permit release retry loop ended unexpectedly".into())
+    })
+    .await
+    .map_err(|_| -> Box<dyn Error + Send + Sync> { "compressed Select did not release its query permit".into() })??;
+    drop(second);
+
+    Ok(())
+}

crates/e2e_test/src/reliant/sql.rs

@@ -122,6 +122,24 @@ async fn select_json_document(client: &Client, key: &str, expression: &str) -> T
     process_select_response(response).await
 }
 
+fn csv_select_request(
+    client: &Client,
+    key: &str,
+) -> aws_sdk_s3::operation::select_object_content::builders::SelectObjectContentFluentBuilder {
+    client
+        .select_object_content()
+        .bucket(BUCKET)
+        .key(key)
+        .expression("SELECT * FROM S3Object")
+        .expression_type(ExpressionType::Sql)
+        .input_serialization(
+            InputSerialization::builder()
+                .csv(CsvInput::builder().file_header_info(FileHeaderInfo::Use).build())
+                .build(),
+        )
+        .output_serialization(OutputSerialization::builder().csv(CsvOutput::builder().build()).build())
+}
+
 async fn process_select_response(
     mut event_stream: aws_sdk_s3::operation::select_object_content::SelectObjectContentOutput,
 ) -> TestResult<String> {
@@ -188,30 +206,42 @@ async fn assert_input_byte_stats(
     let mut last_progress: Option<aws_sdk_s3::types::Progress> = None;
     let mut stats = None;
     let mut saw_end = false;
-    while let Some(event) = payload.recv().await? {
-        match event {
-            aws_sdk_s3::types::SelectObjectContentEventStream::Records(records) => {
-                if let Some(bytes) = records.payload {
-                    records_len = records_len.saturating_add(u64::try_from(bytes.as_ref().len())?);
+    tokio::time::timeout(SELECT_RESPONSE_TIMEOUT, async {
+        // The AWS SDK validates both event-stream CRCs before yielding an event.
+        while let Some(event) = payload.recv().await? {
+            assert!(!saw_end, "Select emitted an event after End");
+            match event {
+                aws_sdk_s3::types::SelectObjectContentEventStream::Records(records) => {
+                    assert!(stats.is_none(), "Select emitted Records after Stats");
+                    if let Some(bytes) = records.payload {
+                        records_len = records_len.saturating_add(u64::try_from(bytes.as_ref().len())?);
+                    }
                 }
-            }
-            aws_sdk_s3::types::SelectObjectContentEventStream::Progress(event) => {
-                let details = event.details.ok_or("Progress event did not contain details")?;
-                if let Some(previous) = last_progress.as_ref() {
-                    assert!(details.bytes_scanned() >= previous.bytes_scanned());
-                    assert!(details.bytes_processed() >= previous.bytes_processed());
-                    assert!(details.bytes_returned() >= previous.bytes_returned());
+                aws_sdk_s3::types::SelectObjectContentEventStream::Progress(event) => {
+                    assert!(stats.is_none(), "Select emitted Progress after Stats");
+                    let details = event.details.ok_or("Progress event did not contain details")?;
+                    if let Some(previous) = last_progress.as_ref() {
+                        assert!(details.bytes_scanned() >= previous.bytes_scanned());
+                        assert!(details.bytes_processed() >= previous.bytes_processed());
+                        assert!(details.bytes_returned() >= previous.bytes_returned());
+                    }
+                    last_progress = Some(details);
                 }
-                last_progress = Some(details);
+                aws_sdk_s3::types::SelectObjectContentEventStream::Stats(event) => {
+                    assert!(stats.is_none(), "Select emitted more than one Stats event");
+                    stats = event.details;
+                }
+                aws_sdk_s3::types::SelectObjectContentEventStream::End(_) => {
+                    assert!(stats.is_some(), "Select emitted End before Stats");
+                    saw_end = true;
+                }
+                _ => assert!(stats.is_none(), "Select emitted a non-terminal event after Stats"),
             }
-            aws_sdk_s3::types::SelectObjectContentEventStream::Stats(event) => stats = event.details,
-            aws_sdk_s3::types::SelectObjectContentEventStream::End(_) => {
-                saw_end = true;
-                break;
-            }
-            _ => {}
         }
-    }
+        Ok::<(), Box<dyn Error + Send + Sync>>(())
+    })
+    .await
+    .map_err(|_| -> Box<dyn Error + Send + Sync> { "Select response timed out".into() })??;
 
     let stats = stats.ok_or("Select response ended without a Stats event")?;
     let input_len = i64::try_from(body.len())?;
@@ -219,10 +249,11 @@ async fn assert_input_byte_stats(
     assert_eq!(stats.bytes_processed(), Some(input_len));
     assert_eq!(stats.bytes_returned(), Some(i64::try_from(records_len)?));
     if progress_enabled {
-        let progress = last_progress.ok_or("Select response ended without a Progress event")?;
-        assert_eq!(progress.bytes_scanned(), stats.bytes_scanned());
-        assert_eq!(progress.bytes_processed(), stats.bytes_processed());
-        assert_eq!(progress.bytes_returned(), stats.bytes_returned());
+        if let Some(progress) = last_progress {
+            assert!(stats.bytes_scanned() >= progress.bytes_scanned());
+            assert!(stats.bytes_processed() >= progress.bytes_processed());
+            assert!(stats.bytes_returned() >= progress.bytes_returned());
+        }
     } else {
         assert!(last_progress.is_none(), "disabled request progress emitted a Progress event");
     }
@@ -231,7 +262,7 @@ async fn assert_input_byte_stats(
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
-async fn test_select_object_content_reports_input_byte_stats() -> TestResult<()> {
+async fn test_select_object_content_http_event_order_crc_and_input_byte_stats() -> TestResult<()> {
     const CSV_BODY: &[u8] = b"name,age\nAlice,30\nBob,25\n";
     const JSON_LINES_BODY: &[u8] = b"{\"name\":\"Alice\"}\n{\"name\":\"Bob\"}\n";
     const JSON_DOCUMENT_BODY: &[u8] = b"[{\"name\":\"Alice\"},{\"name\":\"Bob\"}]";
@@ -289,6 +320,60 @@ async fn test_select_object_content_reports_input_byte_stats() -> TestResult<()>
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn test_select_object_content_http_disconnect_releases_query() -> TestResult<()> {
+    const OBJECT: &str = "disconnect.csv";
+    const ROWS: usize = 16 * 1024;
+    const RELEASE_BACKOFF: Duration = Duration::from_millis(25);
+
+    init_logging();
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server_with_env(vec![], &[("RUSTFS_S3SELECT_MAX_CONCURRENT_QUERIES", "1")])
+        .await?;
+    let client = env.create_s3_client();
+    setup_test_bucket(&client).await?;
+
+    let row = format!("{}\n", "x".repeat(1023));
+    let mut body = Vec::with_capacity("value\n".len() + ROWS * row.len());
+    body.extend_from_slice(b"value\n");
+    for _ in 0..ROWS {
+        body.extend_from_slice(row.as_bytes());
+    }
+    client
+        .put_object()
+        .bucket(BUCKET)
+        .key(OBJECT)
+        .body(Bytes::from(body).into())
+        .send()
+        .await?;
+
+    // Leaving this response body unread fills the bounded HTTP/event channels before the query can finish.
+    let first = csv_select_request(&client, OBJECT).send().await?;
+    let saturated = csv_select_request(&client, OBJECT)
+        .send()
+        .await
+        .expect_err("the first HTTP stream should retain the only query permit");
+    assert_eq!(saturated.as_service_error().and_then(ProvideErrorMetadata::code), Some("SlowDown"));
+
+    drop(first);
+    let second = tokio::time::timeout(Duration::from_secs(5), async {
+        loop {
+            match csv_select_request(&client, OBJECT).send().await {
+                Ok(response) => return Ok::<_, Box<dyn Error + Send + Sync>>(response),
+                Err(error) if error.as_service_error().and_then(ProvideErrorMetadata::code) == Some("SlowDown") => {
+                    tokio::time::sleep(RELEASE_BACKOFF).await;
+                }
+                Err(error) => return Err(format!("unexpected Select error after disconnect: {error}").into()),
+            }
+        }
+    })
+    .await
+    .map_err(|_| -> Box<dyn Error + Send + Sync> { "disconnected Select did not release its query permit".into() })??;
+    drop(second);
+
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn test_select_object_content_csv_basic() -> TestResult<()> {
     let (_env, client) = create_test_environment().await?;