Sequencer calls cmd/transaction-filterer

arbnode/delayed_seq_reorg_test.go

@@ -25,7 +25,8 @@ func TestSequencerReorgFromDelayed(t *testing.T) {
 
 	err = streamer.Start(ctx)
 	Require(t, err)
-	exec.Start(ctx)
+	err = exec.Start(ctx)
+	Require(t, err)
 	init, err := streamer.GetMessage(0)
 	Require(t, err)
 
@@ -225,7 +226,8 @@ func TestSequencerReorgFromLastDelayedMsg(t *testing.T) {
 
 	err = streamer.Start(ctx)
 	Require(t, err)
-	exec.Start(ctx)
+	err = exec.Start(ctx)
+	Require(t, err)
 	init, err := streamer.GetMessage(0)
 	Require(t, err)
 

arbnode/delayed_sequencer.go

@@ -88,7 +88,7 @@ var TestDelayedSequencerConfig = DelayedSequencerConfig{
 	RequireFullFinality:         false,
 	UseMergeFinality:            false,
 	RescanInterval:              time.Millisecond * 100,
-	FilteredTxFullRetryInterval: 30 * time.Second,
+	FilteredTxFullRetryInterval: 1 * time.Second,
 }
 
 func NewDelayedSequencer(l1Reader *headerreader.HeaderReader, reader *InboxReader, exec execution.ExecutionSequencer, coordinator *SeqCoordinator, config DelayedSequencerConfigFetcher) (*DelayedSequencer, error) {

arbnode/inbox_test.go

@@ -142,7 +142,7 @@ func NewTransactionStreamerForTest(t *testing.T, ctx context.Context, ownerAddre
 	}
 
 	transactionStreamerConfigFetcher := func() *TransactionStreamerConfig { return &DefaultTransactionStreamerConfig }
-	execEngine, err := gethexec.NewExecutionEngine(bc, 0, false)
+	execEngine, err := gethexec.NewExecutionEngine(bc, 0, false, nil)
 	if err != nil {
 		Fail(t, err)
 	}
@@ -183,7 +183,8 @@ func TestTransactionStreamer(t *testing.T) {
 
 	err := inbox.Start(ctx)
 	Require(t, err)
-	exec.Start(ctx)
+	err = exec.Start(ctx)
+	Require(t, err)
 
 	maxExpectedGasCost := big.NewInt(l2pricing.InitialBaseFeeWei)
 	maxExpectedGasCost.Mul(maxExpectedGasCost, big.NewInt(2100*2))

changelog/diegoximenes-nit-4252.md

@@ -0,0 +1,2 @@
+### Added
+- Sequencer calls transaction-filterer command if delayed transaction was filtered

cmd/transaction-filterer/api/api.go

@@ -5,6 +5,8 @@ package api
 
 import (
 	"context"
+	"sync"
+	"testing"
 
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/ethereum/go-ethereum/common"
@@ -15,18 +17,22 @@ import (
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/rpc"
 
+	"github.com/offchainlabs/nitro/execution/gethexec"
 	"github.com/offchainlabs/nitro/solgen/go/precompilesgen"
 )
 
-const namespace = "transactionfilterer"
-
 type TransactionFiltererAPI struct {
+	apiMutex sync.Mutex // avoids concurrent transactions with the same nonce
+
 	arbFilteredTransactionsManager *precompilesgen.ArbFilteredTransactionsManager
 	txOpts                         *bind.TransactOpts
 }
 
 // Filter adds the given transaction hash to the filtered transactions set, which is managed by the ArbFilteredTransactionsManager precompile.
 func (t *TransactionFiltererAPI) Filter(ctx context.Context, txHashToFilter common.Hash) (common.Hash, error) {
+	t.apiMutex.Lock()
+	defer t.apiMutex.Unlock()
+
 	txOpts := *t.txOpts
 	txOpts.Context = ctx
 
@@ -41,19 +47,37 @@ func (t *TransactionFiltererAPI) Filter(ctx context.Context, txHashToFilter comm
 	}
 }
 
+// Only used for testing.
+// Sequencer and TransactionFiltererAPI depend on each other, as a workaround for the egg/chicken problem,
+// we set the sequencer client after both are created.
+func (t *TransactionFiltererAPI) SetSequencerClient(_ *testing.T, sequencerClient *ethclient.Client) error {
+	arbFilteredTransactionsManager, err := precompilesgen.NewArbFilteredTransactionsManager(
+		types.ArbFilteredTransactionsManagerAddress,
+		sequencerClient,
+	)
+	if err != nil {
+		return err
+	}
+
+	t.apiMutex.Lock()
+	defer t.apiMutex.Unlock()
+	t.arbFilteredTransactionsManager = arbFilteredTransactionsManager
+	return nil
+}
+
 var DefaultStackConfig = node.Config{
 	DataDir:             "", // ephemeral
 	HTTPPort:            node.DefaultHTTPPort,
 	AuthAddr:            node.DefaultAuthHost,
 	AuthPort:            node.DefaultAuthPort,
 	AuthVirtualHosts:    node.DefaultAuthVhosts,
-	HTTPModules:         []string{namespace},
+	HTTPModules:         []string{gethexec.TransactionFiltererNamespace},
 	HTTPHost:            node.DefaultHTTPHost,
 	HTTPVirtualHosts:    []string{"localhost"},
 	HTTPTimeouts:        rpc.DefaultHTTPTimeouts,
 	WSHost:              node.DefaultWSHost,
 	WSPort:              node.DefaultWSPort,
-	WSModules:           []string{namespace},
+	WSModules:           []string{gethexec.TransactionFiltererNamespace},
 	GraphQLVirtualHosts: []string{"localhost"},
 	P2P: p2p.Config{
 		ListenAddr:  "",
@@ -66,31 +90,31 @@ func NewStack(
 	stackConfig *node.Config,
 	txOpts *bind.TransactOpts,
 	sequencerClient *ethclient.Client,
-) (*node.Node, error) {
+) (*node.Node, *TransactionFiltererAPI, error) {
 	stack, err := node.New(stackConfig)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
 	arbFilteredTransactionsManager, err := precompilesgen.NewArbFilteredTransactionsManager(
 		types.ArbFilteredTransactionsManagerAddress,
 		sequencerClient,
 	)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
 	api := &TransactionFiltererAPI{
 		arbFilteredTransactionsManager: arbFilteredTransactionsManager,
 		txOpts:                         txOpts,
 	}
 	apis := []rpc.API{{
-		Namespace: namespace,
+		Namespace: gethexec.TransactionFiltererNamespace,
 		Version:   "1.0",
 		Service:   api,
 		Public:    true,
 	}}
 	stack.RegisterAPIs(apis)
 
-	return stack, nil
+	return stack, api, nil
 }

cmd/transaction-filterer/main.go

@@ -204,7 +204,7 @@ func mainImpl() int {
 		return 1
 	}
 
-	stack, err := api.NewStack(&stackConf, txOpts, sequencerClient)
+	stack, _, err := api.NewStack(&stackConf, txOpts, sequencerClient)
 	if err != nil {
 		fmt.Fprintf(os.Stderr, "error creating stack: %v\n", err)
 		return 1

consensus/consensusrpcclient/client.go

@@ -35,6 +35,11 @@ func (c *ConsensusRPCClient) Start(ctx_in context.Context) error {
 	return c.client.Start(ctx)
 }
 
+func (c *ConsensusRPCClient) StopAndWait() {
+	c.StopWaiter.StopAndWait()
+	c.client.Close()
+}
+
 func convertError(err error) error {
 	if err == nil {
 		return nil

execution/gethexec/executionengine.go

@@ -169,7 +169,8 @@ type ExecutionEngine struct {
 
 	runningMaintenance atomic.Bool
 
-	addressChecker state.AddressChecker
+	addressChecker               state.AddressChecker
+	transactionFiltererRPCClient *TransactionFiltererRPCClient
 }
 
 func NewL1PriceData() *L1PriceData {
@@ -185,15 +186,22 @@ func init() {
 	}
 }
 
-func NewExecutionEngine(bc *core.BlockChain, syncTillBlock uint64, exposeMultiGas bool) (*ExecutionEngine, error) {
-	return &ExecutionEngine{
-		bc:                bc,
-		resequenceChan:    make(chan []*arbostypes.MessageWithMetadata),
-		newBlockNotifier:  make(chan struct{}, 1),
-		cachedL1PriceData: NewL1PriceData(),
-		exposeMultiGas:    exposeMultiGas,
-		syncTillBlock:     syncTillBlock,
-	}, nil
+func NewExecutionEngine(
+	bc *core.BlockChain,
+	syncTillBlock uint64,
+	exposeMultiGas bool,
+	transactionFiltererRPCClient *TransactionFiltererRPCClient,
+) (*ExecutionEngine, error) {
+	execEngine := &ExecutionEngine{
+		bc:                           bc,
+		resequenceChan:               make(chan []*arbostypes.MessageWithMetadata),
+		newBlockNotifier:             make(chan struct{}, 1),
+		cachedL1PriceData:            NewL1PriceData(),
+		exposeMultiGas:               exposeMultiGas,
+		syncTillBlock:                syncTillBlock,
+		transactionFiltererRPCClient: transactionFiltererRPCClient,
+	}
+	return execEngine, nil
 }
 
 func (s *ExecutionEngine) backlogCallDataUnits() uint64 {
@@ -846,6 +854,20 @@ func (s *ExecutionEngine) createBlockFromNextMessage(msg *arbostypes.MessageWith
 		}
 		// Check if any txs touched filtered addresses but are not in the onchain filter
 		if len(filteringHooks.FilteredTxHashes) > 0 {
+			if s.transactionFiltererRPCClient != nil {
+				s.LaunchThread(func(ctx context.Context) {
+					// Call transaction-filterer sequentially.
+					// To avoid nonce collisions when adding a tx to ArbFilteredTransactionsManager,
+					// transaction-filterer will process only one Filter call at a time.
+					for _, filteredTxHash := range filteringHooks.FilteredTxHashes {
+						_, err := s.transactionFiltererRPCClient.Filter(filteredTxHash).Await(ctx)
+						if err != nil {
+							log.Error("error reporting filtered tx to transaction-filterer", "filteredTxHash", filteredTxHash, "err", err)
+						}
+					}
+				})
+			}
+
 			return nil, nil, nil, &ErrFilteredDelayedMessage{
 				TxHashes:      filteringHooks.FilteredTxHashes,
 				DelayedMsgIdx: msg.DelayedMessagesRead - 1,
@@ -1129,8 +1151,27 @@ func (s *ExecutionEngine) ArbOSVersionForMessageIndex(msgIdx arbutil.MessageInde
 	return containers.NewReadyPromise(extra.ArbOSFormatVersion, nil)
 }
 
-func (s *ExecutionEngine) Start(ctx_in context.Context) {
-	s.StopWaiter.Start(ctx_in, s)
+func (s *ExecutionEngine) StopAndWait() {
+	if s.transactionFiltererRPCClient != nil {
+		s.transactionFiltererRPCClient.StopAndWait()
+	}
+	s.StopWaiter.StopAndWait()
+}
+
+func (s *ExecutionEngine) Start(ctxIn context.Context) error {
+	s.StopWaiter.Start(ctxIn, s)
+
+	ctx, err := s.GetContextSafe()
+	if err != nil {
+		return err
+	}
+
+	if s.transactionFiltererRPCClient != nil {
+		err := s.transactionFiltererRPCClient.Start(ctx)
+		if err != nil {
+			return fmt.Errorf("failed to start transaction filterer RPC client: %w", err)
+		}
+	}
 
 	s.LaunchThread(func(ctx context.Context) {
 		for {
@@ -1186,6 +1227,8 @@ func (s *ExecutionEngine) Start(ctx_in context.Context) {
 			}
 		})
 	}
+
+	return nil
 }
 
 func (s *ExecutionEngine) ShouldTriggerMaintenance(trieLimitBeforeFlushMaintenance time.Duration) bool {