feat: Subscribe to forced inclusion ns events (#3146)

* Use subscription with forced inclusion; refactorings

* Fix compile errors in tests after rebase

* Include timestamp for subscription events

* Doc update

(cherry picked from commit 7a2adbd)

* Apply review feedback

* Updates - WIP

* Updates

* No fetch when in cache

* Updates

* Review feedback

* Review feedback

* Minor fixes

* Minor udpates for tests

* Fix tests

* Fix test

* Review feedback

* Handle edge case in catch up

* Review feedback

* Review feedback

* Linter

* Stop fi retriever in tests

* Update changelog

Author: alpe

CHANGELOG.md

@@ -14,6 +14,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Fix race on startup sync. [#3162](https://github.com/evstack/ev-node/pull/3162)
 - Strict raft state. [#3167](https://github.com/evstack/ev-node/pull/3167)
 
+### Changes
+
+- Subscribe to forced inclusion namespace events [#3146](https://github.com/evstack/ev-node/pull/3146)
+
 ## v1.0.0
 
 ### Fixed

apps/evm/cmd/run.go

@@ -23,7 +23,6 @@ import (
 	"github.com/evstack/ev-node/pkg/config"
 	blobrpc "github.com/evstack/ev-node/pkg/da/jsonrpc"
 	da "github.com/evstack/ev-node/pkg/da/types"
-	"github.com/evstack/ev-node/pkg/genesis"
 	genesispkg "github.com/evstack/ev-node/pkg/genesis"
 	"github.com/evstack/ev-node/pkg/p2p/key"
 	"github.com/evstack/ev-node/pkg/sequencers/based"
@@ -84,7 +83,7 @@ var RunCmd = &cobra.Command{
 		}
 
 		// Create sequencer based on configuration
-		sequencer, err := createSequencer(logger, datastore, nodeConfig, genesis, daClient, executor)
+		sequencer, err := createSequencer(cmd.Context(), logger, datastore, nodeConfig, genesis, daClient, executor)
 		if err != nil {
 			return err
 		}
@@ -145,10 +144,11 @@ func init() {
 // If BasedSequencer is enabled, it creates a based sequencer that fetches transactions from DA.
 // Otherwise, it creates a single (traditional) sequencer.
 func createSequencer(
+	ctx context.Context,
 	logger zerolog.Logger,
 	datastore datastore.Batching,
 	nodeConfig config.Config,
-	genesis genesis.Genesis,
+	genesis genesispkg.Genesis,
 	daClient block.FullDAClient,
 	executor execution.Executor,
 ) (coresequencer.Sequencer, error) {
@@ -158,7 +158,7 @@ func createSequencer(
 			return nil, fmt.Errorf("based sequencer mode requires aggregator mode to be enabled")
 		}
 
-		basedSeq, err := based.NewBasedSequencer(daClient, nodeConfig, datastore, genesis, logger, executor)
+		basedSeq, err := based.NewBasedSequencer(ctx, daClient, nodeConfig, datastore, genesis, logger, executor)
 		if err != nil {
 			return nil, fmt.Errorf("failed to create based sequencer: %w", err)
 		}

apps/evm/server/force_inclusion_test.go

@@ -50,7 +50,7 @@ func (m *mockDA) Get(ctx context.Context, ids []da.ID, namespace []byte) ([]da.B
 	return nil, nil
 }
 
-func (m *mockDA) Subscribe(_ context.Context, _ []byte) (<-chan da.SubscriptionEvent, error) {
+func (m *mockDA) Subscribe(_ context.Context, _ []byte, _ bool) (<-chan da.SubscriptionEvent, error) {
 	// Not needed in these tests; return a closed channel.
 	ch := make(chan da.SubscriptionEvent)
 	close(ch)

apps/grpc/cmd/run.go

@@ -121,7 +121,7 @@ func createSequencer(
 			return nil, fmt.Errorf("based sequencer mode requires aggregator mode to be enabled")
 		}
 
-		basedSeq, err := based.NewBasedSequencer(daClient, nodeConfig, datastore, genesis, logger, executor)
+		basedSeq, err := based.NewBasedSequencer(ctx, daClient, nodeConfig, datastore, genesis, logger, executor)
 		if err != nil {
 			return nil, fmt.Errorf("failed to create based sequencer: %w", err)
 		}

apps/testapp/cmd/run.go

@@ -124,7 +124,7 @@ func createSequencer(
 			return nil, fmt.Errorf("based sequencer mode requires aggregator mode to be enabled")
 		}
 
-		basedSeq, err := based.NewBasedSequencer(daClient, nodeConfig, datastore, genesis, logger, executor)
+		basedSeq, err := based.NewBasedSequencer(ctx, daClient, nodeConfig, datastore, genesis, logger, executor)
 		if err != nil {
 			return nil, fmt.Errorf("failed to create based sequencer: %w", err)
 		}

block/internal/common/event.go

@@ -1,6 +1,10 @@
 package common
 
-import "github.com/evstack/ev-node/types"
+import (
+	"context"
+
+	"github.com/evstack/ev-node/types"
+)
 
 // EventSource represents the origin of a block event
 type EventSource string
@@ -24,3 +28,18 @@ type DAHeightEvent struct {
 	// Optional DA height hints from P2P. first is the DA height hint for the header, second is the DA height hint for the data
 	DaHeightHints [2]uint64
 }
+
+// EventSink receives parsed DA events with backpressure support.
+type EventSink interface {
+	PipeEvent(ctx context.Context, event DAHeightEvent) error
+}
+
+// EventSinkFunc adapts a plain function to the EventSink interface.
+// Useful in tests:
+//
+//	sink := common.EventSinkFunc(func(ctx context.Context, ev common.DAHeightEvent) error { return nil })
+type EventSinkFunc func(ctx context.Context, event DAHeightEvent) error
+
+func (f EventSinkFunc) PipeEvent(ctx context.Context, event DAHeightEvent) error {
+	return f(ctx, event)
+}