Report filtered delayed transactions to filtering-report service - NIT-4644

arbos/block_processor.go

@@ -10,6 +10,7 @@ import (
 	"math"
 	"math/big"
 
+	"github.com/ethereum/go-ethereum/arbitrum/filter"
 	"github.com/ethereum/go-ethereum/arbitrum_types"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core"
@@ -42,9 +43,16 @@ var EmitTicketCreatedEvent func(*vm.EVM, [32]byte) error
 
 // ErrFilteredCascadingRedeem is returned via TxFailed when a redeem's
 // inner execution touches a filtered address, requiring the entire tx group
-// (originating user tx + all its redeems) to be reverted.
+// (originating user tx + all its redeems) to be reverted. All fields are
+// captured before the group rollback so TxFailed can build a fully populated
+// FilteredTxReport without late-filling.
 type ErrFilteredCascadingRedeem struct {
 	OriginatingTxHash common.Hash
+	OriginatingTx     *types.Transaction
+	FilteredAddresses []filter.FilteredAddressRecord
+	BlockNumber       uint64
+	ParentBlockHash   common.Hash
+	PositionInBlock   int // receipt index of the originating user tx
 }
 
 func (e *ErrFilteredCascadingRedeem) Error() string {
@@ -213,10 +221,10 @@ type SequencingHooks interface {
 	// SupportsGroupRollback returns whether the hooks support checkpointing and
 	// rolling back a group of transactions (user tx + its scheduled redeems).
 	SupportsGroupRollback() bool
-	// PreTxFilter rejects a tx before execution.
-	PreTxFilter(*params.ChainConfig, *types.Header, *state.StateDB, *arbosState.ArbosState, *types.Transaction, *arbitrum_types.ConditionalOptions, common.Address, *L1Info) error
-	// PostTxFilter rejects a tx after execution.
-	PostTxFilter(*types.Header, *state.StateDB, *arbosState.ArbosState, *types.Transaction, common.Address, uint64, *core.ExecutionResult) error
+	// PreTxFilter rejects a tx before execution. positionInBlock is len(receipts) at call time.
+	PreTxFilter(*params.ChainConfig, *types.Header, *state.StateDB, *arbosState.ArbosState, *types.Transaction, *arbitrum_types.ConditionalOptions, common.Address, *L1Info, int) error
+	// PostTxFilter rejects a tx after execution. positionInBlock is len(receipts) at call time.
+	PostTxFilter(*types.Header, *state.StateDB, *arbosState.ArbosState, *types.Transaction, common.Address, uint64, *core.ExecutionResult, int) error
 	// BlockFilter rejects an entire block after all txs have been applied.
 	BlockFilter(*types.Header, *state.StateDB, types.Transactions, types.Receipts) error
 	// TxSucceeded records that the last user tx from NextTxToSequence executed successfully.
@@ -244,11 +252,11 @@ func (n *NoopSequencingHooks) NextTxToSequence() (*types.Transaction, *arbitrum_
 
 func (n *NoopSequencingHooks) CanDiscardTx() bool { return false }
 
-func (n *NoopSequencingHooks) PreTxFilter(config *params.ChainConfig, header *types.Header, db *state.StateDB, a *arbosState.ArbosState, transaction *types.Transaction, options *arbitrum_types.ConditionalOptions, address common.Address, info *L1Info) error {
+func (n *NoopSequencingHooks) PreTxFilter(config *params.ChainConfig, header *types.Header, db *state.StateDB, a *arbosState.ArbosState, transaction *types.Transaction, options *arbitrum_types.ConditionalOptions, address common.Address, info *L1Info, positionInBlock int) error {
 	return nil
 }
 
-func (n *NoopSequencingHooks) PostTxFilter(header *types.Header, db *state.StateDB, a *arbosState.ArbosState, transaction *types.Transaction, address common.Address, u uint64, result *core.ExecutionResult) error {
+func (n *NoopSequencingHooks) PostTxFilter(header *types.Header, db *state.StateDB, a *arbosState.ArbosState, transaction *types.Transaction, address common.Address, u uint64, result *core.ExecutionResult, positionInBlock int) error {
 	return nil
 }
 
@@ -426,7 +434,7 @@ func ProduceBlockAdvanced(
 
 			// Writes to statedb object should be avoided to prevent invalid state from permeating as statedb snapshot is not taken
 			if isUserTx {
-				if err = sequencingHooks.PreTxFilter(chainConfig, header, buildState.statedb, buildState.arbState, tx, options, sender, l1Info); err != nil {
+				if err = sequencingHooks.PreTxFilter(chainConfig, header, buildState.statedb, buildState.arbState, tx, options, sender, l1Info, len(buildState.receipts)); err != nil {
 					return nil, nil, err
 				}
 			}
@@ -489,7 +497,7 @@ func ProduceBlockAdvanced(
 				&header.GasUsed,
 				runCtx,
 				func(result *core.ExecutionResult) error {
-					if err := sequencingHooks.PostTxFilter(header, buildState.statedb, buildState.arbState, tx, sender, dataGas, result); err != nil {
+					if err := sequencingHooks.PostTxFilter(header, buildState.statedb, buildState.arbState, tx, sender, dataGas, result, len(buildState.receipts)); err != nil {
 						return err
 					}
 					// Additional post-transaction validity check
@@ -519,11 +527,21 @@ func ProduceBlockAdvanced(
 			// active group checkpoint, roll back the entire group (user tx + all
 			// redeems) to the pre-group state.
 			if !isUserTx && buildState.activeGroupCP != nil && errors.Is(err, state.ErrArbTxFilter) {
-				userTxHash := buildState.activeGroupCP.userTxHash
+				// Capture everything before rollback — addressCheckerStateß
+				cp := buildState.activeGroupCP
+				_, filteredAddresses := buildState.statedb.IsAddressFiltered()
+				originatingTx := buildState.complete[cp.completeLen]
 				if err := buildState.rollbackToGroupCheckpoint(header); err != nil {
 					return nil, nil, nil, err
 				}
-				sequencingHooks.TxFailed(&ErrFilteredCascadingRedeem{OriginatingTxHash: userTxHash})
+				sequencingHooks.TxFailed(&ErrFilteredCascadingRedeem{
+					OriginatingTxHash: cp.userTxHash,
+					OriginatingTx:     originatingTx,
+					FilteredAddresses: filteredAddresses,
+					BlockNumber:       header.Number.Uint64(),
+					ParentBlockHash:   header.ParentHash,
+					PositionInBlock:   cp.receiptsLen,
+				})
 				continue
 			}
 			if isUserTx {

changelog/mnasr-nit-4644.md

@@ -0,0 +1,2 @@
+### Added
+- Report filtered delayed transactions to filtering-report service with structured FilteredTxReport

cmd/filtering-report/api/api.go

@@ -5,15 +5,20 @@ package api
 
 import (
 	"net/http"
+	"sync"
 
 	"github.com/ethereum/go-ethereum/node"
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/rpc"
 
 	"github.com/offchainlabs/nitro/execution/gethexec"
+	"github.com/offchainlabs/nitro/execution/gethexec/addressfilter"
 )
 
-type FilteringReportAPI struct{}
+type FilteringReportAPI struct {
+	mu              sync.Mutex
+	receivedReports []addressfilter.FilteredTxReport
+}
 
 var DefaultStackConfig = node.Config{
 	DataDir:             "", // ephemeral
@@ -36,16 +41,17 @@ var DefaultStackConfig = node.Config{
 	},
 }
 
-func NewStack(stackConfig *node.Config) (*node.Node, error) {
+func NewStack(stackConfig *node.Config) (*node.Node, *FilteringReportAPI, error) {
 	stack, err := node.New(stackConfig)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
+	reportAPI := &FilteringReportAPI{}
 	apis := []rpc.API{{
 		Namespace: gethexec.FilteringReportNamespace,
 		Version:   "1.0",
-		Service:   &FilteringReportAPI{},
+		Service:   reportAPI,
 		Public:    true,
 	}}
 	stack.RegisterAPIs(apis)
@@ -57,5 +63,5 @@ func NewStack(stackConfig *node.Config) (*node.Node, error) {
 		w.WriteHeader(http.StatusOK)
 	}))
 
-	return stack, nil
+	return stack, reportAPI, nil
 }

cmd/filtering-report/api/api_test.go

@@ -23,7 +23,7 @@ func newTestStack(t *testing.T) *node.Node {
 	stackConfig.HTTPPort = 0
 	stackConfig.WSHost = "127.0.0.1"
 	stackConfig.WSPort = 0
-	stack, err := NewStack(&stackConfig)
+	stack, _, err := NewStack(&stackConfig)
 	if err != nil {
 		t.Fatal(err)
 	}

cmd/filtering-report/api/report_filtered_transactions.go

@@ -5,13 +5,18 @@ package api
 
 import (
 	"context"
+	"slices"
 
 	"github.com/ethereum/go-ethereum/log"
 
 	"github.com/offchainlabs/nitro/execution/gethexec/addressfilter"
 )
 
 func (a *FilteringReportAPI) ReportFilteredTransactions(_ context.Context, reports []addressfilter.FilteredTxReport) error {
+	a.mu.Lock()
+	a.receivedReports = append(a.receivedReports, reports...)
+	a.mu.Unlock()
+
 	for _, report := range reports {
 		log.Info("Filtered transaction report",
 			"id", report.ID,
@@ -32,3 +37,10 @@ func (a *FilteringReportAPI) ReportFilteredTransactions(_ context.Context, repor
 	}
 	return nil
 }
+
+// ReceivedReports returns a copy of all reports received so far.
+func (a *FilteringReportAPI) ReceivedReports() []addressfilter.FilteredTxReport {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	return slices.Clone(a.receivedReports)
+}

cmd/filtering-report/main.go

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

execution/gethexec/executionengine.go

@@ -32,6 +32,7 @@ import (
 	"github.com/ethereum/go-ethereum/arbitrum/filter"
 	"github.com/ethereum/go-ethereum/arbitrum/multigas"
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/rawdb"
 	"github.com/ethereum/go-ethereum/core/state"
@@ -50,6 +51,7 @@ import (
 	"github.com/offchainlabs/nitro/arbutil"
 	"github.com/offchainlabs/nitro/consensus"
 	"github.com/offchainlabs/nitro/execution"
+	"github.com/offchainlabs/nitro/execution/gethexec/addressfilter"
 	"github.com/offchainlabs/nitro/execution/gethexec/eventfilter"
 	"github.com/offchainlabs/nitro/util/arbmath"
 	"github.com/offchainlabs/nitro/util/containers"
@@ -94,19 +96,21 @@ func (e *ErrFilteredDelayedMessage) Error() string {
 var ErrDelayedTxFiltered = errors.New("delayed transaction filtered")
 
 // DelayedFilteringSequencingHooks extends NoopSequencingHooks with address filtering
-// for delayed message processing. Collects all tx hashes that touch filtered addresses
-// and are not in the onchain filter. After block production, the caller checks if any
-// hashes were collected and returns ErrFilteredDelayedMessage if so.
+// for delayed message processing. Builds FilteredTxReport entries for txs that touch
+// filtered addresses and are not in the onchain filter. After block production, the
+// caller checks pendingFilteredTxReports and returns ErrFilteredDelayedMessage if any.
 type DelayedFilteringSequencingHooks struct {
 	arbos.NoopSequencingHooks
-	FilteredTxHashes []common.Hash
-	eventFilter      *eventfilter.EventFilter
+	pendingFilteredTxReports []addressfilter.FilteredTxReport
+	eventFilter              *eventfilter.EventFilter
+	inboxRequestId           common.Hash
 }
 
-func NewDelayedFilteringSequencingHooks(txes types.Transactions, ef *eventfilter.EventFilter) *DelayedFilteringSequencingHooks {
+func NewDelayedFilteringSequencingHooks(txes types.Transactions, ef *eventfilter.EventFilter, inboxRequestId common.Hash) *DelayedFilteringSequencingHooks {
 	return &DelayedFilteringSequencingHooks{
 		NoopSequencingHooks: *arbos.NewNoopSequencingHooks(txes),
 		eventFilter:         ef,
+		inboxRequestId:      inboxRequestId,
 	}
 }
 
@@ -127,18 +131,20 @@ func touchAddresses(db *state.StateDB, tx *types.Transaction, sender common.Addr
 }
 
 // PostTxFilter touches To/From addresses and checks IsAddressFiltered.
-// Collects tx hashes that touch filtered addresses but are not in the onchain filter.
-// For redeems, returns ErrArbTxFilter to trigger group rollback.
-func (f *DelayedFilteringSequencingHooks) PostTxFilter(header *types.Header, db *state.StateDB, a *arbosState.ArbosState, tx *types.Transaction, sender common.Address, dataGas uint64, result *core.ExecutionResult) error {
+// Builds a FilteredTxReport and returns ErrArbTxFilter for filtered txs.
+// For redeems, returns ErrArbTxFilter without a report (originating tx is
+// collected in TxFailed after group rollback).
+func (f *DelayedFilteringSequencingHooks) PostTxFilter(header *types.Header, db *state.StateDB, a *arbosState.ArbosState, tx *types.Transaction, sender common.Address, dataGas uint64, result *core.ExecutionResult, positionInBlock int) error {
 	if tx.Type() == types.ArbitrumInternalTxType {
 		return nil
 	}
 	touchAddresses(db, tx, sender)
 	applyEventFilter(f.eventFilter, db)
 
-	if filtered, _ := db.IsAddressFiltered(); filtered {
+	if filtered, filteredAddresses := db.IsAddressFiltered(); filtered {
 		// For redeems, return the filter error so the block processor can
-		// trigger a group rollback.
+		// trigger a group rollback. The block processor captures all report
+		// data before rollback and passes it through ErrFilteredCascadingRedeem.
 		if tx.Type() == types.ArbitrumRetryTxType {
 			return state.ErrArbTxFilter
 		}
@@ -148,24 +154,60 @@ func (f *DelayedFilteringSequencingHooks) PostTxFilter(header *types.Header, db
 		if errors.As(result.Err, &filteredErr) {
 			return nil
 		}
-		// Otherwise, this tx touched a filtered address but wasn't in the
-		// onchain filter - collect it so the caller can halt.
-		f.FilteredTxHashes = append(f.FilteredTxHashes, tx.Hash())
+		txRLP, err := tx.MarshalBinary()
+		if err != nil {
+			log.Error("error marshalling filtered delayed tx to RLP", "txHash", tx.Hash(), "err", err)
+			txRLP = nil
+		}
+		report := addressfilter.FilteredTxReport{
+			ID:                uuid.Must(uuid.NewV7()).String(),
+			TxHash:            tx.Hash(),
+			TxRLP:             txRLP,
+			FilteredAddresses: filteredAddresses,
+			BlockNumber:       header.Number.Uint64(),
+			ParentBlockHash:   header.ParentHash,
+			PositionInBlock:   uint64(positionInBlock), // #nosec G115
+			FilteredAt:        time.Now().UTC(),
+			IsDelayed:         true,
+			DelayedReportData: &addressfilter.DelayedReportData{InboxRequestId: f.inboxRequestId},
+		}
+		f.pendingFilteredTxReports = append(f.pendingFilteredTxReports, report)
 	}
 	return nil
 }
 
 func (f *DelayedFilteringSequencingHooks) SupportsGroupRollback() bool { return true }
 
-// TxFailed extracts the originating tx hash from ErrFilteredCascadingRedeem
-// and appends it to FilteredTxHashes. After ProduceBlockAdvanced returns, the
-// existing check fires ErrFilteredDelayedMessage, causing the delayed sequencer
-// to halt and the transaction-filterer to add the hash to the onchain filter.
+// TxFailed builds a fully populated FilteredTxReport from
+// ErrFilteredCascadingRedeem. The block processor captures all needed data
+// (originating tx, filtered addresses, block metadata, user tx position)
+// before the group rollback and passes it through the error.
 func (f *DelayedFilteringSequencingHooks) TxFailed(err error) {
 	var cascadingErr *arbos.ErrFilteredCascadingRedeem
-	if errors.As(err, &cascadingErr) {
-		f.FilteredTxHashes = append(f.FilteredTxHashes, cascadingErr.OriginatingTxHash)
+	if !errors.As(err, &cascadingErr) {
+		return
 	}
+	var txRLP hexutil.Bytes
+	if cascadingErr.OriginatingTx != nil {
+		if b, marshalErr := cascadingErr.OriginatingTx.MarshalBinary(); marshalErr != nil {
+			log.Error("error marshalling originating tx RLP", "txHash", cascadingErr.OriginatingTxHash, "err", marshalErr)
+		} else {
+			txRLP = b
+		}
+	}
+	report := addressfilter.FilteredTxReport{
+		ID:                uuid.Must(uuid.NewV7()).String(),
+		TxHash:            cascadingErr.OriginatingTxHash,
+		TxRLP:             txRLP,
+		FilteredAddresses: cascadingErr.FilteredAddresses,
+		BlockNumber:       cascadingErr.BlockNumber,
+		ParentBlockHash:   cascadingErr.ParentBlockHash,
+		PositionInBlock:   uint64(cascadingErr.PositionInBlock), // #nosec G115
+		FilteredAt:        time.Now().UTC(),
+		IsDelayed:         true,
+		DelayedReportData: &addressfilter.DelayedReportData{InboxRequestId: f.inboxRequestId},
+	}
+	f.pendingFilteredTxReports = append(f.pendingFilteredTxReports, report)
 }
 
 func applyEventFilter(ef *eventfilter.EventFilter, db *state.StateDB) {
@@ -934,7 +976,11 @@ func (s *ExecutionEngine) createBlockFromNextMessage(msg *arbostypes.MessageWith
 			log.Warn("error parsing incoming message for filtering", "err", err)
 			txes = types.Transactions{}
 		}
-		filteringHooks := NewDelayedFilteringSequencingHooks(txes, s.eventFilter)
+		var inboxRequestId common.Hash
+		if msg.Message.Header.RequestId != nil {
+			inboxRequestId = *msg.Message.Header.RequestId
+		}
+		filteringHooks := NewDelayedFilteringSequencingHooks(txes, s.eventFilter, inboxRequestId)
 
 		block, statedb, receipts, err := arbos.ProduceBlockAdvanced(
 			msg.Message.Header,
@@ -951,10 +997,15 @@ func (s *ExecutionEngine) createBlockFromNextMessage(msg *arbostypes.MessageWith
 			return nil, nil, nil, err
 		}
 		// Check if any txs touched filtered addresses but are not in the onchain filter
-		if len(filteringHooks.FilteredTxHashes) > 0 {
+		if len(filteringHooks.pendingFilteredTxReports) > 0 {
+			filteredTxHashes := make([]common.Hash, len(filteringHooks.pendingFilteredTxReports))
+			for i, report := range filteringHooks.pendingFilteredTxReports {
+				filteredTxHashes[i] = report.TxHash
+			}
+
 			if s.transactionFiltererRPCClient != nil {
 				s.LaunchThread(func(ctx context.Context) {
-					for _, filteredTxHash := range filteringHooks.FilteredTxHashes {
+					for _, filteredTxHash := range filteredTxHashes {
 						_, err := s.transactionFiltererRPCClient.Filter(filteredTxHash).Await(ctx)
 						if err != nil {
 							log.Error("error reporting filtered tx to transaction-filterer", "filteredTxHash", filteredTxHash, "err", err)
@@ -963,8 +1014,18 @@ func (s *ExecutionEngine) createBlockFromNextMessage(msg *arbostypes.MessageWith
 				})
 			}
 
+			// Report structured reports to filtering-report service (non-blocking)
+			if s.filteringReportRPCClient != nil && len(filteringHooks.pendingFilteredTxReports) > 0 {
+				reports := filteringHooks.pendingFilteredTxReports
+				s.LaunchThread(func(ctx context.Context) {
+					if _, err := s.filteringReportRPCClient.ReportFilteredTransactions(reports).Await(ctx); err != nil {
+						log.Error("error reporting filtered delayed txs to filtering-report", "count", len(reports), "err", err)
+					}
+				})
+			}
+
 			return nil, nil, nil, &ErrFilteredDelayedMessage{
-				TxHashes:      filteringHooks.FilteredTxHashes,
+				TxHashes:      filteredTxHashes,
 				DelayedMsgIdx: msg.DelayedMessagesRead - 1,
 			}
 		}