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

Author: alpe

block/internal/syncing/da_follower.go

@@ -1,8 +1,10 @@
 package syncing
 
 import (
+	"bytes"
 	"context"
 	"errors"
+	"fmt"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -45,6 +47,9 @@ type daFollower struct {
 
 	// Namespace to subscribe on (header namespace).
 	namespace []byte
+	// dataNamespace is the data namespace (may equal namespace when header+data
+	// share the same namespace). When different, we subscribe to both and merge.
+	dataNamespace []byte
 
 	// localDAHeight is only written by catchupLoop and read by followLoop
 	// to determine whether a catchup is needed.
@@ -76,18 +81,24 @@ type DAFollowerConfig struct {
 	Logger        zerolog.Logger
 	PipeEvent     func(ctx context.Context, event common.DAHeightEvent) error
 	Namespace     []byte
+	DataNamespace []byte // may be nil or equal to Namespace
 	StartDAHeight uint64
 	DABlockTime   time.Duration
 }
 
 // NewDAFollower creates a new daFollower.
 func NewDAFollower(cfg DAFollowerConfig) DAFollower {
+	dataNs := cfg.DataNamespace
+	if len(dataNs) == 0 {
+		dataNs = cfg.Namespace
+	}
 	f := &daFollower{
 		client:        cfg.Client,
 		retriever:     cfg.Retriever,
 		logger:        cfg.Logger.With().Str("component", "da_follower").Logger(),
 		pipeEvent:     cfg.PipeEvent,
 		namespace:     cfg.Namespace,
+		dataNamespace: dataNs,
 		catchupSignal: make(chan struct{}, 1),
 		daBlockTime:   cfg.DABlockTime,
 	}
@@ -154,12 +165,22 @@ func (f *daFollower) followLoop() {
 	}
 }
 
-// runSubscription opens a single subscription and processes events until the
-// channel is closed or an error occurs.
+// runSubscription opens subscriptions on both header and data namespaces (if
+// different) and processes events until a channel is closed or an error occurs.
 func (f *daFollower) runSubscription() error {
-	ch, err := f.client.Subscribe(f.ctx, f.namespace)
+	headerCh, err := f.client.Subscribe(f.ctx, f.namespace)
 	if err != nil {
-		return err
+		return fmt.Errorf("subscribe header namespace: %w", err)
+	}
+
+	// If namespaces differ, subscribe to the data namespace too and fan-in.
+	ch := headerCh
+	if !bytes.Equal(f.namespace, f.dataNamespace) {
+		dataCh, err := f.client.Subscribe(f.ctx, f.dataNamespace)
+		if err != nil {
+			return fmt.Errorf("subscribe data namespace: %w", err)
+		}
+		ch = f.mergeSubscriptions(headerCh, dataCh)
 	}
 
 	for {
@@ -175,6 +196,41 @@ func (f *daFollower) runSubscription() error {
 	}
 }
 
+// mergeSubscriptions fans two subscription channels into one, concatenating
+// blobs from both namespaces when events arrive at the same DA height.
+func (f *daFollower) mergeSubscriptions(
+	headerCh, dataCh <-chan datypes.SubscriptionEvent,
+) <-chan datypes.SubscriptionEvent {
+	out := make(chan datypes.SubscriptionEvent, 16)
+	go func() {
+		defer close(out)
+		for headerCh != nil || dataCh != nil {
+			var ev datypes.SubscriptionEvent
+			var ok bool
+			select {
+			case <-f.ctx.Done():
+				return
+			case ev, ok = <-headerCh:
+				if !ok {
+					headerCh = nil
+					continue
+				}
+			case ev, ok = <-dataCh:
+				if !ok {
+					dataCh = nil
+					continue
+				}
+			}
+			select {
+			case out <- ev:
+			case <-f.ctx.Done():
+				return
+			}
+		}
+	}()
+	return out
+}
+
 // handleSubscriptionEvent processes a subscription event. When the follower is
 // caught up (ev.Height == localDAHeight) and blobs are available, it processes
 // them inline — avoiding a DA re-fetch round trip. Otherwise, it just updates
@@ -194,11 +250,15 @@ func (f *daFollower) handleSubscriptionEvent(ev datypes.SubscriptionEvent) {
 				return // catchupLoop already signaled via updateHighest
 			}
 		}
-		// Advance local height — we processed this height inline.
-		f.localDAHeight.Store(ev.Height + 1)
-		f.headReached.Store(true)
-		f.logger.Debug().Uint64("da_height", ev.Height).Int("events", len(events)).
-			Msg("processed subscription blobs inline (fast path)")
+		// Only advance if we produced at least one complete event.
+		// With split namespaces, the first namespace's blobs may not produce
+		// events until the second namespace's blobs arrive at the same height.
+		if len(events) != 0 {
+			f.localDAHeight.Store(ev.Height + 1)
+			f.headReached.Store(true)
+			f.logger.Debug().Uint64("da_height", ev.Height).Int("events", len(events)).
+				Msg("processed subscription blobs inline (fast path)")
+		}
 		return
 	}
 

block/internal/syncing/syncer.go

@@ -203,6 +203,7 @@ func (s *Syncer) Start(ctx context.Context) error {
 		Logger:        s.logger,
 		PipeEvent:     s.pipeEvent,
 		Namespace:     s.daClient.GetHeaderNamespace(),
+		DataNamespace: s.daClient.GetDataNamespace(),
 		StartDAHeight: s.daRetrieverHeight.Load(),
 		DABlockTime:   s.config.DA.BlockTime.Duration,
 	})

