feat(block): Event-Driven DA Follower with WebSocket Subscriptions (#3131)

* feat: Replace the Syncer's polling DA worker with an event-driven DAFollower and introduce DA client subscription.

* feat: add inline blob processing to DAFollower for zero-latency follow mode

When the DA subscription delivers blobs at the current local DA height,
the followLoop now processes them inline via ProcessBlobs — avoiding
a round-trip re-fetch from the DA layer.

Architecture:
- followLoop: processes subscription blobs inline when caught up (fast path),
  falls through to catchupLoop when behind (slow path).
- catchupLoop: unchanged — sequential RetrieveFromDA() for bulk sync.

Changes:
- Add Blobs field to SubscriptionEvent for carrying raw blob data
- Add extractBlobData() to DA client Subscribe adapter
- Export ProcessBlobs on DARetriever interface
- Add handleSubscriptionEvent() to DAFollower with inline fast path
- Add TestDAFollower_InlineProcessing with 3 sub-tests

* feat: subscribe to both header and data namespaces for inline processing

When header and data use different DA namespaces, the DAFollower now
subscribes to both and merges events via a fan-in goroutine. This ensures
inline blob processing works correctly for split-namespace configurations.

Changes:
- Add DataNamespace to DAFollowerConfig and daFollower
- Subscribe to both namespaces in runSubscription with mergeSubscriptions fan-in
- Guard handleSubscriptionEvent to only advance localDAHeight when
  ProcessBlobs returns at least one complete event (header+data matched)
- Pass DataNamespace from syncer.go
- Implement Subscribe on DummyDA test helper with subscriber notification

* feat: add subscription watchdog to detect stalled DA subscriptions

If no subscription events arrive within 3× the DA block time (default
30s), the watchdog triggers and returns an error. The followLoop then
reconnects the subscription with the standard backoff. This prevents
the node from silently stopping sync when the DA subscription stalls
(e.g., network partition, DA node freeze).

* fix: security hardening for DA subscription path

* feat: Implement blob subscription for local DA and update JSON-RPC client to use WebSockets, along with E2E test updates for new `evnode` flags and P2P address retrieval.

* WS client constructor

* Merge

* Linter

* Review feedback

* Review feedback

* Review feedbac

* Linter

Author: alpe

apps/evm/cmd/run.go

@@ -60,10 +60,11 @@ var RunCmd = &cobra.Command{
 			return err
 		}
 
-		blobClient, err := blobrpc.NewClient(context.Background(), nodeConfig.DA.Address, nodeConfig.DA.AuthToken, "")
+		blobClient, err := blobrpc.NewWSClient(cmd.Context(), nodeConfig.DA.Address, nodeConfig.DA.AuthToken, "")
 		if err != nil {
 			return fmt.Errorf("failed to create blob client: %w", err)
 		}
+		defer blobClient.Close()
 
 		daClient := block.NewDAClient(blobClient, nodeConfig, logger)
 

apps/evm/server/force_inclusion_test.go

@@ -50,6 +50,13 @@ 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) {
+	// Not needed in these tests; return a closed channel.
+	ch := make(chan da.SubscriptionEvent)
+	close(ch)
+	return ch, nil
+}
+
 func (m *mockDA) Validate(ctx context.Context, ids []da.ID, proofs []da.Proof, namespace []byte) ([]bool, error) {
 	return nil, nil
 }

apps/grpc/cmd/run.go

