Merge pull request #443 from blinklabs-io/feat/protocol-chainsync-server

feat: chainsync server support

Author: agaffney

protocol/chainsync/chainsync.go

@@ -112,16 +112,20 @@ type ChainSync struct {
 
 // Config is used to configure the ChainSync protocol instance
 type Config struct {
-	RollBackwardFunc RollBackwardFunc
-	RollForwardFunc  RollForwardFunc
-	IntersectTimeout time.Duration
-	BlockTimeout     time.Duration
-	PipelineLimit    int
+	RollBackwardFunc  RollBackwardFunc
+	RollForwardFunc   RollForwardFunc
+	FindIntersectFunc FindIntersectFunc
+	RequestNextFunc   RequestNextFunc
+	IntersectTimeout  time.Duration
+	BlockTimeout      time.Duration
+	PipelineLimit     int
 }
 
 // Callback function types
 type RollBackwardFunc func(common.Point, Tip) error
 type RollForwardFunc func(uint, interface{}, Tip) error
+type FindIntersectFunc func([]common.Point) (common.Point, Tip, error)
+type RequestNextFunc func() error
 
 // New returns a new ChainSync object
 func New(protoOptions protocol.ProtocolOptions, cfg *Config) *ChainSync {
@@ -169,6 +173,20 @@ func WithRollForwardFunc(rollForwardFunc RollForwardFunc) ChainSyncOptionFunc {
 	}
 }
 
+// WithFindIntersectFunc specifies the FindIntersect callback function
+func WithFindIntersectFunc(findIntersectFunc FindIntersectFunc) ChainSyncOptionFunc {
+	return func(c *Config) {
+		c.FindIntersectFunc = findIntersectFunc
+	}
+}
+
+// WithRequestNextFunc specifies the RequestNext callback function
+func WithRequestNextFunc(requestNextFunc RequestNextFunc) ChainSyncOptionFunc {
+	return func(c *Config) {
+		c.RequestNextFunc = requestNextFunc
+	}
+}
+
 // WithIntersectTimeout specifies the timeout for intersect operations
 func WithIntersectTimeout(timeout time.Duration) ChainSyncOptionFunc {
 	return func(c *Config) {

protocol/chainsync/client.go

@@ -252,7 +252,7 @@ func (c *Client) handleAwaitReply() error {
 }
 
 func (c *Client) handleRollForward(msgGeneric protocol.Message) error {
-	if c.config.RollForwardFunc == nil && !c.wantFirstBlock {
+	if (c.config == nil || c.config.RollForwardFunc == nil) && !c.wantFirstBlock {
 		return fmt.Errorf(
 			"received chain-sync RollForward message but no callback function is defined",
 		)
@@ -367,6 +367,6 @@ func (c *Client) handleIntersectNotFound(msgGeneric protocol.Message) error {
 		msgIntersectNotFound := msgGeneric.(*MsgIntersectNotFound)
 		c.currentTipChan <- msgIntersectNotFound.Tip
 	}
-	c.intersectResultChan <- IntersectNotFoundError{}
+	c.intersectResultChan <- IntersectNotFoundError
 	return nil
 }

protocol/chainsync/error.go

@@ -14,18 +14,10 @@
 
 package chainsync
 
-import (
-	"fmt"
-)
+import "errors"
 
-// IntersectNotFoundError represents a failure to find a chain intersection
-type IntersectNotFoundError struct {
-}
-
-func (e IntersectNotFoundError) Error() string {
-	return "chain intersection not found"
-}
+var IntersectNotFoundError = errors.New("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")
+var StopSyncProcessError = errors.New("stop sync process")

protocol/chainsync/server.go

@@ -16,7 +16,10 @@ package chainsync
 
 import (
 	"fmt"
+
+	"github.com/blinklabs-io/gouroboros/ledger"
 	"github.com/blinklabs-io/gouroboros/protocol"
+	"github.com/blinklabs-io/gouroboros/protocol/common"
 )
 
 // Server implements the ChainSync server
@@ -54,6 +57,31 @@ func NewServer(protoOptions protocol.ProtocolOptions, cfg *Config) *Server {
 	return s
 }
 
+func (s *Server) RollBackward(point common.Point, tip Tip) error {
+	msg := NewMsgRollBackward(point, tip)
+	return s.SendMessage(msg)
+}
+
+func (s *Server) RollForward(blockType uint, blockData []byte, tip Tip) error {
+	if s.Mode() == protocol.ProtocolModeNodeToNode {
+		eraId := ledger.BlockToBlockHeaderTypeMap[blockType]
+		msg := NewMsgRollForwardNtN(
+			eraId,
+			0,
+			blockData,
+			tip,
+		)
+		return s.SendMessage(msg)
+	} else {
+		msg := NewMsgRollForwardNtC(
+			blockType,
+			blockData,
+			tip,
+		)
+		return s.SendMessage(msg)
+	}
+}
+
 func (s *Server) messageHandler(msg protocol.Message) error {
 	var err error
 	switch msg.Type() {
@@ -74,12 +102,34 @@ func (s *Server) messageHandler(msg protocol.Message) error {
 }
 
 func (s *Server) handleRequestNext(msg protocol.Message) error {
-	// TODO
-	return nil
+	if s.config == nil || s.config.RequestNextFunc == nil {
+		return fmt.Errorf(
+			"received chain-sync RequestNext message but no callback function is defined",
+		)
+	}
+	return s.config.RequestNextFunc()
 }
 
 func (s *Server) handleFindIntersect(msg protocol.Message) error {
-	// TODO
+	if s.config == nil || s.config.FindIntersectFunc == nil {
+		return fmt.Errorf(
+			"received chain-sync FindIntersect message but no callback function is defined",
+		)
+	}
+	msgFindIntersect := msg.(*MsgFindIntersect)
+	point, tip, err := s.config.FindIntersectFunc(msgFindIntersect.Points)
+	if err != nil {
+		if err == IntersectNotFoundError {
+			msgResp := NewMsgIntersectNotFound(tip)
+			if err := s.SendMessage(msgResp); err != nil {
+				return err
+			}
+		}
+	}
+	msgResp := NewMsgIntersectFound(point, tip)
+	if err := s.SendMessage(msgResp); err != nil {
+		return err
+	}
 	return nil
 }