From bb9dd2e57a0b0fb1c19261e0ac38acc322cd097e Mon Sep 17 00:00:00 2001 From: Fabio Barone Date: Thu, 2 Aug 2018 15:03:50 -0500 Subject: [PATCH 1/8] swarm: swap implementation as devp2p2 protocol --- swarm/swap/api.go | 62 ++++++ swarm/swap/protocol.go | 177 ++++++++++++++++ swarm/swap/swap.go | 433 ++++++++++++++++++++++++++++++++++++++++ swarm/swap/swap_test.go | 275 +++++++++++++++++++++++++ swarm/swarm.go | 13 ++ 5 files changed, 960 insertions(+) create mode 100644 swarm/swap/api.go create mode 100644 swarm/swap/protocol.go create mode 100644 swarm/swap/swap.go create mode 100644 swarm/swap/swap_test.go diff --git a/swarm/swap/api.go b/swarm/swap/api.go new file mode 100644 index 0000000000..2de11d9efa --- /dev/null +++ b/swarm/swap/api.go @@ -0,0 +1,62 @@ +// Copyright 2018 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package swap + +import ( + "context" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/hexutil" +) + +// 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 +type API struct { + *Swap +} + +type SwapMetrics struct { +} + +type Cheque struct { +} + +func NewAPI(swap *Swap) *API { + return &API{Swap: swap} +} + +func (swapapi *API) Balance(ctx context.Context) (balance int, err error) { + balance = 0 + err = nil + return +} + +func (swapapi *API) GetSwapMetrics() (*SwapMetrics, error) { + return nil, nil +} + +func (swapapi *API) IssueCheque(recipient *common.Address) (*Cheque, error) { + return nil, nil +} + +func (swapapi *API) RedeemCheque(cheque *Cheque) { +} diff --git a/swarm/swap/protocol.go b/swarm/swap/protocol.go new file mode 100644 index 0000000000..63acbbe929 --- /dev/null +++ b/swarm/swap/protocol.go @@ -0,0 +1,177 @@ +// Copyright 2018 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +// +build !nopssprotocol + +package swap + +import ( + "sync" + "time" + + "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/p2p/protocols" + "github.com/ethereum/go-ethereum/rlp" +) + +const ( + IsActiveProtocol = true +) + +// Convenience wrapper for devp2p protocol messages for transport over pss +type ProtocolMsg struct { + Code uint64 + Size uint32 + Payload []byte + ReceivedAt time.Time +} + +// Creates a ProtocolMsg +func NewProtocolMsg(code uint64, msg interface{}) ([]byte, error) { + + rlpdata, err := rlp.EncodeToBytes(msg) + if err != nil { + return nil, err + } + + // TODO verify that nested structs cannot be used in rlp + smsg := &ProtocolMsg{ + Code: code, + Size: uint32(len(rlpdata)), + Payload: rlpdata, + } + + return rlp.EncodeToBytes(smsg) +} + +// Protocol options to be passed to a new Protocol instance +// +// The parameters specify which encryption schemes to allow +type ProtocolParams struct { +} + +// Convenience object for emulation devp2p over pss +type Protocol struct { + proto *p2p.Protocol + spec *protocols.Spec + RWPoolMu sync.Mutex +} + +/* +// Activates devp2p emulation over a specific pss topic +// +// One or both encryption schemes must be specified. If +// only one is specified, the protocol will not be valid +// for the other, and will make the message handler +// return errors +func RegisterProtocol(ps *Pss, topic *Topic, spec *protocols.Spec, targetprotocol *p2p.Protocol, options *ProtocolParams) (*Protocol, error) { + if !options.Asymmetric && !options.Symmetric { + return nil, fmt.Errorf("specify at least one of asymmetric or symmetric messaging mode") + } + pp := &Protocol{ + Pss: ps, + proto: targetprotocol, + topic: topic, + spec: spec, + pubKeyRWPool: make(map[string]p2p.MsgReadWriter), + symKeyRWPool: make(map[string]p2p.MsgReadWriter), + Asymmetric: options.Asymmetric, + Symmetric: options.Symmetric, + } + return pp, nil +} + +*/ +// Generic handler for incoming messages over devp2p emulation +// +// To be passed to pss.Register() +// +// Will run the protocol on a new incoming peer, provided that +// the encryption key of the message has a match in the internal +// pss keypool +// +// Fails if protocol is not valid for the message encryption scheme, +// if adding a new peer fails, or if the message is not a serialized +// p2p.Msg (which it always will be if it is sent from this object). +func (p *Protocol) Handle(msg []byte, peer *p2p.Peer, asymmetric bool, keyid string) error { + return nil +} + +/* +// Runs an emulated pss Protocol on the specified peer, +// linked to a specific topic +// `key` and `asymmetric` specifies what encryption key +// to link the peer to. +// The key must exist in the pss store prior to adding the peer. +func (p *Protocol) AddPeer(peer *p2p.Peer, topic Topic, asymmetric bool, key string) (p2p.MsgReadWriter, error) { + rw := &PssReadWriter{ + Pss: p.Pss, + rw: make(chan p2p.Msg), + spec: p.spec, + topic: p.topic, + key: key, + } + if asymmetric { + rw.sendFunc = p.Pss.SendAsym + } else { + rw.sendFunc = p.Pss.SendSym + } + if asymmetric { + p.Pss.pubKeyPoolMu.Lock() + if _, ok := p.Pss.pubKeyPool[key]; !ok { + return nil, fmt.Errorf("asym key does not exist: %s", key) + } + p.Pss.pubKeyPoolMu.Unlock() + p.RWPoolMu.Lock() + p.pubKeyRWPool[key] = rw + p.RWPoolMu.Unlock() + } else { + p.Pss.symKeyPoolMu.Lock() + if _, ok := p.Pss.symKeyPool[key]; !ok { + return nil, fmt.Errorf("symkey does not exist: %s", key) + } + p.Pss.symKeyPoolMu.Unlock() + p.RWPoolMu.Lock() + p.symKeyRWPool[key] = rw + p.RWPoolMu.Unlock() + } + go func() { + err := p.proto.Run(peer, rw) + log.Warn(fmt.Sprintf("pss vprotocol quit on %v topic %v: %v", peer, topic, err)) + }() + return rw, nil +} + +func (p *Protocol) RemovePeer(asymmetric bool, key string) { + log.Debug("closing pss peer", "asym", asymmetric, "key", key) + p.RWPoolMu.Lock() + defer p.RWPoolMu.Unlock() + if asymmetric { + rw := p.pubKeyRWPool[key].(*PssReadWriter) + rw.closed = true + delete(p.pubKeyRWPool, key) + } else { + rw := p.symKeyRWPool[key].(*PssReadWriter) + rw.closed = true + delete(p.symKeyRWPool, key) + } +} + +// Uniform translation of protocol specifiers to topic +func ProtocolTopic(spec *protocols.Spec) Topic { + return BytesToTopic([]byte(fmt.Sprintf("%s:%d", spec.Name, spec.Version))) +} +*/ diff --git a/swarm/swap/swap.go b/swarm/swap/swap.go new file mode 100644 index 0000000000..7b58331cbf --- /dev/null +++ b/swarm/swap/swap.go @@ -0,0 +1,433 @@ +// Copyright 2018 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package swap + +import ( + "context" + "crypto/ecdsa" + "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" + "github.com/ethereum/go-ethereum/p2p/protocols" + "github.com/ethereum/go-ethereum/rpc" + "github.com/ethereum/go-ethereum/swarm/log" + whisper "github.com/ethereum/go-ethereum/whisper/whisperv5" +) + +const ( + defaultMaxMsgSize = 1024 * 1024 + swapProtocolName = "swap" + swapVersion = 1 +) + +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 = 100 // threshold that triggers payment {request} (units) + dropAt = 10000 // threshold that triggers disconnect (units) +) + +const ( + chequebookDeployRetries = 5 + chequebookDeployDelay = 1 * time.Second // delay between retries +) + +// SwAP Swarm Accounting Protocol with +// Swift Automatic Payments +// a peer to peer micropayment system +type Swap struct { + lock sync.Mutex // mutex for balance access + balance int // units of chunk/retrieval request + local *Params // local peer's swap parameters + remote *Profile // remote peer's swap profile +} + +// 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 uint // threshold that triggers payment request + DropAt uint // 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) (swap *Swap, err error) { + + swap = &Swap{ + local: local, + } + + //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: uint(payAt), + DropAt: uint(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), + } +} + +///////////////////////////////////////////////////////////////////// +// SECTION: node.Service interface +///////////////////////////////////////////////////////////////////// + +func (p *Swap) Start(srv *p2p.Server) error { + log.Debug("Started swap") + return nil +} + +func (p *Swap) Stop() error { + log.Info("swap shutting down") + return nil +} + +var swapSpec = &protocols.Spec{ + Name: swapProtocolName, + Version: swapVersion, + MaxMsgSize: defaultMaxMsgSize, + Messages: []interface{}{ + SwapMsg{}, + }, +} + +func (p *Swap) Protocols() []p2p.Protocol { + return []p2p.Protocol{ + { + Name: swapSpec.Name, + Version: swapSpec.Version, + Length: swapSpec.Length(), + Run: p.Run, + }, + } +} + +func (p *Swap) Run(peer *p2p.Peer, rw p2p.MsgReadWriter) error { + pp := protocols.NewPeer(peer, rw, swapSpec) + /* + p.fwdPoolMu.Lock() + p.fwdPool[peer.Info().ID] = pp + p.fwdPoolMu.Unlock() + */ + return pp.Run(p.handleSwapMsg) +} + +func (p *Swap) APIs() []rpc.API { + apis := []rpc.API{ + { + Namespace: "swap", + Version: "1.0", + Service: NewAPI(p), + Public: true, + }, + } + return apis +} + +// Filters incoming messages for processing or forwarding. +// Check if address partially matches +// If yes, it CAN be for us, and we process it +// Only passes error to pss protocol handler if payload is not valid pssmsg +func (p *Swap) handleSwapMsg(ctx context.Context, msg interface{}) error { + return nil +} + +// 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 new file mode 100644 index 0000000000..289ba028be --- /dev/null +++ b/swarm/swap/swap_test.go @@ -0,0 +1,275 @@ +// Copyright 2018 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package swap + +import ( + "flag" + "fmt" + "os" + "path/filepath" + "sync" + "testing" + + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/node" + "github.com/ethereum/go-ethereum/rpc" + colorable "github.com/mattn/go-colorable" +) + +var ( + p2pPort = 30100 + ipcpath = ".swarm.ipc" + datadirPrefix = ".data_" + stackW = &sync.WaitGroup{} + loglevel = flag.Int("loglevel", 2, "verbosity of logs") +) + +func init() { + flag.Parse() + + log.PrintOrigins(true) + 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]) + } + } + stack, err := node.New(cfg) + if err != nil { + return nil, fmt.Errorf("ServiceNode create fail: %v", err) + } + return stack, nil +} + +func TestBasicSwap(t *testing.T) { + + // create the two nodes + stack_one, err := newServiceNode(p2pPort, 0, 0) + if err != nil { + log.Crit("Create servicenode #1 fail", "err", err) + } + stack_two, err := newServiceNode(p2pPort+1, 0, 0) + if err != nil { + log.Crit("Create servicenode #2 fail", "err", err) + } + + instance, err := NewSwap(NewDefaultSwapParams().Params) + if err != nil { + t.Fatal("Couldn't create Swap instance") + } + // wrapper function for servicenode to start the service + swapsvc := func(ctx *node.ServiceContext) (node.Service, error) { + return &API{ + Swap: instance, + }, nil + } + + // register adds the service to the services the servicenode starts when started + err = stack_one.Register(swapsvc) + if err != nil { + log.Crit("Register service in servicenode #1 fail", "err", err) + } + err = stack_two.Register(swapsvc) + if err != nil { + log.Crit("Register service in servicenode #2 fail", "err", err) + } + + // start the nodes + err = stack_one.Start() + if err != nil { + log.Crit("servicenode #1 start failed", "err", err) + } + err = stack_two.Start() + if err != nil { + log.Crit("servicenode #2 start failed", "err", err) + } + + // connect to the servicenode RPCs + rpcclient_one, err := rpc.Dial(filepath.Join(stack_one.DataDir(), ipcpath)) + if err != nil { + log.Crit("connect to servicenode #1 IPC fail", "err", err) + } + defer os.RemoveAll(stack_one.DataDir()) + + rpcclient_two, err := rpc.Dial(filepath.Join(stack_two.DataDir(), ipcpath)) + if err != nil { + log.Crit("connect to servicenode #2 IPC fail", "err", err) + } + defer os.RemoveAll(stack_two.DataDir()) + + // display that the initial pong counts are 0 + var balance int + err = rpcclient_one.Call(&balance, "swap_balance") + if err != nil { + log.Crit("servicenode #1 pongcount RPC failed", "err", err) + } + log.Info("servicenode #1 before ping", "balance-1", balance) + + err = rpcclient_two.Call(&balance, "swap_balance") + if err != nil { + log.Crit("servicenode #2 pongcount RPC failed", "err", err) + } + log.Info("servicenode #2 before ping", "balance-2", balance) + + /* + // get the server instances + srv_one := stack_one.Server() + srv_two := stack_two.Server() + + // subscribe to peerevents + eventOneC := make(chan *p2p.PeerEvent) + sub_one := srv_one.SubscribeEvents(eventOneC) + + eventTwoC := make(chan *p2p.PeerEvent) + sub_two := srv_two.SubscribeEvents(eventTwoC) + + // connect the nodes + p2pnode_two := srv_two.Self() + srv_one.AddPeer(p2pnode_two) + + // fork and do the pinging + stackW.Add(2) + pingmax_one := 4 + pingmax_two := 2 + + go func() { + + // 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) + stackW.Done() + return + } + log.Debug("server #1 connected", "peer", ev.Peer) + + // send the pings + for i := 0; i < pingmax_one; i++ { + err := rpcclient_one.Call(nil, "foo_ping", ev.Peer) + if err != nil { + log.Error("server #1 RPC ping fail", "err", err) + stackW.Done() + break + } + } + + // wait for all msgrecv events + // pings we receive, and pongs we expect from pings we sent + for i := 0; i < pingmax_two+pingmax_one; { + ev := <-eventOneC + log.Warn("msg", "type", ev.Type, "i", i) + if ev.Type == "msgrecv" { + i++ + } + } + + stackW.Done() + }() + + // mirrors the previous go func + go func() { + ev := <-eventTwoC + if ev.Type != "add" { + log.Error("expected peer add", "eventtype", ev.Type) + stackW.Done() + return + } + log.Debug("server #2 connected", "peer", ev.Peer) + for i := 0; i < pingmax_two; i++ { + err := rpcclient_two.Call(nil, "foo_ping", ev.Peer) + if err != nil { + log.Error("server #2 RPC ping fail", "err", err) + stackW.Done() + break + } + } + + for i := 0; i < pingmax_one+pingmax_two; { + ev := <-eventTwoC + if ev.Type == "msgrecv" { + log.Warn("msg", "type", ev.Type, "i", i) + i++ + } + } + + stackW.Done() + }() + + // wait for the two ping pong exchanges to finish + stackW.Wait() + + // tell the API to shut down + // this will disconnect the peers and close the channels connecting API and protocol + err = rpcclient_one.Call(nil, "foo_quit", srv_two.Self().ID) + if err != nil { + log.Error("server #1 RPC quit fail", "err", err) + } + err = rpcclient_two.Call(nil, "foo_quit", srv_one.Self().ID) + if err != nil { + log.Error("server #2 RPC quit fail", "err", err) + } + + // disconnect will generate drop events + for { + ev := <-eventOneC + if ev.Type == "drop" { + break + } + } + for { + ev := <-eventTwoC + if ev.Type == "drop" { + break + } + } + + // proudly inspect the results + err = rpcclient_one.Call(&count, "foo_pongCount") + if err != nil { + log.Crit("servicenode #1 pongcount RPC failed", "err", err) + } + log.Info("servicenode #1 after ping", "pongcount", count) + + err = rpcclient_two.Call(&count, "foo_pongCount") + if err != nil { + log.Crit("servicenode #2 pongcount RPC failed", "err", err) + } + log.Info("servicenode #2 after ping", "pongcount", count) + + // bring down the servicenodes + sub_one.Unsubscribe() + sub_two.Unsubscribe() + stack_one.Stop() + stack_two.Stop() + */ +} diff --git a/swarm/swarm.go b/swarm/swarm.go index a895bdfa55..0cf06f3e48 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -51,6 +51,7 @@ import ( "github.com/ethereum/go-ethereum/swarm/storage" "github.com/ethereum/go-ethereum/swarm/storage/mock" "github.com/ethereum/go-ethereum/swarm/storage/mru" + "github.com/ethereum/go-ethereum/swarm/swap" "github.com/ethereum/go-ethereum/swarm/tracing" ) @@ -78,6 +79,7 @@ type Swarm struct { lstore *storage.LocalStore // local store, needs to store for releasing resources after node stopped sfs *fuse.SwarmFS // need this to cleanup all the active mounts on node exit ps *pss.Pss + swap *swap.Swap tracerClose io.Closer } @@ -206,6 +208,10 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e self.bzz = network.NewBzz(bzzconfig, to, stateStore, stream.Spec, self.streamer.Run) + self.swap, err = swap.NewSwap(config.Swap.Params) + if err != nil { + return nil, err + } // Pss = postal service over swarm (devp2p over bzz) self.ps, err = pss.NewPss(to, config.Pss) if err != nil { @@ -363,6 +369,13 @@ func (self *Swarm) Start(srv *p2p.Server) error { } log.Info(fmt.Sprintf("Swarm network started on bzz address: %x", self.bzz.Hive.Overlay.BaseAddr())) + err = self.swap.Start(srv) + if err != nil { + log.Error("swap failed", "err", err) + return err + } + log.Debug("Swap accounting initialized") + if self.ps != nil { self.ps.Start(srv) log.Info("Pss started") From fb8d62f8dbb6993be557ddaf3b6611de1854941b Mon Sep 17 00:00:00 2001 From: Fabio Barone Date: Thu, 9 Aug 2018 12:34:36 -0500 Subject: [PATCH 2/8] swarm/swap: initial version --- swarm/network/stream/delivery.go | 7 + swarm/network/stream/stream.go | 20 ++- swarm/swap/api.go | 6 +- swarm/swap/protocol.go | 274 +++++++++++++++++-------------- swarm/swap/swap.go | 155 ++++++++--------- swarm/swap/swap_test.go | 9 +- swarm/swarm.go | 24 +-- 7 files changed, 266 insertions(+), 229 deletions(-) diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go index 36040339d3..fde8ed0477 100644 --- a/swarm/network/stream/delivery.go +++ b/swarm/network/stream/delivery.go @@ -19,6 +19,7 @@ package stream import ( "context" "errors" + "math/big" "time" "github.com/ethereum/go-ethereum/common" @@ -29,6 +30,7 @@ import ( "github.com/ethereum/go-ethereum/swarm/network" "github.com/ethereum/go-ethereum/swarm/spancontext" "github.com/ethereum/go-ethereum/swarm/storage" + "github.com/ethereum/go-ethereum/swarm/swap" opentracing "github.com/opentracing/opentracing-go" ) @@ -138,6 +140,11 @@ type RetrieveRequestMsg struct { SkipCheck bool } +//TODO: what is the correct price +func (rrm *RetrieveRequestMsg) GetMsgPrice() (*big.Int, swap.EntryDirection) { + return big.NewInt(int64(4096)), swap.CreditEntry +} + func (d *Delivery) handleRetrieveRequestMsg(ctx context.Context, sp *Peer, req *RetrieveRequestMsg) error { log.Trace("received request", "peer", sp.ID(), "hash", req.Addr) handleRetrieveRequestMsgCount.Inc(1) diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index cd0580a0c0..d88303b419 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -35,6 +35,7 @@ import ( "github.com/ethereum/go-ethereum/swarm/spancontext" "github.com/ethereum/go-ethereum/swarm/state" "github.com/ethereum/go-ethereum/swarm/storage" + "github.com/ethereum/go-ethereum/swarm/swap" opentracing "github.com/opentracing/opentracing-go" ) @@ -50,6 +51,7 @@ const ( // Registry registry for outgoing and incoming streamer constructors type Registry struct { + swap *swap.Swap api *API addr *network.BzzAddr skipCheck bool @@ -101,6 +103,12 @@ func NewRegistry(addr *network.BzzAddr, delivery *Delivery, db *storage.DBAPI, i RegisterSwarmSyncerServer(streamer, db) RegisterSwarmSyncerClient(streamer, db) + var err error + streamer.swap, err = swap.NewSwap(swap.NewDefaultSwapParams().Params) + if err != nil { + log.Error(err.Error()) + } + if options.DoSync { // latestIntC function ensures that // - receiving from the in chan is not blocked by processing inside the for loop @@ -390,7 +398,7 @@ func (r *Registry) Run(p *network.BzzPeer) error { } } - return sp.Run(sp.HandleMsg) + return sp.Run(sp.HandleAccountedMsg) } // updateSyncing subscribes to SYNC streams by iterating over the @@ -467,8 +475,16 @@ func (r *Registry) runProtocol(p *p2p.Peer, rw p2p.MsgReadWriter) error { return r.Run(bzzPeer) } +func (p *Peer) HandleAccountedMsg(ctx context.Context, msg interface{}) error { + err := p.handleMsg(ctx, msg) + if _, ok := msg.(swap.SwapAccountedMsgType); ok && err == nil { + p.streamer.swap.AccountForMsg(ctx, msg, p.ID()) + } + return err +} + // HandleMsg is the message handler that delegates incoming messages -func (p *Peer) HandleMsg(ctx context.Context, msg interface{}) error { +func (p *Peer) handleMsg(ctx context.Context, msg interface{}) error { switch msg := msg.(type) { case *SubscribeMsg: diff --git a/swarm/swap/api.go b/swarm/swap/api.go index 2de11d9efa..632a62433f 100644 --- a/swarm/swap/api.go +++ b/swarm/swap/api.go @@ -31,7 +31,7 @@ type APIMsg struct { // Additional public methods accessible through API for pss type API struct { - *Swap + *SwapProtocol } type SwapMetrics struct { @@ -40,8 +40,8 @@ type SwapMetrics struct { type Cheque struct { } -func NewAPI(swap *Swap) *API { - return &API{Swap: swap} +func NewAPI(swap *SwapProtocol) *API { + return &API{SwapProtocol: swap} } func (swapapi *API) Balance(ctx context.Context) (balance int, err error) { diff --git a/swarm/swap/protocol.go b/swarm/swap/protocol.go index 63acbbe929..1744bc9ada 100644 --- a/swarm/swap/protocol.go +++ b/swarm/swap/protocol.go @@ -14,164 +14,186 @@ // You should have received a copy of the GNU Lesser General Public License // along with the go-ethereum library. If not, see . -// +build !nopssprotocol - package swap import ( + "context" + "fmt" "sync" - "time" + "github.com/ethereum/go-ethereum/metrics" "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/p2p/protocols" - "github.com/ethereum/go-ethereum/rlp" + "github.com/ethereum/go-ethereum/rpc" + "github.com/ethereum/go-ethereum/swarm/log" ) const ( IsActiveProtocol = true ) -// Convenience wrapper for devp2p protocol messages for transport over pss -type ProtocolMsg struct { - Code uint64 - Size uint32 - Payload []byte - ReceivedAt time.Time +type SwapProtocol struct { + peersMu sync.RWMutex + peers map[discover.NodeID]*SwapPeer } -// Creates a ProtocolMsg -func NewProtocolMsg(code uint64, msg interface{}) ([]byte, error) { +// Peer is the Peer extension for the streaming protocol +type SwapPeer struct { + *protocols.Peer + swapProtocol *SwapProtocol +} - rlpdata, err := rlp.EncodeToBytes(msg) - if err != nil { - return nil, err +func NewSwapProtocol() *SwapProtocol { + proto := &SwapProtocol{ + peers: make(map[discover.NodeID]*SwapPeer), } + return proto +} - // TODO verify that nested structs cannot be used in rlp - smsg := &ProtocolMsg{ - Code: code, - Size: uint32(len(rlpdata)), - Payload: rlpdata, +// NewPeer is the constructor for Peer +func NewPeer(peer *protocols.Peer, swap *SwapProtocol) *SwapPeer { + p := &SwapPeer{ + Peer: peer, + swapProtocol: swap, } - - return rlp.EncodeToBytes(smsg) + return p } -// Protocol options to be passed to a new Protocol instance -// -// The parameters specify which encryption schemes to allow -type ProtocolParams struct { +type IssueChequeMsg struct { } -// Convenience object for emulation devp2p over pss -type Protocol struct { - proto *p2p.Protocol - spec *protocols.Spec - RWPoolMu sync.Mutex +type RedeemChequeMsg struct { } -/* -// Activates devp2p emulation over a specific pss topic -// -// One or both encryption schemes must be specified. If -// only one is specified, the protocol will not be valid -// for the other, and will make the message handler -// return errors -func RegisterProtocol(ps *Pss, topic *Topic, spec *protocols.Spec, targetprotocol *p2p.Protocol, options *ProtocolParams) (*Protocol, error) { - if !options.Asymmetric && !options.Symmetric { - return nil, fmt.Errorf("specify at least one of asymmetric or symmetric messaging mode") - } - pp := &Protocol{ - Pss: ps, - proto: targetprotocol, - topic: topic, - spec: spec, - pubKeyRWPool: make(map[string]p2p.MsgReadWriter), - symKeyRWPool: make(map[string]p2p.MsgReadWriter), - Asymmetric: options.Asymmetric, - Symmetric: options.Symmetric, - } - return pp, nil -} +///////////////////////////////////////////////////////////////////// +// SECTION: node.Service interface +///////////////////////////////////////////////////////////////////// -*/ -// Generic handler for incoming messages over devp2p emulation -// -// To be passed to pss.Register() -// -// Will run the protocol on a new incoming peer, provided that -// the encryption key of the message has a match in the internal -// pss keypool -// -// Fails if protocol is not valid for the message encryption scheme, -// if adding a new peer fails, or if the message is not a serialized -// p2p.Msg (which it always will be if it is sent from this object). -func (p *Protocol) Handle(msg []byte, peer *p2p.Peer, asymmetric bool, keyid string) error { +func (p *SwapProtocol) Start(srv *p2p.Server) error { + log.Debug("Started swap") return nil } -/* -// Runs an emulated pss Protocol on the specified peer, -// linked to a specific topic -// `key` and `asymmetric` specifies what encryption key -// to link the peer to. -// The key must exist in the pss store prior to adding the peer. -func (p *Protocol) AddPeer(peer *p2p.Peer, topic Topic, asymmetric bool, key string) (p2p.MsgReadWriter, error) { - rw := &PssReadWriter{ - Pss: p.Pss, - rw: make(chan p2p.Msg), - spec: p.spec, - topic: p.topic, - key: key, - } - if asymmetric { - rw.sendFunc = p.Pss.SendAsym - } else { - rw.sendFunc = p.Pss.SendSym - } - if asymmetric { - p.Pss.pubKeyPoolMu.Lock() - if _, ok := p.Pss.pubKeyPool[key]; !ok { - return nil, fmt.Errorf("asym key does not exist: %s", key) - } - p.Pss.pubKeyPoolMu.Unlock() - p.RWPoolMu.Lock() - p.pubKeyRWPool[key] = rw - p.RWPoolMu.Unlock() - } else { - p.Pss.symKeyPoolMu.Lock() - if _, ok := p.Pss.symKeyPool[key]; !ok { - return nil, fmt.Errorf("symkey does not exist: %s", key) - } - p.Pss.symKeyPoolMu.Unlock() - p.RWPoolMu.Lock() - p.symKeyRWPool[key] = rw - p.RWPoolMu.Unlock() - } - go func() { - err := p.proto.Run(peer, rw) - log.Warn(fmt.Sprintf("pss vprotocol quit on %v topic %v: %v", peer, topic, err)) - }() - return rw, nil +func (p *SwapProtocol) Stop() error { + log.Info("swap shutting down") + return nil } -func (p *Protocol) RemovePeer(asymmetric bool, key string) { - log.Debug("closing pss peer", "asym", asymmetric, "key", key) - p.RWPoolMu.Lock() - defer p.RWPoolMu.Unlock() - if asymmetric { - rw := p.pubKeyRWPool[key].(*PssReadWriter) - rw.closed = true - delete(p.pubKeyRWPool, key) - } else { - rw := p.symKeyRWPool[key].(*PssReadWriter) - rw.closed = true - delete(p.symKeyRWPool, key) +var swapSpec = &protocols.Spec{ + Name: swapProtocolName, + Version: swapVersion, + MaxMsgSize: defaultMaxMsgSize, + Messages: []interface{}{ + SwapMsg{}, + }, +} + +func (p *SwapProtocol) Protocols() []p2p.Protocol { + return []p2p.Protocol{ + { + Name: swapSpec.Name, + Version: swapSpec.Version, + Length: swapSpec.Length(), + Run: p.Run, + }, } } -// Uniform translation of protocol specifiers to topic -func ProtocolTopic(spec *protocols.Spec) Topic { - return BytesToTopic([]byte(fmt.Sprintf("%s:%d", spec.Name, spec.Version))) +func (swap *SwapProtocol) DebitByteCount(peer *SwapPeer, numberOfBytes int) error { + return nil +} + +func (swap *SwapProtocol) Run(peer *p2p.Peer, rw p2p.MsgReadWriter) error { + p := protocols.NewPeer(peer, rw, swapSpec) + sp := NewPeer(p, swap) + swap.setPeer(sp) + defer swap.deletePeer(sp) + defer swap.Close() + return sp.Run(sp.handleSwapMsg) +} + +func (p *SwapProtocol) APIs() []rpc.API { + apis := []rpc.API{ + { + Namespace: "swap", + Version: "1.0", + Service: NewAPI(p), + Public: true, + }, + } + return apis +} + +//-------------------- + +func (swap *SwapProtocol) NodeInfo() interface{} { + return nil +} + +func (swap *SwapProtocol) PeerInfo(id discover.NodeID) interface{} { + return nil +} + +func (swap *SwapProtocol) Close() error { + return nil +} + +func (swap *SwapProtocol) getPeer(peerId discover.NodeID) *SwapPeer { + swap.peersMu.RLock() + defer swap.peersMu.RUnlock() + + return swap.peers[peerId] +} + +func (swap *SwapProtocol) setPeer(peer *SwapPeer) { + swap.peersMu.Lock() + defer swap.peersMu.Unlock() + + swap.peers[peer.ID()] = peer + metrics.GetOrRegisterGauge("registry.peers", nil).Update(int64(len(swap.peers))) +} + +func (swap *SwapProtocol) deletePeer(peer *SwapPeer) { + swap.peersMu.Lock() + defer swap.peersMu.Unlock() + + delete(swap.peers, peer.ID()) + metrics.GetOrRegisterGauge("registry.peers", nil).Update(int64(len(swap.peers))) + swap.peersMu.Unlock() +} + +func (swap *SwapProtocol) peersCount() (c int) { + swap.peersMu.Lock() + c = len(swap.peers) + swap.peersMu.Unlock() + return +} + +func (p *SwapPeer) handleSwapMsg(ctx context.Context, msg interface{}) error { + switch msg := msg.(type) { + + case *IssueChequeMsg: + return p.handleIssueChequeMsg(ctx, msg) + + case *RedeemChequeMsg: + return p.handleRedeemChequeMsg(ctx, msg) + + /* + case *QuitMsg: + return p.handleQuitMsg(msg) + */ + + default: + return fmt.Errorf("unknown message type: %T", msg) + } + return nil +} + +func (sp *SwapPeer) handleIssueChequeMsg(ctx context.Context, msg interface{}) (err error) { + return err +} + +func (sp *SwapPeer) handleRedeemChequeMsg(ctx context.Context, msg interface{}) (err error) { + return err } -*/ diff --git a/swarm/swap/swap.go b/swarm/swap/swap.go index 7b58331cbf..b67c53768e 100644 --- a/swarm/swap/swap.go +++ b/swarm/swap/swap.go @@ -32,9 +32,7 @@ import ( "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" - "github.com/ethereum/go-ethereum/p2p/protocols" - "github.com/ethereum/go-ethereum/rpc" + "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/swarm/log" whisper "github.com/ethereum/go-ethereum/whisper/whisperv5" ) @@ -53,8 +51,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 = 100 // threshold that triggers payment {request} (units) - dropAt = 10000 // threshold that triggers disconnect (units) + payAt = big.NewInt(4096 * 10000) // threshold that triggers payment {request} (bytes) + dropAt = big.NewInt(4096 * 10000) // threshold that triggers disconnect (bytes) ) const ( @@ -66,10 +64,38 @@ const ( // Swift Automatic Payments // a peer to peer micropayment system type Swap struct { - lock sync.Mutex // mutex for balance access - balance int // units of chunk/retrieval request - local *Params // local peer's swap parameters - remote *Profile // remote peer's swap profile + lock sync.RWMutex + peers map[discover.NodeID]*swapPeer + local *Params // local peer's swap parameters +} + +type EntryDirection bool + +const ( + DebitEntry EntryDirection = true + CreditEntry EntryDirection = false +) + +type SwapAccountedMsgType interface { + GetMsgPrice() (*big.Int, EntryDirection) +} + +func (swap *Swap) AccountForMsg(ctx context.Context, msg interface{}, peer discover.NodeID) error { + if accounted, ok := msg.(SwapAccountedMsgType); ok { + if _, exists := swap.peers[peer]; !exists { + swap.lock.Lock() + swap.peers[peer] = &swapPeer{ + peer: peer, + swapAccount: swap, + balance: big.NewInt(0), + } + swap.lock.Unlock() + } + price, direction := accounted.GetMsgPrice() + //TODO: Calculate total price and account + swap.peers[peer].AccountMsgForPeer(price, direction) + } + return nil } // Profile - public swap profile @@ -77,8 +103,8 @@ type Swap struct { type Profile struct { BuyAt *big.Int // accepted max price for chunk SellAt *big.Int // offered sale price for chunk - PayAt uint // threshold that triggers payment request - DropAt uint // threshold that triggers disconnect + PayAt *big.Int // threshold that triggers payment request + DropAt *big.Int // threshold that triggers disconnect } // Strategy encapsulates parameters relating to @@ -130,15 +156,41 @@ type PayProfile struct { lock sync.RWMutex } +type swapPeer struct { + lock sync.RWMutex + peer discover.NodeID + swapAccount *Swap + balance *big.Int +} + +func (sp *swapPeer) AccountMsgForPeer(price *big.Int, direction EntryDirection) { + sp.lock.Lock() + defer sp.lock.Unlock() + //the peer is being credited (in its favor), so its balance increases + if direction == CreditEntry { + sp.balance = sp.balance.Add(sp.balance, price) + //the peer is being debited (in local favor), so its balance decreases + } else if direction == DebitEntry { + sp.balance = sp.balance.Sub(sp.balance, price) + } + if sp.balance.Cmp(payAt) > -1 { + //TODO: Issue Cheque + } + if sp.balance.Cmp(dropAt) < 0 { + //TODO: Drop peer + } + log.Error(fmt.Sprintf("balance for peer %s: %s", sp.peer, sp.balance.String())) +} + // New - swap constructor func NewSwap(local *Params) (swap *Swap, err error) { swap = &Swap{ local: local, + peers: make(map[discover.NodeID]*swapPeer), } //swap.SetParams(local) - return } @@ -150,8 +202,8 @@ func NewDefaultSwapParams() *LocalProfile { Profile: &Profile{ BuyAt: buyAt, SellAt: sellAt, - PayAt: uint(payAt), - DropAt: uint(dropAt), + PayAt: payAt, + DropAt: dropAt, }, Strategy: &Strategy{ AutoCashInterval: autoCashInterval, @@ -179,70 +231,6 @@ func (lp *LocalProfile) Init(contract common.Address, prvkey *ecdsa.PrivateKey) } } -///////////////////////////////////////////////////////////////////// -// SECTION: node.Service interface -///////////////////////////////////////////////////////////////////// - -func (p *Swap) Start(srv *p2p.Server) error { - log.Debug("Started swap") - return nil -} - -func (p *Swap) Stop() error { - log.Info("swap shutting down") - return nil -} - -var swapSpec = &protocols.Spec{ - Name: swapProtocolName, - Version: swapVersion, - MaxMsgSize: defaultMaxMsgSize, - Messages: []interface{}{ - SwapMsg{}, - }, -} - -func (p *Swap) Protocols() []p2p.Protocol { - return []p2p.Protocol{ - { - Name: swapSpec.Name, - Version: swapSpec.Version, - Length: swapSpec.Length(), - Run: p.Run, - }, - } -} - -func (p *Swap) Run(peer *p2p.Peer, rw p2p.MsgReadWriter) error { - pp := protocols.NewPeer(peer, rw, swapSpec) - /* - p.fwdPoolMu.Lock() - p.fwdPool[peer.Info().ID] = pp - p.fwdPoolMu.Unlock() - */ - return pp.Run(p.handleSwapMsg) -} - -func (p *Swap) APIs() []rpc.API { - apis := []rpc.API{ - { - Namespace: "swap", - Version: "1.0", - Service: NewAPI(p), - Public: true, - }, - } - return apis -} - -// Filters incoming messages for processing or forwarding. -// Check if address partially matches -// If yes, it CAN be for us, and we process it -// Only passes error to pss protocol handler if payload is not valid pssmsg -func (p *Swap) handleSwapMsg(ctx context.Context, msg interface{}) error { - return nil -} - // Chequebook get's chequebook from the localProfile func (lp *LocalProfile) Chequebook() *chequebook.Chequebook { defer lp.lock.Unlock() @@ -351,12 +339,13 @@ func (lp *LocalProfile) newChequebookFromContract(path string, backend chequeboo 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() + //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)) @@ -379,11 +368,12 @@ func (swap *Swap) Add(n int) error { // Balance accessor func (swap *Swap) Balance() int { - defer swap.lock.Unlock() - swap.lock.Lock() + //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 @@ -431,3 +421,4 @@ func (swap *Swap) Receive(units int, promise Promise) error { return nil } +*/ diff --git a/swarm/swap/swap_test.go b/swarm/swap/swap_test.go index 289ba028be..2be0c5013f 100644 --- a/swarm/swap/swap_test.go +++ b/swarm/swap/swap_test.go @@ -71,7 +71,7 @@ func newServiceNode(port int, httpport int, wsport int, modules ...string) (*nod return stack, nil } -func TestBasicSwap(t *testing.T) { +func TestSwapProtocol(t *testing.T) { // create the two nodes stack_one, err := newServiceNode(p2pPort, 0, 0) @@ -83,14 +83,11 @@ func TestBasicSwap(t *testing.T) { log.Crit("Create servicenode #2 fail", "err", err) } - instance, err := NewSwap(NewDefaultSwapParams().Params) - if err != nil { - t.Fatal("Couldn't create Swap instance") - } + instance := NewSwapProtocol() // wrapper function for servicenode to start the service swapsvc := func(ctx *node.ServiceContext) (node.Service, error) { return &API{ - Swap: instance, + SwapProtocol: instance, }, nil } diff --git a/swarm/swarm.go b/swarm/swarm.go index 0cf06f3e48..95f657bafb 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -208,10 +208,12 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e self.bzz = network.NewBzz(bzzconfig, to, stateStore, stream.Spec, self.streamer.Run) - self.swap, err = swap.NewSwap(config.Swap.Params) - if err != nil { - return nil, err - } + /* + self.swap, err = swap.NewSwap(config.Swap.Params) + if err != nil { + return nil, err + } + */ // Pss = postal service over swarm (devp2p over bzz) self.ps, err = pss.NewPss(to, config.Pss) if err != nil { @@ -369,12 +371,14 @@ func (self *Swarm) Start(srv *p2p.Server) error { } log.Info(fmt.Sprintf("Swarm network started on bzz address: %x", self.bzz.Hive.Overlay.BaseAddr())) - err = self.swap.Start(srv) - if err != nil { - log.Error("swap failed", "err", err) - return err - } - log.Debug("Swap accounting initialized") + /* + err = self.swap.Start(srv) + if err != nil { + log.Error("swap failed", "err", err) + return err + } + log.Debug("Swap accounting initialized") + */ if self.ps != nil { self.ps.Start(srv) From f8ec67a533132b1e7c8cdf155227c6255cacd162 Mon Sep 17 00:00:00 2001 From: Fabio Barone Date: Tue, 14 Aug 2018 19:43:25 -0500 Subject: [PATCH 3/8] swarm/swap: attach swap to swarm, not registry --- swarm/network/stream/common_test.go | 2 +- swarm/network/stream/delivery_test.go | 4 +- swarm/network/stream/intervals_test.go | 2 +- .../network/stream/snapshot_retrieval_test.go | 4 +- swarm/network/stream/snapshot_sync_test.go | 4 +- swarm/network/stream/stream.go | 8 +- swarm/network/stream/syncer_test.go | 2 +- swarm/network_test.go | 115 +++++++++++++++++- swarm/swap/api.go | 7 +- swarm/swap/swap.go | 32 +++-- swarm/swap/swap_test.go | 18 +++ swarm/swarm.go | 13 +- 12 files changed, 176 insertions(+), 35 deletions(-) diff --git a/swarm/network/stream/common_test.go b/swarm/network/stream/common_test.go index 491dc9fd58..0559a942ea 100644 --- a/swarm/network/stream/common_test.go +++ b/swarm/network/stream/common_test.go @@ -109,7 +109,7 @@ func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Registry, *stora db := storage.NewDBAPI(localStore) delivery := NewDelivery(to, db) - streamer := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), nil) + streamer := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), nil, nil) teardown := func() { streamer.Close() removeDataDir() diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go index 972cc859a3..7b2c1f9dcf 100644 --- a/swarm/network/stream/delivery_test.go +++ b/swarm/network/stream/delivery_test.go @@ -328,7 +328,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck kad := network.NewKademlia(addr.Over(), network.NewKadParams()) delivery := NewDelivery(kad, db) - r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), &RegistryOptions{ + r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), nil, &RegistryOptions{ SkipCheck: skipCheck, }) bucket.Store(bucketKeyRegistry, r) @@ -515,7 +515,7 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip kad := network.NewKademlia(addr.Over(), network.NewKadParams()) delivery := NewDelivery(kad, db) - r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), &RegistryOptions{ + r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), nil, &RegistryOptions{ SkipCheck: skipCheck, DoSync: true, SyncUpdateDelay: 0, diff --git a/swarm/network/stream/intervals_test.go b/swarm/network/stream/intervals_test.go index f4294134b9..02626ab65b 100644 --- a/swarm/network/stream/intervals_test.go +++ b/swarm/network/stream/intervals_test.go @@ -74,7 +74,7 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) { kad := network.NewKademlia(addr.Over(), network.NewKadParams()) delivery := NewDelivery(kad, db) - r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), &RegistryOptions{ + r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), nil, &RegistryOptions{ SkipCheck: skipCheck, }) bucket.Store(bucketKeyRegistry, r) diff --git a/swarm/network/stream/snapshot_retrieval_test.go b/swarm/network/stream/snapshot_retrieval_test.go index 4ff947b215..32c1f63c23 100644 --- a/swarm/network/stream/snapshot_retrieval_test.go +++ b/swarm/network/stream/snapshot_retrieval_test.go @@ -133,7 +133,7 @@ func runFileRetrievalTest(nodeCount int) error { kad := network.NewKademlia(addr.Over(), network.NewKadParams()) delivery := NewDelivery(kad, db) - r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), &RegistryOptions{ + r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), nil, &RegistryOptions{ DoSync: true, SyncUpdateDelay: 3 * time.Second, }) @@ -276,7 +276,7 @@ func runRetrievalTest(chunkCount int, nodeCount int) error { kad := network.NewKademlia(addr.Over(), network.NewKadParams()) delivery := NewDelivery(kad, db) - r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), &RegistryOptions{ + r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), nil, &RegistryOptions{ DoSync: true, SyncUpdateDelay: 0, }) diff --git a/swarm/network/stream/snapshot_sync_test.go b/swarm/network/stream/snapshot_sync_test.go index 6acab50af4..1b264011e7 100644 --- a/swarm/network/stream/snapshot_sync_test.go +++ b/swarm/network/stream/snapshot_sync_test.go @@ -139,7 +139,7 @@ func testSyncingViaGlobalSync(t *testing.T, chunkCount int, nodeCount int) { kad := network.NewKademlia(addr.Over(), network.NewKadParams()) delivery := NewDelivery(kad, db) - r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), &RegistryOptions{ + r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), nil, &RegistryOptions{ DoSync: true, SyncUpdateDelay: 3 * time.Second, }) @@ -297,7 +297,7 @@ func testSyncingViaDirectSubscribe(chunkCount int, nodeCount int) error { kad := network.NewKademlia(addr.Over(), network.NewKadParams()) delivery := NewDelivery(kad, db) - r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), nil) + r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), nil, nil) bucket.Store(bucketKeyRegistry, r) fileStore := storage.NewFileStore(storage.NewNetStore(localStore, nil), storage.NewFileStoreParams()) diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index d88303b419..70cb238974 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -75,7 +75,7 @@ type RegistryOptions struct { } // NewRegistry is Streamer constructor -func NewRegistry(addr *network.BzzAddr, delivery *Delivery, db *storage.DBAPI, intervalsStore state.Store, options *RegistryOptions) *Registry { +func NewRegistry(addr *network.BzzAddr, delivery *Delivery, db *storage.DBAPI, intervalsStore state.Store, swap *swap.Swap, options *RegistryOptions) *Registry { if options == nil { options = &RegistryOptions{} } @@ -103,11 +103,7 @@ func NewRegistry(addr *network.BzzAddr, delivery *Delivery, db *storage.DBAPI, i RegisterSwarmSyncerServer(streamer, db) RegisterSwarmSyncerClient(streamer, db) - var err error - streamer.swap, err = swap.NewSwap(swap.NewDefaultSwapParams().Params) - if err != nil { - log.Error(err.Error()) - } + streamer.swap = swap if options.DoSync { // latestIntC function ensures that diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go index f72aa34441..e9ee4df458 100644 --- a/swarm/network/stream/syncer_test.go +++ b/swarm/network/stream/syncer_test.go @@ -108,7 +108,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck delivery := NewDelivery(kad, db) bucket.Store(bucketKeyDelivery, delivery) - r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), &RegistryOptions{ + r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), nil, &RegistryOptions{ SkipCheck: skipCheck, }) diff --git a/swarm/network_test.go b/swarm/network_test.go index 176c635d82..6e3c806773 100644 --- a/swarm/network_test.go +++ b/swarm/network_test.go @@ -41,9 +41,11 @@ import ( ) var ( - loglevel = flag.Int("loglevel", 2, "verbosity of logs") - longrunning = flag.Bool("longrunning", false, "do run long-running tests") - waitKademlia = flag.Bool("waitkademlia", false, "wait for healthy kademlia before checking files availability") + loglevel = flag.Int("loglevel", 2, "verbosity of logs") + longrunning = flag.Bool("longrunning", false, "do run long-running tests") + waitKademlia = flag.Bool("waitkademlia", false, "wait for healthy kademlia before checking files availability") + bucketKeySwap = simulation.BucketKey("swap") + bucketKeySwarm = simulation.BucketKey("swarm") ) func init() { @@ -373,6 +375,113 @@ func testSwarmNetwork(t *testing.T, o *testSwarmNetworkOptions, steps ...testSwa } } +func TestSwapNetworkSymmetricFileUpload(t *testing.T) { + nodeCount := 16 + + sim := simulation.New(map[string]simulation.ServiceFunc{ + "swarm": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) { + config := api.NewConfig() + + dir, err := ioutil.TempDir("", "swap-network-test-node") + if err != nil { + return nil, nil, err + } + cleanup = func() { + err := os.RemoveAll(dir) + if err != nil { + log.Error("cleaning up swarm temp dir", "err", err) + } + } + + config.Path = dir + + privkey, err := crypto.GenerateKey() + if err != nil { + return nil, cleanup, err + } + + config.Init(privkey) + + swarm, err := NewSwarm(config, nil) + if err != nil { + return nil, cleanup, err + } + bucket.Store(bucketKeySwarm, swarm) + log.Info("new swarm", "bzzKey", config.BzzKey, "baseAddr", fmt.Sprintf("%x", swarm.bzz.BaseAddr())) + return swarm, cleanup, nil + }, + }) + defer sim.Close() + + ctx := context.Background() + files := make([]file, 0) + + var checkStatusM sync.Map + var nodeStatusM sync.Map + var totalFoundCount uint64 + + _, err := sim.AddNodesAndConnectChain(nodeCount) + if err != nil { + t.Fatal(err) + } + + result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error { + nodeIDs := sim.UpNodeIDs() + shuffle(len(nodeIDs), func(i, j int) { + nodeIDs[i], nodeIDs[j] = nodeIDs[j], nodeIDs[i] + }) + for _, id := range nodeIDs { + key, data, err := uploadFile(sim.Service("swarm", id).(*Swarm)) + if err != nil { + return err + } + log.Trace("file uploaded", "node", id, "key", key.String()) + files = append(files, file{ + addr: key, + data: data, + nodeID: id, + }) + } + + if _, err := sim.WaitTillHealthy(ctx, 2); err != nil { + return err + } + + // File retrieval check is repeated until all uploaded files are retrieved from all nodes + // or until the timeout is reached. + for { + if retrieve(sim, files, &checkStatusM, &nodeStatusM, &totalFoundCount) == 0 { + return nil + } + } + }) + + for _, node := range sim.NodeIDs() { + item, ok := sim.NodeItem(node, bucketKeySwarm) + if !ok { + log.Error("No swarm") + return + } + swarm := item.(*Swarm) + + for _, n := range sim.NodeIDs() { + if node == n { + continue + } + if swarm.swap.GetPeerBalance(n) != nil { + log.Error(fmt.Sprintf("Balance of node %s to node %s: %s", node.TerminalString(), n.TerminalString(), swarm.swap.GetPeerBalance(n).String())) + } else { + log.Error(fmt.Sprintf("Node %s has no balance with node %s", node.TerminalString(), n.TerminalString())) + } + } + } + + if result.Error != nil { + t.Fatal(result.Error) + } + log.Debug("test terminated") +} + // uploadFile, uploads a short file to the swarm instance // using the api.Put method. func uploadFile(swarm *Swarm) (storage.Address, string, error) { diff --git a/swarm/swap/api.go b/swarm/swap/api.go index 632a62433f..c731cb972b 100644 --- a/swarm/swap/api.go +++ b/swarm/swap/api.go @@ -18,9 +18,11 @@ package swap import ( "context" + "math/big" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/hexutil" + "github.com/ethereum/go-ethereum/p2p/discover" ) // Wrapper for receiving pss messages when using the pss API @@ -32,6 +34,7 @@ type APIMsg struct { // Additional public methods accessible through API for pss type API struct { *SwapProtocol + *Swap } type SwapMetrics struct { @@ -44,8 +47,8 @@ func NewAPI(swap *SwapProtocol) *API { return &API{SwapProtocol: swap} } -func (swapapi *API) Balance(ctx context.Context) (balance int, err error) { - balance = 0 +func (swapapi *API) Balance(ctx context.Context, peer discover.NodeID) (balance *big.Int, err error) { + balance = big.NewInt(0) err = nil return } diff --git a/swarm/swap/swap.go b/swarm/swap/swap.go index b67c53768e..62b8b2bdd0 100644 --- a/swarm/swap/swap.go +++ b/swarm/swap/swap.go @@ -34,6 +34,7 @@ import ( "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/swarm/log" + "github.com/ethereum/go-ethereum/swarm/state" whisper "github.com/ethereum/go-ethereum/whisper/whisperv5" ) @@ -64,9 +65,10 @@ const ( // Swift Automatic Payments // a peer to peer micropayment system type Swap struct { - lock sync.RWMutex - peers map[discover.NodeID]*swapPeer - local *Params // local peer's swap parameters + stateStore state.Store + lock sync.RWMutex + peers map[discover.NodeID]*swapPeer + local *Params // local peer's swap parameters } type EntryDirection bool @@ -83,11 +85,14 @@ type SwapAccountedMsgType interface { func (swap *Swap) AccountForMsg(ctx context.Context, msg interface{}, peer discover.NodeID) error { if accounted, ok := msg.(SwapAccountedMsgType); ok { if _, exists := swap.peers[peer]; !exists { + balance := big.NewInt(0) + swap.stateStore.Get(peer.String()[:24]+"-swap", &balance) swap.lock.Lock() swap.peers[peer] = &swapPeer{ peer: peer, swapAccount: swap, - balance: big.NewInt(0), + balance: balance, + storeID: peer.String()[:24] + "-swap", } swap.lock.Unlock() } @@ -98,6 +103,13 @@ func (swap *Swap) AccountForMsg(ctx context.Context, msg interface{}, peer disco return nil } +func (swap *Swap) GetPeerBalance(peer discover.NodeID) *big.Int { + if p, ok := swap.peers[peer]; ok { + return p.balance + } + return nil +} + // Profile - public swap profile // public parameters for SWAP, serializable config struct passed in handshake type Profile struct { @@ -161,6 +173,7 @@ type swapPeer struct { peer discover.NodeID swapAccount *Swap balance *big.Int + storeID string } func (sp *swapPeer) AccountMsgForPeer(price *big.Int, direction EntryDirection) { @@ -173,21 +186,24 @@ func (sp *swapPeer) AccountMsgForPeer(price *big.Int, direction EntryDirection) } else if direction == DebitEntry { sp.balance = sp.balance.Sub(sp.balance, price) } + //TODO: save to store here? init store? + sp.swapAccount.stateStore.Put(sp.storeID, sp.balance) if sp.balance.Cmp(payAt) > -1 { //TODO: Issue Cheque } if sp.balance.Cmp(dropAt) < 0 { //TODO: Drop peer } - log.Error(fmt.Sprintf("balance for peer %s: %s", sp.peer, sp.balance.String())) + log.Debug(fmt.Sprintf("balance for peer %s: %s", sp.peer, sp.balance.String())) } // New - swap constructor -func NewSwap(local *Params) (swap *Swap, err error) { +func NewSwap(local *Params, stateStore state.Store) (swap *Swap, err error) { swap = &Swap{ - local: local, - peers: make(map[discover.NodeID]*swapPeer), + local: local, + stateStore: stateStore, + peers: make(map[discover.NodeID]*swapPeer), } //swap.SetParams(local) diff --git a/swarm/swap/swap_test.go b/swarm/swap/swap_test.go index 2be0c5013f..060161b897 100644 --- a/swarm/swap/swap_test.go +++ b/swarm/swap/swap_test.go @@ -17,16 +17,28 @@ package swap import ( + "context" + "crypto/rand" "flag" "fmt" + "io/ioutil" "os" "path/filepath" "sync" + "sync/atomic" "testing" + "time" + "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/node" + "github.com/ethereum/go-ethereum/p2p/discover" + "github.com/ethereum/go-ethereum/p2p/simulations/adapters" "github.com/ethereum/go-ethereum/rpc" + "github.com/ethereum/go-ethereum/swarm" + "github.com/ethereum/go-ethereum/swarm/api" + "github.com/ethereum/go-ethereum/swarm/network/simulation" + "github.com/ethereum/go-ethereum/swarm/storage" colorable "github.com/mattn/go-colorable" ) @@ -91,6 +103,12 @@ 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 { diff --git a/swarm/swarm.go b/swarm/swarm.go index 95f657bafb..30db6e1528 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -180,7 +180,12 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e ) delivery := stream.NewDelivery(to, db) - self.streamer = stream.NewRegistry(addr, delivery, db, stateStore, &stream.RegistryOptions{ + self.swap, err = swap.NewSwap(swap.NewDefaultSwapParams().Params, stateStore) + if err != nil { + return nil, err + } + + self.streamer = stream.NewRegistry(addr, delivery, db, stateStore, self.swap, &stream.RegistryOptions{ SkipCheck: config.DeliverySkipCheck, DoSync: config.SyncEnabled, DoRetrieve: true, @@ -208,12 +213,6 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e self.bzz = network.NewBzz(bzzconfig, to, stateStore, stream.Spec, self.streamer.Run) - /* - self.swap, err = swap.NewSwap(config.Swap.Params) - if err != nil { - return nil, err - } - */ // Pss = postal service over swarm (devp2p over bzz) self.ps, err = pss.NewPss(to, config.Pss) if err != nil { From acaeb2d5ea7965fc6dcb0a0e5b04d106dacfd32e Mon Sep 17 00:00:00 2001 From: Fabio Barone Date: Wed, 15 Aug 2018 13:57:03 -0500 Subject: [PATCH 4/8] swarm/swap: swap implemented as protocols/protocol.Peer extension --- swarm/network/stream/delivery.go | 5 +- swarm/network/stream/peer.go | 6 +- swarm/network/stream/stream.go | 12 +-- swarm/swap/protocol.go | 54 +++++++------- swarm/swap/swap.go | 124 +++++++++++++++++++------------ 5 files changed, 113 insertions(+), 88 deletions(-) diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go index fde8ed0477..4ed7a25c69 100644 --- a/swarm/network/stream/delivery.go +++ b/swarm/network/stream/delivery.go @@ -30,7 +30,6 @@ import ( "github.com/ethereum/go-ethereum/swarm/network" "github.com/ethereum/go-ethereum/swarm/spancontext" "github.com/ethereum/go-ethereum/swarm/storage" - "github.com/ethereum/go-ethereum/swarm/swap" opentracing "github.com/opentracing/opentracing-go" ) @@ -141,8 +140,8 @@ type RetrieveRequestMsg struct { } //TODO: what is the correct price -func (rrm *RetrieveRequestMsg) GetMsgPrice() (*big.Int, swap.EntryDirection) { - return big.NewInt(int64(4096)), swap.CreditEntry +func (rrm *RetrieveRequestMsg) GetMsgPrice() *big.Int { + return big.NewInt(int64(4096)) } func (d *Delivery) handleRetrieveRequestMsg(ctx context.Context, sp *Peer, req *RetrieveRequestMsg) error { diff --git a/swarm/network/stream/peer.go b/swarm/network/stream/peer.go index 80b9ab711a..ce861b67d4 100644 --- a/swarm/network/stream/peer.go +++ b/swarm/network/stream/peer.go @@ -30,6 +30,7 @@ import ( "github.com/ethereum/go-ethereum/swarm/spancontext" "github.com/ethereum/go-ethereum/swarm/state" "github.com/ethereum/go-ethereum/swarm/storage" + "github.com/ethereum/go-ethereum/swarm/swap" opentracing "github.com/opentracing/opentracing-go" ) @@ -50,7 +51,8 @@ func (e *notFoundError) Error() string { // Peer is the Peer extension for the streaming protocol type Peer struct { - *protocols.Peer + //*protocols.Peer + *swap.SwapPeer streamer *Registry pq *pq.PriorityQueue serverMu sync.RWMutex @@ -72,7 +74,7 @@ type WrappedPriorityMsg struct { // NewPeer is the constructor for Peer func NewPeer(peer *protocols.Peer, streamer *Registry) *Peer { p := &Peer{ - Peer: peer, + SwapPeer: swap.NewSwapPeer(peer, streamer.swap), pq: pq.New(int(PriorityQueue), PriorityQueueCap), streamer: streamer, servers: make(map[Stream]*server), diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index 70cb238974..9d559eeefd 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -394,7 +394,7 @@ func (r *Registry) Run(p *network.BzzPeer) error { } } - return sp.Run(sp.HandleAccountedMsg) + return sp.RunAccountedProtocol(sp.HandleMsg) } // updateSyncing subscribes to SYNC streams by iterating over the @@ -471,16 +471,8 @@ func (r *Registry) runProtocol(p *p2p.Peer, rw p2p.MsgReadWriter) error { return r.Run(bzzPeer) } -func (p *Peer) HandleAccountedMsg(ctx context.Context, msg interface{}) error { - err := p.handleMsg(ctx, msg) - if _, ok := msg.(swap.SwapAccountedMsgType); ok && err == nil { - p.streamer.swap.AccountForMsg(ctx, msg, p.ID()) - } - return err -} - // HandleMsg is the message handler that delegates incoming messages -func (p *Peer) handleMsg(ctx context.Context, msg interface{}) error { +func (p *Peer) HandleMsg(ctx context.Context, msg interface{}) error { switch msg := msg.(type) { case *SubscribeMsg: diff --git a/swarm/swap/protocol.go b/swarm/swap/protocol.go index 1744bc9ada..b0a27983ec 100644 --- a/swarm/swap/protocol.go +++ b/swarm/swap/protocol.go @@ -35,25 +35,25 @@ const ( type SwapProtocol struct { peersMu sync.RWMutex - peers map[discover.NodeID]*SwapPeer + peers map[discover.NodeID]*SwapProtocolPeer } // Peer is the Peer extension for the streaming protocol -type SwapPeer struct { +type SwapProtocolPeer struct { *protocols.Peer swapProtocol *SwapProtocol } func NewSwapProtocol() *SwapProtocol { proto := &SwapProtocol{ - peers: make(map[discover.NodeID]*SwapPeer), + peers: make(map[discover.NodeID]*SwapProtocolPeer), } return proto } // NewPeer is the constructor for Peer -func NewPeer(peer *protocols.Peer, swap *SwapProtocol) *SwapPeer { - p := &SwapPeer{ +func NewPeer(peer *protocols.Peer, swap *SwapProtocol) *SwapProtocolPeer { + p := &SwapProtocolPeer{ Peer: peer, swapProtocol: swap, } @@ -100,19 +100,6 @@ func (p *SwapProtocol) Protocols() []p2p.Protocol { } } -func (swap *SwapProtocol) DebitByteCount(peer *SwapPeer, numberOfBytes int) error { - return nil -} - -func (swap *SwapProtocol) Run(peer *p2p.Peer, rw p2p.MsgReadWriter) error { - p := protocols.NewPeer(peer, rw, swapSpec) - sp := NewPeer(p, swap) - swap.setPeer(sp) - defer swap.deletePeer(sp) - defer swap.Close() - return sp.Run(sp.handleSwapMsg) -} - func (p *SwapProtocol) APIs() []rpc.API { apis := []rpc.API{ { @@ -125,7 +112,17 @@ func (p *SwapProtocol) APIs() []rpc.API { return apis } -//-------------------- +///////////////////////////////////////////////////////////////////// +// SECTION: p2p.protocol interface +///////////////////////////////////////////////////////////////////// +func (swap *SwapProtocol) Run(peer *p2p.Peer, rw p2p.MsgReadWriter) error { + p := protocols.NewPeer(peer, rw, swapSpec) + sp := NewPeer(p, swap) + swap.setPeer(sp) + defer swap.deletePeer(sp) + defer swap.Close() + return sp.Run(sp.handleSwapMsg) +} func (swap *SwapProtocol) NodeInfo() interface{} { return nil @@ -135,18 +132,23 @@ func (swap *SwapProtocol) PeerInfo(id discover.NodeID) interface{} { return nil } -func (swap *SwapProtocol) Close() error { +//------------------------------------------------------------------------------------------ + +func (swap *SwapProtocol) DebitByteCount(peer *SwapProtocolPeer, numberOfBytes int) error { return nil } -func (swap *SwapProtocol) getPeer(peerId discover.NodeID) *SwapPeer { +func (swap *SwapProtocol) Close() { +} + +func (swap *SwapProtocol) getPeer(peerId discover.NodeID) *SwapProtocolPeer { swap.peersMu.RLock() defer swap.peersMu.RUnlock() return swap.peers[peerId] } -func (swap *SwapProtocol) setPeer(peer *SwapPeer) { +func (swap *SwapProtocol) setPeer(peer *SwapProtocolPeer) { swap.peersMu.Lock() defer swap.peersMu.Unlock() @@ -154,7 +156,7 @@ func (swap *SwapProtocol) setPeer(peer *SwapPeer) { metrics.GetOrRegisterGauge("registry.peers", nil).Update(int64(len(swap.peers))) } -func (swap *SwapProtocol) deletePeer(peer *SwapPeer) { +func (swap *SwapProtocol) deletePeer(peer *SwapProtocolPeer) { swap.peersMu.Lock() defer swap.peersMu.Unlock() @@ -170,7 +172,7 @@ func (swap *SwapProtocol) peersCount() (c int) { return } -func (p *SwapPeer) handleSwapMsg(ctx context.Context, msg interface{}) error { +func (p *SwapProtocolPeer) handleSwapMsg(ctx context.Context, msg interface{}) error { switch msg := msg.(type) { case *IssueChequeMsg: @@ -190,10 +192,10 @@ func (p *SwapPeer) handleSwapMsg(ctx context.Context, msg interface{}) error { return nil } -func (sp *SwapPeer) handleIssueChequeMsg(ctx context.Context, msg interface{}) (err error) { +func (sp *SwapProtocolPeer) handleIssueChequeMsg(ctx context.Context, msg interface{}) (err error) { return err } -func (sp *SwapPeer) handleRedeemChequeMsg(ctx context.Context, msg interface{}) (err error) { +func (sp *SwapProtocolPeer) handleRedeemChequeMsg(ctx context.Context, msg interface{}) (err error) { return err } diff --git a/swarm/swap/swap.go b/swarm/swap/swap.go index 62b8b2bdd0..36b2948677 100644 --- a/swarm/swap/swap.go +++ b/swarm/swap/swap.go @@ -33,6 +33,7 @@ import ( "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" @@ -67,10 +68,19 @@ const ( type Swap struct { stateStore state.Store lock sync.RWMutex - peers map[discover.NodeID]*swapPeer + peers map[discover.NodeID]*SwapPeer local *Params // local peer's swap parameters } +type SwapPeer struct { + *protocols.Peer + lock sync.RWMutex + swapAccount *Swap + handlerFunc func(context.Context, interface{}) error + balance *big.Int + storeID string +} + type EntryDirection bool const ( @@ -79,26 +89,20 @@ const ( ) type SwapAccountedMsgType interface { - GetMsgPrice() (*big.Int, EntryDirection) + GetMsgPrice() *big.Int } -func (swap *Swap) AccountForMsg(ctx context.Context, msg interface{}, peer discover.NodeID) error { +func (sp *SwapPeer) RunAccountedProtocol(protocolHandler func(ctx context.Context, msg interface{}) error) error { + sp.handlerFunc = protocolHandler + + return sp.Run(sp.handleAccountedMsg) +} + +func (sp *SwapPeer) doAccountMsg(ctx context.Context, msg interface{}, direction EntryDirection) error { if accounted, ok := msg.(SwapAccountedMsgType); ok { - if _, exists := swap.peers[peer]; !exists { - balance := big.NewInt(0) - swap.stateStore.Get(peer.String()[:24]+"-swap", &balance) - swap.lock.Lock() - swap.peers[peer] = &swapPeer{ - peer: peer, - swapAccount: swap, - balance: balance, - storeID: peer.String()[:24] + "-swap", - } - swap.lock.Unlock() - } - price, direction := accounted.GetMsgPrice() + price := accounted.GetMsgPrice() //TODO: Calculate total price and account - swap.peers[peer].AccountMsgForPeer(price, direction) + sp.AccountMsgForPeer(price, direction) } return nil } @@ -110,6 +114,61 @@ func (swap *Swap) GetPeerBalance(peer discover.NodeID) *big.Int { return nil } +func (sp *SwapPeer) handleAccountedMsg(ctx context.Context, msg interface{}) error { + err := sp.handlerFunc(ctx, msg) + if _, ok := msg.(SwapAccountedMsgType); ok && err == nil { + sp.doAccountMsg(ctx, msg, CreditEntry) + } + return err +} + +func (sp *SwapPeer) Send(ctx context.Context, msg interface{}) error { + err := sp.Peer.Send(ctx, msg) + if _, ok := msg.(SwapAccountedMsgType); ok && err == nil { + sp.doAccountMsg(ctx, msg, DebitEntry) + } + return err +} + +//The balance is accounted from the point of view of the local node +//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(price *big.Int, direction EntryDirection) { + sp.lock.Lock() + defer sp.lock.Unlock() + //local node is being credited (in its favor), so its balance increases + if direction == CreditEntry { + sp.balance = sp.balance.Add(sp.balance, price) + //local node is being debited (in favor of remote peer), so its balance decreases + } else if direction == DebitEntry { + sp.balance = sp.balance.Sub(sp.balance, price) + } + //TODO: save to store here? init store? + sp.swapAccount.stateStore.Put(sp.storeID, sp.balance) + if sp.balance.Cmp(payAt) > -1 { + //TODO: Issue Cheque + } + if sp.balance.Cmp(dropAt) < 0 { + //TODO: Drop peer + } + log.Debug(fmt.Sprintf("balance for peer %s: %s", sp.ID(), sp.balance.String())) +} + +func NewSwapPeer(peer *protocols.Peer, swap *Swap) *SwapPeer { + balance := big.NewInt(0) + swap.stateStore.Get(peer.String()[:24]+"-swap", &balance) + sp := &SwapPeer{ + Peer: peer, + swapAccount: swap, + balance: balance, + storeID: peer.String()[:24] + "-swap", + } + swap.lock.Lock() + defer swap.lock.Unlock() + swap.peers[peer.ID()] = sp + return sp +} + // Profile - public swap profile // public parameters for SWAP, serializable config struct passed in handshake type Profile struct { @@ -168,42 +227,13 @@ type PayProfile struct { lock sync.RWMutex } -type swapPeer struct { - lock sync.RWMutex - peer discover.NodeID - swapAccount *Swap - balance *big.Int - storeID string -} - -func (sp *swapPeer) AccountMsgForPeer(price *big.Int, direction EntryDirection) { - sp.lock.Lock() - defer sp.lock.Unlock() - //the peer is being credited (in its favor), so its balance increases - if direction == CreditEntry { - sp.balance = sp.balance.Add(sp.balance, price) - //the peer is being debited (in local favor), so its balance decreases - } else if direction == DebitEntry { - sp.balance = sp.balance.Sub(sp.balance, price) - } - //TODO: save to store here? init store? - sp.swapAccount.stateStore.Put(sp.storeID, sp.balance) - if sp.balance.Cmp(payAt) > -1 { - //TODO: Issue Cheque - } - if sp.balance.Cmp(dropAt) < 0 { - //TODO: Drop peer - } - log.Debug(fmt.Sprintf("balance for peer %s: %s", sp.peer, sp.balance.String())) -} - // New - swap constructor func NewSwap(local *Params, stateStore state.Store) (swap *Swap, err error) { swap = &Swap{ local: local, stateStore: stateStore, - peers: make(map[discover.NodeID]*swapPeer), + peers: make(map[discover.NodeID]*SwapPeer), } //swap.SetParams(local) From 18c8bab5db903f2585af2760cc1577c5b35851c2 Mon Sep 17 00:00:00 2001 From: Fabio Barone Date: Thu, 16 Aug 2018 17:30:38 -0500 Subject: [PATCH 5/8] swarm/swap: insufficient funds checks and first working test --- swarm/network_test.go | 38 +++++++++- swarm/swap/swap.go | 157 +++++++++++++++++++++++++++++++++--------- 2 files changed, 158 insertions(+), 37 deletions(-) diff --git a/swarm/network_test.go b/swarm/network_test.go index 6e3c806773..1da677584e 100644 --- a/swarm/network_test.go +++ b/swarm/network_test.go @@ -21,6 +21,7 @@ import ( "flag" "fmt" "io/ioutil" + "math/big" "math/rand" "os" "sync" @@ -44,6 +45,7 @@ var ( loglevel = flag.Int("loglevel", 2, "verbosity of logs") longrunning = flag.Bool("longrunning", false, "do run long-running tests") waitKademlia = flag.Bool("waitkademlia", false, "wait for healthy kademlia before checking files availability") + printStats = flag.Bool("printstats", false, "print accounting stats to STDOUT") bucketKeySwap = simulation.BucketKey("swap") bucketKeySwarm = simulation.BucketKey("swarm") ) @@ -456,6 +458,8 @@ func TestSwapNetworkSymmetricFileUpload(t *testing.T) { } }) + balancesMap := make(map[discover.NodeID]map[discover.NodeID]*big.Int) + for _, node := range sim.NodeIDs() { item, ok := sim.NodeItem(node, bucketKeySwarm) if !ok { @@ -464,14 +468,42 @@ func TestSwapNetworkSymmetricFileUpload(t *testing.T) { } swarm := item.(*Swarm) + subBalances := make(map[discover.NodeID]*big.Int) + for _, n := range sim.NodeIDs() { if node == n { continue } - if swarm.swap.GetPeerBalance(n) != nil { - log.Error(fmt.Sprintf("Balance of node %s to node %s: %s", node.TerminalString(), n.TerminalString(), swarm.swap.GetPeerBalance(n).String())) + balance := swarm.swap.GetPeerBalance(n) + if balance != nil { + subBalances[n] = balance + log.Debug(fmt.Sprintf("Balance of node %s to node %s: %s", node.TerminalString(), n.TerminalString(), swarm.swap.GetPeerBalance(n).String())) } else { - log.Error(fmt.Sprintf("Node %s has no balance with node %s", node.TerminalString(), n.TerminalString())) + log.Debug(fmt.Sprintf("Node %s has no balance with node %s", node.TerminalString(), n.TerminalString())) + } + } + balancesMap[node] = subBalances + } + + if *printStats { + for k, v := range balancesMap { + fmt.Println(fmt.Sprintf("node %s balances:", k.TerminalString())) + for kk, vv := range v { + fmt.Println(fmt.Sprintf(".........with node %s: balance %s", kk.TerminalString(), vv.String())) + } + } + } + + for k, mapForK := range balancesMap { + for n, balanceKwithN := range mapForK { + for subK, mapForSubK := range balancesMap { + if n == subK { + log.Trace(fmt.Sprintf("balance of %s with %s: %s", k.TerminalString(), n.TerminalString(), balanceKwithN)) + log.Trace(fmt.Sprintf("balance of %s with %s: %s", n.TerminalString(), k.TerminalString(), mapForSubK[k])) + if balanceKwithN.CmpAbs(mapForSubK[k]) != 0 && balanceKwithN.Cmp(big.NewInt(0)) != 0 { + log.Error("Expected balances to be |abs| = 0 AND balance1 != 0, but they are not") + } + } } } } diff --git a/swarm/swap/swap.go b/swarm/swap/swap.go index 36b2948677..7570ea486d 100644 --- a/swarm/swap/swap.go +++ b/swarm/swap/swap.go @@ -19,6 +19,7 @@ package swap import ( "context" "crypto/ecdsa" + "errors" "fmt" "math/big" "os" @@ -54,7 +55,10 @@ var ( 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) + dropAt = big.NewInt(-4096 * 10000) // threshold that triggers disconnect (bytes) + + ErrInsufficientFunds = errors.New("Insufficient funds") + ErrNotAccountedMsg = errors.New("Message does not need accounting") ) const ( @@ -92,21 +96,17 @@ type SwapAccountedMsgType interface { GetMsgPrice() *big.Int } +//Handler for received messages func (sp *SwapPeer) RunAccountedProtocol(protocolHandler func(ctx context.Context, msg interface{}) error) error { + //the `peer.Run` function is a loop, so in order to pre-/post-process a message with accounting, + //we need to save the actual handler sp.handlerFunc = protocolHandler + //then run the handler loop function return sp.Run(sp.handleAccountedMsg) } -func (sp *SwapPeer) doAccountMsg(ctx context.Context, msg interface{}, direction EntryDirection) error { - if accounted, ok := msg.(SwapAccountedMsgType); ok { - price := accounted.GetMsgPrice() - //TODO: Calculate total price and account - sp.AccountMsgForPeer(price, direction) - } - return nil -} - +//get a peer's balance func (swap *Swap) GetPeerBalance(peer discover.NodeID) *big.Int { if p, ok := swap.peers[peer]; ok { return p.balance @@ -114,48 +114,137 @@ func (swap *Swap) GetPeerBalance(peer discover.NodeID) *big.Int { return nil } +//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 { - err := sp.handlerFunc(ctx, msg) - if _, ok := msg.(SwapAccountedMsgType); ok && err == nil { - sp.doAccountMsg(ctx, msg, CreditEntry) + var err error + var price *big.Int + + //the message is one which needs accounting... + if _, ok := msg.(SwapAccountedMsgType); 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) + //if not (or some other error occured), return error + if err != nil { + //also, if the error is indeed insufficient funds, then disconnect the peer + if err == ErrInsufficientFunds { + log.Error("Insufficient funds, dropping peer") + sp.Drop(err) + } + return err + } + //at this point we know there are sufficient funds, so process the message + err = sp.handlerFunc(ctx, msg) + if err == nil { + //and if no errors occurred, finally book the entry + sp.AccountMsgForPeer(ctx, msg, price, CreditEntry) + } + } else { + //this message doesn't need accounting, so just process it + err = sp.handlerFunc(ctx, msg) } return err } +//Send a message +//Check if it needs accounting, and if yes, apply accounting logic: +//Check for sufficient funds, perform operation, then account func (sp *SwapPeer) Send(ctx context.Context, msg interface{}) error { - err := sp.Peer.Send(ctx, msg) - if _, ok := msg.(SwapAccountedMsgType); ok && err == nil { - sp.doAccountMsg(ctx, msg, DebitEntry) + var err error + var price *big.Int + + //the message is one which needs accounting... + if _, ok := msg.(SwapAccountedMsgType); 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 + if err != nil { + //also, if the error is indeed insufficient funds, then disconnect the peer + if err == ErrInsufficientFunds { + log.Error("Insufficient funds, dropping peer") + sp.Drop(err) + } + return err + } + //at this point we know there are sufficient funds, so process the message + err = sp.Peer.Send(ctx, msg) + if err == nil { + //and if no errors occurred, finally book the entry + sp.AccountMsgForPeer(ctx, msg, price, DebitEntry) + } + } else { + //this message doesn't need accounting, so just process it + err = sp.Peer.Send(ctx, msg) } return err } +//check that the operation has enough funds available +func (sp *SwapPeer) checkAvailableFunds(ctx context.Context, msg interface{}, direction EntryDirection) (*big.Int, error) { + sp.lock.Lock() + defer sp.lock.Unlock() + + if accounted, ok := msg.(SwapAccountedMsgType); ok { + price := accounted.GetMsgPrice() + //local node is being credited (in its favor), so check upper limit + if direction == CreditEntry { + checkBalance := sp.balance.Add(sp.balance, price) + //(checkBalance *Int) Cmp(payAt) + // -1 if checkBalance < payAt + // 0 if checkBalance == payAt + // +1 if checkBalance > payAt + if checkBalance.Cmp(payAt) == 1 { + return nil, ErrInsufficientFunds + } + } else if direction == DebitEntry { + //(checkBalance *Int) Cmp(dropAt) + // -1 if checkBalance < dropAt + // 0 if checkBalance == dropAt + // +1 if checkBalance > dropAt + checkBalance := sp.balance.Sub(sp.balance, price) + if checkBalance.Cmp(dropAt) == -1 { + return nil, ErrInsufficientFunds + } + } + return price, nil + } + return nil, ErrNotAccountedMsg +} + //The balance is accounted from the point of view of the local node //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(price *big.Int, direction EntryDirection) { - sp.lock.Lock() - defer sp.lock.Unlock() - //local node is being credited (in its favor), so its balance increases - if direction == CreditEntry { - sp.balance = sp.balance.Add(sp.balance, price) - //local node is being debited (in favor of remote peer), so its balance decreases - } else if direction == DebitEntry { - sp.balance = sp.balance.Sub(sp.balance, price) +func (sp *SwapPeer) AccountMsgForPeer(ctx context.Context, msg interface{}, price *big.Int, direction EntryDirection) { + if _, ok := msg.(SwapAccountedMsgType); ok { + sp.lock.Lock() + defer sp.lock.Unlock() + //local node is being credited (in its favor), so its balance increases + if direction == CreditEntry { + //NOTE: do we need to check for sufficient funds again? + //operations are not atomic/transactional, so balance may have changed in the meanwhile! + sp.balance = sp.balance.Add(sp.balance, price) + //local node is being debited (in favor of remote peer), so its balance decreases + } else if direction == DebitEntry { + sp.balance = sp.balance.Sub(sp.balance, price) + } + //TODO: save to store here? init store? + sp.swapAccount.stateStore.Put(sp.storeID, sp.balance) + if sp.balance.Cmp(payAt) > -1 { + //TODO: Issue Cheque + } + if sp.balance.Cmp(dropAt) < 0 { + //TODO: Drop peer + } + log.Debug(fmt.Sprintf("balance for peer %s: %s", sp.ID(), sp.balance.String())) } - //TODO: save to store here? init store? - sp.swapAccount.stateStore.Put(sp.storeID, sp.balance) - if sp.balance.Cmp(payAt) > -1 { - //TODO: Issue Cheque - } - if sp.balance.Cmp(dropAt) < 0 { - //TODO: Drop peer - } - log.Debug(fmt.Sprintf("balance for peer %s: %s", sp.ID(), sp.balance.String())) } +//Create a new swap accounted peer func NewSwapPeer(peer *protocols.Peer, swap *Swap) *SwapPeer { balance := big.NewInt(0) + //check if there is one already in the stateStore and load it swap.stateStore.Get(peer.String()[:24]+"-swap", &balance) sp := &SwapPeer{ Peer: peer, From 9f40c788951cfbcca4db18dfbd1cd6a007c3884e Mon Sep 17 00:00:00 2001 From: Fabio Barone Date: Thu, 16 Aug 2018 18:26:52 -0500 Subject: [PATCH 6/8] 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 +} From 12718bded22f5676df61e2b7dc4c6cc8512913ef Mon Sep 17 00:00:00 2001 From: Fabio Barone Date: Thu, 16 Aug 2018 22:20:45 -0500 Subject: [PATCH 7/8] swarm/swap: improved tests, bug fixes --- swarm/swap/swap.go | 2 +- swarm/swap/swap_test.go | 116 +++++++++++++++++++++++++++++++++++----- 2 files changed, 105 insertions(+), 13 deletions(-) diff --git a/swarm/swap/swap.go b/swarm/swap/swap.go index eb2e07c805..5235d4f596 100644 --- a/swarm/swap/swap.go +++ b/swarm/swap/swap.go @@ -212,7 +212,7 @@ func (sp *SwapPeer) checkAvailableFunds(ctx context.Context, msg interface{}, di // -1 if checkBalance < dropAt // 0 if checkBalance == dropAt // +1 if checkBalance > dropAt - checkBalance := sp.balance.Sub(sp.balance, price) + checkBalance := sp.balance.Sub(sp.balance, price.Abs(price)) if checkBalance.Cmp(dropAt) == -1 { return nil, ErrInsufficientFunds } diff --git a/swarm/swap/swap_test.go b/swarm/swap/swap_test.go index 895134cbd3..62cc9e17c9 100644 --- a/swarm/swap/swap_test.go +++ b/swarm/swap/swap_test.go @@ -18,27 +18,22 @@ package swap import ( "context" - "crypto/rand" "flag" "fmt" "io/ioutil" + "math/big" "os" "path/filepath" "sync" - "sync/atomic" "testing" - "time" - "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/node" - "github.com/ethereum/go-ethereum/p2p/discover" + "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/p2p/protocols" "github.com/ethereum/go-ethereum/p2p/simulations/adapters" "github.com/ethereum/go-ethereum/rpc" - "github.com/ethereum/go-ethereum/swarm" - "github.com/ethereum/go-ethereum/swarm/api" - "github.com/ethereum/go-ethereum/swarm/network/simulation" - "github.com/ethereum/go-ethereum/swarm/storage" + "github.com/ethereum/go-ethereum/swarm/state" colorable "github.com/mattn/go-colorable" ) @@ -50,15 +45,27 @@ var ( loglevel = flag.Int("loglevel", 2, "verbosity of logs") ) +var testSpec = &protocols.Spec{ + Name: "swapTestSpec", + Version: 1, + MaxMsgSize: 10 * 1024 * 1024, + Messages: []interface{}{ + testExceedsPayAtMsg{}, + testExceedsDropAtMsg{}, + }, +} + type testExceedsPayAtMsg struct{} type testExceedsDropAtMsg struct{} func (tmsg *testExceedsPayAtMsg) GetMsgPrice() *big.Int { - return payAt.Sub(big.NewInt(1)) + diff := &big.Int{} + return diff.Sub(payAt, big.NewInt(1)) } func (tmsg *testExceedsDropAtMsg) GetMsgPrice() *big.Int { - return dropAt.Sub(big.NewInt(1)) + diff := &big.Int{} + return diff.Sub(dropAt, big.NewInt(1)) } func init() { @@ -69,12 +76,97 @@ func init() { } func TestLimits(t *testing.T) { - if dropAt >= payAt { + 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())) } } func TestExceedsPayAt(t *testing.T) { + dir, err := ioutil.TempDir("", "swap_test_store") + if err != nil { + panic(err) + } + defer os.RemoveAll(dir) + stateStore, err2 := state.NewDBStore(dir) + if err2 != nil { + panic(err2) + } + swap, err3 := NewSwap(NewDefaultSwapParams().Params, stateStore) + if err3 != nil { + t.Fatal(err3) + } + id := adapters.RandomNodeConfig().ID + testPeer := protocols.NewPeer(p2p.NewPeer(id, "testPeer", nil), nil, nil) + sp := NewSwapPeer(testPeer, swap) + sp.handlerFunc = msgHandler + ctx := context.Background() + sp.handleAccountedMsg(ctx, &testExceedsPayAtMsg{}) +} + +func TestExceedsDropAt(t *testing.T) { + dir, err := ioutil.TempDir("", "swap_test_store") + if err != nil { + panic(err) + } + defer os.RemoveAll(dir) + stateStore, err2 := state.NewDBStore(dir) + if err2 != nil { + panic(err2) + } + swap, err3 := NewSwap(NewDefaultSwapParams().Params, stateStore) + if err3 != nil { + t.Fatal(err3) + } + id := adapters.RandomNodeConfig().ID + testPeer := protocols.NewPeer(p2p.NewPeer(id, "testPeer", nil), nil, nil) + sp := NewSwapPeer(testPeer, swap) + sp.handlerFunc = msgHandler + ctx := context.Background() + err = sp.Send(ctx, &testExceedsDropAtMsg{}) + if err != ErrInsufficientFunds { + t.Fatal("Expected test to fail with insufficient funds, but it didn't") + } +} + +func TestProtocol(t *testing.T) { + /* + dir, err := ioutil.TempDir("", "swap_test_store") + if err != nil { + panic(err) + } + defer os.RemoveAll(dir) + + swap := NewSwap(NewDefaultSwapParams().Params, state.NewDBStore(dir)) + conf := adapters.RandomNodeConfig() + + protocol := func(p *p2p.Peer, rw p2p.MsgReadWriter) error { + peer := protocols.NewPeer(p, rw, testSpec) + runProtocol(peer, swap) + } + protocolTester := p2ptest.NewProtocolTester(t, conf.ID, 2, protocol) + protocolTester.TestExchanges(p2p.TestExchange{ + Label: "", + Expects: []p2ptest.Expect{ + { + Code: 5, + Msg: &RetrieveRequestMsg{ + Addr: hash0[:], + SkipCheck: true, + }, + Peer: peerID, + }, + }, + }) + */ +} + +func runProtocol(peer *protocols.Peer, swap *Swap) { + sp := NewSwapPeer(peer, swap) + sp.Peer.Run(msgHandler) +} + +func msgHandler(ctx context.Context, msg interface{}) error { + return nil } func TestSwapProtocol(t *testing.T) { From ce63aec0c98363102d7552a2c8d705c270add769 Mon Sep 17 00:00:00 2001 From: Fabio Barone Date: Fri, 17 Aug 2018 15:28:21 -0500 Subject: [PATCH 8/8] swarm/swap: very early cheques and test improvements --- swarm/swap/api.go | 32 ++-- swarm/swap/chequemanager.go | 65 +++++++ swarm/swap/protocol.go | 5 +- swarm/swap/swap.go | 26 +-- swarm/swap/swap_test.go | 346 +++++++++++++----------------------- 5 files changed, 223 insertions(+), 251 deletions(-) create mode 100644 swarm/swap/chequemanager.go diff --git a/swarm/swap/api.go b/swarm/swap/api.go index c731cb972b..960530e870 100644 --- a/swarm/swap/api.go +++ b/swarm/swap/api.go @@ -18,13 +18,17 @@ package swap import ( "context" + "errors" "math/big" - "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/hexutil" "github.com/ethereum/go-ethereum/p2p/discover" ) +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 { @@ -34,32 +38,32 @@ type APIMsg struct { // Additional public methods accessible through API for pss type API struct { *SwapProtocol - *Swap } +//TODO: define metrics type SwapMetrics struct { } -type Cheque struct { -} - func NewAPI(swap *SwapProtocol) *API { return &API{SwapProtocol: swap} } -func (swapapi *API) Balance(ctx context.Context, peer discover.NodeID) (balance *big.Int, err error) { +func (swapapi *API) BalanceWithPeer(ctx context.Context, peer discover.NodeID) (balance *big.Int, err error) { + balance = swapapi.swap.peers[peer].balance + if balance == nil { + err = ErrNoSuchPeerAccounting + } + return +} + +func (swapapi *API) Balance(ctx context.Context) (balance *big.Int, err error) { balance = big.NewInt(0) - err = nil + for _, peer := range swapapi.swap.peers { + balance.Add(balance, peer.balance) + } return } func (swapapi *API) GetSwapMetrics() (*SwapMetrics, error) { return nil, nil } - -func (swapapi *API) IssueCheque(recipient *common.Address) (*Cheque, error) { - return nil, nil -} - -func (swapapi *API) RedeemCheque(cheque *Cheque) { -} diff --git a/swarm/swap/chequemanager.go b/swarm/swap/chequemanager.go new file mode 100644 index 0000000000..a9e4494b0a --- /dev/null +++ b/swarm/swap/chequemanager.go @@ -0,0 +1,65 @@ +// Copyright 2018 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package swap + +import ( + "math/big" + + "github.com/ethereum/go-ethereum/p2p/discover" + "github.com/ethereum/go-ethereum/swarm/state" +) + +type ChequeManager struct { + stateStore state.Store + serialPerNode map[discover.NodeID]uint64 + openDebitCheques map[discover.NodeID][]*Cheque + openCreditCheques map[discover.NodeID][]*Cheque +} + +type Cheque struct { + serial uint64 + timeout uint64 + amount *big.Int + sumCumulated *big.Int + beneficiary discover.NodeID //this should probably be common.Address? +} + +func NewChequeManager(stateStore state.Store) *ChequeManager { + return &ChequeManager{ + stateStore: stateStore, + //TODO: restore from state store + serialPerNode: make(map[discover.NodeID]uint64), + openDebitCheques: make(map[discover.NodeID][]*Cheque), + openCreditCheques: make(map[discover.NodeID][]*Cheque), + } +} + +func (mgr *ChequeManager) CreateCheque(beneficiary discover.NodeID, amount *big.Int) *Cheque { + mgr.serialPerNode[beneficiary]++ + cheque := &Cheque{ + serial: mgr.serialPerNode[beneficiary], + beneficiary: beneficiary, + amount: amount, + } + openCheques := mgr.openDebitCheques[beneficiary] + if openCheques == nil { + openCheques = make([]*Cheque, 0) + } + openCheques = append(openCheques, cheque) + mgr.openDebitCheques[beneficiary] = openCheques + return cheque +} diff --git a/swarm/swap/protocol.go b/swarm/swap/protocol.go index 4846750c9b..9863ce20f4 100644 --- a/swarm/swap/protocol.go +++ b/swarm/swap/protocol.go @@ -36,6 +36,7 @@ const ( type SwapProtocol struct { peersMu sync.RWMutex peers map[discover.NodeID]*SwapProtocolPeer + swap *Swap } // Peer is the Peer extension for the streaming protocol @@ -44,9 +45,10 @@ type SwapProtocolPeer struct { swapProtocol *SwapProtocol } -func NewSwapProtocol() *SwapProtocol { +func NewSwapProtocol(swapAccount *Swap) *SwapProtocol { proto := &SwapProtocol{ peers: make(map[discover.NodeID]*SwapProtocolPeer), + swap: swapAccount, } return proto } @@ -61,6 +63,7 @@ func NewPeer(peer *protocols.Peer, swap *SwapProtocol) *SwapProtocolPeer { } type IssueChequeMsg struct { + Cheque *Cheque } type RedeemChequeMsg struct { diff --git a/swarm/swap/swap.go b/swarm/swap/swap.go index 5235d4f596..3334ab9905 100644 --- a/swarm/swap/swap.go +++ b/swarm/swap/swap.go @@ -70,10 +70,11 @@ const ( // Swift Automatic Payments // a peer to peer micropayment system type Swap struct { - stateStore state.Store - lock sync.RWMutex - peers map[discover.NodeID]*SwapPeer - local *Params // local peer's swap parameters + chequeManager *ChequeManager + stateStore state.Store + lock sync.RWMutex + peers map[discover.NodeID]*SwapPeer + local *Params // local peer's swap parameters } type SwapPeer struct { @@ -244,14 +245,14 @@ func (sp *SwapPeer) AccountMsgForPeer(ctx context.Context, msg interface{}, pric // -1 if sp.balance < payAt // 0 if sp.balance == payAt // +1 if sp.balance > payAt - if sp.balance.Cmp(payAt) > 1 { + 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 { + if sp.balance.Cmp(dropAt) == -1 { sp.Drop(ErrInsufficientFunds) } log.Debug(fmt.Sprintf("balance for peer %s: %s", sp.ID(), sp.balance.String())) @@ -259,7 +260,11 @@ func (sp *SwapPeer) AccountMsgForPeer(ctx context.Context, msg interface{}, pric } func (sp *SwapPeer) issueCheque(ctx context.Context) error { - msg := IssueChequeMsg{} + amount := big.NewInt(0) + cheque := sp.swapAccount.chequeManager.CreateCheque(sp.ID(), amount.Abs(payAt)) + msg := IssueChequeMsg{ + Cheque: cheque, + } return sp.Send(ctx, msg) } @@ -342,9 +347,10 @@ type PayProfile struct { func NewSwap(local *Params, stateStore state.Store) (swap *Swap, err error) { swap = &Swap{ - local: local, - stateStore: stateStore, - peers: make(map[discover.NodeID]*SwapPeer), + chequeManager: NewChequeManager(stateStore), + local: local, + stateStore: stateStore, + peers: make(map[discover.NodeID]*SwapPeer), } //swap.SetParams(local) diff --git a/swarm/swap/swap_test.go b/swarm/swap/swap_test.go index 62cc9e17c9..99408d9092 100644 --- a/swarm/swap/swap_test.go +++ b/swarm/swap/swap_test.go @@ -26,6 +26,7 @@ import ( "path/filepath" "sync" "testing" + "time" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/node" @@ -55,6 +56,21 @@ var testSpec = &protocols.Spec{ }, } +type dummyRW struct{} + +func (d *dummyRW) WriteMsg(msg p2p.Msg) error { + return nil +} + +func (d *dummyRW) ReadMsg() (p2p.Msg, error) { + return p2p.Msg{ + Code: 0, + Size: 0, + Payload: nil, + ReceivedAt: time.Now(), + }, nil +} + type testExceedsPayAtMsg struct{} type testExceedsDropAtMsg struct{} @@ -81,107 +97,86 @@ func TestLimits(t *testing.T) { } } +//unit test for exceeds pay limit func TestExceedsPayAt(t *testing.T) { - dir, err := ioutil.TempDir("", "swap_test_store") - if err != nil { - panic(err) - } - defer os.RemoveAll(dir) - stateStore, err2 := state.NewDBStore(dir) - if err2 != nil { - panic(err2) - } - swap, err3 := NewSwap(NewDefaultSwapParams().Params, stateStore) - if err3 != nil { - t.Fatal(err3) - } - id := adapters.RandomNodeConfig().ID - testPeer := protocols.NewPeer(p2p.NewPeer(id, "testPeer", nil), nil, nil) + swap, testDir := createTestSwap(t) + defer os.RemoveAll(testDir) + + testPeer := newDummyPeer() sp := NewSwapPeer(testPeer, swap) - sp.handlerFunc = msgHandler + sp.handlerFunc = dummyMsgHandler + ctx := context.Background() - sp.handleAccountedMsg(ctx, &testExceedsPayAtMsg{}) + sp.Send(ctx, &testExceedsPayAtMsg{}) + + cheques := sp.swapAccount.chequeManager.openDebitCheques[sp.ID()] + if cheques == nil { + t.Fatal("Expected cheques for this peer to be present, but are nil") + } + if len(cheques) == 0 { + t.Fatal("Expected a cheque to have arrived, but len is zero") + } + if cheques[0].serial != 1 { + t.Fatal(fmt.Sprintf("Expected the serial to be one (first message but is: %d", cheques[0].serial)) + } + absPayAt := big.NewInt(0) + if cheques[0].amount.Cmp(absPayAt.Abs(payAt)) != 0 { + t.Fatal(fmt.Sprintf("Expected the serial to be one (first message but is: %d", cheques[0].serial)) + } } +//unit test for exceeds drop limit func TestExceedsDropAt(t *testing.T) { - dir, err := ioutil.TempDir("", "swap_test_store") - if err != nil { - panic(err) - } - defer os.RemoveAll(dir) - stateStore, err2 := state.NewDBStore(dir) - if err2 != nil { - panic(err2) - } - swap, err3 := NewSwap(NewDefaultSwapParams().Params, stateStore) - if err3 != nil { - t.Fatal(err3) - } - id := adapters.RandomNodeConfig().ID - testPeer := protocols.NewPeer(p2p.NewPeer(id, "testPeer", nil), nil, nil) + swap, testDir := createTestSwap(t) + defer os.RemoveAll(testDir) + + testPeer := newDummyPeer() sp := NewSwapPeer(testPeer, swap) - sp.handlerFunc = msgHandler + sp.handlerFunc = dummyMsgHandler + ctx := context.Background() - err = sp.Send(ctx, &testExceedsDropAtMsg{}) + err := sp.Send(ctx, &testExceedsDropAtMsg{}) if err != ErrInsufficientFunds { t.Fatal("Expected test to fail with insufficient funds, but it didn't") } } -func TestProtocol(t *testing.T) { - /* - dir, err := ioutil.TempDir("", "swap_test_store") - if err != nil { - panic(err) - } - defer os.RemoveAll(dir) - - swap := NewSwap(NewDefaultSwapParams().Params, state.NewDBStore(dir)) - conf := adapters.RandomNodeConfig() - - protocol := func(p *p2p.Peer, rw p2p.MsgReadWriter) error { - peer := protocols.NewPeer(p, rw, testSpec) - runProtocol(peer, swap) - } - protocolTester := p2ptest.NewProtocolTester(t, conf.ID, 2, protocol) - protocolTester.TestExchanges(p2p.TestExchange{ - Label: "", - Expects: []p2ptest.Expect{ - { - Code: 5, - Msg: &RetrieveRequestMsg{ - Addr: hash0[:], - SkipCheck: true, - }, - Peer: peerID, - }, - }, - }) - */ +func createTestSwap(t *testing.T) (*Swap, string) { + dir, err := ioutil.TempDir("", "swap_test_store") + if err != nil { + t.Fatal(err) + } + stateStore, err2 := state.NewDBStore(dir) + if err2 != nil { + t.Fatal(err2) + } + swap, err3 := NewSwap(NewDefaultSwapParams().Params, stateStore) + if err3 != nil { + t.Fatal(err3) + } + return swap, dir } func runProtocol(peer *protocols.Peer, swap *Swap) { sp := NewSwapPeer(peer, swap) - sp.Peer.Run(msgHandler) + sp.Peer.Run(dummyMsgHandler) } -func msgHandler(ctx context.Context, msg interface{}) error { +func dummyMsgHandler(ctx context.Context, msg interface{}) error { return nil } -func TestSwapProtocol(t *testing.T) { +func TestSwapRPC(t *testing.T) { + swap, testDir := createTestSwap(t) + defer os.RemoveAll(testDir) // create the two nodes stack_one, err := newServiceNode(p2pPort, 0, 0) if err != nil { - log.Crit("Create servicenode #1 fail", "err", err) - } - stack_two, err := newServiceNode(p2pPort+1, 0, 0) - if err != nil { - log.Crit("Create servicenode #2 fail", "err", err) + t.Fatal("Create servicenode #1 fail", "err", err) } - instance := NewSwapProtocol() + instance := NewSwapProtocol(swap) // wrapper function for servicenode to start the service swapsvc := func(ctx *node.ServiceContext) (node.Service, error) { return &API{ @@ -192,181 +187,80 @@ func TestSwapProtocol(t *testing.T) { // register adds the service to the services the servicenode starts when started err = stack_one.Register(swapsvc) if err != nil { - log.Crit("Register service in servicenode #1 fail", "err", err) + t.Fatal("Register service in servicenode #1 fail", "err", err) } - err = stack_two.Register(swapsvc) - if err != nil { - log.Crit("Register service in servicenode #2 fail", "err", err) - } - // start the nodes err = stack_one.Start() if err != nil { - log.Crit("servicenode #1 start failed", "err", err) - } - err = stack_two.Start() - if err != nil { - log.Crit("servicenode #2 start failed", "err", err) + t.Fatal("servicenode #1 start failed", "err", err) } // connect to the servicenode RPCs rpcclient_one, err := rpc.Dial(filepath.Join(stack_one.DataDir(), ipcpath)) if err != nil { - log.Crit("connect to servicenode #1 IPC fail", "err", err) + t.Fatal("connect to servicenode #1 IPC fail", "err", err) } defer os.RemoveAll(stack_one.DataDir()) - rpcclient_two, err := rpc.Dial(filepath.Join(stack_two.DataDir(), ipcpath)) - if err != nil { - log.Crit("connect to servicenode #2 IPC fail", "err", err) - } - defer os.RemoveAll(stack_two.DataDir()) - - // display that the initial pong counts are 0 - var balance int + var balance *big.Int err = rpcclient_one.Call(&balance, "swap_balance") if err != nil { - log.Crit("servicenode #1 pongcount RPC failed", "err", err) + t.Fatal("servicenode #1 RPC failed", "err", err) } - log.Info("servicenode #1 before ping", "balance-1", balance) + log.Debug("servicenode #1 balance", "balance-1", balance) - err = rpcclient_two.Call(&balance, "swap_balance") + if balance.Cmp(big.NewInt(0)) != 0 { + t.Fatal("Expected balance to be 0 but it is not") + } + + dummyPeer1 := newDummyPeer() + dummyPeer2 := newDummyPeer() + id1 := dummyPeer1.ID() + id2 := dummyPeer2.ID() + + fake1 := int64(234) + fake2 := int64(-100) + fakeBalance1 := big.NewInt(fake1) + fakeBalance2 := big.NewInt(fake2) + + swap.peers[id1] = NewSwapPeer(dummyPeer1, swap) + swap.peers[id2] = NewSwapPeer(dummyPeer2, swap) + swap.peers[id1].balance = fakeBalance1 + swap.peers[id2].balance = fakeBalance2 + + err = rpcclient_one.Call(&balance, "swap_balanceWithPeer", id1) if err != nil { - log.Crit("servicenode #2 pongcount RPC failed", "err", err) + t.Fatal("servicenode #1 RPC failed", "err", err) + } + log.Debug("balance1", "balance-1", balance) + if balance.Cmp(fakeBalance1) != 0 { + t.Fatal(fmt.Sprintf("Expected balance %s to be equal to fake balance %s, but it is not", balance.String(), fakeBalance1.String())) } - log.Info("servicenode #2 before ping", "balance-2", balance) - /* - // get the server instances - srv_one := stack_one.Server() - srv_two := stack_two.Server() + err = rpcclient_one.Call(&balance, "swap_balanceWithPeer", id2) + if err != nil { + t.Fatal("servicenode #1 RPC failed", "err", err) + } + log.Debug("balance2", "balance-2", balance) + if balance.Cmp(fakeBalance2) != 0 { + t.Fatal(fmt.Sprintf("Expected balance %s to be equal to fake balance %s, but it is not", balance.String(), fakeBalance2.String())) + } - // subscribe to peerevents - eventOneC := make(chan *p2p.PeerEvent) - sub_one := srv_one.SubscribeEvents(eventOneC) + err = rpcclient_one.Call(&balance, "swap_balance") + if err != nil { + t.Fatal("servicenode #1 RPC failed", "err", err) + } + log.Debug("balance", "balance", balance) - eventTwoC := make(chan *p2p.PeerEvent) - sub_two := srv_two.SubscribeEvents(eventTwoC) + fakeSum := big.NewInt(fake1 + fake2) + if balance.Cmp(fakeSum) != 0 { + t.Fatal(fmt.Sprintf("Expected balance %s to be equal to sum %s, but it is not", balance.String(), fakeSum.String())) + } +} - // connect the nodes - p2pnode_two := srv_two.Self() - srv_one.AddPeer(p2pnode_two) - - // fork and do the pinging - stackW.Add(2) - pingmax_one := 4 - pingmax_two := 2 - - go func() { - - // 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) - stackW.Done() - return - } - log.Debug("server #1 connected", "peer", ev.Peer) - - // send the pings - for i := 0; i < pingmax_one; i++ { - err := rpcclient_one.Call(nil, "foo_ping", ev.Peer) - if err != nil { - log.Error("server #1 RPC ping fail", "err", err) - stackW.Done() - break - } - } - - // wait for all msgrecv events - // pings we receive, and pongs we expect from pings we sent - for i := 0; i < pingmax_two+pingmax_one; { - ev := <-eventOneC - log.Warn("msg", "type", ev.Type, "i", i) - if ev.Type == "msgrecv" { - i++ - } - } - - stackW.Done() - }() - - // mirrors the previous go func - go func() { - ev := <-eventTwoC - if ev.Type != "add" { - log.Error("expected peer add", "eventtype", ev.Type) - stackW.Done() - return - } - log.Debug("server #2 connected", "peer", ev.Peer) - for i := 0; i < pingmax_two; i++ { - err := rpcclient_two.Call(nil, "foo_ping", ev.Peer) - if err != nil { - log.Error("server #2 RPC ping fail", "err", err) - stackW.Done() - break - } - } - - for i := 0; i < pingmax_one+pingmax_two; { - ev := <-eventTwoC - if ev.Type == "msgrecv" { - log.Warn("msg", "type", ev.Type, "i", i) - i++ - } - } - - stackW.Done() - }() - - // wait for the two ping pong exchanges to finish - stackW.Wait() - - // tell the API to shut down - // this will disconnect the peers and close the channels connecting API and protocol - err = rpcclient_one.Call(nil, "foo_quit", srv_two.Self().ID) - if err != nil { - log.Error("server #1 RPC quit fail", "err", err) - } - err = rpcclient_two.Call(nil, "foo_quit", srv_one.Self().ID) - if err != nil { - log.Error("server #2 RPC quit fail", "err", err) - } - - // disconnect will generate drop events - for { - ev := <-eventOneC - if ev.Type == "drop" { - break - } - } - for { - ev := <-eventTwoC - if ev.Type == "drop" { - break - } - } - - // proudly inspect the results - err = rpcclient_one.Call(&count, "foo_pongCount") - if err != nil { - log.Crit("servicenode #1 pongcount RPC failed", "err", err) - } - log.Info("servicenode #1 after ping", "pongcount", count) - - err = rpcclient_two.Call(&count, "foo_pongCount") - if err != nil { - log.Crit("servicenode #2 pongcount RPC failed", "err", err) - } - log.Info("servicenode #2 after ping", "pongcount", count) - - // bring down the servicenodes - sub_one.Unsubscribe() - sub_two.Unsubscribe() - stack_one.Stop() - stack_two.Stop() - */ +func newDummyPeer() *protocols.Peer { + id := adapters.RandomNodeConfig().ID + return protocols.NewPeer(p2p.NewPeer(id, "testPeer", nil), &dummyRW{}, testSpec) } func newServiceNode(port int, httpport int, wsport int, modules ...string) (*node.Node, error) {