[WIP] add more broadcast logs to scroll-v5.7.11

core/tx_pool.go

@@ -493,6 +493,7 @@ func (pool *TxPool) SetGasPrice(price *big.Int) {
 		// pool.priced is sorted by GasFeeCap, so we have to iterate through pool.all instead
 		drop := pool.all.RemotesBelowTip(price)
 		for _, tx := range drop {
+			log.Error("Dropping transactions below price threshold", "tx", tx.Hash(), "price", price)
 			pool.removeTx(tx.Hash(), false)
 		}
 		pool.priced.Removed(len(drop))
@@ -783,13 +784,13 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (replaced bool, err e
 	// If the transaction is already known, discard it
 	hash := tx.Hash()
 	if pool.all.Get(hash) != nil {
-		log.Trace("Discarding already known transaction", "hash", hash)
+		log.Error("Discarding already known transaction", "hash", hash)
 		knownTxMeter.Mark(1)
 		return false, ErrAlreadyKnown
 	}
 
 	if pool.IsMiner() && rawdb.IsSkippedTransaction(pool.chain.Database(), hash) {
-		log.Trace("Discarding already known skipped transaction", "hash", hash)
+		log.Error("Discarding already known skipped transaction", "hash", hash)
 		knownSkippedTxMeter.Mark(1)
 		return false, ErrAlreadyKnown
 	}
@@ -800,15 +801,15 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (replaced bool, err e
 
 	// If the transaction fails basic validation, discard it
 	if err := pool.validateTx(tx, isLocal); err != nil {
-		log.Trace("Discarding invalid transaction", "hash", hash, "err", err)
+		log.Error("Discarding invalid transaction", "hash", hash, "err", err)
 		invalidTxMeter.Mark(1)
 		return false, err
 	}
 	// If the transaction pool is full, discard underpriced transactions
 	if uint64(pool.all.Slots()+numSlots(tx)) > pool.config.GlobalSlots+pool.config.GlobalQueue {
 		// If the new transaction is underpriced, don't accept it
 		if !isLocal && pool.priced.Underpriced(tx) {
-			log.Trace("Discarding underpriced transaction", "hash", hash, "gasTipCap", tx.GasTipCap(), "gasFeeCap", tx.GasFeeCap())
+			log.Error("Discarding underpriced transaction", "hash", hash, "gasTipCap", tx.GasTipCap(), "gasFeeCap", tx.GasFeeCap())
 			underpricedTxMeter.Mark(1)
 			return false, ErrUnderpriced
 		}
@@ -817,6 +818,7 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (replaced bool, err e
 		// do too many replacements between reorg-runs, so we cap the number of
 		// replacements to 25% of the slots
 		if pool.changesSinceReorg > int(pool.config.GlobalSlots/4) {
+			log.Error("Discarding transaction due to too many changes since reorg", "hash", hash)
 			throttleTxMeter.Mark(1)
 			return false, ErrTxPoolOverflow
 		}
@@ -828,52 +830,67 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (replaced bool, err e
 
 		// Special case, we still can't make the room for the new remote one.
 		if !isLocal && !success {
-			log.Trace("Discarding overflown transaction", "hash", hash)
+			log.Error("Discarding overflown transaction", "hash", hash)
 			overflowedTxMeter.Mark(1)
 			return false, ErrTxPoolOverflow
 		}
 		// Bump the counter of rejections-since-reorg
 		pool.changesSinceReorg += len(drop)
 		// Kick out the underpriced remote transactions.
 		for _, tx := range drop {
-			log.Trace("Discarding freshly underpriced transaction", "hash", tx.Hash(), "gasTipCap", tx.GasTipCap(), "gasFeeCap", tx.GasFeeCap())
+			log.Error("Discarding freshly underpriced transaction", "hash", tx.Hash(), "gasTipCap", tx.GasTipCap(), "gasFeeCap", tx.GasFeeCap())
 			underpricedTxMeter.Mark(1)
 			pool.removeTx(tx.Hash(), false)
 		}
 	}
 	// Try to replace an existing transaction in the pending pool
 	from, _ := types.Sender(pool.signer, tx) // already validated
+
+	log.Info("already validated", "hash", hash, "from", from, "to", tx.To())
+
 	if list := pool.pending[from]; list != nil && list.Overlaps(tx) {
+
+		log.Info("already pending", "hash", hash, "from", from, "to", tx.To())
+
 		// Nonce already pending, check if required price bump is met
 		inserted, old := list.Add(tx, pool.currentState, pool.config.PriceBump, pool.chainconfig, pool.currentHead)
+
+		log.Info("already pending", "hash", hash, "inserted", inserted, "old", old.Hash().Hex())
+
 		if !inserted {
+			log.Error("already pending, ErrReplaceUnderpriced", "hash", hash)
 			pendingDiscardMeter.Mark(1)
 			return false, ErrReplaceUnderpriced
 		}
 		// New transaction is better, replace old one
 		if old != nil {
+			log.Info("already pending, new transaction is better, replace old one", "hash", hash, "old", old.Hash().Hex())
 			pool.all.Remove(old.Hash())
 			pool.calculateTxsLifecycle(types.Transactions{old}, time.Now())
 			pool.priced.Removed(1)
 			pendingReplaceMeter.Mark(1)
 		}
+		log.Info("already pending 1", "hash", hash)
 		pool.all.Add(tx, isLocal)
 		pool.priced.Put(tx, isLocal)
 		pool.journalTx(from, tx)
 		pool.queueTxEvent(tx)
-		log.Trace("Pooled new executable transaction", "hash", hash, "from", from, "to", tx.To())
+		log.Info("Pooled new executable transaction", "hash", hash, "from", from, "to", tx.To())
 
 		// Successful promotion, bump the heartbeat
 		pool.beats[from] = time.Now()
 		return old != nil, nil
 	}
 	// New transaction isn't replacing a pending one, push into queue
+	log.Info("new transaction isn't replacing a pending one, push into queue", "hash", hash, "from", from, "to", tx.To())
 	replaced, err = pool.enqueueTx(hash, tx, isLocal, true)
 	if err != nil {
 		return false, err
 	}
 	// Mark local addresses and journal local transactions
+	log.Info("mark local addresses and journal local transactions", "hash", hash, "from", from, "to", tx.To())
 	if local && !pool.locals.contains(from) {
+		log.Info("Setting new local account", "hash", hash, "from", from, "to", tx.To())
 		log.Info("Setting new local account", "address", from)
 		pool.locals.add(from)
 		pool.priced.Removed(pool.all.RemoteToLocals(pool.locals)) // Migrate the remotes if it's marked as local first time.
@@ -883,7 +900,7 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (replaced bool, err e
 	}
 	pool.journalTx(from, tx)
 
-	log.Trace("Pooled new future transaction", "hash", hash, "from", from, "to", tx.To())
+	log.Info("Pooled new future transaction", "hash", hash, "from", from, "to", tx.To())
 	return replaced, nil
 }
 
@@ -906,6 +923,7 @@ func (pool *TxPool) enqueueTx(hash common.Hash, tx *types.Transaction, local boo
 	// Discard any previous transaction and mark this
 	if old != nil {
 		pool.all.Remove(old.Hash())
+		log.Error("enqueueTx: Discarding any previous transaction", "hash", old.Hash())
 		pool.priced.Removed(1)
 		pool.calculateTxsLifecycle(types.Transactions{old}, time.Now())
 		queuedReplaceMeter.Mark(1)
@@ -955,6 +973,7 @@ func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.T
 	inserted, old := list.Add(tx, pool.currentState, pool.config.PriceBump, pool.chainconfig, pool.currentHead)
 	if !inserted {
 		// An older transaction was better, discard this
+		log.Error("promoteTx: Discarding this transaction, an older transaction was better", "hash", hash)
 		pool.all.Remove(hash)
 		pool.calculateTxsLifecycle(types.Transactions{tx}, time.Now())
 		pool.priced.Removed(1)
@@ -963,6 +982,7 @@ func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.T
 	}
 	// Otherwise discard any previous transaction and mark this
 	if old != nil {
+		log.Error("promoteTx: Discarding any previous transaction and mark this", "old.Hash()", old.Hash())
 		pool.all.Remove(old.Hash())
 		pool.calculateTxsLifecycle(types.Transactions{old}, time.Now())
 		pool.priced.Removed(1)
@@ -1084,6 +1104,7 @@ func (pool *TxPool) addTxsLocked(txs []*types.Transaction, local bool) ([]error,
 		replaced, err := pool.add(tx, local)
 		errs[i] = err
 		if err == nil && !replaced {
+			log.Info("dirty.addTx", "hash", tx.Hash())
 			dirty.addTx(tx)
 		}
 	}
@@ -1145,6 +1166,7 @@ func (pool *TxPool) removeTx(hash common.Hash, outofbound bool) {
 	addr, _ := types.Sender(pool.signer, tx) // already validated during insertion
 
 	// Remove it from the list of known transactions
+	log.Error("removeTx: Remove it from the list of known transactions", "hash", hash)
 	pool.all.Remove(hash)
 	pool.calculateTxsLifecycle(types.Transactions{tx}, time.Now())
 	if outofbound {
@@ -1262,6 +1284,7 @@ func (pool *TxPool) scheduleReorgLoop() {
 			} else {
 				dirtyAccounts.merge(req)
 			}
+			// TODO: print dirtyAccounts related in reorg
 			launchNextRun = true
 			pool.reorgDoneCh <- nextDone
 
@@ -1487,6 +1510,7 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) []*types.Trans
 		forwards := list.Forward(pool.currentState.GetNonce(addr))
 		for _, tx := range forwards {
 			hash := tx.Hash()
+			log.Error("Drop all transactions that are deemed too old (low nonce)", "hash", hash)
 			pool.all.Remove(hash)
 			pool.calculateTxsLifecycle(types.Transactions{tx}, time.Now())
 		}
@@ -1497,6 +1521,7 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) []*types.Trans
 		drops, _ := list.FilterF(costLimit, pool.currentMaxGas, pool.executableTxFilter(costLimit))
 		for _, tx := range drops {
 			hash := tx.Hash()
+			log.Error("Drop all transactions that are too costly (low balance or out of gas)", "hash", hash)
 			pool.all.Remove(hash)
 			pool.calculateTxsLifecycle(types.Transactions{tx}, time.Now())
 		}
@@ -1516,16 +1541,17 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) []*types.Trans
 
 		// Drop all transactions over the allowed limit
 		var caps types.Transactions
-		if !pool.locals.contains(addr) {
-			caps = list.Cap(int(pool.config.AccountQueue))
-			for _, tx := range caps {
-				hash := tx.Hash()
-				pool.all.Remove(hash)
-				pool.calculateTxsLifecycle(types.Transactions{tx}, time.Now())
-				log.Trace("Removed cap-exceeding queued transaction", "hash", hash)
-			}
-			queuedRateLimitMeter.Mark(int64(len(caps)))
-		}
+		// if !pool.locals.contains(addr) {
+		// 	caps = list.Cap(int(pool.config.AccountQueue))
+		// 	for _, tx := range caps {
+		// 		hash := tx.Hash()
+		// 		log.Error("Drop all transactions over the allowed limit", "hash", hash)
+		// 		pool.all.Remove(hash)
+		// 		pool.calculateTxsLifecycle(types.Transactions{tx}, time.Now())
+		// 		log.Trace("Removed cap-exceeding queued transaction", "hash", hash)
+		// 	}
+		// 	queuedRateLimitMeter.Mark(int64(len(caps)))
+		// }
 		// Mark all the items dropped as removed
 		pool.priced.Removed(len(forwards) + len(drops) + len(caps))
 		queuedGauge.Dec(int64(len(forwards) + len(drops) + len(caps)))
@@ -1603,6 +1629,7 @@ func (pool *TxPool) truncatePending() {
 					for _, tx := range caps {
 						// Drop the transaction from the global pools too
 						hash := tx.Hash()
+						log.Error("Drop the transaction from the global pools too 1", "hash", hash)
 						pool.all.Remove(hash)
 						pool.calculateTxsLifecycle(types.Transactions{tx}, time.Now())
 
@@ -1631,6 +1658,7 @@ func (pool *TxPool) truncatePending() {
 				for _, tx := range caps {
 					// Drop the transaction from the global pools too
 					hash := tx.Hash()
+					log.Error("Drop the transaction from the global pools too 2", "hash", hash)
 					pool.all.Remove(hash)
 					pool.calculateTxsLifecycle(types.Transactions{tx}, time.Now())
 
@@ -1713,14 +1741,14 @@ func (pool *TxPool) demoteUnexecutables() {
 			hash := tx.Hash()
 			pool.all.Remove(hash)
 			pool.calculateTxsLifecycle(types.Transactions{tx}, time.Now())
-			log.Trace("Removed old pending transaction", "hash", hash)
+			log.Error("Removed old pending transaction", "hash", hash)
 		}
 		// Drop all transactions that are too costly (low balance or out of gas), and queue any invalids back for later
 		costLimit := pool.currentState.GetBalance(addr)
 		drops, invalids := list.FilterF(costLimit, pool.currentMaxGas, pool.executableTxFilter(costLimit))
 		for _, tx := range drops {
 			hash := tx.Hash()
-			log.Trace("Removed unpayable pending transaction", "hash", hash)
+			log.Error("Removed unpayable pending transaction", "hash", hash)
 			pool.all.Remove(hash)
 			pool.calculateTxsLifecycle(types.Transactions{tx}, time.Now())
 		}

eth/fetcher/tx_fetcher.go

@@ -795,6 +795,10 @@ func (f *TxFetcher) scheduleFetches(timer *mclock.Timer, timeout chan struct{},
 		})
 
 		log.Debug("Scheduling transaction retrieval", "peer", peer, "len(f.announces[peer])", len(f.announces[peer]), "len(hashes)", len(hashes))
+		for _, hash := range hashes {
+			log.Info("Scheduling transaction retrieval", "peer", peer, "len(f.announces[peer])", len(f.announces[peer]), "len(hashes)", len(hashes), "hash", hash)
+		}
+
 		peerAnnounceTxsLenGauge.Update(int64(len(f.announces[peer])))
 		peerRetrievalTxsLenGauge.Update(int64(len(hashes)))
 

eth/handler.go

@@ -519,12 +519,18 @@ func (h *handler) BroadcastTransactions(txs types.Transactions) {
 		directCount += len(hashes)
 		peer.AsyncSendTransactions(hashes)
 		log.Debug("Transactions being broadcasted to", "peer", peer.String(), "len", len(hashes))
+		for _, hash := range hashes {
+			log.Info("Transactions being broadcasted to", "peer", peer.String(), "len", len(hashes), "hash", hash)
+		}
 	}
 	for peer, hashes := range annos {
 		annoPeers++
 		annoCount += len(hashes)
 		peer.AsyncSendPooledTransactionHashes(hashes)
 		log.Debug("Transactions being announced to", "peer", peer.String(), "len", len(hashes))
+		for _, hash := range hashes {
+			log.Info("Transactions being announced to", "peer", peer.String(), "len", len(hashes), "hash", hash)
+		}
 	}
 	log.Debug("Transaction broadcast", "txs", len(txs),
 		"announce packs", annoPeers, "announced hashes", annoCount,

eth/protocols/eth/broadcast.go

@@ -106,9 +106,12 @@ func (p *Peer) broadcastTransactions() {
 				done = make(chan struct{})
 				go func() {
 					log.Debug("Sending transactions", "count", len(txs))
+					for _, tx := range txs {
+						log.Info("Sending transactions", "count", len(txs), "tx", tx.Hash().Hex())
+					}
 					broadcastSendTxsLenGauge.Update(int64(len(txs)))
 					if err := p.SendTransactions(txs); err != nil {
-						log.Debug("Sending transactions", "count", len(txs), "err", err)
+						log.Error("Sending transactions", "count", len(txs), "err", err)
 						broadcastSendTxsFailMeter.Mark(1)
 						fail <- err
 						return
@@ -181,9 +184,12 @@ func (p *Peer) announceTransactions() {
 				done = make(chan struct{})
 				go func() {
 					log.Debug("Sending transaction announcements", "count", len(pending))
+					for _, tx := range pending {
+						log.Info("Sending transaction announcements", "count", len(pending), "tx", tx.Hex())
+					}
 					broadcastAnnoTxsLenGauge.Update(int64(len(pending)))
 					if err := p.sendPooledTransactionHashes(pending); err != nil {
-						log.Debug("Sending transaction announcements", "count", len(pending), "err", err)
+						log.Error("Sending transaction announcements", "count", len(pending), "err", err)
 						broadcastAnnoTxsFailMeter.Mark(1)
 						fail <- err
 						return

eth/protocols/eth/handlers.go

@@ -338,12 +338,15 @@ func handleNewPooledTransactionHashes(backend Backend, msg Decoder, peer *Peer)
 	}
 	ann := new(NewPooledTransactionHashesPacket)
 	if err := msg.Decode(ann); err != nil {
-		log.Debug("Failed to decode `NewPooledTransactionHashesPacket`", "peer", peer.String(), "err", err)
+		log.Error("Failed to decode `NewPooledTransactionHashesPacket`", "peer", peer.String(), "err", err)
 		newPooledTxHashesFailMeter.Mark(1)
 		return fmt.Errorf("%w: message %v: %v", errDecode, msg, err)
 	}
 	// Schedule all the unknown hashes for retrieval
 	log.Debug("handleNewPooledTransactionHashes", "peer", peer.String(), "len(ann)", len(*ann))
+	for _, hash := range *ann {
+		log.Info("handleNewPooledTransactionHashes", "peer", peer.String(), "len(ann)", len(*ann), "hash", hash.Hex())
+	}
 	newPooledTxHashesLenGauge.Update(int64(len(*ann)))
 	for _, hash := range *ann {
 		peer.markTransaction(hash)
@@ -355,12 +358,15 @@ func handleGetPooledTransactions66(backend Backend, msg Decoder, peer *Peer) err
 	// Decode the pooled transactions retrieval message
 	var query GetPooledTransactionsPacket66
 	if err := msg.Decode(&query); err != nil {
-		log.Debug("Failed to decode `GetPooledTransactionsPacket66`", "peer", peer.String(), "err", err)
+		log.Error("Failed to decode `GetPooledTransactionsPacket66`", "peer", peer.String(), "err", err)
 		getPooledTxsFailMeter.Mark(1)
 		return fmt.Errorf("%w: message %v: %v", errDecode, msg, err)
 	}
 	hashes, txs := answerGetPooledTransactions(backend, query.GetPooledTransactionsPacket, peer)
 	log.Debug("handleGetPooledTransactions", "peer", peer.String(), "RequestId", query.RequestId, "len(query)", len(query.GetPooledTransactionsPacket), "retrieved", len(hashes))
+	for _, hash := range hashes {
+		log.Info("handleGetPooledTransactions", "peer", peer.String(), "RequestId", query.RequestId, "len(query)", len(query.GetPooledTransactionsPacket), "retrieved", len(hashes), "hash", hash.Hex())
+	}
 	getPooledTxsQueryLenGauge.Update(int64(len(query.GetPooledTransactionsPacket)))
 	getPooledTxsRetrievedLenGauge.Update(int64(len(hashes)))
 	return peer.ReplyPooledTransactionsRLP(query.RequestId, hashes, txs)
@@ -403,16 +409,19 @@ func handleTransactions(backend Backend, msg Decoder, peer *Peer) error {
 	var txs TransactionsPacket
 	if err := msg.Decode(&txs); err != nil {
 		handleTxsFailMeter.Mark(1)
-		log.Debug("Failed to decode `TransactionsPacket`", "peer", peer.String(), "err", err)
+		log.Error("Failed to decode `TransactionsPacket`", "peer", peer.String(), "err", err)
 		return fmt.Errorf("%w: message %v: %v", errDecode, msg, err)
 	}
 	log.Debug("handleTransactions", "peer", peer.String(), "len(txs)", len(txs))
+	for _, tx := range txs {
+		log.Info("handleTransactions", "peer", peer.String(), "len(txs)", len(txs), "tx", tx.Hash().Hex())
+	}
 	handleTxsLenGauge.Update(int64(len(txs)))
 	for i, tx := range txs {
 		// Validate and mark the remote transaction
 		if tx == nil {
 			handleTxsNilMeter.Mark(1)
-			log.Debug("handleTransactions: transaction is nil", "peer", peer.String(), "i", i)
+			log.Error("handleTransactions: transaction is nil", "peer", peer.String(), "i", i)
 			return fmt.Errorf("%w: transaction %d is nil", errDecode, i)
 		}
 		peer.markTransaction(tx.Hash())
@@ -429,7 +438,7 @@ func handlePooledTransactions66(backend Backend, msg Decoder, peer *Peer) error
 	var txs PooledTransactionsPacket66
 	if err := msg.Decode(&txs); err != nil {
 		pooledTxs66FailMeter.Mark(1)
-		log.Debug("Failed to decode `PooledTransactionsPacket66`", "peer", peer.String(), "err", err)
+		log.Error("Failed to decode `PooledTransactionsPacket66`", "peer", peer.String(), "err", err)
 		return fmt.Errorf("%w: message %v: %v", errDecode, msg, err)
 	}
 	log.Debug("handlePooledTransactions66", "peer", peer.String(), "len(txs)", len(txs.PooledTransactionsPacket))
@@ -438,7 +447,7 @@ func handlePooledTransactions66(backend Backend, msg Decoder, peer *Peer) error
 		// Validate and mark the remote transaction
 		if tx == nil {
 			pooledTxs66NilMeter.Mark(1)
-			log.Debug("handlePooledTransactions: transaction is nil", "peer", peer.String(), "i", i)
+			log.Error("handlePooledTransactions: transaction is nil", "peer", peer.String(), "i", i)
 			return fmt.Errorf("%w: transaction %d is nil", errDecode, i)
 		}
 		peer.markTransaction(tx.Hash())

eth/protocols/eth/peer.go

@@ -426,6 +426,10 @@ func (p *Peer) RequestTxs(hashes []common.Hash) error {
 	id := rand.Uint64()
 
 	log.Debug("Requesting transactions", "RequestId", id, "Peer.id", p.id, "count", len(hashes))
+	for _, hash := range hashes {
+		log.Info("Requesting transaction", "RequestId", id, "Peer.id", p.id, "count", len(hashes), "hash", hash)
+	}
+
 	peerRequestTxsCntGauge.Update(int64(len(hashes)))
 
 	requestTracker.Track(p.id, p.version, GetPooledTransactionsMsg, PooledTransactionsMsg, id)