Fix uninitialized state handlers

README.md

@@ -1,6 +1,12 @@
 ## Running
 
 ### Directly using Go
+
+*Dependencies*
+
+* Go 1.22
+* gRPCurl (for testing)
+
 ```bash
 SIDECAR_DEBUG=false \
 SIDECAR_ETHEREUM_RPC_BASE_URL="http://54.198.82.217:8545" \
@@ -27,7 +33,7 @@ docker run -it --rm \
   -e SIDECAR_SQLITE_DB_FILE_PATH="/sqlite/sidecar.db" \
   -v "$(pwd)/sqlite:/sqlite" \
   --tty -i \
-  public.ecr.aws/z6g0f8n7/go-sidecar:latest /build/bin/sidecar
+  public.ecr.aws/z6g0f8n7/go-sidecar:latest /build/bin/cmd/sidecar
 ```
 
 ### Build and run a container locally

cmd/sidecar/main.go

@@ -8,6 +8,9 @@ import (
 	"github.com/Layr-Labs/sidecar/internal/config"
 	"github.com/Layr-Labs/sidecar/internal/contractManager"
 	"github.com/Layr-Labs/sidecar/internal/contractStore/sqliteContractStore"
+	"github.com/Layr-Labs/sidecar/internal/eigenState/avsOperators"
+	"github.com/Layr-Labs/sidecar/internal/eigenState/operatorShares"
+	"github.com/Layr-Labs/sidecar/internal/eigenState/stakerDelegations"
 	"github.com/Layr-Labs/sidecar/internal/eigenState/stateManager"
 	"github.com/Layr-Labs/sidecar/internal/fetcher"
 	"github.com/Layr-Labs/sidecar/internal/indexer"
@@ -67,6 +70,16 @@ func main() {
 
 	sm := stateManager.NewEigenStateManager(l, grm)
 
+	if _, err := avsOperators.NewAvsOperators(sm, grm, cfg.Network, cfg.Environment, l, cfg); err != nil {
+		l.Sugar().Fatalw("Failed to create AvsOperatorsModel", zap.Error(err))
+	}
+	if _, err := operatorShares.NewOperatorSharesModel(sm, grm, cfg.Network, cfg.Environment, l, cfg); err != nil {
+		l.Sugar().Fatalw("Failed to create OperatorSharesModel", zap.Error(err))
+	}
+	if _, err := stakerDelegations.NewStakerDelegationsModel(sm, grm, cfg.Network, cfg.Environment, l, cfg); err != nil {
+		l.Sugar().Fatalw("Failed to create StakerDelegationsModel", zap.Error(err))
+	}
+
 	fetchr := fetcher.NewFetcher(client, cfg, l)
 
 	idxr := indexer.NewIndexer(mds, contractStore, etherscanClient, cm, client, fetchr, l, cfg)

internal/eigenState/stakerDelegations/stakerDelegations_test.go

@@ -62,7 +62,7 @@ func Test_DelegatedStakersState(t *testing.T) {
 			TransactionIndex: 100,
 			BlockNumber:      blockNumber,
 			Address:          cfg.GetContractsMapForEnvAndNetwork().DelegationManager,
-			Arguments:        `[{"Value": "0x5fc1b61816ddeb33b65a02a42b29059ecd3a20e9" }, { "Value": "0x5accc90436492f24e6af278569691e2c942a676d" }]`,
+			Arguments:        `[{"Name":"staker","Type":"address","Value":"0xbde83df53bc7d159700e966ad5d21e8b7c619459","Indexed":true},{"Name":"operator","Type":"address","Value":"0xbde83df53bc7d159700e966ad5d21e8b7c619459","Indexed":true}]`,
 			EventName:        "StakerDelegated",
 			LogIndex:         400,
 			OutputData:       `{}`,
@@ -83,8 +83,8 @@ func Test_DelegatedStakersState(t *testing.T) {
 		assert.NotNil(t, res)
 
 		typedChange := res.(*AccumulatedStateChange)
-		assert.Equal(t, "0x5fc1b61816ddeb33b65a02a42b29059ecd3a20e9", typedChange.Staker)
-		assert.Equal(t, "0x5accc90436492f24e6af278569691e2c942a676d", typedChange.Operator)
+		assert.Equal(t, "0xbde83df53bc7d159700e966ad5d21e8b7c619459", typedChange.Staker)
+		assert.Equal(t, "0xbde83df53bc7d159700e966ad5d21e8b7c619459", typedChange.Operator)
 
 		teardown(model)
 	})
@@ -97,7 +97,7 @@ func Test_DelegatedStakersState(t *testing.T) {
 			TransactionIndex: 100,
 			BlockNumber:      blockNumber,
 			Address:          cfg.GetContractsMapForEnvAndNetwork().DelegationManager,
-			Arguments:        `[{"Value": "0x5fc1b61816ddeb33b65a02a42b29059ecd3a20e9" }, { "Value": "0x5accc90436492f24e6af278569691e2c942a676d" }]`,
+			Arguments:        `[{"Name":"staker","Type":"address","Value":"0xbde83df53bc7d159700e966ad5d21e8b7c619459","Indexed":true},{"Name":"operator","Type":"address","Value":"0xbde83df53bc7d159700e966ad5d21e8b7c619459","Indexed":true}]`,
 			EventName:        "StakerDelegated",
 			LogIndex:         400,
 			OutputData:       `{}`,
@@ -119,8 +119,8 @@ func Test_DelegatedStakersState(t *testing.T) {
 		assert.NotNil(t, stateChange)
 
 		typedChange := stateChange.(*AccumulatedStateChange)
-		assert.Equal(t, "0x5fc1b61816ddeb33b65a02a42b29059ecd3a20e9", typedChange.Staker)
-		assert.Equal(t, "0x5accc90436492f24e6af278569691e2c942a676d", typedChange.Operator)
+		assert.Equal(t, "0xbde83df53bc7d159700e966ad5d21e8b7c619459", typedChange.Staker)
+		assert.Equal(t, "0xbde83df53bc7d159700e966ad5d21e8b7c619459", typedChange.Operator)
 
 		err = model.CommitFinalState(blockNumber)
 		assert.Nil(t, err)
@@ -155,7 +155,7 @@ func Test_DelegatedStakersState(t *testing.T) {
 				TransactionIndex: 100,
 				BlockNumber:      blocks[0],
 				Address:          cfg.GetContractsMapForEnvAndNetwork().DelegationManager,
-				Arguments:        `[{"Value": "0x5fc1b61816ddeb33b65a02a42b29059ecd3a20e9" }, { "Value": "0x5accc90436492f24e6af278569691e2c942a676d" }]`,
+				Arguments:        `[{"Name":"staker","Type":"address","Value":"0xbde83df53bc7d159700e966ad5d21e8b7c619459","Indexed":true},{"Name":"operator","Type":"address","Value":"0xbde83df53bc7d159700e966ad5d21e8b7c619459","Indexed":true}]`,
 				EventName:        "StakerDelegated",
 				LogIndex:         400,
 				OutputData:       `{}`,
@@ -168,7 +168,7 @@ func Test_DelegatedStakersState(t *testing.T) {
 				TransactionIndex: 100,
 				BlockNumber:      blocks[1],
 				Address:          cfg.GetContractsMapForEnvAndNetwork().DelegationManager,
-				Arguments:        `[{"Value": "0x5fc1b61816ddeb33b65a02a42b29059ecd3a20e9" }, { "Value": "0x5accc90436492f24e6af278569691e2c942a676d" }]`,
+				Arguments:        `[{"Name":"staker","Type":"address","Value":"0xbde83df53bc7d159700e966ad5d21e8b7c619459","Indexed":true},{"Name":"operator","Type":"address","Value":"0xbde83df53bc7d159700e966ad5d21e8b7c619459","Indexed":true}]`,
 				EventName:        "StakerUndelegated",
 				LogIndex:         400,
 				OutputData:       `{}`,

internal/eigenState/stateManager/stateManager.go

@@ -45,9 +45,16 @@ func (e *EigenStateManager) RegisterState(model types.IEigenStateModel, index in
 
 // Given a log, allow each state model to determine if/how to process it
 func (e *EigenStateManager) HandleLogStateChange(log *storage.TransactionLog) error {
+	e.logger.Sugar().Debugw("Handling log state change", zap.String("transactionHash", log.TransactionHash), zap.Uint64("logIndex", log.LogIndex))
 	for _, index := range e.GetSortedModelIndexes() {
 		state := e.StateModels[index]
 		if state.IsInterestingLog(log) {
+			e.logger.Sugar().Debugw("Handling log for model",
+				zap.String("model", state.GetModelName()),
+				zap.String("transactionHash", log.TransactionHash),
+				zap.Uint64("logIndex", log.LogIndex),
+				zap.String("eventName", log.EventName),
+			)
 			_, err := state.HandleStateChange(log)
 			if err != nil {
 				return err

internal/pipeline/pipeline.go

@@ -115,11 +115,6 @@ func (p *Pipeline) RunForBlock(ctx context.Context, blockNumber uint64) error {
 				zap.Uint64("logIndex", log.LogIndex),
 			)
 
-			p.Logger.Sugar().Debugw("Handling log state change",
-				zap.Uint64("blockNumber", blockNumber),
-				zap.String("transactionHash", pt.Transaction.Hash.Value()),
-				zap.Uint64("logIndex", log.LogIndex),
-			)
 			if err := p.stateManager.HandleLogStateChange(indexedLog); err != nil {
 				p.Logger.Sugar().Errorw("Failed to handle log state change",
 					zap.Uint64("blockNumber", blockNumber),

internal/pipeline/pipeline_integration_test.go

@@ -0,0 +1,149 @@
+package pipeline
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"github.com/Layr-Labs/sidecar/internal/clients/ethereum"
+	"github.com/Layr-Labs/sidecar/internal/clients/etherscan"
+	"github.com/Layr-Labs/sidecar/internal/contractManager"
+	"github.com/Layr-Labs/sidecar/internal/contractStore/sqliteContractStore"
+	"github.com/Layr-Labs/sidecar/internal/eigenState/avsOperators"
+	"github.com/Layr-Labs/sidecar/internal/eigenState/operatorShares"
+	"github.com/Layr-Labs/sidecar/internal/eigenState/stakerDelegations"
+	"github.com/Layr-Labs/sidecar/internal/eigenState/stateManager"
+	"github.com/Layr-Labs/sidecar/internal/fetcher"
+	"github.com/Layr-Labs/sidecar/internal/indexer"
+	"github.com/Layr-Labs/sidecar/internal/logger"
+	"github.com/Layr-Labs/sidecar/internal/metrics"
+	"github.com/Layr-Labs/sidecar/internal/sqlite/migrations"
+	"github.com/Layr-Labs/sidecar/internal/storage"
+	sqliteBlockStore "github.com/Layr-Labs/sidecar/internal/storage/sqlite"
+	"github.com/Layr-Labs/sidecar/internal/tests"
+	"github.com/stretchr/testify/assert"
+	"go.uber.org/zap"
+	"gorm.io/gorm"
+	"log"
+	"os"
+	"testing"
+)
+
+var (
+	previousEnv = make(map[string]string)
+)
+
+func replaceEnv() {
+	newEnvValues := map[string]string{
+		"SIDECAR_ENVIRONMENT":           "testnet",
+		"SIDECAR_NETWORK":               "holesky",
+		"SIDECAR_ETHEREUM_RPC_BASE_URL": "http://54.198.82.217:8545",
+		"SIDECAR_ETHERSCAN_API_KEYS":    "QIPXW3YCXPR5NQ9GXTRQ3TSXB9EKMGDE34",
+		"SIDECAR_STATSD_URL":            "localhost:8125",
+		"SIDECAR_DEBUG":                 "true",
+	}
+
+	for k, v := range newEnvValues {
+		previousEnv[k] = os.Getenv(k)
+		os.Setenv(k, v)
+	}
+}
+
+func restoreEnv() {
+	for k, v := range previousEnv {
+		os.Setenv(k, v)
+	}
+}
+
+func setup() (
+	*fetcher.Fetcher,
+	*indexer.Indexer,
+	storage.BlockStore,
+	*stateManager.EigenStateManager,
+	*zap.Logger,
+	*gorm.DB,
+) {
+	cfg := tests.GetConfig()
+	l, _ := logger.NewLogger(&logger.LoggerConfig{Debug: cfg.Debug})
+
+	sdc, err := metrics.InitStatsdClient(cfg.StatsdUrl)
+	if err != nil {
+		l.Sugar().Fatal("Failed to setup statsd client", zap.Error(err))
+	}
+
+	etherscanClient := etherscan.NewEtherscanClient(cfg, l)
+	client := ethereum.NewClient(cfg.EthereumRpcConfig.BaseUrl, l)
+
+	// database
+	grm, err := tests.GetSqliteDatabaseConnection()
+	if err != nil {
+		panic(err)
+	}
+	sqliteMigrator := migrations.NewSqliteMigrator(grm, l)
+	if err := sqliteMigrator.MigrateAll(); err != nil {
+		l.Sugar().Fatalw("Failed to migrate", "error", err)
+	}
+
+	contractStore := sqliteContractStore.NewSqliteContractStore(grm, l, cfg)
+	if err := contractStore.InitializeCoreContracts(); err != nil {
+		log.Fatalf("Failed to initialize core contracts: %v", err)
+	}
+
+	cm := contractManager.NewContractManager(contractStore, etherscanClient, client, sdc, l)
+
+	mds := sqliteBlockStore.NewSqliteBlockStore(grm, l, cfg)
+	if err != nil {
+		log.Fatalln(err)
+	}
+
+	sm := stateManager.NewEigenStateManager(l, grm)
+
+	if _, err := avsOperators.NewAvsOperators(sm, grm, cfg.Network, cfg.Environment, l, cfg); err != nil {
+		l.Sugar().Fatalw("Failed to create AvsOperatorsModel", zap.Error(err))
+	}
+	if _, err := operatorShares.NewOperatorSharesModel(sm, grm, cfg.Network, cfg.Environment, l, cfg); err != nil {
+		l.Sugar().Fatalw("Failed to create OperatorSharesModel", zap.Error(err))
+	}
+	if _, err := stakerDelegations.NewStakerDelegationsModel(sm, grm, cfg.Network, cfg.Environment, l, cfg); err != nil {
+		l.Sugar().Fatalw("Failed to create StakerDelegationsModel", zap.Error(err))
+	}
+
+	fetchr := fetcher.NewFetcher(client, cfg, l)
+
+	idxr := indexer.NewIndexer(mds, contractStore, etherscanClient, cm, client, fetchr, l, cfg)
+
+	return fetchr, idxr, mds, sm, l, grm
+}
+
+func Test_Pipeline_Integration(t *testing.T) {
+	replaceEnv()
+
+	fetchr, idxr, mds, sm, l, grm := setup()
+	t.Run("Should create a new Pipeline", func(t *testing.T) {
+		p := NewPipeline(fetchr, idxr, mds, sm, l)
+		assert.NotNil(t, p)
+	})
+
+	t.Run("Should index a block, transaction with logs", func(t *testing.T) {
+		blockNumber := uint64(1175560)
+		transactionHash := "0x78cc56f0700e7ba5055f124243e6591fc1199cf3c75a17d50f8ea438254c9a76"
+		logIndex := uint64(14)
+
+		fmt.Printf("transactionHash: %s %d\n", transactionHash, logIndex)
+
+		p := NewPipeline(fetchr, idxr, mds, sm, l)
+
+		err := p.RunForBlock(context.Background(), blockNumber)
+		assert.Nil(t, err)
+
+		query := `select * from delegated_stakers where block_number = @blockNumber`
+		delegatedStakers := make([]stakerDelegations.DelegatedStakers, 0)
+		res := grm.Raw(query, sql.Named("blockNumber", blockNumber)).Scan(&delegatedStakers)
+		assert.Nil(t, res.Error)
+
+		assert.Equal(t, 1, len(delegatedStakers))
+	})
+
+	t.Cleanup(func() {
+		restoreEnv()
+	})
+}

internal/sidecar/blockIndexer.go

@@ -102,6 +102,7 @@ func (s *Sidecar) IndexFromCurrentToTip(ctx context.Context) error {
 	blocksProcessed := 0
 	runningAvg := 0
 	totalDurationMs := 0
+	lastBlockParsed := latestBlock
 	for i := latestBlock; i <= int64(ct.Get()); i++ {
 		if shouldShutdown {
 			s.Logger.Sugar().Infow("Shutting down block processor")
@@ -119,6 +120,7 @@ func (s *Sidecar) IndexFromCurrentToTip(ctx context.Context) error {
 				zap.Uint64("blocksRemaining", blocksRemaining),
 				zap.Float64("estimatedTimeRemaining (hrs)", estTimeRemainingHours),
 				zap.Float64("avgBlockProcessTime (ms)", float64(runningAvg)),
+				zap.Uint64("lastBlockParsed", uint64(lastBlockParsed)),
 			)
 		}
 
@@ -130,6 +132,8 @@ func (s *Sidecar) IndexFromCurrentToTip(ctx context.Context) error {
 			)
 			return err
 		}
+
+		lastBlockParsed = i
 		delta := time.Since(startTime).Milliseconds()
 		blocksProcessed++