mirror of
https://github.com/luxfi/coreth.git
synced 2026-10-30 17:00:51 +00:00
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.
628 lines
24 KiB
Go
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
|
|
}
|