Merge 173e211d66 into 7155c65abb

core/txpool/blobpool/blobpool.go

@@ -1577,10 +1577,10 @@ func (p *BlobPool) add(tx *types.Transaction) (err error) {
 	return p.addLocked(tx, true)
 }
 
-// addLocked inserts a new blob transaction into the pool if it passes validation (both
+// addLockedInternal inserts a new blob transaction into the pool if it passes validation (both
 // consensus validity and pool restrictions). It must be called with the pool lock held.
 // Only for internal use.
-func (p *BlobPool) addLocked(tx *types.Transaction, checkGapped bool) (err error) {
+func (p *BlobPool) addLockedInternal(tx *types.Transaction, checkGapped bool) (err error) {
 	// Ensure the transaction is valid from all perspectives
 	if err := p.validateTx(tx); err != nil {
 		log.Trace("Transaction validation failed", "hash", tx.Hash(), "err", err)
@@ -1593,21 +1593,6 @@ func (p *BlobPool) addLocked(tx *types.Transaction, checkGapped bool) (err error
 			addStaleMeter.Mark(1)
 		case errors.Is(err, core.ErrNonceTooHigh):
 			addGappedMeter.Mark(1)
-			// Store the tx in memory, and revalidate later
-			from, _ := types.Sender(p.signer, tx)
-			allowance := p.gappedAllowance(from)
-			if allowance >= 1 && len(p.gapped) < maxGapped {
-				p.gapped[from] = append(p.gapped[from], tx)
-				p.gappedSource[tx.Hash()] = from
-				log.Trace("added tx to gapped blob queue", "allowance", allowance, "hash", tx.Hash(), "from", from, "nonce", tx.Nonce(), "qlen", len(p.gapped[from]))
-				return nil
-			} else {
-				// if maxGapped is reached, it is better to give time to gapped
-				// transactions by keeping the old and dropping this one.
-				// Thus replacing a gapped transaction with another gapped transaction
-				// is discouraged.
-				log.Trace("no gapped blob queue allowance", "allowance", allowance, "hash", tx.Hash(), "from", from, "nonce", tx.Nonce(), "qlen", len(p.gapped[from]))
-			}
 		case errors.Is(err, core.ErrInsufficientFunds):
 			addOverdraftedMeter.Mark(1)
 		case errors.Is(err, txpool.ErrAccountLimitExceeded):
@@ -1737,6 +1722,37 @@ func (p *BlobPool) addLocked(tx *types.Transaction, checkGapped bool) (err error
 			heap.Fix(p.evict, p.evict.index[from])
 		}
 	}
+	return nil
+}
+
+// addLocked inserts a new blob transaction into the pool if it passes validation (both
+// consensus validity and pool restrictions). It handles gapped transactions, underpriced
+// transactions, pool limits, and notifications.
+// It must be called with the pool lock held. Only for internal use.
+func (p *BlobPool) addLocked(tx *types.Transaction, checkGapped bool) (err error) {
+	// check for nonce gaps and add to gapped buffer if needed.
+	if err := p.addLockedInternal(tx, checkGapped); err != nil {
+		if errors.Is(err, core.ErrNonceTooHigh) {
+			addGappedMeter.Mark(1)
+			// Store the tx in memory, and revalidate later
+			from, _ := types.Sender(p.signer, tx) // already validated in addLockedInternal/ValidateTransactionWithState
+			allowance := p.gappedAllowance(from)
+			if allowance >= 1 && len(p.gapped) < maxGapped {
+				p.gapped[from] = append(p.gapped[from], tx)
+				p.gappedSource[tx.Hash()] = from
+				log.Trace("added tx to gapped blob queue", "allowance", allowance, "hash", tx.Hash(), "from", from, "nonce", tx.Nonce(), "qlen", len(p.gapped[from]))
+				return nil
+			} else {
+				// if maxGapped is reached, it is better to give time to gapped
+				// transactions by keeping the old and dropping this one.
+				// Thus replacing a gapped transaction with another gapped transaction
+				// is discouraged.
+				log.Trace("no gapped blob queue allowance", "allowance", allowance, "hash", tx.Hash(), "from", from, "nonce", tx.Nonce(), "qlen", len(p.gapped[from]))
+			}
+		}
+		return err
+	}
+
 	// If the pool went over the allowed data limit, evict transactions until
 	// we're again below the threshold
 	for p.stored > p.config.Datacap {
@@ -1744,6 +1760,14 @@ func (p *BlobPool) addLocked(tx *types.Transaction, checkGapped bool) (err error
 	}
 	p.updateStorageMetrics()
 
+	// If we've just dropped the added transaction, it was clearly underpriced.
+	// We could also try to check for this before adding to the tx pool, but it is
+	// complex because of database internals (RLP encoding, billy shelves).
+	if !p.lookup.exists(tx.Hash()) {
+		addUnderpricedMeter.Mark(1)
+		return txpool.ErrUnderpriced
+	}
+
 	addValidMeter.Mark(1)
 
 	// Transaction was addded successfully, but we only announce if it is (close to being)
@@ -1757,6 +1781,7 @@ func (p *BlobPool) addLocked(tx *types.Transaction, checkGapped bool) (err error
 	}
 
 	//check the gapped queue for this account and try to promote
+	from, _ := types.Sender(p.signer, tx) // already validated in addLockedInternal/ValidateTransactionWithState
 	if gtxs, ok := p.gapped[from]; checkGapped && ok && len(gtxs) > 0 {
 		// We have to add in nonce order, but we want to stable sort to cater for situations
 		// where transactions are replaced, keeping the original receive order for same nonce

core/txpool/blobpool/blobpool_test.go

@@ -1364,9 +1364,10 @@ func TestAdd(t *testing.T) {
 	}
 
 	tests := []struct {
-		seeds map[string]seed
-		adds  []addtx
-		block []addtx
+		seeds   map[string]seed
+		adds    []addtx
+		block   []addtx
+		datacap uint64
 	}{
 		// Transactions from new accounts should be accepted if their initial
 		// nonce matches the expected one from the statedb. Higher or lower must
@@ -1720,6 +1721,66 @@ func TestAdd(t *testing.T) {
 				},
 			},
 		},
+		// Transactions above the Datacap should be rejected
+		{
+			seeds: map[string]seed{
+				"alice": {balance: 10000000},
+			},
+			datacap: 1 * (txAvgSize + blobSize + uint64(txBlobOverhead)), // only allow 1 blob
+			adds: []addtx{
+				{ // Fits in capacity
+					from: "alice",
+					tx:   makeUnsignedTx(0, 1, 1, 1),
+				},
+				{ // Beyond capacity
+					from: "alice",
+					tx:   makeUnsignedTx(1, 1, 1, 1),
+					err:  txpool.ErrUnderpriced,
+				},
+			},
+		},
+		// Transactions above the Datacap should be rejected, use eviction values
+		{
+			seeds: map[string]seed{
+				"alice": {
+					balance: 2000000,
+					//nonce:   1,
+					txs: []*types.BlobTx{
+						makeUnsignedTxWithTestBlob(0, 2, 2, 2, 0),
+						makeUnsignedTxWithTestBlob(1, 2, 2, 2, 1),
+					},
+				},
+				"bob": {
+					balance: 1000000,
+					//nonce:   1,
+					txs: []*types.BlobTx{
+						makeUnsignedTxWithTestBlob(0, 3, 3, 3, 2),
+					},
+				},
+			},
+			datacap: 3 * (txAvgSize + blobSize + uint64(txBlobOverhead)), // only allow 2 blobs
+			adds: []addtx{
+				{ // Beyond capacity, but kicking out one from Alice should make it fit
+					from: "bob",
+					tx:   makeUnsignedTxWithTestBlob(1, 3, 3, 3, 3),
+				},
+				{ // We've just kicked our nonce 1, so this is nonce too high
+					from: "alice",
+					tx:   makeUnsignedTxWithTestBlob(2, 1, 2, 2, 4),
+					err:  core.ErrNonceTooHigh,
+				},
+				{ // This should not succeed, fees are low
+					from: "alice",
+					tx:   makeUnsignedTxWithTestBlob(1, 1, 1, 1, 4),
+					err:  txpool.ErrUnderpriced,
+				},
+				{ // This should also not succeed, because of rolling fee calculation
+					from: "alice",
+					tx:   makeUnsignedTxWithTestBlob(1, 1, 1, 1, 4),
+					err:  txpool.ErrUnderpriced,
+				},
+			},
+		},
 	}
 	for i, tt := range tests {
 		// Create a temporary folder for the persistent backend
@@ -1760,12 +1821,17 @@ func TestAdd(t *testing.T) {
 			blobfee: uint256.NewInt(1),
 			statedb: statedb,
 		}
-		pool := New(Config{Datadir: storage}, chain, nil)
+		pool := New(Config{Datadir: storage, Datacap: tt.datacap}, chain, nil)
 		if err := pool.Init(1, chain.CurrentBlock(), newReserver()); err != nil {
 			t.Fatalf("test %d: failed to create blob pool: %v", i, err)
 		}
 		verifyPoolInternals(t, pool)
 
+		// subscibe to pool events to verify they are emitted correctly
+		txsCh := make(chan core.NewTxsEvent, 1)
+		txsSub := pool.SubscribeTransactions(txsCh, false)
+		defer txsSub.Unsubscribe()
+
 		// Add each transaction one by one, verifying the pool internals in between
 		for j, add := range tt.adds {
 			signed, _ := types.SignNewTx(keys[add.from], types.LatestSigner(params.MainnetChainConfig), add.tx)
@@ -1796,6 +1862,33 @@ func TestAdd(t *testing.T) {
 			}
 			// Verify the pool internals after each addition
 			verifyPoolInternals(t, pool)
+			// verify that if the tx was added, an event was emitted
+			txStatus := pool.Status(signed.Hash())
+			select {
+			case ev := <-txsCh:
+				switch {
+				case add.err == nil && txStatus == txpool.TxStatusPending:
+					if len(ev.Txs) != 1 {
+						t.Errorf("test %d, tx %d: event txs length mismatch: have %d, want 1", i, j, len(ev.Txs))
+					}
+					if ev.Txs[0].Hash() != signed.Hash() {
+						t.Errorf("test %d, tx %d: event tx mismatch: have %v, want %v", i, j, ev.Txs[0].Hash(), signed.Hash())
+					}
+				case add.err == nil && txStatus == txpool.TxStatusQueued:
+					t.Errorf("test %d, tx %d: unexpected new tx event for queued tx", i, j)
+				case add.err != nil:
+					t.Errorf("test %d, tx %d: unexpected new tx event for failed tx", i, j)
+				default:
+					t.Errorf("test %d, tx %d: unexpected test result", i, j)
+				}
+			default:
+				switch {
+				case add.err == nil && txStatus == txpool.TxStatusPending:
+					t.Errorf("test %d, tx %d: expected new tx event, none received", i, j)
+				default:
+					// expected no event
+				}
+			}
 		}
 		verifyPoolInternals(t, pool)