Stream manager

common/stream_services.go

@@ -17,11 +17,6 @@ const (
 
 // Reads the message from the stream and returns a byte of data.
 func ReadMessageFromStream(stream network.Stream) ([]byte, error) {
-	// Set a read deadline
-	if err := stream.SetReadDeadline(time.Now().Add(C_STREAM_TIMEOUT)); err != nil {
-		return nil, errors.Wrap(err, "failed to set read deadline")
-	}
-
 	// First read the length of the incoming message
 	lenBytes := make([]byte, 4)
 	if _, err := io.ReadFull(stream, lenBytes); err != nil {

core/tx_pool.go

@@ -815,7 +815,10 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (replaced bool, err e
 		}
 	}
 	// Try to replace an existing transaction in the pending pool
-	from, _ := types.Sender(pool.signer, tx) // already validated
+	from, err := types.Sender(pool.signer, tx) // already validated
+	if err != nil {
+		return false, err
+	}
 	internal, err := from.InternalAddress()
 	if err != nil {
 		return false, err
@@ -876,7 +879,10 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (replaced bool, err e
 // Note, this method assumes the pool lock is held!
 func (pool *TxPool) enqueueTx(hash common.Hash, tx *types.Transaction, local bool, addAll bool) (bool, error) {
 	// Try to insert the transaction into the future queue
-	from, _ := types.Sender(pool.signer, tx) // already validated
+	from, err := types.Sender(pool.signer, tx) // already validated
+	if err != nil {
+		return false, err
+	}
 	internal, err := from.InternalAddress()
 	if err != nil {
 		return false, err
@@ -1185,7 +1191,10 @@ func (pool *TxPool) Status(hashes []common.Hash) []TxStatus {
 		if tx == nil {
 			continue
 		}
-		from, _ := types.Sender(pool.signer, tx) // already validated
+		from, err := types.Sender(pool.signer, tx) // already validated
+		if err != nil {
+			continue
+		}
 		internal, err := from.InternalAddress()
 		if err != nil {
 			continue
@@ -1222,7 +1231,10 @@ func (pool *TxPool) removeTx(hash common.Hash, outofbound bool) {
 	if tx == nil {
 		return
 	}
-	addr, _ := types.Sender(pool.signer, tx) // already validated during insertion
+	addr, err := types.Sender(pool.signer, tx) // already validated during insertion
+	if err != nil {
+		return
+	}
 	internal, err := addr.InternalAddress()
 	if err != nil {
 		return
@@ -1353,7 +1365,10 @@ func (pool *TxPool) scheduleReorgLoop() {
 		case tx := <-pool.queueTxEventCh:
 			// Queue up the event, but don't schedule a reorg. It's up to the caller to
 			// request one later if they want the events sent.
-			addr, _ := types.Sender(pool.signer, tx)
+			addr, err := types.Sender(pool.signer, tx)
+			if err != nil {
+				continue
+			}
 			internal, err := addr.InternalAddress()
 			if err != nil {
 				pool.logger.WithField("err", err).Debug("Failed to queue transaction")
@@ -1444,7 +1459,10 @@ func (pool *TxPool) runReorg(done chan struct{}, cancel chan struct{}, reset *tx
 
 			// Notify subsystems for newly added transactions
 			for _, tx := range promoted {
-				addr, _ := types.Sender(pool.signer, tx)
+				addr, err := types.Sender(pool.signer, tx)
+				if err != nil {
+					continue
+				}
 				internal, err := addr.InternalAddress()
 				if err != nil {
 					pool.logger.WithField("err", err).Debug("Failed to add transaction event")

core/types/receipt.go

@@ -386,7 +386,10 @@ func (r Receipts) DeriveFields(config *params.ChainConfig, hash common.Hash, num
 		// The contract address can be derived from the transaction itself
 		if r[i].ContractAddress.Equal(common.Address{}) && txs[i].To() == nil {
 			// Deriving the signer is expensive, only do if it's actually needed
-			from, _ := Sender(signer, txs[i])
+			from, err := Sender(signer, txs[i])
+			if err != nil {
+				return err
+			}
 			r[i].ContractAddress = crypto.CreateAddress(from, txs[i].Nonce(), txs[i].Data(), config.Location)
 		}
 		// The used gas can be calculated based on previous r

core/types/stxo.go

@@ -24,6 +24,9 @@ type SpentTxOut struct {
 // countSpentOutputs returns the number of utxos the passed block spends.
 func CountSpentOutputs(block *Block) int {
 	transactions := block.QiTransactions()
+	if len(transactions) == 0 {
+		return 0
+	}
 	if IsCoinBaseTx(transactions[0]) {
 		transactions = transactions[1:]
 	}

core/types/transaction.go

@@ -260,9 +260,6 @@ func (tx *Transaction) ProtoDecode(protoTx *ProtoTransaction, location common.Lo
 		tx.SetInner(&itx)
 
 	case 1:
-		if protoTx.Nonce == nil {
-			return errors.New("missing required field 'Nonce' in ProtoTransaction")
-		}
 		if protoTx.Gas == nil {
 			return errors.New("missing required field 'Gas' in ProtoTransaction")
 		}
@@ -278,12 +275,6 @@ func (tx *Transaction) ProtoDecode(protoTx *ProtoTransaction, location common.Lo
 		if protoTx.To == nil {
 			return errors.New("missing required field 'To' in ProtoTransaction")
 		}
-		if protoTx.GasFeeCap == nil {
-			return errors.New("missing required field 'GasFeeCap' in ProtoTransaction")
-		}
-		if protoTx.GasTipCap == nil {
-			return errors.New("missing required field 'GasTipCap' in ProtoTransaction")
-		}
 
 		var etx ExternalTx
 		etx.AccessList = AccessList{}
@@ -950,7 +941,10 @@ func NewTransactionsByPriceAndNonce(signer Signer, etxs []*Transaction, txs map[
 	}
 
 	for from, accTxs := range txs {
-		acc, _ := Sender(signer, accTxs[0])
+		acc, err := Sender(signer, accTxs[0])
+		if err != nil {
+			continue
+		}
 		wrapped, err := NewTxWithMinerFee(accTxs[0], baseFee, nil)
 		// Remove transaction if sender doesn't match from, or if wrapping fails.
 		if acc.Bytes20() != from || err != nil {

core/types/transaction_signing.go

@@ -106,6 +106,9 @@ func Sender(signer Signer, tx *Transaction) (common.Address, error) {
 	if tx.Type() == ExternalTxType { // External TX does not have a signature
 		return tx.inner.(*ExternalTx).Sender, nil
 	}
+	if tx.Type() == QiTxType {
+		return common.Zero, errors.New("sender field not available for qi type")
+	}
 	if sc := tx.from.Load(); sc != nil {
 		sigCache := sc.(sigCache)
 		// If the signer used to derive from in a previous
@@ -169,6 +172,9 @@ func (s SignerV1) Sender(tx *Transaction) (common.Address, error) {
 	if tx.Type() == ExternalTxType { // External TX does not have a signature
 		return tx.inner.(*ExternalTx).Sender, nil
 	}
+	if tx.Type() == QiTxType {
+		return common.Zero, errors.New("cannot find the sender for a qi transaction type")
+	}
 	V, R, S := tx.GetEcdsaSignatureValues()
 	// DynamicFee txs are defined to use 0 and 1 as their recovery
 	// id, add 27 to become equivalent to unprotected signatures.

core/worker.go

@@ -784,7 +784,10 @@ func (w *worker) commitTransactions(env *environment, txs *types.TransactionsByP
 		// during transaction acceptance is the transaction pool.
 		//
 		// We use the signer regardless of the current hf.
-		from, _ := types.Sender(env.signer, tx)
+		from, err := types.Sender(env.signer, tx)
+		if err != nil {
+			continue
+		}
 		// Start executing the transaction
 		env.state.Prepare(tx.Hash(), env.tcount)
 

p2p/node/api.go

@@ -18,7 +18,6 @@ import (
 
 	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/peer"
-	"github.com/libp2p/go-libp2p/core/protocol"
 
 	"github.com/dominant-strategies/go-quai/common"
 )
@@ -230,8 +229,8 @@ func (p *P2PNode) GetBootPeers() []peer.AddrInfo {
 }
 
 // Opens a new stream to the given peer using the given protocol ID
-func (p *P2PNode) NewStream(peerID peer.ID, protocolID protocol.ID) (network.Stream, error) {
-	return p.Host.NewStream(p.ctx, peerID, protocolID)
+func (p *P2PNode) NewStream(peerID peer.ID) (network.Stream, error) {
+	return p.peerManager.GetStream(peerID)
 }
 
 // Connects to the given peer

p2p/node/node.go

@@ -79,15 +79,9 @@ func NewNode(ctx context.Context) (*P2PNode, error) {
 		return nil, err
 	}
 
-	peerID, err := peer.IDFromPublicKey(getNodeKey().GetPublic())
-	if err != nil {
-		log.Global.Fatalf("error getting self ID: %s", err)
-		return nil, err
-	}
 	// Peer manager handles both connection management and connection gating
 	peerMgr, err := peerManager.NewManager(
 		ctx,
-		peerID,
 		viper.GetInt(utils.MaxPeersFlag.Name), // LowWater
 		2*viper.GetInt(utils.MaxPeersFlag.Name), // HighWater
 		nil,
@@ -183,6 +177,9 @@ func NewNode(ctx context.Context) (*P2PNode, error) {
 	nodeID := host.ID()
 	log.Global.Infof("node created: %s", nodeID)
 
+	// Set peer manager's self ID
+	peerMgr.SetSelfID(nodeID)
+
 	// Create a gossipsub instance with helper functions
 	ps, err := pubsubManager.NewGossipSubManager(ctx, host)
 
@@ -196,7 +193,7 @@ func NewNode(ctx context.Context) (*P2PNode, error) {
 		"headers":      createCache(c_defaultCacheSize),
 	}
 
-	return &P2PNode{
+	p2p := &P2PNode{
 		ctx:            ctx,
 		Host:           host,
 		bootpeers:      bootpeers,
@@ -205,7 +202,12 @@ func NewNode(ctx context.Context) (*P2PNode, error) {
 		peerManager:    peerMgr,
 		requestManager: requestManager.NewManager(),
 		cache:          cache,
-	}, nil
+	}
+
+	// Set the peer manager's backend to the host
+	peerMgr.SetP2PBackend(p2p)
+
+	return p2p, nil
 }
 
 func createCache(size int) *lru.Cache[common.Hash, interface{}] {

p2p/node/p2p_services.go

@@ -4,15 +4,15 @@ import (
 	"errors"
 
 	"github.com/ipfs/go-cid"
-	"github.com/libp2p/go-libp2p/core/network"
+	"github.com/libp2p/go-libp2p/core/host"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/multiformats/go-multihash"
 
 	"github.com/dominant-strategies/go-quai/common"
 	"github.com/dominant-strategies/go-quai/core/types"
 	"github.com/dominant-strategies/go-quai/log"
 	"github.com/dominant-strategies/go-quai/p2p/pb"
-	"github.com/dominant-strategies/go-quai/p2p/protocol"
+	"github.com/dominant-strategies/go-quai/p2p/requestManager"
 	"github.com/dominant-strategies/go-quai/trie"
 )
 
@@ -24,12 +24,14 @@ func (p *P2PNode) requestFromPeer(peerID peer.ID, location common.Location, data
 		"data":     data,
 		"datatype": datatype,
 	}).Trace("Requesting the data from peer")
-	stream, err := p.NewStream(peerID, protocol.ProtocolVersion)
+	stream, err := p.NewStream(peerID)
 	if err != nil {
-		// TODO: should we report this peer for failure to participate?
+		log.Global.WithFields(log.Fields{
+			"peerId": peerID,
+			"error":  err,
+		}).Error("Failed to open stream to peer")
 		return nil, err
 	}
-	defer stream.Close()
 
 	// Get a new request ID
 	id := p.requestManager.CreateRequest()
@@ -40,9 +42,6 @@ func (p *P2PNode) requestFromPeer(peerID peer.ID, location common.Location, data
 		return nil, err
 	}
 
-	// Start listening for the response
-	go p.readLoop(stream, location)
-
 	// Send the request to the peer
 	err = common.WriteMessageToStream(stream, requestBytes)
 	if err != nil {
@@ -90,31 +89,12 @@ func (p *P2PNode) requestFromPeer(peerID peer.ID, location common.Location, data
 	return nil, errors.New("invalid response")
 }
 
-func (p *P2PNode) readLoop(stream network.Stream, location common.Location) {
-	for {
-		message, err := common.ReadMessageFromStream(stream)
-		if err != nil {
-			return
-		}
-
-		recvdID, recvdType, err := pb.DecodeQuaiResponse(message, location)
-		if err != nil {
-			log.Global.WithField(
-				"err", err,
-			).Errorf("error decoding quai response: %s", err)
-			return
-		}
+func (p *P2PNode) GetRequestManager() requestManager.RequestManager {
+	return p.requestManager
+}
 
-		dataChan, err := p.requestManager.GetRequestChan(recvdID)
-		if err != nil {
-			log.Global.WithFields(log.Fields{
-				"requestID": recvdID,
-				"err":       err,
-			}).Error("error associating request ID with data channel")
-			return
-		}
-		dataChan <- recvdType
-	}
+func (p *P2PNode) GetHostBackend() host.Host {
+	return p.Host
 }
 
 // Creates a Cid from a location to be used as DHT key