swarm/swap: very early cheques and test improvements

swarm/swap/api.go

@@ -18,13 +18,17 @@ package swap
 
 import (
 	"context"
+	"errors"
 	"math/big"
 
-	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/ethereum/go-ethereum/p2p/discover"
 )
 
+var (
+	ErrNoSuchPeerAccounting = errors.New("No accounting with that peer")
+)
+
 // Wrapper for receiving pss messages when using the pss API
 // providing access to sender of message
 type APIMsg struct {
@@ -34,32 +38,32 @@ type APIMsg struct {
 // Additional public methods accessible through API for pss
 type API struct {
 	*SwapProtocol
-	*Swap
 }
 
+//TODO: define metrics
 type SwapMetrics struct {
 }
 
-type Cheque struct {
-}
-
 func NewAPI(swap *SwapProtocol) *API {
 	return &API{SwapProtocol: swap}
 }
 
-func (swapapi *API) Balance(ctx context.Context, peer discover.NodeID) (balance *big.Int, err error) {
+func (swapapi *API) BalanceWithPeer(ctx context.Context, peer discover.NodeID) (balance *big.Int, err error) {
+	balance = swapapi.swap.peers[peer].balance
+	if balance == nil {
+		err = ErrNoSuchPeerAccounting
+	}
+	return
+}
+
+func (swapapi *API) Balance(ctx context.Context) (balance *big.Int, err error) {
 	balance = big.NewInt(0)
-	err = nil
+	for _, peer := range swapapi.swap.peers {
+		balance.Add(balance, peer.balance)
+	}
 	return
 }
 
 func (swapapi *API) GetSwapMetrics() (*SwapMetrics, error) {
 	return nil, nil
 }
-
-func (swapapi *API) IssueCheque(recipient *common.Address) (*Cheque, error) {
-	return nil, nil
-}
-
-func (swapapi *API) RedeemCheque(cheque *Cheque) {
-}

swarm/swap/chequemanager.go

@@ -0,0 +1,65 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package swap
+
+import (
+	"math/big"
+
+	"github.com/ethereum/go-ethereum/p2p/discover"
+	"github.com/ethereum/go-ethereum/swarm/state"
+)
+
+type ChequeManager struct {
+	stateStore        state.Store
+	serialPerNode     map[discover.NodeID]uint64
+	openDebitCheques  map[discover.NodeID][]*Cheque
+	openCreditCheques map[discover.NodeID][]*Cheque
+}
+
+type Cheque struct {
+	serial       uint64
+	timeout      uint64
+	amount       *big.Int
+	sumCumulated *big.Int
+	beneficiary  discover.NodeID //this should probably be common.Address?
+}
+
+func NewChequeManager(stateStore state.Store) *ChequeManager {
+	return &ChequeManager{
+		stateStore: stateStore,
+		//TODO: restore from state store
+		serialPerNode:     make(map[discover.NodeID]uint64),
+		openDebitCheques:  make(map[discover.NodeID][]*Cheque),
+		openCreditCheques: make(map[discover.NodeID][]*Cheque),
+	}
+}
+
+func (mgr *ChequeManager) CreateCheque(beneficiary discover.NodeID, amount *big.Int) *Cheque {
+	mgr.serialPerNode[beneficiary]++
+	cheque := &Cheque{
+		serial:      mgr.serialPerNode[beneficiary],
+		beneficiary: beneficiary,
+		amount:      amount,
+	}
+	openCheques := mgr.openDebitCheques[beneficiary]
+	if openCheques == nil {
+		openCheques = make([]*Cheque, 0)
+	}
+	openCheques = append(openCheques, cheque)
+	mgr.openDebitCheques[beneficiary] = openCheques
+	return cheque
+}

swarm/swap/protocol.go

@@ -36,6 +36,7 @@ const (
 type SwapProtocol struct {
 	peersMu sync.RWMutex
 	peers   map[discover.NodeID]*SwapProtocolPeer
+	swap    *Swap
 }
 
 // Peer is the Peer extension for the streaming protocol
@@ -44,9 +45,10 @@ type SwapProtocolPeer struct {
 	swapProtocol *SwapProtocol
 }
 
-func NewSwapProtocol() *SwapProtocol {
+func NewSwapProtocol(swapAccount *Swap) *SwapProtocol {
 	proto := &SwapProtocol{
 		peers: make(map[discover.NodeID]*SwapProtocolPeer),
+		swap:  swapAccount,
 	}
 	return proto
 }
@@ -61,6 +63,7 @@ func NewPeer(peer *protocols.Peer, swap *SwapProtocol) *SwapProtocolPeer {
 }
 
 type IssueChequeMsg struct {
+	Cheque *Cheque
 }
 
 type RedeemChequeMsg struct {

swarm/swap/swap.go

@@ -70,6 +70,7 @@ const (
 //      Swift Automatic  Payments
 // a peer to peer micropayment system
 type Swap struct {
+	chequeManager *ChequeManager
 	stateStore    state.Store
 	lock          sync.RWMutex
 	peers         map[discover.NodeID]*SwapPeer
@@ -244,14 +245,14 @@ func (sp *SwapPeer) AccountMsgForPeer(ctx context.Context, msg interface{}, pric
 		// -1 if sp.balance <  payAt
 		//  0 if sp.balance == payAt
 		// +1 if sp.balance >  payAt
-		if sp.balance.Cmp(payAt) > 1 {
+		if sp.balance.Cmp(payAt) == -1 {
 			err := sp.issueCheque(ctx)
 			if err != nil {
 				//TODO: special error handling, as at this point the accounting has been done
 				//but the cheque could not be sent?
 			}
 		}
-		if sp.balance.Cmp(dropAt) < 0 {
+		if sp.balance.Cmp(dropAt) == -1 {
 			sp.Drop(ErrInsufficientFunds)
 		}
 		log.Debug(fmt.Sprintf("balance for peer %s: %s", sp.ID(), sp.balance.String()))
@@ -259,7 +260,11 @@ func (sp *SwapPeer) AccountMsgForPeer(ctx context.Context, msg interface{}, pric
 }
 
 func (sp *SwapPeer) issueCheque(ctx context.Context) error {
-	msg := IssueChequeMsg{}
+	amount := big.NewInt(0)
+	cheque := sp.swapAccount.chequeManager.CreateCheque(sp.ID(), amount.Abs(payAt))
+	msg := IssueChequeMsg{
+		Cheque: cheque,
+	}
 	return sp.Send(ctx, msg)
 }
 
@@ -342,6 +347,7 @@ type PayProfile struct {
 func NewSwap(local *Params, stateStore state.Store) (swap *Swap, err error) {
 
 	swap = &Swap{
+		chequeManager: NewChequeManager(stateStore),
 		local:         local,
 		stateStore:    stateStore,
 		peers:         make(map[discover.NodeID]*SwapPeer),

swarm/swap/swap_test.go

@@ -26,6 +26,7 @@ import (
 	"path/filepath"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/node"
@@ -55,6 +56,21 @@ var testSpec = &protocols.Spec{
 	},
 }
 
+type dummyRW struct{}
+
+func (d *dummyRW) WriteMsg(msg p2p.Msg) error {
+	return nil
+}
+
+func (d *dummyRW) ReadMsg() (p2p.Msg, error) {
+	return p2p.Msg{
+		Code:       0,
+		Size:       0,
+		Payload:    nil,
+		ReceivedAt: time.Now(),
+	}, nil
+}
+
 type testExceedsPayAtMsg struct{}
 type testExceedsDropAtMsg struct{}
 
@@ -81,107 +97,86 @@ func TestLimits(t *testing.T) {
 	}
 }
 
+//unit test for exceeds pay limit
 func TestExceedsPayAt(t *testing.T) {
-	dir, err := ioutil.TempDir("", "swap_test_store")
+	swap, testDir := createTestSwap(t)
-	if err != nil {
+	defer os.RemoveAll(testDir)
-		panic(err)
+
-	}
+	testPeer := newDummyPeer()
-	defer os.RemoveAll(dir)
-	stateStore, err2 := state.NewDBStore(dir)
-	if err2 != nil {
-		panic(err2)
-	}
-	swap, err3 := NewSwap(NewDefaultSwapParams().Params, stateStore)
-	if err3 != nil {
-		t.Fatal(err3)
-	}
-	id := adapters.RandomNodeConfig().ID
-	testPeer := protocols.NewPeer(p2p.NewPeer(id, "testPeer", nil), nil, nil)
 	sp := NewSwapPeer(testPeer, swap)
-	sp.handlerFunc = msgHandler
+	sp.handlerFunc = dummyMsgHandler
+
 	ctx := context.Background()
-	sp.handleAccountedMsg(ctx, &testExceedsPayAtMsg{})
+	sp.Send(ctx, &testExceedsPayAtMsg{})
+
+	cheques := sp.swapAccount.chequeManager.openDebitCheques[sp.ID()]
+	if cheques == nil {
+		t.Fatal("Expected cheques for this peer to be present, but are nil")
+	}
+	if len(cheques) == 0 {
+		t.Fatal("Expected a cheque to have arrived, but len is zero")
+	}
+	if cheques[0].serial != 1 {
+		t.Fatal(fmt.Sprintf("Expected the serial to be one (first message but is: %d", cheques[0].serial))
+	}
+	absPayAt := big.NewInt(0)
+	if cheques[0].amount.Cmp(absPayAt.Abs(payAt)) != 0 {
+		t.Fatal(fmt.Sprintf("Expected the serial to be one (first message but is: %d", cheques[0].serial))
+	}
 }
 
+//unit test for exceeds drop limit
 func TestExceedsDropAt(t *testing.T) {
-	dir, err := ioutil.TempDir("", "swap_test_store")
+	swap, testDir := createTestSwap(t)
-	if err != nil {
+	defer os.RemoveAll(testDir)
-		panic(err)
+
-	}
+	testPeer := newDummyPeer()
-	defer os.RemoveAll(dir)
-	stateStore, err2 := state.NewDBStore(dir)
-	if err2 != nil {
-		panic(err2)
-	}
-	swap, err3 := NewSwap(NewDefaultSwapParams().Params, stateStore)
-	if err3 != nil {
-		t.Fatal(err3)
-	}
-	id := adapters.RandomNodeConfig().ID
-	testPeer := protocols.NewPeer(p2p.NewPeer(id, "testPeer", nil), nil, nil)
 	sp := NewSwapPeer(testPeer, swap)
-	sp.handlerFunc = msgHandler
+	sp.handlerFunc = dummyMsgHandler
+
 	ctx := context.Background()
-	err = sp.Send(ctx, &testExceedsDropAtMsg{})
+	err := sp.Send(ctx, &testExceedsDropAtMsg{})
 	if err != ErrInsufficientFunds {
 		t.Fatal("Expected test to fail with insufficient funds, but it didn't")
 	}
 }
 
-func TestProtocol(t *testing.T) {
+func createTestSwap(t *testing.T) (*Swap, string) {
-	/*
 	dir, err := ioutil.TempDir("", "swap_test_store")
 	if err != nil {
-			panic(err)
+		t.Fatal(err)
 	}
-		defer os.RemoveAll(dir)
+	stateStore, err2 := state.NewDBStore(dir)
-
+	if err2 != nil {
-		swap := NewSwap(NewDefaultSwapParams().Params, state.NewDBStore(dir))
+		t.Fatal(err2)
-		conf := adapters.RandomNodeConfig()
-
-		protocol := func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
-			peer := protocols.NewPeer(p, rw, testSpec)
-			runProtocol(peer, swap)
 	}
-		protocolTester := p2ptest.NewProtocolTester(t, conf.ID, 2, protocol)
+	swap, err3 := NewSwap(NewDefaultSwapParams().Params, stateStore)
-		protocolTester.TestExchanges(p2p.TestExchange{
+	if err3 != nil {
-			Label: "",
+		t.Fatal(err3)
-			Expects: []p2ptest.Expect{
+	}
-				{
+	return swap, dir
-					Code: 5,
-					Msg: &RetrieveRequestMsg{
-						Addr:      hash0[:],
-						SkipCheck: true,
-					},
-					Peer: peerID,
-				},
-			},
-		})
-	*/
 }
 
 func runProtocol(peer *protocols.Peer, swap *Swap) {
 	sp := NewSwapPeer(peer, swap)
-	sp.Peer.Run(msgHandler)
+	sp.Peer.Run(dummyMsgHandler)
 }
 
-func msgHandler(ctx context.Context, msg interface{}) error {
+func dummyMsgHandler(ctx context.Context, msg interface{}) error {
 	return nil
 }
 
-func TestSwapProtocol(t *testing.T) {
+func TestSwapRPC(t *testing.T) {
+	swap, testDir := createTestSwap(t)
+	defer os.RemoveAll(testDir)
 
 	// create the two nodes
 	stack_one, err := newServiceNode(p2pPort, 0, 0)
 	if err != nil {
-		log.Crit("Create servicenode #1 fail", "err", err)
+		t.Fatal("Create servicenode #1 fail", "err", err)
-	}
-	stack_two, err := newServiceNode(p2pPort+1, 0, 0)
-	if err != nil {
-		log.Crit("Create servicenode #2 fail", "err", err)
 	}
 
-	instance := NewSwapProtocol()
+	instance := NewSwapProtocol(swap)
 	// wrapper function for servicenode to start the service
 	swapsvc := func(ctx *node.ServiceContext) (node.Service, error) {
 		return &API{
@@ -192,181 +187,80 @@ func TestSwapProtocol(t *testing.T) {
 	// register adds the service to the services the servicenode starts when started
 	err = stack_one.Register(swapsvc)
 	if err != nil {
-		log.Crit("Register service in servicenode #1 fail", "err", err)
+		t.Fatal("Register service in servicenode #1 fail", "err", err)
 	}
-	err = stack_two.Register(swapsvc)
-	if err != nil {
-		log.Crit("Register service in servicenode #2 fail", "err", err)
-	}
-
 	// start the nodes
 	err = stack_one.Start()
 	if err != nil {
-		log.Crit("servicenode #1 start failed", "err", err)
+		t.Fatal("servicenode #1 start failed", "err", err)
-	}
-	err = stack_two.Start()
-	if err != nil {
-		log.Crit("servicenode #2 start failed", "err", err)
 	}
 
 	// connect to the servicenode RPCs
 	rpcclient_one, err := rpc.Dial(filepath.Join(stack_one.DataDir(), ipcpath))
 	if err != nil {
-		log.Crit("connect to servicenode #1 IPC fail", "err", err)
+		t.Fatal("connect to servicenode #1 IPC fail", "err", err)
 	}
 	defer os.RemoveAll(stack_one.DataDir())
 
-	rpcclient_two, err := rpc.Dial(filepath.Join(stack_two.DataDir(), ipcpath))
+	var balance *big.Int
-	if err != nil {
-		log.Crit("connect to servicenode #2 IPC fail", "err", err)
-	}
-	defer os.RemoveAll(stack_two.DataDir())
-
-	// display that the initial pong counts are 0
-	var balance int
 	err = rpcclient_one.Call(&balance, "swap_balance")
 	if err != nil {
-		log.Crit("servicenode #1 pongcount RPC failed", "err", err)
+		t.Fatal("servicenode #1 RPC failed", "err", err)
 	}
-	log.Info("servicenode #1 before ping", "balance-1", balance)
+	log.Debug("servicenode #1 balance", "balance-1", balance)
 
-	err = rpcclient_two.Call(&balance, "swap_balance")
+	if balance.Cmp(big.NewInt(0)) != 0 {
+		t.Fatal("Expected balance to be 0 but it is not")
+	}
+
+	dummyPeer1 := newDummyPeer()
+	dummyPeer2 := newDummyPeer()
+	id1 := dummyPeer1.ID()
+	id2 := dummyPeer2.ID()
+
+	fake1 := int64(234)
+	fake2 := int64(-100)
+	fakeBalance1 := big.NewInt(fake1)
+	fakeBalance2 := big.NewInt(fake2)
+
+	swap.peers[id1] = NewSwapPeer(dummyPeer1, swap)
+	swap.peers[id2] = NewSwapPeer(dummyPeer2, swap)
+	swap.peers[id1].balance = fakeBalance1
+	swap.peers[id2].balance = fakeBalance2
+
+	err = rpcclient_one.Call(&balance, "swap_balanceWithPeer", id1)
 	if err != nil {
-		log.Crit("servicenode #2 pongcount RPC failed", "err", err)
+		t.Fatal("servicenode #1 RPC failed", "err", err)
 	}
-	log.Info("servicenode #2 before ping", "balance-2", balance)
+	log.Debug("balance1", "balance-1", balance)
-
+	if balance.Cmp(fakeBalance1) != 0 {
-	/*
+		t.Fatal(fmt.Sprintf("Expected balance %s to be equal to fake balance %s, but it is not", balance.String(), fakeBalance1.String()))
-		// get the server instances
-		srv_one := stack_one.Server()
-		srv_two := stack_two.Server()
-
-			// subscribe to peerevents
-			eventOneC := make(chan *p2p.PeerEvent)
-			sub_one := srv_one.SubscribeEvents(eventOneC)
-
-			eventTwoC := make(chan *p2p.PeerEvent)
-			sub_two := srv_two.SubscribeEvents(eventTwoC)
-
-			// connect the nodes
-			p2pnode_two := srv_two.Self()
-			srv_one.AddPeer(p2pnode_two)
-
-			// fork and do the pinging
-			stackW.Add(2)
-			pingmax_one := 4
-			pingmax_two := 2
-
-			go func() {
-
-				// when we get the add event, we know we are connected
-				ev := <-eventOneC
-				if ev.Type != "add" {
-				log.Error("server #1 expected peer add", "eventtype", ev.Type)
-					stackW.Done()
-					return
 	}
-				log.Debug("server #1 connected", "peer", ev.Peer)
 
-				// send the pings
+	err = rpcclient_one.Call(&balance, "swap_balanceWithPeer", id2)
-				for i := 0; i < pingmax_one; i++ {
-					err := rpcclient_one.Call(nil, "foo_ping", ev.Peer)
 	if err != nil {
-						log.Error("server #1 RPC ping fail", "err", err)
+		t.Fatal("servicenode #1 RPC failed", "err", err)
-						stackW.Done()
-						break
 	}
+	log.Debug("balance2", "balance-2", balance)
+	if balance.Cmp(fakeBalance2) != 0 {
+		t.Fatal(fmt.Sprintf("Expected balance %s to be equal to fake balance %s, but it is not", balance.String(), fakeBalance2.String()))
 	}
 
-				// wait for all msgrecv events
+	err = rpcclient_one.Call(&balance, "swap_balance")
-				// pings we receive, and pongs we expect from pings we sent
-				for i := 0; i < pingmax_two+pingmax_one; {
-					ev := <-eventOneC
-					log.Warn("msg", "type", ev.Type, "i", i)
-					if ev.Type == "msgrecv" {
-						i++
-					}
-				}
-
-				stackW.Done()
-			}()
-
-			// mirrors the previous go func
-			go func() {
-				ev := <-eventTwoC
-				if ev.Type != "add" {
-					log.Error("expected peer add", "eventtype", ev.Type)
-					stackW.Done()
-					return
-				}
-				log.Debug("server #2 connected", "peer", ev.Peer)
-				for i := 0; i < pingmax_two; i++ {
-					err := rpcclient_two.Call(nil, "foo_ping", ev.Peer)
 	if err != nil {
-						log.Error("server #2 RPC ping fail", "err", err)
+		t.Fatal("servicenode #1 RPC failed", "err", err)
-						stackW.Done()
-						break
-					}
 	}
+	log.Debug("balance", "balance", balance)
 
-				for i := 0; i < pingmax_one+pingmax_two; {
+	fakeSum := big.NewInt(fake1 + fake2)
-					ev := <-eventTwoC
+	if balance.Cmp(fakeSum) != 0 {
-					if ev.Type == "msgrecv" {
+		t.Fatal(fmt.Sprintf("Expected balance %s to be equal to sum %s, but it is not", balance.String(), fakeSum.String()))
-						log.Warn("msg", "type", ev.Type, "i", i)
-						i++
-					}
 	}
+}
 
-				stackW.Done()
+func newDummyPeer() *protocols.Peer {
-			}()
+	id := adapters.RandomNodeConfig().ID
-
+	return protocols.NewPeer(p2p.NewPeer(id, "testPeer", nil), &dummyRW{}, testSpec)
-			// wait for the two ping pong exchanges to finish
-			stackW.Wait()
-
-			// tell the API to shut down
-			// this will disconnect the peers and close the channels connecting API and protocol
-			err = rpcclient_one.Call(nil, "foo_quit", srv_two.Self().ID)
-			if err != nil {
-				log.Error("server #1 RPC quit fail", "err", err)
-			}
-			err = rpcclient_two.Call(nil, "foo_quit", srv_one.Self().ID)
-			if err != nil {
-				log.Error("server #2 RPC quit fail", "err", err)
-			}
-
-			// disconnect will generate drop events
-			for {
-				ev := <-eventOneC
-				if ev.Type == "drop" {
-					break
-				}
-			}
-			for {
-				ev := <-eventTwoC
-				if ev.Type == "drop" {
-					break
-				}
-			}
-
-			// proudly inspect the results
-			err = rpcclient_one.Call(&count, "foo_pongCount")
-			if err != nil {
-				log.Crit("servicenode #1 pongcount RPC failed", "err", err)
-			}
-			log.Info("servicenode #1 after ping", "pongcount", count)
-
-			err = rpcclient_two.Call(&count, "foo_pongCount")
-			if err != nil {
-				log.Crit("servicenode #2 pongcount RPC failed", "err", err)
-			}
-			log.Info("servicenode #2 after ping", "pongcount", count)
-
-			// bring down the servicenodes
-			sub_one.Unsubscribe()
-			sub_two.Unsubscribe()
-			stack_one.Stop()
-			stack_two.Stop()
-	*/
 }
 
 func newServiceNode(port int, httpport int, wsport int, modules ...string) (*node.Node, error) {