// Copyright (C) 2019-2025, Lux Industries, Inc. All rights reserved. // See the file LICENSE for licensing terms. package network import ( "context" "sync" "time" "github.com/luxfi/ids" log "github.com/luxfi/log" "github.com/luxfi/math/set" "github.com/luxfi/metric" "github.com/luxfi/p2p" "github.com/luxfi/consensus/version" ) // SyncedNetworkClient is the interface required by the state sync client type SyncedNetworkClient interface { SendSyncedAppRequestAny(ctx context.Context, minVersion *version.Application, request []byte) ([]byte, ids.NodeID, error) SendSyncedAppRequest(ctx context.Context, nodeID ids.NodeID, request []byte) ([]byte, error) Gossip(msg []byte) error TrackBandwidth(nodeID ids.NodeID, bandwidth float64) } // Network handles peer-to-peer networking for the EVM // It embeds p2p.Network to provide the base functionality type Network struct { *p2p.Network sender p2p.Sender codec any // message.Manager — held opaque to avoid plugin/evm/message → network import cycle maxRequests int64 metrics metric.Registerer log log.Logger // Request tracking for sync requestsLock sync.Mutex pendingRequests map[uint32]chan []byte nextRequestID uint32 // Request handler - stores the message.RequestHandler for sync operations // We use interface{} here to avoid import cycles with plugin/evm/message requestHandler interface{} // Shutdown closed bool closeLock sync.Mutex } // NewNetwork creates a new Network instance. The codec argument is the // caller's message.Manager (held opaquely as `any` to avoid pulling the // plugin/evm/message package into this network leaf — it would cycle). // Network never calls into the codec itself; it only stores the handle // for AppRequest dispatchers that pull the value out via SetRequestHandler. func NewNetwork( ctx context.Context, sender p2p.Sender, codec any, maxOutboundActiveRequests int64, metrics metric.Registerer, ) (*Network, error) { logger := log.New() var p2pNet *p2p.Network var err error if sender != nil { p2pNet, err = p2p.NewNetwork(logger, sender, metrics, "evm") if err != nil { return nil, err } } return &Network{ Network: p2pNet, sender: sender, codec: codec, maxRequests: maxOutboundActiveRequests, metrics: metrics, log: logger, pendingRequests: make(map[uint32]chan []byte), }, nil } // Shutdown stops the network func (n *Network) Shutdown() { n.closeLock.Lock() defer n.closeLock.Unlock() n.closed = true // Cancel all pending requests n.requestsLock.Lock() for _, ch := range n.pendingRequests { close(ch) } n.pendingRequests = make(map[uint32]chan []byte) n.requestsLock.Unlock() } // AddHandler registers a handler for a given protocol ID. // Safely handles the case where the embedded p2p.Network is nil (e.g., when sender is nil in tests). func (n *Network) AddHandler(handlerID uint64, handler p2p.Handler) error { if n.Network == nil { // No p2p network available (sender was nil during init) // This is expected in test environments or standalone mode n.log.Debug("Skipping AddHandler: no p2p network available", "handlerID", handlerID) return nil } return n.Network.AddHandler(handlerID, handler) } // NewClient creates a new p2p client for the given handler ID. // Returns a no-op client when the embedded p2p.Network is nil (e.g., when sender is nil in tests). func (n *Network) NewClient(handlerID uint64, options ...p2p.ClientOption) *p2p.Client { if n.Network == nil { // No p2p network available (sender was nil during init) // Return nil - callers must handle nil client gracefully n.log.Debug("NewClient: no p2p network available, returning nil client", "handlerID", handlerID) return nil } return n.Network.NewClient(handlerID, options...) } // P2PValidators returns the p2p validators if available func (n *Network) P2PValidators() *p2p.Validators { // Return nil - validators are set up separately return nil } // SetRequestHandler sets the handler for incoming requests // Accepts any type to avoid import cycles with plugin/evm/message func (n *Network) SetRequestHandler(handler interface{}) { n.requestHandler = handler } // Size returns the number of connected peers func (n *Network) Size() int { if n.Network == nil || n.Network.Peers == nil { return 0 } return len(n.Network.Peers.Sample(1000)) // Approximate count } // SendSyncedAppRequestAny sends a request to any available peer func (n *Network) SendSyncedAppRequestAny(ctx context.Context, minVersion *version.Application, request []byte) ([]byte, ids.NodeID, error) { if n.Network == nil || n.Network.Peers == nil { return nil, ids.EmptyNodeID, ErrNoPeers } peers := n.Network.Peers.Sample(1) if len(peers) == 0 { return nil, ids.EmptyNodeID, ErrNoPeers } // Try the first available peer nodeID := peers[0] response, err := n.SendSyncedAppRequest(ctx, nodeID, request) return response, nodeID, err } // SendSyncedAppRequest sends a request to a specific peer func (n *Network) SendSyncedAppRequest(ctx context.Context, nodeID ids.NodeID, request []byte) ([]byte, error) { if n.sender == nil { return nil, ErrNoSender } n.closeLock.Lock() if n.closed { n.closeLock.Unlock() return nil, ErrNetworkClosed } n.closeLock.Unlock() // Create a channel to receive the response responseChan := make(chan []byte, 1) requestID := n.allocateRequestID(responseChan) defer n.freeRequestID(requestID) // Send the request nodeIDs := set.NewSet[ids.NodeID](1) nodeIDs.Add(nodeID) if err := n.sender.SendRequest(ctx, nodeIDs, requestID, request); err != nil { return nil, err } // Wait for response with timeout select { case <-ctx.Done(): return nil, ctx.Err() case response, ok := <-responseChan: if !ok { return nil, ErrRequestCancelled } return response, nil } } // Gossip broadcasts a message to all peers func (n *Network) Gossip(msg []byte) error { if n.sender == nil { return ErrNoSender } if n.Network == nil || n.Network.Peers == nil { return nil // No peers to gossip to } peers := n.Network.Peers.Sample(100) // Sample up to 100 peers if len(peers) == 0 { return nil // No peers to gossip to } nodeIDs := set.NewSet[ids.NodeID](len(peers)) for _, peer := range peers { nodeIDs.Add(peer) } ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() return n.sender.SendGossip(ctx, p2p.SendConfig{NodeIDs: nodeIDs}, msg) } // TrackBandwidth records bandwidth usage for a peer. // Bandwidth metrics are not yet instrumented; this is a no-op placeholder // required by the network interface. func (n *Network) TrackBandwidth(nodeID ids.NodeID, bandwidth float64) { } // AppResponse handles an incoming response func (n *Network) AppResponse(ctx context.Context, nodeID ids.NodeID, requestID uint32, response []byte) error { n.requestsLock.Lock() ch, ok := n.pendingRequests[requestID] n.requestsLock.Unlock() if ok { select { case ch <- response: default: } } return nil } // allocateRequestID allocates a new request ID and registers the response channel func (n *Network) allocateRequestID(responseChan chan []byte) uint32 { n.requestsLock.Lock() defer n.requestsLock.Unlock() requestID := n.nextRequestID n.nextRequestID++ n.pendingRequests[requestID] = responseChan return requestID } // freeRequestID frees a request ID func (n *Network) freeRequestID(requestID uint32) { n.requestsLock.Lock() defer n.requestsLock.Unlock() delete(n.pendingRequests, requestID) }