use unbuffered channel when indexing transactions

Refs #3227 (comment)

state/txindex/indexer_service.go

@@ -31,55 +31,39 @@ func NewIndexerService(idr TxIndexer, eventBus *types.EventBus) *IndexerService
 // OnStart implements cmn.Service by subscribing for all transactions
 // and indexing them by tags.
 func (is *IndexerService) OnStart() error {
-	blockHeadersSub, err := is.eventBus.Subscribe(context.Background(), subscriber, types.EventQueryNewBlockHeader)
+	// Use SubscribeUnbuffered here to ensure both subscriptions does not get
+	// cancelled due to not pulling messages fast enough. Cause this might
+	// sometimes happen when there are no other subscribers.
+
+	blockHeadersSub, err := is.eventBus.SubscribeUnbuffered(context.Background(), subscriber, types.EventQueryNewBlockHeader)
 	if err != nil {
 		return err
 	}
 
-	// If there're no other subscribers, goroutine below may not pull messages
-	// fast enough. To solve it, we either need to increase the capacity here or
-	// change pubsub to wait a little bit (say, 10ms) before cancelling the
-	// subscription.
-	txsSub, err := is.eventBus.Subscribe(context.Background(), subscriber, types.EventQueryTx, 100)
+	txsSub, err := is.eventBus.SubscribeUnbuffered(context.Background(), subscriber, types.EventQueryTx)
 	if err != nil {
 		return err
 	}
 
 	go func() {
 		for {
-			select {
-			case msg := <-blockHeadersSub.Out():
-				header := msg.Data().(types.EventDataNewBlockHeader).Header
-				batch := NewBatch(header.NumTxs)
-				for i := int64(0); i < header.NumTxs; i++ {
-					select {
-					case msg2 := <-txsSub.Out():
-						txResult := msg2.Data().(types.EventDataTx).TxResult
-						if err = batch.Add(&txResult); err != nil {
-							is.Logger.Error("Can't add tx to batch",
-								"height", header.Height,
-								"index", txResult.Index,
-								"err", err)
-						}
-					case <-txsSub.Cancelled():
-						is.Logger.Error("Failed to index block. txsSub was cancelled. Did the Tendermint stop?",
-							"height", header.Height,
-							"numTxs", header.NumTxs,
-							"numProcessed", i,
-							"err", txsSub.Err(),
-						)
-						return
-					}
-				}
-				if err = is.idr.AddBatch(batch); err != nil {
-					is.Logger.Error("Failed to index block", "height", header.Height, "err", err)
-				} else {
-					is.Logger.Info("Indexed block", "height", header.Height)
+			msg := <-blockHeadersSub.Out()
+			header := msg.Data().(types.EventDataNewBlockHeader).Header
+			batch := NewBatch(header.NumTxs)
+			for i := int64(0); i < header.NumTxs; i++ {
+				msg2 := <-txsSub.Out()
+				txResult := msg2.Data().(types.EventDataTx).TxResult
+				if err = batch.Add(&txResult); err != nil {
+					is.Logger.Error("Can't add tx to batch",
+						"height", header.Height,
+						"index", txResult.Index,
+						"err", err)
 				}
-			case <-blockHeadersSub.Cancelled():
-				is.Logger.Error("blockHeadersSub was cancelled. Did the Tendermint stop?",
-					"err", blockHeadersSub.Err())
-				return
+			}
+			if err = is.idr.AddBatch(batch); err != nil {
+				is.Logger.Error("Failed to index block", "height", header.Height, "err", err)
+			} else {
+				is.Logger.Info("Indexed block", "height", header.Height)
 			}
 		}
 	}()

types/event_bus.go

@@ -62,6 +62,10 @@ func (b *EventBus) Subscribe(ctx context.Context, subscriber string, query tmpub
 	return b.pubsub.Subscribe(ctx, subscriber, query, outCapacity...)
 }
 
+func (b *EventBus) SubscribeUnbuffered(ctx context.Context, subscriber string, query tmpubsub.Query) (Subscription, error) {
+	return b.pubsub.SubscribeUnbuffered(ctx, subscriber, query)
+}
+
 func (b *EventBus) Unsubscribe(ctx context.Context, subscriber string, query tmpubsub.Query) error {
 	return b.pubsub.Unsubscribe(ctx, subscriber, query)
 }