Merge pull request #618 from helium/jg/fix-invalid-witness-md

fix witness metadata on invalid reports

Cargo.toml

@@ -22,6 +22,7 @@ members = [
     "solana",
     "task_manager",
 ]
+resolver = "2"
 
 [workspace.package]
 authors = ["Nova Labs <[email protected]>"]
@@ -58,7 +59,6 @@ sqlx = {version = "0", features = [
   "macros",
   "runtime-tokio-rustls"
 ]}
-
 helium-crypto = {version = "0.6.8", features=["sqlx-postgres", "multisig"]}
 helium-proto = {git = "https://github.com/helium/proto", branch = "master", features = ["services"]}
 hextree = "*"

ingest/src/server_mobile.rs

@@ -279,13 +279,10 @@ pub async fn grpc_server(shutdown: triggered::Listener, settings: &Settings) ->
     let Some(api_token) = settings
         .token
         .as_ref()
-        .and_then(|token| {
-            format!("Bearer {token}")
-                .parse::<MetadataValue<_>>()
-                .ok()
-        }) else {
-            bail!("expected valid api token in settings");
-        };
+        .and_then(|token| format!("Bearer {token}").parse::<MetadataValue<_>>().ok())
+    else {
+        bail!("expected valid api token in settings");
+    };
 
     tracing::info!(
         "grpc listening on {grpc_addr} and server mode {:?}",

iot_config/src/client/mod.rs

@@ -143,7 +143,7 @@ impl gateway_info::GatewayInfoResolver for Client {
             .filter_map(|resp| async move { resp.ok() })
             .map(move |resp| (resp, pubkey.clone()))
             .filter_map(|(resp, pubkey)| async move { resp.verify(&pubkey).map(|_| resp).ok() })
-            .flat_map(|resp| stream::iter(resp.gateways.into_iter()))
+            .flat_map(|resp| stream::iter(resp.gateways))
             .map(gateway_info::GatewayInfo::from)
             .boxed();
 

iot_packet_verifier/src/burner.rs

@@ -88,8 +88,12 @@ where
     pub async fn burn(&mut self) -> Result<(), BurnError<P::Error, S::Error>> {
         // Create burn transaction and execute it:
 
-        let Some(Burn { payer, amount }) = self.pending_burns.fetch_next().await
-            .map_err(BurnError::SqlError)? else {
+        let Some(Burn { payer, amount }) = self
+            .pending_burns
+            .fetch_next()
+            .await
+            .map_err(BurnError::SqlError)?
+        else {
             return Ok(());
         };
 

iot_verifier/src/poc.rs

@@ -289,7 +289,7 @@ impl Poc {
                 0,
                 0,
                 InvalidParticipantSide::Beaconer,
-            ))
+            ));
         };
         // run the witness verifications
         match do_witness_verifications(
@@ -320,9 +320,9 @@ impl Poc {
                 invalid_response.details,
                 &witness_report.report,
                 witness_report.received_timestamp,
-                Some(beaconer_metadata.location),
-                beaconer_metadata.gain,
-                beaconer_metadata.elevation,
+                Some(witness_metadata.location),
+                witness_metadata.gain,
+                witness_metadata.elevation,
                 InvalidParticipantSide::Witness,
             )),
         }

mobile_config/src/client/gateway_client.rs

@@ -98,7 +98,7 @@ impl gateway_info::GatewayInfoResolver for GatewayClient {
                     Err(_) => None,
                 }
             })
-            .flat_map(|res| stream::iter(res.gateways.into_iter()))
+            .flat_map(|res| stream::iter(res.gateways))
             .map(gateway_info::GatewayInfo::from)
             .boxed();
 

solana/src/balance_monitor.rs

@@ -22,7 +22,7 @@ impl BalanceMonitor {
             Some(rpc_client) => {
                 let Ok(keypair) = Keypair::from_bytes(&rpc_client.keypair) else {
                     tracing::error!("sol monitor: keypair failed to deserialize");
-                    return Err(Box::new(SolanaRpcError::InvalidKeypair))
+                    return Err(Box::new(SolanaRpcError::InvalidKeypair));
                 };
                 let app_metric_name = format!("{app_account}-sol-balance");
 

task_manager/src/lib.rs

@@ -124,7 +124,7 @@ fn start_futures(
 ) -> Vec<StopableLocalFuture> {
     shutdown_triggers
         .into_iter()
-        .zip(tasks.into_iter())
+        .zip(tasks)
         .map(
             |((shutdown_trigger, shutdown_listener), task)| StopableLocalFuture {
                 shutdown_trigger,
@@ -134,6 +134,7 @@ fn start_futures(
         .collect()
 }
 
+#[allow(clippy::manual_try_fold)]
 async fn stop_all(futures: Vec<StopableLocalFuture>) -> anyhow::Result<()> {
     futures::stream::iter(futures.into_iter().rev())
         .fold(Ok(()), |last_result, local| async move {