Improve node and services closing (#1554)

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

Resolves #1552
Resolves #1553

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

## Checklist

<!-- 
Please complete the checklist to ensure that the PR is ready to be
reviewed.

IMPORTANT:
PRs should be left in Draft until the below checklist is completed.
-->

- [x] New and updated code has appropriate documentation
- [x] New and updated code has new and/or updated testing
- [x] Required CI checks are passing
- [x] Visual proof for any user facing features like CLI or
documentation updates
- [x] Linked issues closed with keywords


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

- **New Features**
- Introduced a safe closure mechanism for data storage to enhance system
reliability.

- **Bug Fixes**
- Improved error reporting in block submission processes for clearer
understanding of operations.
- Refined the sequence of service shutdowns for more predictable
application behavior.

- **Tests**
- Enhanced full node integration tests with simplified context handling
and ensured clean-up post-execution.

- **Refactor**
- Adjusted the order of operations in service shutdown methods for
better error handling and control flow.
- Updated test cases to better manage the lifecycle of data storage,
ensuring accurate state after restarts.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: tzdybal

block/block_sync.go

@@ -220,17 +220,18 @@ func (bSyncService *BlockSyncService) Start() error {
 }
 
 // Stop is a part of Service interface.
+//
+// `blockStore` is closed last because it's used by other services.
 func (bSyncService *BlockSyncService) Stop() error {
-	err := bSyncService.blockStore.Stop(bSyncService.ctx)
-	err = errors.Join(
-		err,
+	err := errors.Join(
 		bSyncService.p2pServer.Stop(bSyncService.ctx),
 		bSyncService.ex.Stop(bSyncService.ctx),
 		bSyncService.sub.Stop(bSyncService.ctx),
 	)
 	if bSyncService.syncerStatus.isStarted() {
 		err = errors.Join(err, bSyncService.syncer.Stop(bSyncService.ctx))
 	}
+	err = errors.Join(err, bSyncService.blockStore.Stop(bSyncService.ctx))
 	return err
 }
 

block/header_sync.go

@@ -217,17 +217,18 @@ func (hSyncService *HeaderSyncService) Start() error {
 }
 
 // Stop is a part of Service interface.
+//
+// `headerStore` is closed last because it's used by other services.
 func (hSyncService *HeaderSyncService) Stop() error {
-	err := hSyncService.headerStore.Stop(hSyncService.ctx)
-	err = errors.Join(
-		err,
+	err := errors.Join(
 		hSyncService.p2pServer.Stop(hSyncService.ctx),
 		hSyncService.ex.Stop(hSyncService.ctx),
 		hSyncService.sub.Stop(hSyncService.ctx),
 	)
 	if hSyncService.syncerStatus.isStarted() {
 		err = errors.Join(err, hSyncService.syncer.Stop(hSyncService.ctx))
 	}
+	err = errors.Join(err, hSyncService.headerStore.Stop(hSyncService.ctx))
 	return err
 }
 

block/manager.go

@@ -833,7 +833,6 @@ func (m *Manager) submitBlocksToDA(ctx context.Context) error {
 	submittedAllBlocks := false
 	backoff := initialBackoff
 	blocksToSubmit := m.pendingBlocks.getPendingBlocks()
-	numTotalBlocks := len(blocksToSubmit)
 	numSubmittedBlocks := 0
 	attempt := 0
 	for ctx.Err() == nil && !submittedAllBlocks && attempt < maxSubmitAttempts {
@@ -861,10 +860,10 @@ func (m *Manager) submitBlocksToDA(ctx context.Context) error {
 
 	if !submittedAllBlocks {
 		return fmt.Errorf(
-			"failed to submit all blocks to DA layer, submitted %d of %d blocks after %d attempts",
+			"failed to submit all blocks to DA layer, submitted %d blocks (%d left) after %d attempts",
 			numSubmittedBlocks,
-			numTotalBlocks,
-			maxSubmitAttempts,
+			len(blocksToSubmit),
+			attempt,
 		)
 	}
 	return nil

node/full.go

@@ -408,21 +408,25 @@ func (n *FullNode) GetGenesisChunks() ([]string, error) {
 }
 
 // OnStop is a part of Service interface.
+//
+// p2pClient and sync services stop first, ceasing network activities. Then rest of services are halted.
+// Context is cancelled to signal goroutines managed by thread manager to stop.
+// Store is closed last because it's used by other services/goroutines.
 func (n *FullNode) OnStop() {
 	n.Logger.Info("halting full node...")
-	n.cancel()
-	n.threadManager.Wait()
 	n.Logger.Info("shutting down full node sub services...")
-	err := n.p2pClient.Close()
-	err = errors.Join(
-		err,
+	err := errors.Join(
+		n.p2pClient.Close(),
 		n.hSyncService.Stop(),
 		n.bSyncService.Stop(),
 		n.IndexerService.Stop(),
 	)
 	if n.prometheusSrv != nil {
 		err = errors.Join(err, n.prometheusSrv.Shutdown(n.ctx))
 	}
+	n.cancel()
+	n.threadManager.Wait()
+	err = errors.Join(err, n.Store.Close())
 	n.Logger.Error("errors while stopping node:", "errors", err)
 }
 

node/full_node_integration_test.go

@@ -94,26 +94,30 @@ func TestTxGossipingAndAggregation(t *testing.T) {
 	require := require.New(t)
 
 	clientNodes := 4
-	aggCtx, aggCancel := context.WithCancel(context.Background())
-	defer aggCancel()
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
+	aggCtx := context.Background()
+	ctx := context.Background()
 	nodes, apps := createNodes(aggCtx, ctx, clientNodes+1, getBMConfig(), t)
 	startNodes(nodes, apps, t)
+	defer func() {
+		for _, n := range nodes {
+			assert.NoError(n.Stop())
+		}
+	}()
 
 	// wait for nodes to start up and sync up till numBlocksToWaitFor
 	numBlocksToWaitFor := 5
 	for i := 1; i < len(nodes); i++ {
 		require.NoError(waitForAtLeastNBlocks(nodes[i], numBlocksToWaitFor, Store))
 	}
 
-	// Stop all the nodes before checking the calls to ABCI methods were done correctly
+	// Cancel all the nodes before checking the calls to ABCI methods were done correctly
+	// Can't stop here, because we need access to Store to test the state.
 	for _, node := range nodes {
-		assert.NoError(node.Stop())
+		node.Cancel()
 	}
 
-	// Now that the nodes have stopped, it should be safe to access the mock
-	// calls outside of the mutex controlled methods.
+	// Now that the nodes are cancelled, it should be safe to access the mock
+	// calls outside the mutex controlled methods.
 	//
 	// The reason we do this is because in the beginning of the test, we
 	// check that we have produced at least N blocks, which means we could

store/store.go

@@ -39,6 +39,11 @@ func New(ds ds.TxnDatastore) Store {
 	}
 }
 
+// Close safely closes underlying data storage, to ensure that data is actually saved.
+func (s *DefaultStore) Close() error {
+	return s.db.Close()
+}
+
 // SetHeight sets the height saved in the Store if it is higher than the existing height
 func (s *DefaultStore) SetHeight(ctx context.Context, height uint64) {
 	for {

store/store_test.go

@@ -129,21 +129,40 @@ func TestRestart(t *testing.T) {
 	t.Parallel()
 
 	assert := assert.New(t)
+	require := require.New(t)
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	kv, _ := NewDefaultInMemoryKVStore()
+
+	tmpDir, err := os.MkdirTemp("", t.Name())
+	require.NoError(err)
+	defer func() {
+		_ = os.RemoveAll(tmpDir)
+	}()
+
+	kv, err := NewDefaultKVStore(tmpDir, "test", "test")
+	require.NoError(err)
+
 	s1 := New(kv)
 	expectedHeight := uint64(10)
-	err := s1.UpdateState(ctx, types.State{
+	err = s1.UpdateState(ctx, types.State{
 		LastBlockHeight: expectedHeight,
 	})
 	assert.NoError(err)
 
+	err = s1.Close()
+	assert.NoError(err)
+
+	kv, err = NewDefaultKVStore(tmpDir, "test", "test")
+	require.NoError(err)
+
 	s2 := New(kv)
 	_, err = s2.GetState(ctx)
 	assert.NoError(err)
 
+	err = s2.Close()
+	assert.NoError(err)
+
 	assert.Equal(expectedHeight, s2.Height())
 }
 

store/types.go

@@ -40,4 +40,7 @@ type Store interface {
 	UpdateState(ctx context.Context, state types.State) error
 	// GetState returns last state saved with UpdateState.
 	GetState(ctx context.Context) (types.State, error)
+
+	// Close safely closes underlying data storage, to ensure that data is actually saved.
+	Close() error
 }