swarm/swap: corrected limit handling

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
 }

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)

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
-	}
-	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])
+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()))
 	}
 }
-	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 {
@@ -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
+}