diff --git a/Makefile b/Makefile index d84f0e6..393e27a 100644 --- a/Makefile +++ b/Makefile @@ -46,11 +46,14 @@ fakehost: $(BINARY) $(RUN) -vv host --pool "ws://$(FAKEBIND)" --rpc "fakenode://f21f0692b06019ae3f40d78d8b309487fc75f75b76df71d76196c3514272adf30aca4b2451181eb22208757cd4363923e17723d2f2ddf7b0175ecb87dada7ca1?fakepeers=$(FAKEPEERS)" fakehostpool: $(BINARY) - $(RUN) -vv host --pool ":memory:" --enode="enode://f21f0692b06019ae3f40d78d8b309487fc75f75b76df71d76196c3514272adf30aca4b2451181eb22208757cd4363923e17723d2f2ddf7b0175ecb87dada7ca1@[::]:30303?discport=0" --rpc "fakenode://f21f0692b06019ae3f40d78d8b309487fc75f75b76df71d76196c3514272adf30aca4b2451181eb22208757cd4363923e17723d2f2ddf7b0175ecb87dada7ca1?fakepeers=$(FAKEPEERS)" + $(RUN) -vv host --pool ":memory:" --enode="enode://f21f0692b06019ae3f40d78d8b309487fc75f75b76df71d76196c3514272adf30aca4b2451181eb22208757cd4363923e17723d2f2ddf7b0175ecb87dada7ca1@127.0.0.1:30303?discport=0" --rpc "fakenode://f21f0692b06019ae3f40d78d8b309487fc75f75b76df71d76196c3514272adf30aca4b2451181eb22208757cd4363923e17723d2f2ddf7b0175ecb87dada7ca1?fakepeers=$(FAKEPEERS)" fakeclient: $(BINARY) $(RUN) -vv client "http://$(FAKEBIND)" --nodekey=./nodekey --rpc "fakenode://85fbed4332ed4329ca2283f26606618815ae83a870c523bb99b0b2e9dfe5af3b4699c2830ecdeb67519d62362db44aa5a8cafee523e3ab8c76aeef1016f424a4?fakepeers=$(FAKEPEERS)" +poolstatus: + @curl -s "http://$(FAKEBIND)" -H 'Origin: http://localhost:3030/status' --data-binary '{"id":1,"method":"pool_status"}' + release: GOOS=linux GOARCH=amd64 LDFLAGS=$(LDFLAGS) ./build_release "$(PKG)" README.md LICENSE GOOS=linux GOARCH=386 LDFLAGS=$(LDFLAGS) ./build_release "$(PKG)" README.md LICENSE diff --git a/agent/agent.go b/agent/agent.go new file mode 100644 index 0000000..d782871 --- /dev/null +++ b/agent/agent.go @@ -0,0 +1,228 @@ +package agent + +import ( + "context" + "errors" + "sync" + "time" + + "github.com/vipnode/vipnode/ethnode" + "github.com/vipnode/vipnode/pool" + "github.com/vipnode/vipnode/pool/store" +) + +const defaultNumHosts = 3 + +var startTimeout = 10 * time.Second +var updateTimeout = 10 * time.Second + +// Agent is a companion process for nodes that manages the node's communication +// with a Vipnode Pool. +type Agent struct { + ethnode.EthNode + + // NodeURI can be used to override the enode:// connection string that + // the pool should advertise to other peers. Normally, the pool will + // automatically deduce this string from the connection IP and nodeID, but + // we can provide an override if there is a non-standard port or if the + // node runs on a different IP from the vipnode agent. (Optional) + NodeURI string + + // Version is the version of the vipnode agent that the host is running. + Version string + + // BalanceCallback is called whenever the client receives a balance update + // from the pool. It can be used for displaying the current balance to the + // client. (Optional) + BalanceCallback func(store.Balance) + + // PoolMessageCallback is called whenever the client receives a message + // from the pool. This can be a welcome message including rules and + // instructions for how to manage the client's balance. It should be + // displayed to the client. (Optional) + PoolMessageCallback func(string) + + // NumHosts is the minimum number of vipnode hosts the client should + // maintain connections with. (Optional) + NumHosts int + + // Payout is the address to register to associate pool credits towards. + // (Optional) + Payout string + + mu sync.Mutex + started bool + stopCh chan struct{} + waitCh chan error + nodeInfo ethnode.UserAgent // cached during Start +} + +// Start registers the node on the given pool and starts sending peer updates +// every store.KeepaliveInterval. It returns after +// successfully registering with the pool. +func (a *Agent) Start(p pool.Pool) error { + a.mu.Lock() + if a.started { + a.mu.Unlock() + return errors.New("pool already started") + } + if a.stopCh == nil { + a.stopCh = make(chan struct{}) + a.waitCh = make(chan error, 1) + } + a.mu.Unlock() + + startCtx, cancel := context.WithTimeout(context.Background(), startTimeout) + defer cancel() + + enode, err := a.EthNode.Enode(startCtx) + if err != nil { + return err + } + logger.Printf("Connected to local node: %s", enode) + + version := a.Version + if version == "" { + version = "dev" + } + + connectReq := pool.ConnectRequest{ + Payout: a.Payout, + NodeURI: a.NodeURI, + VipnodeVersion: version, + NodeInfo: a.EthNode.UserAgent(), + } + a.nodeInfo = connectReq.NodeInfo + resp, err := p.Connect(startCtx, connectReq) + if err != nil { + return err + } + logger.Printf("Registered on pool: Version %s", resp.PoolVersion) + + if resp.Message != "" && a.PoolMessageCallback != nil { + a.PoolMessageCallback(resp.Message) + } + + if err := a.updatePeers(startCtx, p); err != nil { + return err + } + + go func() { + a.waitCh <- a.serveUpdates(p) + }() + return nil +} + +// Whitelist a peer for this node. +func (a *Agent) Whitelist(ctx context.Context, nodeID string) error { + logger.Printf("Received whitelist request: %s", nodeID) + return a.EthNode.AddTrustedPeer(ctx, nodeID) +} + +// Stop shuts down all the active connections cleanly. +func (a *Agent) Stop() { + a.stopCh <- struct{}{} +} + +// Wait blocks until the agent is stopped. It returns any errors that occur +// during stopping. +func (a *Agent) Wait() error { + return <-a.waitCh +} + +func (a *Agent) serveUpdates(p pool.Pool) error { + ticker := time.Tick(store.KeepaliveInterval) + for { + select { + case <-ticker: + if err := a.updatePeers(context.Background(), p); err != nil { + return err + } + case <-a.stopCh: + a.mu.Lock() + a.started = false + a.mu.Unlock() + + // FIXME: Does it make sense to call a.disconnectPeers(...) here? + return nil + } + } +} + +// disconnectPeers tells the node to disconnect from its peers. +// DEPRECATED: This method is unused. It might be useful at some point though? +// If not, remove later. Or maybe it should be changed to only disconnect from +// vipnode-tracked peers? +func (a *Agent) disconnectPeers(ctx context.Context) error { + peers, err := a.EthNode.Peers(ctx) + if err != nil { + return err + } + for _, node := range peers { + if err := a.EthNode.DisconnectPeer(ctx, node.ID); err != nil { + return err + } + } + return nil +} + +func (a *Agent) updatePeers(ctx context.Context, p pool.Pool) error { + peers, err := a.EthNode.Peers(ctx) + if err != nil { + return err + } + + // Do we need more peers? + // FIXME: Does it make sense to request more peers before sending a vipnode_update? + if needMore := a.NumHosts - len(peers); needMore > 0 { + if err := a.AddPeers(ctx, p, needMore); err != nil { + return err + } + } + + update, err := p.Update(ctx, pool.UpdateRequest{PeerInfo: peers}) + if err != nil { + return err + } + var balance store.Balance + if a.BalanceCallback != nil && update.Balance != nil { + balance = *update.Balance + a.BalanceCallback(balance) + } + if len(update.InvalidPeers) > 0 { + // Client doesn't really need to do anything if the pool stopped + // tracking their host. That means the client is getting a free ride + // and it's up to the host to kick the client when the host deems + // necessary. + logger.Printf("Sent update: %d peers connected, %d expired in pool. Pool response: %s", len(peers), len(update.InvalidPeers), balance.String()) + } else { + logger.Printf("Sent update: %d peers connected. Pool response: %s", len(peers), balance.String()) + } + + return nil +} + +// AddPeers requests num peers from the pool and connects the node to them. +func (a *Agent) AddPeers(ctx context.Context, p pool.Pool, num int) error { + logger.Printf("Requesting %d more %q hosts from pool...", a.EthNode.Kind(), num) + peerResp, err := p.Peer(ctx, pool.PeerRequest{ + Num: num, + Kind: a.nodeInfo.Kind.String(), + }) + if err != nil { + return err + } + nodes := peerResp.Peers + logger.Printf("Received %d host candidates from pool.", len(nodes)) + for _, node := range nodes { + if err := a.EthNode.ConnectPeer(ctx, node.URI); err != nil { + return err + } + } + return nil +} + +// Service is the set of RPC calls exposed by an agent. +type Service interface { + Whitelist(ctx context.Context, nodeID string) error +} diff --git a/agent/agent_test.go b/agent/agent_test.go new file mode 100644 index 0000000..e4ea957 --- /dev/null +++ b/agent/agent_test.go @@ -0,0 +1,48 @@ +package agent + +import ( + "context" + "os" + "testing" + + "github.com/vipnode/vipnode/internal/fakenode" + "github.com/vipnode/vipnode/pool" + "github.com/vipnode/vipnode/pool/store" +) + +func TestAgent(t *testing.T) { + SetLogger(os.Stderr) + + agent := Agent{ + EthNode: &fakenode.FakeNode{ + NodeID: "foo", + }, + NumHosts: 3, + } + + p := &pool.StaticPool{} + if err := agent.Start(p); err != nil { + t.Fatal(err) + } + + if peers, err := agent.EthNode.Peers(context.Background()); err != nil { + t.Fatal(err) + } else if got, want := len(peers), 0; got != want { + t.Errorf("wrong number of peers: got %d; want %d", got, want) + } + + p.Nodes = append(p.Nodes, store.Node{ + URI: "foo", + }) + + // Force update + if err := agent.updatePeers(context.Background(), p); err != nil { + t.Fatal(err) + } + + if peers, err := agent.EthNode.Peers(context.Background()); err != nil { + t.Fatal(err) + } else if got, want := len(peers), 1; got != want { + t.Errorf("wrong number of peers: got %d; want %d", got, want) + } +} diff --git a/host/logger.go b/agent/logger.go similarity index 87% rename from host/logger.go rename to agent/logger.go index 99bf116..8e549af 100644 --- a/host/logger.go +++ b/agent/logger.go @@ -1,4 +1,4 @@ -package host +package agent import ( "io" @@ -11,7 +11,7 @@ var logger *log.Logger // SetLogger overrides the logger output for this package. func SetLogger(w io.Writer) { flags := log.Flags() - prefix := "[host] " + prefix := "[agent] " logger = log.New(w, prefix, flags) } diff --git a/client.go b/client.go index 6673242..706baf3 100644 --- a/client.go +++ b/client.go @@ -8,7 +8,7 @@ import ( "os/signal" "github.com/ethereum/go-ethereum/p2p/discv5" - "github.com/vipnode/vipnode/client" + "github.com/vipnode/vipnode/agent" "github.com/vipnode/vipnode/jsonrpc2" "github.com/vipnode/vipnode/pool" @@ -36,7 +36,10 @@ func runClient(options Options) error { } errChan := make(chan error) - c := client.New(remoteNode) + c := agent.Agent{ + EthNode: remoteNode, + Version: fmt.Sprintf("vipnode/client/%s", Version), + } c.PoolMessageCallback = func(msg string) { logger.Alertf("Message from pool: %s", msg) } diff --git a/client/client.go b/client/client.go deleted file mode 100644 index 5569ce5..0000000 --- a/client/client.go +++ /dev/null @@ -1,138 +0,0 @@ -package client - -import ( - "context" - "errors" - "time" - - "github.com/vipnode/vipnode/ethnode" - "github.com/vipnode/vipnode/pool" - "github.com/vipnode/vipnode/pool/store" -) - -// ErrAlreadyConnected is returned on Connect() if the client is already connected. -var ErrAlreadyConnected = errors.New("client already connected") - -func New(node ethnode.EthNode) *Client { - return &Client{ - EthNode: node, - stopCh: make(chan struct{}), - waitCh: make(chan error, 1), - } -} - -// Client represents a vipnode client which connects to a vipnode host. -type Client struct { - ethnode.EthNode - - // BalanceCallback is called whenever the client receives a balance update - // from the pool. It can be used for displaying the current balance to the - // client. - BalanceCallback func(store.Balance) - - // PoolMessageCallback is called whenever the client receives a message - // from the pool. This can be a welcome message including rules and - // instructions for how to manage the client's balance. It should be - // displayed to the client. - PoolMessageCallback func(string) - - connectedHosts []store.Node - stopCh chan struct{} - waitCh chan error -} - -// Wait blocks until the client is stopped. -func (c *Client) Wait() error { - return <-c.waitCh -} - -// Start retrieves compatible hosts from the pool and connects to them. Start -// blocks until registration is complete, then the keepalive peering updates -// break out into a separate goroutine and Start returns. -func (c *Client) Start(p pool.Pool) error { - logger.Printf("Requesting host candidates...") - starCtx := context.Background() - kind := c.EthNode.Kind().String() - resp, err := p.Client(starCtx, pool.ClientRequest{Kind: kind}) - if err != nil { - return err - } - if resp.Message != "" && c.PoolMessageCallback != nil { - c.PoolMessageCallback(resp.Message) - } - nodes := resp.Hosts - if len(nodes) == 0 { - return pool.NoHostNodesError{} - } - logger.Printf("Received %d host candidates from pool (version %s), connecting...", len(nodes), resp.PoolVersion) - for _, node := range nodes { - if err := c.EthNode.ConnectPeer(starCtx, node.URI); err != nil { - return err - } - } - if err := c.updatePeers(context.Background(), p); err != nil { - return err - } - - go func() { - c.waitCh <- c.serveUpdates(p, nodes) - }() - - return nil -} - -func (c *Client) serveUpdates(p pool.Pool, connectedHosts []store.Node) error { - ticker := time.Tick(store.KeepaliveInterval) - for { - select { - case <-ticker: - if err := c.updatePeers(context.Background(), p); err != nil { - return err - } - case <-c.stopCh: - closeCtx := context.Background() - for _, node := range connectedHosts { - if err := c.EthNode.DisconnectPeer(closeCtx, node.URI); err != nil { - return err - } - } - return nil - } - } -} - -func (c *Client) updatePeers(ctx context.Context, p pool.Pool) error { - peers, err := c.EthNode.Peers(ctx) - if err != nil { - return err - } - peerIDs := make([]string, 0, len(peers)) - for _, p := range peers { - peerIDs = append(peerIDs, p.ID) - } - - update, err := p.Update(ctx, pool.UpdateRequest{Peers: peerIDs}) - if err != nil { - return err - } - if c.BalanceCallback != nil && update.Balance != nil { - c.BalanceCallback(*update.Balance) - } - - if len(update.InvalidPeers) > 0 { - // Client doesn't really need to do anything if the pool stopped - // tracking their host. That means the client is getting a free ride - // and it's up to the host to kick the client when the host deems - // necessary. - logger.Printf("Sent update: %d peers connected, %d expired in pool. Pool response: %s", len(peerIDs), len(update.InvalidPeers), update.Balance.String()) - } else { - logger.Printf("Sent update: %d peers connected. Pool response: %s", len(peerIDs), update.Balance.String()) - } - - return nil -} - -// Disconnect from hosts, also stop serving updates. -func (c *Client) Stop() { - c.stopCh <- struct{}{} -} diff --git a/client/client_test.go b/client/client_test.go deleted file mode 100644 index 380697b..0000000 --- a/client/client_test.go +++ /dev/null @@ -1,27 +0,0 @@ -package client - -import ( - "testing" - - "github.com/vipnode/vipnode/internal/fakenode" - "github.com/vipnode/vipnode/pool" - "github.com/vipnode/vipnode/pool/store" -) - -func TestClient(t *testing.T) { - client := Client{ - EthNode: &fakenode.FakeNode{ - NodeID: "foo", - }, - } - - p := pool.StaticPool{} - err := client.Start(&p) - if _, ok := err.(pool.NoHostNodesError); !ok { - t.Errorf("unexpected no nodes error, got: %q", err) - } - - p.Nodes = append(p.Nodes, store.Node{ - URI: "foo", - }) -} diff --git a/client/logger.go b/client/logger.go deleted file mode 100644 index bb1c28b..0000000 --- a/client/logger.go +++ /dev/null @@ -1,20 +0,0 @@ -package client - -import ( - "io" - "io/ioutil" - "log" -) - -var logger *log.Logger - -// SetLogger overrides the logger output for this package. -func SetLogger(w io.Writer) { - flags := log.Flags() - prefix := "[client] " - logger = log.New(w, prefix, flags) -} - -func init() { - SetLogger(ioutil.Discard) -} diff --git a/ethnode/geth.go b/ethnode/geth.go index e2d288b..47bdcfd 100644 --- a/ethnode/geth.go +++ b/ethnode/geth.go @@ -20,6 +20,7 @@ type codedError interface { var _ EthNode = &gethNode{} type gethNode struct { + agent UserAgent client *rpc.Client } @@ -27,6 +28,10 @@ func (n *gethNode) ContractBackend() bind.ContractBackend { return ethclient.NewClient(n.client) } +func (n *gethNode) UserAgent() UserAgent { + return n.agent +} + func (n *gethNode) Kind() NodeKind { return Geth } diff --git a/ethnode/parity.go b/ethnode/parity.go index 10106f3..17677c5 100644 --- a/ethnode/parity.go +++ b/ethnode/parity.go @@ -16,6 +16,7 @@ type parityPeers struct { } type parityNode struct { + agent UserAgent client *rpc.Client } @@ -23,6 +24,10 @@ func (n *parityNode) ContractBackend() bind.ContractBackend { return ethclient.NewClient(n.client) } +func (n *parityNode) UserAgent() UserAgent { + return n.agent +} + func (n *parityNode) Kind() NodeKind { return Parity } @@ -55,7 +60,7 @@ func (n *parityNode) Peers(ctx context.Context) ([]PeerInfo, error) { if err != nil { return nil, err } - return result.Peers, nil + return filterActivePeers(result.Peers), nil } func (n *parityNode) Enode(ctx context.Context) (string, error) { @@ -73,3 +78,18 @@ func (n *parityNode) BlockNumber(ctx context.Context) (uint64, error) { } return strconv.ParseUint(result, 0, 64) } + +// filterActivePeers filters out any peers that have not completed the +// handshake yet. In Parity, these are peers without any specified Protocols. +func filterActivePeers(peers []PeerInfo) []PeerInfo { + if len(peers) == 0 { + return peers + } + activePeers := make([]PeerInfo, 0, len(peers)) + for _, peer := range peers { + if len(peer.Protocols) > 0 { + activePeers = append(activePeers, peer) + } + } + return activePeers +} diff --git a/ethnode/parity_test.go b/ethnode/parity_test.go new file mode 100644 index 0000000..5b965cd --- /dev/null +++ b/ethnode/parity_test.go @@ -0,0 +1,105 @@ +package ethnode + +import ( + "encoding/json" + "reflect" + "testing" +) + +func TestParityParsePeerInfo(t *testing.T) { + // Some examples of parity_netPeers output + testcases := []struct { + input []byte + want parityPeers + }{ + { + input: []byte(`{ + "active":0, + "connected":1, + "max":25, + "peers":[ + { + "caps":[ + "eth/62", + "eth/63", + "par/1", + "par/2", + "pip/1" + ], + "id":"a117e71696a740f2d0e5427fed5fa0ab1f343799aa873b46c361d405e9b3319689b26867e9b76921b0714ad79470d374ae4ef15f28ad4fe521de4c4f3ce702bc", + "name":"Parity/v1.8.0-beta-9882902-20171015/x86_64-linux-gnu/rustc1.21.0", + "network":{ + "localAddress":"172.18.0.2:41980", + "remoteAddress":"172.18.0.3:30303" + }, + "protocols":{ + "eth":{}, + "pip":{} + } + } + ] + }`), + want: parityPeers{ + Peers: []PeerInfo{ + { + ID: "a117e71696a740f2d0e5427fed5fa0ab1f343799aa873b46c361d405e9b3319689b26867e9b76921b0714ad79470d374ae4ef15f28ad4fe521de4c4f3ce702bc", + Name: "Parity/v1.8.0-beta-9882902-20171015/x86_64-linux-gnu/rustc1.21.0", + Caps: []string{"eth/62", "eth/63", "par/1", "par/2", "pip/1"}, + Network: struct { + LocalAddress string `json:"localAddress"` + RemoteAddress string `json:"remoteAddress"` + }{"172.18.0.2:41980", "172.18.0.3:30303"}, + Protocols: map[string]json.RawMessage{ + "eth": json.RawMessage("{}"), + "pip": json.RawMessage("{}"), + }, + }, + }, + }, + }, + { + input: []byte(`{ + "peers":[ + { + "caps":["foo"], + "id":"someid", + "name":"somename", + "protocols":{} + }, + { + "caps":["bar"], + "id":"anotherid", + "name":"anothername", + "protocols":{"bar": "baz"} + } + ] + }`), + want: parityPeers{ + Peers: []PeerInfo{ + { + ID: "anotherid", + Name: "anothername", + Caps: []string{"bar"}, + Protocols: map[string]json.RawMessage{ + "bar": json.RawMessage(`"baz"`), + }, + }, + }, + }, + }, + } + + for i, tc := range testcases { + var result parityPeers + err := json.Unmarshal(tc.input, &result) + if err != nil { + t.Errorf("[case %d] unexpected error for testcase: %s", i, err) + continue + } + result.Peers = filterActivePeers(result.Peers) + if !reflect.DeepEqual(result, tc.want) { + t.Errorf("[case %d] wrong agent values:\n got: %+v;\n want: %+v", i, result, tc.want) + } + } + +} diff --git a/ethnode/rpc.go b/ethnode/rpc.go index e47d8f4..a5d07d9 100644 --- a/ethnode/rpc.go +++ b/ethnode/rpc.go @@ -2,6 +2,7 @@ package ethnode import ( "context" + "encoding/json" "strconv" "strings" @@ -18,6 +19,17 @@ const ( Parity ) +func ParseNodeKind(s string) NodeKind { + switch strings.ToLower(s) { + case "geth": + return Geth + case "parity": + return Parity + default: + return Unknown + } +} + type NetworkID int const ( @@ -62,13 +74,13 @@ func (n NodeKind) String() string { // UserAgent is the metadata about node client. type UserAgent struct { - Version string // Result of web3_clientVersion - EthProtocol string // Result of eth_protocolVersion + Version string `json:"version"` // Result of web3_clientVersion + EthProtocol string `json:"eth_protocol"` // Result of eth_protocolVersion // Parsed/derived values - Kind NodeKind // Node implementation - Network NetworkID // Network ID - IsFullNode bool // Is this a full node? (or a light client?) + Kind NodeKind `json:"kind"` // Node implementation + Network NetworkID `json:"network"` // Network ID + IsFullNode bool `json:"is_full_node"` // Is this a full node? (or a light client?) } // ParseUserAgent takes string values as output from the web3 RPC for @@ -134,8 +146,16 @@ func DetectClient(client *rpc.Client) (*UserAgent, error) { // PeerInfo stores the node ID and client metadata about a peer. type PeerInfo struct { - ID string `json:"id"` // Unique node identifier (also the encryption pubkey) - Name string `json:"name"` // Name of the node, including client type, version, OS, custom data + ID string `json:"id"` // Unique node identifier (also the encryption pubkey) + Name string `json:"name"` // Name of the node, including client type, version, OS, custom data + Caps []string `json:"caps"` // Capabilities the node is advertising. + Protocols map[string]json.RawMessage `json:"protocols"` // Sub-protocol specific metadata fields + + // FIXME: Do we want to include node-local network address state? Or is it unnecessary security leakage? + Network struct { + LocalAddress string `json:"localAddress"` // Local endpoint of the TCP data connection + RemoteAddress string `json:"remoteAddress"` // Remote endpoint of the TCP data connection + } `json:"network"` } // EthNode is the normalized interface between different kinds of nodes. @@ -144,6 +164,8 @@ type EthNode interface { // Kind returns the kind of node this is. Kind() NodeKind + // UserAgent returns the versions of the client. + UserAgent() UserAgent // Enode returns this node's enode://... Enode(ctx context.Context) (string, error) // AddTrustedPeer adds a nodeID to a set of nodes that can always connect, even @@ -164,17 +186,23 @@ type EthNode interface { // RemoteNode autodetects the node kind and returns the appropriate EthNode // implementation. func RemoteNode(client *rpc.Client) (EthNode, error) { - version, err := DetectClient(client) + agent, err := DetectClient(client) if err != nil { return nil, err } - switch version.Kind { + switch agent.Kind { case Parity: - return &parityNode{client: client}, nil + return &parityNode{ + agent: *agent, + client: client, + }, nil default: // Treat everything else as Geth // FIXME: Is this a bad idea? - node := &gethNode{client: client} + node := &gethNode{ + agent: *agent, + client: client, + } ctx := context.TODO() if err := node.CheckCompatible(ctx); err != nil { return nil, err diff --git a/host.go b/host.go index 482b596..4ebfb4a 100644 --- a/host.go +++ b/host.go @@ -7,7 +7,7 @@ import ( "os/signal" "github.com/ethereum/go-ethereum/p2p/discv5" - "github.com/vipnode/vipnode/host" + "github.com/vipnode/vipnode/agent" "github.com/vipnode/vipnode/jsonrpc2" ws "github.com/vipnode/vipnode/jsonrpc2/ws/gorilla" "github.com/vipnode/vipnode/pool" @@ -37,7 +37,11 @@ func runHost(options Options) error { logger.Warning("No --payout address provided, will not receive pool payments.") } - h := host.New(remoteNode, options.Host.Payout) + h := agent.Agent{ + EthNode: remoteNode, + Payout: options.Host.Payout, + Version: fmt.Sprintf("vipnode/host/%s", Version), + } if options.Host.NodeURI != "" { if err := matchEnode(options.Host.NodeURI, nodeID); err != nil { return err @@ -49,10 +53,20 @@ func runHost(options Options) error { h.NodeURI = remoteEnode } + sigCh := make(chan os.Signal, 1) + signal.Notify(sigCh, os.Interrupt) + go func() { + for _ = range sigCh { + logger.Info("Shutting down...") + h.Stop() + } + }() + if options.Host.Pool == ":memory:" { // Support for in-memory pool. This is primarily for testing. logger.Infof("Starting in-memory vipnode pool.") p := pool.New(memory.New(), nil) + p.Version = fmt.Sprintf("vipnode/pool/%s", Version) rpcPool := &jsonrpc2.Local{} if err := rpcPool.Server.Register("vipnode_", p); err != nil { return err @@ -73,8 +87,9 @@ func runHost(options Options) error { } logger.Infof("Connected to vipnode pool: %s", options.Host.Pool) + // Register reverse-directional RPC calls available on the host rpcServer := &jsonrpc2.Server{} - if err := rpcServer.RegisterMethod("vipnode_whitelist", h, "Whitelist"); err != nil { + if err := rpcServer.RegisterMethod("vipnode_whitelist", &h, "Whitelist"); err != nil { return err } rpcPool := jsonrpc2.Remote{ @@ -98,15 +113,6 @@ func runHost(options Options) error { errChan <- h.Wait() }() - sigCh := make(chan os.Signal, 1) - signal.Notify(sigCh, os.Interrupt) - go func() { - for _ = range sigCh { - logger.Info("Shutting down...") - h.Stop() - } - }() - return <-errChan } diff --git a/host/host.go b/host/host.go deleted file mode 100644 index 9f57ad7..0000000 --- a/host/host.go +++ /dev/null @@ -1,166 +0,0 @@ -package host - -import ( - "context" - "time" - - "github.com/vipnode/vipnode/ethnode" - "github.com/vipnode/vipnode/pool" - "github.com/vipnode/vipnode/pool/store" -) - -var startTimeout = 10 * time.Second -var updateTimeout = 10 * time.Second - -type nodeID string - -type client struct { - nodeID nodeID - expire time.Time -} - -func New(node ethnode.EthNode, payout string) *Host { - return &Host{ - node: node, - payout: payout, - stopCh: make(chan struct{}), - waitCh: make(chan error, 1), - } -} - -type HostService interface { - Whitelist(ctx context.Context, nodeID string) error -} - -// Host represents a single vipnode host. -type Host struct { - // NodeURI can be used to set the enode:// connection string that - // the pool should advertise to clients. Normally, the pool will - // automatically deduce this string from the connection IP and nodeID, but - // we can provide an override if there is a non-standard port or if the - // node runs on a different IP from the vipnode agent. - NodeURI string - - node ethnode.EthNode - payout string - stopCh chan struct{} - waitCh chan error -} - -// Whitelist a client for this host. -func (h *Host) Whitelist(ctx context.Context, nodeID string) error { - logger.Printf("Received whitelist request: %s", nodeID) - return h.node.AddTrustedPeer(ctx, nodeID) -} - -// Disconnect a client from this host and remove from whitelist. -func (h *Host) Disconnect(ctx context.Context, nodeID string) error { - logger.Printf("Received disconnect request: %s", nodeID) - if err := h.node.RemoveTrustedPeer(ctx, nodeID); err != nil { - return err - } - return h.node.DisconnectPeer(ctx, nodeID) -} - -func (h *Host) updatePeers(ctx context.Context, p pool.Pool) error { - block, err := h.node.BlockNumber(ctx) - if err != nil { - return err - } - - peers, err := h.node.Peers(ctx) - if err != nil { - return err - } - peerUpdate := make([]string, 0, len(peers)) - for _, peer := range peers { - peerUpdate = append(peerUpdate, peer.ID) - } - update, err := p.Update(ctx, pool.UpdateRequest{ - Peers: peerUpdate, - BlockNumber: block, - }) - if err != nil { - return err - } - if len(update.InvalidPeers) == 0 { - logger.Printf("Sent update: %d peers. Pool response: %s", len(peerUpdate), update.Balance.String()) - return nil - } - logger.Printf("Sent update: %d peers. Pool response: Disconnect from %d invalid peers, %s", len(peerUpdate), len(update.InvalidPeers), update.Balance.String()) - for _, peerID := range update.InvalidPeers { - // FIXME: Are there recoverable errors here? - if err := h.node.RemoveTrustedPeer(ctx, peerID); err != nil { - return err - } - if err := h.node.DisconnectPeer(ctx, peerID); err != nil { - return err - } - } - return nil -} - -// Stop will terminate the update peers loop, which will cause Start to return. -func (h *Host) Stop() { - h.stopCh <- struct{}{} -} - -// Wait blocks until the host is stopped. It returns any errors that occur -// during stopping. -func (h *Host) Wait() error { - return <-h.waitCh -} - -// Start registers the host on the given pool and starts sending peer updates -// every store.KeepaliveInterval. It returns after -// successfully registering with the pool. -func (h *Host) Start(p pool.Pool) error { - startCtx, cancel := context.WithTimeout(context.Background(), startTimeout) - defer cancel() - - enode, err := h.node.Enode(startCtx) - if err != nil { - return err - } - logger.Printf("Connected to local node: %s", enode) - - hostReq := pool.HostRequest{ - Kind: h.node.Kind().String(), - Payout: h.payout, - NodeURI: h.NodeURI, - } - resp, err := p.Host(startCtx, hostReq) - if err != nil { - return err - } - logger.Printf("Registered on pool: Version %s", resp.PoolVersion) - - // TODO: Resume tracking peers that we care about (in case of interrupted - // shutdown)? - - if err := h.updatePeers(startCtx, p); err != nil { - return err - } - - go func() { - h.waitCh <- h.serveUpdates(p) - }() - return nil -} - -func (h *Host) serveUpdates(p pool.Pool) error { - ticker := time.Tick(store.KeepaliveInterval) - for { - select { - case <-ticker: - ctx, cancel := context.WithTimeout(context.Background(), updateTimeout) - err := h.updatePeers(ctx, p) - cancel() - if err != nil { - return err - } - case <-h.stopCh: - return nil - } - } -} diff --git a/internal/fakecluster/fakecluster.go b/internal/fakecluster/fakecluster.go new file mode 100644 index 0000000..3e3ad6e --- /dev/null +++ b/internal/fakecluster/fakecluster.go @@ -0,0 +1,154 @@ +package fakecluster + +import ( + "crypto/ecdsa" + "fmt" + "io" + "strings" + + "github.com/ethereum/go-ethereum/p2p/discv5" + "github.com/vipnode/vipnode/agent" + "github.com/vipnode/vipnode/internal/fakenode" + "github.com/vipnode/vipnode/jsonrpc2" + "github.com/vipnode/vipnode/pool" + "github.com/vipnode/vipnode/pool/store/memory" +) + +type clusterAgent struct { + *agent.Agent + Node *fakenode.FakeNode + RemotePool pool.Pool + In *jsonrpc2.Remote + Out *jsonrpc2.Remote + Key *ecdsa.PrivateKey +} + +// Cluster represents a set of active hosts and clients connected to a pool. +type Cluster struct { + Clients []clusterAgent + Hosts []clusterAgent + Pool *pool.VipnodePool + + pipes []io.Closer +} + +// New returns a pre-connected pool of hosts and clients. +func New(hostKeys []*ecdsa.PrivateKey, clientKeys []*ecdsa.PrivateKey) (*Cluster, error) { + cluster := &Cluster{ + Hosts: []clusterAgent{}, + Clients: []clusterAgent{}, + pipes: []io.Closer{}, + } + + cluster.Pool = pool.New(memory.New(), nil) + payout := "" + for _, hostKey := range hostKeys { + rpcPool2Host, rpcHost2Pool := jsonrpc2.ServePipe() + cluster.pipes = append(cluster.pipes, rpcPool2Host, rpcHost2Pool) + if err := rpcPool2Host.Server.Register("vipnode_", cluster.Pool); err != nil { + return nil, err + } + + hostNodeID := discv5.PubkeyID(&hostKey.PublicKey).String() + hostNode := fakenode.Node(hostNodeID) + hostNodeURI := fmt.Sprintf("enode://%s@127.0.0.1:30303", hostNodeID) + h := &agent.Agent{EthNode: hostNode, Payout: payout} + if err := rpcHost2Pool.Server.RegisterMethod("vipnode_whitelist", h, "Whitelist"); err != nil { + return nil, err + } + h.NodeURI = hostNodeURI + hostPool := pool.Remote(rpcHost2Pool, hostKey) + + if err := h.Start(hostPool); err != nil { + return nil, err + } + + cluster.Hosts = append(cluster.Hosts, clusterAgent{ + Agent: h, + Node: hostNode, + In: rpcPool2Host, + Out: rpcHost2Pool, + Key: hostKey, + RemotePool: hostPool, + }) + } + + for _, clientKey := range clientKeys { + rpcPool2Client, rpcClient2Pool := jsonrpc2.ServePipe() + cluster.pipes = append(cluster.pipes, rpcPool2Client, rpcClient2Pool) + rpcPool2Client.Server.Register("vipnode_", cluster.Pool) + + clientNodeID := discv5.PubkeyID(&clientKey.PublicKey).String() + clientNode := fakenode.Node(clientNodeID) + c := &agent.Agent{ + EthNode: clientNode, + NumHosts: 3, + } + clientPool := pool.Remote(rpcClient2Pool, clientKey) + if err := c.Start(clientPool); err != nil { + return nil, err + } + cluster.Clients = append(cluster.Clients, clusterAgent{ + Agent: c, + Node: clientNode, + In: rpcPool2Client, + Out: rpcClient2Pool, + Key: clientKey, + RemotePool: clientPool, + }) + } + return cluster, nil +} + +// Close shuts down all the open pipes. +func (c *Cluster) Close() error { + errors := []error{} + for _, pipe := range c.pipes { + if err := pipe.Close(); err != nil { + errors = append(errors, err) + } + } + for _, host := range c.Hosts { + host.Stop() + } + for _, client := range c.Clients { + client.Stop() + } + for _, host := range c.Hosts { + if err := c.Pool.CloseRemote(host.In); err != nil { + errors = append(errors, err) + } + if err := host.Wait(); err != nil { + errors = append(errors, err) + } + } + for _, client := range c.Clients { + if err := client.Wait(); err != nil { + errors = append(errors, err) + } + } + if len(errors) > 0 { + return CloseErrors(errors) + } + return nil +} + +// CloseErrors are used to return a set of errors that occurred while +// attempting to shut down a cluster. +type CloseErrors []error + +func (e CloseErrors) Error() string { + if len(e) == 0 { + return "no close errors" + } + + var s strings.Builder + fmt.Fprintf(&s, "failed to close with %d errors: ", len(e)) + for i, err := range e { + s.WriteString(err.Error()) + if i != len(e)-1 { + s.WriteString("; ") + } + } + return s.String() +} diff --git a/internal/fakenode/fakenode.go b/internal/fakenode/fakenode.go index d329bff..7688cf2 100644 --- a/internal/fakenode/fakenode.go +++ b/internal/fakenode/fakenode.go @@ -2,6 +2,7 @@ package fakenode import ( "context" + "encoding/json" "fmt" "net/url" @@ -23,9 +24,10 @@ func Call(method string, args ...interface{}) call { func Node(nodeID string) *FakeNode { return &FakeNode{ - NodeKind: ethnode.Geth, - NodeID: nodeID, - Calls: Calls{}, + NodeKind: ethnode.Geth, + NodeID: nodeID, + Calls: Calls{}, + IsFullNode: true, } } @@ -36,12 +38,21 @@ type FakeNode struct { Calls Calls FakePeers []ethnode.PeerInfo FakeBlockNumber uint64 + IsFullNode bool } func (n *FakeNode) ContractBackend() bind.ContractBackend { return ðclient.Client{} } +func (n *FakeNode) UserAgent() ethnode.UserAgent { + return ethnode.UserAgent{ + Version: "Geth/Fakenode/Go-tests", + Network: 1, + IsFullNode: n.IsFullNode, + Kind: n.NodeKind, + } +} func (n *FakeNode) Kind() ethnode.NodeKind { return n.NodeKind } func (n *FakeNode) Enode(ctx context.Context) (string, error) { return n.NodeID, nil } func (n *FakeNode) AddTrustedPeer(ctx context.Context, nodeID string) error { @@ -59,7 +70,11 @@ func (n *FakeNode) ConnectPeer(ctx context.Context, nodeURI string) error { return err } n.FakePeers = append(n.FakePeers, ethnode.PeerInfo{ - ID: uri.User.Username(), + ID: uri.User.Username(), + Caps: []string{"fake/1", "eth/62", "eth/63", "les/2"}, + Protocols: map[string]json.RawMessage{ + "fake": json.RawMessage("{}"), + }, }) return nil } diff --git a/internal/keygen/keygen.go b/internal/keygen/keygen.go index e9bce74..ed92644 100644 --- a/internal/keygen/keygen.go +++ b/internal/keygen/keygen.go @@ -13,6 +13,22 @@ var hardcodedKeys = []string{ `OX9lnxz+fWNmEEBXCKfEmEsh5oGhCXXdHJBqilgZPNc=`, `pebDrvrc9iHxN8k7YJva6bvr6Mzimb9ZbFrGpVy/Wb0=`, `cl3X1He2rNZiXbsii3M9zxBTi9B7gB1Tqgk6u5rMytE=`, + `+0HMUDyMBxFbNat59Vl6Sg+3EcVgiXt1y+JNtnjKb18=`, + `exk5qBaBxCex5A5Rx9/0qokyz4tu2aAwCJNzsEtIXnk=`, + `eqS1AIeHCa6xA4WWE2Q8hooTFVyUtQasJeDH3TSxkGU=`, + `DvqsYyCf9KmbmcC34hTwIjzVWSJnXKTxSxJAlKABSBs=`, + `gciM+wfV80vZxh5RAevJzcRBJyGkoxvmWRxyaNkOlpc=`, + `q8Ye3Cq3+/JKASV1KrSKy6wLLwjUdkffM4sBtKcOyPQ=`, + `dAoxmCIoohJePxhAlwL1res/Ict7U+ZhpTHlSXD6zt4=`, + `bx7mVtX5RV8BgsMmWqGs9aPnSNaSRfjB0sPBBKMfex4=`, + `wRQRSoYNlu/7SO75LJYieZv25TlEgH/NE8fyq/OWUQQ=`, + `O5oR0U9/xwfmOQZenln+Vr5rUZe3ooCs2ZFkKYx9VDQ=`, + `MhQtBdZogSKlwB0S7UTffnVrT6G/LXncFiaCaCQBTrQ=`, + `1raBg6tKGFSJ3JVV+cmwezOdNKYRAG1/CwMfw8SdHxQ=`, + `hyHctcYKirNqWFPyU46aTsU1DVOW6SbBH8IR+jLOIa0=`, + `vQzyvauJh+RZojBXy1jzziBr8PBWP+rV3Qefx0ANzsU=`, + `qHC5oCzkoXfYEzNmkS/CN4aJTKUHI1/NoWW9u8AbbQM=`, + `WalBCrYbrAA4Npnaez6nDaEIyX4NMyC6GGpY7Y0iMoM=`, } func HardcodedKey(t *testing.T) *ecdsa.PrivateKey { @@ -34,6 +50,11 @@ func HardcodedKeyIdx(t *testing.T, idx int) *ecdsa.PrivateKey { return privkey } +func EncodeKey(key *ecdsa.PrivateKey) string { + data := crypto.FromECDSA(key) + return base64.StdEncoding.EncodeToString(data) +} + func NewKey(t *testing.T) *ecdsa.PrivateKey { key, err := crypto.GenerateKey() if err != nil { diff --git a/jsonrpc2/remote.go b/jsonrpc2/remote.go index 789b089..8c3f1de 100644 --- a/jsonrpc2/remote.go +++ b/jsonrpc2/remote.go @@ -13,6 +13,8 @@ import ( // ServePipe sets up symmetric server/clients over a net.Pipe() and starts // both in goroutines. Useful for testing. Services still need to be registered. +// FIXME: This is a testing helper, ideally we want to get rid of it. It leaks +// goroutines by design. func ServePipe() (*Remote, *Remote) { c1, c2 := net.Pipe() client := Remote{ diff --git a/jsonrpc2/server.go b/jsonrpc2/server.go index 9a51838..29e42ee 100644 --- a/jsonrpc2/server.go +++ b/jsonrpc2/server.go @@ -17,8 +17,9 @@ var ErrNoPublicMethods = errors.New("no public methods") type Handler interface { // Handle takes a request message and returns a response message. Handle(ctx context.Context, request *Message) (response *Message) + // FIXME: Register* really shouldn't be part of this signature, right? - Register(prefix string, receiver interface{}) error + Register(prefix string, receiver interface{}, onlyMethods ...string) error RegisterMethod(rpcName string, receiver interface{}, methodName string) error } @@ -34,7 +35,7 @@ type Server struct { // Register adds valid methods from the receiver to the registry with the given // prefix. Method names are lowercased. -func (s *Server) Register(prefix string, receiver interface{}) error { +func (s *Server) Register(prefix string, receiver interface{}, onlyMethods ...string) error { s.mu.Lock() defer s.mu.Unlock() @@ -52,13 +53,30 @@ func (s *Server) Register(prefix string, receiver interface{}) error { return ErrNoPublicMethods } + var methodWhitelist map[string]struct{} + if len(onlyMethods) > 0 { + methodWhitelist = map[string]struct{}{} + for _, name := range onlyMethods { + methodWhitelist[name] = struct{}{} + } + } + var buf bytes.Buffer for name, m := range methods { + buf.Reset() buf.WriteString(prefix) buf.WriteRune(unicode.ToLower(rune(name[0]))) buf.WriteString(name[1:]) + + if methodWhitelist != nil { + // Skip methods that are not whitelisted + methodName := buf.String()[len(prefix):] + if _, ok := methodWhitelist[methodName]; !ok { + continue + } + } + s.registry[buf.String()] = m - buf.Reset() } return nil } diff --git a/jsonrpc2/server_test.go b/jsonrpc2/server_test.go index 7a5ce31..eeaa994 100644 --- a/jsonrpc2/server_test.go +++ b/jsonrpc2/server_test.go @@ -9,7 +9,7 @@ import ( func TestServer(t *testing.T) { service := &FruitService{} s := Server{} - if err := s.Register("foo_", service); err != nil { + if err := s.Register("foo_", service, "apple", "banana"); err != nil { t.Error(err) } @@ -42,4 +42,19 @@ func TestServer(t *testing.T) { if string(resp.Response.Result) != "null" { t.Errorf("unexpected result: %q", resp.Result) } + + resp = s.Handle(context.Background(), &Message{ + ID: json.RawMessage([]byte("3")), + Version: Version, + Request: &Request{ + Method: "foo_cherry", + }, + }) + if resp.Error == nil { + t.Errorf("expected error, got: %q", resp) + } + + if resp.Error.Message != "method not found: foo_cherry" { + t.Errorf("unexpected error message: %q", resp.Error) + } } diff --git a/jsonrpc2/types.go b/jsonrpc2/types.go index 4985684..6a71722 100644 --- a/jsonrpc2/types.go +++ b/jsonrpc2/types.go @@ -2,6 +2,7 @@ package jsonrpc2 import ( "encoding/json" + "fmt" ) const Version = "2.0" @@ -24,6 +25,16 @@ type Message struct { Version string `json:"jsonrpc"` // TODO: Replace this with a null-type that encodes to 2.0, like https://go-review.googlesource.com/c/tools/+/136675/1/internal/jsonrpc2/jsonrpc2.go#221 } +func (m Message) String() string { + // This method is here to satisfy vet + b, err := json.Marshal(m) + if err != nil { + // This shouldn't happen. Might even be worth panic'ing? + return fmt.Sprintf("failed to marshal %T: %s", m, err) + } + return string(b) +} + type Request struct { Method string `json:"method"` Params json.RawMessage `json:"params,omitempty"` diff --git a/jsonrpc2/types_test.go b/jsonrpc2/types_test.go new file mode 100644 index 0000000..139b2c6 --- /dev/null +++ b/jsonrpc2/types_test.go @@ -0,0 +1,15 @@ +package jsonrpc2 + +import "testing" + +func TestMessageFormat(t *testing.T) { + msg := &Message{ + ID: []byte("42"), + Version: "2.0", + } + + got, want := msg.String(), `{"id":42,"jsonrpc":"2.0"}` + if got != want { + t.Errorf("wrong message string formatting:\n got: %s;\n want: %s", got, want) + } +} diff --git a/main.go b/main.go index 873f0aa..c696e48 100644 --- a/main.go +++ b/main.go @@ -20,9 +20,8 @@ import ( "github.com/alexcesaro/log/golog" "github.com/ethereum/go-ethereum/crypto" flags "github.com/jessevdk/go-flags" - "github.com/vipnode/vipnode/client" + "github.com/vipnode/vipnode/agent" "github.com/vipnode/vipnode/ethnode" - "github.com/vipnode/vipnode/host" "github.com/vipnode/vipnode/internal/fakenode" "github.com/vipnode/vipnode/internal/pretty" "github.com/vipnode/vipnode/jsonrpc2" @@ -51,11 +50,12 @@ type Options struct { } `command:"client" description:"Connect to a vipnode as a client."` Host struct { - Pool string `long:"pool" description:"Pool to participate in." default:"wss://pool.vipnode.org/"` - RPC string `long:"rpc" description:"RPC path or URL of the host node."` - NodeKey string `long:"nodekey" description:"Path to the host node's private key."` - NodeURI string `long:"enode" description:"Public enode://... URI for clients to connect to. (If node is on a different IP from the vipnode agent)"` - Payout string `long:"payout" description:"Ethereum wallet address to receive pool payments."` + Pool string `long:"pool" description:"Pool to participate in." default:"wss://pool.vipnode.org/"` + RPC string `long:"rpc" description:"RPC path or URL of the host node."` + NodeKey string `long:"nodekey" description:"Path to the host node's private key."` + NodeURI string `long:"enode" description:"Public enode://... URI for clients to connect to. (If node is on a different IP from the vipnode agent)"` + Payout string `long:"payout" description:"Ethereum wallet address to receive pool payments."` + JoinPeers int `long:"join-peers" description:"Whitelist and connect to N other hosts on the pool." default:"0"` } `command:"host" description:"Host a vipnode."` Pool struct { @@ -189,7 +189,7 @@ func subcommand(cmd string, options Options) error { sigCh := make(chan os.Signal, 1) signal.Notify(sigCh, os.Interrupt) - backoff := []int{5, 30, 60, 90, 300} // Backoff in sequence in seconds. + backoff := []int{5, 30, 60, 90, 300} // Backoff sequence in seconds. clearTimeout := time.Second * 300 // Time between attempts before we reset the backoff var err error for i := 0; ; i++ { @@ -281,8 +281,7 @@ func main() { if logLevel == log.Debug { // Enable logging from subpackages pool.SetLogger(logWriter) - client.SetLogger(logWriter) - host.SetLogger(logWriter) + agent.SetLogger(logWriter) payment.SetLogger(logWriter) ethnode.SetLogger(logWriter) jsonrpc2.SetLogger(logWriter) diff --git a/pool.go b/pool.go index 834aea7..1a09e51 100644 --- a/pool.go +++ b/pool.go @@ -191,14 +191,15 @@ func runPool(options Options) error { } handler := &server{ - ws: &ws.Upgrader{}, - header: http.Header{}, + ws: &ws.Upgrader{}, + header: http.Header{}, + onDisconnect: p.CloseRemote, } if options.Pool.AllowOrigin != "" { handler.header.Set("Access-Control-Allow-Origin", options.Pool.AllowOrigin) } - if err := handler.Register("vipnode_", p); err != nil { + if err := handler.Register("vipnode_", p, "connect", "disconnect", "ping", "update", "client", "host"); err != nil { return err } diff --git a/pool/pool.go b/pool/pool.go index a7a5754..4ab389c 100644 --- a/pool/pool.go +++ b/pool/pool.go @@ -3,16 +3,49 @@ package pool import ( "context" + "github.com/vipnode/vipnode/ethnode" "github.com/vipnode/vipnode/pool/store" ) -// TODO: Add HostRequest.Network and ClientRequest.Network? -// TODO: Add HostRequest.HostVersion? +// ConnectRequest is a base request done when a vipnode agent connects to a pool. +// It is common between hosts and clients +type ConnectRequest struct { + // VipnodeVersion is the version string of the vipnode agent + VipnodeVersion string `json:"vipnode_version"` + + // NodeInfo is the metadata of the Ethereum node, includes node kind. + NodeInfo ethnode.UserAgent `json:"node_info"` + + // NodeURI is an optional public node URI override, useful if the vipnode + // agent runs on a separate IP from the actual node host. Otherwise, the + // pool will automatically use the same IP and default port as the host + // connecting. + NodeURI string `json:"node_uri,omitempty"` + + // Payout sets the wallet account to register the host credit towards. (Optional) + Payout string `json:"payout"` +} + +// ConnectResponse is the response a vipnode agent receives from the pool after +// the first connection request. It is common between hosts and clients. +type ConnectResponse struct { + // PoolVersion is the version of vipnode-pool that is running. + PoolVersion string `json:"pool_version"` + // Hosts that have whitelisted the NodeID and are ready for the node to + // connect to. + Hosts []store.Node `json:"hosts,omitempty"` + // Message contains a prompt for the client from the pool, possibly + // instructions for interfacing with this pool. For example, a link to the + // DApp for adding a balance deposit. + Message string `json:"message,omitempty"` +} // HostRequest is the request type for Host RPC calls. +// DEPRECATED: Use ConnectRequest/ConnectResponse type HostRequest struct { // Kind is the type of node the host supports: geth, parity - Kind string `json:"kind"` + Kind string `json:"kind,omitempty"` + // Payout sets the wallet account to register the host credit towards. Payout string `json:"payout"` // Optional public node URI override, useful if the vipnode agent runs on a @@ -22,20 +55,26 @@ type HostRequest struct { } // HostResponse is the response type for Host RPC calls. +// DEPRECATED: Use ConnectRequest/ConnectResponse type HostResponse struct { PoolVersion string `json:"pool_version"` } // ClientRequest is the request type for Client RPC calls. +// DEPRECATED: Use ConnectRequest/ConnectResponse type ClientRequest struct { - Kind string `json:"kind"` - NumHosts int `json:"num_hosts,omitempty"` // NumHosts is the number of hosts to request from the pool. (Optional) + // Kind is the type of node the host supports: geth, parity + Kind string `json:"kind,omitempty"` + + // NumHosts is the number of hosts to request from the pool. (Optional) + NumHosts int `json:"num_hosts,omitempty"` } // ClientResponse is the response type for Client RPC calls. +// DEPRECATED: Use ConnectRequest/ConnectResponse type ClientResponse struct { - // Hosts that have whitelisted the client NodeID and are ready for the - // client to connect. + // Hosts that have whitelisted the NodeID and are ready for the node to + // connect to. Hosts []store.Node `json:"hosts"` // PoolVersion is the version of vipnode-pool that is running. PoolVersion string `json:"pool_version"` @@ -47,33 +86,56 @@ type ClientResponse struct { // UpdateRequest is the request type for Update RPC calls. type UpdateRequest struct { - Peers []string `json:"peers"` - BlockNumber uint64 `json:"block_number"` + Peers []string `json:"peers,omitempty"` // DEPRECATED + PeerInfo []ethnode.PeerInfo `json:"peers_info"` + BlockNumber uint64 `json:"block_number"` } // UpdateResponse is the response type for Update RPC calls. type UpdateResponse struct { Balance *store.Balance `json:"balance,omitempty"` InvalidPeers []string `json:"invalid_peers"` + // TODO: Add PoolPeers []string // PoolPeers is the set of peers that are members of the pool. +} + +// PeerRequest is the request type for Peer RPC calls. +type PeerRequest struct { + // Num is the number of peers + Num int `json:"num"` + // Kind is the type of node we desire, such as "parity" or "geth" (optional) + Kind string `json:"kind,omitempty"` +} + +// PeerResponse is the response type for Peer RPC calls. +type PeerResponse struct { + // Peers that have whitelisted the NodeID and are ready for the node to + // connect to. + Peers []store.Node `json:"peers"` } // Pool represents a vipnode pool for coordinating between clients and hosts. type Pool interface { // Host subscribes a host to receive vipnode_whitelist instructions. + // DEREPCATED: Use Connect Host(ctx context.Context, req HostRequest) (*HostResponse, error) // Client requests for available hosts to connect to as a client. + // DEREPCATED: Use Connect Client(ctx context.Context, req ClientRequest) (*ClientResponse, error) - // Disconnect stops tracking the connection and billing, will prompt a - // disconnect from both ends. - Disconnect(ctx context.Context) error + // Connect subscribes to the active nodes set. + Connect(ctx context.Context, req ConnectRequest) (*ConnectResponse, error) // Update is a keep-alive for sharing the node's peering info. It returns // a list of peers that are no longer corroborated by the pool, and current // balance for the node (if relevant). Update(ctx context.Context, req UpdateRequest) (*UpdateResponse, error) + // Peer initiates a peering request for valid hosts. The returned set of peers + // match the request query and are ready to peer with. By default, it + // should only return full node hosts as peers. + Peer(ctx context.Context, req PeerRequest) (*PeerResponse, error) + // Withdraw prompts a request to settle the node's balance. Withdraw(ctx context.Context) error } diff --git a/pool/remote.go b/pool/remote.go index b300205..82cbb8f 100644 --- a/pool/remote.go +++ b/pool/remote.go @@ -73,6 +73,46 @@ func (p *RemotePool) Client(ctx context.Context, req ClientRequest) (*ClientResp return &resp, nil } +func (p *RemotePool) Connect(ctx context.Context, req ConnectRequest) (*ConnectResponse, error) { + signedReq := request.NodeRequest{ + Method: "vipnode_connect", + NodeID: p.nodeID, + Nonce: p.getNonce(), + ExtraArgs: []interface{}{req}, + } + + args, err := signedReq.SignedArgs(p.privkey) + if err != nil { + return nil, err + } + var resp ConnectResponse + if err := p.client.Call(ctx, &resp, signedReq.Method, args...); err != nil { + return nil, err + } + + return &resp, nil +} + +func (p *RemotePool) Peer(ctx context.Context, req PeerRequest) (*PeerResponse, error) { + signedReq := request.NodeRequest{ + Method: "vipnode_peer", + NodeID: p.nodeID, + Nonce: p.getNonce(), + ExtraArgs: []interface{}{req}, + } + + args, err := signedReq.SignedArgs(p.privkey) + if err != nil { + return nil, err + } + var resp PeerResponse + if err := p.client.Call(ctx, &resp, signedReq.Method, args...); err != nil { + return nil, err + } + + return &resp, nil +} + func (p *RemotePool) Disconnect(ctx context.Context) error { signedReq := request.NodeRequest{ Method: "vipnode_disconnect", diff --git a/pool/service.go b/pool/service.go index 3c4210a..980f27e 100644 --- a/pool/service.go +++ b/pool/service.go @@ -7,6 +7,7 @@ import ( "sync" "time" + "github.com/vipnode/vipnode/ethnode" "github.com/vipnode/vipnode/internal/pretty" "github.com/vipnode/vipnode/jsonrpc2" "github.com/vipnode/vipnode/pool/balance" @@ -30,9 +31,12 @@ func New(storeDriver store.Store, manager balance.Manager) *VipnodePool { manager = balance.NoBalance{} } return &VipnodePool{ - Store: storeDriver, - BalanceManager: manager, - remoteHosts: map[store.NodeID]jsonrpc2.Service{}, + Version: "dev", + + Store: storeDriver, + BalanceManager: manager, + remoteHosts: map[store.NodeID]jsonrpc2.Service{}, + remoteNodeLookup: map[jsonrpc2.Service]store.NodeID{}, } } @@ -46,12 +50,39 @@ type VipnodePool struct { Store store.Store BalanceManager balance.Manager ClientMessager func(nodeID string) string - MaxRequestHosts int // MaxRequestHosts is the maximum number of hosts a client is allowed to request (0 is unlimited) + MaxRequestHosts int // MaxRequestHosts is the maximum number of hosts a client is allowed to request (0 is unlimited) + RestrictNetwork ethnode.NetworkID // TODO: Wire this up + skipWhitelist bool // skipWhitelist is used for testing. + + mu sync.Mutex + remoteHosts map[store.NodeID]jsonrpc2.Service + remoteNodeLookup map[jsonrpc2.Service]store.NodeID // Reverse lookup +} - skipWhitelist bool // skipWhitelist is used for testing. +// TODO: Move CloseRemote and NumRemotes, and remoteHosts etc into a separate struct? - mu sync.Mutex - remoteHosts map[store.NodeID]jsonrpc2.Service +// CloseRemote is to be called when a remote service is disconnected. It is used to clean up state. +func (p *VipnodePool) CloseRemote(remote jsonrpc2.Service) error { + p.mu.Lock() + defer p.mu.Unlock() + + nodeID, ok := p.remoteNodeLookup[remote] + if !ok { + // Nothing to clean up + return nil + } + + delete(p.remoteNodeLookup, remote) + delete(p.remoteHosts, nodeID) + + return nil +} + +// NumRemotes returns the number of remote hosts that the pool is currently maintaining. +func (p *VipnodePool) NumRemotes() int { + p.mu.Lock() + defer p.mu.Unlock() + return len(p.remoteHosts) } func (p *VipnodePool) verify(sig string, method string, nodeID string, nonce int64, args ...interface{}) error { @@ -113,8 +144,17 @@ func (p *VipnodePool) Update(ctx context.Context, sig string, nodeID string, non } nodeBeforeUpdate := *node - peers := req.Peers - inactive, err := p.Store.UpdateNodePeers(store.NodeID(nodeID), peers, req.BlockNumber) + peerTypes := segmentPeers(req.PeerInfo) + if peerTypes.Count == 0 { + // DEPRECATED: Remove this backport for req.Peers once deprecation is complete. + peerTypes.Clients = req.Peers + peerTypes.Count = len(req.Peers) + } + + // FIXME: We only consider peers who are light clients, and effectively + // ignore host peers. It may be useful to take hosts into account in the + // future too. + inactive, err := p.Store.UpdateNodePeers(store.NodeID(nodeID), peerTypes.Clients, req.BlockNumber) if err != nil { return nil, err } @@ -130,9 +170,6 @@ func (p *VipnodePool) Update(ctx context.Context, sig string, nodeID string, non return nil, err } - // FIXME: Is there a bug here when a host is connected to another host? - // TODO: Test InvalidPeers - nodeBalance, err := p.BalanceManager.OnUpdate(nodeBeforeUpdate, validPeers) if err != nil { if _, ok := err.(balance.LowBalanceError); ok { @@ -148,78 +185,100 @@ func (p *VipnodePool) Update(ctx context.Context, sig string, nodeID string, non resp.Balance = &nodeBalance if node.IsHost { - logger.Printf("Host update %q: %d peers, %d active, %d invalid. %s", pretty.Abbrev(nodeID), len(peers), len(validPeers), len(inactive), nodeBalance.String()) + logger.Printf("Host update %q: %d peers, %d active, %d invalid. %s", pretty.Abbrev(nodeID), peerTypes.Count, len(validPeers), len(inactive), nodeBalance.String()) } else { - logger.Printf("Client update %q: %d peers, %d active, %d invalid. %s", pretty.Abbrev(nodeID), len(peers), len(validPeers), len(inactive), nodeBalance.String()) + logger.Printf("Client update %q: %d peers, %d active, %d invalid. %s", pretty.Abbrev(nodeID), peerTypes.Count, len(validPeers), len(inactive), nodeBalance.String()) } return &resp, nil } // Host registers a full node to participate as a vipnode host in this pool. +// DEPRECATED: Use Connect func (p *VipnodePool) Host(ctx context.Context, sig string, nodeID string, nonce int64, req HostRequest) (*HostResponse, error) { - // TODO: Send capabilities? + // This is a backport of Host using Connect behind the scenes. if err := p.verify(sig, "vipnode_host", nodeID, nonce, req); err != nil { return nil, err } - service, err := jsonrpc2.CtxService(ctx) + connectReq := ConnectRequest{ + NodeInfo: ethnode.UserAgent{ + Kind: ethnode.ParseNodeKind(req.Kind), + IsFullNode: true, + }, + Payout: req.Payout, + NodeURI: req.NodeURI, + } + connectResp, err := p.connect(ctx, nodeID, connectReq) if err != nil { return nil, err } - remoteHost := "" - if withAddr, ok := service.(interface{ RemoteAddr() string }); ok { - remoteHost = (&url.URL{Host: withAddr.RemoteAddr()}).Hostname() + resp := &HostResponse{ + PoolVersion: connectResp.PoolVersion, } - defaultPort := "30303" - nodeURI, err := normalizeNodeURI(req.NodeURI, nodeID, remoteHost, defaultPort) + return resp, nil +} + +// Client returns a list of enodes who are ready for the client node to connect. +// DEPRECATED: Use Connect +func (p *VipnodePool) Client(ctx context.Context, sig string, nodeID string, nonce int64, req ClientRequest) (*ClientResponse, error) { + // This is a backport of Client using Connect behind the scenes. + if err := p.verify(sig, "vipnode_client", nodeID, nonce, req); err != nil { + return nil, err + } + connectReq := ConnectRequest{ + NodeInfo: ethnode.UserAgent{ + Kind: ethnode.ParseNodeKind(req.Kind), + IsFullNode: false, + }, + } + connectResp, err := p.connect(ctx, nodeID, connectReq) if err != nil { return nil, err } - // TODO: Confirm that it's a full node, not a light node? Doesn't super matter since if i - // TODO: Check versions? - - logger.Printf("New %q host: %q", req.Kind, nodeURI) - - node := store.Node{ - ID: store.NodeID(nodeID), - URI: nodeURI, - Kind: req.Kind, - LastSeen: time.Now(), - IsHost: true, - Payout: store.Account(req.Payout), - } - err = p.Store.SetNode(node) + // Clients have a default number of hosts they request. Hosts don't. + numRequestHosts := defaultRequestNumHosts + if req.NumHosts > 0 { + numRequestHosts = req.NumHosts + } + hosts, err := p.requestHosts(ctx, nodeID, numRequestHosts, req.Kind) if err != nil { return nil, err } - // FIXME: Clean up disconnected hosts - p.mu.Lock() - p.remoteHosts[node.ID] = service - p.mu.Unlock() - - resp := &HostResponse{ - PoolVersion: p.Version, + resp := &ClientResponse{ + Hosts: hosts, + PoolVersion: connectResp.PoolVersion, + Message: connectResp.Message, } return resp, nil } -// Client returns a list of enodes who are ready for the client node to connect. -func (p *VipnodePool) Client(ctx context.Context, sig string, nodeID string, nonce int64, req ClientRequest) (*ClientResponse, error) { - if err := p.verify(sig, "vipnode_client", nodeID, nonce, req); err != nil { +// Connect returns a list of enodes who are ready for the client node to connect. +func (p *VipnodePool) Connect(ctx context.Context, sig string, nodeID string, nonce int64, req ConnectRequest) (*ConnectResponse, error) { + if err := p.verify(sig, "vipnode_connect", nodeID, nonce, req); err != nil { return nil, err } - kind := req.Kind - numRequestHosts := defaultRequestNumHosts - if p.MaxRequestHosts > 0 && numRequestHosts > p.MaxRequestHosts { - numRequestHosts = p.MaxRequestHosts + return p.connect(ctx, nodeID, req) +} + +// connect is same as Connect without signature verification. Used as a helper. +// TODO: We can inline connect into Connect once Client/Host are removed. +func (p *VipnodePool) connect(ctx context.Context, nodeID string, req ConnectRequest) (*ConnectResponse, error) { + kind := req.NodeInfo.Kind.String() + if kind == "unknown" { + kind = "" } - response := &ClientResponse{ + isHost := req.NodeInfo.IsFullNode + if p.RestrictNetwork != 0 && p.RestrictNetwork != req.NodeInfo.Network { + return nil, fmt.Errorf("node is on the wrong network, pool requires: %s", p.RestrictNetwork) + } + + response := &ConnectResponse{ PoolVersion: p.Version, } if p.ClientMessager != nil { @@ -230,11 +289,39 @@ func (p *VipnodePool) Client(ctx context.Context, sig string, nodeID string, non // get successfully whitelisted, then switched to host status thus bypass // billing? node := store.Node{ - ID: store.NodeID(nodeID), - Kind: kind, - LastSeen: time.Now(), - IsHost: false, + ID: store.NodeID(nodeID), + Kind: kind, + LastSeen: time.Now(), + IsHost: isHost, + Payout: store.Account(req.Payout), + NodeVersion: req.NodeInfo.Version, + VipnodeVersion: req.VipnodeVersion, + } + + if isHost { + // Hosts expose a reverse-RPC for vipnode_whitelist. + service, err := jsonrpc2.CtxService(ctx) + if err != nil { + return nil, err + } + + // We only care about publicly-visible nodeURIs for hosts. + remoteHost := "" + if withAddr, ok := service.(interface{ RemoteAddr() string }); ok { + remoteHost = (&url.URL{Host: withAddr.RemoteAddr()}).Hostname() + } + defaultPort := "30303" + node.URI, err = normalizeNodeURI(req.NodeURI, nodeID, remoteHost, defaultPort) + if err != nil { + return nil, err + } + + p.mu.Lock() + p.remoteHosts[node.ID] = service + p.remoteNodeLookup[service] = node.ID + p.mu.Unlock() } + if err := p.Store.SetNode(node); err != nil { return nil, err } @@ -242,26 +329,58 @@ func (p *VipnodePool) Client(ctx context.Context, sig string, nodeID string, non if err := p.BalanceManager.OnClient(node); err != nil { return nil, err } + logger.Printf("New %q peer: %q", kind, pretty.Abbrev(nodeID)) - r, err := p.Store.ActiveHosts(kind, numRequestHosts) + return response, nil +} + +// Peer returns a list of enodes who are ready for the node to connect. +func (p *VipnodePool) Peer(ctx context.Context, sig string, nodeID string, nonce int64, req PeerRequest) (*PeerResponse, error) { + // TODO: Should we use protocol capability (eth, les, pip) instead of Kind? + // It's hard to get self-reported protocol capability versions though (les/2 vs just les). + hosts, err := p.requestHosts(ctx, nodeID, req.Num, req.Kind) if err != nil { return nil, err } - if len(r) == 0 { - logger.Printf("New %q client: %q (no active hosts found)", kind, pretty.Abbrev(nodeID)) - return nil, NoHostNodesError{} + + // TODO: Move logs here. + + response := &PeerResponse{ + Peers: hosts, + } + return response, nil + +} + +func (p *VipnodePool) requestHosts(ctx context.Context, nodeID string, numRequestHosts int, kind string) ([]store.Node, error) { + if p.MaxRequestHosts > 0 && numRequestHosts > p.MaxRequestHosts { + numRequestHosts = p.MaxRequestHosts + } + + var hosts []store.Node + if numRequestHosts == 0 { + // Nothing left to do + return hosts, nil + } + + r, err := p.Store.ActiveHosts(kind, numRequestHosts) + if err != nil { + return nil, err } if p.skipWhitelist { - logger.Printf("New %q client: %q (%d hosts found, skipping whitelist)", kind, pretty.Abbrev(nodeID), len(r)) - response.Hosts = r - return response, nil + // Bypass whitelisting, used for making testing simpler + return r, nil } errors := []error{} remotes := make([]hostService, 0, len(r)) p.mu.Lock() for _, node := range r { + if node.ID.String() == nodeID { + // Skip self + continue + } remote, ok := p.remoteHosts[node.ID] if ok { remotes = append(remotes, hostService{ @@ -302,18 +421,19 @@ func (p *VipnodePool) Client(ctx context.Context, sig string, nodeID string, non // TODO: Penalize hosts that failed to respond within the deadline? if len(errors) > 0 { - logger.Printf("New %q client: %s (%d hosts found, %d accepted) %s", kind, nodeID[:8], len(remotes), len(accepted), RemoteHostErrors{"vipnode_whitelist", errors}) + err = RemoteHostErrors{"vipnode_whitelist", errors} + logger.Printf("Request %q hosts: %q (%d hosts found, %d accepted); failures: %s", kind, pretty.Abbrev(nodeID), len(remotes), len(accepted), err) } else { - logger.Printf("New %q client: %s (%d hosts found, %d accepted)", kind, nodeID[:8], len(remotes), len(accepted)) + logger.Printf("Request %q hosts: %q (%d hosts found, %d accepted)", kind, pretty.Abbrev(nodeID), len(remotes), len(accepted)) } if len(accepted) >= 1 { - response.Hosts = accepted - return response, nil + // We're okay returning without an error as long as some hosts succeeded. + return accepted, nil } - if len(errors) > 0 { - return nil, RemoteHostErrors{"vipnode_whitelist", errors} + if err != nil { + return nil, err } return nil, NoHostNodesError{len(r)} @@ -323,3 +443,28 @@ func (p *VipnodePool) Client(ctx context.Context, sig string, nodeID string, non func (p *VipnodePool) Ping(ctx context.Context) string { return "pong" } + +// peerTypes stores the subset of peer IDs by type, derived from the capabilities. +type peerTypes struct { + Count int + Hosts []string + Clients []string +} + +func segmentPeers(peers []ethnode.PeerInfo) peerTypes { + r := peerTypes{} + for _, p := range peers { + r.Count++ + + if _, ok := p.Protocols["eth"]; ok { + // Hosts have eth/62 or eth/63 capability (comes up as "eth" protocol) + r.Hosts = append(r.Hosts, p.ID) + } else if len(p.Protocols) > 0 { + // Anything else is probably "les" or "pip" + r.Clients = append(r.Clients, p.ID) + } + // Empty p.Protocols means the peer has not completed the handshake yet, + // so we can ignore them. + } + return r +} diff --git a/pool/service_test.go b/pool/service_test.go index 0e8204f..1cef808 100644 --- a/pool/service_test.go +++ b/pool/service_test.go @@ -71,7 +71,9 @@ func TestPoolService(t *testing.T) { t.Fatal(err) } var result interface{} - if err := client.Call(context.TODO(), &result, req.Method, args...); err.Error() != (NoHostNodesError{}).Error() { + if err := client.Call(context.TODO(), &result, req.Method, args...); err == nil { + t.Error("expected NoHostsNodeError, got nil") + } else if err.Error() != (NoHostNodesError{}).Error() { t.Error(err) } } diff --git a/pool/staticpool.go b/pool/staticpool.go index cd5c328..1e88f99 100644 --- a/pool/staticpool.go +++ b/pool/staticpool.go @@ -32,6 +32,17 @@ func (s *StaticPool) Client(ctx context.Context, req ClientRequest) (*ClientResp return &ClientResponse{Hosts: s.Nodes}, nil } +func (s *StaticPool) Connect(ctx context.Context, req ConnectRequest) (*ConnectResponse, error) { + return &ConnectResponse{ + PoolVersion: "staticpool", + Hosts: s.Nodes, + }, nil +} + +func (s *StaticPool) Peer(ctx context.Context, req PeerRequest) (*PeerResponse, error) { + return &PeerResponse{Peers: s.Nodes}, nil +} + func (s *StaticPool) Disconnect(ctx context.Context) error { return nil } diff --git a/pool/status/status.go b/pool/status/status.go index ab36157..4afea7b 100644 --- a/pool/status/status.go +++ b/pool/status/status.go @@ -19,11 +19,13 @@ type Host struct { LastSeen time.Time `json:"last_seen"` Kind string `json:"kind"` BlockNumber uint64 `json:"block_number"` + NumPeers int `json:"num_peers"` - // TODO: Add peers + NodeVersion string `json:"node_version"` + VipnodeVersion string `json:"vipnode_version"` } -func nodeHost(n store.Node) Host { +func nodeHost(n store.Node, numPeers int) Host { shortID := string(n.ID) if len(shortID) > 12 { shortID = shortID[:12] @@ -33,6 +35,10 @@ func nodeHost(n store.Node) Host { LastSeen: n.LastSeen, Kind: n.Kind, BlockNumber: n.BlockNumber, + NumPeers: numPeers, + + NodeVersion: n.NodeVersion, + VipnodeVersion: n.VipnodeVersion, } } @@ -117,7 +123,12 @@ func (s *PoolStatus) getStatus() (*StatusResponse, error) { r.ActiveHosts = make([]Host, 0, len(nodes)) for _, n := range nodes { - r.ActiveHosts = append(r.ActiveHosts, nodeHost(n)) + peers, err := s.Store.NodePeers(n.ID) + if err != nil { + r.Error = err + return r, err + } + r.ActiveHosts = append(r.ActiveHosts, nodeHost(n, len(peers))) } return r, nil diff --git a/pool/store/store.go b/pool/store/store.go index 29dc913..1c9168f 100644 --- a/pool/store/store.go +++ b/pool/store/store.go @@ -63,6 +63,9 @@ type Node struct { IsHost bool Payout Account BlockNumber uint64 `json:"block_number"` + + NodeVersion string `json:"node_version"` + VipnodeVersion string `json:"vipnode_version"` } // Stats contains various aggregate stats of the store state, used for diff --git a/poolhostclient_test.go b/poolhostclient_test.go index 289c02f..96ab0b8 100644 --- a/poolhostclient_test.go +++ b/poolhostclient_test.go @@ -1,6 +1,8 @@ package main import ( + "context" + "crypto/ecdsa" "fmt" "io" "net" @@ -8,8 +10,8 @@ import ( "testing" "github.com/ethereum/go-ethereum/p2p/discv5" - "github.com/vipnode/vipnode/client" - "github.com/vipnode/vipnode/host" + "github.com/vipnode/vipnode/agent" + "github.com/vipnode/vipnode/internal/fakecluster" "github.com/vipnode/vipnode/internal/fakenode" "github.com/vipnode/vipnode/internal/keygen" "github.com/vipnode/vipnode/jsonrpc2" @@ -21,19 +23,19 @@ func TestPoolHostClient(t *testing.T) { privkey := keygen.HardcodedKeyIdx(t, 0) payout := "" - p := pool.New(memory.New(), nil) + db := memory.New() + p := pool.New(db, nil) rpcPool2Host, rpcHost2Pool := jsonrpc2.ServePipe() defer rpcPool2Host.Close() defer rpcHost2Pool.Close() if err := rpcPool2Host.Server.Register("vipnode_", p); err != nil { t.Fatalf("failed to register vipnode_ rpc for pool: %s", err) } - hostNodeID := discv5.PubkeyID(&privkey.PublicKey).String() hostNode := fakenode.Node(hostNodeID) hostNodeURI := fmt.Sprintf("enode://%s@127.0.0.1:30303", hostNodeID) - h := host.New(hostNode, payout) - if err := rpcHost2Pool.Server.RegisterMethod("vipnode_whitelist", h, "Whitelist"); err != nil { + h := agent.Agent{EthNode: hostNode, Payout: payout} + if err := rpcHost2Pool.Server.RegisterMethod("vipnode_whitelist", &h, "Whitelist"); err != nil { t.Fatalf("failed to register vipnode_ rpc for host: %s", err) } h.NodeURI = hostNodeURI @@ -44,6 +46,12 @@ func TestPoolHostClient(t *testing.T) { } defer h.Stop() + if stats, err := db.Stats(); err != nil { + t.Fatal(err) + } else if stats.NumActiveHosts != 1 { + t.Errorf("wrong number of active hosts: %+v", stats) + } + rpcPool2Client, rpcClient2Pool := jsonrpc2.ServePipe() defer rpcPool2Client.Close() defer rpcClient2Pool.Close() @@ -52,7 +60,11 @@ func TestPoolHostClient(t *testing.T) { clientPrivkey := keygen.HardcodedKeyIdx(t, 1) clientNodeID := discv5.PubkeyID(&clientPrivkey.PublicKey).String() clientNode := fakenode.Node(clientNodeID) - c := client.New(clientNode) + clientNode.IsFullNode = false + c := agent.Agent{ + EthNode: clientNode, + NumHosts: 3, + } clientPool := pool.Remote(rpcClient2Pool, clientPrivkey) if err := c.Start(clientPool); err != nil { t.Fatalf("failed to start client: %s", err) @@ -76,7 +88,6 @@ func TestPoolHostClient(t *testing.T) { } want = fakenode.Calls{ fakenode.Call("ConnectPeer", hostNodeURI), - fakenode.Call("DisconnectPeer", hostNodeURI), } if got := clientNode.Calls; !reflect.DeepEqual(got, want) { t.Errorf("clientNode.Calls:\n got %q;\n want %q", got, want) @@ -88,7 +99,6 @@ func TestPoolHostClient(t *testing.T) { } want = fakenode.Calls{ fakenode.Call("ConnectPeer", hostNodeURI), - fakenode.Call("DisconnectPeer", hostNodeURI), fakenode.Call("ConnectPeer", hostNodeURI), } if got := clientNode.Calls; !reflect.DeepEqual(got, want) { @@ -135,8 +145,8 @@ func TestCloseHost(t *testing.T) { hostNodeID := discv5.PubkeyID(&privkey.PublicKey).String() hostNode := fakenode.Node(hostNodeID) hostNodeURI := fmt.Sprintf("enode://%s@127.0.0.1", hostNodeID) - h := host.New(hostNode, payout) - if err := rpcHost2Pool.Server.RegisterMethod("vipnode_whitelist", h, "Whitelist"); err != nil { + h := agent.Agent{EthNode: hostNode, Payout: payout} + if err := rpcHost2Pool.Server.RegisterMethod("vipnode_whitelist", &h, "Whitelist"); err != nil { t.Fatalf("failed to register vipnode_ rpc for host: %s", err) } h.NodeURI = hostNodeURI @@ -154,3 +164,89 @@ func TestCloseHost(t *testing.T) { h.Stop() h.Wait() } + +func TestPoolHostConnectPeers(t *testing.T) { + hostKeys := []*ecdsa.PrivateKey{} + clientKeys := []*ecdsa.PrivateKey{} + + for i := 0; i < 4; i++ { + hostKeys = append(hostKeys, keygen.HardcodedKeyIdx(t, i)) + } + + cluster, err := fakecluster.New(hostKeys, clientKeys) + if err != nil { + t.Fatal(err) + } + + if got, want := len(cluster.Hosts), 4; got != want { + t.Errorf("wrong number of hosts: got %d; want %d", got, want) + } + + if got, want := cluster.Pool.NumRemotes(), 4; got != want { + t.Errorf("wrong number of remotes: got %d; want %d", got, want) + } + + stats, err := cluster.Pool.Store.Stats() + if err != nil { + t.Error(err) + } + + if stats.NumActiveHosts != 4 { + t.Errorf("wrong stats: %+v", stats) + } + + numHosts := len(hostKeys) + { + // Connect a host to all the peers (numHosts = peers + 1, because it includes self) + host := cluster.Hosts[0] + if peers, err := host.Node.Peers(context.Background()); err != nil { + t.Fatal(err) + } else if len(peers) > 0 { + t.Errorf("host has unexpected peers: %s", peers) + } + + // Request peers + if err := host.AddPeers(context.Background(), host.RemotePool, 10); err != nil { + t.Fatal(err) + } + + if peers, err := host.Node.Peers(context.Background()); err != nil { + t.Fatal(err) + } else if got, want := len(peers), numHosts-1; got != want { + t.Errorf("host has wrong number of peers: got %d; want %d", got, want) + } + + numCalls := 0 + for _, call := range host.Node.Calls { + if call.Method != "ConnectPeer" { + t.Errorf("unexpected call to host: %s", call) + continue + } + numCalls++ + } + if got, want := numCalls, numHosts-1; got != want { + t.Errorf("wrong number of ConnectPeer calls: got %d; want %d", got, want) + } + } + + { + // Check a peer + host := cluster.Hosts[1] + peer := cluster.Hosts[0] + want := fakenode.Calls{fakenode.Call("AddTrustedPeer", peer.Node.NodeID)} + if got := host.Node.Calls; !reflect.DeepEqual(got, want) { + t.Errorf("peer host calls do not match:\n got: %s;\n want: %s", got, want) + } + // We can't check the Peers of the host here because it's a FakeNode so + // it does not actually initiate a connection on Connect. + } + + err = cluster.Close() + if err != nil { + t.Error(err) + } + + if got, want := cluster.Pool.NumRemotes(), 0; got != want { + t.Errorf("wrong number of remotes: got %d; want %d", got, want) + } +} diff --git a/server.go b/server.go index e34ba10..34ea1aa 100644 --- a/server.go +++ b/server.go @@ -14,9 +14,10 @@ type wsHandler interface { type server struct { jsonrpc2.HTTPServer - ws ws.Upgrader - debugLog bool - header http.Header + ws ws.Upgrader + debugLog bool + header http.Header + onDisconnect func(remote jsonrpc2.Service) error } func (s *server) ServeHTTP(w http.ResponseWriter, r *http.Request) { @@ -43,6 +44,8 @@ func (s *server) ServeHTTP(w http.ResponseWriter, r *http.Request) { if s.debugLog { codec = jsonrpc2.DebugCodec(r.RemoteAddr, codec) } + defer codec.Close() + remote := &jsonrpc2.Remote{ Codec: codec, Server: &s.HTTPServer.Server, @@ -54,6 +57,12 @@ func (s *server) ServeHTTP(w http.ResponseWriter, r *http.Request) { if err := remote.Serve(); err != nil && err != io.EOF { logger.Warningf("jsonrpc2.Remote.Serve() error: %s", err) } + + if s.onDisconnect != nil { + if err := s.onDisconnect(remote); err != nil { + logger.Warningf("jsonrpc2.Service disconnect error: %s", err) + } + } default: http.Error(w, "unsupported method", http.StatusUnsupportedMediaType) }