@@ -108,7 +108,7 @@ func createSequencer(
 	genesis genesis.Genesis,
 	executor execution.Executor,
 ) (coresequencer.Sequencer, error) {
-	blobClient, err := blobrpc.NewClient(ctx, nodeConfig.DA.Address, nodeConfig.DA.AuthToken, "")
+	blobClient, err := blobrpc.NewWSClient(ctx, nodeConfig.DA.Address, nodeConfig.DA.AuthToken, "")
 	if err != nil {
 		return nil, fmt.Errorf("failed to create blob client: %w", err)
 	}

apps/testapp/cmd/run.go

@@ -111,7 +111,7 @@ func createSequencer(
 	genesis genesis.Genesis,
 	executor execution.Executor,
 ) (coresequencer.Sequencer, error) {
-	blobClient, err := blobrpc.NewClient(ctx, nodeConfig.DA.Address, nodeConfig.DA.AuthToken, "")
+	blobClient, err := blobrpc.NewWSClient(ctx, nodeConfig.DA.Address, nodeConfig.DA.AuthToken, "")
 	if err != nil {
 		return nil, fmt.Errorf("failed to create blob client: %w", err)
 	}

block/components.go

@@ -114,7 +114,7 @@ func (bc *Components) Stop() error {
 		}
 	}
 	if bc.Syncer != nil {
-		if err := bc.Syncer.Stop(); err != nil {
+		if err := bc.Syncer.Stop(context.Background()); err != nil {
 			errs = errors.Join(errs, fmt.Errorf("failed to stop syncer: %w", err))
 		}
 	}

block/internal/da/client.go

@@ -350,6 +350,71 @@ func (c *client) HasForcedInclusionNamespace() bool {
 	return c.hasForcedNamespace
 }
 
+// Subscribe subscribes to blobs in the given namespace via the celestia-node
+// Subscribe API. It returns a channel that emits a SubscriptionEvent for every
+// DA block containing a matching blob. The channel is closed when ctx is
+// cancelled. The caller must drain the channel after cancellation to avoid
+// goroutine leaks.
+func (c *client) Subscribe(ctx context.Context, namespace []byte) (<-chan datypes.SubscriptionEvent, error) {
+	ns, err := share.NewNamespaceFromBytes(namespace)
+	if err != nil {
+		return nil, fmt.Errorf("invalid namespace: %w", err)
+	}
+
+	rawCh, err := c.blobAPI.Subscribe(ctx, ns)
+	if err != nil {
+		return nil, fmt.Errorf("blob subscribe: %w", err)
+	}
+
+	out := make(chan datypes.SubscriptionEvent, 16)
+	go func() {
+		defer close(out)
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case resp, ok := <-rawCh:
+				if !ok {
+					return
+				}
+				if resp == nil {
+					continue
+				}
+				select {
+				case out <- datypes.SubscriptionEvent{
+					Height: resp.Height,
+					Blobs:  extractBlobData(resp),
+				}:
+				case <-ctx.Done():
+					return
+				}
+			}
+		}
+	}()
+
+	return out, nil
+}
+
+// extractBlobData extracts raw byte slices from a subscription response,
+// filtering out nil blobs, empty data, and blobs exceeding DefaultMaxBlobSize.
+func extractBlobData(resp *blobrpc.SubscriptionResponse) [][]byte {
+	if resp == nil || len(resp.Blobs) == 0 {
+		return nil
+	}
+	blobs := make([][]byte, 0, len(resp.Blobs))
+	for _, blob := range resp.Blobs {
+		if blob == nil {
+			continue
+		}
+		data := blob.Data()
+		if len(data) == 0 || len(data) > common.DefaultMaxBlobSize {
+			continue
+		}
+		blobs = append(blobs, data)
+	}
+	return blobs
+}
+
 // Get fetches blobs by their IDs. Used for visualization and fetching specific blobs.
 func (c *client) Get(ctx context.Context, ids []datypes.ID, namespace []byte) ([]datypes.Blob, error) {
 	if len(ids) == 0 {

block/internal/da/interface.go

@@ -17,6 +17,11 @@ 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)
 
+	// Subscribe returns a channel that emits one SubscriptionEvent per DA block
+	// that contains a blob in the given namespace. The channel is closed when ctx
+	// is cancelled. Callers MUST drain the channel after cancellation.
+	Subscribe(ctx context.Context, namespace []byte) (<-chan datypes.SubscriptionEvent, error)
+
 	// GetLatestDAHeight returns the latest height available on the DA layer.
 	GetLatestDAHeight(ctx context.Context) (uint64, error)
 

block/internal/da/tracing.go

@@ -145,6 +145,9 @@ func (t *tracedClient) GetForcedInclusionNamespace() []byte {
 func (t *tracedClient) HasForcedInclusionNamespace() bool {
 	return t.inner.HasForcedInclusionNamespace()
 }
+func (t *tracedClient) Subscribe(ctx context.Context, namespace []byte) (<-chan datypes.SubscriptionEvent, error) {
+	return t.inner.Subscribe(ctx, namespace)
+}
 
 type submitError struct{ msg string }
 

block/internal/da/tracing_test.go

@@ -22,6 +22,14 @@ type mockFullClient struct {
 	getFn       func(ctx context.Context, ids []datypes.ID, namespace []byte) ([]datypes.Blob, error)
 	getProofsFn func(ctx context.Context, ids []datypes.ID, namespace []byte) ([]datypes.Proof, error)
 	validateFn  func(ctx context.Context, ids []datypes.ID, proofs []datypes.Proof, namespace []byte) ([]bool, error)
+	subscribeFn func(ctx context.Context, namespace []byte) (<-chan datypes.SubscriptionEvent, error)
+}
+
+func (m *mockFullClient) Subscribe(ctx context.Context, namespace []byte) (<-chan datypes.SubscriptionEvent, error) {
+	if m.subscribeFn == nil {
+		panic("not expected to be called")
+	}
+	return m.subscribeFn(ctx, namespace)
 }
 
 func (m *mockFullClient) Submit(ctx context.Context, data [][]byte, gasPrice float64, namespace []byte, options []byte) datypes.ResultSubmit {