Expand peer registry usage and improve reputation system for catchup

daemon/daemon.go

@@ -76,6 +76,7 @@ const (
 	loggerTransactions             = "txs"
 	loggerTxValidator              = "txval"
 	loggerUtxos                    = "utxos"
+	loggerAlert                    = "alert"
 
 	// Service names
 	serviceAlert                   = "alert"

daemon/daemon_services.go

@@ -423,25 +423,25 @@ func (d *Daemon) startRPCService(ctx context.Context, appSettings *settings.Sett
 		return err
 	}
 
-	blockAssemblyClient, err := blockassembly.NewClient(ctx, createLogger("ba"), appSettings)
+	blockAssemblyClient, err := GetBlockAssemblyClient(ctx, createLogger("rpc"), appSettings)
 	if err != nil {
 		return err
 	}
 
-	peerClient, err := peer.NewClient(ctx, createLogger("peer"), appSettings)
+	peerClient, err := peer.NewClient(ctx, createLogger("rpc"), appSettings)
 	if err != nil {
 		return err
 	}
 
-	p2pClient, err := p2p.NewClient(ctx, createLogger("p2p"), appSettings)
+	p2pClient, err := d.daemonStores.GetP2PClient(ctx, createLogger("rpc"), appSettings)
 	if err != nil {
 		return err
 	}
 
 	// Create block validation client for RPC service
 	var blockValidationClient blockvalidation.Interface
 
-	blockValidationClient, err = d.daemonStores.GetBlockValidationClient(ctx, createLogger("blockvalidation"), appSettings)
+	blockValidationClient, err = d.daemonStores.GetBlockValidationClient(ctx, createLogger("rpc"), appSettings)
 	if err != nil {
 		return err
 	}
@@ -471,7 +471,7 @@ func (d *Daemon) startRPCService(ctx context.Context, appSettings *settings.Sett
 	}
 
 	// Add the RPC service to the ServiceManager
-	if err := d.ServiceManager.AddService(serviceRPCFormal, rpcServer); err != nil {
+	if err = d.ServiceManager.AddService(serviceRPCFormal, rpcServer); err != nil {
 		return err
 	}
 
@@ -480,43 +480,43 @@ func (d *Daemon) startRPCService(ctx context.Context, appSettings *settings.Sett
 
 // startAlertService initializes and adds the Alert service to the ServiceManager.
 func (d *Daemon) startAlertService(ctx context.Context, appSettings *settings.Settings,
-	createLogger func(string) ulogger.Logger) error {
+	createLogger func(string) ulogger.Logger) (err error) {
+	var (
+		blockchainClient    blockchain.ClientI
+		utxoStore           utxo.Store
+		blockAssemblyClient blockassembly.ClientI
+		peerClient          peer.ClientI
+		p2pClient           p2p.ClientI
+	)
+
 	// Create the blockchain client for the Alert service
-	blockchainClient, err := d.daemonStores.GetBlockchainClient(
+	blockchainClient, err = d.daemonStores.GetBlockchainClient(
 		ctx, createLogger(loggerBlockchainClient), appSettings, serviceAlert,
 	)
 	if err != nil {
 		return err
 	}
 
 	// Create the UTXO store for the Alert service
-	var utxoStore utxo.Store
-
-	utxoStore, err = d.daemonStores.GetUtxoStore(ctx, createLogger(loggerUtxos), appSettings)
+	utxoStore, err = d.daemonStores.GetUtxoStore(ctx, createLogger(loggerAlert), appSettings)
 	if err != nil {
 		return err
 	}
 
 	// Create the block assembly client for the Alert service
-	var blockAssemblyClient *blockassembly.Client
-
-	blockAssemblyClient, err = blockassembly.NewClient(ctx, createLogger(loggerBlockAssembly), appSettings)
+	blockAssemblyClient, err = GetBlockAssemblyClient(ctx, createLogger(loggerAlert), appSettings)
 	if err != nil {
 		return err
 	}
 
 	// Create the peer client for the Alert service
-	var peerClient peer.ClientI
-
-	peerClient, err = peer.NewClient(ctx, createLogger(loggerPeerClient), appSettings)
+	peerClient, err = peer.NewClient(ctx, createLogger(loggerAlert), appSettings)
 	if err != nil {
 		return err
 	}
 
 	// Create the P2P client for the Alert service
-	var p2pClient p2p.ClientI
-
-	p2pClient, err = p2p.NewClient(ctx, createLogger(loggerP2P), appSettings)
+	p2pClient, err = d.daemonStores.GetP2PClient(ctx, createLogger(loggerAlert), appSettings)
 	if err != nil {
 		return err
 	}
@@ -752,6 +752,14 @@ func (d *Daemon) startValidationService(
 			return err
 		}
 
+		// Create the P2P client for the SubtreeValidation service
+		var p2pClient p2p.ClientI
+
+		p2pClient, err = d.daemonStores.GetP2PClient(ctx, createLogger(loggerP2P), appSettings)
+		if err != nil {
+			return err
+		}
+
 		// Create the SubtreeValidation service
 		var service *subtreevalidation.Server
 
@@ -766,6 +774,7 @@ func (d *Daemon) startValidationService(
 			blockchainClient,
 			subtreeConsumerClient,
 			txMetaConsumerClient,
+			p2pClient,
 		)
 		if err != nil {
 			return err
@@ -800,6 +809,14 @@ func (d *Daemon) startValidationService(
 			return err
 		}
 
+		// Create the P2P client for the BlockValidation service
+		var p2pClient p2p.ClientI
+
+		p2pClient, err = d.daemonStores.GetP2PClient(ctx, createLogger(loggerP2P), appSettings)
+		if err != nil {
+			return err
+		}
+
 		// Create the BlockValidation service
 		d.blockValidationSrv = blockvalidation.New(
 			createLogger(loggerBlockValidation),
@@ -811,6 +828,7 @@ func (d *Daemon) startValidationService(
 			blockchainClient,
 			kafkaConsumerClient,
 			blockAssemblyClient,
+			p2pClient,
 		)
 
 		// Add the BlockValidation service to the ServiceManager

daemon/daemon_stores.go

@@ -8,6 +8,7 @@ import (
 	"github.com/bsv-blockchain/teranode/services/blockassembly"
 	"github.com/bsv-blockchain/teranode/services/blockchain"
 	"github.com/bsv-blockchain/teranode/services/blockvalidation"
+	"github.com/bsv-blockchain/teranode/services/p2p"
 	"github.com/bsv-blockchain/teranode/services/subtreevalidation"
 	"github.com/bsv-blockchain/teranode/services/validator"
 	"github.com/bsv-blockchain/teranode/settings"
@@ -24,6 +25,7 @@ type Stores struct {
 	mainBlockPersisterStore     blob.Store
 	mainBlockStore              blob.Store
 	mainBlockValidationClient   blockvalidation.Interface
+	mainP2PClient               p2p.ClientI
 	mainSubtreeStore            blob.Store
 	mainSubtreeValidationClient subtreevalidation.Interface
 	mainTempStore               blob.Store
@@ -83,6 +85,31 @@ func (d *Stores) GetBlockValidationClient(ctx context.Context, logger ulogger.Lo
 	return d.mainBlockValidationClient, err
 }
 
+// GetP2PClient creates and returns a new P2P client instance. Unlike other store getters, this function
+// always creates a new client instance to maintain source information. The source parameter
+// identifies the origin or purpose of the client.
+//
+// Parameters:
+//   - ctx: The context for managing the client's lifecycle.
+//   - logger: The logger instance for logging client activities.
+//   - appSettings: The application settings containing configuration details.
+//
+// Returns:
+//   - p2p.ClientI: The newly created P2P client instance.
+//   - error: An error object if the client creation fails; otherwise, nil.
+func (d *Stores) GetP2PClient(ctx context.Context, logger ulogger.Logger, appSettings *settings.Settings) (p2p.ClientI, error) {
+	if d.mainP2PClient != nil {
+		return d.mainP2PClient, nil
+	}
+
+	p2pClient, err := p2p.NewClient(ctx, logger, appSettings)
+	if err != nil {
+		return nil, err
+	}
+
+	return p2pClient, nil
+}
+
 // GetBlockchainClient creates and returns a new blockchain client instance. Unlike other store
 // getters, this function always creates a new client instance to maintain source information.
 // The source parameter identifies the origin or purpose of the client.
@@ -376,6 +403,7 @@ func (d *Stores) Cleanup() {
 	d.mainTxStore = nil
 	d.mainUtxoStore = nil
 	d.mainValidatorClient = nil
+	d.mainP2PClient = nil
 
 	// Reset the Aerospike cleanup service singleton if it exists
 	// This prevents state leakage between test runs

services/alert/node.go

@@ -56,7 +56,7 @@ type Node struct {
 
 	// blockassemblyClient handles block assembly operations, allowing the
 	// alert system to interact with block creation processes
-	blockassemblyClient *blockassembly.Client
+	blockassemblyClient blockassembly.ClientI
 
 	// peerClient handles peer operations such as banning and unbanning peers
 	// based on alert system decisions
@@ -86,7 +86,7 @@ type Node struct {
 // Returns:
 //   - config.NodeInterface: A fully initialized Node instance that satisfies the required interface
 func NewNodeConfig(logger ulogger.Logger, blockchainClient blockchain.ClientI, utxoStore utxo.Store,
-	blockassemblyClient *blockassembly.Client, peerClient peer.ClientI, p2pClient p2p.ClientI, tSettings *settings.Settings) config.NodeInterface {
+	blockassemblyClient blockassembly.ClientI, peerClient peer.ClientI, p2pClient p2p.ClientI, tSettings *settings.Settings) config.NodeInterface {
 	return &Node{
 		logger:              logger,
 		blockchainClient:    blockchainClient,

services/alert/server.go

@@ -81,7 +81,7 @@ type Server struct {
 
 	// blockassemblyClient handles block assembly operations,
 	// allowing the alert system to influence block creation when necessary
-	blockassemblyClient *blockassembly.Client
+	blockassemblyClient blockassembly.ClientI
 
 	// appConfig contains alert system specific configuration loaded from
 	// the alert system configuration file, separate from Teranode settings
@@ -109,7 +109,7 @@ type Server struct {
 // Returns:
 //   - *Server: A new Server instance configured with the provided dependencies,
 //     but not yet initialized or started
-func New(logger ulogger.Logger, tSettings *settings.Settings, blockchainClient blockchain.ClientI, utxoStore utxo.Store, blockassemblyClient *blockassembly.Client, peerClient peer.ClientI, p2pClient p2pservice.ClientI) *Server {
+func New(logger ulogger.Logger, tSettings *settings.Settings, blockchainClient blockchain.ClientI, utxoStore utxo.Store, blockassemblyClient blockassembly.ClientI, peerClient peer.ClientI, p2pClient p2pservice.ClientI) *Server {
 	initPrometheusMetrics()
 
 	return &Server{

services/asset/httpimpl/get_catchup_status.go

@@ -0,0 +1,74 @@
+package httpimpl
+
+import (
+	"context"
+	"net/http"
+	"time"
+
+	"github.com/bsv-blockchain/teranode/services/blockvalidation/blockvalidation_api"
+	"github.com/labstack/echo/v4"
+)
+
+// GetCatchupStatus returns the current catchup status from the BlockValidation service
+func (h *HTTP) GetCatchupStatus(c echo.Context) error {
+	ctx, cancel := context.WithTimeout(c.Request().Context(), 5*time.Second)
+	defer cancel()
+
+	// Initialize persistent gRPC clients if not already done
+	h.initGRPCClients(ctx)
+
+	// Check if BlockValidation client connection is available
+	if h.blockvalClientConn == nil {
+		h.logger.Errorf("[GetCatchupStatus] BlockValidation gRPC client not available: %v", h.blockvalClientInitErr)
+		return c.JSON(http.StatusServiceUnavailable, map[string]interface{}{
+			"error":          "BlockValidation service not available",
+			"is_catching_up": false,
+		})
+	}
+
+	// Create BlockValidation API client from persistent connection
+	client := blockvalidation_api.NewBlockValidationAPIClient(h.blockvalClientConn)
+	resp, err := client.GetCatchupStatus(ctx, &blockvalidation_api.EmptyMessage{})
+	if err != nil {
+		h.logger.Errorf("[GetCatchupStatus] Failed to get catchup status: %v", err)
+		return c.JSON(http.StatusInternalServerError, map[string]interface{}{
+			"error":          "Failed to get catchup status",
+			"is_catching_up": false,
+		})
+	}
+
+	// Convert gRPC response to JSON
+	jsonResp := map[string]interface{}{
+		"is_catching_up":         resp.IsCatchingUp,
+		"peer_id":                resp.PeerId,
+		"peer_url":               resp.PeerUrl,
+		"target_block_hash":      resp.TargetBlockHash,
+		"target_block_height":    resp.TargetBlockHeight,
+		"current_height":         resp.CurrentHeight,
+		"total_blocks":           resp.TotalBlocks,
+		"blocks_fetched":         resp.BlocksFetched,
+		"blocks_validated":       resp.BlocksValidated,
+		"start_time":             resp.StartTime,
+		"duration_ms":            resp.DurationMs,
+		"fork_depth":             resp.ForkDepth,
+		"common_ancestor_hash":   resp.CommonAncestorHash,
+		"common_ancestor_height": resp.CommonAncestorHeight,
+	}
+
+	// Add previous attempt if available
+	if resp.PreviousAttempt != nil {
+		jsonResp["previous_attempt"] = map[string]interface{}{
+			"peer_id":             resp.PreviousAttempt.PeerId,
+			"peer_url":            resp.PreviousAttempt.PeerUrl,
+			"target_block_hash":   resp.PreviousAttempt.TargetBlockHash,
+			"target_block_height": resp.PreviousAttempt.TargetBlockHeight,
+			"error_message":       resp.PreviousAttempt.ErrorMessage,
+			"error_type":          resp.PreviousAttempt.ErrorType,
+			"attempt_time":        resp.PreviousAttempt.AttemptTime,
+			"duration_ms":         resp.PreviousAttempt.DurationMs,
+			"blocks_validated":    resp.PreviousAttempt.BlocksValidated,
+		}
+	}
+
+	return c.JSON(http.StatusOK, jsonResp)
+}