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.

Author: alpe

pkg/da/jsonrpc/client.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+	"strings"
 
 	libshare "github.com/celestiaorg/go-square/v3/share"
 	"github.com/filecoin-project/go-jsonrpc"
@@ -23,6 +24,15 @@ func (c *Client) Close() {
 	}
 }
 
+// httpToWS converts an HTTP(S) URL to a WebSocket URL.
+// go-jsonrpc requires WebSocket for channel-based subscriptions (e.g. Subscribe).
+// WebSocket connections also support regular RPC calls, so this is backward-compatible.
+func httpToWS(addr string) string {
+	addr = strings.Replace(addr, "https://", "wss://", 1)
+	addr = strings.Replace(addr, "http://", "ws://", 1)
+	return addr
+}
+
 // NewClient connects to the celestia-node RPC endpoint
 func NewClient(ctx context.Context, addr, token string, authHeaderName string) (*Client, error) {
 	var httpHeader http.Header
@@ -33,16 +43,19 @@ func NewClient(ctx context.Context, addr, token string, authHeaderName string) (
 		httpHeader = http.Header{authHeaderName: []string{fmt.Sprintf("Bearer %s", token)}}
 	}
 
+	// Use WebSocket so that channel-based subscriptions (blob.Subscribe) work.
+	wsAddr := httpToWS(addr)
+
 	var cl Client
 
 	// Connect to the blob namespace
-	blobCloser, err := jsonrpc.NewClient(ctx, addr, "blob", &cl.Blob.Internal, httpHeader)
+	blobCloser, err := jsonrpc.NewClient(ctx, wsAddr, "blob", &cl.Blob.Internal, httpHeader)
 	if err != nil {
 		return nil, fmt.Errorf("failed to connect to blob namespace: %w", err)
 	}
 
 	// Connect to the header namespace
-	headerCloser, err := jsonrpc.NewClient(ctx, addr, "header", &cl.Header.Internal, httpHeader)
+	headerCloser, err := jsonrpc.NewClient(ctx, wsAddr, "header", &cl.Header.Internal, httpHeader)
 	if err != nil {
 		blobCloser()
 		return nil, fmt.Errorf("failed to connect to header namespace: %w", err)

test/e2e/evm_force_inclusion_e2e_test.go

@@ -238,8 +238,10 @@ func TestEvmFullNodeForceInclusionE2E(t *testing.T) {
 	// --- End Sequencer Setup ---
 
 	// --- Start Full Node Setup ---
+	// Get sequencer's full P2P address (including peer ID) for the full node to connect to
+	sequencerP2PAddress := getNodeP2PAddress(t, sut, sequencerHome, env.Endpoints.RollkitRPCPort)
 	// Reuse setupFullNode helper which handles genesis copying and node startup
-	setupFullNode(t, sut, fullNodeHome, sequencerHome, env.FullNodeJWT, env.GenesisHash, env.Endpoints.GetRollkitP2PAddress(), env.Endpoints)
+	setupFullNode(t, sut, fullNodeHome, sequencerHome, env.FullNodeJWT, env.GenesisHash, sequencerP2PAddress, env.Endpoints)
 	t.Log("Full node is up")
 	// --- End Full Node Setup ---
 

test/e2e/evm_test_common.go

@@ -449,15 +449,15 @@ func setupFullNode(t testing.TB, sut *SystemUnderTest, fullNodeHome, sequencerHo
 		"--home", fullNodeHome,
 		"--evm.jwt-secret-file", fullNodeJwtSecretFile,
 		"--evm.genesis-hash", genesisHash,
-		"--rollkit.p2p.peers", sequencerP2PAddress,
+		"--evnode.p2p.peers", sequencerP2PAddress,
 		"--evm.engine-url", endpoints.GetFullNodeEngineURL(),
 		"--evm.eth-url", endpoints.GetFullNodeEthURL(),
-		"--rollkit.da.block_time", DefaultDABlockTime,
-		"--rollkit.da.address", endpoints.GetDAAddress(),
-		"--rollkit.da.namespace", DefaultDANamespace,
-		"--rollkit.da.batching_strategy", "immediate",
-		"--rollkit.rpc.address", endpoints.GetFullNodeRPCListen(),
-		"--rollkit.p2p.listen_address", endpoints.GetFullNodeP2PAddress(),
+		"--evnode.da.block_time", DefaultDABlockTime,
+		"--evnode.da.address", endpoints.GetDAAddress(),
+		"--evnode.da.namespace", DefaultDANamespace,
+		"--evnode.da.batching_strategy", "immediate",
+		"--evnode.rpc.address", endpoints.GetFullNodeRPCListen(),
+		"--evnode.p2p.listen_address", endpoints.GetFullNodeP2PAddress(),
 	}
 	sut.ExecCmd(evmSingleBinaryPath, args...)
 	// Use AwaitNodeLive instead of AwaitNodeUp because in lazy mode scenarios,
@@ -932,4 +932,3 @@ func PrintTraceReport(t testing.TB, label string, spans []TraceSpan) {
 		t.Logf("%-40s %5.1f%% %s", name, pct, bar)
 	}
 }
-

tools/local-da/local.go

@@ -13,6 +13,7 @@ import (
 	"sync"
 	"time"
 
+	libshare "github.com/celestiaorg/go-square/v3/share"
 	"github.com/rs/zerolog"
 
 	blobrpc "github.com/evstack/ev-node/pkg/da/jsonrpc"
@@ -27,6 +28,18 @@ const (
 	DefaultBlockTime = 1 * time.Second
 )
 
+// subscriber holds a registered subscription's channel and namespace filter.
+type subscriber struct {
+	ch chan subscriptionEvent
+	ns libshare.Namespace
+}
+
+// subscriptionEvent is sent to subscribers when a new DA block is produced.
+type subscriptionEvent struct {
+	height uint64
+	blobs  []*blobrpc.Blob
+}
+
 // LocalDA is a simple implementation of in-memory DA. Not production ready! Intended only for testing!
 //
 // Data is stored in a map, where key is a serialized sequence number. This key is returned as ID.
@@ -43,6 +56,10 @@ type LocalDA struct {
 	blockTime   time.Duration
 	lastTime    time.Time // tracks last timestamp to ensure monotonicity
 
+	// Subscriber registry (protected by mu)
+	subscribers map[int]*subscriber
+	nextSubID   int
+
 	logger zerolog.Logger
 }
 
@@ -57,6 +74,7 @@ func NewLocalDA(logger zerolog.Logger, opts ...func(*LocalDA) *LocalDA) *LocalDA
 		data:        make(map[uint64][]kvp),
 		timestamps:  make(map[uint64]time.Time),
 		blobData:    make(map[uint64][]*blobrpc.Blob),
+		subscribers: make(map[int]*subscriber),
 		maxBlobSize: DefaultMaxBlobSize,
 		blockTime:   DefaultBlockTime,
 		lastTime:    time.Now(),
@@ -209,6 +227,7 @@ func (d *LocalDA) SubmitWithOptions(ctx context.Context, blobs []datypes.Blob, g
 
 		d.data[d.height] = append(d.data[d.height], kvp{ids[i], blob})
 	}
+	d.notifySubscribers(d.height)
 	d.logger.Info().Uint64("newHeight", d.height).Int("count", len(ids)).Msg("SubmitWithOptions successful")
 	return ids, nil
 }
@@ -239,6 +258,7 @@ func (d *LocalDA) Submit(ctx context.Context, blobs []datypes.Blob, gasPrice flo
 
 		d.data[d.height] = append(d.data[d.height], kvp{ids[i], blob})
 	}
+	d.notifySubscribers(d.height)
 	d.logger.Info().Uint64("newHeight", d.height).Int("count", len(ids)).Msg("Submit successful")
 	return ids, nil
 }
@@ -335,5 +355,68 @@ func (d *LocalDA) produceEmptyBlock() {
 	defer d.mu.Unlock()
 	d.height++
 	d.timestamps[d.height] = d.monotonicTime()
+	d.notifySubscribers(d.height)
 	d.logger.Debug().Uint64("height", d.height).Msg("produced empty block")
 }
+
+// subscribe registers a new subscriber for blobs matching the given namespace.
+// Returns a read-only channel and a subscription ID for later unsubscription.
+// Must NOT be called with d.mu held.
+func (d *LocalDA) subscribe(ns libshare.Namespace) (<-chan subscriptionEvent, int) {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	id := d.nextSubID
+	d.nextSubID++
+	ch := make(chan subscriptionEvent, 64)
+	d.subscribers[id] = &subscriber{ch: ch, ns: ns}
+	d.logger.Info().Int("subID", id).Str("namespace", hex.EncodeToString(ns.Bytes())).Msg("subscriber registered")
+	return ch, id
+}
+
+// unsubscribe removes a subscriber and closes its channel.
+// Must NOT be called with d.mu held.
+func (d *LocalDA) unsubscribe(id int) {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	if sub, ok := d.subscribers[id]; ok {
+		close(sub.ch)
+		delete(d.subscribers, id)
+		d.logger.Info().Int("subID", id).Msg("subscriber unregistered")
+	}
+}
+
+// notifySubscribers sends a subscriptionEvent to all registered subscribers.
+// For each subscriber, only blobs matching the subscriber's namespace are included.
+// Slow consumers (full channel) are dropped to avoid blocking block production.
+// MUST be called with d.mu held.
+func (d *LocalDA) notifySubscribers(height uint64) {
+	if len(d.subscribers) == 0 {
+		return
+	}
+
+	allBlobs := d.blobData[height] // may be nil for empty blocks
+
+	for id, sub := range d.subscribers {
+		// Filter blobs matching subscriber namespace
+		var matched []*blobrpc.Blob
+		for _, b := range allBlobs {
+			if b != nil && b.Namespace().Equals(sub.ns) {
+				matched = append(matched, b)
+			}
+		}
+
+		evt := subscriptionEvent{
+			height: height,
+			blobs:  matched,
+		}
+
+		select {
+		case sub.ch <- evt:
+		default:
+			// Slow consumer — drop to avoid blocking block production
+			d.logger.Warn().Int("subID", id).Uint64("height", height).Msg("dropping event for slow subscriber")
+		}
+	}
+}

tools/local-da/rpc.go

@@ -31,6 +31,7 @@ func (s *blobServer) Submit(_ context.Context, blobs []*jsonrpc.Blob, _ *jsonrpc
 
 	if len(blobs) == 0 {
 		s.da.timestamps[height] = time.Now()
+		s.da.notifySubscribers(height)
 		return height, nil
 	}
 
@@ -48,6 +49,7 @@ func (s *blobServer) Submit(_ context.Context, blobs []*jsonrpc.Blob, _ *jsonrpc
 		s.da.blobData[height] = append(s.da.blobData[height], b)
 	}
 	s.da.timestamps[height] = time.Now()
+	s.da.notifySubscribers(height)
 
 	return height, nil
 }
@@ -127,11 +129,39 @@ func (s *blobServer) GetCommitmentProof(_ context.Context, _ uint64, _ libshare.
 	return &jsonrpc.CommitmentProof{}, nil
 }
 
-// Subscribe returns a closed channel; LocalDA does not push live updates.
-func (s *blobServer) Subscribe(_ context.Context, _ libshare.Namespace) (<-chan *jsonrpc.SubscriptionResponse, error) {
-	ch := make(chan *jsonrpc.SubscriptionResponse)
-	close(ch)
-	return ch, nil
+// Subscribe streams blobs as they are included for the given namespace.
+// The returned channel emits a SubscriptionResponse for every new DA block.
+// The channel is closed when ctx is cancelled.
+func (s *blobServer) Subscribe(ctx context.Context, namespace libshare.Namespace) (<-chan *jsonrpc.SubscriptionResponse, error) {
+	eventCh, subID := s.da.subscribe(namespace)
+
+	out := make(chan *jsonrpc.SubscriptionResponse, 64)
+	go func() {
+		defer close(out)
+		defer s.da.unsubscribe(subID)
+
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case evt, ok := <-eventCh:
+				if !ok {
+					return
+				}
+				resp := &jsonrpc.SubscriptionResponse{
+					Height: evt.height,
+					Blobs:  evt.blobs,
+				}
+				select {
+				case out <- resp:
+				case <-ctx.Done():
+					return
+				}
+			}
+		}
+	}()
+
+	return out, nil
 }
 
 // startBlobServer starts an HTTP JSON-RPC server on addr serving the blob namespace.