swarm/swap: insufficient funds checks and first working test

swarm/network_test.go

@@ -21,6 +21,7 @@ import (
 	"flag"
 	"fmt"
 	"io/ioutil"
+	"math/big"
 	"math/rand"
 	"os"
 	"sync"
@@ -44,6 +45,7 @@ var (
 	loglevel       = flag.Int("loglevel", 2, "verbosity of logs")
 	longrunning    = flag.Bool("longrunning", false, "do run long-running tests")
 	waitKademlia   = flag.Bool("waitkademlia", false, "wait for healthy kademlia before checking files availability")
+	printStats     = flag.Bool("printstats", false, "print accounting stats to STDOUT")
 	bucketKeySwap  = simulation.BucketKey("swap")
 	bucketKeySwarm = simulation.BucketKey("swarm")
 )
@@ -456,6 +458,8 @@ func TestSwapNetworkSymmetricFileUpload(t *testing.T) {
 		}
 	})
 
+	balancesMap := make(map[discover.NodeID]map[discover.NodeID]*big.Int)
+
 	for _, node := range sim.NodeIDs() {
 		item, ok := sim.NodeItem(node, bucketKeySwarm)
 		if !ok {
@@ -464,14 +468,42 @@ func TestSwapNetworkSymmetricFileUpload(t *testing.T) {
 		}
 		swarm := item.(*Swarm)
 
+		subBalances := make(map[discover.NodeID]*big.Int)
+
 		for _, n := range sim.NodeIDs() {
 			if node == n {
 				continue
 			}
-			if swarm.swap.GetPeerBalance(n) != nil {
+			balance := swarm.swap.GetPeerBalance(n)
-				log.Error(fmt.Sprintf("Balance of node %s to node %s: %s", node.TerminalString(), n.TerminalString(), swarm.swap.GetPeerBalance(n).String()))
+			if balance != nil {
+				subBalances[n] = balance
+				log.Debug(fmt.Sprintf("Balance of node %s to node %s: %s", node.TerminalString(), n.TerminalString(), swarm.swap.GetPeerBalance(n).String()))
 			} else {
-				log.Error(fmt.Sprintf("Node %s has no balance with node %s", node.TerminalString(), n.TerminalString()))
+				log.Debug(fmt.Sprintf("Node %s has no balance with node %s", node.TerminalString(), n.TerminalString()))
+			}
+		}
+		balancesMap[node] = subBalances
+	}
+
+	if *printStats {
+		for k, v := range balancesMap {
+			fmt.Println(fmt.Sprintf("node %s balances:", k.TerminalString()))
+			for kk, vv := range v {
+				fmt.Println(fmt.Sprintf(".........with node %s: balance %s", kk.TerminalString(), vv.String()))
+			}
+		}
+	}
+
+	for k, mapForK := range balancesMap {
+		for n, balanceKwithN := range mapForK {
+			for subK, mapForSubK := range balancesMap {
+				if n == subK {
+					log.Trace(fmt.Sprintf("balance of %s with %s: %s", k.TerminalString(), n.TerminalString(), balanceKwithN))
+					log.Trace(fmt.Sprintf("balance of %s with %s: %s", n.TerminalString(), k.TerminalString(), mapForSubK[k]))
+					if balanceKwithN.CmpAbs(mapForSubK[k]) != 0 && balanceKwithN.Cmp(big.NewInt(0)) != 0 {
+						log.Error("Expected balances to be |abs| = 0 AND balance1 != 0, but they are not")
+					}
+				}
 			}
 		}
 	}

swarm/swap/swap.go

@@ -19,6 +19,7 @@ package swap
 import (
 	"context"
 	"crypto/ecdsa"
+	"errors"
 	"fmt"
 	"math/big"
 	"os"
@@ -54,7 +55,10 @@ var (
 	buyAt                = big.NewInt(20000000000)     // maximum chunk price host is willing to pay (wei)
 	sellAt               = big.NewInt(20000000000)     // minimum chunk price host requires (wei)
 	payAt                = big.NewInt(4096 * 10000)    // threshold that triggers payment {request} (bytes)
-	dropAt               = big.NewInt(4096 * 10000)    // threshold that triggers disconnect (bytes)
+	dropAt               = big.NewInt(-4096 * 10000)   // threshold that triggers disconnect (bytes)
+
+	ErrInsufficientFunds = errors.New("Insufficient funds")
+	ErrNotAccountedMsg   = errors.New("Message does not need accounting")
 )
 
 const (
@@ -92,21 +96,17 @@ type SwapAccountedMsgType interface {
 	GetMsgPrice() *big.Int
 }
 
+//Handler for received messages
 func (sp *SwapPeer) RunAccountedProtocol(protocolHandler func(ctx context.Context, msg interface{}) error) error {
+	//the `peer.Run` function is a loop, so in order to pre-/post-process a message with accounting,
+	//we need to save the actual handler
 	sp.handlerFunc = protocolHandler
 
+	//then run the handler loop function
 	return sp.Run(sp.handleAccountedMsg)
 }
 
-func (sp *SwapPeer) doAccountMsg(ctx context.Context, msg interface{}, direction EntryDirection) error {
+//get a peer's balance
-	if accounted, ok := msg.(SwapAccountedMsgType); ok {
-		price := accounted.GetMsgPrice()
-		//TODO: Calculate total price and account
-		sp.AccountMsgForPeer(price, direction)
-	}
-	return nil
-}
-
 func (swap *Swap) GetPeerBalance(peer discover.NodeID) *big.Int {
 	if p, ok := swap.peers[peer]; ok {
 		return p.balance
@@ -114,48 +114,137 @@ func (swap *Swap) GetPeerBalance(peer discover.NodeID) *big.Int {
 	return nil
 }
 
+//Handle a received message; this is the handler loop function.
+//Check if it needs accounting, and if yes, apply accounting logic:
+//Check for sufficient funds, perform operation, then account
 func (sp *SwapPeer) handleAccountedMsg(ctx context.Context, msg interface{}) error {
-	err := sp.handlerFunc(ctx, msg)
+	var err error
-	if _, ok := msg.(SwapAccountedMsgType); ok && err == nil {
+	var price *big.Int
-		sp.doAccountMsg(ctx, msg, CreditEntry)
+
+	//the message is one which needs accounting...
+	if _, ok := msg.(SwapAccountedMsgType); ok {
+		//..so first check if there are enough funds for the operation available
+		//(for crediting, this means if we are not essentially "overdrafting", or crossing the threshold)
+		price, err = sp.checkAvailableFunds(ctx, msg, CreditEntry)
+		//if not (or some other error occured), return error
+		if err != nil {
+			//also, if the error is indeed insufficient funds, then disconnect the peer
+			if err == ErrInsufficientFunds {
+				log.Error("Insufficient funds, dropping peer")
+				sp.Drop(err)
+			}
+			return err
+		}
+		//at this point we know there are sufficient funds, so process the message
+		err = sp.handlerFunc(ctx, msg)
+		if err == nil {
+			//and if no errors occurred, finally book the entry
+			sp.AccountMsgForPeer(ctx, msg, price, CreditEntry)
+		}
+	} else {
+		//this message doesn't need accounting, so just process it
+		err = sp.handlerFunc(ctx, msg)
 	}
 	return err
 }
 
+//Send a message
+//Check if it needs accounting, and if yes, apply accounting logic:
+//Check for sufficient funds, perform operation, then account
 func (sp *SwapPeer) Send(ctx context.Context, msg interface{}) error {
-	err := sp.Peer.Send(ctx, msg)
+	var err error
-	if _, ok := msg.(SwapAccountedMsgType); ok && err == nil {
+	var price *big.Int
-		sp.doAccountMsg(ctx, msg, DebitEntry)
+
+	//the message is one which needs accounting...
+	if _, ok := msg.(SwapAccountedMsgType); ok {
+		//..so first check if there are enough funds for the operation available
+		price, err = sp.checkAvailableFunds(ctx, msg, DebitEntry)
+		//if not (or some other error occured), return error
+		if err != nil {
+			//also, if the error is indeed insufficient funds, then disconnect the peer
+			if err == ErrInsufficientFunds {
+				log.Error("Insufficient funds, dropping peer")
+				sp.Drop(err)
+			}
+			return err
+		}
+		//at this point we know there are sufficient funds, so process the message
+		err = sp.Peer.Send(ctx, msg)
+		if err == nil {
+			//and if no errors occurred, finally book the entry
+			sp.AccountMsgForPeer(ctx, msg, price, DebitEntry)
+		}
+	} else {
+		//this message doesn't need accounting, so just process it
+		err = sp.Peer.Send(ctx, msg)
 	}
 	return err
 }
 
+//check that the operation has enough funds available
+func (sp *SwapPeer) checkAvailableFunds(ctx context.Context, msg interface{}, direction EntryDirection) (*big.Int, error) {
+	sp.lock.Lock()
+	defer sp.lock.Unlock()
+
+	if accounted, ok := msg.(SwapAccountedMsgType); ok {
+		price := accounted.GetMsgPrice()
+		//local node is being credited (in its favor), so check upper limit
+		if direction == CreditEntry {
+			checkBalance := sp.balance.Add(sp.balance, price)
+			//(checkBalance *Int) Cmp(payAt)
+			// -1 if checkBalance  <  payAt
+			//  0 if checkBalance  == payAt
+			// +1 if checkBalance  >  payAt
+			if checkBalance.Cmp(payAt) == 1 {
+				return nil, ErrInsufficientFunds
+			}
+		} else if direction == DebitEntry {
+			//(checkBalance *Int) Cmp(dropAt)
+			// -1 if checkBalance  <  dropAt
+			//  0 if checkBalance  ==	dropAt
+			// +1 if checkBalance  >  dropAt
+			checkBalance := sp.balance.Sub(sp.balance, price)
+			if checkBalance.Cmp(dropAt) == -1 {
+				return nil, ErrInsufficientFunds
+			}
+		}
+		return price, nil
+	}
+	return nil, ErrNotAccountedMsg
+}
+
 //The balance is accounted from the point of view of the local node
 //Thus, we credit the balance and increase it when the amount is in favor of the local node
 //We debit the balance and decrease it when the amount is in favor of the remote peer
-func (sp *SwapPeer) AccountMsgForPeer(price *big.Int, direction EntryDirection) {
+func (sp *SwapPeer) AccountMsgForPeer(ctx context.Context, msg interface{}, price *big.Int, direction EntryDirection) {
-	sp.lock.Lock()
+	if _, ok := msg.(SwapAccountedMsgType); ok {
-	defer sp.lock.Unlock()
+		sp.lock.Lock()
-	//local node is being credited (in its favor), so its balance increases
+		defer sp.lock.Unlock()
-	if direction == CreditEntry {
+		//local node is being credited (in its favor), so its balance increases
-		sp.balance = sp.balance.Add(sp.balance, price)
+		if direction == CreditEntry {
-		//local node is being debited (in favor of remote peer), so its balance decreases
+			//NOTE: do we need to check for sufficient funds again?
-	} else if direction == DebitEntry {
+			//operations are not atomic/transactional, so balance may have changed in the meanwhile!
-		sp.balance = sp.balance.Sub(sp.balance, price)
+			sp.balance = sp.balance.Add(sp.balance, price)
+			//local node is being debited (in favor of remote peer), so its balance decreases
+		} else if direction == DebitEntry {
+			sp.balance = sp.balance.Sub(sp.balance, price)
+		}
+		//TODO: save to store here? init store?
+		sp.swapAccount.stateStore.Put(sp.storeID, sp.balance)
+		if sp.balance.Cmp(payAt) > -1 {
+			//TODO: Issue Cheque
+		}
+		if sp.balance.Cmp(dropAt) < 0 {
+			//TODO: Drop peer
+		}
+		log.Debug(fmt.Sprintf("balance for peer %s: %s", sp.ID(), sp.balance.String()))
 	}
-	//TODO: save to store here? init store?
-	sp.swapAccount.stateStore.Put(sp.storeID, sp.balance)
-	if sp.balance.Cmp(payAt) > -1 {
-		//TODO: Issue Cheque
-	}
-	if sp.balance.Cmp(dropAt) < 0 {
-		//TODO: Drop peer
-	}
-	log.Debug(fmt.Sprintf("balance for peer %s: %s", sp.ID(), sp.balance.String()))
 }
 
+//Create a new swap accounted peer
 func NewSwapPeer(peer *protocols.Peer, swap *Swap) *SwapPeer {
 	balance := big.NewInt(0)
+	//check if there is one already in the stateStore and load it
 	swap.stateStore.Get(peer.String()[:24]+"-swap", &balance)
 	sp := &SwapPeer{
 		Peer:        peer,