TaskManager spawn helpers (#1065)

* Add `spawn` and `run` helpers

These functions are added in the hopes of providing easy methods to 
ensure our tasks are spawned onto their own threads when starting them 
with TaskManager. 

`run()` is provided for tasks that can not/do not implement `Send`. 
However, effort should be put into allowing these tasks to use 
`spawn()`.

Exporting the type alias `TaskLocalBoxFuture` is a nicety.
It allows us to more easily target the return type if we ever want to 
refactor it into something else. And prevents the proliferation of 
Managed Tasks importing LocalBoxFuture from wherever they please.
- `futures::future::LocalBoxFuture`
- `futures::prelude::future::LocalBoxFuture`
- `futures_util::future::LocalBoxFuture`

* Update impl ManagedTask to use helpers

Everywhere, we replace the return type with the type alias.

Where we can, use `task_manager::spawn()` to ensure tasks are spawned in
their own tasks.

Where we can’t, use `task_manager::run()` to wrap tasks in a pin until 
they can implement `Send` and use `task_manager::spawn()`.

* Update chain-ingest to use task_manager helper

Author: michaeldjeffrey

file_store/src/file_info_poller.rs

@@ -2,7 +2,7 @@ use crate::{traits::MsgDecode, Error, FileInfo, Result};
 use aws_sdk_s3::primitives::ByteStream;
 use chrono::{DateTime, Utc};
 use derive_builder::Builder;
-use futures::{future::LocalBoxFuture, stream::BoxStream, StreamExt};
+use futures::{stream::BoxStream, StreamExt};
 use futures_util::TryFutureExt;
 use retainer::Cache;
 use std::{collections::VecDeque, marker::PhantomData, sync::Arc, time::Duration};
@@ -235,14 +235,8 @@ where
     fn start_task(
         self: Box<Self>,
         shutdown: triggered::Listener,
-    ) -> LocalBoxFuture<'static, anyhow::Result<()>> {
-        let handle = tokio::spawn(self.run(shutdown));
-
-        Box::pin(
-            handle
-                .map_err(anyhow::Error::from)
-                .and_then(|result| async move { result.map_err(anyhow::Error::from) }),
-        )
+    ) -> task_manager::TaskLocalBoxFuture {
+        task_manager::spawn(self.run(shutdown).err_into())
     }
 }
 

file_store/src/file_sink.rs

@@ -2,7 +2,7 @@ use crate::{file_upload::FileUpload, traits::MsgBytes, Error, Result};
 use async_compression::tokio::write::GzipEncoder;
 use bytes::Bytes;
 use chrono::{DateTime, Utc};
-use futures::{future::LocalBoxFuture, SinkExt, TryFutureExt};
+use futures::{SinkExt, TryFutureExt};
 use metrics::Label;
 use std::time::Duration;
 use std::{
@@ -290,14 +290,8 @@ impl<T: MsgBytes + Send + Sync + 'static> ManagedTask for FileSink<T> {
     fn start_task(
         self: Box<Self>,
         shutdown: triggered::Listener,
-    ) -> LocalBoxFuture<'static, anyhow::Result<()>> {
-        let handle = tokio::spawn(self.run(shutdown));
-
-        Box::pin(
-            handle
-                .map_err(anyhow::Error::from)
-                .and_then(|result| async move { result.map_err(anyhow::Error::from) }),
-        )
+    ) -> task_manager::TaskLocalBoxFuture {
+        task_manager::spawn(self.run(shutdown).err_into())
     }
 }
 

file_store/src/file_upload.rs

@@ -1,5 +1,5 @@
 use crate::{Error, Result};
-use futures::{future::LocalBoxFuture, StreamExt, TryFutureExt};
+use futures::{StreamExt, TryFutureExt};
 use std::{
     path::{Path, PathBuf},
     time::Duration,
@@ -54,14 +54,8 @@ impl ManagedTask for FileUploadServer {
     fn start_task(
         self: Box<Self>,
         shutdown: triggered::Listener,
-    ) -> LocalBoxFuture<'static, anyhow::Result<()>> {
-        let handle = tokio::spawn(self.run(shutdown));
-
-        Box::pin(
-            handle
-                .map_err(anyhow::Error::from)
-                .and_then(|result| async move { result.map_err(anyhow::Error::from) }),
-        )
+    ) -> task_manager::TaskLocalBoxFuture {
+        task_manager::spawn(self.run(shutdown).err_into())
     }
 }
 

ingest/src/server_chain.rs

@@ -118,7 +118,7 @@ impl ManagedTask for GrpcServer {
         self: Box<Self>,
         shutdown: triggered::Listener,
     ) -> LocalBoxFuture<'static, anyhow::Result<()>> {
-        Box::pin(self.run(shutdown))
+        task_manager::spawn(self.run(shutdown))
     }
 }
 

ingest/src/server_iot.rs

@@ -6,10 +6,7 @@ use file_store::{
     file_upload,
     traits::{FileSinkCommitStrategy, FileSinkRollTime, FileSinkWriteExt, MsgVerify},
 };
