diff --git a/.github/workflows/test-adapters.yml b/.github/workflows/test-adapters.yml
index b310fa4c37e..4a676bd8737 100644
--- a/.github/workflows/test-adapters.yml
+++ b/.github/workflows/test-adapters.yml
@@ -16,6 +16,11 @@ jobs:
   adapter-tests:
     if: ${{ !contains(vars.CI_SKIP_JOBS, 'adapter-tests') }}
     name: Test Adapters
+    # The suite runs in 3 to 10 minutes. Without a bound, a test that blocks
+    # instead of failing holds a runner for the 6 hour default and stalls the
+    # merge queue behind it; the cancel-if-* sentinels cannot help, because a
+    # job that hangs never reports failure.
+    timeout-minutes: 30
     strategy:
       matrix:
         include:
diff --git a/crates/adapters/src/test.rs b/crates/adapters/src/test.rs
index a2819518fa5..797d252c185 100644
--- a/crates/adapters/src/test.rs
+++ b/crates/adapters/src/test.rs
@@ -74,7 +74,7 @@ pub use mock_dezset::{
     MockDeZSet, MockUpdate, wait_for_output_count, wait_for_output_ordered,
     wait_for_output_unordered,
 };
-pub use mock_input_consumer::{MockInputConsumer, MockInputParser};
+pub use mock_input_consumer::{ErrorCallback, MockInputConsumer, MockInputParser};
 pub use mock_output_consumer::MockOutputConsumer;
 
 pub static DEFAULT_TIMEOUT_MS: u128 = 600_000;
diff --git a/crates/adapters/src/transport/s3.rs b/crates/adapters/src/transport/s3.rs
index 62142bc63a7..305a851dc79 100644
--- a/crates/adapters/src/transport/s3.rs
+++ b/crates/adapters/src/transport/s3.rs
@@ -965,7 +965,10 @@ fn to_s3_config(config: &Arc<S3InputConfig>) -> aws_sdk_s3::Config {
 #[cfg(test)]
 mod test {
     use crate::{
-        test::{MockDeZSet, MockInputConsumer, MockInputParser, mock_parser_pipeline, wait},
+        test::{
+            ErrorCallback, MockDeZSet, MockInputConsumer, MockInputParser, mock_parser_pipeline,
+            wait,
+        },
         transport::s3::{S3InputConfig, S3InputReader},
     };
     use aws_sdk_s3::{
@@ -984,6 +987,12 @@ mod test {
     use mockall::predicate::eq;
     use serde::{Deserialize, Serialize};
     use std::sync::Arc;
+    use std::time::Duration;
+
+    /// How long a test waits for an error the connector is expected to report.
+    /// Generous: it only bounds the wait so a regression fails instead of
+    /// hanging the whole test binary.
+    const ERROR_TIMEOUT: Duration = Duration::from_secs(60);
 
     #[derive(Debug, PartialEq, Eq, Hash, Serialize, Deserialize, Clone, PartialOrd, Ord)]
     struct TestStruct {
@@ -1027,9 +1036,16 @@ mod test {
   }
 }"#;
 
+    /// Builds a reader over `mock`, with `on_error` already installed.
+    ///
+    /// The callback has to be passed in rather than registered afterwards:
+    /// [`S3InputReader::new_inner`] starts listing objects as soon as it is
+    /// called, so a caller that registers later can miss an error the listing
+    /// already reported.
     fn test_setup(
         config_str: &str,
         mock: super::MockS3Client,
+        on_error: ErrorCallback,
     ) -> (
         Box<dyn crate::InputReader>,
         MockInputConsumer,
@@ -1049,7 +1065,7 @@ mod test {
             &config.connector_config.format.unwrap(),
         )
         .unwrap();
-        consumer.on_error(Some(Box::new(|_, _| ())));
+        consumer.on_error(Some(on_error));
         let reader = Box::new(S3InputReader::new_inner(
             &transport_config,
             Box::new(consumer.clone()),
@@ -1061,7 +1077,8 @@ mod test {
     }
 
     fn run_test(config_str: &str, mock: super::MockS3Client, mut test_data: Vec<TestStruct>) {
-        let (reader, consumer, parser, input_handle) = test_setup(config_str, mock);
+        let (reader, consumer, parser, input_handle) =
+            test_setup(config_str, mock, Box::new(|_, _| ()));
         // No outputs should be produced at this point.
         assert!(parser.state().data.is_empty());
         assert!(!consumer.state().eoi);
@@ -1185,7 +1202,8 @@ mod test {
             .with(eq("test-bucket"), eq(""), eq(&None), eq(&None))
             .return_once(|_, _, _, _| Ok((objs, None)));
         let test_data: Vec<TestStruct> = (0..1000).map(|i| TestStruct { i }).collect();
-        let (reader, _consumer, _parser, input_handle) = test_setup(MULTI_KEY_CONFIG_STR, mock);
+        let (reader, _consumer, _parser, input_handle) =
+            test_setup(MULTI_KEY_CONFIG_STR, mock, Box::new(|_, _| ()));
         reader.extend();
         wait(
             || {
@@ -1220,13 +1238,22 @@ mod test {
             .return_once(|_, _, _, _| {
                 Err(ListObjectsV2Error::NoSuchBucket(NoSuchBucketBuilder::default().build()).into())
             });
-        let (reader, consumer, _parser, _) = test_setup(MULTI_KEY_CONFIG_STR, mock);
         let (tx, rx) = std::sync::mpsc::channel();
-        consumer.on_error(Some(Box::new(move |fatal, err| {
-            tx.send((fatal, format!("{err}"))).unwrap()
-        })));
+        let (reader, _consumer, _parser, _) = test_setup(
+            MULTI_KEY_CONFIG_STR,
+            mock,
+            // Ignore a send failure: the receiver is gone once the assertion
+            // below has run, and the reader can still report during teardown.
+            Box::new(move |fatal, err| {
+                let _ = tx.send((fatal, format!("{err}")));
+            }),
+        );
         reader.extend();
-        assert_eq!((true, "NoSuchBucket".to_string()), rx.recv().unwrap());
+        assert_eq!(
+            (true, "NoSuchBucket".to_string()),
+            rx.recv_timeout(ERROR_TIMEOUT)
+                .expect("connector reported no error for a missing bucket"),
+        );
     }
 
     #[test]