From 51b99805823838399c55c317c67f7b10b16ef90c Mon Sep 17 00:00:00 2001 From: Fabio Barone Date: Thu, 16 Aug 2018 18:26:52 -0500 Subject: [PATCH] swarm/swap: corrected limit handling --- swarm/swap/protocol.go | 7 ++-- swarm/swap/swap.go | 32 +++++++++++++++--- swarm/swap/swap_test.go | 74 ++++++++++++++++++++++++----------------- 3 files changed, 73 insertions(+), 40 deletions(-) diff --git a/swarm/swap/protocol.go b/swarm/swap/protocol.go index b0a27983ec..4846750c9b 100644 --- a/swarm/swap/protocol.go +++ b/swarm/swap/protocol.go @@ -133,11 +133,6 @@ func (swap *SwapProtocol) PeerInfo(id discover.NodeID) interface{} { } //------------------------------------------------------------------------------------------ - -func (swap *SwapProtocol) DebitByteCount(peer *SwapProtocolPeer, numberOfBytes int) error { - return nil -} - func (swap *SwapProtocol) Close() { } @@ -193,9 +188,11 @@ func (p *SwapProtocolPeer) handleSwapMsg(ctx context.Context, msg interface{}) e } func (sp *SwapProtocolPeer) handleIssueChequeMsg(ctx context.Context, msg interface{}) (err error) { + log.Debug("SwapProtocolPeer: handleIssueChequeMsg") return err } func (sp *SwapProtocolPeer) handleRedeemChequeMsg(ctx context.Context, msg interface{}) (err error) { + log.Debug("SwapProtocolPeer: handleRedeemChequeMsg") return err } diff --git a/swarm/swap/swap.go b/swarm/swap/swap.go index 7570ea486d..eb2e07c805 100644 --- a/swarm/swap/swap.go +++ b/swarm/swap/swap.go @@ -54,8 +54,8 @@ var ( autoDepositBuffer = big.NewInt(100000000000000) // buffer that is surplus for fork protection etc (wei) 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) + payAt = big.NewInt(-4096 * 10000) // threshold that triggers payment {request} (bytes) + dropAt = big.NewInt(-4096 * 12000) // threshold that triggers disconnect (bytes) ErrInsufficientFunds = errors.New("Insufficient funds") ErrNotAccountedMsg = errors.New("Message does not need accounting") @@ -190,6 +190,10 @@ func (sp *SwapPeer) checkAvailableFunds(ctx context.Context, msg interface{}, di price := accounted.GetMsgPrice() //local node is being credited (in its favor), so check upper limit if direction == CreditEntry { + //TODO: is there are a check needed here? + //It should actually have been done on the client side, the debitor! + //creditor could theoretically go over payAt, but if well done, + //should have been checked on the client side so this shouldn't happen? checkBalance := sp.balance.Add(sp.balance, price) //(checkBalance *Int) Cmp(payAt) // -1 if checkBalance < payAt @@ -199,6 +203,11 @@ func (sp *SwapPeer) checkAvailableFunds(ctx context.Context, msg interface{}, di return nil, ErrInsufficientFunds } } else if direction == DebitEntry { + //NOTE: ErrInsufficientFunds should only be returned + //if the dropAt is exceeded, but should be ignored for payAt, + //as there is a "clemency" margin between triggering the check + //and actually disconnecting the peer + //(checkBalance *Int) Cmp(dropAt) // -1 if checkBalance < dropAt // 0 if checkBalance == dropAt @@ -231,16 +240,29 @@ func (sp *SwapPeer) AccountMsgForPeer(ctx context.Context, msg interface{}, pric } //TODO: save to store here? init store? sp.swapAccount.stateStore.Put(sp.storeID, sp.balance) - if sp.balance.Cmp(payAt) > -1 { - //TODO: Issue Cheque + //(sp.balance *Int) Cmp(payAt) + // -1 if sp.balance < payAt + // 0 if sp.balance == payAt + // +1 if sp.balance > payAt + if sp.balance.Cmp(payAt) > 1 { + err := sp.issueCheque(ctx) + if err != nil { + //TODO: special error handling, as at this point the accounting has been done + //but the cheque could not be sent? + } } if sp.balance.Cmp(dropAt) < 0 { - //TODO: Drop peer + sp.Drop(ErrInsufficientFunds) } log.Debug(fmt.Sprintf("balance for peer %s: %s", sp.ID(), sp.balance.String())) } } +func (sp *SwapPeer) issueCheque(ctx context.Context) error { + msg := IssueChequeMsg{} + return sp.Send(ctx, msg) +} + //Create a new swap accounted peer func NewSwapPeer(peer *protocols.Peer, swap *Swap) *SwapPeer { balance := big.NewInt(0) diff --git a/swarm/swap/swap_test.go b/swarm/swap/swap_test.go index 060161b897..895134cbd3 100644 --- a/swarm/swap/swap_test.go +++ b/swarm/swap/swap_test.go @@ -50,6 +50,17 @@ var ( loglevel = flag.Int("loglevel", 2, "verbosity of logs") ) +type testExceedsPayAtMsg struct{} +type testExceedsDropAtMsg struct{} + +func (tmsg *testExceedsPayAtMsg) GetMsgPrice() *big.Int { + return payAt.Sub(big.NewInt(1)) +} + +func (tmsg *testExceedsDropAtMsg) GetMsgPrice() *big.Int { + return dropAt.Sub(big.NewInt(1)) +} + func init() { flag.Parse() @@ -57,30 +68,13 @@ func init() { log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(*loglevel), log.StreamHandler(colorable.NewColorableStderr(), log.TerminalFormat(true)))) } -func newServiceNode(port int, httpport int, wsport int, modules ...string) (*node.Node, error) { - cfg := &node.DefaultConfig - cfg.P2P.ListenAddr = fmt.Sprintf(":%d", port) - cfg.P2P.EnableMsgEvents = true - cfg.P2P.NoDiscovery = true - cfg.IPCPath = ipcpath - cfg.DataDir = fmt.Sprintf("%s%d", datadirPrefix, port) - if httpport > 0 { - cfg.HTTPHost = node.DefaultHTTPHost - cfg.HTTPPort = httpport +func TestLimits(t *testing.T) { + if dropAt >= payAt { + t.Fatal(fmt.Sprintf("dropAt limit is not lower than payAt limit, dropAt: %s, payAt: %s", dropAt.String(), payAt.String())) } - if wsport > 0 { - cfg.WSHost = node.DefaultWSHost - cfg.WSPort = wsport - cfg.WSOrigins = []string{"*"} - for i := 0; i < len(modules); i++ { - cfg.WSModules = append(cfg.WSModules, modules[i]) - } - } - stack, err := node.New(cfg) - if err != nil { - return nil, fmt.Errorf("ServiceNode create fail: %v", err) - } - return stack, nil +} + +func TestExceedsPayAt(t *testing.T) { } func TestSwapProtocol(t *testing.T) { @@ -103,12 +97,6 @@ func TestSwapProtocol(t *testing.T) { }, nil } - streamersvc := func(ctx *node.ServiceContext) (node.Service, erro) { - return &stream.API{ - streamer: NewRegistry, - }, nil - } - // register adds the service to the services the servicenode starts when started err = stack_one.Register(swapsvc) if err != nil { @@ -182,7 +170,7 @@ func TestSwapProtocol(t *testing.T) { // when we get the add event, we know we are connected ev := <-eventOneC if ev.Type != "add" { - log.Error("server #1 expected peer add", "eventtype", ev.Type) + log.Error("server #1 expected peer add", "eventtype", ev.Type) stackW.Done() return } @@ -288,3 +276,29 @@ func TestSwapProtocol(t *testing.T) { stack_two.Stop() */ } + +func newServiceNode(port int, httpport int, wsport int, modules ...string) (*node.Node, error) { + cfg := &node.DefaultConfig + cfg.P2P.ListenAddr = fmt.Sprintf(":%d", port) + cfg.P2P.EnableMsgEvents = true + cfg.P2P.NoDiscovery = true + cfg.IPCPath = ipcpath + cfg.DataDir = fmt.Sprintf("%s%d", datadirPrefix, port) + if httpport > 0 { + cfg.HTTPHost = node.DefaultHTTPHost + cfg.HTTPPort = httpport + } + if wsport > 0 { + cfg.WSHost = node.DefaultWSHost + cfg.WSPort = wsport + cfg.WSOrigins = []string{"*"} + for i := 0; i < len(modules); i++ { + cfg.WSModules = append(cfg.WSModules, modules[i]) + } + } + stack, err := node.New(cfg) + if err != nil { + return nil, fmt.Errorf("ServiceNode create fail: %v", err) + } + return stack, nil +}