Swap pricing

api/api.go

@@ -266,7 +266,6 @@ func (a *API) Resolve(ctx context.Context, address string) (storage.Address, err
 	return resolved[:], nil
 }
 
-//
 func tld(address string) (tld string) {
 	splitAddress := strings.Split(address, ".")
 	if len(splitAddress) > 1 {

api/config.go

@@ -54,8 +54,8 @@ type Config struct {
 	// Swap configs
 	SwapBackendURL          string         // Ethereum API endpoint
 	SwapEnabled             bool           // whether SWAP incentives are enabled
-	SwapPaymentThreshold    uint64         // honey amount at which a payment is triggered
-	SwapDisconnectThreshold uint64         // honey amount at which a peer disconnects
+	SwapPaymentThreshold    uint           // honey amount at which a payment is triggered
+	SwapDisconnectThreshold uint           // honey amount at which a peer disconnects
 	SwapInitialDeposit      uint64         // initial deposit amount to the chequebook
 	SwapLogPath             string         // dir to swap related audit logs
 	Contract                common.Address // address of the chequebook contract

bzzeth/bzzeth_test.go

@@ -87,7 +87,7 @@ func newTestNetworkStore(t *testing.T) (prvkey *ecdsa.PrivateKey, netStore *stor
 	}
 
 	netStore = storage.NewNetStore(localStore, bzzAddr, enode.ID{})
-	r := retrieval.New(kad, netStore, bzzAddr, nil)
+	r := retrieval.New(kad, netStore, bzzAddr, nil, 0)
 	netStore.RemoteGet = r.RequestFromPeers
 
 	cleanup = func() {

cmd/swarm/config.go

@@ -213,10 +213,10 @@ func flagsOverride(currentConfig *bzzapi.Config, ctx *cli.Context) *bzzapi.Confi
 	if initialDepo := ctx.GlobalUint64(SwarmSwapInitialDepositFlag.Name); initialDepo != 0 {
 		currentConfig.SwapInitialDeposit = initialDepo
 	}
-	if paymentThreshold := ctx.GlobalUint64(SwarmSwapPaymentThresholdFlag.Name); paymentThreshold != 0 {
+	if paymentThreshold := ctx.GlobalUint(SwarmSwapPaymentThresholdFlag.Name); paymentThreshold != 0 {
 		currentConfig.SwapPaymentThreshold = paymentThreshold
 	}
-	if disconnectThreshold := ctx.GlobalUint64(SwarmSwapDisconnectThresholdFlag.Name); disconnectThreshold != 0 {
+	if disconnectThreshold := ctx.GlobalUint(SwarmSwapDisconnectThresholdFlag.Name); disconnectThreshold != 0 {
 		currentConfig.SwapDisconnectThreshold = disconnectThreshold
 	}
 	if ctx.GlobalIsSet(SwarmNoSyncFlag.Name) {

cmd/swarm/config_test.go

@@ -238,8 +238,8 @@ func TestConfigCmdLineOverrides(t *testing.T) {
 		fmt.Sprintf("--%s", utils.DataDirFlag.Name), dir,
 		fmt.Sprintf("--%s", utils.IPCPathFlag.Name), conf.IPCPath,
 		"--verbosity", fmt.Sprintf("%d", *testutil.Loglevel),
-		fmt.Sprintf("--%s", SwarmSwapPaymentThresholdFlag.Name), strconv.FormatUint(swap.DefaultPaymentThreshold+1, 10),
-		fmt.Sprintf("--%s", SwarmSwapDisconnectThresholdFlag.Name), strconv.FormatUint(swap.DefaultDisconnectThreshold+1, 10),
+		fmt.Sprintf("--%s", SwarmSwapPaymentThresholdFlag.Name), strconv.FormatUint(uint64(swap.DefaultPaymentThreshold)+1, 10),
+		fmt.Sprintf("--%s", SwarmSwapDisconnectThresholdFlag.Name), strconv.FormatUint(uint64(swap.DefaultDisconnectThreshold)+1, 10),
 		fmt.Sprintf("--%s", SwarmEnablePinningFlag.Name),
 	}
 

network/retrieval/peer.go

@@ -28,21 +28,26 @@ import (
 	"github.com/ethersphere/swarm/storage"
 )
 
-// Peer wraps BzzPeer with a contextual logger and tracks open
+var ErrCannotFindRUID = errors.New("cannot find ruid")
+var ErrNoAddressMatch = errors.New("retrieve request found but address does not match")
+
+// Peer wraps BzzPeer with a contextual logger and tracks open. TODO: update comment
 // retrievals for that peer
 type Peer struct {
 	*network.BzzPeer
-	logger     log.Logger             // logger with base and peer address
-	mtx        sync.Mutex             // synchronize retrievals
-	retrievals map[uint]chunk.Address // current ongoing retrievals
+	logger           log.Logger             // logger with base and peer address
+	mtx              sync.Mutex             // synchronize retrievals
+	retrievals       map[uint]chunk.Address // current ongoing retrievals
+	priceInformation []uint                 // price per Proximity order
 }
 
 // NewPeer is the constructor for Peer
 func NewPeer(peer *network.BzzPeer, baseKey []byte) *Peer {
 	return &Peer{
-		BzzPeer:    peer,
-		logger:     log.New("base", hex.EncodeToString(baseKey)[:16], "peer", peer.ID().String()[:16]),
-		retrievals: make(map[uint]chunk.Address),
+		BzzPeer:          peer,
+		logger:           log.New("base", hex.EncodeToString(baseKey)[:16], "peer", peer.ID().String()[:16]),
+		retrievals:       make(map[uint]chunk.Address),
+		priceInformation: make([]uint, 10), //TODO: more intelligent way of creating this array. Load historical data from store
 	}
 }
 
@@ -57,16 +62,23 @@ func (p *Peer) addRetrieval(ruid uint, addr storage.Address) {
 // chunkReceived is called upon ChunkDelivery message reception
 // it is meant to idenfify unsolicited chunk deliveries
 func (p *Peer) checkRequest(ruid uint, addr storage.Address) error {
+	// 0 means synthetic chunk
+	if ruid == uint(0) {
+		return nil
+	}
 	p.mtx.Lock()
 	defer p.mtx.Unlock()
 	v, ok := p.retrievals[ruid]
 	if !ok {
-		return errors.New("cannot find ruid")
+		return ErrCannotFindRUID
 	}
-	delete(p.retrievals, ruid) // since we got the delivery we wanted - it is safe to delete the retrieve request
 	if !bytes.Equal(v, addr) {
-		return errors.New("retrieve request found but address does not match")
+		return ErrNoAddressMatch
 	}
 
 	return nil
 }
+
+func (p *Peer) deleteRequest(ruid uint) {
+	delete(p.retrievals, ruid)
+}

network/retrieval/retrieve.go

@@ -39,7 +39,6 @@ import (
 	"github.com/ethersphere/swarm/p2p/protocols"
 	"github.com/ethersphere/swarm/spancontext"
 	"github.com/ethersphere/swarm/storage"
-	"github.com/ethersphere/swarm/swap"
 	opentracing "github.com/opentracing/opentracing-go"
 	olog "github.com/opentracing/opentracing-go/log"
 )
@@ -63,30 +62,20 @@ var (
 		Messages: []interface{}{
 			ChunkDelivery{},
 			RetrieveRequest{},
+			NewPrice{},
 		},
 	}
 
 	ErrNoPeerFound = errors.New("no peer found")
 )
 
-// Price is the method through which a message type marks itself
-// as implementing the protocols.Price protocol and thus
-// as swap-enabled message
-func (rr *RetrieveRequest) Price() *protocols.Price {
-	return &protocols.Price{
-		Value:   swap.RetrieveRequestPrice,
-		PerByte: false,
-		Payer:   protocols.Sender,
-	}
-}
-
 // Price is the method through which a message type marks itself
 // as implementing the protocols.Price protocol and thus
 // as swap-enabled message
 func (cd *ChunkDelivery) Price() *protocols.Price {
 	return &protocols.Price{
-		Value:   swap.ChunkDeliveryPrice,
-		PerByte: true,
+		Value:   cd.price,
+		PerByte: false,
 		Payer:   protocols.Receiver,
 	}
 }
@@ -95,6 +84,7 @@ func (cd *ChunkDelivery) Price() *protocols.Price {
 type Retrieval struct {
 	netStore *storage.NetStore
 	kad      *network.Kademlia
+	margin   uint
 	mtx      sync.RWMutex       // protect peer map
 	peers    map[enode.ID]*Peer // compatible peers
 	spec     *protocols.Spec    // protocol spec
@@ -103,10 +93,11 @@ type Retrieval struct {
 }
 
 // New returns a new instance of the retrieval protocol handler
-func New(kad *network.Kademlia, ns *storage.NetStore, baseKey []byte, balance protocols.Balance) *Retrieval {
+func New(kad *network.Kademlia, ns *storage.NetStore, baseKey []byte, balance protocols.Balance, margin uint) *Retrieval {
 	r := &Retrieval{
 		netStore: ns,
 		kad:      kad,
+		margin:   margin,
 		peers:    make(map[enode.ID]*Peer),
 		spec:     spec,
 		logger:   log.New("base", hex.EncodeToString(baseKey)[:16]),
@@ -140,6 +131,29 @@ func (r *Retrieval) getPeer(id enode.ID) *Peer {
 	return r.peers[id]
 }
 
+// get the pric for requesting a chunk from peer
+func (r *Retrieval) getPeerPrice(peer enode.ID, chunkAddr chunk.Address) uint {
+	r.mtx.RLock()
+	defer r.mtx.RUnlock()
+	//TODO: does this return 0 if the price was never initialized?
+	return r.peers[peer].priceInformation[chunk.Proximity(chunkAddr, peer.Bytes())] //TODO: is peer.Bytes() the correct address to use?
+}
+
+func (r *Retrieval) setPeerPrice(peer enode.ID, chunkAddr chunk.Address, newPrice uint) {
+	r.mtx.Lock()
+	defer r.mtx.Unlock()
+	r.peers[peer].priceInformation[chunk.Proximity(chunkAddr, peer.Bytes())] = newPrice
+}
+
+// returns true if the price
+func (r *Retrieval) verifyPrice(offeredPrice uint, wantedPrice uint) bool {
+	if offeredPrice >= wantedPrice {
+		return true
+	} else {
+		return false
+	}
+}
+
 // Run is being dispatched when 2 nodes connect
 func (r *Retrieval) Run(bp *network.BzzPeer) error {
 	sp := NewPeer(bp, r.kad.BaseAddr())
@@ -158,6 +172,9 @@ func (r *Retrieval) handleMsg(p *Peer) func(context.Context, interface{}) error
 			go r.handleRetrieveRequest(ctx, p, msg)
 		case *ChunkDelivery:
 			go r.handleChunkDelivery(ctx, p, msg)
+
+		case *NewPrice:
+			go r.handleNewPrice(ctx, p, msg)
 		}
 		return nil
 	}
@@ -298,10 +315,31 @@ func (r *Retrieval) findPeer(ctx context.Context, req *storage.Request) (retPeer
 	return retPeer, nil
 }
 
+// handleNewPrice updates the priceInformation for the particular peer. If there are outstanding requests for the particular chunk, we handle them.
+func (r *Retrieval) handleNewPrice(ctx context.Context, p *Peer, msg *NewPrice) {
+	// update price
+	r.setPeerPrice(p.ID(), msg.Addr, msg.Price)
+	err := r.peers[p.ID()].checkRequest(msg.Ruid, msg.Addr)
+	// outstanding request
+	if err == nil {
+		// TODO: compare newPrice with margin
+		// if newPrice + margin >= outstanding request price => issue new request
+		// else => send NewPrice message
+	}
+	// TODO: look up outstanding chunk requests, verify wether the newPrice + margin >= the price of the outstanding request. If so, resend with new price, if not, send newPrice message ourself to Origininator of request
+}
+
 // handleRetrieveRequest handles an incoming retrieve request from a certain Peer
 // if the chunk is found in the localstore it is served immediately, otherwise
 // it results in a new retrieve request to candidate peers in our kademlia
 func (r *Retrieval) handleRetrieveRequest(ctx context.Context, p *Peer, msg *RetrieveRequest) {
+	// return directly with a NewPrice msg if msg.price is not bigger than margin
+	wantedPrice := r.margin
+	ok := r.verifyPrice(msg.Price, wantedPrice)
+	if !ok {
+		r.createAndSendNewPrice(ctx, p, msg.Ruid, msg.Addr, wantedPrice)
+	}
+
 	p.logger.Debug("retrieval.handleRetrieveRequest", "ref", msg.Addr)
 	handleRetrieveRequestMsgCount.Inc(1)
 
@@ -313,13 +351,15 @@ func (r *Retrieval) handleRetrieveRequest(ctx context.Context, p *Peer, msg *Ret
 
 	defer osp.Finish()
 
-	ctx, cancel := context.WithTimeout(ctx, timeouts.FetcherGlobalTimeout)
-	defer cancel()
-
 	req := &storage.Request{
 		Addr:   msg.Addr,
+		Ruid: 	msg.Ruid,
+		Price:  msg.Price,
 		Origin: p.ID(),
 	}
+	ctx, cancel := context.WithTimeout(ctx, timeouts.FetcherGlobalTimeout)
+	defer cancel()
+
 	chunk, err := r.netStore.Get(ctx, chunk.ModeGetRequest, req)
 	if err != nil {
 		retrieveChunkFail.Inc(1)
@@ -328,9 +368,9 @@ func (r *Retrieval) handleRetrieveRequest(ctx context.Context, p *Peer, msg *Ret
 	}
 
 	p.logger.Trace("retrieval.handleRetrieveRequest - delivery", "ref", msg.Addr)
-
 	deliveryMsg := &ChunkDelivery{
 		Ruid:  msg.Ruid,
+		price: msg.Price,
 		Addr:  chunk.Address(),
 		SData: chunk.Data(),
 	}
@@ -350,6 +390,9 @@ func (r *Retrieval) handleRetrieveRequest(ctx context.Context, p *Peer, msg *Ret
 func (r *Retrieval) handleChunkDelivery(ctx context.Context, p *Peer, msg *ChunkDelivery) {
 	p.logger.Debug("retrieval.handleChunkDelivery", "ref", msg.Addr)
 	err := p.checkRequest(msg.Ruid, msg.Addr)
+	if err == ErrNoAddressMatch {
+		p.deleteRequest(msg.Ruid)
+	}
 	if err != nil {
 		unsolicitedChunkDelivery.Inc(1)
 		p.logger.Error("unsolicited chunk delivery from peer", "ruid", msg.Ruid, "addr", msg.Addr, "err", err)
@@ -397,7 +440,7 @@ func (r *Retrieval) RequestFromPeers(ctx context.Context, req *storage.Request,
 
 	const maxFindPeerRetries = 5
 	retries := 0
-
+	//TODO: make FINDPEER request based on r.Peer.priceInformation
 FINDPEER:
 	sp, err := r.findPeer(ctx, req)
 	if err != nil {
@@ -418,9 +461,23 @@ FINDPEER:
 		goto FINDPEER
 	}
 
+	//check if the price is ok, if not ok send a NewPrice msg
+	wantedPrice := r.margin + r.getPeerPrice(sp.ID(), req.Addr)
+	ok := r.verifyPrice(req.Price, wantedPrice)
+	if !ok {
+		r.createAndSendNewPrice(ctx, protoPeer, req.Ruid, req.Addr, wantedPrice)
+	}
+
+	// create a non-zero ruid (zero specifices synthetic chunk)
+	ruid := uint(rand.Uint32())
+	for ruid == 0 {
+		ruid = uint(rand.Uint32())
+	}
+
 	ret := &RetrieveRequest{
-		Ruid: uint(rand.Uint32()),
-		Addr: req.Addr,
+		Ruid:  ruid,
+		Price: r.getPeerPrice(sp.ID(), req.Addr),
+		Addr:  req.Addr,
 	}
 	protoPeer.logger.Trace("sending retrieve request", "ref", ret.Addr, "origin", localID, "ruid", ret.Ruid)
 	protoPeer.addRetrieval(ret.Ruid, ret.Addr)
@@ -434,6 +491,15 @@ FINDPEER:
 	return &spID, nil
 }
 
+func (r *Retrieval) createAndSendNewPrice(ctx context.Context, destinationPeer *Peer, ruid uint, addr storage.Address, wantedPrice uint) error {
+	newPriceMsg := &NewPrice{
+		Ruid:  ruid,
+		Price: wantedPrice,
+		Addr:  addr,
+	}
+	return destinationPeer.Send(ctx, newPriceMsg)
+}
+
 func (r *Retrieval) Start(server *p2p.Server) error {
 	r.logger.Info("starting bzz-retrieve")
 	return nil

network/retrieval/retrieve_test.go

@@ -449,7 +449,7 @@ func TestRequestFromPeers(t *testing.T) {
 
 	to.On(peer)
 
-	s := New(to, nil, to.BaseAddr(), nil)
+	s := New(to, nil, to.BaseAddr(), nil, 0)
 
 	req := storage.NewRequest(storage.Address(hash0[:]))
 	id, err := s.findPeer(context.Background(), req)
@@ -465,14 +465,9 @@ func TestRequestFromPeers(t *testing.T) {
 //TestHasPriceImplementation is to check that Retrieval provides priced messages
 func TestHasPriceImplementation(t *testing.T) {
 	price := (&ChunkDelivery{}).Price()
-	if price == nil || price.Value == 0 {
+	if price == nil {
 		t.Fatal("No prices set for chunk delivery msg")
 	}
-
-	price = (&RetrieveRequest{}).Price()
-	if price == nil || price.Value == 0 {
-		t.Fatal("No prices set for retrieve requests")
-	}
 }
 
 func newBzzRetrieveWithLocalstore(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) {
@@ -502,7 +497,7 @@ func newBzzRetrieveWithLocalstore(ctx *adapters.ServiceContext, bucket *sync.Map
 		return nil, nil, err
 	}
 
-	r := New(kad, netStore, kad.BaseAddr(), nil)
+	r := New(kad, netStore, kad.BaseAddr(), nil, 0)
 	netStore.RemoteGet = r.RequestFromPeers
 	bucket.Store(bucketKeyFileStore, fileStore)
 	bucket.Store(bucketKeyNetstore, netStore)
@@ -621,7 +616,7 @@ func newRetrievalTester(t *testing.T, prvkey *ecdsa.PrivateKey, netStore *storag
 		prvkey = key
 	}
 
-	r := New(kad, netStore, kad.BaseAddr(), nil)
+	r := New(kad, netStore, kad.BaseAddr(), nil, 0)
 	protocolTester := p2ptest.NewProtocolTester(prvkey, 1, r.runProtocol)
 
 	return protocolTester, r, protocolTester.Stop, nil