Merge pull request #310 from bitcoin-sv/fix/prometheus-server

Fix/prometheus server

CHANGELOG.md

@@ -30,7 +30,7 @@ All notable changes to this project will be documented in this file. The format
 
 ### Added
 
-- Callbacks for status `SEEN_ON_NETWORK` and `SEEN_IN_ORPHAN_MEMPOOL` if new request header `X-SkipFeeValidation` is given with value `true`.
+- Callbacks for status `SEEN_ON_NETWORK` and `SEEN_IN_ORPHAN_MEMPOOL` if new request header `X-FullStatusUpdates` is given with value `true`.
 
 ## [1.0.62] - 2023-11-23
 

config.yaml

@@ -4,6 +4,7 @@ logFormat: text # format of logging. Value can be one of text | json | tint
 profilerAddr: localhost:9999 # address to start profiler server on
 statisticsServerAddress: localhost:9005 # address to start statistics server on
 prometheusEndpoint: /metrics # endpoint for prometheus metrics
+prometheusAddr: :2112
 tracing: true # enable trancing
 grpcMessageSize: 100000000 # maximum grpc message size
 network: regtest # bitcoin network to connect to

main.go

@@ -119,10 +119,15 @@ func run() error {
 		}
 	}()
 
+	prometheusAddr := viper.GetString("prometheusAddr")
 	prometheusEndpoint := viper.GetString("prometheusEndpoint")
