Merge pull request #567 from helium/jg/stream-locked-route-stuffs

IoT Config locked route updates and shutdown listeners

iot_config/src/gateway_service.rs

@@ -35,6 +35,7 @@ pub struct GatewayService {
     region_map: RegionMapReader,
     signing_key: Arc<Keypair>,
     delegate_cache: watch::Receiver<org::DelegateCache>,
+    shutdown: triggered::Listener,
 }
 
 impl GatewayService {
@@ -44,6 +45,7 @@ impl GatewayService {
         region_map: RegionMapReader,
         auth_cache: AuthCache,
         delegate_cache: watch::Receiver<org::DelegateCache>,
+        shutdown: triggered::Listener,
     ) -> Result<Self> {
         let gateway_cache = Arc::new(Cache::new());
         let cache_clone = gateway_cache.clone();
@@ -56,6 +58,7 @@ impl GatewayService {
             region_map,
             signing_key: Arc::new(settings.signing_keypair()?),
             delegate_cache,
+            shutdown,
         })
     }
 
@@ -275,18 +278,21 @@ impl iot_config::Gateway for GatewayService {
         let signing_key = self.signing_key.clone();
         let batch_size = request.batch_size;
         let region_map = self.region_map.clone();
+        let shutdown_listener = self.shutdown.clone();
 
         let (tx, rx) = tokio::sync::mpsc::channel(20);
 
         tokio::spawn(async move {
-            stream_all_gateways_info(
-                &pool,
-                tx.clone(),
-                &signing_key,
-                region_map.clone(),
-                batch_size,
-            )
-            .await
+            tokio::select! {
+                _ = shutdown_listener => (),
+                _ = stream_all_gateways_info(
+                    &pool,
+                    tx.clone(),
+                    &signing_key,
+                    region_map.clone(),
+                    batch_size,
+                ) => (),
+            }
         });
 
         Ok(Response::new(GrpcStreamResult::new(rx)))
@@ -313,24 +319,24 @@ async fn stream_all_gateways_info(
             })
             .collect();
 
-        let mut response = GatewayInfoStreamResV1 {
+        let mut gateway = GatewayInfoStreamResV1 {
             gateways: gateway_infos,
             timestamp,
             signer: signer.clone(),
             signature: vec![],
         };
 
-        response = match signing_key.sign(&response.encode_to_vec()) {
+        gateway = match signing_key.sign(&gateway.encode_to_vec()) {
             Ok(signature) => GatewayInfoStreamResV1 {
                 signature,
-                ..response
+                ..gateway
             },
             Err(_) => {
                 continue;
             }
         };
 
-        tx.send(Ok(response)).await?;
+        tx.send(Ok(gateway)).await?;
     }
     Ok(())
 }

iot_config/src/lib.rs

@@ -36,14 +36,17 @@ pub fn update_channel<T: Clone>() -> broadcast::Sender<T> {
     update_tx
 }
 
