fix: ensure node and services are watcing for the same shutdown signal from the context cancel (#1406)

<!--
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.
-->

## 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. 
-->

The nodes were creating a new context with cancel after passing the
original context through to the services. This meant that when the node
shutdown and cancelled its context, its services did not receive a
shutdown signal until the original caller cancelled its context. This
was leading to panics in testing about writing to a logger test file
that was already closed.

This change was verified by reverting the changes in #1402 and testing
in a loop 100 times with no panics.

**EDIT 1**
Since no good deed goes unpunished 🙃 some timeouts surfaced. This was
due to `context.Background()` being used for subscribe and unsubscribe
events. Under the hood those events are watching for `ctx.Done()`
events, so when `context.Background()` is passed in, they can hang
indefinitely if the other signals aren't triggered. The test that was
most prone to this timeout was run in a loop 1000 times to verify the
issue was fixed.

Additionally, added some more checks for `ctx.Done()` for faster
shutdowns.

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

- **New Features**
- Enhanced node shutdown process with context cancellation for improved
service termination.
  
- **Refactor**
- Updated error handling and context management in node creation
functions.
- Improved context usage in block synchronization and event subscription
methods.
  
- **Tests**
- Adjusted full node integration tests to reflect new context
management.
  
- **Chores**
  - Removed redundant node cancellation in test cleanup function.
  
- **Documentation**
  - No visible changes to end-users.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: MSevey

block/manager.go

@@ -386,6 +386,11 @@ func (m *Manager) sendNonBlockingSignalToRetrieveCh() {
 // If commit for block h+1 is available, we proceed with sync process, and remove synced block from sync cache.
 func (m *Manager) trySyncNextBlock(ctx context.Context, daHeight uint64) error {
 	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+		}
 		currentHeight := m.store.Height()
 		b, ok := m.blockCache.getBlock(currentHeight + 1)
 		if !ok {
@@ -516,13 +521,24 @@ func (m *Manager) RetrieveLoop(ctx context.Context) {
 }
 
 func (m *Manager) processNextDABlock(ctx context.Context) error {
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	default:
+	}
+
 	// TODO(tzdybal): extract configuration option
 	maxRetries := 10
 	daHeight := atomic.LoadUint64(&m.daHeight)
 
 	var err error
 	m.logger.Debug("trying to retrieve block from DA", "daHeight", daHeight)
 	for r := 0; r < maxRetries; r++ {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+		}
 		blockResp, fetchErr := m.fetchBlock(ctx, daHeight)
 		if fetchErr == nil {
 			if blockResp.Code == da.StatusNotFound {
@@ -616,6 +632,12 @@ func (m *Manager) IsProposer() (bool, error) {
 }
 
 func (m *Manager) publishBlock(ctx context.Context) error {
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	default:
+	}
+
 	var lastCommit *types.Commit
 	var lastHeaderHash types.Hash
 	var err error

node/full.go

@@ -95,7 +95,17 @@ func newFullNode(
 	clientCreator proxy.ClientCreator,
 	genesis *cmtypes.GenesisDoc,
 	logger log.Logger,
-) (*FullNode, error) {
+) (fn *FullNode, err error) {
+	// Create context with cancel so that all services using the context can
+	// catch the cancel signal when the node shutdowns
+	ctx, cancel := context.WithCancel(ctx)
+	defer func() {
+		// If there is an error, cancel the context
+		if err != nil {
+			cancel()
+		}
+	}()
+
 	proxyApp, err := initProxyApp(clientCreator, logger)
 	if err != nil {
 		return nil, err
@@ -147,8 +157,6 @@ func newFullNode(
 		return nil, err
 	}
 
-	ctx, cancel := context.WithCancel(ctx)
-
 	node := &FullNode{
 		proxyApp:       proxyApp,
 		eventBus:       eventBus,

node/full_client.go

@@ -112,7 +112,7 @@ func (c *FullClient) BroadcastTxCommit(ctx context.Context, tx cmtypes.Tx) (*cty
 		return nil, err
 	}
 	defer func() {
-		if err := c.EventBus.Unsubscribe(context.Background(), subscriber, q); err != nil {
+		if err := c.EventBus.Unsubscribe(ctx, subscriber, q); err != nil {
 			c.Logger.Error("Error unsubscribing from eventBus", "err", err)
 		}
 	}()
@@ -885,7 +885,7 @@ func (c *FullClient) resubscribe(subscriber string, q cmpubsub.Query) cmtypes.Su
 			return nil
 		}
 
-		sub, err := c.EventBus.Subscribe(context.Background(), subscriber, q)
+		sub, err := c.EventBus.Subscribe(c.node.ctx, subscriber, q)
 		if err == nil {
 			return sub
 		}

node/full_node_integration_test.go

@@ -43,8 +43,6 @@ func prepareProposalResponse(_ context.Context, req *abci.RequestPrepareProposal
 func TestCentralizedSequencer(t *testing.T) {
 	require := require.New(t)
 	assert := assert.New(t)
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
 	genDoc, privkey := types.GetGenesisWithPrivkey()
 	genDoc.AppHash = make([]byte, 32)
 	nodeKey := &p2p.NodeKey{
@@ -70,7 +68,9 @@ func TestCentralizedSequencer(t *testing.T) {
 		DAStartHeight: 1,
 		DABlockTime:   1 * time.Second,
 	}
-	node, err := newFullNode(ctx, config.NodeConfig{DAAddress: MockServerAddr, Aggregator: false, BlockManagerConfig: blockManagerConfig}, signingKey, signingKey, proxy.NewLocalClientCreator(app), genDoc, log.TestingLogger())
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	node, err := newFullNode(ctx, config.NodeConfig{DAAddress: MockServerAddr, Aggregator: false, BlockManagerConfig: blockManagerConfig}, signingKey, signingKey, proxy.NewLocalClientCreator(app), genDoc, test.NewFileLogger(t))
 	require.NoError(err)
 	node.dalc = dalc
 	node.blockManager.SetDALC(dalc)
@@ -100,8 +100,7 @@ func TestCentralizedSequencer(t *testing.T) {
 	submitResp := dalc.SubmitBlocks(ctx, []*types.Block{validBlock, junkProposerBlock, sigInvalidBlock})
 	fmt.Println(submitResp)
 	require.Equal(submitResp.Code, da.StatusSuccess)
-
-	require.NoError(testutils.Retry(3000, 100*time.Millisecond, func() error {
+	require.NoError(testutils.Retry(300, 100*time.Millisecond, func() error {
 		block, err := node.Store.GetBlock(1)
 		if err != nil {
 			return err
@@ -482,16 +481,20 @@ func TestSubmitBlocksToDA(t *testing.T) {
 		assert.NoError(seq.Stop())
 	}()
 
-	timer := time.NewTimer(5 * seq.nodeConfig.DABlockTime)
-	<-timer.C
-
 	numberOfBlocksToSyncTill := seq.Store.Height()
 
 	//Make sure all produced blocks made it to DA
 	for i := uint64(1); i <= numberOfBlocksToSyncTill; i++ {
-		block, err := seq.Store.GetBlock(i)
-		require.NoError(err)
-		require.True(seq.blockManager.IsDAIncluded(block.Hash()), block.Height())
+		require.NoError(testutils.Retry(300, 100*time.Millisecond, func() error {
+			block, err := seq.Store.GetBlock(i)
+			if err != nil {
+				return err
+			}
+			if !seq.blockManager.IsDAIncluded(block.Hash()) {
+				return fmt.Errorf("block %d not DA included", block.Height())
+			}
+			return nil
+		}))
 	}
 }
 

node/light.go

@@ -49,7 +49,17 @@ func newLightNode(
 	clientCreator proxy.ClientCreator,
 	genesis *cmtypes.GenesisDoc,
 	logger log.Logger,
-) (*LightNode, error) {
+) (ln *LightNode, err error) {
+	// Create context with cancel so that all services using the context can
+	// catch the cancel signal when the node shutdowns
+	ctx, cancel := context.WithCancel(ctx)
+	defer func() {
+		// If there is an error, cancel the context
+		if err != nil {
+			cancel()
+		}
+	}()
+
 	// Create the proxyApp and establish connections to the ABCI app (consensus, mempool, query).
 	proxyApp := proxy.NewAppConns(clientCreator, proxy.NopMetrics())
 	proxyApp.SetLogger(logger.With("module", "proxy"))
@@ -71,8 +81,6 @@ func newLightNode(
 		return nil, fmt.Errorf("error while initializing HeaderSyncService: %w", err)
 	}
 
-	ctx, cancel := context.WithCancel(ctx)
-
 	node := &LightNode{
 		P2P:          client,
 		proxyApp:     proxyApp,

node/node_test.go

@@ -19,9 +19,6 @@ var MockServerAddr = ":7980"
 
 // cleanUpNode stops the node and checks if it is running
 func cleanUpNode(node Node, t *testing.T) {
-	defer func() {
-		node.Cancel()
-	}()
 	assert.NoError(t, node.Stop())
 	assert.False(t, node.IsRunning())
 }

state/txindex/indexer_service.go

@@ -47,14 +47,14 @@ func (is *IndexerService) OnStart() error {
 	// canceled due to not pulling messages fast enough. Cause this might
 	// sometimes happen when there are no other subscribers.
 	blockSub, err := is.eventBus.SubscribeUnbuffered(
-		context.Background(),
+		is.ctx,
 		subscriber,
 		types.EventQueryNewBlockEvents)
 	if err != nil {
 		return err
 	}
 
-	txsSub, err := is.eventBus.SubscribeUnbuffered(context.Background(), subscriber, types.EventQueryTx)
+	txsSub, err := is.eventBus.SubscribeUnbuffered(is.ctx, subscriber, types.EventQueryTx)
 	if err != nil {
 		return err
 	}
@@ -130,6 +130,6 @@ func (is *IndexerService) OnStart() error {
 // OnStop implements service.Service by unsubscribing from all transactions.
 func (is *IndexerService) OnStop() {
 	if is.eventBus.IsRunning() {
-		_ = is.eventBus.UnsubscribeAll(context.Background(), subscriber)
+		_ = is.eventBus.UnsubscribeAll(is.ctx, subscriber)
 	}
 }

test/context/context.go

@@ -0,0 +1,31 @@
+package context
+
+import (
+	"context"
+	"fmt"
+	"runtime"
+	"strings"
+)
+
+// WithDebugCancelFunc returns a custom cancel function with debugging information.
+func WithDebugCancelFunc(ctx context.Context) (context.Context, context.CancelFunc) {
+	childCtx, cancelFunc := context.WithCancel(ctx)
+	debugCancelFunc := func() {
+		var pcs [5]uintptr // Capture the last 5 frames
+		n := runtime.Callers(1, pcs[:])
+		frames := runtime.CallersFrames(pcs[:n])
+
+		var trace []string
+		for {
+			frame, more := frames.Next()
+			trace = append(trace, fmt.Sprintf("%s:%d %s", frame.File, frame.Line, frame.Function))
+			if !more {
+				break
+			}
+		}
+
+		fmt.Printf("Context canceled by the following last 5 levels of the stack trace:\n%s\n", strings.Join(trace, "\n"))
+		cancelFunc() // Call the standard context cancel function
+	}
+	return childCtx, debugCancelFunc
+}