Bumps rust toolchain and resolves linting errors

.cargo/config → .cargo/config.toml

@@ -8,3 +8,6 @@ ABQ_WORKSPACE_DIR = { value = "", relative = true }
 # easily hit the number of open FDs (especially on MacOS) during our tests.
 # Revisit this when we've made everything async on the worker side.
 RUST_TEST_THREADS = "1"
+
+[workspace]
+resolver = "2"

.github/workflows/build_and_upload.yml

@@ -5,26 +5,26 @@ on:
   workflow_dispatch:
     inputs:
       ref:
-        description: 'ref to build'
+        description: "ref to build"
         required: true
         type: string
       release_channel:
-        description: 'release channel'
+        description: "release channel"
         required: true
         type: choice
         options:
-        - "v1"
-        - "unstable"
+          - "v1"
+          - "unstable"
 
   # for automatic release from main on the unstable release channel
   workflow_call:
     inputs:
       ref:
-        description: 'ref to build'
+        description: "ref to build"
         required: true
         type: string
       release_channel:
-        description: 'release channel'
+        description: "release channel"
         required: true
         type: string
     secrets:
@@ -61,24 +61,24 @@ jobs:
             deprecated-platform: linux_x86-64
             os: linux
             architecture: x86_64
-            cross-target: 'x86_64-unknown-linux-musl'
+            cross-target: "x86_64-unknown-linux-musl"
             install-musl-tools: true
           - runs-on: ubuntu-latest
             deprecated-platform: linux_aarch64
             os: linux
             architecture: aarch64
-            cross-target: 'aarch64-unknown-linux-musl'
+            cross-target: "aarch64-unknown-linux-musl"
             container: messense/rust-musl-cross:aarch64-musl@sha256:777bd4c61179c38dc213bb8472500584646d28fd4a7c3e0b30b9ef70cb446d58
           - runs-on: macos-11 # use an older version for broader osx support
             deprecated-platform: darwin_x86-64
             os: darwin
             architecture: x86_64
-            cross-target: ''
+            cross-target: ""
           - runs-on: macos-11 # first OS X to support arm64 -- so the first os for cross compilation
             deprecated-platform: darwin_aarch64
             os: darwin
             architecture: aarch64
-            cross-target: 'aarch64-apple-darwin'
+            cross-target: "aarch64-apple-darwin"
     runs-on: ${{ matrix.runs-on }}
     container: ${{ matrix.container }}
     outputs:
@@ -122,7 +122,7 @@ jobs:
       - name: Install Rust toolchain
         uses: rwx-research/rust-toolchain@abq
         with:
-          toolchain: 1.65.0
+          toolchain: 1.81.0
           target: ${{ matrix.cross-target }}
 
       # We don't build a musl ABQ on MacOS
@@ -132,7 +132,7 @@ jobs:
           sudo apt-get install -y musl-tools
 
       - name: Build release
-        if: '!matrix.cross-target'
+        if: "!matrix.cross-target"
         run: cargo build --release --all-features
 
       - name: Build release

.github/workflows/test_and_package_development.yml

@@ -76,7 +76,7 @@ jobs:
       - name: Install Rust toolchain
         uses: rwx-research/rust-toolchain@abq
         with:
-          toolchain: 1.65.0
+          toolchain: 1.81.0
           target: ${{ env.RUST_TARGET }}
           components: clippy, rustfmt
 

Cargo.toml

@@ -1,4 +1,5 @@
 [workspace]
+resolver = "2"
 
 members = [
   "crates/abq_cli",

crates/abq_cli/src/instance.rs

@@ -19,7 +19,6 @@ use signal_hook::consts::TERM_SIGNALS;
 use signal_hook::iterator::Signals;
 use std::net::{IpAddr, SocketAddr};
 use std::thread;
-use tempfile::TempDir;
 use tokio_cron_scheduler::JobScheduler;
 
 use thiserror::Error;
@@ -280,12 +279,7 @@ enum AbqLocator {
         queue_negotiator: QueueNegotiatorHandle,
         server_addr: SocketAddr,
     },
-    Local(Abq, EphemeralAbqGuards),
-}
-
-struct EphemeralAbqGuards {
-    _manifests_path: TempDir,
-    _results_path: TempDir,
+    Local(Abq),
 }
 
 #[derive(Debug, Error)]