-use futures::{
-    future::{LocalBoxFuture, TryFutureExt},
-    Stream, StreamExt,
-};
+use futures::{future::TryFutureExt, Stream, StreamExt};
 use helium_crypto::{Network, PublicKey};
 use helium_proto::services::poc_lora::{
     self, lora_stream_request_v1::Request as StreamRequest,
@@ -181,9 +178,9 @@ impl ManagedTask for GrpcServer {
     fn start_task(
         self: Box<Self>,
         shutdown: triggered::Listener,
-    ) -> LocalBoxFuture<'static, anyhow::Result<()>> {
+    ) -> task_manager::TaskLocalBoxFuture {
         let address = self.address;
-        Box::pin(async move {
+        task_manager::spawn(async move {
             let grpc_server = transport::Server::builder()
                 .layer(custom_tracing::grpc_layer::new_with_span(make_span))
                 .layer(poc_metrics::request_layer!("ingest_server_iot_connection"))

ingest/src/server_mobile.rs

@@ -6,7 +6,6 @@ use file_store::{
     file_upload,
     traits::{FileSinkCommitStrategy, FileSinkRollTime, FileSinkWriteExt, MsgVerify},
 };
-use futures::future::LocalBoxFuture;
 use futures_util::TryFutureExt;
 use helium_crypto::{Network, PublicKey, PublicKeyBinary};
 use helium_proto::services::poc_mobile::{
@@ -73,8 +72,8 @@ where
     fn start_task(
         self: Box<Self>,
         shutdown: triggered::Listener,
-    ) -> LocalBoxFuture<'static, anyhow::Result<()>> {
-        Box::pin(self.run(shutdown))
+    ) -> task_manager::TaskLocalBoxFuture {
+        task_manager::spawn(self.run(shutdown))
     }
 }
 

iot_config/src/db_cleaner.rs

@@ -1,5 +1,4 @@
 use chrono::{DateTime, Utc};
-use futures::TryFutureExt;
 use sqlx::{PgConnection, Pool, Postgres, Transaction};
 use std::time::Duration;
 use task_manager::ManagedTask;
@@ -15,12 +14,8 @@ impl ManagedTask for DbCleaner {
     fn start_task(
         self: Box<Self>,
         shutdown: triggered::Listener,
-    ) -> futures::future::LocalBoxFuture<'static, anyhow::Result<()>> {
-        Box::pin(
-            tokio::spawn(self.run(shutdown))
-                .map_err(anyhow::Error::from)
-                .and_then(|result| async move { result }),
-        )
+    ) -> task_manager::TaskLocalBoxFuture {
+        task_manager::spawn(self.run(shutdown))
     }
 }
 

iot_config/src/main.rs

@@ -1,6 +1,5 @@
 use anyhow::{Error, Result};
 use clap::Parser;
-use futures::future::LocalBoxFuture;
 use futures_util::TryFutureExt;
 use helium_proto::services::{
     iot_config::{AdminServer, GatewayServer, OrgServer, RouteServer},
@@ -144,8 +143,8 @@ impl ManagedTask for GrpcServer {
     fn start_task(
         self: Box<Self>,
         shutdown: triggered::Listener,
-    ) -> LocalBoxFuture<'static, anyhow::Result<()>> {
-        Box::pin(async move {
+    ) -> task_manager::TaskLocalBoxFuture {
+        task_manager::spawn(async move {
             let grpc_server = transport::Server::builder()
                 .http2_keepalive_interval(Some(Duration::from_secs(250)))
                 .http2_keepalive_timeout(Some(Duration::from_secs(60)))

iot_packet_verifier/src/burner.rs

@@ -4,7 +4,7 @@ use crate::{
         confirm_pending_txns, Burn, ConfirmPendingError, PendingTables, PendingTablesTransaction,
     },
 };
-use futures::{future::LocalBoxFuture, TryFutureExt};
+use futures::TryFutureExt;
 use solana::{burn::SolanaNetwork, GetSignature, SolanaRpcError};
 use std::time::Duration;
 use task_manager::ManagedTask;
@@ -25,14 +25,8 @@ where
     fn start_task(
         self: Box<Self>,
         shutdown: triggered::Listener,
-    ) -> LocalBoxFuture<'static, anyhow::Result<()>> {
-        let handle = tokio::spawn(self.run(shutdown));
-
-        Box::pin(
-            handle
-                .map_err(anyhow::Error::from)
-                .and_then(|result| async move { result.map_err(anyhow::Error::from) }),
-        )
+    ) -> task_manager::TaskLocalBoxFuture {
+        task_manager::spawn(self.run(shutdown).err_into())
     }
 }
 

iot_packet_verifier/src/daemon.rs

@@ -41,8 +41,8 @@ where
     fn start_task(
         self: Box<Self>,
         shutdown: triggered::Listener,
-    ) -> futures::future::LocalBoxFuture<'static, anyhow::Result<()>> {
-        Box::pin(self.run(shutdown))
+    ) -> task_manager::TaskLocalBoxFuture {
+        task_manager::spawn(self.run(shutdown))
     }
 }