For loop audit (cosmos#1117)

<!--
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
I did a quick audit of the code for infinite for loops to catch any
remaining places where we might be hanging after a shutdown.

This PR adds some comments to explain some test helpers and adds two
select statements that watch for cancellation of subscriptions to help
with a cleaner shutdown.

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

closes cosmos#1069 

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

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

da/mock/mock.go

@@ -73,6 +73,10 @@ func (m *DataAvailabilityLayerClient) Init(_ types.NamespaceID, config []byte, d
 func (m *DataAvailabilityLayerClient) Start() error {
 	m.logger.Debug("Mock Data Availability Layer Client starting")
 	go func() {
+		// This infinite loop is expected to be stopped and cleaned up
+		// by the GC. This is OK because it is only a test helper and
+		// the block time is sufficiently long that there should not be
+		// a lot of wasted cycles.
 		for {
 			time.Sleep(m.config.BlockTime)
 			m.updateDAHeight()

da/mock/util.go

@@ -174,6 +174,10 @@ func RandNamespace() share.Namespace {
 	rb := make([]byte, namespace.NamespaceVersionZeroIDSize)
 	r.Read(rb) // nolint:gosec
 	for {
+		// This infinite loop is here to ensure that we generate a valid
+		// namespace for testing. We don't care about errors from
+		// invalid namespaces and the chances of this loop firing a
+		// large number of times is very low.
 		namespace, _ := share.NewBlobNamespaceV0(rb)
 		if err := namespace.ValidateForData(); err != nil {
 			continue

state/txindex/indexer_service.go

@@ -58,35 +58,44 @@ func (is *IndexerService) OnStart() error {
 
 	go func() {
 		for {
-			msg := <-blockHeadersSub.Out()
-			eventDataHeader := msg.Data().(types.EventDataNewBlockHeader)
-			height := eventDataHeader.Header.Height
-			batch := NewBatch(eventDataHeader.NumTxs)
-
-			for i := int64(0); i < eventDataHeader.NumTxs; i++ {
-				msg2 := <-txsSub.Out()
-				txResult := msg2.Data().(types.EventDataTx).TxResult
-
-				if err = batch.Add(&txResult); err != nil {
-					is.Logger.Error(
-						"failed to add tx to batch",
-						"height", height,
-						"index", txResult.Index,
-						"err", err,
-					)
+			select {
+			case <-blockHeadersSub.Cancelled():
+				return
+			case msg := <-blockHeadersSub.Out():
+				eventDataHeader := msg.Data().(types.EventDataNewBlockHeader)
+				height := eventDataHeader.Header.Height
+				batch := NewBatch(eventDataHeader.NumTxs)
+
+				for i := int64(0); i < eventDataHeader.NumTxs; i++ {
+					select {
+					case <-txsSub.Cancelled():
+						return
+					case msg2 := <-txsSub.Out():
+						txResult := msg2.Data().(types.EventDataTx).TxResult
+
+						if err = batch.Add(&txResult); err != nil {
+							is.Logger.Error(
+								"failed to add tx to batch",
+								"height", height,
+								"index", txResult.Index,
+								"err", err,
+							)
+						}
+					}
 				}
-			}
 
-			if err := is.blockIdxr.Index(eventDataHeader); err != nil {
-				is.Logger.Error("failed to index block", "height", height, "err", err)
-			} else {
-				is.Logger.Info("indexed block", "height", height)
-			}
+				if err := is.blockIdxr.Index(eventDataHeader); err != nil {
+					is.Logger.Error("failed to index block", "height", height, "err", err)
+				} else {
+					is.Logger.Info("indexed block", "height", height)
+				}
+
+				if err = is.txIdxr.AddBatch(batch); err != nil {
+					is.Logger.Error("failed to index block txs", "height", height, "err", err)
+				} else {
+					is.Logger.Debug("indexed block txs", "height", height, "num_txs", eventDataHeader.NumTxs)
+				}
 
-			if err = is.txIdxr.AddBatch(batch); err != nil {
-				is.Logger.Error("failed to index block txs", "height", height, "err", err)
-			} else {
-				is.Logger.Debug("indexed block txs", "height", height, "num_txs", eventDataHeader.NumTxs)
 			}
 		}
 	}()