Gossip Timeout

internal/node/node.go

@@ -197,7 +197,8 @@ func NewKoinosP2PNode(ctx context.Context, listenAddr string, localRPC rpc.Local
 		node.PeerErrorChan,
 		node.Host.ID(),
 		node,
-		node.Applicator)
+		node.Applicator,
+		config.GossipOptions)
 
 	node.ConnectionManager = p2p.NewConnectionManager(
 		node.Host,

internal/options/config.go

@@ -8,6 +8,7 @@ type Config struct {
 	GossipToggleOptions      GossipToggleOptions
 	ApplicatorOptions        ApplicatorOptions
 	ConnectionManagerOptions ConnectionManagerOptions
+	GossipOptions            GossipOptions
 }
 
 // NewConfig creates a new Config
@@ -19,6 +20,7 @@ func NewConfig() *Config {
 		GossipToggleOptions:      *NewGossipToggleOptions(),
 		ApplicatorOptions:        *NewApplicatorOptions(),
 		ConnectionManagerOptions: *NewConnectionManagerOptions(),
+		GossipOptions:            *NewGossipOptions(),
 	}
 	return &config
 }

internal/options/gossip_options.go

@@ -0,0 +1,22 @@
+package options
+
+import "time"
+
+const (
+	blockTimeoutDefault       = 2 * time.Second
+	transactionTimeoutDefault = 1 * time.Second
+)
+
+// GossipOptions are options for Gossip
+type GossipOptions struct {
+	BlockTimeout       time.Duration
+	TransactionTimeout time.Duration
+}
+
+// New GossipOptions returns default initialized GossipOptions
+func NewGossipOptions() *GossipOptions {
+	return &GossipOptions{
+		BlockTimeout:       blockTimeoutDefault,
+		TransactionTimeout: transactionTimeoutDefault,
+	}
+}

internal/options/peer_connection_options.go

@@ -20,7 +20,7 @@ const (
 	blockRequestTimeoutDefault   = time.Second * time.Duration(blockRequestBatchSizeDefault*0.3) // Roughly calculated considering 30Mbps minimum upload speed and 1MB blocks
 	handshakeRetryTimeDefault    = time.Second * 6
 	syncedBlockDeltaDefault      = 5
-	syncedPingTimeDefault        = time.Second * 10
+	syncedSleepTimeDefault       = time.Second * 10
 )
 
 // PeerConnectionOptions are options for PeerConnection
@@ -33,7 +33,7 @@ type PeerConnectionOptions struct {
 	BlockRequestTimeout   time.Duration
 	HandshakeRetryTime    time.Duration
 	SyncedBlockDelta      uint64
-	SyncedPingTime        time.Duration
+	SyncedSleepTime       time.Duration
 }
 
 // NewPeerConnectionOptions returns default initialized PeerConnectionOptions
@@ -47,6 +47,6 @@ func NewPeerConnectionOptions() *PeerConnectionOptions {
 		BlockRequestTimeout:   blockRequestTimeoutDefault,
 		HandshakeRetryTime:    handshakeRetryTimeDefault,
 		SyncedBlockDelta:      syncedBlockDeltaDefault,
-		SyncedPingTime:        syncedPingTimeDefault,
+		SyncedSleepTime:       syncedSleepTimeDefault,
 	}
 }

internal/p2p/gossip.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	log "github.com/koinos/koinos-log-golang/v2"
+	"github.com/koinos/koinos-p2p/internal/options"
 	"github.com/koinos/koinos-p2p/internal/p2perrors"
 	"github.com/koinos/koinos-p2p/internal/rpc"
 	"github.com/koinos/koinos-proto-golang/v2/koinos/canonical"
@@ -159,7 +160,8 @@ type KoinosGossip struct {
 	myPeerID      peer.ID
 	libProvider   LastIrreversibleBlockProvider
 	applicator    *Applicator
-	reportCancel  *context.CancelFunc
+	opts          options.GossipOptions
+	gossipCancel  *context.CancelFunc
 	recentBlocks  uint32
 	recentTrxs    uint32
 }
