fix: send fewer transactions if all not included

Author: meyer9

runner/network/sequencer_benchmark.go

@@ -238,10 +238,12 @@ func (nb *sequencerBenchmark) Run(ctx context.Context, metricsCollector metrics.
 
 		blockMetrics := metrics.NewBlockMetrics()
 
+		pendingTxs := 0
+
 		// run for a few blocks
 		for i := 0; i < params.NumBlocks; i++ {
 			blockMetrics.SetBlockNumber(uint64(i) + 1)
-			err := transactionWorker.SendTxs(benchmarkCtx)
+			txsSent, err := transactionWorker.SendTxs(benchmarkCtx, pendingTxs)
 			if err != nil {
 				nb.log.Warn("failed to send transactions", "err", err)
 				errChan <- err
@@ -259,6 +261,17 @@ func (nb *sequencerBenchmark) Run(ctx context.Context, metricsCollector metrics.
 				return
 			}
 
+			// Track how many user txs are still pending in the node's mempool.
+			// payload.Transactions includes the L1 info deposit tx, so user txs = total - 1.
+			userTxsIncluded := len(payload.Transactions) - 1
+			if userTxsIncluded < 0 {
+				userTxsIncluded = 0
+			}
+			pendingTxs = pendingTxs + txsSent - userTxsIncluded
+			if pendingTxs < 0 {
+				pendingTxs = 0
+			}
+
 			time.Sleep(1000 * time.Millisecond)
 
 			err = metricsCollector.Collect(benchmarkCtx, blockMetrics)

runner/payload/contract/contract_worker.go

@@ -286,13 +286,13 @@ func (t *contractPayloadWorker) sendContractTx(ctx context.Context) error {
 	return nil
 }
 
-func (t *contractPayloadWorker) SendTxs(ctx context.Context) error {
+func (t *contractPayloadWorker) SendTxs(ctx context.Context, _ int) (int, error) {
 	for i := 0; i < t.params.CallsPerBlock; i++ {
 		err := t.sendContractTx(ctx)
 
 		if err != nil {
 			t.log.Error("Failed to send transaction", "error", err)
-			return err
+			return 0, err
 		}
 
 		debugResult, err := t.debugContract()
@@ -301,5 +301,5 @@ func (t *contractPayloadWorker) SendTxs(ctx context.Context) error {
 		}
 	}
 
-	return nil
+	return t.params.CallsPerBlock, nil
 }

runner/payload/simulator/worker.go

@@ -574,14 +574,22 @@ func (t *simulatorPayloadWorker) waitForReceipt(ctx context.Context, txHash comm
 	})
 }
 
-func (t *simulatorPayloadWorker) sendTxs(ctx context.Context) error {
+func (t *simulatorPayloadWorker) sendTxs(ctx context.Context, pendingTxs int) (int, error) {
 	txs := make([]*types.Transaction, 0, t.numCallers)
 
 	gas := t.params.GasLimit - 100_000
 
 	sendTxsStartTime := time.Now()
 
-	for i := uint64(0); i < uint64(math.Ceil(float64(t.numCallsPerBlock)*t.scaleFactor)); i++ {
+	targetCalls := uint64(math.Ceil(float64(t.numCallsPerBlock) * t.scaleFactor))
+	callsToSend := targetCalls
+	if uint64(pendingTxs) >= targetCalls {
+		callsToSend = 0
+	} else {
+		callsToSend = targetCalls - uint64(pendingTxs)
+	}
+
+	for i := uint64(0); i < callsToSend; i++ {
 		actual := t.actualNumConfig
 		expected := t.payloadParams.Mul(float64(t.numCalls+1) * t.scaleFactor)
 
@@ -600,7 +608,7 @@ func (t *simulatorPayloadWorker) sendTxs(ctx context.Context) error {
 		transferTx, err := t.createCallTx(t.transactors[callerIdx], t.callerKeys[callerIdx], blockCounts)
 		if err != nil {
 			t.log.Error("Failed to create transfer transaction", "err", err)
-			return err
+			return 0, err
 		}
 
 		t.gasUsedCache[blockCounts.Hash()] = transferTx.Gas()
@@ -625,8 +633,8 @@ func (t *simulatorPayloadWorker) sendTxs(ctx context.Context) error {
 
 	t.mempool.AddTransactions(txs)
 	sendTxsDuration := time.Since(sendTxsStartTime)
-	log.Info("Send transactions duration", "duration", sendTxsDuration, "numCalls", uint64(math.Ceil(float64(t.numCallsPerBlock)*t.scaleFactor)))
-	return nil
+	log.Info("Send transactions duration", "duration", sendTxsDuration, "targetCalls", targetCalls, "callsSent", len(txs), "pendingTxs", pendingTxs)
+	return len(txs), nil
 }
 
 func (t *simulatorPayloadWorker) createCallTx(transactor *bind.TransactOpts, fromPriv *ecdsa.PrivateKey, config *simulatorstats.Stats) (*types.Transaction, error) {
@@ -663,10 +671,11 @@ func (t *simulatorPayloadWorker) createDeployTx(fromPriv *ecdsa.PrivateKey) (*co
 	return &deployAddr, deployTx, nil
 }
 
-func (t *simulatorPayloadWorker) SendTxs(ctx context.Context) error {
-	if err := t.sendTxs(ctx); err != nil {
+func (t *simulatorPayloadWorker) SendTxs(ctx context.Context, pendingTxs int) (int, error) {
+	n, err := t.sendTxs(ctx, pendingTxs)
+	if err != nil {
 		t.log.Error("Failed to send transactions", "err", err)
-		return err
+		return 0, err
 	}
-	return nil
+	return n, nil
 }

runner/payload/transferonly/worker.go

@@ -246,14 +246,16 @@ func (t *transferOnlyPayloadWorker) waitForReceipt(ctx context.Context, txHash c
 	})
 }
 
-func (t *transferOnlyPayloadWorker) sendTxs(ctx context.Context) error {
+func (t *transferOnlyPayloadWorker) sendTxs(ctx context.Context, pendingTxs int) (int, error) {
 	numAccounts := t.numAccounts()
-	gasUsed := uint64(0)
 	txs := make([]*types.Transaction, 0, numAccounts)
 	acctIdx := 0
 
 	randomInt := rand.Uint64()
 
+	// Account for gas that pending transactions (still in the node mempool) will consume.
+	gasUsed := uint64(pendingTxs) * 21000
+
 	for gasUsed < (t.params.GasLimit - 100_000) {
 
 		dest := t.addresses[(acctIdx+1)%numAccounts]
@@ -273,20 +275,19 @@ func (t *transferOnlyPayloadWorker) sendTxs(ctx context.Context) error {
 		transferTx, err := t.createTransferTx(t.privateKeys[acctIdx], t.nextNonce[t.addresses[acctIdx]], dest, big.NewInt(1))
 		if err != nil {
 			t.log.Error("Failed to create transfer transaction", "err", err)
-			return err
+			return 0, err
 		}
 
 		txs = append(txs, transferTx)
 
 		gasUsed += transferTx.Gas()
 
 		t.nextNonce[t.addresses[acctIdx]]++
-		// 21000 gas per transfer
 		acctIdx = (acctIdx + 1) % numAccounts
 	}
 
 	t.mempool.AddTransactions(txs)
-	return nil
+	return len(txs), nil
 }
 
 func (t *transferOnlyPayloadWorker) createTransferTx(fromPriv *ecdsa.PrivateKey, nonce uint64, toAddr common.Address, amount *big.Int) (*types.Transaction, error) {
@@ -305,10 +306,11 @@ func (t *transferOnlyPayloadWorker) createTransferTx(fromPriv *ecdsa.PrivateKey,
 	return tx, nil
 }
 
-func (t *transferOnlyPayloadWorker) SendTxs(ctx context.Context) error {
-	if err := t.sendTxs(ctx); err != nil {
+func (t *transferOnlyPayloadWorker) SendTxs(ctx context.Context, pendingTxs int) (int, error) {
+	n, err := t.sendTxs(ctx, pendingTxs)
+	if err != nil {
 		t.log.Error("Failed to send transactions", "err", err)
-		return err
+		return 0, err
 	}
-	return nil
+	return n, nil
 }

runner/payload/txfuzz/tx_fuzz_worker.go

@@ -81,11 +81,11 @@ func (t *txFuzzPayloadWorker) Stop(ctx context.Context) error {
 	return nil
 }
 
-func (t *txFuzzPayloadWorker) SendTxs(ctx context.Context) error {
+func (t *txFuzzPayloadWorker) SendTxs(ctx context.Context, _ int) (int, error) {
 	t.log.Info("Sending txs in tx-fuzz mode")
-	pendingTxs := t.proxyServer.PendingTxs()
+	pending := t.proxyServer.PendingTxs()
 	t.proxyServer.ClearPendingTxs()
 
-	t.mempool.AddTransactions(pendingTxs)
-	return nil
+	t.mempool.AddTransactions(pending)
+	return len(pending), nil
 }

runner/payload/worker/types.go

@@ -9,7 +9,12 @@ import (
 // Note: Payload workers are responsible keeping track of gas in a block and sending transactions to the mempool.
 type Worker interface {
 	Setup(ctx context.Context) error
-	SendTxs(ctx context.Context) error
+	// SendTxs generates and queues transactions for the next block.
+	// pendingTxs is the number of previously-sent transactions still in the node's
+	// mempool; implementations should reduce their output accordingly so the
+	// mempool stays close to one block's worth of work.
+	// Returns the number of transactions actually queued.
+	SendTxs(ctx context.Context, pendingTxs int) (int, error)
 	Stop(ctx context.Context) error
 	Mempool() mempool.FakeMempool
 }