chore: make sure all blocking chan calls are also watching for shutdown signals (#1328)

In the spirit of shutdown handling, I did a quick audit of the codebase
for places where we had blocking channel calls and put them into select
statements so that we can also watch for a shutdown signal.

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

- **New Features**
- Introduced a new "DA Included" status for blocks, indicating a higher
level of finality.
- Added support for out-of-order rollup blocks on DA and a termination
condition related to sequencer double-signing.
- Implemented a new `BlockSyncService` with functions to start syncing
and handle block storage and broadcasting.
- Created a `NewPendingBlocks` function to manage blocks that need to be
published to the DA layer.

- **Improvements**
- Improved error handling by checking for context completion before
sending data to channels, ensuring more informative error messages.
- Enhanced the block confirmation process and block retrieval mechanism
with updated terminology and functionality.

- **Refactor**
- Renamed various functions and methods to better reflect their purpose,
such as `store.GetState`, `Manager.IsDAIncluded`, and
`Manager.BlockStoreRetrieveLoop`.
- Removed unused functions like `Manager.saveValidatorsToStore` and
`Manager.getLastStateValidators`.

- **Bug Fixes**
- Fixed potential goroutine leaks by handling context cancellation in
`newTxValidator`, `BroadcastTxCommit`, and `BroadcastTxSync` functions.

- **Tests**
- Updated test assertions from `assert` to `require` to ensure immediate
failure upon encountering errors during testing.

- **Documentation**
- Updated documentation to reflect changes in block confirmation
terminology and the introduction of the "DA Included" concept.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: MSevey

block/manager.go

@@ -16,6 +16,7 @@ import (
 	"github.com/cometbft/cometbft/proxy"
 	cmtypes "github.com/cometbft/cometbft/types"
 	"github.com/libp2p/go-libp2p/core/crypto"
+	"github.com/pkg/errors"
 	"go.uber.org/multierr"
 
 	"github.com/rollkit/rollkit/config"
@@ -426,6 +427,17 @@ func (m *Manager) BlockStoreRetrieveLoop(ctx context.Context) {
 			}
 			daHeight := atomic.LoadUint64(&m.daHeight)
 			for _, block := range blocks {
+				// Check for shut down event prior to logging
+				// and sending block to blockInCh. The reason
+				// for checking for the shutdown event
+				// separately is due to the inconsistent nature
+				// of the select statement when multiple cases
+				// are satisfied.
+				select {
+				case <-ctx.Done():
+					return
+				default:
+				}
 				m.logger.Debug("block retrieved from p2p block sync", "blockHeight", block.Height(), "daHeight", daHeight)
 				m.blockInCh <- newBlockEvent{block, daHeight}
 			}
@@ -501,6 +513,17 @@ func (m *Manager) processNextDABlock(ctx context.Context) error {
 				m.blockCache.setDAIncluded(blockHash)
 				m.logger.Info("block marked as DA included", "blockHeight", block.Height(), "blockHash", blockHash)
 				if !m.blockCache.isSeen(blockHash) {
+					// Check for shut down event prior to logging
+					// and sending block to blockInCh. The reason
+					// for checking for the shutdown event
+					// separately is due to the inconsistent nature
+					// of the select statement when multiple cases
+					// are satisfied.
+					select {
+					case <-ctx.Done():
+						return errors.WithMessage(ctx.Err(), "unable to send block to blockInCh, context done")
+					default:
+					}
 					m.blockInCh <- newBlockEvent{block, daHeight}
 				}
 			}
@@ -689,11 +712,13 @@ func (m *Manager) publishBlock(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-
-	// Check if the node has shutdown prior to publishing to channels
+	// Check for shut down event prior to sending the header and block to
+	// their respective channels. The reason for checking for the shutdown
+	// event separately is due to the inconsistent nature of the select
+	// statement when multiple cases are satisfied.
 	select {
 	case <-ctx.Done():
-		return nil
+		return errors.WithMessage(ctx.Err(), "unable to send header and block, context done")
 	default:
 	}
 

go.mod

@@ -21,6 +21,7 @@ require (
 	github.com/libp2p/go-libp2p-kad-dht v0.23.0
 	github.com/libp2p/go-libp2p-pubsub v0.9.3
 	github.com/multiformats/go-multiaddr v0.12.0
+	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.17.0
 	github.com/rollkit/go-da v0.0.0-20231117151938-ee3b613d7a3a
 	github.com/rs/cors v1.10.1
@@ -132,7 +133,6 @@ require (
 	github.com/pbnjay/memory v0.0.0-20210728143218-7b4eea64cf58 // indirect
 	github.com/pelletier/go-toml/v2 v2.1.0 // indirect
 	github.com/petermattis/goid v0.0.0-20230317030725-371a4b8eda08 // indirect
-	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/polydawn/refmt v0.89.0 // indirect
 	github.com/prometheus/client_model v0.4.1-0.20230718164431-9a2bf3000d16 // indirect

node/full.go

@@ -400,7 +400,11 @@ func (n *FullNode) newTxValidator() p2p.GossipValidator {
 		n.Logger.Debug("transaction received", "bytes", len(m.Data))
 		checkTxResCh := make(chan *abci.ResponseCheckTx, 1)
 		err := n.Mempool.CheckTx(m.Data, func(resp *abci.ResponseCheckTx) {
-			checkTxResCh <- resp
+			select {
+			case <-n.ctx.Done():
+				return
+			case checkTxResCh <- resp:
+			}
 		}, mempool.TxInfo{
 			SenderID:    n.mempoolIDs.GetForPeer(m.From),
 			SenderP2PID: corep2p.ID(m.From),

node/full_client.go

@@ -120,7 +120,11 @@ func (c *FullClient) BroadcastTxCommit(ctx context.Context, tx cmtypes.Tx) (*cty
 	// add to mempool and wait for CheckTx result
 	checkTxResCh := make(chan *abci.ResponseCheckTx, 1)
 	err = c.node.Mempool.CheckTx(tx, func(res *abci.ResponseCheckTx) {
-		checkTxResCh <- res
+		select {
+		case <-ctx.Done():
+			return
+		case checkTxResCh <- res:
+		}
 	}, mempool.TxInfo{})
 	if err != nil {
 		c.Logger.Error("Error on broadcastTxCommit", "err", err)
@@ -198,7 +202,11 @@ func (c *FullClient) BroadcastTxAsync(ctx context.Context, tx cmtypes.Tx) (*ctyp
 func (c *FullClient) BroadcastTxSync(ctx context.Context, tx cmtypes.Tx) (*ctypes.ResultBroadcastTx, error) {
 	resCh := make(chan *abci.ResponseCheckTx, 1)
 	err := c.node.Mempool.CheckTx(tx, func(res *abci.ResponseCheckTx) {
-		resCh <- res
+		select {
+		case <-ctx.Done():
+			return
+		case resCh <- res:
+		}
 	}, mempool.TxInfo{})
 	if err != nil {
 		return nil, err

rpc/json/ws_test.go

@@ -47,25 +47,25 @@ func TestWebSockets(t *testing.T) {
     }
 }
 `))
-	assert.NoError(err)
+	require.NoError(err)
 
 	err = conn.SetReadDeadline(time.Now().Add(1 * time.Second))
-	assert.NoError(err)
+	require.NoError(err)
 	typ, msg, err := conn.ReadMessage()
-	assert.NoError(err)
+	require.NoError(err)
 	assert.Equal(websocket.TextMessage, typ)
 	assert.NotEmpty(msg)
 
 	// wait for new block event
 	err = conn.SetReadDeadline(time.Now().Add(3 * time.Second))
-	assert.NoError(err)
+	require.NoError(err)
 	typ, msg, err = conn.ReadMessage()
-	assert.NoError(err)
+	require.NoError(err)
 	assert.Equal(websocket.TextMessage, typ)
 	assert.NotEmpty(msg)
 	var payload cmtypes.EventDataNewBlock
 	err = json.Unmarshal(msg, &payload)
-	assert.NoError(err)
+	require.NoError(err)
 	assert.NotNil(payload.ResultFinalizeBlock)
 	assert.NotNil(payload.Block)
 	assert.GreaterOrEqual(payload.Block.Height, int64(1))
@@ -79,6 +79,6 @@ func TestWebSockets(t *testing.T) {
 	handler.ServeHTTP(rsp, req)
 	assert.Equal(http.StatusOK, rsp.Code)
 	jsonResp := response{}
-	assert.NoError(json.Unmarshal(rsp.Body.Bytes(), &jsonResp))
+	require.NoError(json.Unmarshal(rsp.Body.Bytes(), &jsonResp))
 	assert.Nil(jsonResp.Error)
 }