@@ -172,7 +174,8 @@ func NewKoinosGossip(
 	peerErrorChan chan<- PeerError,
 	id peer.ID,
 	libProvider LastIrreversibleBlockProvider,
-	applicator *Applicator) *KoinosGossip {
+	applicator *Applicator,
+	opts options.GossipOptions) *KoinosGossip {
 
 	block := NewGossipManager(ps, peerErrorChan, BlockTopicName)
 	transaction := NewGossipManager(ps, peerErrorChan, TransactionTopicName)
@@ -185,6 +188,7 @@ func NewKoinosGossip(
 		myPeerID:      id,
 		libProvider:   libProvider,
 		applicator:    applicator,
+		opts:          opts,
 	}
 
 	return &kg
@@ -208,10 +212,10 @@ func (kg *KoinosGossip) EnableGossip(ctx context.Context, enable bool) {
 // StartGossip enables gossip of blocks and transactions
 func (kg *KoinosGossip) StartGossip(ctx context.Context) {
 	log.Info("Starting gossip mode")
-	kg.startBlockGossip(ctx)
-	kg.startTransactionGossip(ctx)
-	reportCtx, reportCancel := context.WithCancel(ctx)
-	kg.reportCancel = &reportCancel
+	gossipCtx, gossipCancel := context.WithCancel(ctx)
+	kg.gossipCancel = &gossipCancel
+	kg.startBlockGossip(gossipCtx)
+	kg.startTransactionGossip(gossipCtx)
 
 	go func() {
 		for {
@@ -223,7 +227,7 @@ func (kg *KoinosGossip) StartGossip(ctx context.Context) {
 				if numBlocks > 0 || numTrxs > 0 {
 					log.Infof("Recently gossiped %v block(s) and %v transaction(s)", numBlocks, numTrxs)
 				}
-			case <-reportCtx.Done():
+			case <-gossipCtx.Done():
 				return
 			}
 		}
@@ -235,9 +239,9 @@ func (kg *KoinosGossip) StopGossip() {
 	log.Info("Stopping gossip mode")
 	kg.block.Stop()
 	kg.transaction.Stop()
-	if kg.reportCancel != nil {
-		(*kg.reportCancel)()
-		kg.reportCancel = nil
+	if kg.gossipCancel != nil {
+		(*kg.gossipCancel)()
+		kg.gossipCancel = nil
 	}
 }
 
@@ -254,7 +258,7 @@ func (kg *KoinosGossip) PublishTransaction(ctx context.Context, transaction *pro
 
 		log.Debugf("Publishing transaction - %s", util.TransactionString(transaction))
 		atomic.AddUint32(&kg.recentTrxs, 1)
-		kg.transaction.PublishMessage(context.Background(), binary)
+		kg.transaction.PublishMessage(ctx, binary)
 	}
 
 	return nil
@@ -273,7 +277,7 @@ func (kg *KoinosGossip) PublishBlock(ctx context.Context, block *protocol.Block)
 
 		log.Debugf("Publishing block - %s", util.BlockString(block))
 		atomic.AddUint32(&kg.recentBlocks, 1)
-		kg.block.PublishMessage(context.Background(), binary)
+		kg.block.PublishMessage(ctx, binary)
 	}
 
 	return nil
@@ -361,8 +365,10 @@ func (kg *KoinosGossip) applyBlock(ctx context.Context, pid peer.ID, msg *pubsub
 
 	log.Debugf("Pushing gossip block - %s from peer %v", util.BlockString(block), msg.ReceivedFrom)
 
-	// TODO: Fix nil argument
-	if err := kg.applicator.ApplyBlock(ctx, block); err != nil {
+	applyBlockContext, cancelApplyBlock := context.WithTimeout(ctx, kg.opts.BlockTimeout)
+	defer cancelApplyBlock()
+
+	if err := kg.applicator.ApplyBlock(applyBlockContext, block); err != nil {
 		return fmt.Errorf("%w - %s, %v", p2perrors.ErrBlockApplication, util.BlockString(block), err.Error())
 	}
 
@@ -426,7 +432,10 @@ func (kg *KoinosGossip) applyTransaction(ctx context.Context, pid peer.ID, msg *
 		return fmt.Errorf("%w, gossiped transaction missing id", p2perrors.ErrDeserialization)
 	}
 
-	if err := kg.applicator.ApplyTransaction(ctx, transaction); err != nil {
+	applyTransactionContext, cancelApplyTransaction := context.WithTimeout(ctx, kg.opts.TransactionTimeout)
+	defer cancelApplyTransaction()
+
+	if err := kg.applicator.ApplyTransaction(applyTransactionContext, transaction); err != nil {
 		if errors.Is(err, p2perrors.ErrInvalidNonce) {
 			return fmt.Errorf("%w - %s, %v", p2perrors.ErrInvalidNonce, util.TransactionString(transaction), err.Error())
 		}

internal/p2p/peer_connection.go

@@ -16,17 +16,13 @@ import (
 	"github.com/multiformats/go-multihash"
 )
 
-type signalRequestBlocks struct{}
-
 // PeerConnection handles the sync portion of a connection to a peer
 type PeerConnection struct {
 	id       peer.ID
 	version  *semver.Version
 	isSynced bool
 	opts     *options.PeerConnectionOptions
 
-	requestBlockChan chan signalRequestBlocks
-
 	libProvider     LastIrreversibleBlockProvider
 	localRPC        rpc.LocalRPC
 	peerRPC         rpc.RemoteRPC
@@ -35,10 +31,6 @@ type PeerConnection struct {
 	versionProvider ProtocolVersionProvider
 }
 
-func (p *PeerConnection) requestBlocks() {
-	p.requestBlockChan <- signalRequestBlocks{}
-}
-
 func (p *PeerConnection) handshake(ctx context.Context) error {
 	// Check Peer's protocol version
 	version, err := p.versionProvider.GetProtocolVersion(ctx, p.id)
@@ -97,7 +89,7 @@ func (p *PeerConnection) handshake(ctx context.Context) error {
 	return nil
 }
 
-func (p *PeerConnection) handleRequestBlocks(ctx context.Context) error {
+func (p *PeerConnection) requestSyncBlocks(ctx context.Context) error {
 	// Get my last irreversible block
 	lib := p.libProvider.GetLastIrreversibleBlock()
 
@@ -200,26 +192,28 @@ func (p *PeerConnection) handleRequestBlocks(ctx context.Context) error {
 
 func (p *PeerConnection) connectionLoop(ctx context.Context) {
 	for {
+		// Request sync blocks.
+		// If there is an error, report it
+		err := p.requestSyncBlocks(ctx)
+		if err != nil {
+			select {
+			case p.peerErrorChan <- PeerError{id: p.id, err: err}:
+			case <-ctx.Done():
+				return
+			}
+		}
+
+		// Get sleep time if we are synced or not
+		sleepTime := time.Second
+		if p.isSynced {
+			sleepTime = p.opts.SyncedSleepTime
+		}
+
+		// Sleep and then repeat
 		select {
+		case <-time.After(sleepTime):
 		case <-ctx.Done():
 			return
-		case <-p.requestBlockChan:
-			err := p.handleRequestBlocks(ctx)
-			if err != nil {
-				go time.AfterFunc(time.Second, p.requestBlocks)
-				go func() {
-					select {
-					case p.peerErrorChan <- PeerError{id: p.id, err: err}:
-					case <-ctx.Done():
-					}
-				}()
-			} else {
-				if p.isSynced {
-					go time.AfterFunc(p.opts.SyncedPingTime, p.requestBlocks)
-				} else {
-					go p.requestBlocks()
-				}
-			}
 		}
 	}
 }
@@ -240,7 +234,6 @@ func (p *PeerConnection) Start(ctx context.Context) {
 				}()
 			} else {
 				go p.connectionLoop(ctx)
-				go p.requestBlocks()
 				return
 			}
 			select {
@@ -263,15 +256,14 @@ func NewPeerConnection(
 	applicator *Applicator,
 	versionProvider ProtocolVersionProvider) *PeerConnection {
 	return &PeerConnection{
-		id:               id,
-		isSynced:         false,
-		opts:             opts,
-		requestBlockChan: make(chan signalRequestBlocks),
-		libProvider:      libProvider,
-		localRPC:         localRPC,
-		peerRPC:          peerRPC,
-		applicator:       applicator,
-		peerErrorChan:    peerErrorChan,
-		versionProvider:  versionProvider,
+		id:              id,
+		isSynced:        false,
+		opts:            opts,
+		libProvider:     libProvider,
+		localRPC:        localRPC,
+		peerRPC:         peerRPC,
+		applicator:      applicator,
+		peerErrorChan:   peerErrorChan,
+		versionProvider: versionProvider,
 	}
 }