@@ -312,14 +306,14 @@ impl AbqInstance {
             AbqLocator::Remote {
                 queue_negotiator, ..
             } => *queue_negotiator,
-            AbqLocator::Local(abq, _) => abq.get_negotiator_handle(),
+            AbqLocator::Local(abq) => abq.get_negotiator_handle(),
         }
     }
 
     pub fn server_addr(&self) -> SocketAddr {
         match &self.locator {
             AbqLocator::Remote { server_addr, .. } => *server_addr,
-            AbqLocator::Local(abq, _) => abq.server_addr(),
+            AbqLocator::Local(abq) => abq.server_addr(),
         }
     }
 
@@ -359,13 +353,9 @@ impl AbqInstance {
         config.server_options = ServerOptions::new(server_auth, server_tls);
 
         let queue = Abq::start(config).await;
-        let guards = EphemeralAbqGuards {
-            _manifests_path: manifests_path,
-            _results_path: results_path,
-        };
 
         AbqInstance {
-            locator: AbqLocator::Local(queue, guards),
+            locator: AbqLocator::Local(queue),
             client_options: ClientOptions::new(client_auth, client_tls),
         }
     }

crates/abq_queue/src/queue.rs

@@ -17,7 +17,7 @@ use abq_utils::net_protocol::queue::{
     AssociatedTestResults, CancelReason, GroupId, NativeRunnerInfo, NegotiatorInfo, Request,
     TestResultsResponse, TestSpec, TestStrategy,
 };
