Files
coreth/network/network.go
zeekay 8ff2228467 fix(evm): add ZAP nil-safety, genesis trie flush, and Fortuna gas model
Add nil-safety guards for SharedMemory and ValidatorState when running
under ZAP transport (no shared memory proxy). Force trie flush via
triedb.Cap(0) after genesis regeneration to prevent missing trie node
errors. Add Fortuna (ACP-176) gas model support returning Big0 post-fork.
Clean up plugin main entry point. Update consensus to v1.22.67.
2026-02-10 18:54:58 -08:00

628 lines
24 KiB
Go

// Copyright (C) 2019-2025, Lux Industries, Inc. All rights reserved.
// See the file LICENSE for licensing terms.
package network
import (
"context"
"errors"
"fmt"
"sync"
"time"
"golang.org/x/sync/semaphore"
log "github.com/luxfi/log"
"github.com/luxfi/metric"
"github.com/luxfi/atomic"
"github.com/luxfi/codec"
consensuscontext "github.com/luxfi/consensus/context"
validators "github.com/luxfi/consensus/validator"
consensusversion "github.com/luxfi/consensus/version"
"github.com/luxfi/constants"
"github.com/luxfi/ids"
"github.com/luxfi/math/set"
"github.com/luxfi/p2p"
"github.com/luxfi/coreth/network/stats"
"github.com/luxfi/coreth/plugin/evm/message"
)
// AppSender provides the interface for sending application-level messages
type AppSender interface {
SendAppRequest(nodeID ids.NodeID, requestID uint32, request []byte) error
SendAppResponse(nodeID ids.NodeID, requestID uint32, response []byte) error
SendCrossChainAppRequest(chainID ids.ID, requestID uint32, request []byte) error
SendCrossChainAppResponse(chainID ids.ID, requestID uint32, response []byte) error
}
// AppHandler handles application-level messages
type AppHandler interface {
AppRequest(ctx context.Context, nodeID ids.NodeID, requestID uint32, deadline time.Time, request []byte) error
AppResponse(ctx context.Context, nodeID ids.NodeID, requestID uint32, response []byte) error
AppRequestFailed(ctx context.Context, nodeID ids.NodeID, requestID uint32, appErr *p2p.Error) error
AppGossip(ctx context.Context, nodeID ids.NodeID, msg []byte) error
CrossChainAppRequest(ctx context.Context, chainID ids.ID, requestID uint32, deadline time.Time, msg []byte) error
CrossChainAppResponse(ctx context.Context, chainID ids.ID, requestID uint32, msg []byte) error
CrossChainAppRequestFailed(ctx context.Context, chainID ids.ID, requestID uint32, appErr *p2p.Error) error
}
// Minimum amount of time to handle a request
const (
minRequestHandlingDuration = 100 * time.Millisecond
maxValidatorSetStaleness = time.Minute
)
var (
errAcquiringSemaphore = errors.New("error acquiring semaphore")
errExpiredRequest = errors.New("expired request")
_ Network = (*network)(nil)
_ validators.Connector = (*network)(nil)
_ AppHandler = (*network)(nil)
)
// SyncedNetworkClient defines ability to send request / response through the Network
type SyncedNetworkClient interface {
// SendSyncedAppRequestAny synchronously sends request to an arbitrary peer with a
// node version greater than or equal to minVersion.
// Returns response bytes, the ID of the chosen peer, and ErrRequestFailed if
// the request should be retried.
SendSyncedAppRequestAny(ctx context.Context, minVersion *consensusversion.Application, request []byte) ([]byte, ids.NodeID, error)
// SendSyncedAppRequest synchronously sends request to the selected nodeID
// Returns response bytes, and ErrRequestFailed if the request should be retried.
SendSyncedAppRequest(ctx context.Context, nodeID ids.NodeID, request []byte) ([]byte, error)
// TrackBandwidth should be called for each valid request with the bandwidth
// (length of response divided by request time), and with 0 if the response is invalid.
TrackBandwidth(nodeID ids.NodeID, bandwidth float64)
}
type Network interface {
validators.Connector
AppHandler
SyncedNetworkClient
// SendAppRequestAny sends request to an arbitrary peer with a
// node version greater than or equal to minVersion.
// Returns the ID of the chosen peer, and an error if the request could not
// be sent to a peer with the desired [minVersion].
SendAppRequestAny(ctx context.Context, minVersion *consensusversion.Application, message []byte, handler message.ResponseHandler) (ids.NodeID, error)
// SendAppRequest sends message to given nodeID, notifying handler when there's a response or timeout
SendAppRequest(ctx context.Context, nodeID ids.NodeID, message []byte, handler message.ResponseHandler) error
// Shutdown stops all peer channel listeners and marks the node to have stopped
// n.Start() can be called again but the peers will have to be reconnected
// by calling OnPeerConnected for each peer
Shutdown()
// SetRequestHandler sets the provided request handler as the request handler
SetRequestHandler(handler message.RequestHandler)
// Size returns the size of the network in number of connected peers
Size() uint32
// NewClient returns a client to send messages with for the given protocol
NewClient(protocol uint64, options ...p2p.ClientOption) *p2p.Client
// AddHandler registers a server handler for an application protocol
AddHandler(protocol uint64, handler p2p.Handler) error
P2PValidators() *p2p.Validators
}
// network is an implementation of Network that processes message requests for
// each peer in linear fashion
type network struct {
lock sync.RWMutex // lock for mutating state of this Network struct
self ids.NodeID // NodeID of this node
requestIDGen uint32 // requestID counter used to track outbound requests
outstandingRequestHandlers map[uint32]message.ResponseHandler // maps luxd requestID => message.ResponseHandler
activeAppRequests *semaphore.Weighted // controls maximum number of active outbound requests
sdkNetwork *p2p.Network // SDK network (luxd p2p) for sending messages to peers
appSender AppSender // luxd AppSender for sending messages
codec codec.Manager // Codec used for parsing messages
appRequestHandler message.RequestHandler // maps request type => handler
peers *peerTracker // tracking of peers & bandwidth
appStats stats.RequestHandlerStats // Provide request handler metrics
// Set to true when Shutdown is called, after which all operations on this
// struct are no-ops.
//
// Invariant: Even though `closed` is an atomic, `lock` is required to be
// held when sending requests to guarantee that the network isn't closed
// during these calls. This is because closing the network cancels all
// outstanding requests, which means we must guarantee never to register a
// request that will never be fulfilled or cancelled.
closed atomic.Atomic[bool]
p2pValidators *p2p.Validators
}
func NewNetwork(
ctx *consensuscontext.Context,
appSender AppSender,
codec codec.Manager,
maxActiveAppRequests int64,
registerer metric.Registerer,
) (Network, error) {
// Create a wrapper that implements p2p.Sender
p2pSender := &appSenderWrapper{appSender: appSender}
p2pNetwork, err := p2p.NewNetwork(ctx.Log.(log.Logger), p2pSender, registerer, "p2p")
if err != nil {
return nil, fmt.Errorf("failed to initialize p2p network: %w", err)
}
// ValidatorState may be nil when using ZAP transport (default)
// which doesn't pass full Runtime context. Use ok-pattern to
// avoid panic on nil interface assertion.
var p2pVals *p2p.Validators
if validatorState, ok := ctx.ValidatorState.(validators.State); ok {
p2pVals = p2p.NewValidators(p2pNetwork.Peers, ctx.Log.(log.Logger), constants.PrimaryNetworkID, validatorState, maxValidatorSetStaleness)
}
return &network{
appSender: appSender,
codec: codec,
self: ctx.NodeID,
outstandingRequestHandlers: make(map[uint32]message.ResponseHandler),
activeAppRequests: semaphore.NewWeighted(maxActiveAppRequests),
sdkNetwork: p2pNetwork,
appRequestHandler: message.NoopRequestHandler{},
peers: NewPeerTracker(),
appStats: stats.NewRequestHandlerStats(),
p2pValidators: p2pVals,
}, nil
}
// SendAppRequestAny synchronously sends request to an arbitrary peer with a
// node version greater than or equal to minVersion. If minVersion is nil,
// the request will be sent to any peer regardless of their version.
// Returns the ID of the chosen peer, and an error if the request could not
// be sent to a peer with the desired [minVersion].
func (n *network) SendAppRequestAny(ctx context.Context, minVersion *consensusversion.Application, request []byte, handler message.ResponseHandler) (ids.NodeID, error) {
// If the context was cancelled, we can skip sending this request.
if err := ctx.Err(); err != nil {
return ids.EmptyNodeID, err
}
// Take a slot from total [activeAppRequests] and block until a slot becomes available.
if err := n.activeAppRequests.Acquire(ctx, 1); err != nil {
return ids.EmptyNodeID, errAcquiringSemaphore
}
n.lock.Lock()
defer n.lock.Unlock()
if nodeID, ok := n.peers.GetAnyPeer(minVersion); ok {
return nodeID, n.sendAppRequest(ctx, nodeID, request, handler)
}
n.activeAppRequests.Release(1)
return ids.EmptyNodeID, fmt.Errorf("no peers found matching version %s out of %d peers", minVersion, n.peers.Size())
}
// SendAppRequest sends request message bytes to specified nodeID, notifying the responseHandler on response or failure
func (n *network) SendAppRequest(ctx context.Context, nodeID ids.NodeID, request []byte, responseHandler message.ResponseHandler) error {
if nodeID == ids.EmptyNodeID {
return fmt.Errorf("cannot send request to empty nodeID, nodeID=%s, requestLen=%d", nodeID, len(request))
}
// If the context was cancelled, we can skip sending this request.
if err := ctx.Err(); err != nil {
return err
}
// Take a slot from total [activeAppRequests] and block until a slot becomes available.
if err := n.activeAppRequests.Acquire(ctx, 1); err != nil {
return errAcquiringSemaphore
}
n.lock.Lock()
defer n.lock.Unlock()
return n.sendAppRequest(ctx, nodeID, request, responseHandler)
}
// sendAppRequest sends request message bytes to specified nodeID and adds [responseHandler] to [outstandingRequestHandlers]
// so that it can be invoked when the network receives either a response or failure message.
// Assumes [nodeID] is never [self] since we guarantee [self] will not be added to the [peers] map.
// Releases active requests semaphore if there was an error in sending the request
// Returns an error if [appSender] is unable to make the request.
// Assumes write lock is held
func (n *network) sendAppRequest(ctx context.Context, nodeID ids.NodeID, request []byte, responseHandler message.ResponseHandler) error {
if n.closed.Get() {
n.activeAppRequests.Release(1)
return nil
}
// If the context was cancelled, we can skip sending this request.
if err := ctx.Err(); err != nil {
n.activeAppRequests.Release(1)
return err
}
log.Debug("sending request to peer", "nodeID", nodeID, "requestLen", len(request))
n.peers.TrackPeer(nodeID)
requestID := n.nextRequestID()
n.outstandingRequestHandlers[requestID] = responseHandler
nodeIDs := set.NewSet[ids.NodeID](1)
nodeIDs.Add(nodeID)
// Send app request to [nodeID].
// On failure, release the slot from [activeAppRequests] and delete request
// from [outstandingRequestHandlers]
//
// Cancellation is removed from this context to avoid erroring unexpectedly.
// SendAppRequest should be non-blocking and any error other than context
// cancellation is unexpected.
//
// This guarantees that the network should never receive an unexpected
// AppResponse.
if err := n.appSender.SendAppRequest(nodeIDs.List()[0], requestID, request); err != nil {
log.Error(
"request to peer failed",
"nodeID", nodeID,
"requestID", requestID,
"requestLen", len(request),
"error", err,
)
n.activeAppRequests.Release(1)
delete(n.outstandingRequestHandlers, requestID)
return err
}
log.Debug("sent request message to peer", "nodeID", nodeID, "requestID", requestID)
return nil
}
// AppRequest is called by luxd -> VM when there is an incoming AppRequest from a peer
// error returned by this function is expected to be treated as fatal by the engine
// returns error if the requestHandler returns an error
// sends a response back to the sender if length of response returned by the handler is >0
// expects the deadline to not have been passed
func (n *network) AppRequest(ctx context.Context, nodeID ids.NodeID, requestID uint32, deadline time.Time, request []byte) error {
if n.closed.Get() {
return nil
}
log.Debug("received AppRequest from node", "nodeID", nodeID, "requestID", requestID, "requestLen", len(request))
if !IsNetworkRequest(requestID) {
log.Debug("forwarding AppRequest to SDK network", "nodeID", nodeID, "requestID", requestID, "requestLen", len(request))
_, p2pErr := n.sdkNetwork.Request(ctx, nodeID, requestID, deadline, request)
if p2pErr != nil {
return errors.New(p2pErr.Message)
}
return nil
}
bufferedDeadline, err := calculateTimeUntilDeadline(deadline, n.appStats)
if err != nil {
log.Debug("deadline to process AppRequest has expired, skipping", "nodeID", nodeID, "requestID", requestID, "err", err)
return nil
}
var req message.Request
if _, err := n.codec.Unmarshal(request, &req); err != nil {
log.Debug("failed to unmarshal AppRequest", "nodeID", nodeID, "requestID", requestID, "err", err)
return nil
}
log.Debug("processing incoming request", "nodeID", nodeID, "requestID", requestID, "req", req)
// We make a new context here because we don't want to cancel the context
// passed into n.AppSender.SendAppResponse below
handleCtx, cancel := context.WithDeadline(context.Background(), bufferedDeadline)
defer cancel()
responseBytes, err := req.Handle(handleCtx, nodeID, requestID, n.appRequestHandler)
switch {
case err != nil && err != context.DeadlineExceeded:
return err // Return a fatal error
case responseBytes != nil:
return n.appSender.SendAppResponse(nodeID, requestID, responseBytes) // Propagate fatal error
default:
return nil
}
}
// AppResponse is invoked when there is a response received from a peer regarding a request
// Error returned by this function is expected to be treated as fatal by the engine
// If [requestID] is not known, this function will emit a log and return a nil error.
// If the response handler returns an error it is propagated as a fatal error.
func (n *network) AppResponse(ctx context.Context, nodeID ids.NodeID, requestID uint32, response []byte) error {
log.Debug("received AppResponse from peer", "nodeID", nodeID, "requestID", requestID)
handler, exists := n.markRequestFulfilled(requestID)
if !exists {
log.Debug("forwarding AppResponse to SDK network", "nodeID", nodeID, "requestID", requestID, "responseLen", len(response))
return n.sdkNetwork.Response(ctx, nodeID, requestID, response)
}
// We must release the slot
n.activeAppRequests.Release(1)
return handler.OnResponse(response)
}
// AppRequestFailed can be called by the luxd -> VM in following cases:
// - node is benched
// - failed to send message to [nodeID] due to a network issue
// - request times out before a response is provided
// error returned by this function is expected to be treated as fatal by the engine
// returns error only when the response handler returns an error
func (n *network) AppRequestFailed(ctx context.Context, nodeID ids.NodeID, requestID uint32, appErr *p2p.Error) error {
log.Debug("received AppRequestFailed from peer", "nodeID", nodeID, "requestID", requestID)
handler, exists := n.markRequestFulfilled(requestID)
if !exists {
log.Debug("forwarding AppRequestFailed to SDK network", "nodeID", nodeID, "requestID", requestID)
return n.sdkNetwork.RequestFailed(ctx, nodeID, requestID, appErr)
}
// We must release the slot
n.activeAppRequests.Release(1)
return handler.OnFailure()
}
// calculateTimeUntilDeadline calculates the time until deadline and drops it if we missed he deadline to response.
// This function updates metrics for app requests.
// This is called by [AppRequest].
func calculateTimeUntilDeadline(deadline time.Time, stats stats.RequestHandlerStats) (time.Time, error) {
// calculate how much time is left until the deadline
timeTillDeadline := time.Until(deadline)
stats.UpdateTimeUntilDeadline(timeTillDeadline)
// bufferedDeadline is half the time till actual deadline so that the message has a reasonable chance
// of completing its processing and sending the response to the peer.
bufferedDeadline := time.Now().Add(timeTillDeadline / 2)
// check if we have enough time to handle this request
if time.Until(bufferedDeadline) < minRequestHandlingDuration {
// Drop the request if we already missed the deadline to respond.
stats.IncDeadlineDroppedRequest()
return time.Time{}, errExpiredRequest
}
return bufferedDeadline, nil
}
// markRequestFulfilled fetches the handler for [requestID] and marks the request with [requestID] as having been fulfilled.
// This is called by either [AppResponse] or [AppRequestFailed].
// Assumes that the write lock is not held.
func (n *network) markRequestFulfilled(requestID uint32) (message.ResponseHandler, bool) {
n.lock.Lock()
defer n.lock.Unlock()
handler, exists := n.outstandingRequestHandlers[requestID]
if !exists {
return nil, false
}
// mark message as processed
delete(n.outstandingRequestHandlers, requestID)
return handler, true
}
// AppGossip is called by luxd -> VM when there is an incoming AppGossip
// from a peer. An error returned by this function is treated as fatal by the
// engine.
func (n *network) AppGossip(ctx context.Context, nodeID ids.NodeID, gossipBytes []byte) error {
return n.sdkNetwork.Gossip(ctx, nodeID, gossipBytes)
}
// Connected adds the given nodeID to the peer list so that it can receive messages
func (n *network) Connected(ctx context.Context, nodeID ids.NodeID, nodeVersion *consensusversion.Application) error {
log.Debug("adding new peer", "nodeID", nodeID)
n.lock.Lock()
defer n.lock.Unlock()
if n.closed.Get() {
return nil
}
if nodeID != n.self {
// The legacy peer tracker doesn't expect to be connected to itself.
n.peers.Connected(nodeID, nodeVersion)
}
return n.sdkNetwork.Connected(ctx, nodeID, nodeVersion)
}
// Disconnected removes given [nodeID] from the peer list
func (n *network) Disconnected(ctx context.Context, nodeID ids.NodeID) error {
log.Debug("disconnecting peer", "nodeID", nodeID)
n.lock.Lock()
defer n.lock.Unlock()
if n.closed.Get() {
return nil
}
if nodeID != n.self {
// The legacy peer tracker doesn't expect to be connected to itself.
n.peers.Disconnected(nodeID)
}
return n.sdkNetwork.Disconnected(ctx, nodeID)
}
// Shutdown disconnects all peers
func (n *network) Shutdown() {
n.lock.Lock()
defer n.lock.Unlock()
// clean up any pending requests
for requestID, handler := range n.outstandingRequestHandlers {
_ = handler.OnFailure() // make sure all waiting threads are unblocked
delete(n.outstandingRequestHandlers, requestID)
}
n.peers = NewPeerTracker() // reset peers
n.closed.Set(true) // mark network as closed
}
func (n *network) SetRequestHandler(handler message.RequestHandler) {
n.lock.Lock()
defer n.lock.Unlock()
n.appRequestHandler = handler
}
func (n *network) Size() uint32 {
n.lock.RLock()
defer n.lock.RUnlock()
return uint32(n.peers.Size())
}
func (n *network) TrackBandwidth(nodeID ids.NodeID, bandwidth float64) {
n.lock.Lock()
defer n.lock.Unlock()
n.peers.TrackBandwidth(nodeID, bandwidth)
}
// SendSyncedAppRequestAny synchronously sends request to an arbitrary peer with a
// node version greater than or equal to minVersion.
// Returns response bytes, the ID of the chosen peer, and ErrRequestFailed if
// the request should be retried.
func (n *network) SendSyncedAppRequestAny(ctx context.Context, minVersion *consensusversion.Application, request []byte) ([]byte, ids.NodeID, error) {
waitingHandler := newWaitingResponseHandler()
nodeID, err := n.SendAppRequestAny(ctx, minVersion, request, waitingHandler)
if err != nil {
return nil, nodeID, err
}
response, err := waitingHandler.WaitForResult(ctx)
return response, nodeID, err
}
// SendSyncedAppRequest synchronously sends request to the specified nodeID
// Returns response bytes and ErrRequestFailed if the request should be retried.
func (n *network) SendSyncedAppRequest(ctx context.Context, nodeID ids.NodeID, request []byte) ([]byte, error) {
waitingHandler := newWaitingResponseHandler()
if err := n.SendAppRequest(ctx, nodeID, request, waitingHandler); err != nil {
return nil, err
}
return waitingHandler.WaitForResult(ctx)
}
func (n *network) NewClient(protocol uint64, options ...p2p.ClientOption) *p2p.Client {
return n.sdkNetwork.NewClient(protocol, options...)
}
func (n *network) AddHandler(protocol uint64, handler p2p.Handler) error {
return n.sdkNetwork.AddHandler(protocol, handler)
}
// P2PValidators returns the p2p validators
func (n *network) P2PValidators() *p2p.Validators {
return n.p2pValidators
}
// invariant: peer/network must use explicitly even request ids.
// for this reason, [n.requestID] is initialized as zero and incremented by 2.
// This is for backwards-compatibility while the SDK router exists with the
// legacy coreth handlers to avoid a (very) narrow edge case where request ids
// can overlap, resulting in a dropped timeout.
func (n *network) nextRequestID() uint32 {
next := n.requestIDGen
n.requestIDGen += 2
return next
}
// IsNetworkRequest checks if the given requestID is a request for this network handler (even-numbered requestIDs)
// SDK requests are odd-numbered requestIDs
// (see invariant: https://github.com/luxfi/node/blob/v1.13.0/network/p2p/router.go#L83)
func IsNetworkRequest(requestID uint32) bool {
return requestID%2 == 0
}
// CrossChainAppRequest handles a cross-chain request
func (n *network) CrossChainAppRequest(ctx context.Context, chainID ids.ID, requestID uint32, deadline time.Time, msg []byte) error {
// For now, we don't handle cross-chain requests
return nil
}
// CrossChainAppResponse handles a cross-chain response to a request this node sent
func (n *network) CrossChainAppResponse(ctx context.Context, chainID ids.ID, requestID uint32, msg []byte) error {
// For now, we don't handle cross-chain responses
return nil
}
// CrossChainAppRequestFailed notifies that a CrossChainAppRequest failed or timed out
func (n *network) CrossChainAppRequestFailed(ctx context.Context, chainID ids.ID, requestID uint32, appErr *p2p.Error) error {
// For now, we don't handle cross-chain request failures
return nil
}
// appSenderWrapper wraps an AppSender to implement p2p.Sender
type appSenderWrapper struct {
appSender AppSender
}
func (w *appSenderWrapper) SendRequest(ctx context.Context, nodeIDs set.Set[ids.NodeID], requestID uint32, appRequestBytes []byte) error {
// AppSender takes single nodeID at a time
for nodeID := range nodeIDs {
if err := w.appSender.SendAppRequest(nodeID, requestID, appRequestBytes); err != nil {
return err
}
}
return nil
}
func (w *appSenderWrapper) SendResponse(ctx context.Context, nodeID ids.NodeID, requestID uint32, appResponseBytes []byte) error {
return w.appSender.SendAppResponse(nodeID, requestID, appResponseBytes)
}
func (w *appSenderWrapper) SendError(ctx context.Context, nodeID ids.NodeID, requestID uint32, errorCode int32, errorMessage string) error {
// AppSender doesn't have SendAppError, so we just log it
log.Debug("app error", "nodeID", nodeID, "requestID", requestID, "code", errorCode, "message", errorMessage)
return nil
}
func (w *appSenderWrapper) SendGossip(ctx context.Context, config p2p.SendConfig, appGossipBytes []byte) error {
// AppSender doesn't have SendAppGossip
return nil
}
// P2PSenderAdapter adapts p2p.Sender to network.AppSender
type P2PSenderAdapter struct {
sender p2p.Sender
}
var _ AppSender = (*P2PSenderAdapter)(nil)
// NewP2PSenderAdapter creates an adapter that wraps p2p.Sender
func NewP2PSenderAdapter(sender p2p.Sender) AppSender {
return &P2PSenderAdapter{sender: sender}
}
// SendAppRequest sends to a single node via p2p.Sender
func (a *P2PSenderAdapter) SendAppRequest(nodeID ids.NodeID, requestID uint32, request []byte) error {
nodeIDs := set.Of(nodeID)
return a.sender.SendRequest(context.Background(), nodeIDs, requestID, request)
}
// SendAppResponse forwards response via p2p.Sender
func (a *P2PSenderAdapter) SendAppResponse(nodeID ids.NodeID, requestID uint32, response []byte) error {
return a.sender.SendResponse(context.Background(), nodeID, requestID, response)
}
// SendCrossChainAppRequest - cross-chain not supported via p2p.Sender
func (a *P2PSenderAdapter) SendCrossChainAppRequest(chainID ids.ID, requestID uint32, request []byte) error {
return nil
}
// SendCrossChainAppResponse - cross-chain not supported via p2p.Sender
func (a *P2PSenderAdapter) SendCrossChainAppResponse(chainID ids.ID, requestID uint32, response []byte) error {
return nil
}