Improve Lightnodes to use Phi, support caching requests and more

.circleci/config.yml

@@ -1,6 +1,6 @@
 version: 2.1
 orbs:
-  core: ren/circleci-orbs@dev:first
+  core: ren/core@0.0.1
 executors:
   go_exec:
     docker:
@@ -12,7 +12,34 @@ jobs:
       - checkout
       - restore_cache: # Restore saved cache if no changes are detected since last run
           key: go-mod-v1-{{ checksum "go.sum" }}
-      - run: go test -v ./...
+      - run:
+          name: Install tools
+          command: |
+            go get -u golang.org/x/lint/golint
+            go get -u github.com/onsi/ginkgo/ginkgo
+            go get -u github.com/loongy/covermerge
+            go get -u github.com/mattn/goveralls
+      - run:
+          name: Run tests
+          command: |
+            # make sure all the tests pass
+            go test -v ./...
+            # all the tests passed so update coverage
+            CI=true /go/bin/ginkgo -v --race --cover --coverprofile coverprofile.out ./...
+            /go/bin/covermerge \
+            validator/coverprofile.out \
+            coverprofile.out \
+            dispatcher/coverprofile.out \
+            server/ratelimiter/coverprofile.out \
+            server/coverprofile.out \
+            updater/coverprofile.out \
+            client/coverprofile.out \
+            cacher/coverprofile.out \
+            store/coverprofile.out > covermerge.out
+            goveralls -coverprofile=covermerge.out -service=circleci
+      - run:
+          name: Run linter
+          command: golint ./...
       - save_cache:
           key: go-mod-v1-{{ checksum "go.sum" }}
           paths:
@@ -42,4 +69,4 @@ workflows:
                 - master
     jobs:
       - core/merge_beta:
-          executor: go_exec
+          executor: go_exec

.gitignore

@@ -0,0 +1,7 @@
+# Test coverage
+*.coverprofile
+*.out
+*.html
+
+# Environment variables
+.env

README.md

@@ -1,6 +1,7 @@
 # `🚀 lightnode`
 
