Add last submitted da heights to raft

Author: alpe

block/internal/cache/manager.go

@@ -81,7 +81,9 @@ type PendingManager interface {
 	GetPendingHeaders(ctx context.Context) ([]*types.SignedHeader, error)
 	GetPendingData(ctx context.Context) ([]*types.SignedData, error)
 	SetLastSubmittedHeaderHeight(ctx context.Context, height uint64)
+	GetLastSubmittedHeaderHeight() uint64
 	SetLastSubmittedDataHeight(ctx context.Context, height uint64)
+	GetLastSubmittedDataHeight() uint64
 	NumPendingHeaders() uint64
 	NumPendingData() uint64
 }
@@ -347,10 +349,18 @@ func (m *implementation) GetPendingData(ctx context.Context) ([]*types.SignedDat
 	return signedDataList, nil
 }
 
+func (m *implementation) GetLastSubmittedHeaderHeight() uint64 {
+	return m.pendingHeaders.GetLastSubmittedDataHeight()
+}
+
 func (m *implementation) SetLastSubmittedHeaderHeight(ctx context.Context, height uint64) {
 	m.pendingHeaders.SetLastSubmittedHeaderHeight(ctx, height)
 }
 
+func (m *implementation) GetLastSubmittedDataHeight() uint64 {
+	return m.pendingData.GetLastSubmittedDataHeight()
+}
+
 func (m *implementation) SetLastSubmittedDataHeight(ctx context.Context, height uint64) {
 	m.pendingData.SetLastSubmittedDataHeight(ctx, height)
 }

block/internal/cache/pending_base.go

@@ -71,6 +71,10 @@ func (pb *pendingBase[T]) numPending() uint64 {
 	return height - pb.lastHeight.Load()
 }
 
