indexer.bridge.grouping

indexer/bigint/bigint.go

@@ -25,35 +25,3 @@ func Clamp(start, end *big.Int, size uint64) *big.Int {
 func Matcher(num int64) func(*big.Int) bool {
 	return func(bi *big.Int) bool { return bi.Int64() == num }
 }
-
-type Range struct {
-	Start *big.Int
-	End   *big.Int
-}
-
-// Grouped will return a slice of inclusive ranges from (start, end),
-// capped to the supplied size from `(start, end)`.
-func Grouped(start, end *big.Int, size uint64) []Range {
-	if end.Cmp(start) < 0 || size == 0 {
-		return nil
-	}
-	bigMaxDiff := big.NewInt(int64(size - 1))
-
-	groups := []Range{}
-	for start.Cmp(end) <= 0 {
-		diff := new(big.Int).Sub(end, start)
-		switch {
-		case diff.Uint64()+1 <= size:
-			// re-use allocated diff as the next start
-			groups = append(groups, Range{start, end})
-			start = diff.Add(end, One)
-		default:
-			// re-use allocated diff as the next start
-			end := new(big.Int).Add(start, bigMaxDiff)
-			groups = append(groups, Range{start, end})
-			start = diff.Add(end, One)
-		}
-	}
-
-	return groups
-}

indexer/bigint/bigint_test.go

@@ -27,46 +27,3 @@ func TestClamp(t *testing.T) {
 	require.False(t, end == result)
 	require.Equal(t, uint64(5), result.Uint64())
 }
-
-func TestGrouped(t *testing.T) {
-	// base cases
-	require.Nil(t, Grouped(One, Zero, 1))
-	require.Nil(t, Grouped(Zero, One, 0))
-
-	// Same Start/End
-	group := Grouped(One, One, 1)
-	require.Len(t, group, 1)
-	require.Equal(t, One, group[0].Start)
-	require.Equal(t, One, group[0].End)
-
-	Three, Five := big.NewInt(3), big.NewInt(5)
-
-	// One at a time
-	group = Grouped(One, Three, 1)
-	require.Equal(t, One, group[0].End)
-	require.Equal(t, int64(1), group[0].End.Int64())
-	require.Equal(t, int64(2), group[1].Start.Int64())
-	require.Equal(t, int64(2), group[1].End.Int64())
-	require.Equal(t, int64(3), group[2].Start.Int64())
-	require.Equal(t, int64(3), group[2].End.Int64())
-
-	// Split groups
-	group = Grouped(One, Five, 3)
-	require.Len(t, group, 2)
-	require.Equal(t, One, group[0].Start)
-	require.Equal(t, int64(3), group[0].End.Int64())
-	require.Equal(t, int64(4), group[1].Start.Int64())
-	require.Equal(t, Five, group[1].End)
-
-	// Encompasses the range
-	group = Grouped(One, Five, 5)
-	require.Len(t, group, 1)
-	require.Equal(t, One, group[0].Start, Zero)
-	require.Equal(t, Five, group[0].End)
-
-	// Size larger than the entire range
-	group = Grouped(One, Five, 100)
-	require.Len(t, group, 1)
-	require.Equal(t, One, group[0].Start, Zero)
-	require.Equal(t, Five, group[0].End)
-}

indexer/database/blocks.go

@@ -2,8 +2,10 @@ package database
 
 import (
 	"errors"
+	"fmt"
 	"math/big"
 
+	"github.com/ethereum-optimism/optimism/indexer/bigint"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 
@@ -51,7 +53,7 @@ type BlocksView interface {
 	L2BlockHeaderWithFilter(BlockHeader) (*L2BlockHeader, error)
 	L2LatestBlockHeader() (*L2BlockHeader, error)
 
-	LatestEpoch() (*Epoch, error)
+	LatestObservedEpoch(*big.Int, uint64) (*Epoch, error)
 }
 
 type BlocksDB interface {
@@ -155,36 +157,74 @@ type Epoch struct {
 	L2BlockHeader L2BlockHeader `gorm:"embedded"`
 }
 
-// LatestEpoch return the latest epoch, seen on  L1 & L2. In other words
-// this returns the latest indexed L1 block that has a corresponding
-// indexed L2 block with a matching L1Origin (equal timestamps).
+// LatestObservedEpoch return the marker for latest epoch, observed on  L1 & L2, within
+// the specified bounds. In other words this returns the latest indexed L1 block that has
+// a corresponding indexed L2 block with a matching L1Origin (equal timestamps).
+//
+// If `fromL1Height` (inclusive) is not specified, the search will start from genesis and
+// continue all the way to latest indexed heights if `maxL1Range == 0`.
 //
 // For more, see the protocol spec:
 //   - https://github.com/ethereum-optimism/optimism/blob/develop/specs/derivation.md
-func (db *blocksDB) LatestEpoch() (*Epoch, error) {
-	latestL1Header, err := db.L1LatestBlockHeader()
-	if err != nil {
-		return nil, err
-	} else if latestL1Header == nil {
-		return nil, nil
+func (db *blocksDB) LatestObservedEpoch(fromL1Height *big.Int, maxL1Range uint64) (*Epoch, error) {
+	// We use timestamps since that translates to both L1 & L2
+	var fromTimestamp, toTimestamp uint64
+
+	if fromL1Height == nil {
+		fromL1Height = bigint.Zero
 	}
 
-	latestL2Header, err := db.L2LatestBlockHeader()
-	if err != nil {
-		return nil, err
-	} else if latestL2Header == nil {
-		return nil, nil
+	// Lower Bound (the default `fromTimestamp = 0` suffices genesis representation)
+	if fromL1Height.BitLen() > 0 {
+		var header L1BlockHeader
+		result := db.gorm.Where("number = ?", fromL1Height).Take(&header)
+		if result.Error != nil {
+			if errors.Is(result.Error, gorm.ErrRecordNotFound) {
+				return nil, nil
+			}
+			return nil, result.Error
+		}
+
+		fromTimestamp = header.Timestamp
 	}
 
-	minTime := latestL1Header.Timestamp
-	if latestL2Header.Timestamp < minTime {
-		minTime = latestL2Header.Timestamp
+	// Upper Bound (lowest timestamp indexed between L1/L2 bounded by `maxL1Range`)
+	{
+		l1QueryFilter := fmt.Sprintf("timestamp >= %d", fromTimestamp)
+		if maxL1Range > 0 {
+			maxHeight := new(big.Int).Add(fromL1Height, big.NewInt(int64(maxL1Range)))
+			l1QueryFilter = fmt.Sprintf("%s AND number <= %d", l1QueryFilter, maxHeight)
+		}
+
+		var l1Header L1BlockHeader
+		result := db.gorm.Where(l1QueryFilter).Order("timestamp DESC").Take(&l1Header)
+		if result.Error != nil {
+			if errors.Is(result.Error, gorm.ErrRecordNotFound) {
+				return nil, nil
+			}
+			return nil, result.Error
+		}
+
+		toTimestamp = l1Header.Timestamp
+
+		var l2Header L2BlockHeader
+		result = db.gorm.Where("timestamp <= ?", toTimestamp).Order("timestamp DESC").Take(&l2Header)
+		if result.Error != nil {
+			if errors.Is(result.Error, gorm.ErrRecordNotFound) {
+				return nil, nil
+			}
+			return nil, result.Error
+		}
+
+		if l2Header.Timestamp < toTimestamp {
+			toTimestamp = l2Header.Timestamp
+		}
 	}
 
-	// This is a faster query than doing an INNER JOIN between l1_block_headers and l2_block_headers
-	// which requires a full table scan to compute the resulting table.
-	l1Query := db.gorm.Table("l1_block_headers").Where("timestamp <= ?", minTime)
-	l2Query := db.gorm.Table("l2_block_headers").Where("timestamp <= ?", minTime)
+	// Search for the latest indexed epoch within range. This is a faster query than doing an INNER JOIN between
+	// l1_block_headers and l2_block_headers which requires a full table scan to compute the resulting table.
+	l1Query := db.gorm.Table("l1_block_headers").Where("timestamp >= ? AND timestamp <= ?", fromTimestamp, toTimestamp)
+	l2Query := db.gorm.Table("l2_block_headers").Where("timestamp >= ? AND timestamp <= ?", fromTimestamp, toTimestamp)
 	query := db.gorm.Raw(`SELECT * FROM (?) AS l1_block_headers, (?) AS l2_block_headers
 		WHERE l1_block_headers.timestamp = l2_block_headers.timestamp
 		ORDER BY l2_block_headers.number DESC LIMIT 1`, l1Query, l2Query)

indexer/database/mocks.go

 package database
 
 import (
+	"math/big"
+
 	"github.com/ethereum/go-ethereum/common"
 
 	"github.com/stretchr/testify/mock"
@@ -51,7 +53,7 @@ func (m *MockBlocksView) L2LatestBlockHeader() (*L2BlockHeader, error) {
 	return args.Get(0).(*L2BlockHeader), args.Error(1)
 }
 
-func (m *MockBlocksView) LatestEpoch() (*Epoch, error) {
+func (m *MockBlocksView) LatestObservedEpoch(*big.Int, uint64) (*Epoch, error) {
 	args := m.Called()
 	return args.Get(0).(*Epoch), args.Error(1)
 }

indexer/e2e_tests/setup.go

@@ -41,21 +41,24 @@ func createE2ETestSuite(t *testing.T) E2ETestSuite {
 	dbUser := os.Getenv("DB_USER")
 	dbName := setupTestDatabase(t)
 
-	// Discard the Global Logger as each component
-	// has its own configured logger
+	// Rollup System Configuration. Unless specified,
+	// omit logs emitted by the various components. Maybe
+	// we can eventually dump these logs to a temp file
 	log.Root().SetHandler(log.DiscardHandler())
-
-	// Rollup System Configuration and Start
 	opCfg := op_e2e.DefaultSystemConfig(t)
-	opCfg.DeployConfig.FinalizationPeriodSeconds = 2
+	if len(os.Getenv("ENABLE_ROLLUP_LOGS")) == 0 {
+		t.Log("set env 'ENABLE_ROLLUP_LOGS' to show rollup logs")
+		for name, logger := range opCfg.Loggers {
+			t.Logf("discarding logs for %s", name)
+			logger.SetHandler(log.DiscardHandler())
+		}
+	}
+
+	// Rollup Start
 	opSys, err := opCfg.Start(t)
 	require.NoError(t, err)
 	t.Cleanup(func() { opSys.Close() })
 
-	// E2E tests can run on the order of magnitude of minutes. Once
-	// the system is running, mark this test for Parallel execution
-	t.Parallel()
-
 	// Indexer Configuration and Start
 	indexerCfg := config.Config{
 		DB: config.DBConfig{
@@ -86,8 +89,14 @@ func createE2ETestSuite(t *testing.T) E2ETestSuite {
 		MetricsServer: config.ServerConfig{Host: "127.0.0.1", Port: 0},
 	}
 
-	// Emit debug log levels
-	db, err := database.NewDB(testlog.Logger(t, log.LvlDebug).New("role", "db"), indexerCfg.DB)
+	// E2E tests can run on the order of magnitude of minutes. Once
+	// the system is running, mark this test for Parallel execution
+	t.Parallel()
+
+	// provide a DB for the unit test. disable logging
+	silentLog := testlog.Logger(t, log.LvlInfo)
+	silentLog.SetHandler(log.DiscardHandler())
+	db, err := database.NewDB(silentLog, indexerCfg.DB)
 	require.NoError(t, err)
 	t.Cleanup(func() { db.Close() })
 
@@ -138,7 +147,6 @@ func setupTestDatabase(t *testing.T) string {
 		User:     user,
 		Password: "",
 	}
-	// NewDB will create the database schema
 
 	silentLog := log.New()
 	silentLog.SetHandler(log.DiscardHandler())