Expand peer registry usage and improve reputation system for catchup (#20)

Co-authored-by: oskarszoon <[email protected]>
Co-authored-by: Siggi <[email protected]>
Co-authored-by: Siggi <[email protected]>

Author: 3 people

CLAUDE.md

@@ -193,6 +193,53 @@ The project includes a Bitcoin expert agent (`.claude/agents/bitcoin-expert.md`)
 - Don't use mock blockchain client/store - you can use a real one using the sqlitememory store
 - Don't use mock kafka - you can use in_memory_kafka.go
 
+### Service Interface Design Pattern
+
+When creating or updating service interfaces and clients, follow this pattern to avoid exposing protobuf/gRPC types:
+
+**Interface Layer** (`Interface.go`):
+- Define interfaces using native Go types and existing domain types (e.g., `*PeerInfo`, `[]string`, `bool`, `error`)
+- Do NOT expose protobuf types (e.g., `*p2p_api.GetPeersResponse`) in interface signatures
+- Use simple, idiomatic Go return types: `error` for success/fail, `bool` for yes/no, `[]string` for lists
+- Prefer existing domain structs over creating new minimal types
+
+**Client Layer** (`Client.go`):
+- Keep the protobuf/gRPC import for internal use (e.g., `import "github.com/bsv-blockchain/teranode/services/p2p/p2p_api"`)
+- Maintain internal gRPC client field (e.g., `client p2p_api.PeerServiceClient`)
+- Public methods match the interface signatures (native types)
+- Convert between native types and protobuf types internally using helper functions
+
+**Benefits**:
+- Cleaner API boundaries between services
+- Reduces coupling to gRPC implementation details
+- Makes interfaces more testable (no protobuf dependencies needed for mocks)
+- Uses idiomatic Go types that are easier to work with
+
+**Example**:
+```go
+// Interface.go - Clean, no protobuf types
+type ClientI interface {
+    GetPeers(ctx context.Context) ([]*PeerInfo, error)
+    BanPeer(ctx context.Context, peerID string, duration int64, reason string) error
+    IsBanned(ctx context.Context, peerID string) (bool, error)
+    ListBanned(ctx context.Context) ([]string, error)
+}
+
+// Client.go - Internal conversion
+type Client struct {
+    client p2p_api.PeerServiceClient // gRPC client
+}
+
+func (c *Client) GetPeers(ctx context.Context) ([]*PeerInfo, error) {
+    resp, err := c.client.GetPeers(ctx, &emptypb.Empty{})
+    if err != nil {
+        return nil, err
+    }
+    // Convert p2p_api types to native PeerInfo
+    return convertFromAPIResponse(resp), nil
+}
+```
+
 ## Git Workflow (Fork Mode)
 
 All developers work in forked repositories with `upstream` remote pointing to the original repo.

cmd/filereader/file_reader.go

@@ -212,7 +212,7 @@ func handleSubtreeData(br *bufio.Reader, logger ulogger.Logger, settings *settin
 		return errors.NewProcessingError("error reading subtree", err)
 	}
 
-	var sd *subtree.SubtreeData
+	var sd *subtree.Data
 
 	sd, err = subtree.NewSubtreeDataFromReader(st, br)
 	if err != nil {
@@ -258,7 +258,7 @@ func handleSubtreeMeta(br *bufio.Reader, logger ulogger.Logger, settings *settin
 		return errors.NewProcessingError("error reading subtree", err)
 	}
 
-	var subtreeMeta *subtree.SubtreeMeta
+	var subtreeMeta *subtree.Meta
 
 	subtreeMeta, err = subtree.NewSubtreeMetaFromReader(st, br)
 	if err != nil {

daemon/daemon.go

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

daemon/daemon_services.go

@@ -401,6 +401,16 @@ func (d *Daemon) startAssetService(ctx context.Context, appSettings *settings.Se
 		return err
 	}
 
+	// Get the P2P client for the Asset service
+	var p2pClient p2p.ClientI
+
+	p2pClient, err = d.daemonStores.GetP2PClient(
+		ctx, createLogger(loggerP2P), appSettings,
+	)
+	if err != nil {
+		return err
+	}
+
 	// Initialize the Asset service with the necessary parts
 	return d.ServiceManager.AddService(serviceAssetFormal, asset.NewServer(
 		createLogger(serviceAsset),
@@ -411,6 +421,7 @@ func (d *Daemon) startAssetService(ctx context.Context, appSettings *settings.Se
 		blockPersisterStore,
 		blockchainClient,
 		blockvalidationClient,
+		p2pClient,
 	))
 }
 
@@ -433,25 +444,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
 	}
@@ -481,7 +492,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
 	}
 
@@ -490,43 +501,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
 	}
@@ -762,6 +773,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
 
@@ -776,6 +795,7 @@ func (d *Daemon) startValidationService(
 			blockchainClient,
 			subtreeConsumerClient,
 			txMetaConsumerClient,
+			p2pClient,
 		)
 		if err != nil {
 			return err
@@ -810,6 +830,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),
@@ -821,6 +849,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,33 @@ 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
+	}
+
+	d.mainP2PClient = p2pClient
+
+	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 +405,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

daemon/test_daemon.go