-use abq_utils::net_protocol::results::{self, OpaqueLazyAssociatedTestResults};
+use abq_utils::net_protocol::results::{self};
 use abq_utils::net_protocol::runners::{Manifest, MetadataMap, StdioOutput};
 use abq_utils::net_protocol::work_server::{self, RetryManifestResponse};
 use abq_utils::net_protocol::workers::{
@@ -2315,10 +2315,6 @@ impl QueueServer {
         entity: Entity,
         mut stream: Box<dyn net_async::ServerStream>,
     ) -> OpaqueResult<()> {
-        enum Response {
-            One(TestResultsResponse),
-            Chunk(OpaqueLazyAssociatedTestResults),
-        }
         let results_cell = match queues.get_read_results_cell(&run_id).located(here!()) {
             Ok(state) => match state {
                 ReadResultsState::ReadFromCell(cell) => cell,

crates/abq_test_support/native_runner_simulator/Cargo.toml

@@ -21,7 +21,19 @@ serde.workspace = true
 serde_derive.workspace = true
 serde_json.workspace = true
 
-tokio.workspace = true
+tokio = { version = "1.26.0", features = [
+  "fs",
+  "io-util",
+  "io-std",
+  "net",
+  "rt",
+  "rt-multi-thread",
+  "macros",
+  "sync",
+  "time",
+  "process",
+] }
+
 tempfile.workspace = true
 
 abq_utils = { path = "../../abq_utils" }

crates/abq_utils/src/net_async/tcp.rs

@@ -11,8 +11,6 @@ use tokio::net::ToSocketAddrs;
 use super::UnverifiedServerStream;
 use crate::auth::{ClientAuthStrategy, Role, ServerAuthStrategy};
 
-pub struct RawServerStream(tokio::net::TcpStream);
-
 #[derive(Debug)]
 #[repr(transparent)]
 pub struct ClientStream(tokio::net::TcpStream);

crates/abq_utils/src/net_async/tls.rs

@@ -15,8 +15,6 @@ use tokio::net::ToSocketAddrs;
 use super::UnverifiedServerStream;
 use crate::auth::{ClientAuthStrategy, Role, ServerAuthStrategy};
 
-pub struct RawServerStream(tokio::net::TcpStream);
-
 #[derive(Debug)]
 pub struct ServerStream(tokio_tls::server::TlsStream<tokio::net::TcpStream>, Role);
 

crates/abq_workers/src/negotiate.rs

@@ -25,7 +25,7 @@ use abq_utils::{
     auth::User,
     error::{EntityfulError, ErrorEntity, ResultLocation},
     exit::ExitCode,
-    here, log_entityful, net, net_async,
+    here, log_entityful, net_async,
     net_opt::ClientOptions,
     net_protocol::{
         self,
@@ -139,7 +139,7 @@ pub enum WorkersNegotiateError {
 }
 
 /// The worker pool side of the negotiation.
-pub struct WorkersNegotiator(Box<dyn net::ClientStream>, WorkerContext);
+pub struct WorkersNegotiator();
 
 pub enum NegotiatedWorkers {
     /// No more workers were created, because there is no more work to be done.

crates/abq_workers/src/workers.rs

@@ -799,15 +799,6 @@ async fn attempt_test_id_for_test_like_runner(
                     vec![test]
                 }
             }
-            #[cfg(feature = "test-test_ids")]
-            (R::EchoOnRetry(succeed_on), s) => {
-                if succeed_on == _attempt {
-                    let result = echo::EchoWorker::run(echo::EchoWork { message: s });
-                    vec![result]
-                } else {
-                    panic!("Failed to echo!");
-                }
-            }
             (runner, test_id) => unreachable!(
                 "Invalid runner/test_id combination: {:?} and {:?}",
                 runner, test_id
@@ -1229,139 +1220,6 @@ mod test {
         test_echo_n(proto, 2, 8).await;
     }
 
-    #[test]
-    #[cfg(feature = "test-test_ids")]
-    fn test_timeout() {
-        let (write_work, get_next_tests) = work_writer();
-        let (results, results_handler) = results_collector();
-
-        let run_id = RunId::new();
-        let manifest = ManifestMessage {
-            test_ids: vec![TestId::Echo("mona lisa".to_string())],
-        };
-
-        let (default_config, manifest_collector) = setup_pool(
-            TestLikeRunner::InduceTimeout,
-            run_id,
-            manifest,
-            get_next_tests,
-            results_handler,
-        );
-
-        let timeout = Duration::from_millis(1);
-        let config = WorkerPoolConfig {
-            work_timeout: timeout,
-            work_retries: 0,
-            ..default_config
-        };
-        let mut pool = WorkerPool::new(config);
-
-        for test_id in await_manifest_test_specs(manifest_collector) {
-            write_work(local_work(test_id, run_id, WorkId("id1".to_string())));
-        }
-
-        write_work(NextWork::EndOfWork);
-
-        await_results(results, |results| {
-            let results = results.lock().unwrap();
-            if results.is_empty() {
-                return false;
-            }
-
-            results.get("id1").unwrap() == &WorkerResult::Timeout(timeout)
-        });
-
-        pool.shutdown();
-    }
-
-    #[test]
-    #[cfg(feature = "test-test_ids")]
-    fn test_panic_no_retries() {
-        let (write_work, get_next_tests) = work_writer();
-        let (results, results_handler) = results_collector();
-
-        let run_id = RunId::new();
-        let manifest = ManifestMessage {
-            test_ids: vec![TestId::Echo("".to_string())],
-        };
-
-        let (default_config, manifest_collector) = setup_pool(
-            TestLikeRunner::EchoOnRetry(10),
-            run_id,
-            manifest,
-            get_next_tests,
-            results_handler,
-        );
-
-        let config = WorkerPoolConfig {
-            work_retries: 0,
-            ..default_config
-        };
-        let mut pool = WorkerPool::new(config);
-
-        for test_id in await_manifest_test_specs(manifest_collector) {
-            write_work(local_work(test_id, run_id, WorkId("id1".to_string())));
-        }
-        write_work(NextWork::EndOfWork);
-
-        await_results(results, |results| {
-            let results = results.lock().unwrap();
-            if results.is_empty() {
-                return false;
-            }
-
-            results.get("id1").unwrap() == &WorkerResult::Panic("Failed to echo!".to_string())
-        });
-
-        pool.shutdown();
-    }
-
-    #[test]
-    #[cfg(feature = "test-test_ids")]
-    fn test_panic_succeed_after_retry() {
-        let (write_work, get_next_tests) = work_writer();
-        let (results, results_handler) = results_collector();
-
-        let run_id = RunId::new();
-        let manifest = ManifestMessage {
-            test_ids: vec![TestId::Echo("okay".to_string())],
-        };
-
-        let (default_config, manifest_collector) = setup_pool(
-            TestLikeRunner::EchoOnRetry(2),
-            run_id,
-            manifest,
-            get_next_tests,
-            results_handler,
-        );
-
-        let config = WorkerPoolConfig {
-            work_retries: 1,
-            ..default_config
-        };
-        let mut pool = WorkerPool::new(config);
-
-        for test_id in await_manifest_test_specs(manifest_collector) {
-            write_work(local_work(test_id, run_id, WorkId("id1".to_string())));
-        }
-        write_work(NextWork::EndOfWork);
-
-        await_results(results, |results| {
-            let results = results.lock().unwrap();
-            if results.is_empty() {
-                return false;
-            }
-
-            results.get("id1").unwrap()
-                == &WorkerResult::Output(Output {
-                    success: true,
-                    message: "okay".to_string(),
-                })
-        });
-
-        pool.shutdown();
-    }
-
     #[tokio::test]
     #[traced_test]
     async fn bad_message_doesnt_take_down_queue_negotiator_server() {

rust-toolchain.toml

@@ -1,3 +1,3 @@
 [toolchain]
-channel = "1.65.0"
+channel = "1.81.0"
 profile = "default"