-	if prometheusEndpoint != "" {
+	if prometheusEndpoint != "" && prometheusAddr != "" {
 		logger.Info("Starting prometheus", slog.String("endpoint", prometheusEndpoint))
 		http.Handle(prometheusEndpoint, promhttp.Handler())
+		err = http.ListenAndServe(prometheusAddr, nil)
+		if err != nil {
+			logger.Error("failed to start prometheus server", slog.String("err", err.Error()))
+		}
 	}
 
 	tracingOn := viper.GetBool("tracing")

metamorph/processor.go

@@ -327,7 +327,7 @@ func (p *Processor) processExpiredTransactions() {
 
 		// Before re-requesting/re-announcing txs check if they had been mined or seen in mempool in the meantime
 		minedOrSeen, err := p.store.GetMinedOrSeen(context.Background(), hashSlice)
-		if err == nil {
+		if err == nil && len(minedOrSeen) > 0 {
 			// if tx has been mined or seen in the meantime delete both from processor response map and expired transactions
 			p.logger.Info("found mined or seen txs", slog.Int("number", len(minedOrSeen)))
 

test/double_spend_test.go

@@ -49,17 +49,17 @@ func TestDoubleSpend(t *testing.T) {
 			tx, err := createTx(privateKey, address, utxos[0])
 			require.NoError(t, err)
 
-			url := "http://arc:9090/"
+			url := arcEndpoint
 
 			arcClient, err := api.NewClientWithResponses(url)
 			require.NoError(t, err)
 
 			//submit first transaction
-			postTxChecksStatus(t, arcClient, tx, "SEEN_ON_NETWORK", tc.extFormat)
+			postTxChecksStatus(t, arcClient, tx, metamorph_api.Status_SEEN_ON_NETWORK, tc.extFormat)
 
 			// send double spending transaction when first tx is in mempool
 			txMempool := createTxToNewAddress(t, privateKey, utxos[0])
-			postTxChecksStatus(t, arcClient, txMempool, "REJECTED", tc.extFormat)
+			postTxChecksStatus(t, arcClient, txMempool, metamorph_api.Status_REJECTED, tc.extFormat)
 
 			generate(t, 10)
 
@@ -71,12 +71,12 @@ func TestDoubleSpend(t *testing.T) {
 
 			// send double spending transaction when first tx was mined
 			txMined := createTxToNewAddress(t, privateKey, utxos[0])
-			postTxChecksStatus(t, arcClient, txMined, "SEEN_IN_ORPHAN_MEMPOOL", tc.extFormat)
+			postTxChecksStatus(t, arcClient, txMined, metamorph_api.Status_SEEN_IN_ORPHAN_MEMPOOL, tc.extFormat)
 		})
 	}
 }
 
-func postTxChecksStatus(t *testing.T, client *api.ClientWithResponses, tx *bt.Tx, expectedStatus string, extFormat bool) {
+func postTxChecksStatus(t *testing.T, client *api.ClientWithResponses, tx *bt.Tx, expectedStatus metamorph_api.Status, extFormat bool) {
 	rawTxString := hex.EncodeToString(tx.Bytes())
 	if extFormat {
 		rawTxString = hex.EncodeToString(tx.ExtendedBytes())
@@ -86,7 +86,7 @@ func postTxChecksStatus(t *testing.T, client *api.ClientWithResponses, tx *bt.Tx
 	}
 
 	ctx := context.Background()
-	waitForStatus := api.WaitForStatus(metamorph_api.Status_SEEN_ON_NETWORK)
+	waitForStatus := api.WaitForStatus(expectedStatus)
 	params := &api.POSTTransactionParams{
 		XWaitForStatus: &waitForStatus,
 	}
@@ -95,7 +95,7 @@ func postTxChecksStatus(t *testing.T, client *api.ClientWithResponses, tx *bt.Tx
 
 	require.Equal(t, http.StatusOK, response.StatusCode())
 	require.NotNil(t, response.JSON200)
-	require.Equalf(t, expectedStatus, response.JSON200.TxStatus, "status of response: %s does not match expected status: %s for tx ID %s", response.JSON200.TxStatus, expectedStatus, tx.TxID())
+	require.Equalf(t, expectedStatus.String(), response.JSON200.TxStatus, "status of response: %s does not match expected status: %s for tx ID %s", response.JSON200.TxStatus, expectedStatus.String(), tx.TxID())
 }
 
 func createTxToNewAddress(t *testing.T, privateKey string, utxo NodeUnspentUtxo) *bt.Tx {

test/endpoint_test.go

@@ -30,6 +30,12 @@ import (
 
 const (
 	feeSat = 10
+
+	arcEndpoint      = "http://arc:9090/"
+	v1Tx             = "v1/tx"
+	v1Txs            = "v1/txs"
+	arcEndpointV1Tx  = arcEndpoint + v1Tx
+	arcEndpointV1Txs = arcEndpoint + v1Txs
 )
 
 type Response struct {
@@ -162,10 +168,8 @@ func TestBatchChainedTxs(t *testing.T) {
 
 			buffer := bytes.NewBuffer(payLoad)
 
-			url := "http://arc:9090/v1/txs"
-
 			// Send POST request
-			req, err := http.NewRequest("POST", url, buffer)
+			req, err := http.NewRequest("POST", arcEndpointV1Txs, buffer)
 			require.NoError(t, err)
 
 			req.Header.Set("Content-Type", "application/json")
@@ -292,9 +296,7 @@ func TestPostCallbackToken(t *testing.T) {
 			tx, err := createTx(privateKey, address, utxos[0])
 			require.NoError(t, err)
 
-			url := "http://arc:9090/"
-
-			arcClient, err := api.NewClientWithResponses(url)
+			arcClient, err := api.NewClientWithResponses(arcEndpoint)
 			require.NoError(t, err)
 
 			ctx := context.Background()
@@ -392,7 +394,7 @@ func TestPostCallbackToken(t *testing.T) {
 
 			generate(t, 10)
 
-			time.Sleep(1 * time.Second) // give ARC time to perform the status update on DB
+			time.Sleep(5 * time.Second) // give ARC time to perform the status update on DB
 
 			var statusResponse *api.GETTransactionStatusResponse
 			statusResponse, err = arcClient.GETTransactionStatusWithResponse(ctx, response.JSON200.Txid)
@@ -412,7 +414,6 @@ func TestPostCallbackToken(t *testing.T) {
 					require.Equal(t, statusResponse.JSON200.Txid, callback.Txid)
 					require.Equal(t, *statusResponse.JSON200.BlockHeight, *callback.BlockHeight)
 					require.Equal(t, *statusResponse.JSON200.BlockHash, *callback.BlockHash)
-					require.Equal(t, *statusResponse.JSON200.TxStatus, *callback.TxStatus)
 					require.Equal(t, "MINED", *callback.TxStatus)
 					require.NotNil(t, statusResponse.JSON200.MerklePath)
 					_, err = bc.NewBUMPFromStr(*statusResponse.JSON200.MerklePath)
@@ -464,7 +465,7 @@ func TestPostSkipFee(t *testing.T) {
 
 			fmt.Println("Transaction with Zero fee:", tx)
 
-			url := "http://arc:9090/"
+			url := arcEndpoint
 
 			arcClient, err := api.NewClientWithResponses(url)
 			require.NoError(t, err)
@@ -530,7 +531,7 @@ func TestPostSkipTxValidation(t *testing.T) {
 
 			fmt.Println("Transaction with Zero fee:", tx)
 
-			url := "http://arc:9090/"
+			url := arcEndpoint
 
 			arcClient, err := api.NewClientWithResponses(url)
 			require.NoError(t, err)
@@ -587,10 +588,9 @@ func Test_E2E_Success(t *testing.T) {
 
 	tx := createTxHexStringExtended(t)
 	jsonPayload := fmt.Sprintf(`{"rawTx": "%s"}`, hex.EncodeToString(tx.ExtendedBytes()))
-	url := "http://arc:9090/v1/tx"
 
 	// Send POST request
-	req, err := http.NewRequest("POST", url, strings.NewReader(jsonPayload))
+	req, err := http.NewRequest("POST", arcEndpointV1Tx, strings.NewReader(jsonPayload))
 	require.NoError(t, err)
 
 	req.Header.Set("Content-Type", "application/json")
@@ -604,14 +604,14 @@ func Test_E2E_Success(t *testing.T) {
 	require.Equal(t, txID, txIDRepeat)
 
 	// Check transaction status
-	statusUrl := fmt.Sprintf("http://arc:9090/v1/tx/%s", txID)
+	statusUrl := fmt.Sprintf("%s/%s", arcEndpointV1Tx, txID)
 	statusResp, err := http.Get(statusUrl)
 	require.NoError(t, err)
 	defer statusResp.Body.Close()
 
 	var statusResponse TxStatusResponse
 	require.NoError(t, json.NewDecoder(statusResp.Body).Decode(&statusResponse))
-	require.Equal(t, "SEEN_ON_NETWORK", statusResponse.TxStatus, "Expected txStatus to be 'SEEN_ON_NETWORK'")
+	require.Equalf(t, "SEEN_ON_NETWORK", statusResponse.TxStatus, "Expected txStatus to be 'SEEN_ON_NETWORK' for tx id %s", txID)
 
 	t.Logf("Transaction status: %s", statusResponse.TxStatus)
 
@@ -698,8 +698,7 @@ func TestPostTx_BadRequestBodyFormat(t *testing.T) {
 }
 
 func postTx(t *testing.T, jsonPayload string, headers map[string]string) (*http.Response, error) {
-	url := "http://arc:9090/v1/tx"
-	req, err := http.NewRequest("POST", url, strings.NewReader(jsonPayload))
+	req, err := http.NewRequest("POST", arcEndpointV1Tx, strings.NewReader(jsonPayload))
 	if err != nil {
 		t.Fatalf("Error creating HTTP request: %s", err)
 	}

tracing/peer_handler_collector.go

@@ -19,19 +19,6 @@ type PeerHandlerStats struct {
 	BlockProcessingMs       atomic.Uint64
 }
 
-func NewPeerHandlerStats() *PeerHandlerStats {
-	return &PeerHandlerStats{
-		TransactionSent:         atomic.Uint64{},
-		TransactionAnnouncement: atomic.Uint64{},
-		TransactionRejection:    atomic.Uint64{},
-		TransactionGet:          atomic.Uint64{},
-		Transaction:             atomic.Uint64{},
-		BlockAnnouncement:       atomic.Uint64{},
-		Block:                   atomic.Uint64{},
-		BlockProcessingMs:       atomic.Uint64{},
-	}
-}
-
 type PeerHandlerCollector struct {
 	service                 string
 	stats                   *safemap.Safemap[string, *PeerHandlerStats]