diff --git a/core/txpool/blobpool/blobpool.go b/core/txpool/blobpool/blobpool.go index 4030a0c339..9e6952dad5 100644 --- a/core/txpool/blobpool/blobpool.go +++ b/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 diff --git a/core/txpool/blobpool/blobpool_test.go b/core/txpool/blobpool/blobpool_test.go index 7c57755401..f4a329aad3 100644 --- a/core/txpool/blobpool/blobpool_test.go +++ b/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)