-[![CircleCI](https://circleci.com/gh/renproject/lightnode/tree/master.svg?style=svg)](https://circleci.com/gh/renproject/lightnode/tree/master)
+[![CircleCI](https://circleci.com/gh/renproject/lightnode/tree/master.svg?style=shield)](https://circleci.com/gh/renproject/lightnode/tree/master)
+[![Coverage Status](https://coveralls.io/repos/github/renproject/lightnode/badge.svg?branch=master)](https://coveralls.io/github/renproject/lightnode?branch=master)
 
 A node used for querying Darknodes using JSON-RPC 2.0 interfaces. Featuring query caching (for performance) as well as retrying for failed requests (for reliability).
 

cacher/cacher.go

@@ -0,0 +1,125 @@
+package cacher
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/renproject/darknode/jsonrpc"
+	"github.com/renproject/kv"
+	"github.com/renproject/lightnode/server"
+	"github.com/renproject/phi"
+	"github.com/sirupsen/logrus"
+	"golang.org/x/crypto/sha3"
+)
+
+// ID is a key for a cached response.
+type ID [32]byte
+
+func (id ID) String() string {
+	return string(id[:32])
+}
+
+// Cacher is a task responsible for caching responses for corresponding
+// requests. Upon receiving a request (in the current architecture this request
+// comes from the `Validator`) it will check its cache to see if it has a
+// cached response. If it does, it will write this immediately as a repsonse,
+// otherwise it will forward the request on to the `Dispatcher`. Once the
+// `Dispatcher` has a response ready, the `Cacher` will store this response in
+// its cache with a key derived from the request, and then pass the repsonse
+// along to be given to the client. Currently, idempotent requests are stored
+// in a LRU cache, and non-idempotent requests are stored in a TTL cache.
+type Cacher struct {
+	logger     logrus.FieldLogger
+	dispatcher phi.Sender
+
+	ttlCache kv.Iterable
+}
+
+// New constructs a new `Cacher` as a `phi.Task` which can be `Run()`.
+func New(dispatcher phi.Sender, logger logrus.FieldLogger, cap int, ttl time.Duration, opts phi.Options) phi.Task {
+	ttlCache, err := kv.NewTTLCache(kv.NewJSON(kv.NewMemDB()), ttl)
+	if err != nil {
+		logger.Panicf("[cacher] cannot create TTL cache: %v", err)
+	}
+	return phi.New(&Cacher{logger, dispatcher, ttlCache}, opts)
+}
+
+// Handle implements the `phi.Handler` interface.
+func (cacher *Cacher) Handle(_ phi.Task, message phi.Message) {
+	msg, ok := message.(server.RequestWithResponder)
+	if !ok {
+		cacher.logger.Panicf("[cacher] unexpected message type %T", message)
+	}
+
+	params, err := msg.Request.Params.MarshalJSON()
+	if err != nil {
+		cacher.logger.Errorf("[cacher] cannot marshal request to json: %v", err)
+	}
+
+	data := append(params, []byte(msg.Request.Method)...)
+	reqID := hash(data)
+
+	cachable := isCachable(msg.Request.Method)
+	response, cached := cacher.get(reqID, msg.DarknodeID)
+	if cachable && cached {
+		msg.Responder <- response
+	} else {
+		responder := make(chan jsonrpc.Response, 1)
+		cacher.dispatcher.Send(server.RequestWithResponder{
+			Request:    msg.Request,
+			Responder:  responder,
+			DarknodeID: msg.DarknodeID,
+		})
+
+		// TODO: What do we do when a second request comes in that is already
+		// being fetched at the moment? Currently it will also send it to the
+		// dispatcher, which is probably not ideal.
+		go func() {
+			response := <-responder
+			// TODO: Consider thread safety of insertion.
+			cacher.insert(reqID, msg.DarknodeID, response, msg.Request.Method)
+			msg.Responder <- response
+		}()
+	}
+}
+
+func (cacher *Cacher) insert(reqID ID, darknodeID string, response jsonrpc.Response, method string) {
+	id := reqID.String() + darknodeID
+	if err := cacher.ttlCache.Insert(id, response); err != nil {
+		cacher.logger.Panicf("[cacher] could not insert response into TTL cache: %v", err)
+	}
+}
+
+func (cacher *Cacher) get(reqID ID, darknodeID string) (jsonrpc.Response, bool) {
+	id := reqID.String() + darknodeID
+
+	var response jsonrpc.Response
+	if err := cacher.ttlCache.Get(id, &response); err == nil {
+		return response, true
+	}
+
+	return jsonrpc.Response{}, false
+}
+
+func isCachable(method string) bool {
+	switch method {
+	case jsonrpc.MethodQueryBlock,
+		jsonrpc.MethodQueryBlocks,
+		jsonrpc.MethodQueryNumPeers,
+		jsonrpc.MethodQueryPeers,
+		jsonrpc.MethodQueryEpoch,
+		jsonrpc.MethodQueryStat:
+		return true
+	case jsonrpc.MethodSubmitTx,
+		jsonrpc.MethodQueryTx:
+		// TODO: We need to make sure these are the only methods that we want to
+		// avoid caching.
+		return false
+	default:
+		panic(fmt.Sprintf("[cacher] unsupported method %s encountered which should have been rejected by the previous checks", method))
+	}
+}
+
+func hash(data []byte) ID {
+	return sha3.Sum256(data)
+}

cacher/cacher_suite_test.go

@@ -0,0 +1,13 @@
+package cacher_test
+
+import (
+	"testing"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+func TestCacher(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Cacher Suite")
+}

cacher/cacher_test.go

@@ -0,0 +1,108 @@
+package cacher_test
+
+import (
+	"context"
+	"time"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	"github.com/renproject/darknode/jsonrpc"
+	"github.com/renproject/lightnode/cacher"
+	"github.com/renproject/lightnode/server"
+	"github.com/renproject/lightnode/testutils"
+	"github.com/renproject/phi"
+	"github.com/sirupsen/logrus"
+)
+
+func initCacher(ctx context.Context, cacheCap int, ttl time.Duration) (phi.Sender, <-chan phi.Message) {
+	opts := phi.Options{Cap: 10}
+	logger := logrus.New()
+	inspector, messages := testutils.NewInspector(10)
+	cacher := cacher.New(inspector, logger, cacheCap, ttl, opts)
+
+	go cacher.Run(ctx)
+	go inspector.Run(ctx)
+
+	return cacher, messages
+}
+
+var _ = Describe("Cacher", func() {
+	Context("When receving a request that does not have a response in the cache", func() {
+		It("Should pass the request through", func() {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			cacher, messages := initCacher(ctx, 10, time.Second)
+
+			for method, _ := range jsonrpc.RPCs {
+				// TODO: This method is not supported right now, but when it is
+				// this case should be tested too.
+				if method == jsonrpc.MethodQueryEpoch {
+					continue
+				}
+
+				request := testutils.ValidRequest(method)
+				cacher.Send(server.NewRequestWithResponder(request, ""))
+
+				select {
+				case <-time.After(time.Second):
+					Fail("timeout")
+				case message := <-messages:
+					req, ok := message.(server.RequestWithResponder)
+					Expect(ok).To(BeTrue())
+					Expect(req.Request).To(Equal(request))
+					Expect(req.Responder).To(Not(BeNil()))
+					Eventually(req.Responder).ShouldNot(Receive())
+				}
+			}
+		})
+	})
+
+	Context("when receiving a request that has a response in the cache", func() {
+		It("Should return the cached response", func() {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			cacher, messages := initCacher(ctx, 10, 1*time.Second)
+
+			for method, _ := range jsonrpc.RPCs {
+				// TODO: This method is not supported right now, but when it is
+				// this case should be tested too.
+				if method == jsonrpc.MethodQueryEpoch {
+					continue
+				}
+
+				// We have disabled caching for these methods.
+				if method == jsonrpc.MethodSubmitTx || method == jsonrpc.MethodQueryTx {
+					continue
+				}
+
+				// Send the first request
+				request := testutils.ValidRequest(method)
+				reqWithRes := server.NewRequestWithResponder(request, "")
+				cacher.Send(reqWithRes)
+				forwardedReq := <-messages
+				res := testutils.ErrorResponse(request.ID)
+				forwardedReq.(server.RequestWithResponder).Responder <- res
+
+				select {
+				case <-time.After(time.Second):
+					Fail("timeout")
+				case response := <-reqWithRes.Responder:
+					Expect(response).To(Equal(res))
+				}
+
+				// Send the second request and expect a cached response
+				request = testutils.ValidRequest(method)
+				reqWithRes = server.NewRequestWithResponder(request, "")
+				cacher.Send(reqWithRes)
+
+				select {
+				case <-time.After(time.Second):
+					Fail("timeout")
+				case response := <-reqWithRes.Responder:
+					Expect(response).To(Equal(res))
+					Eventually(messages).ShouldNot(Receive())
+				}
+			}
+		})
+	})
+})

client/client.go

@@ -0,0 +1,67 @@
+package client
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/renproject/darknode/addr"
+	"github.com/renproject/darknode/jsonrpc"
+)
+
+// SendToDarknode sends a given request to the darknode at the given url. The
+// timeout is the timeout for the request.
+//
+// NOTE: If err is not nil, it is expected that the caller will construct an
+// appropriate error response message.
+func SendToDarknode(url string, req jsonrpc.Request, timeout time.Duration) (jsonrpc.Response, error) {
+	httpClient := new(http.Client)
+	httpClient.Timeout = timeout
+
+	// Construct HTTP request.
+	body, err := json.Marshal(req)
+	if err != nil {
+		return jsonrpc.Response{}, fmt.Errorf("[client] could not marshal request: %v", err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), httpClient.Timeout)
+	defer cancel()
+	r, err := http.NewRequest(http.MethodPost, url, bytes.NewBuffer(body))
+	if err != nil {
+		return jsonrpc.Response{}, fmt.Errorf("[client] could not create http request: %v", err)
+	}
+	r = r.WithContext(ctx)
+	r.Header.Set("Content-Type", "application/json")
+
+	endTime := time.Now().Add(timeout)
+	for {
+		// Retry until timeout.
+		if time.Now().After(endTime) {
+			return jsonrpc.Response{}, fmt.Errorf("timeout: %v", err)
+		}
+
+		// Send request.
+		response, err := httpClient.Do(r)
+		if err != nil {
+			continue
+		}
+
+		// Read response.
+		var resp jsonrpc.Response
+		if err := json.NewDecoder(response.Body).Decode(&resp); err != nil {
+			continue
+		}
+
+		return resp, nil
+	}
+}
+
+// URLFromMulti converts a `addr.MultiAddress` to a url string appropriate for
+// sending a JSON-RPC request to a darknode. The port is defined in the multi
+// address is incremented by one because of darknode specific logic about what
+// the JSON-RPC port is.
+func URLFromMulti(addr addr.MultiAddress) string {
+	return fmt.Sprintf("http://%s:%v", addr.IP4(), addr.Port()+1)
+}

client/client_suite_test.go

@@ -0,0 +1,13 @@
+package client_test
+
+import (
+	"testing"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+func TestClient(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Client Suite")
+}