feat: improve tests, add a ticker

bitcoin-sv · Aug 22, 2024 · a93fe48 · a93fe48
1 parent ca40840
commit a93fe48
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 16 deletions.
diff --git a/internal/blocktx/processor.go b/internal/blocktx/processor.go
@@ -350,7 +350,7 @@ func (ph *Processor) StartProcessRequestTxs() {
 				err = ph.publishMinedTxs(txHashes)
 				if err != nil {
 					ph.logger.Error("failed to publish mined txs", slog.String("err", err.Error()))
-					continue
+					continue // retry, don't clear the txHashes slice
 				}
 
 				txHashes = make([]*chainhash.Hash, 0, ph.registerRequestTxsBatchSize)
@@ -364,7 +364,8 @@ func (ph *Processor) StartProcessRequestTxs() {
 				err := ph.publishMinedTxs(txHashes)
 				if err != nil {
 					ph.logger.Error("failed to publish mined txs", slog.String("err", err.Error()))
-					continue
+					ticker.Reset(ph.registerRequestTxsInterval)
+					continue // retry, don't clear the txHashes slice
 				}
 
 				txHashes = make([]*chainhash.Hash, 0, ph.registerRequestTxsBatchSize)
@@ -674,9 +675,7 @@ func ExtractHeightFromCoinbaseTx(tx *bt.Tx) uint64 {
 }
 
 func (ph *Processor) Shutdown() {
-	if ph.cancelAll != nil {
-		ph.cancelAll()
-	}
+	ph.cancelAll()
 	ph.waitGroup.Wait()
 }
 

diff --git a/internal/blocktx/processor_test.go b/internal/blocktx/processor_test.go
@@ -194,7 +194,6 @@ func TestHandleBlock(t *testing.T) {
 			peerHandler := blocktx.NewPeerHandler(logger, blockRequestCh, blockProcessCh)
 			processor, err := blocktx.NewProcessor(logger, storeMock, blockRequestCh, blockProcessCh, blocktx.WithTransactionBatchSize(batchSize), blocktx.WithMessageQueueClient(mq))
 			require.NoError(t, err)
-			defer processor.Shutdown()
 
 			processor.StartBlockProcessing()
 
@@ -258,6 +257,8 @@ func TestHandleBlock(t *testing.T) {
 			err = peerHandler.HandleBlock(blockMessage, peer)
 			require.NoError(t, err)
 			time.Sleep(20 * time.Millisecond)
+			processor.Shutdown()
+
 			require.ElementsMatch(t, expectedInsertedTransactions, insertedBlockTransactions)
 		})
 	}
@@ -547,7 +548,7 @@ func TestStartProcessRequestTxs(t *testing.T) {
 			getMinedErr: errors.New("get mined error"),
 			requestedTx: testdata.TX1Hash[:],
 
-			expectedGetMinedCalls:     4,
+			expectedGetMinedCalls:     4, // 3 times on the channel message, 1 time on ticker
 			expectedPublishMinedCalls: 0,
 		},
 		{
@@ -556,7 +557,7 @@ func TestStartProcessRequestTxs(t *testing.T) {
 			publishMinedErr: errors.New("publish mined error"),
 			requestedTx:     testdata.TX1Hash[:],
 
-			expectedGetMinedCalls:     4,
+			expectedGetMinedCalls:     4, // 3 times on the channel message, 1 time on ticker
 			expectedPublishMinedCalls: 4,
 		},
 		{
@@ -587,8 +588,6 @@ func TestStartProcessRequestTxs(t *testing.T) {
 
 	for _, tc := range tt {
 		t.Run(tc.name, func(t *testing.T) {
-			publishMinedErrTest := tc.publishMinedErr
-			getMinedErrTest := tc.getMinedErr
 			storeMock := &storeMocks.BlocktxStoreMock{
 				GetMinedTransactionsFunc: func(ctx context.Context, hashes []*chainhash.Hash) ([]store.GetMinedTransactionResult, error) {
 					for _, hash := range hashes {
@@ -599,13 +598,13 @@ func TestStartProcessRequestTxs(t *testing.T) {
 						TxHash:      testdata.TX1Hash[:],
 						BlockHash:   testdata.Block1Hash[:],
 						BlockHeight: 1,
-					}}, getMinedErrTest
+					}}, tc.getMinedErr
 				},
 			}
 
 			mq := &mocks.MessageQueueClientMock{
 				PublishMarshalFunc: func(topic string, m protoreflect.ProtoMessage) error {
-					return publishMinedErrTest
+					return tc.publishMinedErr
 				},
 			}
 
@@ -614,7 +613,7 @@ func TestStartProcessRequestTxs(t *testing.T) {
 			logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo}))
 			processor, err := blocktx.NewProcessor(logger, storeMock,
 				nil, nil,
-				blocktx.WithRegisterRequestTxsInterval(20*time.Millisecond),
+				blocktx.WithRegisterRequestTxsInterval(15*time.Millisecond),
 				blocktx.WithRegisterRequestTxsBatchSize(3),
 				blocktx.WithRequestTxChan(requestTxChannel),
 				blocktx.WithMessageQueueClient(mq))
@@ -624,10 +623,8 @@ func TestStartProcessRequestTxs(t *testing.T) {
 				requestTxChannel <- tc.requestedTx
 			}
 
-			processor.StartProcessRequestTxs()
-
 			// call tested function
-			require.NoError(t, err)
+			processor.StartProcessRequestTxs()
 			time.Sleep(20 * time.Millisecond)
 			processor.Shutdown()