Merge pull request #10 from coinbase/patrick/mempool-support

[Services] Add /mempool support

README.md

@@ -168,7 +168,7 @@ and run one of the following commands:
 
 ## Future Work
 * Publish benchamrks for sync speed, storage usage, and load testing
-* Rosetta API `/mempool/*` implementation
+* [Rosetta API `/mempool/transaction`](https://www.rosetta-api.org/docs/MempoolApi.html#mempooltransaction) implementation
 * Add CI test using `rosetta-cli` to run on each PR (likely on a regtest network)
 * Add performance mode to use unlimited RAM (implementation currently optimized to use <= 16 GB of RAM)
 * Support Multi-Sig Sends

bitcoin/client.go

@@ -78,6 +78,9 @@ const (
 	// https://developer.bitcoin.org/reference/rpc/estimatesmartfee.html
 	requestMethodEstimateSmartFee requestMethod = "estimatesmartfee"
 
+	// https://developer.bitcoin.org/reference/rpc/getrawmempool.html
+	requestMethodRawMempool requestMethod = "getrawmempool"
+
 	// blockNotFoundErrCode is the RPC error code when a block cannot be found
 	blockNotFoundErrCode = -5
 )
@@ -316,6 +319,23 @@ func (b *Client) PruneBlockchain(
 	return response.Result, nil
 }
 
+// RawMempool returns an array of all transaction
+// hashes currently in the mempool.
+func (b *Client) RawMempool(
+	ctx context.Context,
+) ([]string, error) {
+	// Parameters:
+	//   1. verbose
+	params := []interface{}{false}
+
+	response := &rawMempoolResponse{}
+	if err := b.post(ctx, requestMethodRawMempool, params, response); err != nil {
+		return nil, fmt.Errorf("%w: error getting raw mempool", err)
+	}
+
+	return response.Result, nil
+}
+
 // getPeerInfo performs the `getpeerinfo` JSON-RPC request
 func (b *Client) getPeerInfo(
 	ctx context.Context,

bitcoin/client_fixtures/raw_mempool.json

@@ -0,0 +1,9 @@
+{
+  "result": [
+    "9cec12d170e97e21a876fa2789e6bfc25aa22b8a5e05f3f276650844da0c33ab",
+    "37b4fcc8e0b229412faeab8baad45d3eb8e4eec41840d6ac2103987163459e75",
+    "7bbb29ae32117597fcdf21b464441abd571dad52d053b9c2f7204f8ea8c4762e"
+  ],
+  "error": null,
+  "id": "curltest"
+}

bitcoin/client_test.go

@@ -1243,6 +1243,72 @@ func TestSuggestedFeeRate(t *testing.T) {
 	}
 }
 
+func TestRawMempool(t *testing.T) {
+	tests := map[string]struct {
+		responses []responseFixture
+
+		expectedTransactions []string
+		expectedError        error
+	}{
+		"successful": {
+			responses: []responseFixture{
+				{
+					status: http.StatusOK,
+					body:   loadFixture("raw_mempool.json"),
+					url:    url,
+				},
+			},
+			expectedTransactions: []string{
+				"9cec12d170e97e21a876fa2789e6bfc25aa22b8a5e05f3f276650844da0c33ab",
+				"37b4fcc8e0b229412faeab8baad45d3eb8e4eec41840d6ac2103987163459e75",
+				"7bbb29ae32117597fcdf21b464441abd571dad52d053b9c2f7204f8ea8c4762e",
+			},
+		},
+		"500 error": {
+			responses: []responseFixture{
+				{
+					status: http.StatusInternalServerError,
+					body:   "{}",
+					url:    url,
+				},
+			},
+			expectedError: errors.New("invalid response: 500 Internal Server Error"),
+		},
+	}
+
+	for name, test := range tests {
+		t.Run(name, func(t *testing.T) {
+			var (
+				assert = assert.New(t)
+			)
+
+			responses := make(chan responseFixture, len(test.responses))
+			for _, response := range test.responses {
+				responses <- response
+			}
+
+			ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				response := <-responses
+				assert.Equal("application/json", r.Header.Get("Content-Type"))
+				assert.Equal("POST", r.Method)
+				assert.Equal(response.url, r.URL.RequestURI())
+
+				w.WriteHeader(response.status)
+				fmt.Fprintln(w, response.body)
+			}))
+
+			client := NewClient(ts.URL, MainnetGenesisBlockIdentifier, MainnetCurrency)
+			txs, err := client.RawMempool(context.Background())
+			if test.expectedError != nil {
+				assert.Contains(err.Error(), test.expectedError.Error())
+			} else {
+				assert.NoError(err)
+				assert.Equal(test.expectedTransactions, txs)
+			}
+		})
+	}
+}
+
 // loadFixture takes a file name and returns the response fixture.
 func loadFixture(fileName string) string {
 	content, err := ioutil.ReadFile(fmt.Sprintf("client_fixtures/%s", fileName))

bitcoin/types.go

@@ -478,6 +478,25 @@ func (s suggestedFeeRateResponse) Err() error {
 	)
 }
 
