Minor updates

Author: alpe

block/internal/executing/executor_restart_test.go

@@ -79,7 +79,7 @@ func TestExecutor_RestartUsesPendingHeader(t *testing.T) {
 	require.NoError(t, exec1.initializeState())
 
 	// Set up context for first executor
-	exec1.ctx, exec1.cancel = context.WithCancel(context.Background())
+	exec1.ctx, exec1.cancel = context.WithCancel(t.Context())
 
 	// First executor produces a block normally
 	mockSeq1.EXPECT().GetNextBatch(mock.Anything, mock.AnythingOfType("sequencer.GetNextBatchRequest")).
@@ -101,12 +101,12 @@ func TestExecutor_RestartUsesPendingHeader(t *testing.T) {
 	require.NoError(t, err)
 
 	// Verify first block was produced
-	h1, err := memStore.Height(context.Background())
+	h1, err := memStore.Height(t.Context())
 	require.NoError(t, err)
 	assert.Equal(t, uint64(1), h1)
 
 	// Store the produced block data for later verification
-	originalHeader, originalData, err := memStore.GetBlockData(context.Background(), 1)
+	originalHeader, originalData, err := memStore.GetBlockData(t.Context(), 1)
 	require.NoError(t, err)
 	assert.Equal(t, 2, len(originalData.Txs), "first block should have 2 transactions")
 
@@ -158,8 +158,7 @@ func TestExecutor_RestartUsesPendingHeader(t *testing.T) {
 	pendingHeader.DataHash = pendingData.DACommitment()
 
 	// Save pending block data (this is what would happen during a crash)
-	// We use savePendingBlock directly which writes to the metadata keys expected by the executor on restart
-	err = exec1.savePendingBlock(context.Background(), pendingHeader, pendingData)
+	err = exec1.savePendingBlock(t.Context(), pendingHeader, pendingData)
 	require.NoError(t, err)
 
 	// Stop first executor (simulating crash/restart)
@@ -196,7 +195,7 @@ func TestExecutor_RestartUsesPendingHeader(t *testing.T) {
 	require.NoError(t, exec2.initializeState())
 
 	// Set up context for second executor
-	exec2.ctx, exec2.cancel = context.WithCancel(context.Background())
+	exec2.ctx, exec2.cancel = context.WithCancel(t.Context())
 	defer exec2.cancel()
 
 	// Verify that the state is at height 1 (pending block at height 2 wasn't committed)
@@ -218,12 +217,12 @@ func TestExecutor_RestartUsesPendingHeader(t *testing.T) {
 	require.NoError(t, err)
 
 	// Verify height advanced to 2
-	h2, err := memStore.Height(context.Background())
+	h2, err := memStore.Height(t.Context())
 	require.NoError(t, err)
 	assert.Equal(t, uint64(2), h2, "height should advance to 2 using pending block")
 
 	// Verify the block at height 2 matches the pending block data
-	finalHeader, finalData, err := memStore.GetBlockData(context.Background(), 2)
+	finalHeader, finalData, err := memStore.GetBlockData(t.Context(), 2)
 	require.NoError(t, err)
 	assert.Equal(t, 3, len(finalData.Txs), "should use pending block with 3 transactions")
 	assert.Equal(t, []byte("pending_tx1"), []byte(finalData.Txs[0]))
@@ -385,7 +384,7 @@ func TestExecutor_RestartNoPendingHeader(t *testing.T) {
 	require.NoError(t, err)
 
 	// Verify normal operation
-	h, err := memStore.Height(context.Background())
+	h, err := memStore.Height(t.Context())
 	require.NoError(t, err)
 	assert.Equal(t, uint64(numBlocks+1), h)
 

block/internal/syncing/syncer.go

@@ -722,7 +722,7 @@ func (s *Syncer) TrySyncNextBlock(ctx context.Context, event *common.DAHeightEve
 	currentState := s.getLastState()
 	headerHash := header.Hash().String()
 
-	s.logger.Info().Uint64("height", nextHeight).Msg("syncing block started")
+	s.logger.Info().Uint64("height", nextHeight).Msg("syncing block")
 
 	// Compared to the executor logic where the current block needs to be applied first,
 	// here only the previous block needs to be applied to proceed to the verification.
@@ -732,17 +732,6 @@ func (s *Syncer) TrySyncNextBlock(ctx context.Context, event *common.DAHeightEve
 		s.cache.RemoveHeaderDAIncluded(headerHash)
 		s.cache.RemoveDataDAIncluded(data.DACommitment().String())
 
-		s.logger.Warn().
-			Err(err).
-			Uint64("height", header.Height()).
-			Uint64("time", uint64(header.Time().Unix())).
-			Hex("proposer", header.ProposerAddress).
-			Str("data_hash", hex.EncodeToString(header.DataHash)).
-			Str("app_hash", hex.EncodeToString(header.AppHash)).
-			Hex("last_header_hash", header.LastHeaderHash).
-			Int("len signature", len(header.Signature)).
-			Msg("block validation failed")
-
 		if !errors.Is(err, errInvalidState) && !errors.Is(err, errInvalidBlock) {
 			return errors.Join(errInvalidBlock, err)
 		}
@@ -810,7 +799,7 @@ func (s *Syncer) TrySyncNextBlock(ctx context.Context, event *common.DAHeightEve
 	if s.p2pHandler != nil {
 		s.p2pHandler.SetProcessedHeight(newState.LastBlockHeight)
 	}
-	s.logger.Info().Uint64("height", nextHeight).Msg("syncing block completed")
+
 	return nil
 }
 

node/failover.go

@@ -189,23 +189,17 @@ func (f *failoverState) Run(pCtx context.Context) (multiErr error) {
 		return nil
 	})
 
-	// P2P client persists across mode switches (started/closed by FullNode.Run).
-	// Reconfigure() was already called in setupFailoverState to re-bootstrap DHT.
-
-	// Start header and data sync services concurrently. Each service's
-	// initFromP2PWithRetry can block up to 30s when peers have no blocks
-	// (e.g. lazy mode sequencer at height 0). Running them in parallel
-	// avoids a 60s cumulative startup delay.
+	// start header and data sync services concurrently to avoid cumulative startup delay.
 	syncWg, syncCtx := errgroup.WithContext(ctx)
 	syncWg.Go(func() error {
 		if err := f.headerSyncService.Start(syncCtx); err != nil {
-			return fmt.Errorf("error while starting header sync service: %w", err)
+			return fmt.Errorf("header sync service: %w", err)
 		}
 		return nil
 	})
 	syncWg.Go(func() error {
 		if err := f.dataSyncService.Start(syncCtx); err != nil {
-			return fmt.Errorf("error while starting data sync service: %w", err)
+			return fmt.Errorf("data sync service: %w", err)
 		}
 		return nil
 	})

node/full.go

@@ -104,8 +104,6 @@ func newFullNode(
 		}
 	}
 
-	// The p2p client is fully configured and started before leader election.
-	// SyncService.getPeerIDs() gates peer usage on conf.Node.Aggregator.
 	leaderFactory := func() (raftpkg.Runnable, error) {
 		logger.Info().Msg("Starting aggregator-MODE")
 		nodeConfig.Node.Aggregator = true
@@ -283,8 +281,7 @@ func (n *FullNode) Run(parentCtx context.Context) error {
 		n.prometheusSrv, n.pprofSrv = n.startInstrumentationServer()
 	}
 
-	// Start the P2P client once. It persists across mode switches so that
-	// the host and PubSub (including externally registered topics) survive.
+	// Start the P2P client once. It persists across mode switches
 	if err := n.p2pClient.Start(ctx); err != nil {
 		return fmt.Errorf("start p2p: %w", err)
 	}

node/node.go

@@ -28,9 +28,6 @@ type NodeOptions struct {
 // NewNode returns a new Full or Light Node based on the config.
 // This is the entry point for composing a node, when compiling a node, you need to provide an executor.
 // Example executors can be found in apps/
-//
-// The p2pClient owns the node identity (private key) and is shared across
-// mode switches. It supports in-place reconfiguration via Reconfigure().
 func NewNode(
 	conf config.Config,
 	exec coreexecutor.Executor,

pkg/cmd/run_node.go

@@ -164,8 +164,6 @@ func StartNode(
 		executor = telemetry.WithTracingExecutor(executor)
 	}
 
-	// Create the P2P client. It is long-lived and reconfigured in-place
-	// on mode switches, avoiding costly teardown of the libp2p stack.
 	p2pClient, err := p2p.NewClient(nodeConfig.P2P, nodeKey.PrivKey, datastore, genesis.ChainID, logger, nil)
 	if err != nil {
 		return fmt.Errorf("create p2p client: %w", err)

pkg/p2p/client.go

@@ -49,8 +49,7 @@ type Client struct {
 	chainID string
 	privKey crypto.PrivKey
 
-	rawHost host.Host // unwrapped libp2p host, stored to avoid double-wrapping via routedhost
-	host    host.Host // may be wrapped with routedhost after DHT setup
+	host    host.Host
 	dht     *dht.IpfsDHT
 	disc    *discovery.RoutingDiscovery
 	gater   *conngater.BasicConnectionGater
@@ -124,20 +123,18 @@ func NewClientWithHost(
 // 4. Use active peer discovery to look for peers from same ORU network.
 func (c *Client) Start(ctx context.Context) error {
 	if c.started {
-		return nil // already started — called from FullNode.Run()
+		return nil
 	}
 	c.logger.Debug().Msg("starting P2P client")
 
 	if c.host != nil {
-		c.rawHost = c.host
 		return c.startWithHost(ctx, c.host)
 	}
 
 	h, err := c.listen()
 	if err != nil {
 		return err
 	}
-	c.rawHost = h
 	return c.startWithHost(ctx, h)
 }
 
@@ -185,21 +182,10 @@ func (c *Client) Close() error {
 	if c.host != nil {
 		err = errors.Join(err, c.host.Close())
 	}
+	c.started = false
 	return err
 }
 
-// PrivKey returns the node's private key.
-func (c *Client) PrivKey() crypto.PrivKey {
-	return c.privKey
-}
-
-// Reconfigure updates the mutable P2P configuration without tearing down
-// the libp2p host, PubSub, or DHT. Currently this only updates the
-// stored config; the sync service gates peer usage on conf.Node.Aggregator.
-func (c *Client) Reconfigure(conf config.P2PConfig) {
-	c.conf = conf
-}
-
 // Addrs returns listen addresses of Client.
 func (c *Client) Addrs() []multiaddr.Multiaddr {
 	return c.host.Addrs()