x

Author: alpe

block/components.go

@@ -119,6 +119,11 @@ func (bc *Components) Stop() error {
 			errs = errors.Join(errs, fmt.Errorf("failed to stop submitter: %w", err))
 		}
 	}
+	if bc.Cache != nil {
+		if err := bc.Cache.SaveToDisk(); err != nil {
+			errs = errors.Join(errs, fmt.Errorf("failed to save caches: %w", err))
+		}
+	}
 
 	return errs
 }

block/internal/executing/executor.go

@@ -408,6 +408,30 @@ func (e *Executor) produceBlock() error {
 		return fmt.Errorf("failed to validate block: %w", err)
 	}
 
+	// Propose block to raft to share state in the cluster
+	if e.raftNode != nil {
+		headerBytes, err := header.MarshalBinary()
+		if err != nil {
+			return fmt.Errorf("failed to marshal header: %w", err)
+		}
+		dataBytes, err := data.MarshalBinary()
+		if err != nil {
+			return fmt.Errorf("failed to marshal data: %w", err)
+		}
+
+		raftState := &raft.RaftBlockState{
+			Height:    newHeight,
+			Hash:      header.Hash(),
+			Timestamp: header.BaseHeader.Time,
+			Header:    headerBytes,
+			Data:      dataBytes,
+		}
+		if err := e.raftNode.Broadcast(e.ctx, raftState); err != nil {
+			return fmt.Errorf("failed to propose block to raft: %w", err)
+		}
+		e.logger.Debug().Uint64("height", newHeight).Msg("proposed block to raft")
+	}
+
 	batch, err := e.store.NewBatch(e.ctx)
 	if err != nil {
 		return fmt.Errorf("failed to create batch: %w", err)
@@ -433,30 +457,6 @@ func (e *Executor) produceBlock() error {
 	e.setLastState(newState)
 	e.metrics.Height.Set(float64(newState.LastBlockHeight))
 
-	// Propose block to raft before p2p broadcast if raft is enabled
-	if e.raftNode != nil {
-		headerBytes, err := header.MarshalBinary()
-		if err != nil {
-			return fmt.Errorf("failed to marshal header: %w", err)
-		}
-		dataBytes, err := data.MarshalBinary()
-		if err != nil {
-			return fmt.Errorf("failed to marshal data: %w", err)
-		}
-
-		raftState := &raft.RaftBlockState{
-			Height:    newHeight,
-			Hash:      header.Hash(),
-			Timestamp: header.BaseHeader.Time,
-			Header:    headerBytes,
-			Data:      dataBytes,
-		}
-		if err := e.raftNode.Broadcast(e.ctx, raftState); err != nil {
-			return fmt.Errorf("failed to propose block to raft: %w", err)
-		}
-		e.logger.Debug().Uint64("height", newHeight).Msg("proposed block to raft")
-	}
-
 	// broadcast header and data to P2P network
 	g, ctx := errgroup.WithContext(e.ctx)
 	g.Go(func() error { return e.headerBroadcaster.WriteToStoreAndBroadcast(ctx, header) })
@@ -465,7 +465,6 @@ func (e *Executor) produceBlock() error {
 		e.logger.Error().Err(err).Msg("failed to broadcast header and/data")
 		// don't fail block production on broadcast error
 	}
-
 	e.recordBlockMetrics(data)
 
 	e.logger.Info().

block/internal/submitting/da_submitter.go

@@ -191,7 +191,15 @@ func (s *DASubmitter) SubmitHeaders(ctx context.Context, cache cache.Manager) er
 		return nil
 	}
 
-	s.logger.Info().Int("count", len(headers)).Msg("submitting headers to DA")
+	heights := make([]uint64, len(headers))
+	for i, sd := range headers {
+		heights[i] = sd.Height()
+		if len(sd.Header.ProposerAddress) == 0 || bytes.Equal(sd.Header.ProposerAddress, make([]byte, 32)) {
+			panic("empty  proposer address")
+		}
+	}
+
+	s.logger.Info().Uints64("heights", heights).Int("count", len(headers)).Msg("submitting headers to DA")
 
 	return submitToDA(s, ctx, headers,
 		func(header *types.SignedHeader) ([]byte, error) {
@@ -239,7 +247,11 @@ func (s *DASubmitter) SubmitData(ctx context.Context, cache cache.Manager, signe
 		return nil // No non-empty data to submit
 	}
 
-	s.logger.Info().Int("count", len(signedDataList)).Msg("submitting data to DA")
+	heights := make([]uint64, len(signedDataList))
+	for i, sd := range signedDataList {
+		heights[i] = sd.Height()
+	}
+	s.logger.Info().Uints64("heights", heights).Int("count", len(signedDataList)).Msg("submitting data to DA")
 
 	return submitToDA(s, ctx, signedDataList,
 		func(signedData *types.SignedData) ([]byte, error) {

block/internal/submitting/submitter.go

@@ -103,6 +103,7 @@ func (s *Submitter) Start(ctx context.Context) error {
 
 	// Start DA submission loop if signer is available (aggregator nodes only)
 	if s.signer != nil {
+		s.logger.Info().Msg("starting DA submission loop")
 		s.wg.Add(1)
 		go func() {
 			defer s.wg.Done()
@@ -113,7 +114,14 @@ func (s *Submitter) Start(ctx context.Context) error {
 	// Start DA inclusion processing loop (both sync and aggregator nodes)
 	s.wg.Add(1)
 	go func() {
-		defer s.wg.Done()
+		defer func() {
+			// fetch all from DA to ensure final height is set
+			// todo: revisit this for a graceful shutdown scenario
+			tCtx, cancelFunc := context.WithTimeout(context.Background(), 2*time.Second)
+			defer cancelFunc()
+			s.processAllFromDA(tCtx)
+			s.wg.Done()
+		}()
 		s.processDAInclusionLoop()
 	}()
 
@@ -193,48 +201,63 @@ func (s *Submitter) processDAInclusionLoop() {
 		case <-s.ctx.Done():
 			return
 		case <-ticker.C:
-			currentDAIncluded := s.GetDAIncludedHeight()
+			s.processAllFromDA(s.ctx)
+		}
+	}
+}
 
-			for {
-				nextHeight := currentDAIncluded + 1
+func (s *Submitter) processAllFromDA(ctx context.Context) {
+	currentDAIncluded := s.GetDAIncludedHeight()
 
-				// Get block data first
-				header, data, err := s.store.GetBlockData(s.ctx, nextHeight)
-				if err != nil {
-					break
-				}
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
 
-				// Check if this height is DA included
-				if included, err := s.IsHeightDAIncluded(nextHeight, header, data); err != nil || !included {
-					break
-				}
+		nextHeight := currentDAIncluded + 1
+
+		// Get block data first
+		header, data, err := s.store.GetBlockData(s.ctx, nextHeight)
+		if err != nil {
+			return
+		}
 
-				s.logger.Debug().Uint64("height", nextHeight).Msg("advancing DA included height")
+		// Check if this height is DA included
+		if included, err := s.IsHeightDAIncluded(nextHeight, header, data); err != nil || !included {
+			return
+		}
 
-				// Set sequencer height to DA height mapping using already retrieved data
-				if err := s.setSequencerHeightToDAHeight(s.ctx, nextHeight, header, data, currentDAIncluded == 0); err != nil {
-					s.logger.Error().Err(err).Uint64("height", nextHeight).Msg("failed to set sequencer height to DA height mapping")
-					break
-				}
+		s.logger.Debug().Uint64("height", nextHeight).Msg("advancing DA included height")
 
-				// Set final height in executor
-				if err := s.setFinalWithRetry(nextHeight); err != nil {
-					s.sendCriticalError(fmt.Errorf("failed to set final height: %w", err))
-					s.logger.Error().Err(err).Uint64("height", nextHeight).Msg("failed to set final height")
-					break
-				}
+		// Set sequencer height to DA height mapping using already retrieved data
+		if err := s.setSequencerHeightToDAHeight(s.ctx, nextHeight, header, data, currentDAIncluded == 0); err != nil {
+			s.logger.Error().Err(err).Uint64("height", nextHeight).Msg("failed to set sequencer height to DA height mapping")
+			return
+		}
 
-				// Update DA included height
-				s.SetDAIncludedHeight(nextHeight)
-				currentDAIncluded = nextHeight
+		// Set final height in executor
+		if err := s.setFinalWithRetry(nextHeight); err != nil {
+			s.sendCriticalError(fmt.Errorf("failed to set final height: %w", err))
+			s.logger.Error().Err(err).Uint64("height", nextHeight).Msg("failed to set final height")
+			return
+		}
 
-				// Persist DA included height
-				bz := make([]byte, 8)
-				binary.LittleEndian.PutUint64(bz, nextHeight)
-				if err := s.store.SetMetadata(s.ctx, store.DAIncludedHeightKey, bz); err != nil {
-					s.logger.Error().Err(err).Uint64("height", nextHeight).Msg("failed to persist DA included height")
-				}
-			}
+		// Update DA included height
+		s.SetDAIncludedHeight(nextHeight)
+		currentDAIncluded = nextHeight
+
+		// Persist DA included height
+		bz := make([]byte, 8)
+		binary.LittleEndian.PutUint64(bz, nextHeight)
+		if err := s.store.SetMetadata(s.ctx, store.DAIncludedHeightKey, bz); err != nil {
+			s.logger.Error().Err(err).Uint64("height", nextHeight).Msg("failed to persist DA included height")
+		}
+		if s.signer == nil {
+			// bump height to keep cache in sync when submission loop is not active
+			s.cache.SetLastSubmittedDataHeight(s.ctx, currentDAIncluded)
+			s.cache.SetLastSubmittedHeaderHeight(s.ctx, currentDAIncluded)
 		}
 	}
 }
@@ -280,6 +303,7 @@ func (s *Submitter) SetDAIncludedHeight(height uint64) {
 // initializeDAIncludedHeight loads the DA included height from store
 func (s *Submitter) initializeDAIncludedHeight(ctx context.Context) error {
 	if height, err := s.store.GetMetadata(ctx, store.DAIncludedHeightKey); err == nil && len(height) == 8 {
+		s.logger.Debug().Uint64("height", binary.LittleEndian.Uint64(height)).Msg("Initial DA included height loaded from store")
 		s.SetDAIncludedHeight(binary.LittleEndian.Uint64(height))
 	}
 	return nil
@@ -367,7 +391,7 @@ func (s *Submitter) IsHeightDAIncluded(height uint64, header *types.SignedHeader
 	_, headerIncluded := s.cache.GetHeaderDAIncluded(headerHash)
 	_, dataIncluded := s.cache.GetDataDAIncluded(dataHash)
 
-	dataIncluded = bytes.Equal(data.DACommitment(), common.DataHashForEmptyTxs) || dataIncluded
+	dataIncluded = dataIncluded || bytes.Equal(data.DACommitment(), common.DataHashForEmptyTxs)
 
 	return headerIncluded && dataIncluded, nil
 }

block/internal/syncing/syncer.go

@@ -421,7 +421,12 @@ func (s *Syncer) processHeightEvent(event *common.DAHeightEvent) {
 	if err := s.trySyncNextBlock(event); err != nil {
 		s.logger.Error().Err(err).Msg("failed to sync next block")
 		// If the error is not due to an validation error, re-store the event as pending
-		if !errors.Is(err, errInvalidBlock) {
+		switch {
+		case errors.Is(err, errInvalidBlock):
+			// do not reschedule
+		case errors.Is(err, errInvalidState):
+			s.logger.Fatal().Err(err).Msg("Invalid state, shutting down")
+		default:
 			s.cache.SetPendingEvent(height, event)
 		}
 		return
@@ -450,7 +455,7 @@ func (s *Syncer) trySyncNextBlock(event *common.DAHeightEvent) error {
 	// Compared to the executor logic where the current block needs to be applied first,
 	// here only the previous block needs to be applied to proceed to the verification.
 	// The header validation must be done before applying the block to avoid executing gibberish
-	if err := s.validateBlock(header, data); err != nil {
+	if err := s.validateBlock(header, data, currentState); err != nil {
 		// remove header as da included (not per se needed, but keep cache clean)
 		s.cache.RemoveHeaderDAIncluded(header.Hash().String())
 		return errors.Join(errInvalidBlock, fmt.Errorf("failed to validate block: %w", err))
@@ -555,14 +560,13 @@ func (s *Syncer) executeTxsWithRetry(ctx context.Context, rawTxs [][]byte, heade
 	return nil, nil
 }
 
+var errInvalidState = errors.New("invalid state")
+
 // validateBlock validates a synced block
 // NOTE: if the header was gibberish and somehow passed all validation prior but the data was correct
 // or if the data was gibberish and somehow passed all validation prior but the header was correct
 // we are still losing both in the pending event. This should never happen.
-func (s *Syncer) validateBlock(
-	header *types.SignedHeader,
-	data *types.Data,
-) error {
+func (s *Syncer) validateBlock(header *types.SignedHeader, data *types.Data, state types.State, ) error {
 	// Set custom verifier for aggregator node signature
 	header.SetCustomVerifierForSyncNode(s.options.SyncNodeSignatureBytesProvider)
 
@@ -575,7 +579,23 @@ func (s *Syncer) validateBlock(
 	if err := types.Validate(header, data); err != nil {
 		return fmt.Errorf("header-data validation failed: %w", err)
 	}
+	if state.LastBlockHeight < s.genesis.InitialHeight {
+		return nil
+	}
+	// Validate header against state
+	if header.Height() != state.LastBlockHeight+1 {
+		return fmt.Errorf("%w: invalid block height - got: %d, want: %d", errInvalidState, header.Height(), state.LastBlockHeight+1)
+	}
 
+	if !header.Time().After(state.LastBlockTime) {
+		return fmt.Errorf("%w: invalid block time - got: %v, last: %v", errInvalidState, header.Time(), state.LastBlockTime)
+	}
+	if !bytes.Equal(header.LastHeaderHash, state.LastHeaderHash) {
+		return fmt.Errorf("%w: invalid last header hash - got: %x, want: %x", errInvalidState, header.LastHeaderHash, header.LastHeaderHash)
+	}
+	if !bytes.Equal(header.AppHash, state.AppHash) {
+		return fmt.Errorf("%w: invalid last appHash hash - got: %x, want: %x", errInvalidState, header.AppHash, header.AppHash)
+	}
 	return nil
 }
 

node/full.go

@@ -105,7 +105,7 @@ func newFullNode(
 		nodeConfig.P2P.Peers = "" // raft cluster provides most up to date blocks. No need for p2p peers.
 	case nodeConfig.Node.Aggregator && !nodeConfig.Raft.Enable:
 		leaderElection = AlwaysLeader{}
-	case !nodeConfig.Node.Aggregator:
+	case !nodeConfig.Node.Aggregator && !nodeConfig.Raft.Enable:
 		leaderElection = AlwaysFollower{}
 	default:
 		return nil, fmt.Errorf("raft config must be used in sequencer setup only")
@@ -120,7 +120,7 @@ func newFullNode(
 		aggregatorComponentFunc: func(ctx context.Context) error {
 			logger.Info().Msg("Starting aggregator-MODE")
 			nodeConfig.Node.Aggregator = true
-			nodeConfig.P2P.Peers = ""
+			nodeConfig.P2P.Peers = "" // peers are not supported in aggregator mode
 			m, err := newAggregatorMode(nodeConfig, nodeKey, signer, genesis, database, exec, sequencer, da, logger, rktStore, mainKV, blockMetrics, nodeOpts, raftNode)
 			if err != nil {
 				return err