op-dispute-mon: Fetch output root agreement in extract stage (#10464)

* op-dispute-mon: Enrich games in parallel

Reports any failure to retrieve data in the failed metric since it results in the game being skipped.

* op-dispute-mon: Make max concurrency configurable

Simplify the code a bit.

* op-dispute-mon: Add numbers to log

* op-dispute-mon: Reduce default max concurrency

* op-dispute-mon: Add metric for monitor duration

* op-dispute-mon: Fetch output root agreement in extract stage

Removes the last HTTP call from the transform stage and allows the calls to be done in parallel.

op-dispute-mon/mon/validator.go → op-dispute-mon/mon/extract/agreement_enricher.go

-package mon
+package extract
 
 import (
 	"context"
@@ -6,6 +6,8 @@ import (
 	"strings"
 	"time"
 
+	monTypes "github.com/ethereum-optimism/optimism/op-dispute-mon/mon/types"
+	"github.com/ethereum-optimism/optimism/op-service/sources/batching/rpcblock"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/log"
 
@@ -21,47 +23,49 @@ type OutputMetrics interface {
 	RecordOutputFetchTime(float64)
 }
 
-type outputValidator struct {
+type AgreementEnricher struct {
 	log     log.Logger
 	metrics OutputMetrics
 	client  OutputRollupClient
 }
 
-func newOutputValidator(logger log.Logger, metrics OutputMetrics, client OutputRollupClient) *outputValidator {
-	return &outputValidator{
+func NewAgreementEnricher(logger log.Logger, metrics OutputMetrics, client OutputRollupClient) *AgreementEnricher {
+	return &AgreementEnricher{
 		log:     logger,
 		metrics: metrics,
 		client:  client,
 	}
 }
 
-// CheckRootAgreement validates the specified root claim against the output at the given block number.
-func (o *outputValidator) CheckRootAgreement(ctx context.Context, l1HeadNum uint64, l2BlockNum uint64, rootClaim common.Hash) (bool, common.Hash, error) {
-	output, err := o.client.OutputAtBlock(ctx, l2BlockNum)
+// Enrich validates the specified root claim against the output at the given block number.
+func (o *AgreementEnricher) Enrich(ctx context.Context, block rpcblock.Block, caller GameCaller, game *monTypes.EnrichedGameData) error {
+	output, err := o.client.OutputAtBlock(ctx, game.L2BlockNumber)
 	if err != nil {
 		// string match as the error comes from the remote server so we can't use Errors.Is sadly.
 		if strings.Contains(err.Error(), "not found") {
 			// Output root doesn't exist, so we must disagree with it.
-			return false, common.Hash{}, nil
+			game.AgreeWithClaim = false
+			return nil
 		}
-		return false, common.Hash{}, fmt.Errorf("failed to get output at block: %w", err)
+		return fmt.Errorf("failed to get output at block: %w", err)
 	}
 	o.metrics.RecordOutputFetchTime(float64(time.Now().Unix()))
-	expected := common.Hash(output.OutputRoot)
-	rootMatches := rootClaim == expected
+	game.ExpectedRootClaim = common.Hash(output.OutputRoot)
+	rootMatches := game.RootClaim == game.ExpectedRootClaim
 	if !rootMatches {
-		return false, expected, nil
+		game.AgreeWithClaim = false
+		return nil
 	}
 
 	// If the root matches, also check that l2 block is safe at the L1 head
-	safeHead, err := o.client.SafeHeadAtL1Block(ctx, l1HeadNum)
+	safeHead, err := o.client.SafeHeadAtL1Block(ctx, game.L1HeadNum)
 	if err != nil {
-		o.log.Warn("Unable to verify proposed block was safe", "l1HeadNum", l1HeadNum, "l2BlockNum", l2BlockNum, "err", err)
+		o.log.Warn("Unable to verify proposed block was safe", "l1HeadNum", game.L1HeadNum, "l2BlockNum", game.L2BlockNumber, "err", err)
 		// If safe head data isn't available, assume the output root was safe
 		// Avoids making the dispute mon dependent on safe head db being available
-		//
-		return true, expected, nil
+		game.AgreeWithClaim = true
+		return nil
 	}
-	isSafe := safeHead.SafeHead.Number >= l2BlockNum
-	return isSafe, expected, nil
+	game.AgreeWithClaim = safeHead.SafeHead.Number >= game.L2BlockNumber
+	return nil
 }

op-dispute-mon/mon/validator_test.go → op-dispute-mon/mon/extract/agreement_enricher_test.go

-package mon
+package extract
 
 import (
 	"context"
 	"errors"
 	"testing"
 
+	"github.com/ethereum-optimism/optimism/op-dispute-mon/mon/types"
 	"github.com/ethereum-optimism/optimism/op-service/eth"
+	"github.com/ethereum-optimism/optimism/op-service/sources/batching/rpcblock"
 	"github.com/ethereum-optimism/optimism/op-service/testlog"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/stretchr/testify/require"
 )
 
-var (
-	mockRootClaim = common.HexToHash("0x10")
-)
-
 func TestDetector_CheckRootAgreement(t *testing.T) {
 	t.Parallel()
 
 	t.Run("OutputFetchFails", func(t *testing.T) {
 		validator, rollup, metrics := setupOutputValidatorTest(t)
 		rollup.outputErr = errors.New("boom")
-		agree, fetched, err := validator.CheckRootAgreement(context.Background(), 100, 0, mockRootClaim)
+		game := &types.EnrichedGameData{
+			L1HeadNum:     100,
+			L2BlockNumber: 0,
+			RootClaim:     mockRootClaim,
+		}
+		err := validator.Enrich(context.Background(), rpcblock.Latest, nil, game)
 		require.ErrorIs(t, err, rollup.outputErr)
-		require.Equal(t, common.Hash{}, fetched)
-		require.False(t, agree)
+		require.Equal(t, common.Hash{}, game.ExpectedRootClaim)
+		require.False(t, game.AgreeWithClaim)
 		require.Zero(t, metrics.fetchTime)
 	})
 
 	t.Run("OutputMismatch_Safe", func(t *testing.T) {
 		validator, _, metrics := setupOutputValidatorTest(t)
-		agree, fetched, err := validator.CheckRootAgreement(context.Background(), 100, 0, common.Hash{})
+		game := &types.EnrichedGameData{
+			L1HeadNum:     100,
+			L2BlockNumber: 0,
+			RootClaim:     common.Hash{},
+		}
+		err := validator.Enrich(context.Background(), rpcblock.Latest, nil, game)
 		require.NoError(t, err)
-		require.Equal(t, mockRootClaim, fetched)
-		require.False(t, agree)
+		require.Equal(t, mockRootClaim, game.ExpectedRootClaim)
+		require.False(t, game.AgreeWithClaim)
 		require.NotZero(t, metrics.fetchTime)
 	})
 
 	t.Run("OutputMatches_Safe", func(t *testing.T) {
 		validator, _, metrics := setupOutputValidatorTest(t)
-		agree, fetched, err := validator.CheckRootAgreement(context.Background(), 200, 0, mockRootClaim)
+		game := &types.EnrichedGameData{
+			L1HeadNum:     200,
+			L2BlockNumber: 0,
+			RootClaim:     mockRootClaim,
+		}
+		err := validator.Enrich(context.Background(), rpcblock.Latest, nil, game)
 		require.NoError(t, err)
-		require.Equal(t, mockRootClaim, fetched)
-		require.True(t, agree)
+		require.Equal(t, mockRootClaim, game.ExpectedRootClaim)
+		require.True(t, game.AgreeWithClaim)
 		require.NotZero(t, metrics.fetchTime)
 	})
 
 	t.Run("OutputMismatch_NotSafe", func(t *testing.T) {
 		validator, client, metrics := setupOutputValidatorTest(t)
 		client.safeHeadNum = 99
-		agree, fetched, err := validator.CheckRootAgreement(context.Background(), 100, 0, common.Hash{})
+		game := &types.EnrichedGameData{
+			L1HeadNum:     100,
+			L2BlockNumber: 0,
+			RootClaim:     common.Hash{},
+		}
+		err := validator.Enrich(context.Background(), rpcblock.Latest, nil, game)
 		require.NoError(t, err)
-		require.Equal(t, mockRootClaim, fetched)
-		require.False(t, agree)
+		require.Equal(t, mockRootClaim, game.ExpectedRootClaim)
+		require.False(t, game.AgreeWithClaim)
 		require.NotZero(t, metrics.fetchTime)
 	})
 
 	t.Run("OutputMatches_SafeHeadError", func(t *testing.T) {
 		validator, client, metrics := setupOutputValidatorTest(t)
 		client.safeHeadErr = errors.New("boom")
-		agree, fetched, err := validator.CheckRootAgreement(context.Background(), 200, 0, mockRootClaim)
+		game := &types.EnrichedGameData{
+			L1HeadNum:     200,
+			L2BlockNumber: 0,
+			RootClaim:     mockRootClaim,
+		}
+		err := validator.Enrich(context.Background(), rpcblock.Latest, nil, game)
 		require.NoError(t, err)
-		require.Equal(t, mockRootClaim, fetched)
-		require.True(t, agree) // Assume safe if we can't retrieve the safe head so monitoring isn't dependent on safe head db
+		require.Equal(t, mockRootClaim, game.ExpectedRootClaim)
+		require.True(t, game.AgreeWithClaim) // Assume safe if we can't retrieve the safe head so monitoring isn't dependent on safe head db
 		require.NotZero(t, metrics.fetchTime)
 	})
 
 	t.Run("OutputMismatch_SafeHeadError", func(t *testing.T) {
 		validator, client, metrics := setupOutputValidatorTest(t)
 		client.safeHeadErr = errors.New("boom")
-		agree, fetched, err := validator.CheckRootAgreement(context.Background(), 100, 0, common.Hash{})
+		game := &types.EnrichedGameData{
+			L1HeadNum:     100,
+			L2BlockNumber: 0,
+		}
+		err := validator.Enrich(context.Background(), rpcblock.Latest, nil, game)
 		require.NoError(t, err)
-		require.Equal(t, mockRootClaim, fetched)
-		require.False(t, agree) // Not agreed because the root doesn't match
+		require.Equal(t, mockRootClaim, game.ExpectedRootClaim)
+		require.False(t, game.AgreeWithClaim) // Not agreed because the root doesn't match
 		require.NotZero(t, metrics.fetchTime)
 	})
 
 	t.Run("OutputMatches_NotSafe", func(t *testing.T) {
 		validator, client, metrics := setupOutputValidatorTest(t)
 		client.safeHeadNum = 99
-		agree, fetched, err := validator.CheckRootAgreement(context.Background(), 200, 100, mockRootClaim)
+		game := &types.EnrichedGameData{
+			L1HeadNum:     200,
+			L2BlockNumber: 100,
+			RootClaim:     mockRootClaim,
+		}
+		err := validator.Enrich(context.Background(), rpcblock.Latest, nil, game)
 		require.NoError(t, err)
-		require.Equal(t, mockRootClaim, fetched)
-		require.False(t, agree)
+		require.Equal(t, mockRootClaim, game.ExpectedRootClaim)
+		require.False(t, game.AgreeWithClaim)
 		require.NotZero(t, metrics.fetchTime)
 	})
 
@@ -91,19 +123,24 @@ func TestDetector_CheckRootAgreement(t *testing.T) {
 		validator, rollup, metrics := setupOutputValidatorTest(t)
 		// This crazy error is what we actually get back from the API
 		rollup.outputErr = errors.New("failed to get L2 block ref with sync status: failed to determine L2BlockRef of height 42984924, could not get payload: not found")
-		agree, fetched, err := validator.CheckRootAgreement(context.Background(), 100, 42984924, mockRootClaim)
+		game := &types.EnrichedGameData{
+			L1HeadNum:     100,
+			L2BlockNumber: 42984924,
+			RootClaim:     mockRootClaim,
+		}
+		err := validator.Enrich(context.Background(), rpcblock.Latest, nil, game)
 		require.NoError(t, err)
-		require.Equal(t, common.Hash{}, fetched)
-		require.False(t, agree)
+		require.Equal(t, common.Hash{}, game.ExpectedRootClaim)
+		require.False(t, game.AgreeWithClaim)
 		require.Zero(t, metrics.fetchTime)
 	})
 }
 
-func setupOutputValidatorTest(t *testing.T) (*outputValidator, *stubRollupClient, *stubOutputMetrics) {
+func setupOutputValidatorTest(t *testing.T) (*AgreementEnricher, *stubRollupClient, *stubOutputMetrics) {
 	logger := testlog.Logger(t, log.LvlInfo)
 	client := &stubRollupClient{safeHeadNum: 99999999999}
 	metrics := &stubOutputMetrics{}
-	validator := newOutputValidator(logger, metrics, client)
+	validator := NewAgreementEnricher(logger, metrics, client)
 	return validator, client, metrics
 }
 
@@ -122,7 +159,7 @@ type stubRollupClient struct {
 	safeHeadNum uint64
 }
 
-func (s *stubRollupClient) OutputAtBlock(ctx context.Context, blockNum uint64) (*eth.OutputResponse, error) {
+func (s *stubRollupClient) OutputAtBlock(_ context.Context, blockNum uint64) (*eth.OutputResponse, error) {
 	s.blockNum = blockNum
 	return &eth.OutputResponse{OutputRoot: eth.Bytes32(mockRootClaim)}, s.outputErr
 }

op-dispute-mon/mon/forecast.go

 package mon
 
 import (
-	"context"
 	"errors"
-	"fmt"
 
 	"github.com/ethereum-optimism/optimism/op-challenger/game/types"
 	"github.com/ethereum-optimism/optimism/op-dispute-mon/metrics"
 	"github.com/ethereum-optimism/optimism/op-dispute-mon/mon/transform"
 	monTypes "github.com/ethereum-optimism/optimism/op-dispute-mon/mon/types"
-	"github.com/ethereum/go-ethereum/common"
-
 	"github.com/ethereum/go-ethereum/log"
 )
 
@@ -18,10 +14,6 @@ var (
 	ErrRootAgreement = errors.New("failed to check root agreement")
 )
 
-type OutputValidator interface {
-	CheckRootAgreement(ctx context.Context, l1HeadNum uint64, l2BlockNum uint64, root common.Hash) (bool, common.Hash, error)
-}
-
 type ForecastMetrics interface {
 	RecordGameAgreement(status metrics.GameAgreementStatus, count int)
 	RecordLatestInvalidProposal(timestamp uint64)
@@ -44,23 +36,21 @@ type forecastBatch struct {
 }
 
 type Forecast struct {
-	logger    log.Logger
-	metrics   ForecastMetrics
-	validator OutputValidator
+	logger  log.Logger
+	metrics ForecastMetrics
 }
 
-func NewForecast(logger log.Logger, metrics ForecastMetrics, validator OutputValidator) *Forecast {
+func NewForecast(logger log.Logger, metrics ForecastMetrics) *Forecast {
 	return &Forecast{
-		logger:    logger,
-		metrics:   metrics,
-		validator: validator,
+		logger:  logger,
+		metrics: metrics,
 	}
 }
 
-func (f *Forecast) Forecast(ctx context.Context, games []*monTypes.EnrichedGameData, ignoredCount, failedCount int) {
+func (f *Forecast) Forecast(games []*monTypes.EnrichedGameData, ignoredCount, failedCount int) {
 	batch := forecastBatch{}
 	for _, game := range games {
-		if err := f.forecastGame(ctx, game, &batch); err != nil {
+		if err := f.forecastGame(game, &batch); err != nil {
 			f.logger.Error("Failed to forecast game", "err", err)
 		}
 	}
@@ -84,12 +74,10 @@ func (f *Forecast) recordBatch(batch forecastBatch, ignoredCount, failedCount in
 	f.metrics.RecordFailedGames(failedCount)
 }
 
-func (f *Forecast) forecastGame(ctx context.Context, game *monTypes.EnrichedGameData, metrics *forecastBatch) error {
+func (f *Forecast) forecastGame(game *monTypes.EnrichedGameData, metrics *forecastBatch) error {
 	// Check the root agreement.
-	agreement, expected, err := f.validator.CheckRootAgreement(ctx, game.L1HeadNum, game.L2BlockNumber, game.RootClaim)
-	if err != nil {
-		return fmt.Errorf("%w: %w", ErrRootAgreement, err)
-	}
+	agreement := game.AgreeWithClaim
+	expected := game.ExpectedRootClaim
 
 	expectedResult := types.GameStatusDefenderWon
 	if !agreement {

op-dispute-mon/mon/monitor.go

@@ -13,7 +13,7 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 )
 
-type ForecastResolution func(ctx context.Context, games []*types.EnrichedGameData, ignoredCount, failedCount int)
+type ForecastResolution func(games []*types.EnrichedGameData, ignoredCount, failedCount int)
 type Bonds func(games []*types.EnrichedGameData)
 type Resolutions func(games []*types.EnrichedGameData)
 type MonitorClaims func(games []*types.EnrichedGameData)
@@ -100,7 +100,7 @@ func (m *gameMonitor) monitorGames() error {
 		return fmt.Errorf("failed to load games: %w", err)
 	}
 	m.resolutions(enrichedGames)
-	m.forecast(m.ctx, enrichedGames, ignored, failed)
+	m.forecast(enrichedGames, ignored, failed)
 	m.bonds(enrichedGames)
 	m.claims(enrichedGames)
 	m.withdrawals(enrichedGames)

op-dispute-mon/mon/monitor_test.go

@@ -172,7 +172,7 @@ type mockForecast struct {
 	calls int
 }
 
-func (m *mockForecast) Forecast(_ context.Context, _ []*monTypes.EnrichedGameData, _, _ int) {
+func (m *mockForecast) Forecast(_ []*monTypes.EnrichedGameData, _, _ int) {
 	m.calls++
 }
 

op-dispute-mon/mon/service.go

@@ -44,7 +44,6 @@ type Service struct {
 	claims       *ClaimMonitor
 	withdrawals  *WithdrawalMonitor
 	rollupClient *sources.RollupClient
-	validator    *outputValidator
 
 	l1Client *ethclient.Client
 
@@ -90,7 +89,6 @@ func (s *Service) initFromConfig(ctx context.Context, cfg *config.Config) error
 	s.initResolutionMonitor()
 	s.initWithdrawalMonitor()
 
-	s.initOutputValidator()   // Must be called before initForecast
 	s.initGameCallerCreator() // Must be called before initForecast
 
 	s.initExtractor(cfg)
@@ -118,10 +116,6 @@ func (s *Service) initWithdrawalMonitor() {
 	s.withdrawals = NewWithdrawalMonitor(s.logger, s.metrics)
 }
 
-func (s *Service) initOutputValidator() {
-	s.validator = newOutputValidator(s.logger, s.metrics, s.rollupClient)
-}
-
 func (s *Service) initGameCallerCreator() {
 	s.game = extract.NewGameCallerCreator(s.metrics, batching.NewMultiCaller(s.l1Client.Client(), batching.DefaultBatchSize))
 }
@@ -139,11 +133,12 @@ func (s *Service) initExtractor(cfg *config.Config) {
 		extract.NewBondEnricher(),
 		extract.NewBalanceEnricher(),
 		extract.NewL1HeadBlockNumEnricher(s.l1Client),
+		extract.NewAgreementEnricher(s.logger, s.metrics, s.rollupClient),
 	)
 }
 
 func (s *Service) initForecast(cfg *config.Config) {
-	s.forecast = NewForecast(s.logger, s.metrics, s.validator)
+	s.forecast = NewForecast(s.logger, s.metrics)
 }
 
 func (s *Service) initBonds() {

op-dispute-mon/mon/types/types.go

@@ -25,6 +25,9 @@ type EnrichedGameData struct {
 	MaxClockDuration uint64
 	Claims           []EnrichedClaim
 
+	AgreeWithClaim    bool
+	ExpectedRootClaim common.Hash
+
 	// Recipients maps addresses to true if they are a bond recipient in the game.
 	Recipients map[common.Address]bool