les: lespay API

les/api.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"reflect"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common/hexutil"
@@ -376,10 +377,13 @@ func (api *PrivateLespayAPI) makeCall(ctx context.Context, remote bool, nodeStr
 		freeID string
 		peer   *peer
 		node   *enode.Node
+		err    error
 	)
 	if nodeStr != "" {
-		if peer = api.peerSet.Peer(nodeStr); peer != nil {
-			id = peer.ID()
+		if id, err = enode.ParseID(nodeStr); err == nil {
+			if peer = api.peerSet.Peer(peerIdToString(id)); peer == nil {
+				return nil, errors.New("peer not connected")
+			}
 			freeID = peer.freeClientId()
 		} else {
 			var err error
@@ -416,18 +420,25 @@ func (api *PrivateLespayAPI) makeCall(ctx context.Context, remote bool, nodeStr
 				return nil, errors.New("UDP DHT not available")
 			}
 			cancelFn = api.dht.SendTalkRequest(node, "lespay", [][]byte{cmd}, func(payload interface{}) bool {
-				if replies, ok := payload.([][]byte); ok && len(replies) == 1 {
-					reply = replies[0]
+				fmt.Println("dht delivered", payload, reflect.TypeOf(payload))
+				if replies, ok := payload.([]interface{}); ok && len(replies) == 1 {
+					reply, ok = replies[0].([]byte)
 				}
 				close(delivered)
 				return reply != nil
 			})
 		}
 		select {
+		case <-time.After(time.Second * 5):
+			cancelFn()
+			return nil, errors.New("timeout")
 		case <-ctx.Done():
 			cancelFn()
 			return nil, ctx.Err()
 		case <-delivered:
+			if len(reply) == 0 {
+				return nil, errors.New("unknown command")
+			}
 			return reply, nil
 		}
 	} else {
@@ -444,10 +455,13 @@ func (api *PrivateLespayAPI) Connection(ctx context.Context, remote bool, node s
 	params := tsConnectionParams{requestedCapacity, stayConnected, paymentModule, setCap}
 	enc, _ := rlp.EncodeToBytes(&params)
 	var resEnc []byte
-	resEnc, err = api.makeCall(ctx, remote, node, enc)
+	fmt.Println("makeCall", remote, node, enc)
+	resEnc, err = api.makeCall(ctx, remote, node, append([]byte{tsConnection}, enc...))
 	if err != nil {
+		fmt.Println("makeCall err", err)
 		return
 	}
 	err = rlp.DecodeBytes(resEnc, &results)
+	fmt.Println("decode err", err)
 	return
 }

les/client_handler.go

@@ -17,6 +17,7 @@
 package les
 
 import (
+	"fmt"
 	"math/big"
 	"sync"
 	"time"
@@ -316,19 +317,24 @@ func (h *clientHandler) handleMsg(p *peer) error {
 		p.freezeServer(false)
 		p.Log().Debug("Service resumed")
 	case LespayReplyMsg:
+		fmt.Println("LespayReply received")
 		p.Log().Trace("Received tx status response")
 		var resp struct {
 			ReqID   uint64
 			Replies [][]byte
 		}
 		if err := msg.Decode(&resp); err != nil {
+			fmt.Println("LespayReply decode err", err)
 			return errResp(ErrDecode, "msg %v: %v", msg, err)
 		}
+		fmt.Println("LespayReply decoded", resp)
 		h.lespayReplyLock.Lock()
 		if handler := h.lespayReplyHandlers[resp.ReqID]; handler != nil {
+			fmt.Println("handler found")
 			delete(h.lespayReplyHandlers, resp.ReqID)
 			responseError = !handler(resp.Replies)
 		} else {
+			fmt.Println("handler not found")
 			responseError = true
 		}
 		h.lespayReplyLock.Unlock()
@@ -354,9 +360,16 @@ func (h *clientHandler) handleMsg(p *peer) error {
 
 func (h *clientHandler) makeLespayCall(p *peer, cmds [][]byte, handler func([][]byte) bool) func() bool {
 	reqID := genReqID()
+	h.lespayReplyLock.Lock()
+	h.lespayReplyHandlers[reqID] = handler
+	h.lespayReplyLock.Unlock()
 	if p.SendLespay(reqID, cmds) != nil {
+		h.lespayReplyLock.Lock()
+		delete(h.lespayReplyHandlers, reqID)
+		h.lespayReplyLock.Unlock()
 		return nil
 	}
+	fmt.Println("Lespay sent")
 	return func() bool {
 		h.lespayReplyLock.Lock()
 		cancel := h.lespayReplyHandlers[reqID] != nil

les/clientpool.go

@@ -594,19 +594,21 @@ func (f *clientPool) setCapacity(id enode.ID, freeID string, capacity uint64, mi
 	if missing != 0 {
 		return missing, capacity, errNoPriority
 	}
-	if setCap && c == nil {
-		return missing, capacity, fmt.Errorf("client %064x is not connected", c.id[:])
-	}
 	// capacity update is possible
-	f.connectedCap += capacity - c.capacity
-	f.updateFullRatio()
-	f.priorityConnected += capacity - c.capacity
-	c.capacity = capacity
-	c.balanceTracker.setCapacity(capacity)
-	f.connectedQueue.Update(c.queueIndex)
-	totalConnectedGauge.Update(int64(f.connectedCap))
-	updatePriceFactors(&c.balanceTracker, c.posFactors, c.negFactors, c.capacity)
-	c.peer.updateCapacity(c.capacity)
+	if setCap {
+		if c == nil {
+			return 0, capacity, fmt.Errorf("client %064x is not connected", c.id[:])
+		}
+		f.connectedCap += capacity - c.capacity
+		f.updateFullRatio()
+		f.priorityConnected += capacity - c.capacity
+		c.capacity = capacity
+		c.balanceTracker.setCapacity(capacity)
+		f.connectedQueue.Update(c.queueIndex)
+		totalConnectedGauge.Update(int64(f.connectedCap))
+		updatePriceFactors(&c.balanceTracker, c.posFactors, c.negFactors, c.capacity)
+		c.peer.updateCapacity(c.capacity)
+	}
 	return 0, capacity, nil
 }
 

les/server_handler.go

@@ -20,7 +20,9 @@ import (
 	"encoding/binary"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"net"
+	"reflect"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -130,7 +132,7 @@ func (h *serverHandler) handle(p *peer) error {
 	}
 	// Reject light clients if server is not synced.
 	if !h.synced() {
-		return p2p.DiscRequested
+		//return p2p.DiscRequested
 	}
 	defer p.fcClient.Disconnect()
 
@@ -982,11 +984,21 @@ func (h *serverHandler) broadcastHeaders() {
 }
 
 func (h *serverHandler) talkRequestHandler(id enode.ID, addr *net.UDPAddr, payload interface{}) (interface{}, bool) {
-	cmds, ok := payload.([][]byte)
+	fmt.Println("talkRequestHandler", id, addr, payload, reflect.TypeOf(payload))
+	c, ok := payload.([]interface{})
 	if !ok {
 		return nil, false
 	}
+	cmds := make([][]byte, len(c))
+	for i, c := range c {
+		cmds[i], ok = c.([]byte)
+		if !ok {
+			fmt.Println("type err", reflect.TypeOf(c))
+			return nil, false
+		}
+	}
+	fmt.Println("ok", ok)
 	results := h.server.tokenSale.runCommands(cmds, id, addr.IP.String())
-	res, _ := rlp.EncodeToBytes(&results)
-	return res, true
+	fmt.Println("results", results)
+	return results, true
 }

les/tokensale.go

@@ -367,7 +367,9 @@ func (t *tokenSale) runCommand(cmd []byte, id enode.ID, freeID string) []byte {
 		)
 		if err := rlp.DecodeBytes(cmd[1:], &params); err == nil {
 			results.PcValue, results.PcBalance, err = t.deposit(id, params.PaymentModule, params.ProofOfPayment)
-			results.Err = err.Error()
+			if err != nil {
+				results.Err = err.Error()
+			}
 			res, _ = rlp.EncodeToBytes(&results)
 		}
 	case tsBuyTokens:
@@ -388,7 +390,9 @@ func (t *tokenSale) runCommand(cmd []byte, id enode.ID, freeID string) []byte {
 		if err := rlp.DecodeBytes(cmd[1:], &params); err == nil {
 			results.AvailableCapacity, results.TokenBalance, results.TokensMissing, results.PcBalance, results.PcMissing, results.PaymentRequired, err =
 				t.connection(id, freeID, params.RequestedCapacity, time.Duration(params.StayConnected)*time.Second, params.PaymentModule, params.SetCap)
-			results.Err = err.Error()
+			if err != nil {
+				results.Err = err.Error()
+			}
 			res, _ = rlp.EncodeToBytes(&results)
 		}
 	}

p2p/discv5/net.go

@@ -432,9 +432,11 @@ loop:
 
 		// Ingress packet handling.
 		case pkt := <-net.read:
-			//fmt.Println("read", pkt.ev)
 			log.Trace("<-net.read")
 			n := net.internNode(&pkt)
+			if pkt.ev == talkRequestPacket {
+				fmt.Println("read trp", n.state, pkt)
+			}
 			prestate := n.state
 			status := "ok"
 			if err := net.handle(n, pkt.ev, &pkt); err != nil {
@@ -722,11 +724,17 @@ func (net *Network) refresh(done chan<- struct{}) {
 
 func (net *Network) internNode(pkt *ingressPacket) *Node {
 	if n := net.nodes[pkt.remoteID]; n != nil {
+		if pkt.ev == talkRequestPacket {
+			fmt.Println("node exists")
+		}
 		n.IP = pkt.remoteAddr.IP
 		n.UDP = uint16(pkt.remoteAddr.Port)
 		n.TCP = uint16(pkt.remoteAddr.Port)
 		return n
 	}
+	if pkt.ev == talkRequestPacket {
+		fmt.Println("node created")
+	}
 	n := NewNode(pkt.remoteID, pkt.remoteAddr.IP, uint16(pkt.remoteAddr.Port), uint16(pkt.remoteAddr.Port))
 	n.state = unknown
 	net.nodes[pkt.remoteID] = n
@@ -1061,7 +1069,7 @@ func (net *Network) handle(n *Node, ev nodeEvent, pkt *ingressPacket) error {
 	//fmt.Println("handle", n.addr().String(), n.state, ev)
 	if pkt != nil {
 		if err := net.checkPacket(n, ev, pkt); err != nil {
-			//fmt.Println("check err:", err)
+			//fmt.Println("check err:", err, pkt)
 			return err
 		}
 		// Start the background expiration goroutine after the first
@@ -1076,6 +1084,7 @@ func (net *Network) handle(n *Node, ev nodeEvent, pkt *ingressPacket) error {
 	if n.state == nil {
 		n.state = unknown //???
 	}
+	//fmt.Println("old state:", n.state)
 	next, err := n.state.handle(net, n, ev, pkt)
 	net.transition(n, next)
 	//fmt.Println("new state:", n.state)
@@ -1241,8 +1250,10 @@ func (net *Network) handleQueryEvent(n *Node, ev nodeEvent, pkt *ingressPacket)
 		net.talkRequestSubLock.RLock()
 		subFn := net.talkRequestSubs[string(p.TalkID)]
 		net.talkRequestSubLock.RUnlock()
+		fmt.Println("trp", p)
 		if subFn != nil {
 			resp, ok := subFn(enode.ID(n.sha), n.addr(), p.Payload)
+			fmt.Println("subFn", ok)
 			if ok {
 				net.conn.send(n, talkResponsePacket, talkResponse{ReplyTok: pkt.hash, Payload: resp})
 			} else {
@@ -1367,7 +1378,7 @@ func (q *talkQuery) start(net *Network) bool {
 		q.key = string(q.remote.sha[:]) + string(hash[:])
 		net.talkResponseSubs[q.key] = q.handler
 		net.talkResponseSubLock.Unlock()
-		fmt.Println("sent")
+		fmt.Println("sent", q.remote.state)
 		return true
 	}
 	// If the node is not known yet, it won't accept queries.
@@ -1397,6 +1408,9 @@ func (net *Network) SendTalkRequest(to *enode.Node, talkID string, payload inter
 		net.nodes[nodeID] = node
 	}
 	q := &talkQuery{remote: node, talkID: talkID, payload: payload, handler: handler}
+	if node.state == nil || !node.state.canQuery {
+		net.ping(node, node.addr())
+	}
 	select {
 	case net.queryReq <- q:
 	case <-net.closed:

p2p/enode/node.go

@@ -217,7 +217,7 @@ func (n ID) MarshalText() ([]byte, error) {
 
 // UnmarshalText implements the encoding.TextUnmarshaler interface.
 func (n *ID) UnmarshalText(text []byte) error {
-	id, err := parseID(string(text))
+	id, err := ParseID(string(text))
 	if err != nil {
 		return err
 	}
@@ -229,14 +229,14 @@ func (n *ID) UnmarshalText(text []byte) error {
 // The string may be prefixed with 0x.
 // It panics if the string is not a valid ID.
 func HexID(in string) ID {
-	id, err := parseID(in)
+	id, err := ParseID(in)
 	if err != nil {
 		panic(err)
 	}
 	return id
 }
 
-func parseID(in string) (ID, error) {
+func ParseID(in string) (ID, error) {
 	var id ID
 	b, err := hex.DecodeString(strings.TrimPrefix(in, "0x"))
 	if err != nil {