x

Author: alpe

block/components.go

@@ -100,6 +100,7 @@ func (bc *Components) Start(ctx context.Context) error {
 func (bc *Components) Stop() error {
 	var errs error
 	if bc.Executor != nil {
+		println("+++ stopping executor")
 		if err := bc.Executor.Stop(); err != nil {
 			errs = errors.Join(errs, fmt.Errorf("failed to stop executor: %w", err))
 		}

node/failover.go

@@ -165,16 +165,14 @@ func setupFailoverState(
 }
 
 func (f *failoverState) Run(ctx context.Context) (multiErr error) {
-	//wg, ctx := errgroup.WithContext(ctx)
-	var wg = errgroup.Group{}
+	var wg errgroup.Group
 	wg.Go(func() error {
 		f.logger.Info().Str("addr", f.rpcServer.Addr).Msg("Started RPC server")
 		if err := f.rpcServer.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
 			return err
 		}
 		return nil
 	})
-	defer f.rpcServer.Shutdown(context.Background()) // nolint: errcheck
 
 	if err := f.p2pClient.Start(ctx); err != nil {
 		return fmt.Errorf("start p2p: %w", err)
@@ -184,8 +182,12 @@ func (f *failoverState) Run(ctx context.Context) (multiErr error) {
 	if err := f.headerSyncService.Start(ctx); err != nil {
 		return fmt.Errorf("error while starting header sync service: %w", err)
 	}
+
 	defer func() {
-		if err := f.headerSyncService.Stop(context.Background()); err != nil && !errors.Is(err, context.Canceled) {
+		shutdownCtx, done := context.WithTimeout(context.Background(), 3*time.Second)
+		defer done()
+
+		if err := f.headerSyncService.Stop(shutdownCtx); err != nil && !errors.Is(err, context.Canceled) {
 			multiErr = errors.Join(multiErr, fmt.Errorf("stopping header sync: %w", err))
 		}
 	}()
@@ -194,7 +196,10 @@ func (f *failoverState) Run(ctx context.Context) (multiErr error) {
 		return fmt.Errorf("error while starting data sync service: %w", err)
 	}
 	defer func() {
-		if err := f.dataSyncService.Stop(context.Background()); err != nil && !errors.Is(err, context.Canceled) {
+		shutdownCtx, done := context.WithTimeout(context.Background(), 3*time.Second)
+		defer done()
+
+		if err := f.dataSyncService.Stop(shutdownCtx); err != nil && !errors.Is(err, context.Canceled) {
 			multiErr = errors.Join(multiErr, fmt.Errorf("stopping data sync: %w", err))
 		}
 	}()
@@ -211,6 +216,12 @@ func (f *failoverState) Run(ctx context.Context) (multiErr error) {
 		}
 	}()
 
+	defer func() { // shutdown first
+		shutdownCtx, done := context.WithTimeout(context.Background(), 3*time.Second)
+		defer done()
+		_ = f.rpcServer.Shutdown(shutdownCtx)
+	}()
+
 	return wg.Wait()
 }
 

node/full_node_integration_test.go

@@ -84,16 +84,14 @@ func TestTxGossipingMultipleNodesDAIncluded(t *testing.T) {
 
 	numNodes := 4
 	nodes, cleanups := createNodesWithCleanup(t, numNodes, config)
-	for _, cleanup := range cleanups {
-		defer cleanup()
-	}
 
 	ctxs, cancels := createNodeContexts(numNodes)
 	var runningWg sync.WaitGroup
 
 	errChan := make(chan error, numNodes)
 	// Start only the sequencer first
 	startNodeInBackground(t, nodes, ctxs, &runningWg, 0, errChan)
+	t.Cleanup(func() { shutdownAndWait(t, cleanups, &runningWg, 10*time.Second) })
 
 	// Wait for the first block to be produced by the sequencer
 	err := waitForFirstBlock(nodes[0], Header)
@@ -165,16 +163,13 @@ func TestFastDASync(t *testing.T) {
 	config.DA.BlockTime = evconfig.DurationWrapper{Duration: 200 * time.Millisecond}
 
 	nodes, cleanups := createNodesWithCleanup(t, 2, config)
-	for _, cleanup := range cleanups {
-		defer cleanup()
-	}
-
 	ctxs, cancels := createNodeContexts(len(nodes))
 	var runningWg sync.WaitGroup
 
 	errChan := make(chan error, len(nodes))
 	// Start only the first node
 	startNodeInBackground(t, nodes, ctxs, &runningWg, 0, errChan)
+	t.Cleanup(func() { shutdownAndWait(t, cleanups, &runningWg, 10*time.Second) })
 
 	// Wait for the first node to produce a few blocks
 	blocksToWaitFor := uint64(2)
@@ -185,6 +180,7 @@ func TestFastDASync(t *testing.T) {
 
 	// Now start the second node and time its sync
 	startNodeInBackground(t, nodes, ctxs, &runningWg, 1, errChan)
+
 	start := time.Now()
 	// Wait for the second node to catch up to the first node
 	require.NoError(waitForAtLeastNBlocks(nodes[1], blocksToWaitFor, Store))
@@ -327,6 +323,7 @@ func testSingleSequencerSingleFullNode(t *testing.T, source Source) {
 
 	// Start the sequencer first
 	startNodeInBackground(t, nodes, ctxs, &runningWg, 0, errChan)
+	t.Cleanup(func() { shutdownAndWait(t, cancels, &runningWg, 10*time.Second) })
 
 	// Wait for the sequencer to produce at first block
 	require.NoError(waitForFirstBlock(nodes[0], source))
@@ -346,9 +343,6 @@ func testSingleSequencerSingleFullNode(t *testing.T, source Source) {
 
 	// Verify both nodes are synced using the helper
 	require.NoError(verifyNodesSynced(nodes[0], nodes[1], source))
-
-	// Cancel all node contexts to signal shutdown and wait
-	shutdownAndWait(t, cancels, &runningWg, 5*time.Second)
 }
 
 // testSingleSequencerTwoFullNodes sets up a single sequencer and two full nodes, starts the sequencer, waits for it to produce a block, then starts the full nodes.
@@ -370,6 +364,7 @@ func testSingleSequencerTwoFullNodes(t *testing.T, source Source) {
 
 	// Start the sequencer first
 	startNodeInBackground(t, nodes, ctxs, &runningWg, 0, errChan)
+	t.Cleanup(func() { shutdownAndWait(t, cancels, &runningWg, 10*time.Second) })
 
 	// Wait for the sequencer to produce at first block
 	require.NoError(waitForFirstBlock(nodes[0], source))
@@ -397,9 +392,6 @@ func testSingleSequencerTwoFullNodes(t *testing.T, source Source) {
 	for i := 1; i < numNodes; i++ {
 		require.NoError(verifyNodesSynced(nodes[0], nodes[i], source))
 	}
-
-	// Cancel all node contexts to signal shutdown and wait
-	shutdownAndWait(t, cancels, &runningWg, 5*time.Second)
 }
 
 // testSingleSequencerSingleFullNodeTrustedHash sets up a single sequencer and a single full node with a trusted hash, starts the sequencer, waits for it to produce a block, then starts the full node with the trusted hash.
@@ -421,6 +413,7 @@ func testSingleSequencerSingleFullNodeTrustedHash(t *testing.T, source Source) {
 
 	// Start the sequencer first
 	startNodeInBackground(t, nodes, ctxs, &runningWg, 0, errChan)
+	t.Cleanup(func() { shutdownAndWait(t, cancels, &runningWg, 10*time.Second) })
 
 	// Wait for the sequencer to produce at first block
 	require.NoError(waitForFirstBlock(nodes[0], source))
@@ -461,9 +454,6 @@ func testSingleSequencerSingleFullNodeTrustedHash(t *testing.T, source Source) {
 
 	// Verify both nodes are synced using the helper
 	require.NoError(verifyNodesSynced(nodes[0], nodes[1], source))
-
-	// Cancel all node contexts to signal shutdown and wait
-	shutdownAndWait(t, cancels, &runningWg, 5*time.Second)
 }
 
 // TestTwoChainsInOneNamespace verifies that two chains in the same namespace can coexist without any issues.

node/helpers_test.go

@@ -107,6 +107,10 @@ func newTestNode(
 	remoteSigner, err := remote_signer.NewNoopSigner(genesisValidatorKey)
 	require.NoError(t, err)
 
+	logger := zerolog.Nop()
+	if testing.Verbose() {
+		logger = zerolog.New(zerolog.NewTestWriter(t))
+	}
 	node, err := NewNode(
 		config,
 		executor,
@@ -117,7 +121,7 @@ func newTestNode(
 		genesis,
 		ds,
 		DefaultMetricsProvider(evconfig.DefaultInstrumentationConfig()),
-		zerolog.Nop(),
+		logger,
 		NodeOptions{},
 	)
 	require.NoError(t, err)
@@ -190,12 +194,6 @@ func createNodesWithCleanup(t *testing.T, num int, config evconfig.Config) ([]*F
 	// Update cleanup to cancel the context instead of calling Stop
 	cleanup := func() {
 		stopDAHeightTicker()
-		// slow down shutdown to let caches persist
-		for _, n := range nodes {
-			if n.IsRunning() {
-				time.Sleep(time.Second / 10)
-			}
-		}
 	}
 
 	nodes[0], cleanups[0] = aggNode.(*FullNode), cleanup
@@ -268,7 +266,7 @@ func startNodeInBackground(t *testing.T, nodes []*FullNode, ctxs []context.Conte
 }
 
 // Helper to cancel all contexts and wait for goroutines with timeout
-func shutdownAndWait(t *testing.T, cancels []context.CancelFunc, wg *sync.WaitGroup, timeout time.Duration) {
+func shutdownAndWait[T ~func()](t *testing.T, cancels []T, wg *sync.WaitGroup, timeout time.Duration) {
 	for _, cancel := range cancels {
 		cancel()
 	}

node/single_sequencer_integration_test.go

@@ -6,6 +6,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"net/http"
 	"sync"
 	"testing"
@@ -68,6 +69,7 @@ func (s *FullNodeTestSuite) SetupTest() {
 
 	// Start the node in a goroutine using Run instead of Start
 	s.startNodeInBackground(s.node)
+	s.T().Cleanup(func() { shutdownAndWait(s.T(), []context.CancelFunc{s.cancel}, &s.runningWg, 10*time.Second) })
 
 	// Verify that the node is running and producing blocks
 	err := waitForFirstBlock(s.node, Header)
@@ -242,6 +244,7 @@ func TestStateRecovery(t *testing.T) {
 
 	// Start the sequencer first
 	startNodeInBackground(t, []*FullNode{node}, []context.Context{ctx}, &runningWg, 0, nil)
+	t.Cleanup(func() { shutdownAndWait(t, []context.CancelFunc{cancel}, &runningWg, 10*time.Second) })
 
 	blocksToWaitFor := uint64(20)
 	// Wait for the sequencer to produce at first block
@@ -285,6 +288,7 @@ func TestMaxPendingHeadersAndData(t *testing.T) {
 
 	var runningWg sync.WaitGroup
 	startNodeInBackground(t, []*FullNode{node}, []context.Context{ctx}, &runningWg, 0, nil)
+	t.Cleanup(func() { shutdownAndWait(t, []context.CancelFunc{cancel}, &runningWg, 10*time.Second) })
 
 	// Wait blocks to be produced up to max pending
 	numExtraBlocks := uint64(5)
@@ -294,9 +298,6 @@ func TestMaxPendingHeadersAndData(t *testing.T) {
 	height, err := getNodeHeight(node, Store)
 	require.NoError(err)
 	require.LessOrEqual(height, config.Node.MaxPendingHeadersAndData)
-
-	// Stop the node and wait for shutdown
-	shutdownAndWait(t, []context.CancelFunc{cancel}, &runningWg, 10*time.Second)
 }
 
 // TestBatchQueueThrottlingWithDAFailure tests that when DA layer fails and MaxPendingHeadersAndData
@@ -334,7 +335,9 @@ func TestBatchQueueThrottlingWithDAFailure(t *testing.T) {
 	var runningWg sync.WaitGroup
 	errChan := make(chan error, 1)
 	startNodeInBackground(t, []*FullNode{node}, []context.Context{ctx}, &runningWg, 0, errChan)
+	t.Cleanup(func() { shutdownAndWait(t, []context.CancelFunc{cancel}, &runningWg, 10*time.Second) })
 	require.Len(errChan, 0, "Expected no errors when starting node")
+
 	// Wait for the node to start producing blocks
 	waitForBlockN(t, 1, node, config.Node.BlockTime.Duration)
 
@@ -387,13 +390,6 @@ func TestBatchQueueThrottlingWithDAFailure(t *testing.T) {
 	require.NoError(err)
 	t.Logf("Final height: %d", finalHeight)
 	cancel() // stop the node
-	if v, ok := node.leaderElection.(*singleRoleElector); ok {
-		// skip cache persistence to avoid race condition with shutdown
-		v.runnable.(*failoverState).bc.Cache = nil
-	} else {
-		time.Sleep(time.Second)
-	}
-
 	// The height should not have increased much due to MaxPendingHeadersAndData limit
 	// Allow at most 3 additional blocks due to timing and pending blocks in queue
 	heightIncrease := finalHeight - heightAfterDAFailure
@@ -427,6 +423,7 @@ func waitForBlockN(t *testing.T, n uint64, node *FullNode, blockInterval time.Du
 		return got >= n
 	}, timeout[0], blockInterval/2)
 }
+
 func TestReadinessEndpointWhenBlockProductionStops(t *testing.T) {
 	require := require.New(t)
 
@@ -443,14 +440,20 @@ func TestReadinessEndpointWhenBlockProductionStops(t *testing.T) {
 	defer cancel()
 
 	var runningWg sync.WaitGroup
-	startNodeInBackground(t, []*FullNode{node}, []context.Context{ctx}, &runningWg, 0, nil)
+	errChan := make(chan error, 1)
+	startNodeInBackground(t, []*FullNode{node}, []context.Context{ctx}, &runningWg, 0, errChan)
+	t.Cleanup(func() { shutdownAndWait(t, []context.CancelFunc{cancel}, &runningWg, 10*time.Second) })
+	require.Len(errChan, 0, "Expected no errors when starting node")
 
-	waitForBlockN(t, 3, node, config.Node.BlockTime.Duration)
+	waitForBlockN(t, 2, node, config.Node.BlockTime.Duration)
+	require.Len(errChan, 0, "Expected no errors when starting node")
 
 	resp, err := http.Get("http://" + config.RPC.Address + "/health/ready")
 	require.NoError(err)
-	require.Equal(http.StatusOK, resp.StatusCode, "Readiness should be READY while producing blocks")
+	body, err := io.ReadAll(resp.Body)
+	require.NoError(err)
 	resp.Body.Close()
+	require.Equal(http.StatusOK, resp.StatusCode, "Readiness should be READY while producing blocks: %s", body)
 
 	time.Sleep(time.Duration(config.Node.MaxPendingHeadersAndData+2) * config.Node.BlockTime.Duration)
 
@@ -466,6 +469,4 @@ func TestReadinessEndpointWhenBlockProductionStops(t *testing.T) {
 		defer resp.Body.Close()
 		return resp.StatusCode == http.StatusServiceUnavailable
 	}, 10*time.Second, 100*time.Millisecond, "Readiness should be UNREADY after aggregator stops producing blocks (5x block time)")
-
-	shutdownAndWait(t, []context.CancelFunc{cancel}, &runningWg, 10*time.Second)
 }

scripts/test.mk

@@ -18,7 +18,7 @@ test-all: test test-docker-e2e
 ## test-integration: Running integration tests
 test-integration:
 	@echo "--> Running integration tests"
-	@cd node && go test -mod=readonly -failfast -timeout=15m -tags='integration' ./...
+	@cd node && go test -mod=readonly -failfast -v -timeout=15m -tags='integration' ./...
 .PHONY: test-integration
 
 ## test-e2e: Running e2e tests