swap: refactor lastReceivedCheque, lastSentCheque, balances to peer (ethersphere#1725)

* swap: move lastReceivedCheque, lastSentCheque and balances to peer
* swap: move sendCheque and createCheque to peer
* swap: remove accounting and store lock in favour of a per-peer lock

Author: ralph-pichler

swap/peer.go

@@ -17,10 +17,14 @@
 package swap
 
 import (
+	"context"
 	"errors"
+	"fmt"
+	"strconv"
+	"sync"
 
 	"github.com/ethereum/go-ethereum/common"
-
+	"github.com/ethersphere/swarm/log"
 	"github.com/ethersphere/swarm/p2p/protocols"
 )
 
@@ -30,18 +34,140 @@ var ErrDontOwe = errors.New("no negative balance")
 // Peer is a devp2p peer for the Swap protocol
 type Peer struct {
 	*protocols.Peer
+	lock               sync.RWMutex
 	swap               *Swap
 	beneficiary        common.Address
 	contractAddress    common.Address
 	lastReceivedCheque *Cheque
+	lastSentCheque     *Cheque
+	balance            int64
 }
 
 // NewPeer creates a new swap Peer instance
-func NewPeer(p *protocols.Peer, s *Swap, beneficiary common.Address, contractAddress common.Address) *Peer {
-	return &Peer{
+func NewPeer(p *protocols.Peer, s *Swap, beneficiary common.Address, contractAddress common.Address) (peer *Peer, err error) {
+	peer = &Peer{
 		Peer:            p,
 		swap:            s,
 		beneficiary:     beneficiary,
 		contractAddress: contractAddress,
 	}
+
+	if peer.lastReceivedCheque, err = s.loadLastReceivedCheque(p.ID()); err != nil {
+		return nil, err
+	}
+
+	if peer.lastSentCheque, err = s.loadLastSentCheque(p.ID()); err != nil {
+		return nil, err
+	}
+
+	if peer.balance, err = s.loadBalance(p.ID()); err != nil {
+		return nil, err
+	}
+	return peer, nil
+}
+
+func (p *Peer) getLastReceivedCheque() *Cheque {
+	return p.lastReceivedCheque
+}
+
+func (p *Peer) getLastSentCheque() *Cheque {
+	return p.lastSentCheque
+}
+
+func (p *Peer) setLastReceivedCheque(cheque *Cheque) error {
+	p.lastReceivedCheque = cheque
+	return p.swap.saveLastReceivedCheque(p.ID(), cheque)
+}
+
+func (p *Peer) setLastSentCheque(cheque *Cheque) error {
+	p.lastSentCheque = cheque
+	return p.swap.saveLastSentCheque(p.ID(), cheque)
+}
+
+func (p *Peer) getLastCumulativePayout() uint64 {
+	lastCheque := p.getLastReceivedCheque()
+	if lastCheque != nil {
+		return lastCheque.CumulativePayout
+	}
+	return 0
+}
+
+func (p *Peer) setBalance(balance int64) error {
+	p.balance = balance
+	return p.swap.saveBalance(p.ID(), balance)
+}
+
+func (p *Peer) getBalance() int64 {
+	return p.balance
+}
+
+// To be called with mutex already held
+func (p *Peer) updateBalance(amount int64) error {
+	//adjust the balance
+	//if amount is negative, it will decrease, otherwise increase
+	newBalance := p.getBalance() + amount
+	if err := p.setBalance(newBalance); err != nil {
+		return err
+	}
+	log.Debug("balance for peer after accounting", "peer", p.ID().String(), "balance", strconv.FormatInt(newBalance, 10))
+	return nil
+}
+
+// createCheque creates a new cheque whose beneficiary will be the peer and
+// whose amount is based on the last cheque and current balance for this peer
+// The cheque will be signed and point to the issuer's contract
+// To be called with mutex already held
+// Caller must be careful that the same resources aren't concurrently read and written by multiple routines
+func (p *Peer) createCheque() (*Cheque, error) {
+	var cheque *Cheque
+	var err error
+
+	if p.getBalance() >= 0 {
+		return nil, fmt.Errorf("expected negative balance, found: %d", p.getBalance())
+	}
+	// the balance should be negative here, we take the absolute value:
+	honey := uint64(-p.getBalance())
+
+	amount, err := p.swap.oracle.GetPrice(honey)
+	if err != nil {
+		return nil, fmt.Errorf("error getting price from oracle: %v", err)
+	}
+
+	total := p.getLastCumulativePayout()
+
+	cheque = &Cheque{
+		ChequeParams: ChequeParams{
+			CumulativePayout: total + amount,
+			Contract:         p.swap.owner.Contract,
+			Beneficiary:      p.beneficiary,
+		},
+		Honey: honey,
+	}
+	cheque.Signature, err = cheque.Sign(p.swap.owner.privateKey)
+
+	return cheque, err
+}
+
+// sendCheque sends a cheque to peer
+// To be called with mutex already held
+// Caller must be careful that the same resources aren't concurrently read and written by multiple routines
+func (p *Peer) sendCheque() error {
+	cheque, err := p.createCheque()
+	if err != nil {
+		return fmt.Errorf("error while creating cheque: %v", err)
+	}
+
+	if err := p.setLastSentCheque(cheque); err != nil {
+		return fmt.Errorf("error while storing the last cheque: %v", err)
+	}
+
+	if err := p.updateBalance(int64(cheque.Honey)); err != nil {
+		return err
+	}
+
+	log.Info("sending cheque", "honey", cheque.Honey, "cumulativePayout", cheque.ChequeParams.CumulativePayout, "beneficiary", cheque.Beneficiary, "contract", cheque.Contract)
+
+	return p.Send(context.Background(), &EmitChequeMsg{
+		Cheque: cheque,
+	})
 }

swap/protocol.go

@@ -121,8 +121,10 @@ func (s *Swap) run(p *p2p.Peer, rw p2p.MsgReadWriter) error {
 		return err
 	}
 
-	swapPeer := NewPeer(protoPeer, s, beneficiary, response.ContractAddress)
-	s.addPeer(swapPeer)
+	swapPeer, err := s.addPeer(protoPeer, beneficiary, response.ContractAddress)
+	if err != nil {
+		return err
+	}
 	defer s.removePeer(swapPeer)
 
 	return swapPeer.Run(s.handleMsg(swapPeer))
@@ -134,17 +136,22 @@ func (s *Swap) removePeer(p *Peer) {
 	delete(s.peers, p.ID())
 }
 
-func (s *Swap) addPeer(p *Peer) {
+func (s *Swap) addPeer(protoPeer *protocols.Peer, beneficiary common.Address, contractAddress common.Address) (*Peer, error) {
 	s.peersLock.Lock()
 	defer s.peersLock.Unlock()
+	p, err := NewPeer(protoPeer, s, beneficiary, contractAddress)
+	if err != nil {
+		return nil, err
+	}
 	s.peers[p.ID()] = p
+	return p, nil
 }
 
-func (s *Swap) getPeer(id enode.ID) (*Peer, bool) {
+func (s *Swap) getPeer(id enode.ID) *Peer {
 	s.peersLock.RLock()
 	defer s.peersLock.RUnlock()
-	peer, ok := s.peers[id]
-	return peer, ok
+	peer := s.peers[id]
+	return peer
 }
 
 type swapAPI interface {

swap/protocol_test.go

@@ -63,7 +63,7 @@ func TestHandshake(t *testing.T) {
 	creditor := protocolTester.Nodes[1]
 
 	// set balance artifially
-	swap.balances[creditor.ID()] = -42
+	swap.saveBalance(creditor.ID(), -42)
 
 	// create the expected cheque to be received
 	cheque := newTestCheque()
@@ -135,14 +135,17 @@ func TestEmitCheque(t *testing.T) {
 	// create the debitor peer
 	dPtpPeer := p2p.NewPeer(enode.ID{}, "debitor", []p2p.Cap{})
 	dProtoPeer := protocols.NewPeer(dPtpPeer, nil, Spec)
-	debitor := NewPeer(dProtoPeer, creditorSwap, debitorSwap.owner.address, debitorSwap.owner.Contract)
+	debitor, err := creditorSwap.addPeer(dProtoPeer, debitorSwap.owner.address, debitorSwap.owner.Contract)
+	if err != nil {
+		t.Fatal(err)
+	}
 
 	// set balance artificially
-	creditorSwap.balances[debitor.ID()] = 42
-	log.Debug("balance", "balance", creditorSwap.balances[debitor.ID()])
+	debitor.setBalance(42)
+	log.Debug("balance", "balance", debitor.getBalance())
 	// a safe check: at this point no cheques should be in the swap
-	if len(creditorSwap.cheques) != 0 {
-		t.Fatalf("Expected no cheques at creditor, but there are %d:", len(creditorSwap.cheques))
+	if debitor.getLastReceivedCheque() != nil {
+		t.Fatalf("Expected no cheques at creditor, but there is %v:", debitor.getLastReceivedCheque())
 	}
 
 	log.Debug("create a cheque")
@@ -179,15 +182,15 @@ func TestEmitCheque(t *testing.T) {
 	case <-time.After(4 * time.Second):
 		t.Fatalf("Timeout waiting for cash transaction to complete")
 	}
-	log.Debug("balance", "balance", creditorSwap.balances[debitor.ID()])
+	log.Debug("balance", "balance", debitor.getBalance())
 	// check that the balance has been reset
-	if creditorSwap.balances[debitor.ID()] != 0 {
-		t.Fatalf("Expected debitor balance to have been reset to %d, but it is %d", 0, creditorSwap.balances[debitor.ID()])
+	if debitor.getBalance() != 0 {
+		t.Fatalf("Expected debitor balance to have been reset to %d, but it is %d", 0, debitor.getBalance())
 	}
-	recvCheque := creditorSwap.loadLastReceivedCheque(debitor)
+	recvCheque := debitor.getLastReceivedCheque()
 	log.Debug("expected cheque", "cheque", recvCheque)
 	if recvCheque != cheque {
-		t.Fatalf("Expected exactly one cheque at creditor, but there are %d:", len(creditorSwap.cheques))
+		t.Fatalf("Expected cheque at creditor, but it was %v:", recvCheque)
 	}
 }
 
@@ -205,48 +208,48 @@ func TestTriggerPaymentThreshold(t *testing.T) {
 
 	// create a dummy pper
 	cPeer := newDummyPeerWithSpec(Spec)
-	creditor := NewPeer(cPeer.Peer, debitorSwap, common.Address{}, common.Address{})
-	// set the creditor as peer into the debitor's swap
-	debitorSwap.peers[creditor.ID()] = creditor
+	creditor, err := debitorSwap.addPeer(cPeer.Peer, common.Address{}, common.Address{})
+	if err != nil {
+		t.Fatal(err)
+	}
 
 	// set the balance to manually be at PaymentThreshold
 	overDraft := 42
-	debitorSwap.balances[creditor.ID()] = 0 - DefaultPaymentThreshold
+	creditor.setBalance(-DefaultPaymentThreshold)
 
 	// we expect a cheque at the end of the test, but not yet
-	lenCheques := len(debitorSwap.cheques)
-	if lenCheques != 0 {
-		t.Fatalf("Expected no cheques yet, but there are %d", lenCheques)
+	if creditor.getLastSentCheque() != nil {
+		t.Fatalf("Expected no cheques yet, but there is %v:", creditor.getLastSentCheque())
 	}
 	// do some accounting, no error expected, just a WARN
-	err := debitorSwap.Add(int64(-overDraft), creditor.Peer)
+	err = debitorSwap.Add(int64(-overDraft), creditor.Peer)
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	// we should now have a cheque
-	lenCheques = len(debitorSwap.cheques)
-	if lenCheques != 1 {
-		t.Fatalf("Expected one cheque, but there are %d", lenCheques)
+	if creditor.getLastSentCheque() == nil {
+		t.Fatal("Expected one cheque, but there is none")
 	}
-	cheque := debitorSwap.cheques[creditor.ID()]
+
+	cheque := creditor.getLastSentCheque()
 	expectedAmount := uint64(overDraft) + DefaultPaymentThreshold
 	if cheque.CumulativePayout != expectedAmount {
 		t.Fatalf("Expected cheque cumulative payout to be %d, but is %d", expectedAmount, cheque.CumulativePayout)
 	}
 
 	// because no other accounting took place in the meantime the balance should be exactly 0
-	if debitorSwap.balances[creditor.ID()] != 0 {
-		t.Fatalf("Expected debitorSwap balance to be 0, but is %d", debitorSwap.balances[creditor.ID()])
+	if creditor.getBalance() != 0 {
+		t.Fatalf("Expected debitorSwap balance to be 0, but is %d", creditor.getBalance())
 	}
 
 	// do some accounting again to trigger a second cheque
 	if err = debitorSwap.Add(int64(-DefaultPaymentThreshold), creditor.Peer); err != nil {
 		t.Fatal(err)
 	}
 
-	if debitorSwap.balances[creditor.ID()] != 0 {
-		t.Fatalf("Expected debitorSwap balance to be 0, but is %d", debitorSwap.balances[creditor.ID()])
+	if creditor.getBalance() != 0 {
+		t.Fatalf("Expected debitorSwap balance to be 0, but is %d", creditor.getBalance())
 	}
 }
 
@@ -260,34 +263,33 @@ func TestTriggerDisconnectThreshold(t *testing.T) {
 
 	// create a dummy pper
 	cPeer := newDummyPeerWithSpec(Spec)
-	debitor := NewPeer(cPeer.Peer, creditorSwap, common.Address{}, common.Address{})
-	// set the debitor as peer into the creditor's swap
-	creditorSwap.peers[debitor.ID()] = debitor
+	debitor, err := creditorSwap.addPeer(cPeer.Peer, common.Address{}, common.Address{})
+	if err != nil {
+		t.Fatal(err)
+	}
 
 	// set the balance to manually be at DisconnectThreshold
 	overDraft := 42
 	expectedBalance := int64(DefaultDisconnectThreshold)
 	// we don't expect any change after the test
-	creditorSwap.balances[debitor.ID()] = expectedBalance
+	debitor.setBalance(expectedBalance)
 	// we also don't expect any cheques yet
-	lenCheques := len(creditorSwap.cheques)
-	if lenCheques != 0 {
-		t.Fatalf("Expected no cheques yet, but there are %d", lenCheques)
+	if debitor.getLastSentCheque() != nil {
+		t.Fatalf("Expected no cheques yet, but there is %v", debitor.getLastSentCheque())
 	}
 	// now do some accounting
-	err := creditorSwap.Add(int64(overDraft), debitor.Peer)
+	err = creditorSwap.Add(int64(overDraft), debitor.Peer)
 	// it should fail due to overdraft
 	if err == nil {
 		t.Fatal("Expected an error due to overdraft, but did not get any")
 	}
 	// no balance change expected
-	if creditorSwap.balances[debitor.ID()] != expectedBalance {
-		t.Fatalf("Expected balance to be %d, but is %d", expectedBalance, creditorSwap.balances[debitor.ID()])
+	if debitor.getBalance() != expectedBalance {
+		t.Fatalf("Expected balance to be %d, but is %d", expectedBalance, debitor.getBalance())
 	}
 	// still no cheques expected
-	lenCheques = len(creditorSwap.cheques)
-	if lenCheques != 0 {
-		t.Fatalf("Expected still no cheque, but there are %d", lenCheques)
+	if debitor.getLastSentCheque() != nil {
+		t.Fatalf("Expected still no cheques yet, but there is %v", debitor.getLastSentCheque())
 	}
 
 	// let's do the whole thing again (actually a bit silly, it's somehow simulating the peer would have been dropped)
@@ -296,13 +298,12 @@ func TestTriggerDisconnectThreshold(t *testing.T) {
 		t.Fatal("Expected an error due to overdraft, but did not get any")
 	}
 
-	if creditorSwap.balances[debitor.ID()] != expectedBalance {
-		t.Fatalf("Expected balance to be %d, but is %d", expectedBalance, creditorSwap.balances[debitor.ID()])
+	if debitor.getBalance() != expectedBalance {
+		t.Fatalf("Expected balance to be %d, but is %d", expectedBalance, debitor.getBalance())
 	}
 
-	lenCheques = len(creditorSwap.cheques)
-	if lenCheques != 0 {
-		t.Fatalf("Expected still no cheque, but there are %d", lenCheques)
+	if debitor.getLastSentCheque() != nil {
+		t.Fatalf("Expected no cheques yet, but there is %v", debitor.getLastSentCheque())
 	}
 }