fix flaky catalyst tests by giving enough time for the pool to reset to the newly-inserted test chain before continuing the test case

core/txpool/legacypool/legacypool.go

@@ -978,6 +978,7 @@ func (pool *LegacyPool) Add(txs []*types.Transaction, sync bool) []error {
 	pool.requestPromoteExecutables(req)
 	if sync {
 		<-req.done
+		fmt.Printf("added %x\n", txs[0].Hash())
 	}
 	return errs
 }
@@ -1204,6 +1205,7 @@ func (pool *LegacyPool) scheduleReorgLoop() {
 
 		select {
 		case req := <-pool.reqResetCh:
+			fmt.Printf("request reset to %d\n", req.newHead.Number)
 			// Reset request: update head if request is already pending.
 			if reset == nil {
 				reset = req
@@ -1215,17 +1217,7 @@ func (pool *LegacyPool) scheduleReorgLoop() {
 			//pool.reorgDoneCh <- nextDone
 
 		case req := <-pool.reqPromoteCh:
-			select {
+			fmt.Println("request promote")
-			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)
-				if _, ok := queuedEvents[addr]; !ok {
-					queuedEvents[addr] = NewSortedMap()
-				}
-				queuedEvents[addr].Put(tx)
-			default:
-			}
 			// Promote request: update address set if request is already pending.
 			if dirtyAccounts == nil {
 				dirtyAccounts = req.set
@@ -1246,24 +1238,18 @@ func (pool *LegacyPool) scheduleReorgLoop() {
 			queuedEvents[addr].Put(tx)
 
 		case <-curDone:
-			fmt.Println("reorg done")
 			for _, ch := range curDones {
-				fmt.Println(ch)
 				close(ch)
 			}
-			fmt.Println("end reorg done")
 			curDone = nil
 
 		case <-pool.reorgShutdownCh:
 			// Wait for current run to finish.
 			if curDone != nil {
-				fmt.Println("shutdown reorg done")
 				<-curDone
 				for _, ch := range curDones {
-					fmt.Println(ch)
 					close(ch)
 				}
-				fmt.Println("end reorg done")
 			}
 			close(nextDone)
 			return
@@ -1324,6 +1310,7 @@ func (pool *LegacyPool) runReorg(done chan struct{}, reset *txpoolResetRequest,
 		for addr, list := range pool.pending {
 			highestPending := list.LastElement()
 			nonces[addr] = highestPending.Nonce() + 1
+			fmt.Printf("set pending nonce %x <- %d\n", addr, highestPending.Nonce()+1)
 		}
 		pool.pendingNonces.setAll(nonces)
 	}
@@ -1465,6 +1452,7 @@ func (pool *LegacyPool) promoteExecutables(accounts []common.Address) []*types.T
 		if list == nil {
 			continue // Just in case someone calls with a non existing account
 		}
+
 		// Drop all transactions that are deemed too old (low nonce)
 		forwards := list.Forward(pool.currentState.GetNonce(addr))
 		for _, tx := range forwards {
@@ -1479,14 +1467,23 @@ func (pool *LegacyPool) promoteExecutables(accounts []common.Address) []*types.T
 		log.Trace("Removed unpayable queued transactions", "count", len(drops))
 		queuedNofundsMeter.Mark(int64(len(drops)))
 
+		fmt.Printf("pending nonce of %x is %d\n", addr, pool.pendingNonces.get(addr))
+		fmt.Printf("in-state nonce of %x is %d\n", addr, pool.currentState.GetNonce(addr))
 		// Gather all executable transactions and promote them
 		readies := list.Ready(pool.pendingNonces.get(addr))
+		fmt.Println(len(readies))
+
 		for _, tx := range readies {
 			hash := tx.Hash()
 			if pool.promoteTx(addr, hash, tx) {
 				promoted = append(promoted, tx)
 			}
 		}
+		fmt.Println(len(promoted))
+		if len(promoted) > 0 {
+			fmt.Println(promoted[0].Hash())
+			fmt.Println(promoted[0].Nonce())
+		}
 		log.Trace("Promoted queued transactions", "count", len(promoted))
 		queuedGauge.Dec(int64(len(readies)))
 

eth/catalyst/api_test.go

@@ -139,7 +139,7 @@ func assembleWithTransactions(api *ConsensusAPI, parentHash common.Hash, params
 			return nil, err
 		}
 		if have, want := len(execData.Transactions), want; have != want {
-			err = fmt.Errorf("invalid number of transactions, have %d want %d", have, want)
+			err = fmt.Errorf("block %d: invalid number of transactions, have %d want %d", execData.Number, have, want)
 			continue
 		}
 		return execData, nil
@@ -448,6 +448,7 @@ func startEthService(t *testing.T, genesis *core.Genesis, blocks []*types.Block)
 		n.Close()
 		t.Fatal("can't import test blocks:", err)
 	}
+	time.Sleep(100 * time.Millisecond)
 
 	ethservice.SetSynced()
 	return n, ethservice