Improve poller state management

src/bundle/mod.rs

@@ -154,22 +154,45 @@ impl Unbundler {
   async fn poll(&self) -> Result<()> {
     info!("Unbundler: Checking for updates...");
 
-    let mut state = self.state.write().map_err(|_| Error::LockWrite)?;
+    // Update state on a spearate scope to avoid holding a write lock while
+    // the poller runs.
+    //
+    // Lock will be released when `initial_state` goes out of scope.
+    let active_bundle = {
+      let mut initial_state = self.state.write().map_err(|_| Error::LockWrite)?;
+
+      let active_bundle = initial_state.active_bundle.clone();
+      if let None = active_bundle {
+        initial_state.status = UnbundlerStatus::InitialPolling;
+      } else {
+        initial_state.status = UnbundlerStatus::Polling;
+      }
 
-    if let None = state.active_bundle {
-      state.status = UnbundlerStatus::InitialPolling;
-    } else {
-      state.status = UnbundlerStatus::Polling;
-    }
+      active_bundle
+    };
 
-    let result = self
-      .poller
-      .poll(&state.active_bundle)
-      .await
-      .context(PollError)?;
+    // Invoke the poller.
+    let result = match self.poller.poll(&active_bundle).await {
+      Err(err) => {
+        let mut state = self.state.write().map_err(|_| Error::LockWrite)?;
+
+        // Rollback status if the poll fails.
+        if let None = active_bundle {
+          state.status = UnbundlerStatus::Initialized;
+        } else {
+          state.status = UnbundlerStatus::Ready;
+        }
+
+        Err(err)
+      }
+      res => res,
+    }
+    .context(PollError)?;
 
     match result {
       poller::PollResult::Skip => {
+        let mut state = self.state.write().map_err(|_| Error::LockWrite)?;
+
         state.status = UnbundlerStatus::Ready;
 
         info!("Unbundler: No updates from poller.");
@@ -177,6 +200,8 @@ impl Unbundler {
         Ok(())
       }
       poller::PollResult::StaticUpdateReady { etag, path } => {
+        let mut state = self.state.write().map_err(|_| Error::LockWrite)?;
+
         // Replacing active bundle.
         state.active_bundle = Some(Bundle { etag });
         state.staging_bundle = None;
@@ -185,11 +210,12 @@ impl Unbundler {
         let mut serve_dir = self.serve_dir.write().map_err(|_| Error::LockWrite)?;
 
         serve_dir.replace(path);
-        // TODO: Update path.
 
         Ok(())
       }
       poller::PollResult::UpdateReady { etag } => {
+        let mut state = self.state.write().map_err(|_| Error::LockWrite)?;
+
         if state.rundir.subdir_exists(&etag).context(SubDirError)? {
           warn!("Unbundler: Skipping update. Subdir already exists.");