feat: FLY-2237 bitcoin/rootstock peer watchers

Author: volodymyrzahanych

.mockery.yaml

@@ -115,6 +115,7 @@ packages:
     interfaces:
       EclipseCheckUseCase:
       UpdateBtcReleaseUseCase:
+      NodePeerAlertUseCase:
   github.com/rsksmart/liquidity-provider-server/internal/adapters/entrypoints/watcher/monitoring:
     interfaces:
       GetAssetReportUseCase:

cmd/application/lps/application.go

@@ -172,6 +172,8 @@ func (app *Application) prepareWatchers() ([]watcher.Watcher, error) {
 		app.watcherRegistry.PenalizationAlertWatcher,
 		app.watcherRegistry.PegoutBridgeWatcher,
 		app.watcherRegistry.BtcReleaseWatcher,
+		app.watcherRegistry.BitcoinPeerWatcher,
+		app.watcherRegistry.RootstockPeerWatcher,
 		app.watcherRegistry.QuoteMetricsWatcher,
 		app.watcherRegistry.AssetReportWatcher,
 	}

docker-compose/local/docker-compose.lps.yml

@@ -87,6 +87,8 @@ services:
       - BTC_RELEASE_WATCHER_START_BLOCK
       - BTC_RELEASE_WATCHER_PAGE_SIZE
       - BTC_RELEASE_CHECK_TIMEOUT
+      - RSK_MIN_PEERS
+      - BITCOIN_MIN_PEERS
     ports:
       - "8080:8080"
     volumes:

internal/adapters/dataproviders/bitcoin/btcclient/binding.go

@@ -50,6 +50,7 @@ type RpcClient interface {
 	CreateWallet(name string, opts ...rpcclient.CreateWalletOpt) (*btcjson.CreateWalletResult, error)
 	LoadWallet(walletName string) (*btcjson.LoadWalletResult, error)
 	EstimateSmartFee(confTarget int64, mode *btcjson.EstimateSmartFeeMode) (*btcjson.EstimateSmartFeeResult, error)
+	GetConnectionCount() (int64, error)
 }
 
 type ClientAdapter interface {

internal/adapters/dataproviders/bitcoin/mempool_space/api.go

@@ -470,6 +470,10 @@ func (api *MempoolSpaceApi) GetZeroAddress(addressType blockchain.BtcAddressType
 	return "", errors.New("not supported in MempoolSpace API")
 }
 
+func (api *MempoolSpaceApi) GetConnectionCount() (int64, error) {
+	return 0, errors.New("GetConnectionCount is not supported in MempoolSpace API")
+}
+
 func (api *MempoolSpaceApi) getBlock(blockHash string) (*btcutil.Block, error) {
 	req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, fmt.Sprintf("%s/block/%s/raw", api.url, blockHash), nil)
 	if err != nil {

internal/adapters/dataproviders/bitcoin/rpc.go

@@ -261,6 +261,10 @@ func (rpc *bitcoindRpc) GetZeroAddress(addressType blockchain.BtcAddressType) (s
 	return addresses.Address(addressType)
 }
 
+func (rpc *bitcoindRpc) GetConnectionCount() (int64, error) {
+	return rpc.conn.client.GetConnectionCount()
+}
+
 func (rpc *bitcoindRpc) getTxBlock(txHash string) (*wire.MsgBlock, *chainhash.Hash, error) {
 	parsedTxHash, err := chainhash.NewHashFromStr(txHash)
 	if err != nil {

internal/adapters/dataproviders/bitcoin/rpc_test.go

@@ -707,3 +707,23 @@ func TestBitcoindRpc_GetZeroAddress(t *testing.T) {
 		assert.Empty(t, result)
 	})
 }
+
+func TestBitcoindRpc_GetConnectionCount(t *testing.T) {
+	t.Run("Should return connection count", func(t *testing.T) {
+		client := &mocks.ClientAdapterMock{}
+		client.EXPECT().GetConnectionCount().Return(int64(8), nil).Once()
+		rpc := bitcoin.NewBitcoindRpc(bitcoin.NewConnection(&chaincfg.MainNetParams, client))
+		count, err := rpc.GetConnectionCount()
+		require.NoError(t, err)
+		assert.Equal(t, int64(8), count)
+		client.AssertExpectations(t)
+	})
+	t.Run("Should return error when client fails", func(t *testing.T) {
+		client := &mocks.ClientAdapterMock{}
+		client.EXPECT().GetConnectionCount().Return(int64(0), assert.AnError).Once()
+		rpc := bitcoin.NewBitcoindRpc(bitcoin.NewConnection(&chaincfg.MainNetParams, client))
+		_, err := rpc.GetConnectionCount()
+		require.Error(t, err)
+		client.AssertExpectations(t)
+	})
+}

internal/adapters/dataproviders/rootstock/bindings.go

@@ -29,6 +29,7 @@ type RpcClientBinding interface {
 	BlockNumber(ctx context.Context) (uint64, error)
 	BlockByHash(ctx context.Context, hash common.Hash) (*types.Block, error)
 	BlockByNumber(ctx context.Context, number *big.Int) (*types.Block, error)
+	PeerCount(ctx context.Context) (uint64, error)
 }
 
 type RskBridgeBinding interface {

internal/adapters/dataproviders/rootstock/rpc.go

@@ -193,6 +193,13 @@ func (rpc *rskjRpcServer) GetBlockByNumber(ctx context.Context, blockNumber *big
 	}, nil
 }
 
+func (rpc *rskjRpcServer) PeerCount(ctx context.Context) (uint64, error) {
+	return rskRetry(rpc.retryParams.Retries, rpc.retryParams.Sleep,
+		func() (uint64, error) {
+			return rpc.client.PeerCount(ctx)
+		})
+}
+
 func (rpc *rskjRpcServer) ChainId(ctx context.Context) (uint64, error) {
 	result, err := rskRetry(rpc.retryParams.Retries, rpc.retryParams.Sleep,
 		func() (uint64, error) {

internal/adapters/dataproviders/rootstock/rpc_test.go

@@ -469,3 +469,24 @@ func TestRskjRpcServer_GetBlockByNumber(t *testing.T) {
 		client.AssertExpectations(t)
 	})
 }
+
+func TestRskjRpcServer_PeerCount(t *testing.T) {
+	t.Run("should return peer count", func(t *testing.T) {
+		client := &mocks.RpcClientBindingMock{}
+		client.EXPECT().PeerCount(mock.Anything).Return(uint64(5), nil)
+		rpc := rootstock.NewRskjRpcServer(rootstock.NewRskClient(client), rootstock.RetryParams{})
+		count, err := rpc.PeerCount(context.Background())
+		require.NoError(t, err)
+		assert.Equal(t, uint64(5), count)
+		client.AssertExpectations(t)
+	})
+	t.Run("should handle errors", func(t *testing.T) {
+		client := &mocks.RpcClientBindingMock{}
+		client.EXPECT().PeerCount(mock.Anything).Return(uint64(0), assert.AnError)
+		rpc := rootstock.NewRskjRpcServer(rootstock.NewRskClient(client), rootstock.RetryParams{})
+		count, err := rpc.PeerCount(context.Background())
+		require.Error(t, err)
+		assert.Equal(t, uint64(0), count)
+		client.AssertExpectations(t)
+	})
+}