Init chain on empty state; Forward runtime errors

Author: alpe

block/components.go

@@ -46,7 +46,8 @@ func (bc *Components) GetLastState() types.State {
 	return types.State{}
 }
 
-// Start starts all components and monitors for critical errors
+// Start starts all components and monitors for critical errors.
+// It is blocking and returns when the context is cancelled or an error occurs
 func (bc *Components) Start(ctx context.Context) error {
 	ctxWithCancel, cancel := context.WithCancel(ctx)
 

block/internal/executing/executor.go

@@ -199,7 +199,7 @@ func (e *Executor) initializeState() error {
 			LastBlockHeight: e.genesis.InitialHeight - 1,
 			LastBlockTime:   e.genesis.StartTime,
 			AppHash:         stateRoot,
-			DAHeight:        0,
+			DAHeight:        e.genesis.DAStartHeight,
 		}
 	}
 

block/internal/syncing/syncer.go

@@ -183,21 +183,34 @@ func (s *Syncer) initializeState() error {
 	// Load state from store
 	state, err := s.store.GetState(s.ctx)
 	if err != nil {
-		// Use genesis state if no state exists
+		// Initialize new chain state for a fresh full node (no prior state on disk)
+		// Mirror executor initialization to ensure AppHash matches headers produced by the sequencer.
+		stateRoot, _, initErr := s.exec.InitChain(
+			s.ctx,
+			s.genesis.StartTime,
+			s.genesis.InitialHeight,
+			s.genesis.ChainID,
+		)
+		if initErr != nil {
+			return fmt.Errorf("failed to initialize execution client: %w", initErr)
+		}
+
 		state = types.State{
 			ChainID:         s.genesis.ChainID,
 			InitialHeight:   s.genesis.InitialHeight,
 			LastBlockHeight: s.genesis.InitialHeight - 1,
 			LastBlockTime:   s.genesis.StartTime,
-			DAHeight:        0,
+			DAHeight:        s.genesis.DAStartHeight,
+			AppHash:         stateRoot,
 		}
 	}
-
+	if state.DAHeight < s.genesis.DAStartHeight {
+		return fmt.Errorf("DA height (%d) is lower than DA start height (%d)", state.DAHeight, s.genesis.DAStartHeight)
+	}
 	s.SetLastState(state)
 
 	// Set DA height
-	daHeight := max(state.DAHeight, s.genesis.DAStartHeight)
-	s.SetDAHeight(daHeight)
+	s.SetDAHeight(state.DAHeight)
 
 	s.logger.Info().
 		Uint64("height", state.LastBlockHeight).
@@ -442,6 +455,7 @@ func (s *Syncer) trySyncNextBlock(event *common.DAHeightEvent) error {
 		if !errors.Is(err, errInvalidState) && !errors.Is(err, errInvalidBlock) {
 			return errors.Join(errInvalidBlock, err)
 		}
+		return err
 	}
 
 	// Apply block

block/internal/syncing/syncer_test.go

@@ -111,6 +111,7 @@ func TestSyncer_validateBlock_DataHashMismatch(t *testing.T) {
 	cfg := config.DefaultConfig()
 	gen := genesis.Genesis{ChainID: "tchain", InitialHeight: 1, StartTime: time.Now().Add(-time.Second), ProposerAddress: addr}
 	mockExec := testmocks.NewMockExecutor(t)
+	mockExec.EXPECT().InitChain(mock.Anything, mock.Anything, uint64(1), "tchain").Return([]byte("app0"), uint64(1024), nil).Once()
 
 	s := NewSyncer(
 		st,
@@ -159,7 +160,9 @@ func TestProcessHeightEvent_SyncsAndUpdatesState(t *testing.T) {
 	gen := genesis.Genesis{ChainID: "tchain", InitialHeight: 1, StartTime: time.Now().Add(-time.Second), ProposerAddress: addr}
 
 	mockExec := testmocks.NewMockExecutor(t)
+	mockExec.EXPECT().InitChain(mock.Anything, mock.Anything, uint64(1), "tchain").Return([]byte("app0"), uint64(1024), nil).Once()
 
+	errChan := make(chan error, 1)
 	s := NewSyncer(
 		st,
 		mockExec,
@@ -172,7 +175,7 @@ func TestProcessHeightEvent_SyncsAndUpdatesState(t *testing.T) {
 		common.NewMockBroadcaster[*types.Data](t),
 		zerolog.Nop(),
 		common.DefaultBlockOptions(),
-		make(chan error, 1),
+		errChan,
 	)
 
 	require.NoError(t, s.initializeState())
@@ -190,6 +193,7 @@ func TestProcessHeightEvent_SyncsAndUpdatesState(t *testing.T) {
 	evt := common.DAHeightEvent{Header: hdr, Data: data, DaHeight: 1}
 	s.processHeightEvent(&evt)
 
+	requireEmptyChan(t, errChan)
 	h, err := st.Height(context.Background())
 	require.NoError(t, err)
 	assert.Equal(t, uint64(1), h)
@@ -209,7 +213,9 @@ func TestSequentialBlockSync(t *testing.T) {
 	gen := genesis.Genesis{ChainID: "tchain", InitialHeight: 1, StartTime: time.Now().Add(-time.Second), ProposerAddress: addr}
 
 	mockExec := testmocks.NewMockExecutor(t)
+	mockExec.EXPECT().InitChain(mock.Anything, mock.Anything, uint64(1), "tchain").Return([]byte("app0"), uint64(1024), nil).Once()
 
+	errChan := make(chan error, 1)
 	s := NewSyncer(
 		st,
 		mockExec,
@@ -222,7 +228,7 @@ func TestSequentialBlockSync(t *testing.T) {
 		common.NewMockBroadcaster[*types.Data](t),
 		zerolog.Nop(),
 		common.DefaultBlockOptions(),
-		make(chan error, 1),
+		errChan,
 	)
 	require.NoError(t, s.initializeState())
 	s.ctx = context.Background()
@@ -265,6 +271,7 @@ func TestSequentialBlockSync(t *testing.T) {
 	assert.True(t, ok)
 	_, ok = cm.GetDataDAIncluded(data2.DACommitment().String())
 	assert.True(t, ok)
+	requireEmptyChan(t, errChan)
 }
 
 func TestSyncer_sendNonBlockingSignal(t *testing.T) {
@@ -328,14 +335,16 @@ func TestSyncer_processPendingEvents(t *testing.T) {
 func TestSyncLoopPersistState(t *testing.T) {
 	ds := dssync.MutexWrap(datastore.NewMapDatastore())
 	st := store.New(ds)
-	cm, err := cache.NewManager(config.DefaultConfig(), st, zerolog.Nop())
+	cfg := config.DefaultConfig()
+	cfg.ClearCache = true
+
+	cacheMgr, err := cache.NewManager(cfg, st, zerolog.Nop())
 	require.NoError(t, err)
 
-	myDAHeightOffset := uint64(1)
-	myFutureDAHeight := uint64(9)
+	const myDAHeightOffset = uint64(1)
+	const numBlocks = uint64(5)
 
 	addr, pub, signer := buildSyncTestSigner(t)
-	cfg := config.DefaultConfig()
 	gen := genesis.Genesis{ChainID: "tchain", InitialHeight: 1, StartTime: time.Now().Add(-time.Second), ProposerAddress: addr, DAStartHeight: myDAHeightOffset}
 
 	dummyExec := execution.NewDummyExecutor()
@@ -353,19 +362,20 @@ func TestSyncLoopPersistState(t *testing.T) {
 	mockP2PDataStore := common.NewMockBroadcaster[*types.Data](t)
 	mockP2PDataStore.EXPECT().Store().Return(mockDataStore).Maybe()
 
+	errorCh := make(chan error, 1)
 	syncerInst1 := NewSyncer(
 		st,
 		dummyExec,
 		nil,
-		cm,
+		cacheMgr,
 		common.NopMetrics(),
 		cfg,
 		gen,
 		mockP2PHeaderStore,
 		mockP2PDataStore,
 		zerolog.Nop(),
 		common.DefaultBlockOptions(),
-		make(chan error, 1),
+		errorCh,
 	)
 	require.NoError(t, syncerInst1.initializeState())
 
@@ -378,13 +388,15 @@ func TestSyncLoopPersistState(t *testing.T) {
 
 	// with n da blobs fetched
 	var prevHeaderHash, prevAppHash []byte
-	for i := range myFutureDAHeight - myDAHeightOffset {
-		chainHeight, daHeight := i+1, i+myDAHeightOffset
+	prevAppHash, _, _ = execution.NewDummyExecutor().InitChain(t.Context(), gen.StartTime, gen.DAStartHeight, gen.ChainID)
+	for i := range numBlocks {
+		chainHeight, daHeight := gen.InitialHeight+i, i+myDAHeightOffset
+		blockTime := gen.StartTime.Add(time.Duration(chainHeight+1) * time.Second)
 		emptyData := &types.Data{
 			Metadata: &types.Metadata{
 				ChainID: gen.ChainID,
 				Height:  chainHeight,
-				Time:    uint64(time.Now().Add(time.Duration(chainHeight) * time.Second).UnixNano()),
+				Time:    uint64(blockTime.UnixNano()),
 			},
 		}
 		_, sigHeader := makeSignedHeaderBytes(t, gen.ChainID, chainHeight, addr, pub, signer, prevAppHash, emptyData, prevHeaderHash)
@@ -401,6 +413,7 @@ func TestSyncLoopPersistState(t *testing.T) {
 	}
 
 	// stop at next height
+	myFutureDAHeight := myDAHeightOffset + numBlocks
 	daRtrMock.On("RetrieveFromDA", mock.Anything, myFutureDAHeight).
 		Run(func(_ mock.Arguments) {
 			// wait for consumer to catch up
@@ -412,35 +425,39 @@ func TestSyncLoopPersistState(t *testing.T) {
 		Return(nil, coreda.ErrHeightFromFuture)
 
 	go syncerInst1.processLoop()
-	// dssync from DA until stop height reached
+	// sync from DA until stop height reached
 	syncerInst1.syncLoop()
+	requireEmptyChan(t, errorCh)
+
 	t.Log("syncLoop on instance1 completed")
+	require.Equal(t, myFutureDAHeight, syncerInst1.GetDAHeight())
+	lastStateDAHeight := syncerInst1.GetLastState().DAHeight
 
 	// wait for all events consumed
-	require.NoError(t, cm.SaveToDisk())
+	require.NoError(t, cacheMgr.SaveToDisk())
 	t.Log("processLoop on instance1 completed")
 
 	// then
 	daRtrMock.AssertExpectations(t)
 	p2pHndlMock.AssertExpectations(t)
+	require.Len(t, syncerInst1.heightInCh, 0)
 
 	// and all processed - verify no events remain at heights we tested
-	event1 := syncerInst1.cache.GetNextPendingEvent(1)
-	assert.Nil(t, event1)
-	event2 := syncerInst1.cache.GetNextPendingEvent(2)
-	assert.Nil(t, event2)
-	assert.Len(t, syncerInst1.heightInCh, 0)
-
+	for i := range numBlocks {
+		blockHeight := gen.InitialHeight + i + 1
+		event := syncerInst1.cache.GetNextPendingEvent(blockHeight)
+		require.Nil(t, event, "event at height %d should have been removed", blockHeight)
+	}
 	// and when new instance is up on restart
-	cm, err = cache.NewManager(config.DefaultConfig(), st, zerolog.Nop())
+	cacheMgr, err = cache.NewManager(cfg, st, zerolog.Nop())
 	require.NoError(t, err)
-	require.NoError(t, cm.LoadFromDisk())
+	require.NoError(t, cacheMgr.LoadFromDisk())
 
 	syncerInst2 := NewSyncer(
 		st,
 		dummyExec,
 		nil,
-		cm,
+		cacheMgr,
 		common.NopMetrics(),
 		cfg,
 		gen,
@@ -451,7 +468,7 @@ func TestSyncLoopPersistState(t *testing.T) {
 		make(chan error, 1),
 	)
 	require.NoError(t, syncerInst2.initializeState())
-	require.Equal(t, myFutureDAHeight-1, syncerInst2.GetDAHeight())
+	require.Equal(t, lastStateDAHeight, syncerInst2.GetDAHeight())
 
 	ctx, cancel = context.WithCancel(t.Context())
 	t.Cleanup(cancel)
@@ -465,7 +482,7 @@ func TestSyncLoopPersistState(t *testing.T) {
 		Run(func(arg mock.Arguments) {
 			cancel()
 			// retrieve last one again
-			assert.Equal(t, myFutureDAHeight-1, arg.Get(1).(uint64))
+			assert.Equal(t, lastStateDAHeight, arg.Get(1).(uint64))
 		}).
 		Return(nil, nil)
 
@@ -622,3 +639,12 @@ func TestSyncer_InitializeState_CallsReplayer(t *testing.T) {
 	// Verify that GetLatestHeight was called (proves Replayer was invoked)
 	mockExec.AssertCalled(t, "GetLatestHeight", mock.Anything)
 }
+
+func requireEmptyChan(t *testing.T, errorCh chan error) {
+	t.Helper()
+	select {
+	case err := <-errorCh:
+		t.Fatalf("syncLoop on instance1 failed: %v", err)
+	default:
+	}
+}