chore: connect visualiser (#2705)

<!--
Please read and fill out this form before submitting your PR.

Please make sure you have reviewed our contributors guide before
submitting your
first PR.

NOTE: PR titles should follow semantic commits:
https://www.conventionalcommits.org/en/v1.0.0/
-->

## Overview

<!-- 
Please provide an explanation of the PR, including the appropriate
context,
background, goal, and rationale. If there is an issue with this
information,
please provide a tl;dr and link the issue. 

Ex: Closes #<issue number>
-->

Author: tac0turtle

block/internal/submitting/da_submitter.go

@@ -134,13 +134,19 @@ func NewDASubmitter(
 	options common.BlockOptions,
 	logger zerolog.Logger,
 ) *DASubmitter {
+	daSubmitterLogger := logger.With().Str("component", "da_submitter").Logger()
+
+	if config.RPC.EnableDAVisualization {
+		visualizerLogger := logger.With().Str("component", "da_visualization").Logger()
+		server.SetDAVisualizationServer(server.NewDAVisualizationServer(da, visualizerLogger, config.Node.Aggregator))
+	}
 
 	return &DASubmitter{
 		da:              da,
 		config:          config,
 		genesis:         genesis,
 		options:         options,
-		logger:          logger.With().Str("component", "da_submitter").Logger(),
+		logger:          daSubmitterLogger,
 		namespaceBz:     coreda.NamespaceFromString(config.DA.GetNamespace()).Bytes(),
 		namespaceDataBz: coreda.NamespaceFromString(config.DA.GetDataNamespace()).Bytes(),
 	}

block/internal/submitting/da_submitter_test.go

@@ -18,6 +18,7 @@ import (
 	coreda "github.com/evstack/ev-node/core/da"
 	"github.com/evstack/ev-node/pkg/config"
 	"github.com/evstack/ev-node/pkg/genesis"
+	"github.com/evstack/ev-node/pkg/rpc/server"
 	"github.com/evstack/ev-node/pkg/signer"
 	"github.com/evstack/ev-node/pkg/signer/noop"
 	"github.com/evstack/ev-node/pkg/store"
@@ -83,6 +84,27 @@ func TestDASubmitter_NewDASubmitter(t *testing.T) {
 	assert.NotNil(t, submitter.genesis)
 }
 
+func TestNewDASubmitterSetsVisualizerWhenEnabled(t *testing.T) {
+	t.Helper()
+	defer server.SetDAVisualizationServer(nil)
+
+	cfg := config.DefaultConfig()
+	cfg.RPC.EnableDAVisualization = true
+	cfg.Node.Aggregator = true
+
+	dummyDA := coreda.NewDummyDA(10_000_000, 0, 0, 10*time.Millisecond)
+
+	NewDASubmitter(
+		dummyDA,
+		cfg,
+		genesis.Genesis{},
+		common.DefaultBlockOptions(),
+		zerolog.Nop(),
+	)
+
+	require.NotNil(t, server.GetDAVisualizationServer())
+}
+
 func TestDASubmitter_SubmitHeaders_Success(t *testing.T) {
 	submitter, st, cm, _, gen := setupDASubmitterTest(t)
 	ctx := context.Background()

block/internal/submitting/submitter_test.go

@@ -19,6 +19,7 @@ import (
 	"github.com/evstack/ev-node/block/internal/common"
 	"github.com/evstack/ev-node/pkg/config"
 	"github.com/evstack/ev-node/pkg/genesis"
+	"github.com/evstack/ev-node/pkg/rpc/server"
 	"github.com/evstack/ev-node/pkg/signer"
 	"github.com/evstack/ev-node/pkg/store"
 	testmocks "github.com/evstack/ev-node/test/mocks"
@@ -141,11 +142,16 @@ func TestSubmitter_processDAInclusionLoop_advances(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
+	// Clean up any existing visualization server
+	defer server.SetDAVisualizationServer(nil)
+	server.SetDAVisualizationServer(nil)
+
 	cm, st := newTestCacheAndStore(t)
 
 	// small block time to tick quickly
 	cfg := config.DefaultConfig()
 	cfg.DA.BlockTime.Duration = 5 * time.Millisecond
+	cfg.RPC.EnableDAVisualization = false // Ensure visualization is disabled
 	metrics := common.PrometheusMetrics("test")
 
 	exec := testmocks.NewMockExecutor(t)

docs/learn/specs/block-manager.md

@@ -595,20 +595,20 @@ When both header and data are available for a height:
 
 The components communicate through well-defined interfaces:
 
-#### Executor ↔ Core Executor:
+#### Executor ↔ Core Executor
 
 - `InitChain`: initializes the chain state with the given genesis time, initial height, and chain ID using `InitChainSync` on the executor to obtain initial `appHash` and initialize the state.
 - `CreateBlock`: prepares a block with transactions from the provided batch data.
 - `ApplyBlock`: validates the block, executes the block (apply transactions), captures validator updates, and returns updated state.
 - `SetFinal`: marks the block as final when both its header and data are confirmed on the DA layer.
 - `GetTxs`: retrieves transactions from the application (used by Reaper component).
 
-#### Reaper ↔ Sequencer:
+#### Reaper ↔ Sequencer
 
 - `GetNextBatch`: retrieves the next batch of transactions to include in a block.
 - `VerifyBatch`: validates that a batch came from the expected sequencer.
 
-#### Submitter/Syncer ↔ DA Layer:
+#### Submitter/Syncer ↔ DA Layer
 
 - `Submit`: submits headers or data blobs to the DA network.
 - `Get`: retrieves headers or data blobs from the DA network.
@@ -617,17 +617,20 @@ The components communicate through well-defined interfaces:
 ## Assumptions and Considerations
 
 ### Component Architecture
+
 - The block package uses a modular component architecture instead of a monolithic manager
 - Components are created based on node type: aggregator nodes get all components, non-aggregator nodes only get synchronization components
 - Each component has a specific responsibility and communicates through well-defined interfaces
 - Components share a common Cache Manager for coordination and state tracking
 
 ### Initialization and State Management
+
 - Components load the initial state from the local store and use genesis if not found in the local store, when the node (re)starts
 - The default mode for aggregator nodes is normal (not lazy)
 - Components coordinate through channels and shared cache structures
 
 ### Block Production (Executor Component)
+
 - The Executor can produce empty blocks
 - In lazy aggregation mode, the Executor maintains consistency with the DA layer by producing empty blocks at regular intervals, ensuring a 1:1 mapping between DA layer blocks and execution layer blocks
 - The lazy aggregation mechanism uses a dual timer approach:
@@ -637,6 +640,7 @@ The components communicate through well-defined interfaces:
 - Transaction notifications from the `Reaper` to the `Executor` are handled via a non-blocking notification channel (`txNotifyCh`) to prevent backpressure
 
 ### DA Submission (Submitter Component)
+
 - The Submitter enforces `MaxPendingHeadersAndData` limit to prevent unbounded growth of pending queues during DA submission issues
 - Headers and data are submitted separately to the DA layer using different namespaces, supporting the header/data separation architecture
 - The Cache Manager uses persistent caches for headers and data to track seen items and DA inclusion status
@@ -645,16 +649,19 @@ The components communicate through well-defined interfaces:
 - Gas price management in the Submitter includes automatic adjustment with `GasMultiplier` on DA submission retries
 
 ### Storage and Persistence
+
 - Components use persistent storage (disk) when the `root_dir` and `db_path` configuration parameters are specified in `config.yaml` file under the app directory. If these configuration parameters are not specified, the in-memory storage is used, which will not be persistent if the node stops
 - The Syncer does not re-apply blocks when they transition from soft confirmed to DA included status. The block is only marked DA included in the caches
 - Header and data stores use separate prefixes for isolation in the underlying database
 - The genesis `ChainID` is used to create separate `PubSubTopID`s for headers and data in go-header
 
 ### P2P and Synchronization
+
 - Block sync over the P2P network works only when a full node is connected to the P2P network by specifying the initial seeds to connect to via `P2PConfig.Seeds` configuration parameter when starting the full node
 - Node's context is passed down to all components to support graceful shutdown and cancellation
 
 ### Architecture Design Decisions
+
 - The Executor supports custom signature payload providers for headers, enabling flexible signing schemes
 - The component architecture supports the separation of header and data structures in Evolve. This allows for expanding the sequencing scheme beyond single sequencing and enables the use of a decentralized sequencer mode. For detailed information on this architecture, see the [Header and Data Separation ADR](../../adr/adr-014-header-and-data-separation.md)
 - Components process blocks with a minimal header format, which is designed to eliminate dependency on CometBFT's header format and can be used to produce an execution layer tailored header if needed. For details on this header structure, see the [Evolve Minimal Header](../../adr/adr-015-rollkit-minimal-header.md) specification

docs/learn/specs/full_node.md

@@ -35,13 +35,15 @@ The [Store] is initialized with `DefaultStore`, an implementation of the [store
 The [Block Components] provide a modular architecture for managing block-related operations. Instead of a single monolithic manager, the system uses specialized components:
 
 **For Aggregator Nodes:**
+
 - **Executor**: Block production (normal and lazy modes) and state transitions
 - **Reaper**: Transaction collection and submission to sequencer
 - **Submitter**: Header and data submission to DA layer
 - **Syncer**: Block retrieval and synchronization from DA and P2P
 - **Cache Manager**: Coordination and tracking across all components
 
 **For Non-Aggregator Nodes:**
+
 - **Syncer**: Block retrieval and synchronization from DA and P2P
 - **Cache Manager**: Tracking and caching of synchronized blocks
 

pkg/config/defaults.go

@@ -74,7 +74,8 @@ func DefaultConfig() Config {
 			SignerPath: "config",
 		},
 		RPC: RPCConfig{
-			Address: "127.0.0.1:7331",
+			Address:               "127.0.0.1:7331",
+			EnableDAVisualization: false,
 		},
 	}
 }