test/testda/dummy.go

@@ -28,6 +28,12 @@ func (h *Header) Time() time.Time {
 	return h.Timestamp
 }
 
+// subscriber holds a channel and the context for a single Subscribe caller.
+type subscriber struct {
+	ch  chan datypes.SubscriptionEvent
+	ctx context.Context
+}
+
 // DummyDA is a test implementation of the DA client interface.
 // It supports blob storage, height simulation, failure injection, and header retrieval.
 type DummyDA struct {
@@ -38,13 +44,50 @@ type DummyDA struct {
 	headers    map[uint64]*Header             // height -> header (with timestamp)
 	failSubmit atomic.Bool
 
+	// subscribers tracks active Subscribe callers.
+	subscribers []*subscriber
+
 	tickerMu   sync.Mutex
 	tickerStop chan struct{}
 }
 
-func (d *DummyDA) Subscribe(ctx context.Context, namespace []byte) (<-chan datypes.SubscriptionEvent, error) {
-	//TODO implement me
-	panic("implement me")
+// Subscribe returns a channel that emits a SubscriptionEvent for every new DA
+// height produced by Submit or StartHeightTicker. The channel is closed when
+// ctx is cancelled or Reset is called.
+func (d *DummyDA) Subscribe(ctx context.Context, _ []byte) (<-chan datypes.SubscriptionEvent, error) {
+	ch := make(chan datypes.SubscriptionEvent, 64)
+	sub := &subscriber{ch: ch, ctx: ctx}
+
+	d.mu.Lock()
+	d.subscribers = append(d.subscribers, sub)
+	d.mu.Unlock()
+
+	// Remove subscriber and close channel when ctx is cancelled.
+	go func() {
+		<-ctx.Done()
+		d.mu.Lock()
+		defer d.mu.Unlock()
+		for i, s := range d.subscribers {
+			if s == sub {
+				d.subscribers = append(d.subscribers[:i], d.subscribers[i+1:]...)
+				break
+			}
+		}
+		close(ch)
+	}()
+
+	return ch, nil
+}
+
+// notifySubscribers sends an event to all active subscribers. Must be called
+// with d.mu held.
+func (d *DummyDA) notifySubscribers(ev datypes.SubscriptionEvent) {
+	for _, sub := range d.subscribers {
+		select {
+		case sub.ch <- ev:
+		case <-sub.ctx.Done():
+		}
+	}
 }
 
 // Option configures a DummyDA instance.
@@ -116,6 +159,10 @@ func (d *DummyDA) Submit(_ context.Context, data [][]byte, _ float64, namespace
 			Timestamp: now,
 		}
 	}
+	d.notifySubscribers(datypes.SubscriptionEvent{
+		Height: height,
+		Blobs:  data,
+	})
 	d.mu.Unlock()
 
 	return datypes.ResultSubmit{
@@ -258,6 +305,9 @@ func (d *DummyDA) StartHeightTicker(interval time.Duration) func() {
 						Timestamp: now,
 					}
 				}
+				d.notifySubscribers(datypes.SubscriptionEvent{
+					Height: height,
+				})
 				d.mu.Unlock()
 			case <-stopCh:
 				return
@@ -282,6 +332,11 @@ func (d *DummyDA) Reset() {
 	d.blobs = make(map[uint64]map[string][][]byte)
 	d.headers = make(map[uint64]*Header)
 	d.failSubmit.Store(false)
+	// Close all subscriber channels.
+	for _, sub := range d.subscribers {
+		close(sub.ch)
+	}
+	d.subscribers = nil
 	d.mu.Unlock()
 
 	d.tickerMu.Lock()