run unit tests in both modes

.github/workflows/_build-and-test-locally.yml

@@ -229,8 +229,13 @@ jobs:
           ${cov_prefix} cargo nextest run $CARGO_FLAGS $CARGO_FEATURES -E '!package(pageserver)'
 
           # run pageserver tests with different settings
-          for io_engine in std-fs tokio-epoll-uring ; do
-            NEON_PAGESERVER_UNIT_TEST_VIRTUAL_FILE_IOENGINE=$io_engine ${cov_prefix} cargo nextest run $CARGO_FLAGS $CARGO_FEATURES  -E 'package(pageserver)'
+          for get_vectored_concurrent_io in sequential sidecar-task; do
+            for io_engine in std-fs tokio-epoll-uring ; do
+              NEON_PAGESERVER_UNIT_TEST_GET_VECTORED_CONCURRENT_IO=$get_vectored_concurrent_io \
+                NEON_PAGESERVER_UNIT_TEST_VIRTUAL_FILE_IOENGINE=$io_engine \
+                ${cov_prefix} \
+                cargo nextest run $CARGO_FLAGS $CARGO_FEATURES  -E 'package(pageserver)'
+            done
           done
 
           # Run separate tests for real S3

libs/utils/src/env.rs

@@ -2,6 +2,7 @@
 
 use std::{fmt::Display, str::FromStr};
 
+/// For types `V` that implement [`FromStr`].
 pub fn var<V, E>(varname: &str) -> Option<V>
 where
     V: FromStr<Err = E>,
@@ -10,7 +11,9 @@ where
     match std::env::var(varname) {
         Ok(s) => Some(
             s.parse()
-                .map_err(|e| format!("failed to parse env var {varname}: {e:#}"))
+                .map_err(|e| {
+                    format!("failed to parse env var {varname} using FromStr::parse: {e:#}")
+                })
                 .unwrap(),
         ),
         Err(std::env::VarError::NotPresent) => None,
@@ -19,3 +22,24 @@ where
         }
     }
 }
+
+/// For types `V` that implement [`serde::de::DeserializeOwned`].
+pub fn var_serde_json_string<V>(varname: &str) -> Option<V>
+where
+    V: serde::de::DeserializeOwned,
+{
+    match std::env::var(varname) {
+        Ok(s) => Some({
+            let value = serde_json::Value::String(s);
+            serde_json::from_value(value)
+                .map_err(|e| {
+                    format!("failed to parse env var {varname} as a serde_json json string: {e:#}")
+                })
+                .unwrap()
+        }),
+        Err(std::env::VarError::NotPresent) => None,
+        Err(std::env::VarError::NotUnicode(_)) => {
+            panic!("env var {varname} is not unicode")
+        }
+    }
+}

pageserver/src/tenant.rs

@@ -5708,7 +5708,7 @@ mod tests {
     use pageserver_api::value::Value;
     use pageserver_compaction::helpers::overlaps_with;
     use rand::{thread_rng, Rng};
-    use storage_layer::{IoConcurrency, PersistentLayerKey, SelectedIoConcurrency};
+    use storage_layer::{IoConcurrency, PersistentLayerKey};
     use tests::storage_layer::ValuesReconstructState;
     use tests::timeline::{GetVectoredError, ShutdownMode};
     use timeline::{CompactOptions, DeltaLayerTestDesc};

pageserver/src/tenant/storage_layer.rs

@@ -497,6 +497,7 @@ impl IoConcurrency {
     #[cfg(test)]
     pub(crate) fn spawn_for_test() -> impl std::ops::DerefMut<Target = Self> {
         use std::ops::{Deref, DerefMut};
+        use tracing::info;
         use utils::sync::gate::Gate;
 
         // Spawn needs a Gate, give it one.
@@ -518,10 +519,41 @@ impl IoConcurrency {
             }
         }
         let gate = Box::new(Gate::default());
+
+        // The default behavior when running Rust unit tests without any further
+        // flags is to use the new behavior.
+        // The CI uses the following environment variable to unit test both old
+        // and new behavior.
+        // NB: the Python regression & perf tests take the `else` branch
+        // below and have their own defaults management.
+        let selected = {
+            // The pageserver_api::config type is unsuitable because it's internally tagged.
+            #[derive(serde::Deserialize)]
+            #[serde(rename_all = "kebab-case")]
+            enum TestOverride {
+                Sequential,
+                SidecarTask,
+            }
+            use once_cell::sync::Lazy;
+            static TEST_OVERRIDE: Lazy<TestOverride> = Lazy::new(|| {
+                utils::env::var_serde_json_string(
+                    "NEON_PAGESERVER_UNIT_TEST_GET_VECTORED_CONCURRENT_IO",
+                )
+                .unwrap_or(TestOverride::SidecarTask)
+            });
+
+            match *TEST_OVERRIDE {
+                TestOverride::Sequential => SelectedIoConcurrency::Sequential,
+                TestOverride::SidecarTask => {
+                    SelectedIoConcurrency::SidecarTask(gate.enter().expect("just created it"))
+                }
+            }
+        };
+
+        info!(?selected, "get_vectored_concurrent_io test");
+
         Wrapper {
-            inner: Self::spawn(SelectedIoConcurrency::SidecarTask(
-                gate.enter().expect("just created it"),
-            )),
+            inner: Self::spawn(selected),
             gate,
         }
     }
@@ -1112,6 +1144,7 @@ mod tests2 {
     /// ```text
     /// RUST_LOG=trace cargo nextest run  --features testing  --no-capture test_io_concurrency_noise
     /// running 1 test
+    /// 2025-01-21T17:42:01.335679Z  INFO get_vectored_concurrent_io test selected=SidecarTask
     /// 2025-01-21T17:42:01.335680Z TRACE spawning sidecar task task_id=0
     /// 2025-01-21T17:42:01.335937Z TRACE IoConcurrency_sidecar{task_id=0}: start
     /// 2025-01-21T17:42:01.335972Z TRACE IoConcurrency_sidecar{task_id=0}: received new io future
@@ -1137,6 +1170,13 @@ mod tests2 {
         crate::tenant::harness::setup_logging();
 
         let io_concurrency = IoConcurrency::spawn_for_test();
+        match *io_concurrency {
+            IoConcurrency::Sequential => {
+                // This test asserts behavior in sidecar mode, doesn't make sense in sequential mode.
+                return;
+            }
+            IoConcurrency::SidecarTask { .. } => {}
+        }
         let mut reconstruct_state = ValuesReconstructState::new(io_concurrency.clone());
 
         let (io_fut_is_waiting_tx, io_fut_is_waiting) = tokio::sync::oneshot::channel();