fetch DA height

Author: julienrbrt

apps/evm/server/force_inclusion_test.go

@@ -73,6 +73,10 @@ func (m *mockDA) HasForcedInclusionNamespace() bool {
 	return true
 }
 
+func (m *mockDA) GetLatestDAHeight(_ context.Context) (uint64, error) {
+	return 0, nil
+}
+
 func TestForceInclusionServer_handleSendRawTransaction_Success(t *testing.T) {
 	testHeight := uint64(100)
 

block/internal/da/client.go

@@ -299,6 +299,23 @@ func (c *client) Retrieve(ctx context.Context, height uint64, namespace []byte)
 	}
 }
 
+// GetLatestDAHeight returns the latest height available on the DA layer by
+// querying the network head.
+func (c *client) GetLatestDAHeight(ctx context.Context) (uint64, error) {
+	headCtx, cancel := context.WithTimeout(ctx, c.defaultTimeout)
+	defer cancel()
+
+	header, err := c.headerAPI.NetworkHead(headCtx)
+	if err != nil {
+		return 0, fmt.Errorf("failed to get DA network head: %w", err)
+	}
+	if header == nil {
+		return 0, fmt.Errorf("DA network head returned nil header")
+	}
+
+	return header.Height, nil
+}
+
 // RetrieveForcedInclusion retrieves blobs from the forced inclusion namespace at the specified height.
 func (c *client) RetrieveForcedInclusion(ctx context.Context, height uint64) datypes.ResultRetrieve {
 	if !c.hasForcedNamespace {

block/internal/da/interface.go

@@ -17,6 +17,9 @@ type Client interface {
 	// Get retrieves blobs by their IDs. Used for visualization and fetching specific blobs.
 	Get(ctx context.Context, ids []datypes.ID, namespace []byte) ([]datypes.Blob, error)
 
+	// GetLatestDAHeight returns the latest height available on the DA layer..
+	GetLatestDAHeight(ctx context.Context) (uint64, error)
+
 	// Namespace accessors.
 	GetHeaderNamespace() []byte
 	GetDataNamespace() []byte

block/internal/da/tracing.go

@@ -123,6 +123,20 @@ func (t *tracedClient) Validate(ctx context.Context, ids []datypes.ID, proofs []
 	return res, nil
 }
 
+func (t *tracedClient) GetLatestDAHeight(ctx context.Context) (uint64, error) {
+	ctx, span := t.tracer.Start(ctx, "DA.GetLatestDAHeight")
+	defer span.End()
+
+	height, err := t.inner.GetLatestDAHeight(ctx)
+	if err != nil {
+		span.RecordError(err)
+		span.SetStatus(codes.Error, err.Error())
+		return 0, err
+	}
+	span.SetAttributes(attribute.Int64("da.latest_height", int64(height)))
+	return height, nil
+}
+
 func (t *tracedClient) GetHeaderNamespace() []byte { return t.inner.GetHeaderNamespace() }
 func (t *tracedClient) GetDataNamespace() []byte   { return t.inner.GetDataNamespace() }
 func (t *tracedClient) GetForcedInclusionNamespace() []byte {

block/internal/da/tracing_test.go

@@ -54,10 +54,11 @@ func (m *mockFullClient) Validate(ctx context.Context, ids []datypes.ID, proofs
 	}
 	return nil, nil
 }
-func (m *mockFullClient) GetHeaderNamespace() []byte          { return []byte{0x01} }
-func (m *mockFullClient) GetDataNamespace() []byte            { return []byte{0x02} }
-func (m *mockFullClient) GetForcedInclusionNamespace() []byte { return []byte{0x03} }
-func (m *mockFullClient) HasForcedInclusionNamespace() bool   { return true }
+func (m *mockFullClient) GetLatestDAHeight(_ context.Context) (uint64, error) { return 0, nil }
+func (m *mockFullClient) GetHeaderNamespace() []byte                          { return []byte{0x01} }
+func (m *mockFullClient) GetDataNamespace() []byte                            { return []byte{0x02} }
+func (m *mockFullClient) GetForcedInclusionNamespace() []byte                 { return []byte{0x03} }
+func (m *mockFullClient) HasForcedInclusionNamespace() bool                   { return true }
 
 // setup a tracer provider + span recorder
 func setupDATrace(t *testing.T, inner FullClient) (FullClient, *tracetest.SpanRecorder) {

pkg/sequencers/single/sequencer.go

@@ -21,6 +21,7 @@ import (
 	"github.com/evstack/ev-node/pkg/genesis"
 	seqcommon "github.com/evstack/ev-node/pkg/sequencers/common"
 	"github.com/evstack/ev-node/pkg/store"
+	"github.com/evstack/ev-node/types"
 )
 
 // ErrInvalidId is returned when the chain id is invalid
@@ -57,6 +58,11 @@ type Sequencer struct {
 	// inclusion transactions, no mempool) before resuming normal sequencing.
 	// This ensures the sequencer produces the same blocks that nodes running in
 	// base sequencing mode would have produced during the downtime.
+	//
+	// catchingUp is true when the sequencer is replaying missed DA epochs.
+	// It is set when we detect (via GetLatestDAHeight) that the DA layer is more
+	// than one epoch ahead of our checkpoint, and cleared when we hit
+	// ErrHeightFromFuture (meaning we've reached the DA head).
 	catchingUp bool
 	// currentDAEndTime is the DA epoch end timestamp from the last fetched epoch.
 	// Used as the block timestamp during catch-up to match based sequencing behavior.
@@ -421,14 +427,20 @@ func (c *Sequencer) IsCatchingUp() bool {
 }
 
 // fetchNextDAEpoch fetches transactions from the next DA epoch using checkpoint.
-// It also updates the catch-up state based on the DA epoch timestamp:
-//   - If the fetched epoch's timestamp is significantly in the past (more than
-//     one epoch's wall-clock duration), the sequencer enters catch-up mode.
+// It also updates the catch-up state based on DA heights:
+//   - Before the first fetch, it queries GetLatestDAHeight to determine if the
+//     sequencer has missed more than one DA epoch. If so, catch-up mode is
+//     entered and only forced-inclusion blocks (no mempool) are produced.
 //   - If the DA height is from the future (not yet produced), the sequencer
 //     exits catch-up mode as it has reached the DA head.
 func (c *Sequencer) fetchNextDAEpoch(ctx context.Context, maxBytes uint64) (uint64, error) {
 	currentDAHeight := c.checkpoint.DAHeight
 
+	// Determine catch-up state before the (potentially expensive) epoch fetch.
+	// This is done once per sequencer lifecycle — subsequent catch-up exits are
+	// handled by ErrHeightFromFuture below.
+	c.updateCatchUpState(ctx)
+
 	c.logger.Debug().
 		Uint64("da_height", currentDAHeight).
 		Uint64("tx_index", c.checkpoint.TxIndex).
@@ -466,11 +478,6 @@ func (c *Sequencer) fetchNextDAEpoch(ctx context.Context, maxBytes uint64) (uint
 		c.currentDAEndTime = forcedTxsEvent.Timestamp.UTC()
 	}
 
-	// Determine catch-up state based on epoch timestamp.
-	// If the epoch we just fetched ended more than one epoch's wall-clock duration ago,
-	// we are behind the DA head and must catch up by replaying missed epochs.
-	c.updateCatchUpState(forcedTxsEvent)
-
 	// Validate and filter transactions
 	validTxs := make([][]byte, 0, len(forcedTxsEvent.Txs))
 	skippedTxs := 0
@@ -501,60 +508,75 @@ func (c *Sequencer) fetchNextDAEpoch(ctx context.Context, maxBytes uint64) (uint
 	return forcedTxsEvent.EndDaHeight, nil
 }
 
-// updateCatchUpState determines whether the sequencer is catching up to the DA head.
+// updateCatchUpState determines whether the sequencer needs to catch up to the
+// DA head by comparing the sequencer's checkpoint DA height against the latest
+// DA height.
 //
-// The sequencer is considered to be catching up when the DA epoch it just fetched
-// has a timestamp that is significantly in the past — specifically, more than one
-// full epoch's wall-clock duration ago. This means other nodes likely switched to
-// base sequencing during the sequencer's downtime, and the sequencer must replay
-// those missed epochs before resuming normal block production.
+// The detection is purely height-based: we query GetLatestDAHeight once (on the
+// first epoch fetch) and calculate how many epochs the sequencer has missed. If
+// the gap exceeds one epoch, the sequencer enters catch-up mode and replays
+// missed epochs with forced-inclusion transactions only (no mempool). It remains
+// in catch-up until fetchNextDAEpoch hits ErrHeightFromFuture, meaning we've
+// reached the DA head.
 //
-// When the epoch timestamp is recent (within one epoch duration), the sequencer
-// has reached the DA head and can resume normal operation.
-func (c *Sequencer) updateCatchUpState(event *block.ForcedInclusionEvent) {
-	if event == nil || event.Timestamp.IsZero() {
-		// No timestamp available (e.g., empty epoch) — don't change catch-up state.
-		// If we were already catching up, we remain in that state until we see a
-		// recent timestamp or hit HeightFromFuture.
+// This check is performed only once per sequencer lifecycle. If the downtime was
+// short enough that the sequencer is still within the current or next epoch, no
+// catch-up is needed and the (lightweight) GetLatestDAHeight call is the only
+// overhead.
+func (c *Sequencer) updateCatchUpState(ctx context.Context) {
+	// Already catching up — nothing to do. We'll exit via ErrHeightFromFuture.
+	if c.catchingUp {
 		return
 	}
 
-	if c.genesis.DAEpochForcedInclusion == 0 {
+	epochSize := c.genesis.DAEpochForcedInclusion
+	if epochSize == 0 {
 		// No epoch-based forced inclusion configured — catch-up is irrelevant.
-		c.catchingUp = false
 		return
 	}
 
-	// Calculate how long one DA epoch takes in wall-clock time.
-	epochWallDuration := time.Duration(c.genesis.DAEpochForcedInclusion) * c.cfg.DA.BlockTime.Duration
+	currentDAHeight := c.checkpoint.DAHeight
+	daStartHeight := c.genesis.DAStartHeight
+
+	latestDAHeight, err := c.daClient.GetLatestDAHeight(ctx)
+	if err != nil {
+		c.logger.Warn().Err(err).
+			Msg("failed to get latest DA height for catch-up detection, skipping check")
+		return
+	}
 
-	// Use a minimum threshold to avoid false positives from minor delays.
-	catchUpThreshold := epochWallDuration
-	if catchUpThreshold < 30*time.Second {
-		catchUpThreshold = 30 * time.Second
+	if latestDAHeight <= currentDAHeight {
+		// DA hasn't moved beyond our position — nothing to catch up.
+		c.logger.Debug().
+			Uint64("checkpoint_da_height", currentDAHeight).
+			Uint64("latest_da_height", latestDAHeight).
+			Msg("sequencer is at or ahead of DA head, no catch-up needed")
+		return
 	}
 
-	timeSinceEpoch := time.Since(event.Timestamp)
-	wasCatchingUp := c.catchingUp
+	// Calculate epoch numbers for current position and DA head.
+	currentEpoch := types.CalculateEpochNumber(currentDAHeight, daStartHeight, epochSize)
+	latestEpoch := types.CalculateEpochNumber(latestDAHeight, daStartHeight, epochSize)
+	missedEpochs := latestEpoch - currentEpoch
 
-	if timeSinceEpoch > catchUpThreshold {
-		c.catchingUp = true
-		if !wasCatchingUp {
-			c.logger.Warn().
-				Dur("time_since_epoch", timeSinceEpoch).
-				Dur("threshold", catchUpThreshold).
-				Uint64("epoch_start", event.StartDaHeight).
-				Uint64("epoch_end", event.EndDaHeight).
-				Msg("entering catch-up mode: DA epoch is behind head, replaying missed epochs with forced inclusion txs only")
-		}
-	} else {
-		c.catchingUp = false
-		if wasCatchingUp {
-			c.logger.Info().
-				Dur("time_since_epoch", timeSinceEpoch).
-				Uint64("epoch_start", event.StartDaHeight).
-				Uint64("epoch_end", event.EndDaHeight).
-				Msg("exiting catch-up mode: reached DA head, resuming normal sequencing")
-		}
+	if missedEpochs <= 1 {
+		// Within the current or next epoch — normal operation, no catch-up.
+		c.logger.Debug().
+			Uint64("checkpoint_da_height", currentDAHeight).
+			Uint64("latest_da_height", latestDAHeight).
+			Uint64("current_epoch", currentEpoch).
+			Uint64("latest_epoch", latestEpoch).
+			Msg("sequencer within one epoch of DA head, no catch-up needed")
+		return
 	}
+
+	// The DA layer is more than one epoch ahead. Enter catch-up mode.
+	c.catchingUp = true
+	c.logger.Warn().
+		Uint64("checkpoint_da_height", currentDAHeight).
+		Uint64("latest_da_height", latestDAHeight).
+		Uint64("current_epoch", currentEpoch).
+		Uint64("latest_epoch", latestEpoch).
+		Uint64("missed_epochs", missedEpochs).
+		Msg("entering catch-up mode: DA layer is multiple epochs ahead, replaying missed epochs with forced inclusion txs only")
 }