upgrade to Rust 1.45.0 (#10396)

### Problem

Keep up with Rust releases.

### Solution

Upgrade to Rust v1.45.0.

Clippy-related changes:
- `impl FnMut(...) -> ()`. Drop the () as extraneous.
- `Result.and_then(|x| Ok(y))` -> `Result.map(|x| y)`
- `Result.and_then(|x| Err(y))` -> `Result.map_err(|x| y)`
- `clippy::identity_conversion` renamed to `clippy::useless_conversion`

Other changes:
- Bump type length limit in `src/rust/engine/process_executor/src/main.rs` due to compile error.

### Result

Code compiles. Tests pass.

rust-toolchain

@@ -1 +1 @@
-1.44.1
+1.45.0

src/rust/engine/graph/src/lib.rs

@@ -947,15 +947,15 @@ impl<N: Node> Graph<N> {
     &self,
     roots: &[N],
     context: &N::Context,
-    mut f: impl FnMut(&N, N::Item) -> (),
+    mut f: impl FnMut(&N, N::Item),
   ) {
     let inner = self.inner.lock();
     for (n, v) in inner.live_reachable(roots, context) {
       f(n, v);
     }
   }
 
-  pub fn visit_live(&self, context: &N::Context, mut f: impl FnMut(&N, N::Item) -> ()) {
+  pub fn visit_live(&self, context: &N::Context, mut f: impl FnMut(&N, N::Item)) {
     let inner = self.inner.lock();
     for (n, v) in inner.live(context) {
       f(n, v);

src/rust/engine/nailgun/src/lib.rs

@@ -196,7 +196,7 @@ impl Server {
     self
       .exited_receiver
       .await
-      .or_else(|_| Err("Server exited uncleanly.".to_owned()))?
+      .map_err(|_| "Server exited uncleanly.".to_owned())?
   }
 }
 

src/rust/engine/process_execution/src/local.rs

@@ -174,7 +174,7 @@ impl StreamedHermeticCommand {
       .stderr(Stdio::piped())
       .spawn()
       .map_err(|e| format!("Error launching process: {:?}", e))
-      .and_then(|mut child| {
+      .map(|mut child| {
         debug!("spawned local process as {} for {:?}", child.id(), req);
         let stdout_stream = FramedRead::new(child.stdout.take().unwrap(), BytesCodec::new())
           .map_ok(|bytes| ChildOutput::Stdout(bytes.into()))
@@ -194,11 +194,9 @@ impl StreamedHermeticCommand {
           })
           .boxed();
 
-        Ok(
-          futures::stream::select_all(vec![stdout_stream, stderr_stream, exit_stream])
-            .map_err(|e| format!("Failed to consume process outputs: {:?}", e))
-            .boxed(),
-        )
+        futures::stream::select_all(vec![stdout_stream, stderr_stream, exit_stream])
+          .map_err(|e| format!("Failed to consume process outputs: {:?}", e))
+          .boxed()
       })
   }
 }

src/rust/engine/process_execution/src/nailgun/nailgun_pool.rs

@@ -242,10 +242,10 @@ impl NailgunPool {
       nailgun_server_fingerprint,
       build_id,
     )
-    .and_then(move |process| {
+    .map(move |process| {
       let port = process.port;
       processes.insert(name.clone(), process);
-      Ok(port)
+      port
     })
   }
 }
@@ -316,19 +316,19 @@ impl NailgunProcess {
         let port = read_port(&mut child);
         port.map(|port| (child, port))
       })
-      .and_then(|(child, port)| {
+      .map(|(child, port)| {
         debug!(
           "Created nailgun server process with pid {} and port {}",
           child.id(),
           port
         );
-        Ok(NailgunProcess {
+        NailgunProcess {
           port,
           fingerprint: nailgun_server_fingerprint,
           name,
           handle: Arc::new(Mutex::new(child)),
           build_id,
-        })
+        }
       })
   }
 }

src/rust/engine/process_executor/src/main.rs

@@ -26,7 +26,7 @@
 #![allow(clippy::new_without_default, clippy::new_ret_no_self)]
 // Arc<Mutex> can be more clear than needing to grok Orderings:
 #![allow(clippy::mutex_atomic)]
-#![type_length_limit = "1076739"]
+#![type_length_limit = "1076744"]
 
 use std::collections::{BTreeMap, BTreeSet};
 use std::convert::TryFrom;

src/rust/engine/sharded_lmdb/src/lib.rs

@@ -375,7 +375,7 @@ impl ShardedLmdb {
       .await
   }
 
-  #[allow(clippy::identity_conversion)] // False positive: https://github.com/rust-lang/rust-clippy/issues/3913
+  #[allow(clippy::useless_conversion)] // False positive: https://github.com/rust-lang/rust-clippy/issues/3913
   pub fn compact(&self) -> Result<(), String> {
     for (env, old_dir, _) in ShardedLmdb::envs(&self.root_path, self.max_size)? {
       let new_dir = TempDir::new_in(old_dir.parent().unwrap()).expect("TODO");