fix(pipeline): pipeline waits on stale tip (#380)

Author: kariy

crates/sync/pipeline/src/lib.rs

@@ -354,7 +354,6 @@ impl Pipeline {
                 result = self.run_loop() => {
                     if let Err(error) = result {
                         error!(target: "pipeline", %error, "Pipeline finished due to error.");
-                        break;
                     }
                 }
             }
@@ -544,21 +543,25 @@ impl Pipeline {
                 }
             } else {
                 info!(target: "pipeline", "Waiting to receive new tip.");
-            }
+                self.cmd_rx.changed().await.map_err(|_| Error::CommandChannelClosed)?;
 
-            if let Some(PipelineCommand::SetTip(new_tip)) = *self
-                .cmd_rx
-                .wait_for(|c| matches!(c, &Some(PipelineCommand::SetTip(_))))
-                .await
-                .map_err(|_| Error::CommandChannelClosed)?
-            {
-                info!(target: "pipeline", tip = %new_tip, "A new tip has been set.");
-                self.tip = Some(new_tip);
-                self.metrics.set_sync_target(new_tip);
+                match *self.cmd_rx.borrow_and_update() {
+                    Some(PipelineCommand::SetTip(new_tip)) => {
+                        info!(target: "pipeline", tip = %new_tip, "A new tip has been set.");
+                        self.tip = Some(new_tip);
+                        self.metrics.set_sync_target(new_tip);
+                    }
+
+                    Some(PipelineCommand::Stop) => break,
+
+                    _ => {}
+                }
             }
 
             yield_now().await;
         }
+
+        Ok(())
     }
 }