feat(syncing): add grace period for missing force txs inclusion

Author: julienrbrt

block/internal/common/metrics.go

@@ -65,6 +65,10 @@ type Metrics struct {
 	DAInclusionHeight       metrics.Gauge
 	PendingHeadersCount     metrics.Gauge
 	PendingDataCount        metrics.Gauge
+
+	// Forced inclusion metrics
+	ForcedInclusionTxsInGracePeriod metrics.Gauge   // Number of forced inclusion txs currently in grace period
+	ForcedInclusionTxsMalicious     metrics.Counter // Total number of forced inclusion txs marked as malicious
 }
 
 // PrometheusMetrics returns Metrics built using Prometheus client library
@@ -182,6 +186,21 @@ func PrometheusMetrics(namespace string, labelsAndValues ...string) *Metrics {
 		Help:      "Number of data blocks pending DA submission",
 	}, labels).With(labelsAndValues...)
 
+	// Forced inclusion metrics
+	m.ForcedInclusionTxsInGracePeriod = prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
+		Namespace: namespace,
+		Subsystem: MetricsSubsystem,
+		Name:      "forced_inclusion_txs_in_grace_period",
+		Help:      "Number of forced inclusion transactions currently in grace period (past epoch end but within grace boundary)",
+	}, labels).With(labelsAndValues...)
+
+	m.ForcedInclusionTxsMalicious = prometheus.NewCounterFrom(stdprometheus.CounterOpts{
+		Namespace: namespace,
+		Subsystem: MetricsSubsystem,
+		Name:      "forced_inclusion_txs_malicious_total",
+		Help:      "Total number of forced inclusion transactions marked as malicious (past grace boundary)",
+	}, labels).With(labelsAndValues...)
+
 	// DA Submitter metrics
 	m.DASubmitterPendingBlobs = prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
 		Namespace: namespace,
@@ -246,6 +265,10 @@ func NopMetrics() *Metrics {
 		DASubmitterLastFailure:  make(map[DASubmitterFailureReason]metrics.Gauge),
 		DASubmitterPendingBlobs: discard.NewGauge(),
 		DASubmitterResends:      discard.NewCounter(),
+
+		// Forced inclusion metrics
+		ForcedInclusionTxsInGracePeriod: discard.NewGauge(),
+		ForcedInclusionTxsMalicious:     discard.NewCounter(),
 	}
 
 	// Initialize maps with no-op metrics

block/internal/syncing/syncer.go

@@ -741,16 +741,44 @@ func (s *Syncer) verifyForcedInclusionTxs(currentState types.State, data *types.
 	}
 
 	// Check if we've moved past any epoch boundaries with pending txs
+	// Grace period: Allow forced inclusion txs from epoch N to be included in epoch N+1, N+2, etc.
+	// Only flag as malicious if past grace boundary to prevent false positives during DA unavailability.
 	var maliciousTxs, remainingPending []pendingForcedInclusionTx
