swarm/swap, p2p/protocols: finalized first phase of swap accounting

p2p/protocols/accounting.go

@@ -20,29 +20,47 @@ import (
 	"github.com/ethereum/go-ethereum/metrics"
 )
 
+//define some metrics
 var (
 	//NOTE: these metrics just define the interfaces and are currently *NOT persisted* over sessions
-	mBalanceCredit   = metrics.NewRegisteredCounterForced("account.balance.credit", nil)
+	//All metrics are cumulative
-	mBalanceDebit    = metrics.NewRegisteredCounterForced("account.balance.debit", nil)
+
-	mBytesCredit     = metrics.NewRegisteredCounterForced("account.bytes.credit", nil)
+	//total amount of units credited
-	mBytesDebit      = metrics.NewRegisteredCounterForced("account.bytes.debit", nil)
+	mBalanceCredit = metrics.NewRegisteredCounterForced("account.balance.credit", nil)
-	mMsgCredit       = metrics.NewRegisteredCounterForced("account.msg.credit", nil)
+	//total amount of units debited
-	mMsgDebit        = metrics.NewRegisteredCounterForced("account.msg.debit", nil)
+	mBalanceDebit = metrics.NewRegisteredCounterForced("account.balance.debit", nil)
-	mPeerDrops       = metrics.NewRegisteredCounterForced("account.peerdrops", nil)
+	//total amount of bytes credited
-	mSelfDrops       = metrics.NewRegisteredCounterForced("account.selfdrops", nil)
+	mBytesCredit = metrics.NewRegisteredCounterForced("account.bytes.credit", nil)
-	mChequesIssued   = metrics.NewRegisteredCounterForced("account.cheques.issued", nil)
+	//total amount of bytes debited
-	mChequesReceived = metrics.NewRegisteredCounterForced("account.cheques.received", nil)
+	mBytesDebit = metrics.NewRegisteredCounterForced("account.bytes.debit", nil)
+	//total amount of credited messages
+	mMsgCredit = metrics.NewRegisteredCounterForced("account.msg.credit", nil)
+	//total amount of debited messages
+	mMsgDebit = metrics.NewRegisteredCounterForced("account.msg.debit", nil)
+	//how many times local node had to drop remote peers
+	mPeerDrops = metrics.NewRegisteredCounterForced("account.peerdrops", nil)
+	//how many times local node overdrafted and dropped
+	mSelfDrops = metrics.NewRegisteredCounterForced("account.selfdrops", nil)
+	//how many cheques have been issued
+	//mChequesIssued = metrics.NewRegisteredCounterForced("account.cheques.issued", nil)
+	//how many cheques have been received
+	//mChequesReceived = metrics.NewRegisteredCounterForced("account.cheques.received", nil)
 )
 
+//Prices defines how prices are being passed on to the accounting instance
 type Prices interface {
+	//Return the Price for a message
 	Price(interface{}) *Price
 }
 
+//Price represents the costs of a message
 type Price struct {
 	Value   int64 //Positive if sender pays, negative if receiver pays
 	PerByte bool  //True if the price is per byte or for unit
 }
 
