diff --git a/swarm/swap/api.go b/swarm/swap/api.go index 960530e870..c224d43f2f 100644 --- a/swarm/swap/api.go +++ b/swarm/swap/api.go @@ -21,7 +21,6 @@ import ( "errors" "math/big" - "github.com/ethereum/go-ethereum/common/hexutil" "github.com/ethereum/go-ethereum/p2p/discover" ) @@ -29,25 +28,22 @@ var ( ErrNoSuchPeerAccounting = errors.New("No accounting with that peer") ) -// Wrapper for receiving pss messages when using the pss API -// providing access to sender of message -type APIMsg struct { - Msg hexutil.Bytes -} - -// Additional public methods accessible through API for pss +//This is the API definition to access swarm swap accounting data via RPC type API struct { - *SwapProtocol + *Protocol } //TODO: define metrics +//Get metrics about swap for this node type SwapMetrics struct { } -func NewAPI(swap *SwapProtocol) *API { - return &API{SwapProtocol: swap} +//Create a new API instance +func NewAPI(swap *Protocol) *API { + return &API{Protocol: swap} } +//Get the balance for this node with a specific peer func (swapapi *API) BalanceWithPeer(ctx context.Context, peer discover.NodeID) (balance *big.Int, err error) { balance = swapapi.swap.peers[peer].balance if balance == nil { @@ -56,6 +52,10 @@ func (swapapi *API) BalanceWithPeer(ctx context.Context, peer discover.NodeID) ( return } +//Get the overall balance of the node +//Iterates over all peers this node is having accounted interaction +//and just adds up balances. +//It assumes that if a disfavorable balance is represented as a negative value func (swapapi *API) Balance(ctx context.Context) (balance *big.Int, err error) { balance = big.NewInt(0) for _, peer := range swapapi.swap.peers { @@ -64,6 +64,7 @@ func (swapapi *API) Balance(ctx context.Context) (balance *big.Int, err error) { return } +//Just return the Swap metrics func (swapapi *API) GetSwapMetrics() (*SwapMetrics, error) { return nil, nil } diff --git a/swarm/swap/protocol.go b/swarm/swap/protocol.go index 78517e82e4..0ee3fc575d 100644 --- a/swarm/swap/protocol.go +++ b/swarm/swap/protocol.go @@ -33,52 +33,63 @@ const ( IsActiveProtocol = true ) -type SwapProtocol struct { +//Here we define the Swap p2p.protocol +//We use it to standardize the interaction between nodes who need clear their accounts. +//Accounting itself is separate (see swarm/swap/swap.go) +//This protocol focuses on sending and receiving cheques and +//cheque management/clearing +//For a better understanding, please read the Swap network paper "Generalized Swap swear and swindle games" +type Protocol struct { peersMu sync.RWMutex - peers map[discover.NodeID]*SwapProtocolPeer + peers map[discover.NodeID]*Peer swap *Swap } -// Peer is the Peer extension for the streaming protocol -type SwapProtocolPeer struct { +//This is the peer representing a participant in this protocol +type Peer struct { *protocols.Peer - swapProtocol *SwapProtocol + swapProtocol *Protocol } -func NewSwapProtocol(swapAccount *Swap) *SwapProtocol { - proto := &SwapProtocol{ - peers: make(map[discover.NodeID]*SwapProtocolPeer), +//Create a new protocol instance +func NewSwapProtocol(swapAccount *Swap) *Protocol { + proto := &Protocol{ + peers: make(map[discover.NodeID]*Peer), swap: swapAccount, } return proto } -// NewPeer is the constructor for Peer -func NewPeer(peer *protocols.Peer, swap *SwapProtocol) *SwapProtocolPeer { - p := &SwapProtocolPeer{ +// NewPeer is the constructor for the protocol Peer +func NewPeer(peer *protocols.Peer, swap *Protocol) *Peer { + p := &Peer{ Peer: peer, swapProtocol: swap, } return p } +//In a peer exchange, if node A gets too indebted with node B, +//node A issues a cheque and sends it to B type IssueChequeMsg struct { Cheque *Cheque } +//In a scenario where B received a cheque from A, node B can +//redeem a cheque, which means it kicks off the process to cash it in. +//In this case, this message is sent to peer A type RedeemChequeMsg struct { } ///////////////////////////////////////////////////////////////////// // SECTION: node.Service interface ///////////////////////////////////////////////////////////////////// - -func (p *SwapProtocol) Start(srv *p2p.Server) error { +func (p *Protocol) Start(srv *p2p.Server) error { log.Debug("Started swap") return nil } -func (p *SwapProtocol) Stop() error { +func (p *Protocol) Stop() error { log.Info("swap shutting down") return nil } @@ -88,11 +99,12 @@ var swapSpec = &protocols.Spec{ Version: swapVersion, MaxMsgSize: defaultMaxMsgSize, Messages: []interface{}{ - SwapMsg{}, + IssueChequeMsg{}, + RedeemChequeMsg{}, }, } -func (p *SwapProtocol) Protocols() []p2p.Protocol { +func (p *Protocol) Protocols() []p2p.Protocol { return []p2p.Protocol{ { Name: swapSpec.Name, @@ -103,7 +115,7 @@ func (p *SwapProtocol) Protocols() []p2p.Protocol { } } -func (p *SwapProtocol) APIs() []rpc.API { +func (p *Protocol) APIs() []rpc.API { apis := []rpc.API{ { Namespace: "swap", @@ -118,7 +130,7 @@ func (p *SwapProtocol) APIs() []rpc.API { ///////////////////////////////////////////////////////////////////// // SECTION: p2p.protocol interface ///////////////////////////////////////////////////////////////////// -func (swap *SwapProtocol) Run(peer *p2p.Peer, rw p2p.MsgReadWriter) error { +func (swap *Protocol) Run(peer *p2p.Peer, rw p2p.MsgReadWriter) error { p := protocols.NewPeer(peer, rw, swapSpec) sp := NewPeer(p, swap) swap.setPeer(sp) @@ -127,50 +139,51 @@ func (swap *SwapProtocol) Run(peer *p2p.Peer, rw p2p.MsgReadWriter) error { return sp.Run(sp.handleSwapMsg) } -func (swap *SwapProtocol) NodeInfo() interface{} { +func (swap *Protocol) NodeInfo() interface{} { return nil } -func (swap *SwapProtocol) PeerInfo(id discover.NodeID) interface{} { +func (swap *Protocol) PeerInfo(id discover.NodeID) interface{} { return nil } //------------------------------------------------------------------------------------------ -func (swap *SwapProtocol) Close() { +func (swap *Protocol) Close() { } -func (swap *SwapProtocol) getPeer(peerId discover.NodeID) *SwapProtocolPeer { +func (swap *Protocol) getPeer(peerId discover.NodeID) *Peer { swap.peersMu.RLock() defer swap.peersMu.RUnlock() return swap.peers[peerId] } -func (swap *SwapProtocol) setPeer(peer *SwapProtocolPeer) { +func (swap *Protocol) setPeer(peer *Peer) { swap.peersMu.Lock() defer swap.peersMu.Unlock() swap.peers[peer.ID()] = peer - metrics.GetOrRegisterGauge("registry.peers", nil).Update(int64(len(swap.peers))) + metrics.GetOrRegisterGauge("swap.peers", nil).Update(int64(len(swap.peers))) } -func (swap *SwapProtocol) deletePeer(peer *SwapProtocolPeer) { +func (swap *Protocol) deletePeer(peer *Peer) { swap.peersMu.Lock() defer swap.peersMu.Unlock() delete(swap.peers, peer.ID()) - metrics.GetOrRegisterGauge("registry.peers", nil).Update(int64(len(swap.peers))) + metrics.GetOrRegisterGauge("swap.peers", nil).Update(int64(len(swap.peers))) swap.peersMu.Unlock() } -func (swap *SwapProtocol) peersCount() (c int) { +func (swap *Protocol) peersCount() (c int) { swap.peersMu.Lock() c = len(swap.peers) swap.peersMu.Unlock() return } -func (p *SwapProtocolPeer) handleSwapMsg(ctx context.Context, msg interface{}) error { +//Protocol message handler for handling cheque messages +func (p *Peer) handleSwapMsg(ctx context.Context, msg interface{}) error { switch msg := msg.(type) { case *IssueChequeMsg: @@ -179,22 +192,19 @@ func (p *SwapProtocolPeer) handleSwapMsg(ctx context.Context, msg interface{}) e case *RedeemChequeMsg: return p.handleRedeemChequeMsg(ctx, msg) - /* - case *QuitMsg: - return p.handleQuitMsg(msg) - */ - default: return fmt.Errorf("unknown message type: %T", msg) } } -func (sp *SwapProtocolPeer) handleIssueChequeMsg(ctx context.Context, msg interface{}) (err error) { +//A IssueChequeMsg has been received +func (sp *Peer) 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) { +//A RedeemChequeMsg has been received +func (sp *Peer) 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 f0f27443e1..c31c528906 100644 --- a/swarm/swap/swap.go +++ b/swarm/swap/swap.go @@ -18,26 +18,16 @@ package swap import ( "context" - "crypto/ecdsa" "errors" "fmt" "math/big" - "os" - "path/filepath" "sync" "time" - "github.com/ethereum/go-ethereum/accounts/abi/bind" - "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/contracts/chequebook" - "github.com/ethereum/go-ethereum/contracts/chequebook/contract" - "github.com/ethereum/go-ethereum/core/types" - "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/p2p/protocols" "github.com/ethereum/go-ethereum/swarm/log" "github.com/ethereum/go-ethereum/swarm/state" - whisper "github.com/ethereum/go-ethereum/whisper/whisperv5" ) const ( @@ -47,15 +37,8 @@ const ( ) var ( - autoCashInterval = 300 * time.Second // default interval for autocash - autoCashThreshold = big.NewInt(50000000000000) // threshold that triggers autocash (wei) - autoDepositInterval = 300 * time.Second // default interval for autocash - autoDepositThreshold = big.NewInt(50000000000000) // threshold that triggers autodeposit (wei) - 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 * 12000) // 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") @@ -66,17 +49,19 @@ const ( chequebookDeployDelay = 1 * time.Second // delay between retries ) -// SwAP Swarm Accounting Protocol with -// Swift Automatic Payments +// SwAP Swarm Accounting Protocol // a peer to peer micropayment system +// A node maintains an individual balance with every peer +// Only messages which have a price will be accounted for type Swap struct { chequeManager *ChequeManager stateStore state.Store lock sync.RWMutex peers map[discover.NodeID]*SwapPeer - local *Params // local peer's swap parameters } +//Protocols which want to send and handle priced messages will need to use +//this peer instead of protocols.Peer, which is embedded type SwapPeer struct { *protocols.Peer lock sync.RWMutex @@ -86,6 +71,7 @@ type SwapPeer struct { storeID string } +//This defines if a price will be debited or credited to an account type EntryDirection bool const ( @@ -93,8 +79,9 @@ const ( CreditEntry EntryDirection = false ) -type SwapAccountedMsgType interface { - GetMsgPrice() *big.Int +//A message which needs accounting needs to implement this interface +type PricedMsg interface { + Price() *big.Int } //Handler for received messages @@ -104,7 +91,7 @@ func (sp *SwapPeer) RunAccountedProtocol(protocolHandler func(ctx context.Contex sp.handlerFunc = protocolHandler //then run the handler loop function - return sp.Run(sp.handleAccountedMsg) + return sp.Run(sp.handle) } //get a peer's balance @@ -118,12 +105,12 @@ func (swap *Swap) GetPeerBalance(peer discover.NodeID) *big.Int { //Handle a received message; this is the handler loop function. //Check if it needs accounting, and if yes, apply accounting logic: //Check for sufficient funds, perform operation, then account -func (sp *SwapPeer) handleAccountedMsg(ctx context.Context, msg interface{}) error { +func (sp *SwapPeer) handle(ctx context.Context, msg interface{}) error { var err error var price *big.Int //the message is one which needs accounting... - if _, ok := msg.(SwapAccountedMsgType); ok { + if _, ok := msg.(PricedMsg); ok { //..so first check if there are enough funds for the operation available //(for crediting, this means if we are not essentially "overdrafting", or crossing the threshold) price, err = sp.checkAvailableFunds(ctx, msg, CreditEntry) @@ -157,7 +144,7 @@ func (sp *SwapPeer) Send(ctx context.Context, msg interface{}) error { var price *big.Int //the message is one which needs accounting... - if _, ok := msg.(SwapAccountedMsgType); ok { + if _, ok := msg.(PricedMsg); ok { //..so first check if there are enough funds for the operation available price, err = sp.checkAvailableFunds(ctx, msg, DebitEntry) //if not (or some other error occured), return error @@ -187,8 +174,8 @@ func (sp *SwapPeer) checkAvailableFunds(ctx context.Context, msg interface{}, di sp.lock.Lock() defer sp.lock.Unlock() - if accounted, ok := msg.(SwapAccountedMsgType); ok { - price := accounted.GetMsgPrice() + if accounted, ok := msg.(PricedMsg); ok { + price := accounted.Price() //local node is being credited (in its favor), so check upper limit if direction == CreditEntry { //TODO: is there a check needed here? @@ -229,7 +216,7 @@ func (sp *SwapPeer) checkAvailableFunds(ctx context.Context, msg interface{}, di //Thus, we credit the balance and increase it when the amount is in favor of the local node //We debit the balance and decrease it when the amount is in favor of the remote peer func (sp *SwapPeer) AccountMsgForPeer(ctx context.Context, msg interface{}, price *big.Int, direction EntryDirection) { - if _, ok := msg.(SwapAccountedMsgType); ok { + if _, ok := msg.(PricedMsg); ok { sp.lock.Lock() defer sp.lock.Unlock() //local node is being credited (in its favor), so its balance increases @@ -252,6 +239,7 @@ func (sp *SwapPeer) AccountMsgForPeer(ctx context.Context, msg interface{}, pric if err != nil { //TODO: special error handling, as at this point the accounting has been done //but the cheque could not be sent? + log.Warn("Payment threshold exceeded, but error sending cheque!", "err", err) } } if sp.balance.Cmp(dropAt) == -1 { @@ -261,12 +249,15 @@ func (sp *SwapPeer) AccountMsgForPeer(ctx context.Context, msg interface{}, pric } } +//Issue a cheque for the remote peer. Happens if we are indebted with the peer +//and crossed the payment threshold func (sp *SwapPeer) issueCheque(ctx context.Context) error { - amount := big.NewInt(0) + amount := &big.Int{} cheque := sp.swapAccount.chequeManager.CreateCheque(sp.ID(), amount.Abs(payAt)) msg := IssueChequeMsg{ Cheque: cheque, } + //TODO: This should now be via the actual SwapProtocol return sp.Send(ctx, msg) } @@ -287,303 +278,14 @@ func NewSwapPeer(peer *protocols.Peer, swap *Swap) *SwapPeer { return sp } -// Profile - public swap profile -// public parameters for SWAP, serializable config struct passed in handshake -type Profile struct { - BuyAt *big.Int // accepted max price for chunk - SellAt *big.Int // offered sale price for chunk - PayAt *big.Int // threshold that triggers payment request - DropAt *big.Int // threshold that triggers disconnect -} - -// Strategy encapsulates parameters relating to -// automatic deposit and automatic cashing -type Strategy struct { - AutoCashInterval time.Duration // default interval for autocash - AutoCashThreshold *big.Int // threshold that triggers autocash (wei) - AutoDepositInterval time.Duration // default interval for autocash - AutoDepositThreshold *big.Int // threshold that triggers autodeposit (wei) - AutoDepositBuffer *big.Int // buffer that is surplus for fork protection etc (wei) -} - -// SwapMsg encapsulates messages transported over pss. -type SwapMsg struct { - To []byte - Control []byte - Expire uint32 - Payload *whisper.Envelope -} - -// Params extends the public profile with private parameters relating to -// automatic deposit and automatic cashing -type Params struct { - *Profile - *Strategy -} - -// LocalProfile combines a PayProfile with *swap.Params -type LocalProfile struct { - *Params - *PayProfile -} - -// RemoteProfile combines a PayProfile with *swap.Profile -type RemoteProfile struct { - *Profile - *PayProfile -} - -// PayProfile is a container for relevant chequebook and beneficiary options -type PayProfile struct { - PublicKey string // check against signature of promise - Contract common.Address // address of chequebook contract - Beneficiary common.Address // recipient address for swarm sales revenue - privateKey *ecdsa.PrivateKey - publicKey *ecdsa.PublicKey - owner common.Address - chbook *chequebook.Chequebook - lock sync.RWMutex -} - // New - swap constructor -func NewSwap(local *Params, stateStore state.Store) (swap *Swap, err error) { +func New(stateStore state.Store) (swap *Swap, err error) { swap = &Swap{ chequeManager: NewChequeManager(stateStore), - local: local, stateStore: stateStore, peers: make(map[discover.NodeID]*SwapPeer), } - //swap.SetParams(local) return } - -// NewDefaultSwapParams create params with default values -func NewDefaultSwapParams() *LocalProfile { - return &LocalProfile{ - PayProfile: &PayProfile{}, - Params: &Params{ - Profile: &Profile{ - BuyAt: buyAt, - SellAt: sellAt, - PayAt: payAt, - DropAt: dropAt, - }, - Strategy: &Strategy{ - AutoCashInterval: autoCashInterval, - AutoCashThreshold: autoCashThreshold, - AutoDepositInterval: autoDepositInterval, - AutoDepositThreshold: autoDepositThreshold, - AutoDepositBuffer: autoDepositBuffer, - }, - }, - } -} - -// Init this can only finally be set after all config options (file, cmd line, env vars) -// have been evaluated -func (lp *LocalProfile) Init(contract common.Address, prvkey *ecdsa.PrivateKey) { - pubkey := &prvkey.PublicKey - - lp.PayProfile = &PayProfile{ - PublicKey: common.ToHex(crypto.FromECDSAPub(pubkey)), - Contract: contract, - Beneficiary: crypto.PubkeyToAddress(*pubkey), - privateKey: prvkey, - publicKey: pubkey, - owner: crypto.PubkeyToAddress(*pubkey), - } -} - -// Chequebook get's chequebook from the localProfile -func (lp *LocalProfile) Chequebook() *chequebook.Chequebook { - defer lp.lock.Unlock() - lp.lock.Lock() - return lp.chbook -} - -// PrivateKey accessor -func (lp *LocalProfile) PrivateKey() *ecdsa.PrivateKey { - return lp.privateKey -} - -// func (self *LocalProfile) PublicKey() *ecdsa.PublicKey { -// return self.publicKey -// } - -// SetKey set's private and public key on localProfile -func (lp *LocalProfile) SetKey(prvkey *ecdsa.PrivateKey) { - lp.privateKey = prvkey - lp.publicKey = &prvkey.PublicKey -} - -// SetChequebook wraps the chequebook initialiser and sets up autoDeposit to cover spending. -func (lp *LocalProfile) SetChequebook(ctx context.Context, backend chequebook.Backend, path string) error { - lp.lock.Lock() - swapContract := lp.Contract - lp.lock.Unlock() - - valid, err := chequebook.ValidateCode(ctx, backend, swapContract) - if err != nil { - return err - } else if valid { - return lp.newChequebookFromContract(path, backend) - } - return lp.deployChequebook(ctx, backend, path) -} - -// deployChequebook deploys the localProfile Chequebook -func (lp *LocalProfile) deployChequebook(ctx context.Context, backend chequebook.Backend, path string) error { - opts := bind.NewKeyedTransactor(lp.privateKey) - opts.Value = lp.AutoDepositBuffer - opts.Context = ctx - - log.Info(fmt.Sprintf("Deploying new chequebook (owner: %v)", opts.From.Hex())) - address, err := deployChequebookLoop(opts, backend) - if err != nil { - log.Error(fmt.Sprintf("unable to deploy new chequebook: %v", err)) - return err - } - log.Info(fmt.Sprintf("new chequebook deployed at %v (owner: %v)", address.Hex(), opts.From.Hex())) - - // need to save config at this point - lp.lock.Lock() - lp.Contract = address - err = lp.newChequebookFromContract(path, backend) - lp.lock.Unlock() - if err != nil { - log.Warn(fmt.Sprintf("error initialising cheque book (owner: %v): %v", opts.From.Hex(), err)) - } - return err -} - -// deployChequebookLoop repeatedly tries to deploy a chequebook. -func deployChequebookLoop(opts *bind.TransactOpts, backend chequebook.Backend) (addr common.Address, err error) { - var tx *types.Transaction - for try := 0; try < chequebookDeployRetries; try++ { - if try > 0 { - time.Sleep(chequebookDeployDelay) - } - if _, tx, _, err = contract.DeployChequebook(opts, backend); err != nil { - log.Warn(fmt.Sprintf("can't send chequebook deploy tx (try %d): %v", try, err)) - continue - } - if addr, err = bind.WaitDeployed(opts.Context, backend, tx); err != nil { - log.Warn(fmt.Sprintf("chequebook deploy error (try %d): %v", try, err)) - continue - } - return addr, nil - } - return addr, err -} - -// newChequebookFromContract - initialise the chequebook from a persisted json file or create a new one -// caller holds the lock -func (lp *LocalProfile) newChequebookFromContract(path string, backend chequebook.Backend) error { - hexkey := common.Bytes2Hex(lp.Contract.Bytes()) - err := os.MkdirAll(filepath.Join(path, "chequebooks"), os.ModePerm) - if err != nil { - return fmt.Errorf("unable to create directory for chequebooks: %v", err) - } - - chbookpath := filepath.Join(path, "chequebooks", hexkey+".json") - lp.chbook, err = chequebook.LoadChequebook(chbookpath, lp.privateKey, backend, true) - - if err != nil { - lp.chbook, err = chequebook.NewChequebook(chbookpath, lp.Contract, lp.privateKey, backend) - if err != nil { - log.Warn(fmt.Sprintf("unable to initialise chequebook (owner: %v): %v", lp.owner.Hex(), err)) - return fmt.Errorf("unable to initialise chequebook (owner: %v): %v", lp.owner.Hex(), err) - } - } - - lp.chbook.AutoDeposit(lp.AutoDepositInterval, lp.AutoDepositThreshold, lp.AutoDepositBuffer) - log.Info(fmt.Sprintf("auto deposit ON for %v -> %v: interval = %v, threshold = %v, buffer = %v)", crypto.PubkeyToAddress(*(lp.publicKey)).Hex()[:8], lp.Contract.Hex()[:8], lp.AutoDepositInterval, lp.AutoDepositThreshold, lp.AutoDepositBuffer)) - - return nil -} - -/* -// Add (n) -// n > 0 called when promised/provided n units of service -// n < 0 called when used/requested n units of service -func (swap *Swap) Add(n int) error { - //defer swap.lock.Unlock() - //swap.lock.Lock() - swap.balance += n - if !swap.Sells && swap.balance > 0 { - log.Trace(fmt.Sprintf("<%v> remote peer cannot have debt (balance: %v)", swap.proto, swap.balance)) - swap.proto.Drop() - return fmt.Errorf("[SWAP] <%v> remote peer cannot have debt (balance: %v)", swap.proto, swap.balance) - } - if !swap.Buys && swap.balance < 0 { - log.Trace(fmt.Sprintf("<%v> we cannot have debt (balance: %v)", swap.proto, swap.balance)) - return fmt.Errorf("[SWAP] <%v> we cannot have debt (balance: %v)", swap.proto, swap.balance) - } - if swap.balance >= int(swap.local.DropAt) { - log.Trace(fmt.Sprintf("<%v> remote peer has too much debt (balance: %v, disconnect threshold: %v)", swap.proto, swap.balance, swap.local.DropAt)) - swap.proto.Drop() - return fmt.Errorf("[SWAP] <%v> remote peer has too much debt (balance: %v, disconnect threshold: %v)", swap.proto, swap.balance, swap.local.DropAt) - } else if swap.balance <= -int(swap.remote.PayAt) { - swap.send() - } - return nil -} - -// Balance accessor -func (swap *Swap) Balance() int { - //defer swap.lock.Unlock() - //swap.lock.Lock() - return swap.balance -} - -/* -// send (units) is called when payment is due -// In case of insolvency no promise is issued and sent, safe against fraud -// No return value: no error = payment is opportunistic = hang in till dropped -func (swap *Swap) send() { - if swap.local.BuyAt != nil && swap.balance < 0 { - amount := big.NewInt(int64(-swap.balance)) - amount.Mul(amount, swap.remote.SellAt) - promise, err := swap.Out.Issue(amount) - if err != nil { - log.Warn(fmt.Sprintf("<%v> cannot issue cheque (amount: %v, channel: %v): %v", swap.proto, amount, swap.Out, err)) - } else { - log.Warn(fmt.Sprintf("<%v> cheque issued (amount: %v, channel: %v)", swap.proto, amount, swap.Out)) - swap.proto.Pay(-swap.balance, promise) - swap.balance = 0 - } - } -} - -// Receive (units, promise) is called by the protocol when a payment msg is received -// returns error if promise is invalid. -func (swap *Swap) Receive(units int, promise Promise) error { - if units <= 0 { - return fmt.Errorf("invalid units: %v <= 0", units) - } - - price := new(big.Int).SetInt64(int64(units)) - price.Mul(price, swap.local.SellAt) - - amount, err := swap.In.Receive(promise) - - if err != nil { - err = fmt.Errorf("invalid promise: %v", err) - } else if price.Cmp(amount) != 0 { - // verify amount = units * unit sale price - return fmt.Errorf("invalid amount: %v = %v * %v (units sent in msg * agreed sale unit price) != %v (signed in cheque)", price, units, swap.local.SellAt, amount) - } - if err != nil { - log.Trace(fmt.Sprintf("<%v> invalid promise (amount: %v, channel: %v): %v", swap.proto, amount, swap.In, err)) - return err - } - - // credit remote peer with units - swap.Add(-units) - log.Trace(fmt.Sprintf("<%v> received promise (amount: %v, channel: %v): %v", swap.proto, amount, swap.In, promise)) - - return nil -} -*/ diff --git a/swarm/swap/swap_test.go b/swarm/swap/swap_test.go index 403716125f..e6b3f6da98 100644 --- a/swarm/swap/swap_test.go +++ b/swarm/swap/swap_test.go @@ -57,6 +57,9 @@ var testSpec = &protocols.Spec{ }, } +//dummy implementation of a MsgReadWriter +//this allows for quick and easy unit tests without +//having to build up the complete protocol type dummyRW struct{} func (d *dummyRW) WriteMsg(msg p2p.Msg) error { @@ -72,21 +75,27 @@ func (d *dummyRW) ReadMsg() (p2p.Msg, error) { }, nil } +//define a couple of messages for tests type testExceedsPayAtMsg struct{} type testExceedsDropAtMsg struct{} type testCheapMsg struct{} -func (tmsg *testExceedsPayAtMsg) GetMsgPrice() *big.Int { +//this message is just one unit more expensive than the payment threshold +func (tmsg *testExceedsPayAtMsg) Price() *big.Int { diff := &big.Int{} - return diff.Sub(payAt, big.NewInt(1)) + diff = diff.Abs(payAt) + return diff.Add(diff, big.NewInt(1)) } -func (tmsg *testExceedsDropAtMsg) GetMsgPrice() *big.Int { +//this message is just one unit more expensive than the disconnect threshold +func (tmsg *testExceedsDropAtMsg) Price() *big.Int { diff := &big.Int{} - return diff.Sub(dropAt, big.NewInt(1)) + diff = diff.Abs(dropAt) + return diff.Add(diff, big.NewInt(1)) } -func (tmsg *testCheapMsg) GetMsgPrice() *big.Int { +//a message with an arbitrary cost +func (tmsg *testCheapMsg) Price() *big.Int { return big.NewInt(100) } @@ -97,6 +106,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 func TestLimits(t *testing.T) { if dropAt.Cmp(payAt) > -1 { t.Fatal(fmt.Sprintf("dropAt limit is not lower than payAt limit, dropAt: %s, payAt: %s", dropAt.String(), payAt.String())) @@ -104,17 +114,29 @@ func TestLimits(t *testing.T) { } //unit test for exceeds pay limit +//when the payment threshold is reached, a cheque will be issued +//this test checks that a cheque is present if a message is sent +//which exceeds the payment threshold +//(note: the details of cheque handling will need to be fleshed out +//in future iterations, current implementation is very primitive) func TestExceedsPayAt(t *testing.T) { + //create a test swap account swap, testDir := createTestSwap(t) defer os.RemoveAll(testDir) + //create a dummy peer testPeer := newDummyPeer() sp := NewSwapPeer(testPeer, swap) sp.handlerFunc = dummyMsgHandler + //send the message ctx := context.Background() - sp.Send(ctx, &testExceedsPayAtMsg{}) + err := sp.Send(ctx, &testExceedsPayAtMsg{}) + if err != nil { + t.Fatal("Unecpected error on sending message", "err", err) + } + //check that a cheque is present cheques := sp.swapAccount.chequeManager.openDebitCheques[sp.ID()] if cheques == nil { t.Fatal("Expected cheques for this peer to be present, but are nil") @@ -132,6 +154,8 @@ func TestExceedsPayAt(t *testing.T) { } //unit test for exceeds drop limit +//tests that a message is being sent which crosses the drop limit +//in that case, we should receive a InsufficientFunds error func TestExceedsDropAt(t *testing.T) { swap, testDir := createTestSwap(t) defer os.RemoveAll(testDir) @@ -147,15 +171,19 @@ func TestExceedsDropAt(t *testing.T) { } } +//send a message with cost, +//then check that the balance has the expected amount func TestSendCheapMessage(t *testing.T) { swap, testDir := createTestSwap(t) defer os.RemoveAll(testDir) testPeer := newDummyPeer() sp := NewSwapPeer(testPeer, swap) + //set an arbitrary test balance value testBalance := big.NewInt(1234567890) sp.balance = testBalance + //send the message msg := &testCheapMsg{} ctx := context.Background() err := sp.Send(ctx, msg) @@ -163,19 +191,30 @@ func TestSendCheapMessage(t *testing.T) { t.Fatal("Unexpected error sending message") } - if sp.balance.Cmp(testBalance.Sub(testBalance, msg.GetMsgPrice())) != 0 { + //check the new balance + if sp.balance.Cmp(testBalance.Sub(testBalance, msg.Price())) != 0 { t.Fatal(fmt.Sprintf("Unexpected balance value after sending cheap message test. Expected balance: %s, balance is: %s", - testBalance.Sub(testBalance, msg.GetMsgPrice()).String(), sp.balance.String())) + testBalance.Sub(testBalance, msg.Price()).String(), sp.balance.String())) } } +//try restoring a balance from state store +//this is simulated by creating a node, +//assigning it an arbitrary balance, +//send a message (triggers to save to store), +//then create a different SwapPeer instance with same peerID, +//which will try to load a balance from the stateStore func TestRestoreBalanceFromStateStore(t *testing.T) { swap, testDir := createTestSwap(t) defer os.RemoveAll(testDir) + //create the dummy p2p protocol Peer testPeer := newDummyPeer() + //create the "source" swap peer sp := NewSwapPeer(testPeer, swap) + //create a reference an arbitrary balance testBalance := big.NewInt(1234567890) + //assign the same value to the peer sp.balance = big.NewInt(1234567890) //send a message, should trigger saving to stateStore @@ -187,16 +226,22 @@ func TestRestoreBalanceFromStateStore(t *testing.T) { t.Fatal("Unexpected error sending message") } + //create a new peer with same underlying protocols.Peer + //this will try to load the balance from the stateStore, + //as it is the same discover.NodeID sp2 := NewSwapPeer(testPeer, swap) + //compare the balances expectedBalance := &big.Int{} - expectedBalance.Sub(testBalance, msg.GetMsgPrice()) + expectedBalance.Sub(testBalance, msg.Price()) if sp2.balance.Cmp(expectedBalance) != 0 { t.Fatal(fmt.Sprintf("Unexpected balance value after sending cheap message test. Expected balance: %s, balance is: %s", expectedBalance.String(), sp2.balance.String())) } } +//create a test swap account +//creates a stateStore for persistence and a Swap account func createTestSwap(t *testing.T) (*Swap, string) { dir, err := ioutil.TempDir("", "swap_test_store") if err != nil { @@ -206,22 +251,19 @@ func createTestSwap(t *testing.T) (*Swap, string) { if err2 != nil { t.Fatal(err2) } - swap, err3 := NewSwap(NewDefaultSwapParams().Params, stateStore) + swap, err3 := New(stateStore) if err3 != nil { t.Fatal(err3) } return swap, dir } -func runProtocol(peer *protocols.Peer, swap *Swap) { - sp := NewSwapPeer(peer, swap) - sp.Peer.Run(dummyMsgHandler) -} - +//dummy message handler (needed or we will have a panic in the accounting) func dummyMsgHandler(ctx context.Context, msg interface{}) error { return nil } +//tests some basic things over RPC func TestSwapRPC(t *testing.T) { swap, testDir := createTestSwap(t) defer os.RemoveAll(testDir) @@ -236,7 +278,7 @@ func TestSwapRPC(t *testing.T) { // wrapper function for servicenode to start the service swapsvc := func(ctx *node.ServiceContext) (node.Service, error) { return &API{ - SwapProtocol: instance, + Protocol: instance, }, nil } @@ -314,11 +356,13 @@ func TestSwapRPC(t *testing.T) { } } +//creates a dummy protocols.Peer with dummy MsgReadWriter func newDummyPeer() *protocols.Peer { id := adapters.RandomNodeConfig().ID return protocols.NewPeer(p2p.NewPeer(id, "testPeer", nil), &dummyRW{}, testSpec) } +//creates a p2p.Service node stub func newServiceNode(port int, httpport int, wsport int, modules ...string) (*node.Node, error) { cfg := &node.DefaultConfig cfg.P2P.ListenAddr = fmt.Sprintf(":%d", port)