fix: Prevent blockwatcher from missing contract events on startup

core/core.go

@@ -337,7 +337,10 @@ func newWithPrivateConfig(ctx context.Context, config Config, pConfig privateCon
 		Topics:          topics,
 		Client:          blockWatcherClient,
 	}
-	blockWatcher := blockwatch.New(blockRetentionLimit, blockWatcherConfig)
+	blockWatcher, err := blockwatch.New(blockRetentionLimit, blockWatcherConfig)
+	if err != nil {
+		return nil, err
+	}
 
 	// Initialize the order validator
 	orderValidator, err := ordervalidator.New(

ethereum/blockwatch/block_watcher.go

@@ -92,15 +92,19 @@ type Watcher struct {
 }
 
 // New creates a new Watcher instance.
-func New(retentionLimit int, config Config) *Watcher {
+func New(retentionLimit int, config Config) (*Watcher, error) {
+	existingMiniHeaders, err := config.DB.FindMiniHeaders(nil)
+	if err != nil {
+		return nil, err
+	}
 	return &Watcher{
 		pollingInterval: config.PollingInterval,
 		db:              config.DB,
-		stack:           simplestack.New(retentionLimit, []*types.MiniHeader{}),
+		stack:           simplestack.New(retentionLimit, existingMiniHeaders),
 		client:          config.Client,
 		withLogs:        config.WithLogs,
 		topics:          config.Topics,
-	}
+	}, nil
 }
 
 func (w *Watcher) GetNumberOfBlocksBehind(ctx context.Context) (int, int, error) {

ethereum/blockwatch/block_watcher_test.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"math/big"
+	"strconv"
 	"strings"
 	"testing"
 	"time"
@@ -35,6 +36,43 @@ func dbOptions() *db.Options {
 	return options
 }
 
+// TestNewWatcher ensures that any existing blocks in the database are added to
+// simplestack when New is called.
+func TestNewWatcher(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	dbOpts := dbOptions()
+	database, err := db.New(ctx, dbOpts)
+	require.NoError(t, err)
+	config.DB = database
+
+	expectedMiniHeaders := make([]*types.MiniHeader, dbOpts.MaxMiniHeaders)
+	for i := range expectedMiniHeaders {
+		expectedMiniHeaders[i] = &types.MiniHeader{
+			Hash:   common.HexToHash(strconv.Itoa(i)),
+			Number: big.NewInt(int64(i)),
+		}
+
+	}
+	added, _, err := database.AddMiniHeaders(expectedMiniHeaders)
+	require.NoError(t, err)
+	require.Len(t, added, len(expectedMiniHeaders))
+
+	watcher, err := New(dbOpts.MaxMiniHeaders, config)
+	require.NoError(t, err)
+	actualMiniHeaders := watcher.stack.PeekAll()
+	for _, expectedMiniHeader := range expectedMiniHeaders {
+		found := false
+		for _, actualMiniHeader := range actualMiniHeaders {
+			if expectedMiniHeader.Number.Cmp(actualMiniHeader.Number) == 0 {
+				assert.Equal(t, expectedMiniHeader, actualMiniHeader)
+				found = true
+			}
+		}
+		assert.True(t, found, "miniHeader with hash %q was not stored in watcher.stack", expectedMiniHeader.Hash.Hex())
+	}
+}
+
 func TestWatcher(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -519,7 +557,9 @@ func setupOrderWatcher(t *testing.T, ctx context.Context, client Client) *Watche
 	require.NoError(t, err)
 	config.Client = client
 	config.DB = database
-	return New(blockRetentionLimit, config)
+	watcher, err := New(blockRetentionLimit, config)
+	require.NoError(t, err)
+	return watcher
 }
 
 func aRange(from, to int) string {

zeroex/orderwatch/order_watcher_test.go

@@ -1959,7 +1959,7 @@ func setupOrderWatcherWithValidator(ctx context.Context, t *testing.T, ethRPCCli
 		Topics:          topics,
 		Client:          blockWatcherClient,
 	}
-	blockWatcher := blockwatch.New(maxMiniHeaders, blockWatcherConfig)
+	blockWatcher, err := blockwatch.New(maxMiniHeaders, blockWatcherConfig)
 	require.NoError(t, err)
 	orderWatcher, err := New(Config{
 		DB:                database,