feat: multicast

cmd/arc/services/blocktx.go

@@ -1,6 +1,7 @@
 package cmd
 
 import (
+	"errors"
 	"fmt"
 	"log/slog"
 	"os"
@@ -13,11 +14,13 @@ import (
 	"github.com/bitcoin-sv/arc/internal/message_queue/nats/client/nats_jetstream"
 	"github.com/bitcoin-sv/arc/internal/message_queue/nats/nats_connection"
 	"github.com/bitcoin-sv/arc/internal/tracing"
+	"github.com/libsv/go-p2p/wire"
 
 	"github.com/bitcoin-sv/arc/config"
 	"github.com/bitcoin-sv/arc/internal/blocktx"
 	"github.com/bitcoin-sv/arc/internal/blocktx/bcnet"
 	"github.com/bitcoin-sv/arc/internal/blocktx/bcnet/blocktx_p2p"
+	"github.com/bitcoin-sv/arc/internal/blocktx/bcnet/mcast"
 	"github.com/bitcoin-sv/arc/internal/blocktx/store"
 	"github.com/bitcoin-sv/arc/internal/blocktx/store/postgresql"
 	"github.com/bitcoin-sv/arc/internal/p2p"
@@ -37,13 +40,14 @@ func StartBlockTx(logger *slog.Logger, arcConfig *config.ArcConfig) (func(), err
 	btxConfig := arcConfig.Blocktx
 
 	var (
-		blockStore   store.BlocktxStore
-		mqClient     blocktx.MessageQueueClient
-		processor    *blocktx.Processor
-		pm           *p2p.PeerManager
-		server       *blocktx.Server
-		healthServer *grpc_opts.GrpcServer
-		workers      *blocktx.BackgroundWorkers
+		blockStore    store.BlocktxStore
+		mqClient      blocktx.MessageQueueClient
+		processor     *blocktx.Processor
+		pm            *p2p.PeerManager
+		mcastListener *mcast.Listener
+		server        *blocktx.Server
+		healthServer  *grpc_opts.GrpcServer
+		workers       *blocktx.BackgroundWorkers
 
 		err error
 	)
@@ -71,15 +75,10 @@ func StartBlockTx(logger *slog.Logger, arcConfig *config.ArcConfig) (func(), err
 
 	stopFn := func() {
 		logger.Info("Shutting down blocktx")
-		disposeBlockTx(logger, server, processor, pm, mqClient, blockStore, healthServer, workers, shutdownFns)
+		disposeBlockTx(logger, server, processor, pm, mcastListener, mqClient, blockStore, healthServer, workers, shutdownFns)
 		logger.Info("Shutdown complete")
 	}
 
-	network, err := config.GetNetwork(arcConfig.Network)
-	if err != nil {
-		return nil, err
-	}
-
 	blockStore, err = NewBlocktxStore(logger, btxConfig.Db, arcConfig.Tracing)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create blocktx store: %v", err)
@@ -143,58 +142,18 @@ func StartBlockTx(logger *slog.Logger, arcConfig *config.ArcConfig) (func(), err
 	err = processor.Start(arcConfig.Prometheus.IsEnabled())
 	if err != nil {
 		stopFn()
-		return nil, fmt.Errorf("failed to start peer handler: %v", err)
-	}
-
-	// p2p global setting
-	p2p.SetExcessiveBlockSize(maximumBlockSize)
-
-	pmOpts := []p2p.PeerManagerOptions{}
-	if arcConfig.Blocktx.MonitorPeers {
-		pmOpts = append(pmOpts, p2p.WithRestartUnhealthyPeers())
-	}
-
-	pm = p2p.NewPeerManager(logger.With(slog.String("module", "peer-mng")), network, pmOpts...)
-	peers := make([]p2p.PeerI, len(arcConfig.Broadcasting.Unicast.Peers))
-
-	peerHandler := blocktx_p2p.NewMsgHandler(logger, blockRequestCh, blockProcessCh)
-
-	peerOpts := []p2p.PeerOptions{
-		p2p.WithMaximumMessageSize(maximumBlockSize),
-		p2p.WithPingInterval(30*time.Second, 1*time.Minute),
-		p2p.WithReadBufferSize(arcConfig.Blocktx.P2pReadBufferSize),
-	}
-
-	if version.Version != "" {
-		peerOpts = append(peerOpts, p2p.WithUserAgent("ARC", version.Version))
+		return nil, fmt.Errorf("failed to start prometheus: %v", err)
 	}
 
-	for i, peerSetting := range arcConfig.Broadcasting.Unicast.Peers {
-		peerURL, err := peerSetting.GetP2PUrl()
-		if err != nil {
-			stopFn()
-			return nil, fmt.Errorf("error getting peer url: %v", err)
-		}
-
-		peer := p2p.NewPeer(logger.With(slog.String("module", "peer")), peerHandler, peerURL, network, peerOpts...)
-		ok := peer.Connect()
-		if !ok {
-			stopFn()
-			return nil, fmt.Errorf("error creating peer %s: %v", peerURL, err)
-		}
-
-		err = pm.AddPeer(peer)
-		if err != nil {
-			stopFn()
-			return nil, fmt.Errorf("error adding peer: %v", err)
-		}
-
-		peers[i] = peer
+	pm, mcastListener, err = setupBcNetworkCommunication(logger, arcConfig, blockStore, blockRequestCh, blockProcessCh)
+	if err != nil {
+		stopFn()
+		return nil, fmt.Errorf("failed to establish connection with network: %v", err)
 	}
 
 	if btxConfig.FillGaps != nil && btxConfig.FillGaps.Enabled {
 		workers = blocktx.NewBackgroundWorkers(blockStore, logger)
-		workers.StartFillGaps(peers, btxConfig.FillGaps.Interval, btxConfig.RecordRetentionDays, blockRequestCh)
+		workers.StartFillGaps(pm.GetPeers(), btxConfig.FillGaps.Interval, btxConfig.RecordRetentionDays, blockRequestCh)
 	}
 
 	server, err = blocktx.NewServer(arcConfig.Prometheus.Endpoint, arcConfig.GrpcMessageSize, logger,
@@ -250,8 +209,151 @@ func NewBlocktxStore(logger *slog.Logger, dbConfig *config.DbConfig, tracingConf
 	return s, err
 }
 
+// setupBcNetworkCommunication initializes the Bloctx blockchain network communication layer, configuring it
+// to operate in either classic (P2P-only) or hybrid (P2P and multicast) mode.
+//
+// Parameters:
+// - `l *slog.Logger`: Logger instance for logging events.
+// - `arcConfig *config.ArcConfig`: Configuration object containing blockchain network settings.
+// - `store store.BlocktxStore`: A storage interface for blockchain transactions.
+// - `blockRequestCh chan<- blocktx_p2p.BlockRequest`: Channel for handling block requests.
+// - `blockProcessCh chan<- *bcnet.BlockMessage`: Channel for processing block messages.
+//
+// Returns:
+// - `manager *p2p.PeerManager`: Manages P2P peers.
+// - `mcastListener *mcast.Listener`: Handles multicast communication, or `nil` if not in hybrid mode.
+// - `err error`: Error if any issue occurs during setup.
+//
+// Key Details:
+// - **Mode Handling**:
+//   - `"classic"` mode uses `blocktx_p2p.NewMsgHandler` for P2P communication only.
+//   - `"hybrid"` mode uses `blocktx_p2p.NewHybridMsgHandler` for P2P and multicast communication.
+//
+// - **Error Cleanup**: Ensures resources like peers and multicast listeners are properly cleaned up on errors.
+// - **Peer Management**: Connects to configured peers and initializes a PeerManager for handling P2P connections.
+// - **Multicast Communication**: In hybrid mode, joins a multicast group for blockchain updates and uses correct P2P message handler.
+//
+// Message Handlers:
+// - `blocktx_p2p.NewMsgHandler`: Used in classic mode, handles all blockchain communication exclusively via P2P.
+// - `blocktx_p2p.NewHybridMsgHandler`: Used in hybrid mode, seamlessly integrates P2P communication with multicast group updates.
+func setupBcNetworkCommunication(l *slog.Logger, arcConfig *config.ArcConfig, store store.BlocktxStore, blockRequestCh chan<- blocktx_p2p.BlockRequest, blockProcessCh chan<- *bcnet.BlockMessage) (manager *p2p.PeerManager, mcastListener *mcast.Listener, err error) {
+	defer func() {
+		// cleanup on error
+		if err == nil {
+			return
+		}
+
+		if manager != nil {
+			manager.Shutdown()
+		}
+
+		if mcastListener != nil {
+			mcastListener.Disconnect()
+		}
+	}()
+
+	// p2p global setting
+	p2p.SetExcessiveBlockSize(maximumBlockSize)
+
+	cfg := arcConfig.Blocktx.BlockchainNetwork
+	network, err := config.GetNetwork(cfg.Network)
+	if err != nil {
+		return
+	}
+
+	var msgHandler p2p.MessageHandlerI
+
+	if cfg.Mode == "classic" {
+		msgHandler = blocktx_p2p.NewMsgHandler(l, blockRequestCh, blockProcessCh)
+	} else if cfg.Mode == "hybrid" {
+		l.Info("!!! Blocktx will communicate with blockchain in HYBRID mode (via p2p and multicast groups) !!!")
+		msgHandler = blocktx_p2p.NewHybridMsgHandler(l, blockProcessCh)
+	} else {
+		return nil, nil, fmt.Errorf("unsupported communication type: %s", cfg.Mode)
+	}
+
+	// connect to peers
+	var managerOpts []p2p.PeerManagerOptions
+	if arcConfig.Blocktx.MonitorPeers {
+		managerOpts = append(managerOpts, p2p.WithRestartUnhealthyPeers())
+	}
+
+	manager = p2p.NewPeerManager(l.With(slog.String("module", "peer-mng")), network, managerOpts...)
+	peers, err := connectToPeers(l, network, msgHandler, cfg.Peers, p2p.WithMaximumMessageSize(maximumBlockSize))
+	if err != nil {
+		return
+	}
+
+	for _, p := range peers {
+		if err = manager.AddPeer(p); err != nil {
+			return
+		}
+	}
+
+	// connect to mcast
+	if cfg.Mode == "hybrid" {
+		if cfg.Mcast == nil {
+			return manager, mcastListener, errors.New("mcast config is required")
+		}
+
+		// TODO: add net interfaces
+		mcastListener = mcast.NewMcastListener(l, cfg.Mcast.McastBlock.Address, network, store, blockProcessCh)
+		ok := mcastListener.Connect()
+		if !ok {
+			return manager, nil, fmt.Errorf("error connecting to mcast %s: %w", cfg.Mcast.McastBlock, err)
+		}
+	}
+
+	return
+}
+
+func connectToPeers(l *slog.Logger, network wire.BitcoinNet, msgHandler p2p.MessageHandlerI, peersConfig []*config.PeerConfig, additionalOpts ...p2p.PeerOptions) (peers []*p2p.Peer, err error) {
+	defer func() {
+		// cleanup on error
+		if err == nil {
+			return
+		}
+
+		for _, p := range peers {
+			p.Shutdown()
+		}
+	}()
+
+	opts := []p2p.PeerOptions{
+		p2p.WithPingInterval(30*time.Second, 2*time.Minute),
+	}
+
+	if version.Version != "" {
+		opts = append(opts, p2p.WithUserAgent("ARC", version.Version))
+	}
+
+	opts = append(opts, additionalOpts...)
+
+	for _, settings := range peersConfig {
+		url, err := settings.GetP2PUrl()
+		if err != nil {
+			return nil, fmt.Errorf("error getting peer url: %w", err)
+		}
+
+		p := p2p.NewPeer(
+			l.With(slog.String("module", "peer")),
+			msgHandler,
+			url,
+			network,
+			opts...)
+		ok := p.Connect()
+		if !ok {
+			return nil, fmt.Errorf("error connecting peer %s: %w", url, err)
+		}
+
+		peers = append(peers, p)
+	}
+
+	return
+}
+
 func disposeBlockTx(l *slog.Logger, server *blocktx.Server, processor *blocktx.Processor,
-	pm *p2p.PeerManager, mqClient blocktx.MessageQueueClient,
+	pm *p2p.PeerManager, mcastListener *mcast.Listener, mqClient blocktx.MessageQueueClient,
 	store store.BlocktxStore, healthServer *grpc_opts.GrpcServer, workers *blocktx.BackgroundWorkers,
 	shutdownFns []func(),
 ) {
@@ -277,6 +379,9 @@ func disposeBlockTx(l *slog.Logger, server *blocktx.Server, processor *blocktx.P
 	if pm != nil {
 		pm.Shutdown()
 	}
+	if mcastListener != nil {
+		mcastListener.Disconnect()
+	}
 	if mqClient != nil {
 		mqClient.Shutdown()
 	}

cmd/arc/services/callbacker.go

@@ -27,6 +27,7 @@ import (
 	"log/slog"
 	"net/http"
 	"os"
+	"reflect"
 	"time"
 
 	"github.com/bitcoin-sv/arc/config"
@@ -195,7 +196,7 @@ func dispose(l *slog.Logger, server *callbacker.Server, workers *callbacker.Back
 		processor.GracefulStop()
 	}
 
-	if mqClient != nil {
+	if mqClient != nil && !(reflect.ValueOf(mqClient).IsNil()) {
 		mqClient.Shutdown()
 	}