@@ -1399,7 +1399,7 @@ func createAndSaveSubtrees(ctx context.Context, subtreeStore blob.Store, txs []*
 }
 
 // storeSubtreeFiles serializes and stores the subtree, subtree data, and subtree meta in the provided subtree store.
-func storeSubtreeFiles(ctx context.Context, subtreeStore blob.Store, subtree *subtreepkg.Subtree, subtreeData *subtreepkg.SubtreeData, subtreeMeta *subtreepkg.SubtreeMeta) error {
+func storeSubtreeFiles(ctx context.Context, subtreeStore blob.Store, subtree *subtreepkg.Subtree, subtreeData *subtreepkg.Data, subtreeMeta *subtreepkg.Meta) error {
 	subtreeBytes, err := subtree.Serialize()
 	if err != nil {
 		return err
@@ -1874,22 +1874,22 @@ func (td *TestDaemon) ConnectToPeer(t *testing.T, peer *TestDaemon) {
 
 			return
 		case <-ticker.C:
-			r, err := td.P2PClient.GetPeers(td.Ctx)
+			peers, err := td.P2PClient.GetPeers(td.Ctx)
 			if err != nil {
 				// If there's an error calling RPC, log it and continue retrying
 				t.Logf("Error calling getpeerinfo: %v. Retrying...", err)
 				continue
 			}
 
-			if len(r.Peers) == 0 {
+			if len(peers) == 0 {
 				t.Logf("getpeerinfo returned empty peer list. Retrying...")
 				continue
 			}
 
 			found := false
 
-			for _, p := range r.Peers {
-				if p != nil && p.Id == peer.Settings.P2P.PeerID {
+			for _, p := range peers {
+				if p != nil && p.ID.String() == peer.Settings.P2P.PeerID {
 					found = true
 					break
 				}

docs/references/services/blockvalidation_reference.md

@@ -284,7 +284,7 @@ Validates a block and returns validation results without adding it to the blockc
 #### processBlockFound
 
 ```go
-func (u *Server) processBlockFound(ctx context.Context, hash *chainhash.Hash, baseURL string, peerID string, useBlock ...*model.Block) error
+func (u *Server) processBlockFound(ctx context.Context, hash *chainhash.Hash, peerID string, baseURL string, useBlock ...*model.Block) error
 ```
 
 Internal method that processes a newly discovered block. Handles block retrieval, validation, and integration with the blockchain state.

docs/references/settings/services/p2p_settings.md

@@ -28,17 +28,10 @@
 | BanDuration | time.Duration | 24h | p2p_ban_duration | Ban duration |
 | ForceSyncPeer | string | "" | p2p_force_sync_peer | **CRITICAL** - Forced sync peer override |
 | SharePrivateAddresses | bool | true | p2p_share_private_addresses | Private address advertisement |
-| PeerHealthCheckInterval | time.Duration | 30s | p2p_health_check_interval | **CRITICAL** - Health check timing |
-| PeerHealthHTTPTimeout | time.Duration | 5s | p2p_health_http_timeout | **CRITICAL** - Health check HTTP timeout |
-| PeerHealthRemoveAfterFailures | int | 3 | p2p_health_remove_after_failures | **CRITICAL** - Failure threshold for peer removal |
 | AllowPrunedNodeFallback | bool | true | p2p_allow_pruned_node_fallback | **CRITICAL** - Pruned node fallback behavior |
 
 ## Configuration Dependencies
 
-### Peer Health Management
-- `PeerHealthCheckInterval`, `PeerHealthHTTPTimeout`, and `PeerHealthRemoveAfterFailures` work together
-- Controls peer health monitoring and removal behavior
-
 ### Forced Sync Peer Selection
 - `ForceSyncPeer` overrides automatic peer selection
 - `AllowPrunedNodeFallback` affects fallback behavior when forced peer unavailable

errors/Error_types.go

@@ -17,6 +17,7 @@ var (
 	ErrBlockInvalidFormat         = New(ERR_BLOCK_INVALID_FORMAT, "block format is invalid")
 	ErrBlockNotFound              = New(ERR_BLOCK_NOT_FOUND, "block not found")
 	ErrBlockParentNotMined        = New(ERR_BLOCK_PARENT_NOT_MINED, "block parent not mined")
+	ErrCatchupInProgress          = New(ERR_CATCHUP_IN_PROGRESS, "catchup in progress")
 	ErrConfiguration              = New(ERR_CONFIGURATION, "configuration error")
 	ErrContextCanceled            = New(ERR_CONTEXT_CANCELED, "context canceled")
 	ErrError                      = New(ERR_ERROR, "generic error")
@@ -286,6 +287,11 @@ func NewStateInitializationError(message string, params ...interface{}) *Error {
 	return New(ERR_STATE_INITIALIZATION, message, params...)
 }
 
+// NewCatchupInProgressError creates a new error with the catchup in progress error code.
+func NewCatchupInProgressError(message string, params ...interface{}) *Error {
+	return New(ERR_CATCHUP_IN_PROGRESS, message, params...)
+}
+
 // NewStateError creates a new error with the state error code.
 func NewStateError(message string, params ...interface{}) *Error {
 	return New(ERR_STATE_ERROR, message, params...)