diff --git a/p2p/protocols/accounting.go b/p2p/protocols/accounting.go index 469a43657d..06a1a58454 100644 --- a/p2p/protocols/accounting.go +++ b/p2p/protocols/accounting.go @@ -61,7 +61,13 @@ type Price struct { Payer Payer } -//ForSender gives back the price for sending a message +//For gives back the price for a message +//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. +//Thus: If Sending and sender pays, amount positive, otherwise negative +//If Receiving, and receiver pays, amount positive, otherwise negative func (p *Price) For(payer Payer, size uint32) int64 { price := p.Value if p.PerByte { @@ -114,7 +120,7 @@ func (ah *Accounting) Send(peer *Peer, size uint32, msg interface{}) error { costToLocalNode := price.For(Sender, size) //do the accounting err := ah.Add(costToLocalNode, peer) - //record metrics + //record metrics: just increase counters for user-facing metrics ah.doMetrics(costToLocalNode, size, err) return err } @@ -134,18 +140,24 @@ func (ah *Accounting) Receive(peer *Peer, size uint32, msg interface{}) error { costToLocalNode := price.For(Receiver, size) //do the accounting err := ah.Add(costToLocalNode, peer) - //record metrics + //record metrics: just increase counters for user-facing metrics ah.doMetrics(costToLocalNode, size, err) return err } //record some metrics +//this is not an error handling. `err` is returned by both `Send` and `Receive` +//`err` will only be non-nil if a limit has been violated (overdraft), in which case the peer has been dropped. +//if the limit has been violated and `err` is thus not nil: +// * if the price is positive, local node has been credited; thus `err` implicitly signals the REMOTE has been dropped +// * if the price is negative, local node has been debited, thus `err` implicitly signals LOCAL node "overdraft" func (ah *Accounting) doMetrics(price int64, size uint32, err error) { if price > 0 { mBalanceCredit.Inc(price) mBytesCredit.Inc(int64(size)) mMsgCredit.Inc(1) if err != nil { + //increase the number of times a remote node has been dropped due to "overdraft" mPeerDrops.Inc(1) } } else { @@ -153,6 +165,7 @@ func (ah *Accounting) doMetrics(price int64, size uint32, err error) { mBytesDebit.Inc(int64(size)) mMsgDebit.Inc(1) if err != nil { + //increase the number of times the local node has done an "overdraft" in respect to other nodes mSelfDrops.Inc(1) } } diff --git a/p2p/protocols/protocol.go b/p2p/protocols/protocol.go index b6329aa2c0..7dddd852fa 100644 --- a/p2p/protocols/protocol.go +++ b/p2p/protocols/protocol.go @@ -32,7 +32,6 @@ import ( "bufio" "bytes" "context" - "errors" "fmt" "io" "reflect" @@ -361,11 +360,6 @@ func (p *Peer) handleIncoming(handle func(ctx context.Context, msg interface{}) //if the accounting hook is set, call it if p.spec.Hook != nil { - if wmsg.Size != uint32(len(wmsg.Payload)) { - errMsg := "Advertised 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 { return err diff --git a/p2p/protocols/protocol_test.go b/p2p/protocols/protocol_test.go index be7c619923..2874af48d2 100644 --- a/p2p/protocols/protocol_test.go +++ b/p2p/protocols/protocol_test.go @@ -235,9 +235,13 @@ func TestProtocolHook(t *testing.T) { runFunc := func(p *p2p.Peer, rw p2p.MsgReadWriter) error { peer := NewPeer(p, rw, spec) ctx := context.TODO() - peer.Send(ctx, &dummyMsg{ + err := peer.Send(ctx, &dummyMsg{ Content: "handshake"}) + if err != nil { + t.Fatal(err) + } + handle := func(ctx context.Context, msg interface{}) error { return nil } @@ -335,7 +339,10 @@ func TestNoHook(t *testing.T) { ctx := context.TODO() msg := &perBytesMsgSenderPays{Content: "testBalance"} //send a message - peer.Send(ctx, msg) + err := peer.Send(ctx, msg) + if err != nil { + t.Fatal(err) + } //simulate receiving a message rw.msg = msg peer.handleIncoming(func(ctx context.Context, msg interface{}) error {