swarm: swap implementation as devp2p2 protocol

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 <http://www.gnu.org/licenses/>.
+
+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) {
+}

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 <http://www.gnu.org/licenses/>.
+
+// +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)))
+}
+*/

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 <http://www.gnu.org/licenses/>.
+
+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
+}

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 <http://www.gnu.org/licenses/>.
+
+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()
+	*/
+}

swarm/swarm.go

@@ -51,6 +51,8 @@ import (
 	"github.com/ethereum/go-ethereum/swarm/storage"
 	"github.com/ethereum/go-ethereum/swarm/storage/feed"
 	"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 +80,7 @@ type Swarm struct {
 	netStore    *storage.NetStore
 	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 +209,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 {
@@ -361,6 +368,13 @@ func (self *Swarm) Start(srv *p2p.Server) error {
 	}
 	log.Info("Swarm network started", "bzzaddr", fmt.Sprintf("%x", self.bzz.Hive.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)
 	}