+func (pb *pendingBase[T]) getLastSubmittedHeight() uint64 {
+	return pb.lastHeight.Load()
+}
+
 func (pb *pendingBase[T]) setLastSubmittedHeight(ctx context.Context, newLastSubmittedHeight uint64) {
 	lsh := pb.lastHeight.Load()
 	if newLastSubmittedHeight > lsh && pb.lastHeight.CompareAndSwap(lsh, newLastSubmittedHeight) {

block/internal/cache/pending_data.go

@@ -2,6 +2,7 @@ package cache
 
 import (
 	"context"
+	"errors"
 
 	"github.com/rs/zerolog"
 
@@ -28,8 +29,17 @@ type PendingData struct {
 	base *pendingBase[*types.Data]
 }
 
+var errInFlightData = errors.New("inflight data")
+
 func fetchData(ctx context.Context, store store.Store, height uint64) (*types.Data, error) {
 	_, data, err := store.GetBlockData(ctx, height)
+	if err != nil {
+		return nil, err
+	}
+	// in the executor, WIP data is temporary stored. skip them until the process is completed
+	if data.Height() == 0 {
+		return nil, errInFlightData
+	}
 	return data, err
 }
 
@@ -58,3 +68,6 @@ func (pd *PendingData) NumPendingData() uint64 {
 func (pd *PendingData) SetLastSubmittedDataHeight(ctx context.Context, newLastSubmittedDataHeight uint64) {
 	pd.base.setLastSubmittedHeight(ctx, newLastSubmittedDataHeight)
 }
+func (pd *PendingData) GetLastSubmittedDataHeight() uint64 {
+	return pd.base.getLastSubmittedHeight()
+}

block/internal/cache/pending_headers.go

@@ -2,6 +2,7 @@ package cache
 
 import (
 	"context"
+	"errors"
 
 	"github.com/rs/zerolog"
 
@@ -25,8 +26,17 @@ type PendingHeaders struct {
 	base *pendingBase[*types.SignedHeader]
 }
 
+var errInFlightHeader = errors.New("inflight header")
+
 func fetchSignedHeader(ctx context.Context, store storepkg.Store, height uint64) (*types.SignedHeader, error) {
 	header, err := store.GetHeader(ctx, height)
+	if err != nil {
+		return nil, err
+	}
+	// in the executor, WIP headers are temporary stored. skip them until the process is completed
+	if header.Height() == 0 {
+		return nil, errInFlightHeader
+	}
 	return header, err
 }
 
@@ -55,3 +65,7 @@ func (ph *PendingHeaders) SetLastSubmittedHeaderHeight(ctx context.Context, newL
 func (ph *PendingHeaders) init() error {
 	return ph.base.init()
 }
+
+func (ph *PendingHeaders) GetLastSubmittedDataHeight() uint64 {
+	return ph.base.getLastSubmittedHeight()
+}

block/internal/executing/executor.go

@@ -443,17 +443,18 @@ func (e *Executor) produceBlock() error {
 		}
 
 		raftState := &raft.RaftBlockState{
-			Height:    newHeight,
-			Hash:      header.Hash(),
-			Timestamp: header.BaseHeader.Time,
-			Header:    headerBytes,
-			Data:      dataBytes,
+			Height:                      newHeight,
+			Hash:                        header.Hash(),
+			Timestamp:                   header.BaseHeader.Time,
+			Header:                      headerBytes,
+			Data:                        dataBytes,
+			LastSubmittedDAHeaderHeight: e.cache.GetLastSubmittedHeaderHeight(),
+			LastSubmittedDADataHeight:   e.cache.GetLastSubmittedDataHeight(),
 		}
 		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")
-
 	}
 	if err := batch.Commit(); err != nil {
 		return fmt.Errorf("failed to commit batch: %w", err)

block/internal/syncing/raft_retriever.go

@@ -22,12 +22,14 @@ func (e eventProcessorFn) handle(ctx context.Context, event common.DAHeightEvent
 	return e(ctx, event)
 }
 
+type raftStatePreProcessor func(ctx context.Context, state *raft.RaftBlockState) error
 type raftRetriever struct {
-	raftNode       common.RaftNode
-	wg             sync.WaitGroup
-	logger         zerolog.Logger
-	genesis        genesis.Genesis
-	eventProcessor eventProcessor
+	raftNode              common.RaftNode
+	wg                    sync.WaitGroup
+	logger                zerolog.Logger
+	genesis               genesis.Genesis
+	eventProcessor        eventProcessor
+	raftBlockPreProcessor raftStatePreProcessor
 
 	mtx    sync.Mutex
 	cancel context.CancelFunc
@@ -38,12 +40,14 @@ func newRaftRetriever(
 	genesis genesis.Genesis,
 	logger zerolog.Logger,
 	eventProcessor eventProcessor,
+	raftBlockPostProcessor raftStatePreProcessor,
 ) *raftRetriever {
 	return &raftRetriever{
-		raftNode:       raftNode,
-		genesis:        genesis,
-		logger:         logger,
-		eventProcessor: eventProcessor,
+		raftNode:              raftNode,
+		genesis:               genesis,
+		logger:                logger,
+		eventProcessor:        eventProcessor,
+		raftBlockPreProcessor: raftBlockPostProcessor,
 	}
 }
 
@@ -98,6 +102,9 @@ func (r *raftRetriever) raftApplyLoop(ctx context.Context, applyCh <-chan raft.R
 // consumeRaftBlock applies a block received from raft consensus
 func (r *raftRetriever) consumeRaftBlock(ctx context.Context, state *raft.RaftBlockState) error {
 	r.logger.Debug().Uint64("height", state.Height).Msg("applying raft block")
+	if err := r.raftBlockPreProcessor(ctx, state); err != nil {
+		return err
+	}
 
 	// Unmarshal header and data
 	var header types.SignedHeader
@@ -122,7 +129,7 @@ func (r *raftRetriever) consumeRaftBlock(ctx context.Context, state *raft.RaftBl
 	event := common.DAHeightEvent{
 		Header:   &header,
 		Data:     &data,
-		DaHeight: 0, // raft events don't have DA height context
+		DaHeight: 0, // raft events don't have DA height context, yet as DA submission is asynchronous
 	}
 	return r.eventProcessor.handle(ctx, event)
 }

block/internal/syncing/syncer.go

@@ -11,6 +11,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/evstack/ev-node/pkg/raft"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/rs/zerolog"
 	"golang.org/x/sync/errgroup"
@@ -79,7 +80,7 @@ func NewSyncer(
 	store store.Store,
 	exec coreexecutor.Executor,
 	daClient da.Client,
-	cache cache.CacheManager,
+	cache cache.Manager,
 	metrics *common.Metrics,
 	config config.Config,
 	genesis genesis.Genesis,
@@ -108,7 +109,12 @@ func NewSyncer(
 		logger:            logger.With().Str("component", "syncer").Logger(),
 	}
 	if raftNode != nil && !reflect.ValueOf(raftNode).IsNil() {
-		s.raftRetriever = newRaftRetriever(raftNode, genesis, logger, eventProcessorFn(s.pipeEvent))
+		s.raftRetriever = newRaftRetriever(raftNode, genesis, logger, eventProcessorFn(s.pipeEvent), func(ctx context.Context, state *raft.RaftBlockState) error {
+			s.logger.Info().Uint64("header_height", state.LastSubmittedDAHeaderHeight).Uint64("data_height", state.LastSubmittedDADataHeight).Msg("+++ received raft block state")
+			cache.SetLastSubmittedHeaderHeight(ctx, state.LastSubmittedDAHeaderHeight)
+			cache.SetLastSubmittedDataHeight(ctx, state.LastSubmittedDADataHeight)
+			return nil
+		})
 	}
 	return s
 }

pkg/raft/types.go

@@ -4,11 +4,13 @@ import "fmt"
 
 // RaftBlockState represents a replicated block state
 type RaftBlockState struct {
-	Height    uint64
-	Hash      []byte
-	Timestamp uint64
-	Header    []byte
-	Data      []byte
+	Height                      uint64 `json:"height"`
+	LastSubmittedDAHeaderHeight uint64 `json:"last_submitted_da_header_height"`
+	LastSubmittedDADataHeight   uint64 `json:"last_submitted_da_data_height"`
+	Hash                        []byte `json:"hash"`
+	Timestamp                   uint64 `json:"timestamp"`
+	Header                      []byte `json:"header"`
+	Data                        []byte `json:"data"`
 }
 
 // assertValid checks basic constraints but does not ensure that no gaps exist or chain continuity

test/e2e/failover_e2e_test.go

@@ -13,7 +13,6 @@ import (
 	"os"
 	"path/filepath"
 	"slices"
-	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -230,7 +229,7 @@ func TestLeaseFailoverE2E(t *testing.T) {
 	}, 2*must(time.ParseDuration(DefaultDABlockTime)), 100*time.Millisecond)
 
 	t.Log("+++ Verifying no DA gaps...")
-	verifyBlocks(t, daStartHeight, lastDABlockNewLeader, jwtSecret, testEndpoints.GetDAAddress(), genesisHeight, state.LastBlockHeight)
+	verifyDABlocks(t, daStartHeight, lastDABlockNewLeader, jwtSecret, testEndpoints.GetDAAddress(), genesisHeight, state.LastBlockHeight)
 
 	// Cleanup processes
 	clusterNodes.killAll()
@@ -260,13 +259,13 @@ func verifyNoDoubleSigning(t *testing.T, clusterNodes *raftClusterNodes, genesis
 				require.NoError(t, err)
 				t.Logf("%s: %v", nodes[0], rsp.Block)
 			}
-			//t.FailNow()
+			t.FailNow()
 		}
 	}
 }
 
-// verifyBlocks checks that DA block heights form a continuous sequence without gaps
-func verifyBlocks(t *testing.T, daStartHeight, lastDABlock uint64, jwtSecret string, daAddress string, genesisHeight, lastEVBlock uint64) {
+// verifyDABlocks checks that DA block heights form a continuous sequence without gaps
+func verifyDABlocks(t *testing.T, daStartHeight, lastDABlock uint64, jwtSecret string, daAddress string, genesisHeight, lastEVBlock uint64) {
 	t.Helper()
 	rpcClient, err := jsonrpc.NewClient(t.Context(), zerolog.Nop(), daAddress, jwtSecret, defaultMaxBlobSize)
 	require.NoError(t, err)
@@ -287,8 +286,8 @@ func verifyBlocks(t *testing.T, daStartHeight, lastDABlock uint64, jwtSecret str
 		require.NoError(t, err)
 
 		for _, blob := range blobs {
-			if evHeight, hash := extractBlockHeight(t, blob); evHeight != 0 {
-				t.Logf("extracting block height from blob (da height %d): %x", daHeight, evHeight)
+			if evHeight, hash, blobType := extractBlockHeight(t, blob); evHeight != 0 {
+				t.Logf("extracting block height from blob (da height %d): %4d (%s)", daHeight, evHeight, blobType)
 				if height, ok := deduplicationCache[hash.String()]; ok {
 					require.Equal(t, evHeight, height)
 					continue
@@ -309,16 +308,21 @@ func verifyBlocks(t *testing.T, daStartHeight, lastDABlock uint64, jwtSecret str
 }
 
 // extractBlockHeight attempts to decode a blob as SignedHeader or SignedData and extract the block height
-func extractBlockHeight(t *testing.T, blob []byte) (uint64, types.Hash) {
+func extractBlockHeight(t *testing.T, blob []byte) (uint64, types.Hash, string) {
 	t.Helper()
+	if len(blob) == 0 {
+		t.Log("empty blob, skipping")
+		return 0, nil, ""
+	}
 	var headerPb pb.SignedHeader
 	if err := proto.Unmarshal(blob, &headerPb); err == nil {
 		var signedHeader types.SignedHeader
 		if err := signedHeader.FromProto(&headerPb); err == nil {
 			if err := signedHeader.Header.ValidateBasic(); err == nil {
-				return signedHeader.Height(), signedHeader.Hash()
+				return signedHeader.Height(), signedHeader.Hash(), "header"
 			} else {
-				t.Logf("invalid header: %v", err)
+				jsonBZ, _ := json.MarshalIndent(signedHeader.Header, "", "  ")
+				t.Logf("invalid header: %v: %s", err, string(jsonBZ))
 			}
 		} else {
 			t.Logf("failed to unmarshal signed header: %v", err)
@@ -330,12 +334,12 @@ func extractBlockHeight(t *testing.T, blob []byte) (uint64, types.Hash) {
 	var signedData types.SignedData
 	if err := signedData.UnmarshalBinary(blob); err == nil {
 		if signedData.Metadata != nil {
-			return signedData.Height(), signedData.Hash()
+			return signedData.Height(), signedData.Hash(), "data"
 		}
 	} else {
 		t.Logf("failed to unmarshal signed data: %v", err)
 	}
-	return 0, nil
+	return 0, nil, ""
 }
 
 func initChain(t *testing.T, sut *SystemUnderTest, workDir string) string {
@@ -399,7 +403,7 @@ func setupRaftSequencerNode(
 		"--evnode.raft.node_id="+nodeID,
 		"--evnode.raft.raft_addr="+raftAddr,
 		"--evnode.raft.raft_dir="+raftDir,
-		"--evnode.raft.bootstrap="+strconv.FormatBool(bootstrap),
+		"--evnode.raft.bootstrap=true",
 		"--evnode.raft.peers="+strings.Join(raftPeers, ","),
 		"--evnode.raft.snap_count=10",
 		"--evnode.raft.send_timeout=300ms",