+// rawMempoolResponse is the response body for `getrawmempool` requests.
+type rawMempoolResponse struct {
+	Result []string       `json:"result"`
+	Error  *responseError `json:"error"`
+}
+
+func (r rawMempoolResponse) Err() error {
+	if r.Error == nil {
+		return nil
+	}
+
+	return fmt.Errorf(
+		"%w: error JSON RPC response, code: %d, message: %s",
+		ErrJSONRPCError,
+		r.Error.Code,
+		r.Error.Message,
+	)
+}
+
 // CoinIdentifier converts a tx hash and vout into
 // the canonical CoinIdentifier.Identifier used in
 // rosetta-bitcoin.

mocks/services/client.go

@@ -38,6 +38,29 @@ func (_m *Client) NetworkStatus(_a0 context.Context) (*types.NetworkStatusRespon
 	return r0, r1
 }
 
+// RawMempool provides a mock function with given fields: _a0
+func (_m *Client) RawMempool(_a0 context.Context) ([]string, error) {
+	ret := _m.Called(_a0)
+
+	var r0 []string
+	if rf, ok := ret.Get(0).(func(context.Context) []string); ok {
+		r0 = rf(_a0)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]string)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context) error); ok {
+		r1 = rf(_a0)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // SendRawTransaction provides a mock function with given fields: _a0, _a1
 func (_m *Client) SendRawTransaction(_a0 context.Context, _a1 string) (string, error) {
 	ret := _m.Called(_a0, _a1)

services/mempool_service.go

@@ -22,11 +22,15 @@ import (
 )
 
 // MempoolAPIService implements the server.MempoolAPIServicer interface.
-type MempoolAPIService struct{}
+type MempoolAPIService struct {
+	client Client
+}
 
 // NewMempoolAPIService creates a new instance of a MempoolAPIService.
-func NewMempoolAPIService() server.MempoolAPIServicer {
+func NewMempoolAPIService(client Client) server.MempoolAPIServicer {
-	return &MempoolAPIService{}
+	return &MempoolAPIService{
+		client: client,
+	}
 }
 
 // Mempool implements the /mempool endpoint.
@@ -34,7 +38,19 @@ func (s *MempoolAPIService) Mempool(
 	ctx context.Context,
 	request *types.NetworkRequest,
 ) (*types.MempoolResponse, *types.Error) {
-	return nil, wrapErr(ErrUnimplemented, nil)
+	mempoolTransactions, err := s.client.RawMempool(ctx)
+	if err != nil {
+		return nil, wrapErr(ErrBitcoind, err)
+	}
+
+	transactionIdentifiers := make([]*types.TransactionIdentifier, len(mempoolTransactions))
+	for i, mempoolTransaction := range mempoolTransactions {
+		transactionIdentifiers[i] = &types.TransactionIdentifier{Hash: mempoolTransaction}
+	}
+
+	return &types.MempoolResponse{
+		TransactionIdentifiers: transactionIdentifiers,
+	}, nil
 }
 
 // MempoolTransaction implements the /mempool/transaction endpoint.

services/mempool_service_test.go

@@ -18,20 +18,37 @@ import (
 	"context"
 	"testing"
 
+	mocks "github.com/coinbase/rosetta-bitcoin/mocks/services"
+
+	"github.com/coinbase/rosetta-sdk-go/types"
 	"github.com/stretchr/testify/assert"
 )
 
 func TestMempoolEndpoints(t *testing.T) {
-	servicer := NewMempoolAPIService()
+	mockClient := &mocks.Client{}
+	servicer := NewMempoolAPIService(mockClient)
 	ctx := context.Background()
 
+	mockClient.On("RawMempool", ctx).Return([]string{
+		"tx1",
+		"tx2",
+	}, nil)
 	mem, err := servicer.Mempool(ctx, nil)
-	assert.Nil(t, mem)
+	assert.Nil(t, err)
-	assert.Equal(t, ErrUnimplemented.Code, err.Code)
+	assert.Equal(t, &types.MempoolResponse{
-	assert.Equal(t, ErrUnimplemented.Message, err.Message)
+		TransactionIdentifiers: []*types.TransactionIdentifier{
+			{
+				Hash: "tx1",
+			},
+			{
+				Hash: "tx2",
+			},
+		},
+	}, mem)
 
 	memTransaction, err := servicer.MempoolTransaction(ctx, nil)
 	assert.Nil(t, memTransaction)
 	assert.Equal(t, ErrUnimplemented.Code, err.Code)
 	assert.Equal(t, ErrUnimplemented.Message, err.Message)
+	mockClient.AssertExpectations(t)
 }

services/router.go

@@ -55,7 +55,7 @@ func NewBlockchainRouter(
 		asserter,
 	)
 
-	mempoolAPIService := NewMempoolAPIService()
+	mempoolAPIService := NewMempoolAPIService(client)
 	mempoolAPIController := server.NewMempoolAPIController(
 		mempoolAPIService,
 		asserter,

services/types.go

@@ -28,6 +28,7 @@ type Client interface {
 	NetworkStatus(context.Context) (*types.NetworkStatusResponse, error)
 	SendRawTransaction(context.Context, string) (string, error)
 	SuggestedFeeRate(context.Context, int64) (float64, error)
+	RawMempool(context.Context) ([]string, error)
 }
 
 // Indexer is used by the servicers to get block and account data.