+	var txsInGracePeriod int
 	for _, pending := range stillPending {
-		// If current DA height is past this epoch's end, these txs should have been included
-		if currentState.DAHeight > pending.EpochEnd {
+		// Calculate grace boundary: epoch end + (grace periods × epoch size)
+		graceBoundary := pending.EpochEnd + (s.genesis.ForcedInclusionGracePeriod * s.genesis.DAEpochForcedInclusion)
+
+		// If current DA height is past the grace boundary, these txs should have been included
+		if currentState.DAHeight > graceBoundary {
 			maliciousTxs = append(maliciousTxs, pending)
+			s.logger.Warn().
+				Uint64("current_da_height", currentState.DAHeight).
+				Uint64("epoch_end", pending.EpochEnd).
+				Uint64("grace_boundary", graceBoundary).
+				Uint64("grace_periods", s.genesis.ForcedInclusionGracePeriod).
+				Str("tx_hash", pending.TxHash[:16]).
+				Msg("forced inclusion transaction past grace boundary - marking as malicious")
 		} else {
 			remainingPending = append(remainingPending, pending)
+
+			// Track if we're in the grace period (past epoch end but within grace boundary)
+			if currentState.DAHeight > pending.EpochEnd {
+				txsInGracePeriod++
+				s.logger.Info().
+					Uint64("current_da_height", currentState.DAHeight).
+					Uint64("epoch_end", pending.EpochEnd).
+					Uint64("grace_boundary", graceBoundary).
+					Uint64("grace_periods", s.genesis.ForcedInclusionGracePeriod).
+					Str("tx_hash", pending.TxHash[:16]).
+					Msg("forced inclusion transaction in grace period - not yet malicious")
+			}
 		}
 	}
 
+	// Update metrics for grace period tracking
+	s.metrics.ForcedInclusionTxsInGracePeriod.Set(float64(txsInGracePeriod))
+
 	// Update pending map - clear old entries and store only remaining pending
 	s.pendingForcedInclusionTxs.Range(func(key, value any) bool {
 		s.pendingForcedInclusionTxs.Delete(key)
@@ -760,14 +788,18 @@ func (s *Syncer) verifyForcedInclusionTxs(currentState types.State, data *types.
 		s.pendingForcedInclusionTxs.Store(pending.TxHash, pending)
 	}
 
-	// If there are transactions from past epochs that weren't included, sequencer is malicious
+	// If there are transactions past grace boundary that weren't included, sequencer is malicious
 	if len(maliciousTxs) > 0 {
+		// Update metrics for malicious detection
+		s.metrics.ForcedInclusionTxsMalicious.Add(float64(len(maliciousTxs)))
+
 		s.logger.Error().
 			Uint64("height", data.Height()).
 			Uint64("current_da_height", currentState.DAHeight).
 			Int("malicious_count", len(maliciousTxs)).
-			Msg("SEQUENCER IS MALICIOUS: forced inclusion transactions from past epoch(s) not included")
-		return errors.Join(errMaliciousProposer, fmt.Errorf("sequencer is malicious: %d forced inclusion transactions from past epoch(s) not included", len(maliciousTxs)))
+			Uint64("grace_periods", s.genesis.ForcedInclusionGracePeriod).
+			Msg("SEQUENCER IS MALICIOUS: forced inclusion transactions past grace boundary not included")
+		return errors.Join(errMaliciousProposer, fmt.Errorf("sequencer is malicious: %d forced inclusion transactions past grace boundary (grace_periods=%d) not included", len(maliciousTxs), s.genesis.ForcedInclusionGracePeriod))
 	}
 
 	// Log current state

block/internal/syncing/syncer_forced_inclusion_test.go

@@ -206,7 +206,7 @@ func TestVerifyForcedInclusionTxs_MissingTransactions(t *testing.T) {
 	err = s.verifyForcedInclusionTxs(currentState, data2)
 	require.Error(t, err)
 	require.Contains(t, err.Error(), "sequencer is malicious")
-	require.Contains(t, err.Error(), "forced inclusion transactions from past epoch(s) not included")
+	require.Contains(t, err.Error(), "past grace boundary")
 }
 
 func TestVerifyForcedInclusionTxs_PartiallyIncluded(t *testing.T) {
@@ -309,7 +309,7 @@ func TestVerifyForcedInclusionTxs_PartiallyIncluded(t *testing.T) {
 	err = s.verifyForcedInclusionTxs(currentState, data2)
 	require.Error(t, err)
 	require.Contains(t, err.Error(), "sequencer is malicious")
-	require.Contains(t, err.Error(), "forced inclusion transactions from past epoch(s) not included")
+	require.Contains(t, err.Error(), "past grace boundary")
 }
 
 func TestVerifyForcedInclusionTxs_NoForcedTransactions(t *testing.T) {
@@ -759,5 +759,137 @@ func TestVerifyForcedInclusionTxs_MaliciousAfterEpochEnd(t *testing.T) {
 	err = s.verifyForcedInclusionTxs(currentState, data3)
 	require.Error(t, err)
 	require.Contains(t, err.Error(), "sequencer is malicious")
-	require.Contains(t, err.Error(), "forced inclusion transactions from past epoch(s) not included")
+	require.Contains(t, err.Error(), "past grace boundary")
+}
+
+// TestVerifyForcedInclusionTxs_SmoothingExceedsEpoch tests the critical scenario where
+// forced inclusion transactions cannot all be included before an epoch ends.
+// This demonstrates that the system correctly detects malicious behavior when
+// transactions remain pending after the epoch boundary.
+func TestVerifyForcedInclusionTxs_SmoothingExceedsEpoch(t *testing.T) {
+	ds := dssync.MutexWrap(datastore.NewMapDatastore())
+	st := store.New(ds)
+
+	cm, err := cache.NewCacheManager(config.DefaultConfig(), zerolog.Nop())
+	require.NoError(t, err)
+
+	addr, pub, signer := buildSyncTestSigner(t)
+	gen := genesis.Genesis{
+		ChainID:                "tchain",
+		InitialHeight:          1,
+		StartTime:              time.Now().Add(-time.Second),
+		ProposerAddress:        addr,
+		DAStartHeight:          100,
+		DAEpochForcedInclusion: 3, // Epoch: [100, 102]
+	}
+
+	cfg := config.DefaultConfig()
+	cfg.DA.ForcedInclusionNamespace = "nsForcedInclusion"
+
+	mockExec := testmocks.NewMockExecutor(t)
+	mockExec.EXPECT().InitChain(mock.Anything, mock.Anything, uint64(1), "tchain").
+		Return([]byte("app0"), uint64(1024), nil).Once()
+
+	mockDA := testmocks.NewMockDA(t)
+
+	daClient := da.NewClient(da.Config{
+		DA:                       mockDA,
+		Logger:                   zerolog.Nop(),
+		Namespace:                cfg.DA.Namespace,
+		DataNamespace:            cfg.DA.DataNamespace,
+		ForcedInclusionNamespace: cfg.DA.ForcedInclusionNamespace,
+	})
+	daRetriever := NewDARetriever(daClient, cm, gen, zerolog.Nop())
+	fiRetriever := da.NewForcedInclusionRetriever(daClient, gen, zerolog.Nop())
+
+	s := NewSyncer(
+		st,
+		mockExec,
+		daClient,
+		cm,
+		common.NopMetrics(),
+		cfg,
+		gen,
+		common.NewMockBroadcaster[*types.SignedHeader](t),
+		common.NewMockBroadcaster[*types.Data](t),
+		zerolog.Nop(),
+		common.DefaultBlockOptions(),
+		make(chan error, 1),
+	)
+	s.daRetriever = daRetriever
+	s.fiRetriever = fiRetriever
+
+	require.NoError(t, s.initializeState())
+	s.ctx = context.Background()
+
+	namespaceForcedInclusionBz := coreda.NamespaceFromString(cfg.DA.GetForcedInclusionNamespace()).Bytes()
+
+	// Create 3 forced inclusion transactions
+	dataBin1, _ := makeSignedDataBytes(t, gen.ChainID, 10, addr, pub, signer, 2)
+	dataBin2, _ := makeSignedDataBytes(t, gen.ChainID, 11, addr, pub, signer, 2)
+	dataBin3, _ := makeSignedDataBytes(t, gen.ChainID, 12, addr, pub, signer, 2)
+
+	// Mock DA retrieval for Epoch 1: [100, 102]
+	mockDA.EXPECT().GetIDs(mock.Anything, uint64(100), mock.MatchedBy(func(ns []byte) bool {
+		return bytes.Equal(ns, namespaceForcedInclusionBz)
+	})).Return(&coreda.GetIDsResult{
+		IDs:       [][]byte{[]byte("fi1"), []byte("fi2"), []byte("fi3")},
+		Timestamp: time.Now(),
+	}, nil).Once()
+
+	mockDA.EXPECT().Get(mock.Anything, mock.Anything, mock.MatchedBy(func(ns []byte) bool {
+		return bytes.Equal(ns, namespaceForcedInclusionBz)
+	})).Return([][]byte{dataBin1, dataBin2, dataBin3}, nil).Once()
+
+	for height := uint64(101); height <= 102; height++ {
+		mockDA.EXPECT().GetIDs(mock.Anything, height, mock.MatchedBy(func(ns []byte) bool {
+			return bytes.Equal(ns, namespaceForcedInclusionBz)
+		})).Return(&coreda.GetIDsResult{IDs: [][]byte{}, Timestamp: time.Now()}, nil).Once()
+	}
+
+	// Block at DA height 102 (epoch end): Only includes 2 of 3 txs
+	// The third tx remains pending - legitimate within the epoch
+	data1 := makeData(gen.ChainID, 1, 2)
+	data1.Txs[0] = types.Tx(dataBin1)
+	data1.Txs[1] = types.Tx(dataBin2)
+
+	currentState := s.GetLastState()
+	currentState.DAHeight = 102 // At epoch end
+
+	err = s.verifyForcedInclusionTxs(currentState, data1)
+	require.NoError(t, err, "smoothing within epoch should be allowed")
+
+	// Verify 1 tx still pending
+	pendingCount := 0
+	s.pendingForcedInclusionTxs.Range(func(key, value any) bool {
+		pendingCount++
+		return true
+	})
+	require.Equal(t, 1, pendingCount, "should have 1 pending forced inclusion tx")
+
+	// === CRITICAL TEST: Move to next epoch WITHOUT including the pending tx ===
+	// Mock DA for next epoch [103, 105] with no forced txs
+	mockDA.EXPECT().GetIDs(mock.Anything, uint64(103), mock.MatchedBy(func(ns []byte) bool {
+		return bytes.Equal(ns, namespaceForcedInclusionBz)
+	})).Return(&coreda.GetIDsResult{IDs: [][]byte{}, Timestamp: time.Now()}, nil).Once()
+
+	mockDA.EXPECT().GetIDs(mock.Anything, uint64(104), mock.MatchedBy(func(ns []byte) bool {
+		return bytes.Equal(ns, namespaceForcedInclusionBz)
+	})).Return(&coreda.GetIDsResult{IDs: [][]byte{}, Timestamp: time.Now()}, nil).Once()
+
+	mockDA.EXPECT().GetIDs(mock.Anything, uint64(105), mock.MatchedBy(func(ns []byte) bool {
+		return bytes.Equal(ns, namespaceForcedInclusionBz)
+	})).Return(&coreda.GetIDsResult{IDs: [][]byte{}, Timestamp: time.Now()}, nil).Once()
+
+	// Block at DA height 105 (next epoch end): Doesn't include the pending tx
+	data2 := makeData(gen.ChainID, 2, 1)
+	data2.Txs[0] = types.Tx([]byte("regular_tx_only"))
+
+	currentState.DAHeight = 105 // Past previous epoch boundary [100, 102]
+
+	// Should FAIL - forced tx from previous epoch wasn't included before epoch ended
+	err = s.verifyForcedInclusionTxs(currentState, data2)
+	require.Error(t, err, "should detect malicious sequencer when forced tx exceeds epoch")
+	require.Contains(t, err.Error(), "sequencer is malicious")
+	require.Contains(t, err.Error(), "past grace boundary")
 }