feat: chainsync bulk mode

Fixes #18

Author: agaffney

input/chainsync/chainsync.go

@@ -23,6 +23,7 @@ import (
 
 	ouroboros "github.com/blinklabs-io/gouroboros"
 	"github.com/blinklabs-io/gouroboros/ledger"
+	"github.com/blinklabs-io/gouroboros/protocol/blockfetch"
 	ochainsync "github.com/blinklabs-io/gouroboros/protocol/chainsync"
 	ocommon "github.com/blinklabs-io/gouroboros/protocol/common"
 )
@@ -34,13 +35,16 @@ type ChainSync struct {
 	address          string
 	socketPath       string
 	ntcTcp           bool
+	bulkMode         bool
 	intersectTip     bool
 	intersectPoints  []ocommon.Point
 	includeCbor      bool
 	statusUpdateFunc StatusUpdateFunc
 	status           *ChainSyncStatus
 	errorChan        chan error
 	eventChan        chan event.Event
+	bulkRangeStart   ocommon.Point
+	bulkRangeEnd     ocommon.Point
 }
 
 type ChainSyncStatus struct {
@@ -76,15 +80,26 @@ func (c *ChainSync) Start() error {
 	if c.oConn.BlockFetch() != nil {
 		c.oConn.BlockFetch().Client.Start()
 	}
-	if c.intersectTip {
-		tip, err := c.oConn.ChainSync().Client.GetCurrentTip()
+	if c.bulkMode && !c.intersectTip && c.oConn.BlockFetch() != nil {
+		var err error
+		c.bulkRangeStart, c.bulkRangeEnd, err = c.oConn.ChainSync().Client.GetAvailableBlockRange(c.intersectPoints)
 		if err != nil {
 			return err
 		}
-		c.intersectPoints = []ocommon.Point{tip.Point}
-	}
-	if err := c.oConn.ChainSync().Client.Sync(c.intersectPoints); err != nil {
-		return err
+		if err := c.oConn.BlockFetch().Client.GetBlockRange(c.bulkRangeStart, c.bulkRangeEnd); err != nil {
+			return err
+		}
+	} else {
+		if c.intersectTip {
+			tip, err := c.oConn.ChainSync().Client.GetCurrentTip()
+			if err != nil {
+				return err
+			}
+			c.intersectPoints = []ocommon.Point{tip.Point}
+		}
+		if err := c.oConn.ChainSync().Client.Sync(c.intersectPoints); err != nil {
+			return err
+		}
 	}
 	return nil
 }
@@ -158,6 +173,11 @@ func (c *ChainSync) setupConnection() error {
 				ochainsync.WithRollBackwardFunc(c.handleRollBackward),
 			),
 		),
+		ouroboros.WithBlockFetchConfig(
+			blockfetch.NewConfig(
+				blockfetch.WithBlockFunc(c.handleBlockFetchBlock),
+			),
+		),
 	)
 	if err != nil {
 		return err
@@ -208,6 +228,19 @@ func (c *ChainSync) handleRollForward(blockType uint, blockData interface{}, tip
 	return nil
 }
 
+func (c *ChainSync) handleBlockFetchBlock(block ledger.Block) error {
+	evt := event.New("chainsync.block", time.Now(), NewBlockEvent(block, c.includeCbor))
+	c.eventChan <- evt
+	c.updateStatus(block.SlotNumber(), block.BlockNumber(), block.Hash(), c.bulkRangeEnd.Slot, hex.EncodeToString(c.bulkRangeEnd.Hash))
+	// Start normal chain-sync if we've reached the last block of our bulk range
+	if block.SlotNumber() == c.bulkRangeEnd.Slot {
+		if err := c.oConn.ChainSync().Client.Sync(c.intersectPoints); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (c *ChainSync) updateStatus(slotNumber uint64, blockNumber uint64, blockHash string, tipSlotNumber uint64, tipBlockHash string) {
 	c.status.SlotNumber = slotNumber
 	c.status.BlockNumber = blockNumber

input/chainsync/options.go

@@ -83,3 +83,10 @@ func WithStatusUpdateFunc(statusUpdateFunc StatusUpdateFunc) ChainSyncOptionFunc
 		c.statusUpdateFunc = statusUpdateFunc
 	}
 }
+
+// WithBulkMode specifies whether to use the "bulk" sync mode with NtN (node-to-node). This should only be used against your own nodes for resource usage reasons
+func WithBulkMode(bulkMode bool) ChainSyncOptionFunc {
+	return func(c *ChainSync) {
+		c.bulkMode = bulkMode
+	}
+}

input/chainsync/plugin.go

@@ -15,7 +15,13 @@
 package chainsync
 
 import (
+	"encoding/hex"
+	"strconv"
+	"strings"
+
 	"github.com/blinklabs-io/snek/plugin"
+
+	ocommon "github.com/blinklabs-io/gouroboros/protocol/common"
 )
 
 var cmdlineOptions struct {
@@ -24,6 +30,7 @@ var cmdlineOptions struct {
 	address        string
 	socketPath     string
 	ntcTcp         bool
+	bulkMode       bool
 	intersectTip   bool
 	intersectPoint string
 	includeCbor    bool
@@ -74,14 +81,27 @@ func init() {
 					DefaultValue: false,
 					Dest:         &(cmdlineOptions.ntcTcp),
 				},
+				{
+					Name:         "bulk-mode",
+					Type:         plugin.PluginOptionTypeBool,
+					Description:  "use the 'bulk' sync mode with NtN (node-to-node). This should only be used against your own nodes for resource usage reasons",
+					DefaultValue: false,
+					Dest:         &(cmdlineOptions.bulkMode),
+				},
 				{
 					Name:         "intersect-tip",
 					Type:         plugin.PluginOptionTypeBool,
 					Description:  "start syncing at the chain tip (defaults to chain genesis)",
 					DefaultValue: true,
 					Dest:         &(cmdlineOptions.intersectTip),
 				},
-				// TODO: intersect-point
+				{
+					Name:         "intersect-point",
+					Type:         plugin.PluginOptionTypeString,
+					Description:  "start syncing at the specified chain point(s) in '<slot>.<hash>' format",
+					DefaultValue: "",
+					Dest:         &(cmdlineOptions.intersectPoint),
+				},
 				{
 					Name:         "include-cbor",
 					Type:         plugin.PluginOptionTypeBool,
@@ -95,15 +115,48 @@ func init() {
 }
 
 func NewFromCmdlineOptions() plugin.Plugin {
-	p := New(
+	opts := []ChainSyncOptionFunc{
 		WithNetwork(cmdlineOptions.network),
 		WithNetworkMagic(uint32(cmdlineOptions.networkMagic)),
 		WithAddress(cmdlineOptions.address),
 		WithSocketPath(cmdlineOptions.socketPath),
 		WithNtcTcp(cmdlineOptions.ntcTcp),
-		WithIntersectTip(cmdlineOptions.intersectTip),
-		// TODO: WithIntersectPoints
+		WithBulkMode(cmdlineOptions.bulkMode),
 		WithIncludeCbor(cmdlineOptions.includeCbor),
-	)
+	}
+	if cmdlineOptions.intersectPoint != "" {
+		intersectPoints := []ocommon.Point{}
+		for _, point := range strings.Split(cmdlineOptions.intersectPoint, ",") {
+			intersectPointParts := strings.Split(point, ".")
+			if len(intersectPointParts) != 2 {
+				panic("invalid intersect point format")
+			}
+			intersectSlot, err := strconv.ParseUint(intersectPointParts[0], 10, 64)
+			if err != nil {
+				panic("invalid intersect point format")
+			}
+			intersectHashBytes, err := hex.DecodeString(intersectPointParts[1])
+			if err != nil {
+				panic("invalid intersect point format")
+			}
+			intersectPoints = append(
+				intersectPoints,
+				ocommon.Point{
+					Slot: intersectSlot,
+					Hash: intersectHashBytes[:],
+				},
+			)
+		}
+		opts = append(
+			opts,
+			WithIntersectPoints(intersectPoints),
+		)
+	} else {
+		opts = append(
+			opts,
+			WithIntersectTip(cmdlineOptions.intersectTip),
+		)
+	}
+	p := New(opts...)
 	return p
 }