+//Price can be different depending on if it is the sender or receiver who pays
+//ForSender gives back the price for sending a message
 func (p *Price) ForSender(size uint32) int64 {
 	price := p.Value
 	if p.PerByte {
@@ -51,6 +69,8 @@ func (p *Price) ForSender(size uint32) int64 {
 	return price
 }
 
+//Price can be different depending on if it is the sender or receiver who pays
+//ForReceiver gives back the price for  receiving a message
 func (p *Price) ForReceiver(size uint32) int64 {
 	price := p.Value
 	if p.PerByte {
@@ -59,6 +79,9 @@ func (p *Price) ForReceiver(size uint32) int64 {
 	return 0 - price
 }
 
+//Balance is the actual accounting instance
+//Balance defines the operations needed for accounting
+//Implementations internally maintain the balance for every peer
 type Balance interface {
 	//Adds amount to the local balance with remote node `peer`;
 	//positive amount = credit local node
@@ -66,9 +89,12 @@ type Balance interface {
 	Add(amount int64, peer *Peer) error
 }
 
+//Accounting implements the Hook interface
+//It interfaces to the balances through the Balance interface,
+//while interfacing with protocols and its prices through the Prices interface
 type Accounting struct {
-	Balance
+	Balance //interface to accounting logic
-	Prices
+	Prices  //interface to prices logic
 }
 
 func NewAccounting(mgr Balance, po Prices) *Accounting {
@@ -79,28 +105,41 @@ func NewAccounting(mgr Balance, po Prices) *Accounting {
 	return ah
 }
 
+//Implement Hook.Send
 func (ah *Accounting) Send(peer *Peer, size uint32, msg interface{}) error {
+	//get the price for a message (through the protocol spec)
 	price := ah.Price(msg)
+	//this message doesn't need accounting
 	if price == nil {
 		return nil
 	}
+	//evaluate the price for sending messages
 	finalPrice := price.ForSender(size)
+	//do the accounting
 	err := ah.Add(finalPrice, peer)
+	//record metrics
 	ah.doMetrics(finalPrice, size, err)
 	return err
 }
 
+//Implement Hook.Receive
 func (ah *Accounting) Receive(peer *Peer, size uint32, msg interface{}) error {
+	//get the price for a message (through the protocol spec)
 	price := ah.Price(msg)
+	//this message doesn't need accounting
 	if price == nil {
 		return nil
 	}
+	//evaluate the price for receiving messages
 	finalPrice := price.ForReceiver(size)
+	//do the accounting
 	err := ah.Add(finalPrice, peer)
+	//record metrics
 	ah.doMetrics(finalPrice, size, err)
 	return err
 }
 
+//record some metrics
 func (ah *Accounting) doMetrics(price int64, size uint32, err error) {
 	if price > 0 {
 		mBalanceCredit.Inc(int64(price))

p2p/protocols/accounting_test.go

@@ -27,27 +27,36 @@ import (
 	"github.com/ethereum/go-ethereum/rlp"
 )
 
+//dummy Balance implementation
 type dummyBalance struct {
 	amount int64
 	peer   *Peer
 }
 
+//dummy Prices implementation
 type dummyPrices struct{}
 
+//a dummy message which needs size based accounting
 type perBytesMsg struct {
 	Content string
 }
 
+//a dummy message which is paid for per unit
 type perUnitMsg struct{}
+
+//a dummy message which doesn't need any payment
 type zeroMsg struct{}
 
+//return the price for the defined messages
 func (d *dummyPrices) Price(msg interface{}) *Price {
 	switch msg.(type) {
+	//size based message cost, sender pays
 	case *perBytesMsg:
 		return &Price{
 			PerByte: true,
 			Value:   int64(100),
 		}
+		//unitary cost, receiver pays
 	case *perUnitMsg:
 		return &Price{
 			PerByte: false,
@@ -57,71 +66,173 @@ func (d *dummyPrices) Price(msg interface{}) *Price {
 	return nil
 }
 
+//dummy accounting implementation, only stores values for later check
 func (d *dummyBalance) Add(amount int64, peer *Peer) error {
 	d.amount = amount
 	d.peer = peer
 	return nil
 }
 
+//We need to test that if the hook is not defined, then message infrastructure
+//(send,receive) still works
 func TestNoHook(t *testing.T) {
+	//create a test spec
 	spec := createTestSpec()
+	//a random node
 	id := adapters.RandomNodeConfig().ID
+	//a peer
 	p := p2p.NewPeer(id, "testPeer", nil)
-	peer := NewPeer(p, &dummyRW{}, spec)
+	rw := &dummyRW{}
+	peer := NewPeer(p, rw, spec)
 	ctx := context.TODO()
 	msg := &perBytesMsg{Content: "testBalance"}
+	//send a message
 	peer.Send(ctx, msg)
+	//simulate receiving a message
+	rw.msg = msg
 	peer.handleIncoming(func(ctx context.Context, msg interface{}) error {
 		return nil
 	})
+	//all should just work and not result in any error
 }
 
-func TestSendBalance(t *testing.T) {
+//lowest level unit test
+func TestSend(t *testing.T) {
+	//create instances
 	balance := &dummyBalance{}
 	prices := &dummyPrices{}
-
+	//create the spec
 	spec := createTestSpec()
+	//create the accounting hook for the spec
 	spec.Hook = NewAccounting(balance, prices)
+	//create a peer
+	id := adapters.RandomNodeConfig().ID
+	p := p2p.NewPeer(id, "testPeer", nil)
+	peer := NewPeer(p, &dummyRW{}, spec)
+	msg := &perBytesMsg{Content: "testBalance"}
 
+	size, _ := rlp.EncodeToBytes(msg)
+	spec.Hook.Send(peer, uint32(len(size)), msg)
+	if balance.amount != int64(len(size)*100) {
+		t.Fatalf("Expected balance to be %d but is %d", (len(size) * 100), balance.amount)
+	}
+
+	//send a unitary, receiver pays message
+	msg2 := &perUnitMsg{}
+	spec.Hook.Send(peer, 0, msg2)
+	//check that the balance is actually the expected
+	if balance.amount != int64(-99) {
+		t.Fatalf("Expected balance to be %d but is %d", -99, balance.amount)
+	}
+
+	//set arbitrary balance
+	balance.amount = 77
+	//send a non accounted message, balance should not change
+	msg3 := &zeroMsg{}
+	spec.Hook.Send(peer, 0, msg3)
+	if balance.amount != int64(77) {
+		t.Fatalf("Expected balance to be %d but is %d", 77, balance.amount)
+	}
+}
+
+//lowest level unit test
+func TestReceive(t *testing.T) {
+	//create instances
+	balance := &dummyBalance{}
+	prices := &dummyPrices{}
+	//create the spec
+	spec := createTestSpec()
+	//create the accounting hook for the spec
+	spec.Hook = NewAccounting(balance, prices)
+	//create a peer
+	id := adapters.RandomNodeConfig().ID
+	p := p2p.NewPeer(id, "testPeer", nil)
+	peer := NewPeer(p, &dummyRW{}, spec)
+	msg := &perBytesMsg{Content: "testBalance"}
+
+	size, _ := rlp.EncodeToBytes(msg)
+	spec.Hook.Receive(peer, uint32(len(size)), msg)
+	if balance.amount != int64(len(size)*-100) {
+		t.Fatalf("Expected balance to be %d but is %d", (len(size) * 100), balance.amount)
+	}
+
+	//send a unitary, receiver pays message
+	msg2 := &perUnitMsg{}
+	spec.Hook.Receive(peer, 0, msg2)
+	//check that the balance is actually the expected
+	if balance.amount != int64(99) {
+		t.Fatalf("Expected balance to be %d but is %d", -99, balance.amount)
+	}
+
+	//set arbitrary balance
+	balance.amount = 77
+	//send a non accounted message, balance should not change
+	msg3 := &zeroMsg{}
+	spec.Hook.Receive(peer, 0, msg3)
+	if balance.amount != int64(77) {
+		t.Fatalf("Expected balance to be %d but is %d", 77, balance.amount)
+	}
+}
+
+//Test sending with a peer, higher level
+func TestSendWithPeer(t *testing.T) {
+	//create instances
+	balance := &dummyBalance{}
+	prices := &dummyPrices{}
+	//create the spec
+	spec := createTestSpec()
+	//create the accounting hook for the spec
+	spec.Hook = NewAccounting(balance, prices)
+	//create a peer
 	id := adapters.RandomNodeConfig().ID
 	p := p2p.NewPeer(id, "testPeer", nil)
 	peer := NewPeer(p, &dummyRW{}, spec)
 	ctx := context.TODO()
 	msg := &perBytesMsg{Content: "testBalance"}
 	size, _ := rlp.EncodeToBytes(msg)
+	//send a size based message
 	peer.Send(ctx, msg)
+	//check that the balance is actually the expected
 	if balance.amount != int64((len(size) * 100)) {
-		t.Fatalf("Expected price to be %d but is %d", (len(size) * 100), balance.amount)
+		t.Fatalf("Expected balance to be %d but is %d", (len(size) * 100), balance.amount)
 	}
 
+	//send a unitary, receiver pays message
 	msg2 := &perUnitMsg{}
 	peer.Send(ctx, msg2)
+	//check that the balance is actually the expected
 	if balance.amount != int64(-99) {
-		t.Fatalf("Expected price to be %d but is %d", -99, balance.amount)
+		t.Fatalf("Expected balance to be %d but is %d", -99, balance.amount)
 	}
 
+	//set arbitrary balance
 	balance.amount = 77
+	//send a non accounted message, balance should not change
 	msg3 := &zeroMsg{}
 	peer.Send(ctx, msg3)
 	if balance.amount != int64(77) {
-		t.Fatalf("Expected price to be %d but is %d", 77, balance.amount)
+		t.Fatalf("Expected balance to be %d but is %d", 77, balance.amount)
 	}
 }
 
-func TestReceiveBalance(t *testing.T) {
+//Test receiving with a peer, higher level
+func TestReceiveWithPeer(t *testing.T) {
+	//create instances
 	balance := &dummyBalance{}
 	prices := &dummyPrices{}
-
+	//create the spec
 	spec := createTestSpec()
+	//create the accounting hook for the spec
 	spec.Hook = NewAccounting(balance, prices)
-
+	//create a peer
 	id := adapters.RandomNodeConfig().ID
 	p := p2p.NewPeer(id, "testPeer", nil)
 	rw := &dummyRW{}
 	peer := NewPeer(p, rw, spec)
+
+	//simulate receiving a size based, sender-pays message
 	msg := &perBytesMsg{Content: "testBalance"}
 	size, _ := rlp.EncodeToBytes(msg)
-
 	rw.msg = msg
 	rw.code, _ = spec.GetCode(msg)
 	err := peer.handleIncoming(func(ctx context.Context, msg interface{}) error {
@@ -131,9 +242,10 @@ func TestReceiveBalance(t *testing.T) {
 		t.Fatalf("Expected no error, but got error: %v", err)
 	}
 	if balance.amount != int64((len(size) * (-100))) {
-		t.Fatalf("Expected price to be %d but is %d", (len(size) * (-100)), balance.amount)
+		t.Fatalf("Expected balance to be %d but is %d", (len(size) * (-100)), balance.amount)
 	}
 
+	//simulate receiving a size based, sender-pays message
 	msg2 := &perUnitMsg{}
 	rw.msg = msg2
 	rw.code, _ = spec.GetCode(msg2)
@@ -144,23 +256,24 @@ func TestReceiveBalance(t *testing.T) {
 		t.Fatalf("Expected no error, but got error: %v", err)
 	}
 	if balance.amount != int64(99) {
-		t.Fatalf("Expected price to be %d but is %d", 99, balance.amount)
+		t.Fatalf("Expected balance to be %d but is %d", 99, balance.amount)
 	}
 
+	//set arbitrary balance an
 	msg3 := &zeroMsg{}
 	rw.msg = msg3
 	rw.code, _ = spec.GetCode(msg3)
 	//need to reset cause no accounting won't overwrite
 	balance.amount = -888
+	//simulate receiving a non accounted message, balance should not change
 	err = peer.handleIncoming(func(ctx context.Context, msg interface{}) error {
 		return nil
 	})
 	if err != nil {
 		t.Fatalf("Expected no error, but got error: %v", err)
 	}
-
 	if balance.amount != int64(-888) {
-		t.Fatalf("Expected price to be %d but is %d", -888, balance.amount)
+		t.Fatalf("Expected balance to be %d but is %d", -888, balance.amount)
 	}
 }
 
@@ -200,6 +313,7 @@ func (d *dummyRW) getDummyMsg() []byte {
 	return rr
 }
 
+//create a test spec
 func createTestSpec() *Spec {
 	spec := &Spec{
 		Name:       "test",

p2p/protocols/protocol.go

@@ -123,9 +123,14 @@ type WrappedMsg struct {
 	Payload []byte
 }
 
+//For accounting, the design is to allow the Spec to describe which and how its messages are priced
+//To access this functionality, we provide a Hook interface which will call accounting methods
+//NOTE: there could be more such (horizontal) hooks in the future
 type Hook interface {
-	Send(*Peer, uint32, interface{}) error
+	//A hook for sending messages
-	Receive(*Peer, uint32, interface{}) error
+	Send(peer *Peer, size uint32, msg interface{}) error
+	//A hook for receiving messages
+	Receive(peer *Peer, size uint32, msg interface{}) error
 }
 
 // Spec is a protocol specification including its name and version as well as
@@ -147,6 +152,7 @@ type Spec struct {
 	// each message must have a single unique data type
 	Messages []interface{}
 
+	//hook for accounting (could be extended to multiple hooks in the future)
 	Hook Hook
 
 	initOnce sync.Once
@@ -287,6 +293,7 @@ func (p *Peer) Send(ctx context.Context, msg interface{}) error {
 		return errorf(ErrInvalidMsgType, "%v", code)
 	}
 
+	//if the accounting hook is set, call it
 	if p.spec.Hook != nil {
 		err := p.spec.Hook.Send(p, wmsg.Size, msg)
 		if err != nil {
@@ -353,10 +360,12 @@ func (p *Peer) handleIncoming(handle func(ctx context.Context, msg interface{})
 		return errorf(ErrDecode, "<= %v: %v", msg, err)
 	}
 
+	//if the accounting hook is set, call it
 	if p.spec.Hook != nil {
 		if wmsg.Size != uint32(len(wmsg.Payload)) {
-			log.Warn("Advertised message size and payload length don't match")
+			errMsg := "Advertised message size and payload length don't match"
-			p.Drop(errors.New("message size and payload length don't match"))
+			log.Warn(errMsg)
+			p.Drop(errors.New(errMsg))
 		}
 		err := p.spec.Hook.Receive(p, wmsg.Size, val)
 		if err != nil {

swarm/swap/swap.go

@@ -38,15 +38,18 @@ type Swap struct {
 	balances   map[enode.ID]int64 //map of balances for each peer
 }
 
-//Credit us and debit remote
+//Swap implements the protocols.Balance interface
+//Add is the (sole) accounting function
 func (s *Swap) Add(amount int64, peer *protocols.Peer) (err error) {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
+	//load existing balances from the state store
 	s.loadState(peer)
-
+	//adjust the balance
+	//if amount is negative, it will decrease, otherwise increase
 	s.balances[peer.ID()] += amount
-
+	//save the new balance to the state store
 	peerBalance := s.balances[peer.ID()]
 	s.stateStore.Put(peer.ID().String(), &peerBalance)
 
@@ -54,7 +57,7 @@ func (s *Swap) Add(amount int64, peer *protocols.Peer) (err error) {
 	return err
 }
 
-//get a peer's balance
+//Get a peer's balance
 func (swap *Swap) GetPeerBalance(peer enode.ID) (int64, error) {
 	swap.lock.RLock()
 	defer swap.lock.RUnlock()
@@ -64,9 +67,12 @@ func (swap *Swap) GetPeerBalance(peer enode.ID) (int64, error) {
 	return 0, errors.New("Peer not found")
 }
 
+//load balances from the state store (persisted)
 func (s *Swap) loadState(peer *protocols.Peer) {
 	var peerBalance int64
 	peerID := peer.ID()
+	//only load if the current instance doesn't already have this peer's
+	//balance in memory
 	if _, ok := s.balances[peerID]; !ok {
 		s.stateStore.Get(peerID.String(), &peerBalance)
 		s.balances[peerID] = peerBalance

swarm/swap/swap_test.go

@@ -43,7 +43,7 @@ func init() {
 	log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(*loglevel), log.StreamHandler(colorable.NewColorableStderr(), log.TerminalFormat(true))))
 }
 
-//check that the disconnect threshold is below the payment threshold
+//Test that repeated bookings do correct accounting
 func TestRepeatedBookings(t *testing.T) {
 	//create a test swap account
 	swap, testDir := createTestSwap(t)