-pub async fn broadcast_update<T>(
+pub async fn broadcast_update<T: std::fmt::Debug>(
     message: T,
     sender: broadcast::Sender<T>,
 ) -> Result<(), broadcast::error::SendError<T>> {
     while !enqueue_update(sender.len()) {
         tokio::time::sleep(tokio::time::Duration::from_millis(25)).await
     }
-    sender.send(message).map(|_| ())
+    sender.send(message).map(|_| ()).map_err(|err| {
+        tracing::error!(error = ?err, "failed to broadcast routing update");
+        err
+    })
 }
 
 fn enqueue_update(queue_size: usize) -> bool {

iot_config/src/main.rs

@@ -95,6 +95,7 @@ impl Daemon {
             region_map.clone(),
             auth_cache.clone(),
             delegate_key_cache,
+            shutdown_listener.clone(),
         )?;
         let route_svc = RouteService::new(
             settings,
@@ -108,6 +109,7 @@ impl Daemon {
             pool.clone(),
             route_svc.clone_update_channel(),
             delegate_key_updater,
+            shutdown_listener.clone(),
         )?;
         let admin_svc = AdminService::new(
             settings,

iot_config/src/org_service.rs

@@ -1,6 +1,6 @@
 use crate::{
     admin::{AuthCache, KeyType},
-    helium_netids, lora_field, org,
+    broadcast_update, helium_netids, lora_field, org,
     route::list_routes,
     telemetry, verify_public_key, GrpcResult, Settings,
 };
@@ -26,6 +26,7 @@ pub struct OrgService {
     route_update_tx: broadcast::Sender<RouteStreamResV1>,
     signing_key: Keypair,
     delegate_updater: watch::Sender<org::DelegateCache>,
+    shutdown: triggered::Listener,
 }
 
 #[derive(Clone, Debug, PartialEq)]
@@ -41,13 +42,15 @@ impl OrgService {
         pool: Pool<Postgres>,
         route_update_tx: broadcast::Sender<RouteStreamResV1>,
         delegate_updater: watch::Sender<org::DelegateCache>,
+        shutdown: triggered::Listener,
     ) -> Result<Self> {
         Ok(Self {
             auth_cache,
             pool,
             route_update_tx,
             signing_key: settings.signing_keypair()?,
             delegate_updater,
+            shutdown,
         })
     }
 
@@ -111,6 +114,38 @@ impl OrgService {
             .sign(response)
             .map_err(|_| Status::internal("response signing error"))
     }
+
+    async fn stream_org_routes_enable_disable(&self, oui: u64) -> Result<(), Status> {
+        let routes = list_routes(oui, &self.pool).await.map_err(|err| {
+            tracing::error!(org = oui, reason = ?err, "failed to list org routes for streaming update");
+            Status::internal(format!("error retrieving routes for updated org: {}", oui))
+        })?;
+        let timestamp = Utc::now().encode_timestamp();
+        let signer: Vec<u8> = self.signing_key.public_key().into();
+        for route in routes {
+            let route_id = route.id.clone();
+            let mut update = RouteStreamResV1 {
+                action: ActionV1::Add.into(),
+                data: Some(route_stream_res_v1::Data::Route(route.into())),
+                timestamp,
+                signer: signer.clone(),
+                signature: vec![],
+            };
+            update.signature = self.sign_response(&update.encode_to_vec())?;
+            if broadcast_update(update, self.route_update_tx.clone())
+                .await
+                .is_err()
+            {
+                tracing::info!(
+                    route_id,
+                    "all subscribers disconnected; org routes update incomplete"
+                );
+                break;
+            };
+            tracing::debug!(route_id, "route updated");
+        }
+        Ok(())
+    }
 }
 
 #[tonic::async_trait]
@@ -431,38 +466,9 @@ impl iot_config::Org for OrgService {
                     Status::internal(format!("org disable failed for: {}", request.oui))
                 })?;
 
-            let org_routes = list_routes(request.oui, &self.pool).await.map_err(|err| {
-                tracing::error!(
-                    org = request.oui,
-                    reason = ?err,
-                    "failed to list org routes for streaming disable update"
-                );
-                Status::internal(format!(
-                    "error retrieving routes for disabled org: {}",
-                    request.oui
-                ))
-            })?;
-
-            let timestamp = Utc::now().encode_timestamp();
-            let signer: Vec<u8> = self.signing_key.public_key().into();
-            for route in org_routes {
-                let route_id = route.id.clone();
-                let mut update = RouteStreamResV1 {
-                    action: ActionV1::Add.into(),
-                    data: Some(route_stream_res_v1::Data::Route(route.into())),
-                    timestamp,
-                    signer: signer.clone(),
-                    signature: vec![],
-                };
-                update.signature = self.sign_response(&update.encode_to_vec())?;
-                if self.route_update_tx.send(update).is_err() {
-                    tracing::info!(
-                        route_id = route_id,
-                        "all subscribers disconnected; route disable incomplete"
-                    );
-                    break;
-                };
-                tracing::debug!(route_id = route_id, "route disabled");
+            tokio::select! {
+                _ = self.shutdown.clone() => return Err(Status::unavailable("service shutting down")),
+                result = self.stream_org_routes_enable_disable(request.oui) => result?
             }
         }
 
@@ -499,38 +505,9 @@ impl iot_config::Org for OrgService {
                     Status::internal(format!("org enable failed for: {}", request.oui))
                 })?;
 
-            let org_routes = list_routes(request.oui, &self.pool).await.map_err(|err| {
-                tracing::error!(
-                    org = request.oui,
-                    reason = ?err,
-                    "failed to list routes for streaming enable update"
-                );
-                Status::internal(format!(
-                    "error retrieving routes for enabled org: {}",
-                    request.oui
-                ))
-            })?;
-
-            let timestamp = Utc::now().encode_timestamp();
-            let signer: Vec<u8> = self.signing_key.public_key().into();
-            for route in org_routes {
-                let route_id = route.id.clone();
-                let mut update = RouteStreamResV1 {
-                    action: ActionV1::Add.into(),
-                    data: Some(route_stream_res_v1::Data::Route(route.into())),
-                    timestamp,
-                    signer: signer.clone(),
-                    signature: vec![],
-                };
-                update.signature = self.sign_response(&update.encode_to_vec())?;
-                if self.route_update_tx.send(update).is_err() {
-                    tracing::info!(
-                        route_id = route_id,
-                        "all subscribers disconnected; route enable incomplete"
-                    );
-                    break;
-                };
-                tracing::debug!(route_id = route_id, "route enabled");
+            tokio::select! {
+                _ = self.shutdown.clone() => return Err(Status::unavailable("service shutting down")),
+                result = self.stream_org_routes_enable_disable(request.oui) => result?
             }
         }
 

iot_config/src/route.rs

@@ -137,28 +137,28 @@ pub async fn create_route(
 
     transaction.commit().await?;
 
-    if new_route.active && !new_route.locked {
-        let timestamp = Utc::now().encode_timestamp();
-        let signer = signing_key.public_key().into();
-        let mut update = proto::RouteStreamResV1 {
-            action: proto::ActionV1::Add.into(),
-            data: Some(proto::route_stream_res_v1::Data::Route(
-                new_route.clone().into(),
-            )),
-            timestamp,
-            signer,
-            signature: vec![],
-        };
-        _ = signing_key
-            .sign(&update.encode_to_vec())
-            .map_err(|err| tracing::error!("error signing route stream response: {err:?}"))
-            .and_then(|signature| {
-                update.signature = signature;
-                update_tx.send(update).map_err(|err| {
-                    tracing::warn!("error broadcasting route stream response: {err:?}")
-                })
-            });
+    let timestamp = Utc::now().encode_timestamp();
+    let signer = signing_key.public_key().into();
+    let mut update = proto::RouteStreamResV1 {
+        action: proto::ActionV1::Add.into(),
+        data: Some(proto::route_stream_res_v1::Data::Route(
+            new_route.clone().into(),
+        )),
+        timestamp,
+        signer,
+        signature: vec![],
     };
+    _ = futures::future::ready(signing_key.sign(&update.encode_to_vec()))
+        .map_err(|err| {
+            tracing::error!(error = ?err, "error signing route create");
+            anyhow!("error signing route create")
+        })
+        .and_then(|signature| {
+            update.signature = signature;
+            broadcast_update(update, update_tx)
+                .map_err(|_| anyhow!("failed broadcasting route create"))
+        })
+        .await;
 
     Ok(new_route)
 }
@@ -213,15 +213,17 @@ pub async fn update_route(
         signature: vec![],
     };
 
-    _ = signing_key
-        .sign(&update_res.encode_to_vec())
-        .map_err(|err| tracing::error!("error signing route stream response: {err:?}"))
+    _ = futures::future::ready(signing_key.sign(&update_res.encode_to_vec()))
+        .map_err(|err| {
+            tracing::error!(error = ?err, "error signing route update");
+            anyhow!("error signing route update")
+        })
         .and_then(|signature| {
             update_res.signature = signature;
-            update_tx
-                .send(update_res)
-                .map_err(|err| tracing::warn!("error broadcasting route stream response: {err:?}"))
-        });
+            broadcast_update(update_res, update_tx)
+                .map_err(|_| anyhow!("failed broadcasting route update"))
+        })
+        .await;
 
     Ok(updated_route)
 }
@@ -523,7 +525,6 @@ pub fn active_route_stream<'a>(
         select r.id, r.oui, r.net_id, r.max_copies, r.server_host, r.server_port, r.server_protocol_opts, r.active, r.ignore_empty_skf, o.locked
             from routes r
             join organizations o on r.oui = o.oui
-            where o.locked = false and r.active = true
             group by r.id, o.locked
         "#,
     )