Merge pull request #237 from blinklabs-io/feat/stop-chainsync

feat: method to stop chainsync process

Author: agaffney

protocol/chainsync/client.go

@@ -147,9 +147,12 @@ func (c *Client) Sync(intersectPoints []common.Point) error {
 func (c *Client) syncLoop() {
 	for {
 		// Wait for a block to be received
-		if _, ok := <-c.readyForNextBlockChan; !ok {
+		if ready, ok := <-c.readyForNextBlockChan; !ok {
 			// Channel is closed, which means we're shutting down
 			return
+		} else if !ready {
+			// Sync was cancelled
+			return
 		}
 		c.busyMutex.Lock()
 		// Request the next block
@@ -172,10 +175,7 @@ func (c *Client) handleRollForward(msgGeneric protocol.Message) error {
 	if c.config.RollForwardFunc == nil {
 		return fmt.Errorf("received chain-sync RollForward message but no callback function is defined")
 	}
-	// Signal that we're ready for the next block after we finish handling this one
-	defer func() {
-		c.readyForNextBlockChan <- true
-	}()
+	var callbackErr error
 	if c.Mode() == protocol.ProtocolModeNodeToNode {
 		msg := msgGeneric.(*MsgRollForwardNtN)
 		var blockHeader interface{}
@@ -206,16 +206,23 @@ func (c *Client) handleRollForward(msgGeneric protocol.Message) error {
 			}
 		}
 		// Call the user callback function
-		return c.config.RollForwardFunc(blockType, blockHeader, msg.Tip)
+		callbackErr = c.config.RollForwardFunc(blockType, blockHeader, msg.Tip)
 	} else {
 		msg := msgGeneric.(*MsgRollForwardNtC)
 		blk, err := ledger.NewBlockFromCbor(msg.BlockType(), msg.BlockCbor())
 		if err != nil {
 			return err
 		}
 		// Call the user callback function
-		return c.config.RollForwardFunc(msg.BlockType(), blk, msg.Tip)
+		callbackErr = c.config.RollForwardFunc(msg.BlockType(), blk, msg.Tip)
 	}
+	if callbackErr == StopSyncProcessError {
+		// Signal that we're cancelling the sync
+		c.readyForNextBlockChan <- false
+	}
+	// Signal that we're ready for the next block
+	c.readyForNextBlockChan <- true
+	return nil
 }
 
 func (c *Client) handleRollBackward(msgGeneric protocol.Message) error {
@@ -228,7 +235,14 @@ func (c *Client) handleRollBackward(msgGeneric protocol.Message) error {
 		c.readyForNextBlockChan <- true
 	}()
 	// Call the user callback function
-	return c.config.RollBackwardFunc(msg.Point, msg.Tip)
+	callbackErr := c.config.RollBackwardFunc(msg.Point, msg.Tip)
+	if callbackErr == StopSyncProcessError {
+		// Signal that we're cancelling the sync
+		c.readyForNextBlockChan <- false
+	}
+	// Signal that we're ready for the next block
+	c.readyForNextBlockChan <- true
+	return nil
 }
 
 func (c *Client) handleIntersectFound(msgGeneric protocol.Message) error {

protocol/chainsync/error.go

@@ -1,9 +1,17 @@
 package chainsync
 
+import (
+	"fmt"
+)
+
 // IntersectNotFoundError represents a failure to find a chain intersection
 type IntersectNotFoundError struct {
 }
 
 func (e IntersectNotFoundError) Error() string {
 	return "chain intersection not found"
 }
+
+// StopChainSync is used as a special return value from a RollForward or RollBackward handler function
+// to signify that the sync process should be stopped
+var StopSyncProcessError = fmt.Errorf("stop sync process")