P2p validate accounting (#2051)

* p2p/protocols: change sequence of message accounting

Author: holisticode

p2p/protocols/accounting.go

@@ -71,7 +71,7 @@ type Price struct {
 // A protocol provides the message price in absolute value
 // This method then returns the correct signed amount,
 // depending on who pays, which is identified by the `payer` argument:
-// `Send` will pass a `Sender` payer, `Receive` will pass the `Receiver` argument.
+// Sending will pass a `Sender` payer, receiving will pass the `Receiver` argument.
 // Thus: If Sending and sender pays, amount negative, otherwise positive
 // If Receiving, and receiver pays, amount negative, otherwise positive
 func (p *Price) For(payer Payer, size uint32) int64 {
@@ -93,6 +93,10 @@ type Balance interface {
 	// positive amount = credit local node
 	// negative amount = debit local node
 	Add(amount int64, peer *Peer) error
+	// Check is a dry-run for the Add operation:
+	// As the accounting takes place **after** the actual send/receive operation happens,
+	// we want to make sure that that operation would not result in any problem
+	Check(amount int64, peer *Peer) error
 }
 
 // Accounting implements the Hook interface
@@ -118,44 +122,35 @@ func SetupAccountingMetrics(reportInterval time.Duration, path string) *Accounti
 	return NewAccountingMetrics(metrics.AccountingRegistry, reportInterval, path)
 }
 
-// Send takes a peer, a size and a msg and
-//   - calculates the cost for the local node sending a msg of size to peer querying the message for its price
-//   - credits/debits local node using balance interface
-func (ah *Accounting) Send(peer *Peer, size uint32, msg interface{}) error {
-	// get the price for a message
-	var pricedMessage PricedMessage
-	var ok bool
-	// if the msg implements `Price`, it is an accounted message
-	if pricedMessage, ok = msg.(PricedMessage); !ok {
-		return nil
-	}
-	// evaluate the price for sending messages
-	costToLocalNode := pricedMessage.Price().For(Sender, size)
+// Apply takes a peer, the signed cost for the local node and the msg size and credits/debits local node using balance interface
+func (ah *Accounting) Apply(peer *Peer, costToLocalNode int64, size uint32) error {
 	// do the accounting
 	err := ah.Add(costToLocalNode, peer)
 	// record metrics: just increase counters for user-facing metrics
 	ah.doMetrics(costToLocalNode, size, err)
 	return err
 }
 
-// Receive takes a peer, a size and a msg and
-//   - calculates the cost for the local node receiving a msg of size from peer querying the message for its price
-//   - credits/debits local node using balance interface
-func (ah *Accounting) Receive(peer *Peer, size uint32, msg interface{}) error {
+// Validate calculates the cost for the local node sending or receiving a msg to/from a peer querying the message for its price.
+// It returns either the signed cost for the local node as int64 or an error, signaling that the accounting operation would fail
+// (no change has been applied at this point)
+func (ah *Accounting) Validate(peer *Peer, size uint32, msg interface{}, payer Payer) (int64, error) {
 	// get the price for a message (by querying the message type via the PricedMessage interface)
 	var pricedMessage PricedMessage
 	var ok bool
 	// if the msg implements `Price`, it is an accounted message
 	if pricedMessage, ok = msg.(PricedMessage); !ok {
-		return nil
+		return 0, nil
 	}
 	// evaluate the price for receiving messages
-	costToLocalNode := pricedMessage.Price().For(Receiver, size)
-	// do the accounting
-	err := ah.Add(costToLocalNode, peer)
-	// record metrics: just increase counters for user-facing metrics
-	ah.doMetrics(costToLocalNode, size, err)
-	return err
+	costToLocalNode := pricedMessage.Price().For(payer, size)
+	// check that the operation would perform correctly
+	err := ah.Check(costToLocalNode, peer)
+	if err != nil {
+		// signal to caller that the operation would fail
+		return 0, err
+	}
+	return costToLocalNode, nil
 }
 
 // record some metrics

p2p/protocols/accounting_simulation_test.go

@@ -71,7 +71,6 @@ func TestAccountingSimulation(t *testing.T) {
 	}
 	defer os.RemoveAll(dir)
 	SetupAccountingMetrics(1*time.Second, filepath.Join(dir, "metrics.db"))
-	//define the node.Service for this test
 	services := adapters.Services{
 		"accounting": func(ctx *adapters.ServiceContext) (node.Service, error) {
 			return bal.newNode(), nil
@@ -82,8 +81,9 @@ func TestAccountingSimulation(t *testing.T) {
 	net := simulations.NewNetwork(adapter, &simulations.NetworkConfig{DefaultService: "accounting"})
 	defer net.Shutdown()
 
-	// we send msgs messages per node, wait for all messages to arrive
-	bal.wg.Add(*nodes * *msgs)
+	// we send msgs messages per node, wait for all messages to have been accounted
+	// (Add runs for each send and receive, so we need double the amount)
+	bal.wg.Add(*nodes * *msgs * 2)
 	trigger := make(chan enode.ID)
 	go func() {
 		// wait for all of them to arrive
@@ -168,13 +168,14 @@ func newMatrix(n int) *matrix {
 }
 
 // called from the testBalance's Add accounting function: register balance change
-func (m *matrix) add(i, j int, v int64) error {
+func (m *matrix) add(i, j int, v int64, wg *sync.WaitGroup) error {
 	// index for the balance of local node i with remote nodde j is
 	// i * number of nodes + remote node
 	mi := i*m.n + j
 	// register that balance
 	m.lock.Lock()
 	m.m[mi] += v
+	wg.Done()
 	m.lock.Unlock()
 	return nil
 }
@@ -229,13 +230,17 @@ type testNode struct {
 	peerCount int
 }
 
+func (t *testNode) Check(a int64, p *Peer) error {
+	return nil
+}
+
 // do the accounting for the peer's test protocol
 // testNode implements protocols.Balance
 func (t *testNode) Add(a int64, p *Peer) error {
 	//get the index for the remote peer
 	remote := t.bal.id2n[p.ID()]
 	log.Debug("add", "local", t.i, "remote", remote, "amount", a)
-	return t.bal.add(t.i, remote, a)
+	return t.bal.add(t.i, remote, a, t.bal.wg)
 }
 
 //run the p2p protocol
@@ -260,7 +265,6 @@ func (t *testNode) run(p *p2p.Peer, rw p2p.MsgReadWriter) error {
 
 // p2p message receive handler function
 func (tp *testPeer) handle(ctx context.Context, msg interface{}) error {
-	tp.wg.Done()
 	log.Debug("receive", "from", tp.remote, "to", tp.local, "type", reflect.TypeOf(msg), "msg", msg)
 	return nil
 }

p2p/protocols/accounting_test.go

@@ -95,6 +95,11 @@ func (m *zeroPriceMsg) Price() *Price {
 	}
 }
 
+//dummy accounting implementation, only stores values for later check
+func (d *dummyBalance) Check(amount int64, peer *Peer) error {
+	return nil
+}
+
 //dummy accounting implementation, only stores values for later check
 func (d *dummyBalance) Add(amount int64, peer *Peer) error {
 	d.amount = amount
@@ -168,16 +173,25 @@ func TestBalance(t *testing.T) {
 }
 
 func checkAccountingTestCases(t *testing.T, cases []testCase, acc *Accounting, peer *Peer, balance *dummyBalance, send bool) {
+	t.Helper()
 	for _, c := range cases {
 		var err error
-		var expectedResult int64
+		var expectedResult, cost int64
 		//reset balance before every check
 		balance.amount = 0
 		if send {
-			err = acc.Send(peer, c.size, c.msg)
+			cost, err = acc.Validate(peer, c.size, c.msg, Sender)
+			if err != nil {
+				t.Fatal(err)
+			}
+			err = acc.Apply(peer, cost, c.size)
 			expectedResult = c.sendResult
 		} else {
-			err = acc.Receive(peer, c.size, c.msg)
+			cost, err = acc.Validate(peer, c.size, c.msg, Receiver)
+			if err != nil {
+				t.Fatal(err)
+			}
+			err = acc.Apply(peer, cost, c.size)
 			expectedResult = c.recvResult
 		}
 

p2p/protocols/protocol.go

@@ -115,10 +115,10 @@ func errorf(code int, format string, params ...interface{}) *Error {
 //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 {
-	//A hook for sending messages
-	Send(peer *Peer, size uint32, msg interface{}) error
-	//A hook for receiving messages
-	Receive(peer *Peer, size uint32, msg interface{}) error
+	// A hook for applying accounting
+	Apply(peer *Peer, costToLocalNode int64, size uint32) error
+	// Run some validation before applying accounting
+	Validate(peer *Peer, size uint32, msg interface{}, payer Payer) (int64, error)
 }
 
 // Spec is a protocol specification including its name and version as well as
@@ -203,6 +203,7 @@ type Peer struct {
 	spec      *Spec
 	encode    func(context.Context, interface{}) (interface{}, int, error)
 	decode    func(p2p.Msg) (context.Context, []byte, error)
+	lock      sync.Mutex
 }
 
 // NewPeer constructs a new peer
@@ -277,15 +278,31 @@ func (p *Peer) Send(ctx context.Context, msg interface{}) error {
 		}
 		size = len(r)
 	}
-	// if the accounting hook is set, call it
+
+	// if the accounting hook is set, do accounting logic
 	if p.spec.Hook != nil {
-		err = p.spec.Hook.Send(p, uint32(size), msg)
+
+		// let's lock, we want to avoid that after validating, a separate call might interfere
+		p.lock.Lock()
+		defer p.lock.Unlock()
+		// validate that this operation would succeed...
+		costToLocalNode, err := p.spec.Hook.Validate(p, uint32(size), wmsg, Sender)
 		if err != nil {
+			// ...because if it would fail, we return and don't send the message
 			return err
 		}
+		// seems like accounting would succeed, thus send the message first...
+		err = p2p.Send(p.rw, code, wmsg)
+		if err != nil {
+			return err
+		}
+		// ...and finally apply (write) the accounting change
+		err = p.spec.Hook.Apply(p, costToLocalNode, uint32(size))
+	} else {
+		err = p2p.Send(p.rw, code, wmsg)
 	}
 
-	return p2p.Send(p.rw, code, wmsg)
+	return err
 }
 
 // handleIncoming(code)
@@ -322,23 +339,40 @@ 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 the accounting hook is set, do accounting logic
 	if p.spec.Hook != nil {
-		err := p.spec.Hook.Receive(p, uint32(len(msgBytes)), val)
+
+		p.lock.Lock()
+		defer p.lock.Unlock()
+
+		size := uint32(len(msgBytes))
+
+		// validate that the accounting call would succeed...
+		costToLocalNode, err := p.spec.Hook.Validate(p, size, val, Receiver)
 		if err != nil {
+			// ...because if it would fail, we return and don't handle the message
 			return err
 		}
-	}
 
-	// call the registered handler callbacks
-	// a registered callback take the decoded message as argument as an interface
-	// which the handler is supposed to cast to the appropriate type
-	// it is entirely safe not to check the cast in the handler since the handler is
-	// chosen based on the proper type in the first place
-	if err := handle(ctx, val); err != nil {
-		return errorf(ErrHandler, "(msg code %v): %v", msg.Code, err)
+		// seems like accounting would be fine, so handle the message
+		if err := handle(ctx, val); err != nil {
+			return errorf(ErrHandler, "(msg code %v): %v", msg.Code, err)
+		}
+
+		// handling succeeded, finally apply accounting
+		err = p.spec.Hook.Apply(p, costToLocalNode, size)
+	} else {
+		// call the registered handler callbacks
+		// a registered callback take the decoded message as argument as an interface
+		// which the handler is supposed to cast to the appropriate type
+		// it is entirely safe not to check the cast in the handler since the handler is
+		// chosen based on the proper type in the first place
+		if err := handle(ctx, val); err != nil {
+			return errorf(ErrHandler, "(msg code %v): %v", msg.Code, err)
+		}
 	}
-	return nil
+
+	return err
 }
 
 // Handshake negotiates a handshake on the peer connection

p2p/protocols/protocol_test.go

@@ -209,26 +209,25 @@ type dummyMsg struct {
 	Content string
 }
 
-func (d *dummyHook) Send(peer *Peer, size uint32, msg interface{}) error {
+func (d *dummyHook) Validate(peer *Peer, size uint32, msg interface{}, payer Payer) (int64, error) {
 	d.mu.Lock()
 	defer d.mu.Unlock()
-
 	d.peer = peer
 	d.size = size
 	d.msg = msg
-	d.send = true
-	return d.err
+	if payer == Sender {
+		d.send = true
+	} else {
+		d.send = false
+		d.waitC <- struct{}{}
+	}
+	return 0, d.err
 }
 
-func (d *dummyHook) Receive(peer *Peer, size uint32, msg interface{}) error {
+func (d *dummyHook) Apply(peer *Peer, cost int64, size uint32) error {
 	d.mu.Lock()
 	defer d.mu.Unlock()
 
-	d.peer = peer
-	d.size = size
-	d.msg = msg
-	d.send = false
-	d.waitC <- struct{}{}
 	return d.err
 }
 

swap/swap.go

@@ -304,7 +304,7 @@ func keyToID(key string, prefix string) enode.ID {
 	return enode.HexID(key[len(prefix):])
 }
 
-// createOwner assings keys and addresses
+// createOwner assigns keys and addresses
 func createOwner(prvkey *ecdsa.PrivateKey) *Owner {
 	pubkey := &prvkey.PublicKey
 	return &Owner{
@@ -314,20 +314,45 @@ func createOwner(prvkey *ecdsa.PrivateKey) *Owner {
 	}
 }
 
+// modifyBalanceOk checks that the amount would not result in crossing the disconnection threshold
+func (s *Swap) modifyBalanceOk(amount int64, swapPeer *Peer) (err error) {
+	// check if balance with peer is over the disconnect threshold and if the message would increase the existing debt
+	balance := swapPeer.getBalance()
+	if balance >= s.params.DisconnectThreshold && amount > 0 {
+		return fmt.Errorf("balance for peer %s is over the disconnect threshold %d and cannot incur more debt, disconnecting", swapPeer.ID().String(), s.params.DisconnectThreshold)
+	}
+
+	return nil
+}
+
+// Check is called as a *dry run* before applying the actual accounting to an operation.
+// It only checks that performing a given accounting operation would not incur in an error.
+// If it returns no error, this signals to the caller that the operation is safe
+func (s *Swap) Check(amount int64, peer *protocols.Peer) (err error) {
+	swapPeer := s.getPeer(peer.ID())
+	if swapPeer == nil {
+		return fmt.Errorf("peer %s not a swap enabled peer", peer.ID().String())
+	}
+
+	swapPeer.lock.Lock()
+	defer swapPeer.lock.Unlock()
+	// currently this is the only real check needed:
+	return s.modifyBalanceOk(amount, swapPeer)
+}
+
 // Add is the (sole) accounting function
 // Swap implements the protocols.Balance interface
 func (s *Swap) Add(amount int64, peer *protocols.Peer) (err error) {
 	swapPeer := s.getPeer(peer.ID())
 	if swapPeer == nil {
 		return fmt.Errorf("peer %s not a swap enabled peer", peer.ID().String())
 	}
+
 	swapPeer.lock.Lock()
 	defer swapPeer.lock.Unlock()
-
-	// check if balance with peer is over the disconnect threshold and if the message would increase the existing debt
-	balance := swapPeer.getBalance()
-	if balance >= s.params.DisconnectThreshold && amount > 0 {
-		return fmt.Errorf("balance for peer %s is over the disconnect threshold %d and cannot incur more debt, disconnecting", peer.ID().String(), s.params.DisconnectThreshold)
+	// we should probably check here again:
+	if err = s.modifyBalanceOk(amount, swapPeer); err != nil {
+		return err
 	}
 
 	if err = swapPeer.updateBalance(amount); err != nil {