darkirc: monitor network disconnection and resync accordingly

bin/darkirc/src/main.rs

@@ -26,7 +26,7 @@ use darkfi::{
         jsonrpc::JsonSubscriber,
         server::{listen_and_serve, RequestHandler},
     },
-    system::{sleep, StoppableTask, StoppableTaskPtr},
+    system::{sleep, StoppableTask, StoppableTaskPtr, Subscription},
     util::path::{expand_path, get_config_path},
     Error, Result,
 };
@@ -401,32 +401,22 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
     info!("Starting P2P network");
     p2p.clone().start().await?;
 
-    let comms_timeout = p2p.settings().read().await.outbound_connect_timeout;
+    // Initial DAG sync
+    sync_task(&p2p, &event_graph, args.skip_dag_sync).await?;
 
-    loop {
-        if p2p.is_connected() {
-            info!("Got peer connection");
-            // We'll attempt to sync for ever
-            if !args.skip_dag_sync {
-                info!("Syncing event DAG");
-                match event_graph.dag_sync().await {
-                    Ok(()) => break,
-                    Err(e) => {
-                        // TODO: Maybe at this point we should prune or something?
-                        // TODO: Or maybe just tell the user to delete the DAG from FS.
-                        error!("Failed syncing DAG ({}), retrying in {}s...", e, comms_timeout);
-                        sleep(comms_timeout).await;
-                    }
-                }
-            } else {
-                *event_graph.synced.write().await = true;
-                break
+    // Stoppable task to monitor network and resync on disconnect.
+    let sync_mon_task = StoppableTask::new();
+    sync_mon_task.clone().start(
+        sync_and_monitor(p2p.clone(), event_graph.clone(), args.skip_dag_sync),
+        |res| async move {
+            match res {
+                Ok(()) | Err(Error::DetachedTaskStopped) => { /* TODO: */ }
+                Err(e) => error!("Failed sync task: {}", e),
             }
-        } else {
-            info!("Waiting for some P2P connections...");
-            sleep(comms_timeout).await;
-        }
-    }
+        },
+        Error::DetachedTaskStopped,
+        ex.clone(),
+    );
 
     // Signal handling for graceful termination.
     let (signals_handler, signals_task) = SignalHandler::new(ex)?;
@@ -452,3 +442,64 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
     info!("Shut down successfully");
     Ok(())
 }
+
+/// Async task to monitor network disconnections.
+async fn monitor_network(subscription: &Subscription<Error>) -> Result<()> {
+    Err(subscription.receive().await)
+}
+
+/// Async task to endlessly try to sync DAG, returns Ok if done.
+async fn sync_task(p2p: &P2pPtr, event_graph: &EventGraphPtr, skip_dag_sync: bool) -> Result<()> {
+    let comms_timeout = p2p.settings().read().await.outbound_connect_timeout;
+
+    loop {
+        if p2p.is_connected() {
+            info!("Got peer connection");
+            // We'll attempt to sync for ever
+            if !skip_dag_sync {
+                info!("Syncing event DAG");
+                match event_graph.dag_sync().await {
+                    Ok(()) => break,
+                    Err(e) => {
+                        // TODO: Maybe at this point we should prune or something?
+                        // TODO: Or maybe just tell the user to delete the DAG from FS.
+                        error!("Failed syncing DAG ({}), retrying in {}s...", e, comms_timeout);
+                        sleep(comms_timeout).await;
+                    }
+                }
+            } else {
+                *event_graph.synced.write().await = true;
+                break;
+            }
+        } else {
+            info!("Waiting for some P2P connections...");
+            sleep(comms_timeout).await;
+        }
+    }
+
+    Ok(())
+}
+
+/// Async task to monitor the network and force resync on disconnections
+async fn sync_and_monitor(
+    p2p: P2pPtr,
+    event_graph: EventGraphPtr,
+    skip_dag_sync: bool,
+) -> Result<()> {
+    loop {
+        let net_subscription = p2p.hosts().subscribe_disconnect().await;
+        let result = monitor_network(&net_subscription).await;
+        net_subscription.unsubscribe().await;
+
+        match result {
+            Ok(_) => return Ok(()),
+            Err(Error::NetworkNotConnected) => {
+                // Sync node again
+                info!("Network disconnection detected, resyncing...");
+                *event_graph.synced.write().await = false;
+                sync_task(&p2p, &event_graph, skip_dag_sync).await?;
+            }
+            Err(e) => return Err(e),
+        }
+    }
+}