les: rework client db

les/api.go

@@ -104,12 +104,16 @@ func (api *PrivateLightServerAPI) clientInfo(c *clientInfo, id enode.ID) map[str
 		info["capacity"] = c.capacity
 		pb, nb := c.balanceTracker.getBalance(now)
 		info["pricing/balance"], info["pricing/negBalance"] = pb, nb
-		info["pricing/balanceMeta"] = c.balanceMetaInfo
+
+		cb := api.server.clientPool.ndb.getCurrencyBalance(id)
+		info["pricing/currency"] = cb.amount
 		info["priority"] = pb.base != 0
 	} else {
 		info["isConnected"] = false
-		pb := api.server.clientPool.ndb.getOrNewPB(id)
+		pb := api.server.clientPool.ndb.getOrNewBalance(id.Bytes(), false)
-		info["pricing/balance"], info["pricing/balanceMeta"] = pb.value, pb.meta
+
+		cb := api.server.clientPool.ndb.getCurrencyBalance(id)
+		info["pricing/balance"], info["pricing/currency"] = pb.value, cb.amount
 		info["priority"] = pb.value.base != 0
 	}
 	return info
@@ -172,8 +176,8 @@ func (api *PrivateLightServerAPI) setParams(params map[string]interface{}, clien
 
 // AddBalance updates the balance of a client (either overwrites it or adds to it).
 // It also updates the balance meta info string.
-func (api *PrivateLightServerAPI) AddBalance(id enode.ID, value int64, meta string) ([2]uint64, error) {
+func (api *PrivateLightServerAPI) AddBalance(id enode.ID, value int64) ([2]uint64, error) {
-	oldBalance, newBalance, err := api.server.clientPool.addBalance(id, value, meta)
+	oldBalance, newBalance, err := api.server.clientPool.addBalance(id, value)
 	return [2]uint64{oldBalance, newBalance}, err
 }
 

les/clientdb.go

@@ -0,0 +1,279 @@
+// Copyright 2020 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 les
+
+import (
+	"bytes"
+	"encoding/binary"
+	"io"
+	"math"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/mclock"
+	"github.com/ethereum/go-ethereum/core/rawdb"
+	"github.com/ethereum/go-ethereum/ethdb"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/p2p/enode"
+	"github.com/ethereum/go-ethereum/rlp"
+	lru "github.com/hashicorp/golang-lru"
+)
+
+const balanceCacheLimit = 8192 // the maximum number of cached items in service token balance queue
+
+// tokenBalance is a wrapper of expiredValue which represents the service token
+// balance of clients. The balance value will decay exponentially over time and
+// can be deleted when the amount is small enough.
+type tokenBalance struct {
+	value expiredValue
+}
+
+// EncodeRLP implements rlp.Encoder
+func (b *tokenBalance) EncodeRLP(w io.Writer) error {
+	return rlp.Encode(w, []interface{}{b.value.base, b.value.exp})
+}
+
+// DecodeRLP implements rlp.Decoder
+func (b *tokenBalance) DecodeRLP(s *rlp.Stream) error {
+	var entry struct {
+		Base, Exp uint64
+	}
+	if err := s.Decode(&entry); err != nil {
+		return err
+	}
+	b.value = expiredValue{base: entry.Base, exp: entry.Exp}
+	return nil
+}
+
+// currencyBalance represents the client's currency balance.
+type currencyBalance struct {
+	amount uint64
+	typ    string
+}
+
+// EncodeRLP implements rlp.Encoder
+func (b *currencyBalance) EncodeRLP(w io.Writer) error {
+	return rlp.Encode(w, []interface{}{b.amount, b.typ})
+}
+
+// DecodeRLP implements rlp.Decoder
+func (b *currencyBalance) DecodeRLP(s *rlp.Stream) error {
+	var entry struct {
+		Amount uint64
+		Type   string
+	}
+	if err := s.Decode(&entry); err != nil {
+		return err
+	}
+	b.amount, b.typ = entry.Amount, entry.Type
+	return nil
+}
+
+const (
+	// nodeDBVersion is the version identifier of the node data in db
+	//
+	// Changelog:
+	// * Replace `lastTotal` with `meta` in positive balance: version 0=>1
+	// * Rework balance, add currency balance: version 1=>2
+	nodeDBVersion = 2
+
+	// dbCleanupCycle is the cycle of db for useless data cleanup
+	dbCleanupCycle = time.Hour
+)
+
+var (
+	posBalancePrefix = []byte("pb:")         // dbVersion(uint16 big endian) + posBalancePrefix + id -> positive balance
+	negBalancePrefix = []byte("nb:")         // dbVersion(uint16 big endian) + negBalancePrefix + ip -> negative balance
+	curBalancePrefix = []byte("cb:")         // dbVersion(uint16 big endian) + curBalancePrefix + id -> currency balance
+	expirationKey    = []byte("expiration:") // dbVersion(uint16 big endian) + expirationKey -> posExp, negExp
+)
+
+type nodeDB struct {
+	db            ethdb.Database
+	cache         *lru.Cache
+	clock         mclock.Clock
+	closeCh       chan struct{}
+	evictCallBack func(mclock.AbsTime, bool, tokenBalance) bool // Callback to determine whether the balance can be evicted.
+	cleanupHook   func()                                        // Test hook used for testing
+}
+
+func newNodeDB(db ethdb.Database, clock mclock.Clock) *nodeDB {
+	var buff [2]byte
+	binary.BigEndian.PutUint16(buff[:], uint16(nodeDBVersion))
+
+	cache, _ := lru.New(balanceCacheLimit)
+	ndb := &nodeDB{
+		db:      rawdb.NewTable(db, string(buff[:])),
+		cache:   cache,
+		clock:   clock,
+		closeCh: make(chan struct{}),
+	}
+	go ndb.expirer()
+	return ndb
+}
+
+func (db *nodeDB) close() {
+	close(db.closeCh)
+}
+
+func (db *nodeDB) key(id []byte, neg bool) []byte {
+	prefix := posBalancePrefix
+	if neg {
+		prefix = negBalancePrefix
+	}
+	return append(prefix, id...)
+}
+
+func (db *nodeDB) getExpiration() (float64, float64) {
+	blob, err := db.db.Get(expirationKey)
+	if err != nil || len(blob) != 16 {
+		return 0, 0
+	}
+	return math.Float64frombits(binary.BigEndian.Uint64(blob[:8])), math.Float64frombits(binary.BigEndian.Uint64(blob[8:16]))
+}
+
+func (db *nodeDB) setExpiration(pos, neg float64) {
+	var buff [16]byte
+	binary.BigEndian.PutUint64(buff[:8], math.Float64bits(pos))
+	binary.BigEndian.PutUint64(buff[8:16], math.Float64bits(neg))
+	db.db.Put(expirationKey, buff[:16])
+}
+
+func (db *nodeDB) getCurrencyBalance(id enode.ID) currencyBalance {
+	var b currencyBalance
+	enc, err := db.db.Get(append(curBalancePrefix, id.Bytes()...))
+	if err != nil || len(enc) == 0 {
+		return b
+	}
+	if err := rlp.DecodeBytes(enc, &b); err != nil {
+		log.Crit("Failed to decode positive balance", "err", err)
+	}
+	return b
+}
+
+func (db *nodeDB) setCurrencyBalance(id enode.ID, b currencyBalance) {
+	enc, err := rlp.EncodeToBytes(&(b))
+	if err != nil {
+		log.Crit("Failed to encode currency balance", "err", err)
+	}
+	db.db.Put(append(curBalancePrefix, id.Bytes()...), enc)
+}
+
+func (db *nodeDB) getOrNewBalance(id []byte, neg bool) tokenBalance {
+	key := db.key(id, neg)
+	item, exist := db.cache.Get(string(key))
+	if exist {
+		return item.(tokenBalance)
+	}
+	var b tokenBalance
+	enc, err := db.db.Get(key)
+	if err != nil || len(enc) == 0 {
+		return b
+	}
+	if err := rlp.DecodeBytes(enc, &b); err != nil {
+		log.Crit("Failed to decode positive balance", "err", err)
+	}
+	db.cache.Add(string(key), b)
+	return b
+}
+
+func (db *nodeDB) setBalance(id []byte, neg bool, b tokenBalance) {
+	key := db.key(id, neg)
+	enc, err := rlp.EncodeToBytes(&(b))
+	if err != nil {
+		log.Crit("Failed to encode positive balance", "err", err)
+	}
+	db.db.Put(key, enc)
+	db.cache.Add(string(key), b)
+}
+
+func (db *nodeDB) delBalance(id []byte, neg bool) {
+	key := db.key(id, neg)
+	db.db.Delete(key)
+	db.cache.Remove(string(key))
+}
+
+// getPosBalanceIDs returns a lexicographically ordered list of IDs of accounts
+// with a positive balance
+func (db *nodeDB) getPosBalanceIDs(start, stop enode.ID, maxCount int) (result []enode.ID) {
+	if maxCount <= 0 {
+		return
+	}
+	it := db.db.NewIteratorWithStart(db.key(start.Bytes(), false))
+	defer it.Release()
+	for i := len(stop[:]) - 1; i >= 0; i-- {
+		stop[i]--
+		if stop[i] != 255 {
+			break
+		}
+	}
+	stopKey := db.key(stop.Bytes(), false)
+	keyLen := len(stopKey)
+
+	for it.Next() {
+		var id enode.ID
+		if len(it.Key()) != keyLen || bytes.Compare(it.Key(), stopKey) == 1 {
+			return
+		}
+		copy(id[:], it.Key()[keyLen-len(id):])
+		result = append(result, id)
+		if len(result) == maxCount {
+			return
+		}
+	}
+	return
+}
+
+func (db *nodeDB) expirer() {
+	for {
+		select {
+		case <-db.clock.After(dbCleanupCycle):
+			db.expireNodes()
+		case <-db.closeCh:
+			return
+		}
+	}
+}
+
+// expireNodes iterates the whole node db and checks whether the
+// token balances can deleted.
+func (db *nodeDB) expireNodes() {
+	var (
+		visited int
+		deleted int
+		start   = time.Now()
+	)
+	for index, prefix := range [][]byte{posBalancePrefix, negBalancePrefix} {
+		iter := db.db.NewIteratorWithPrefix(prefix)
+		for iter.Next() {
+			visited += 1
+			var balance tokenBalance
+			if err := rlp.DecodeBytes(iter.Value(), &balance); err != nil {
+				log.Crit("Failed to decode negative balance", "err", err)
+			}
+			if db.evictCallBack != nil && db.evictCallBack(db.clock.Now(), index != 0, balance) {
+				deleted += 1
+				db.db.Delete(iter.Key())
+			}
+		}
+	}
+	// Invoke testing hook if it's not nil.
+	if db.cleanupHook != nil {
+		db.cleanupHook()
+	}
+	log.Debug("Expire nodes", "visited", visited, "deleted", deleted, "elapsed", common.PrettyDuration(time.Since(start)))
+}

les/clientdb_test.go

@@ -0,0 +1,139 @@
+// Copyright 2020 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 les
+
+import (
+	"reflect"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common/mclock"
+	"github.com/ethereum/go-ethereum/core/rawdb"
+	"github.com/ethereum/go-ethereum/p2p/enode"
+)
+
+func TestNodeDB(t *testing.T) {
+	ndb := newNodeDB(rawdb.NewMemoryDatabase(), mclock.System{})
+	defer ndb.close()
+
+	var cases = []struct {
+		id       enode.ID
+		ip       string
+		balance  tokenBalance
+		positive bool
+	}{
+		{enode.ID{0x00, 0x01, 0x02}, "", tokenBalance{value: expval(100)}, true},
+		{enode.ID{0x00, 0x01, 0x02}, "", tokenBalance{value: expval(200)}, true},
+		{enode.ID{}, "127.0.0.1", tokenBalance{value: expval(100)}, false},
+		{enode.ID{}, "127.0.0.1", tokenBalance{value: expval(200)}, false},
+	}
+	for _, c := range cases {
+		if c.positive {
+			ndb.setBalance(c.id.Bytes(), false, c.balance)
+			if pb := ndb.getOrNewBalance(c.id.Bytes(), false); !reflect.DeepEqual(pb, c.balance) {
+				t.Fatalf("Positive balance mismatch, want %v, got %v", c.balance, pb)
+			}
+		} else {
+			ndb.setBalance([]byte(c.ip), true, c.balance)
+			if nb := ndb.getOrNewBalance([]byte(c.ip), true); !reflect.DeepEqual(nb, c.balance) {
+				t.Fatalf("Negative balance mismatch, want %v, got %v", c.balance, nb)
+			}
+		}
+	}
+	for _, c := range cases {
+		if c.positive {
+			ndb.delBalance(c.id.Bytes(), false)
+			if pb := ndb.getOrNewBalance(c.id.Bytes(), false); !reflect.DeepEqual(pb, tokenBalance{}) {
+				t.Fatalf("Positive balance mismatch, want %v, got %v", tokenBalance{}, pb)
+			}
+		} else {
+			ndb.delBalance([]byte(c.ip), true)
+			if nb := ndb.getOrNewBalance([]byte(c.ip), true); !reflect.DeepEqual(nb, tokenBalance{}) {
+				t.Fatalf("Negative balance mismatch, want %v, got %v", tokenBalance{}, nb)
+			}
+		}
+	}
+	posExp, negExp := 100.1234, 200.5678
+	ndb.setExpiration(posExp, negExp)
+	if pos, neg := ndb.getExpiration(); pos != posExp || neg != negExp {
+		t.Fatalf("Expiration mismatch, want %v / %v, got %v / %v", posExp, negExp, pos, neg)
+	}
+	curBalance := currencyBalance{typ: "ETH", amount: 10000}
+	ndb.setCurrencyBalance(enode.ID{0x01, 0x02}, curBalance)
+	if got := ndb.getCurrencyBalance(enode.ID{0x01, 0x02}); !reflect.DeepEqual(got, curBalance) {
+		t.Fatalf("Currency balance mismatch, want %v, got %v", curBalance, got)
+	}
+}
+
+func TestNodeDBExpiration(t *testing.T) {
+	var (
+		iterated int
+		done     = make(chan struct{}, 1)
+	)
+	callback := func(now mclock.AbsTime, neg bool, b tokenBalance) bool {
+		iterated += 1
+		return true
+	}
+	clock := &mclock.Simulated{}
+	ndb := newNodeDB(rawdb.NewMemoryDatabase(), clock)
+	defer ndb.close()
+	ndb.evictCallBack = callback
+	ndb.cleanupHook = func() { done <- struct{}{} }
+
+	var cases = []struct {
+		id      []byte
+		neg     bool
+		balance tokenBalance
+	}{
+		{[]byte{0x01, 0x02}, false, tokenBalance{value: expval(1)}},
+		{[]byte{0x03, 0x04}, false, tokenBalance{value: expval(1)}},
+		{[]byte{0x05, 0x06}, false, tokenBalance{value: expval(1)}},
+		{[]byte{0x07, 0x08}, false, tokenBalance{value: expval(1)}},
+
+		{[]byte("127.0.0.1"), true, tokenBalance{value: expval(1)}},
+		{[]byte("127.0.0.2"), true, tokenBalance{value: expval(1)}},
+		{[]byte("127.0.0.3"), true, tokenBalance{value: expval(1)}},
+		{[]byte("127.0.0.4"), true, tokenBalance{value: expval(1)}},
+	}
+	for _, c := range cases {
+		ndb.setBalance(c.id, c.neg, c.balance)
+	}
+	clock.WaitForTimers(1)
+	clock.Run(time.Hour + time.Minute)
+	select {
+	case <-done:
+	case <-time.NewTimer(time.Second).C:
+		t.Fatalf("timeout")
+	}
+	if iterated != 8 {
+		t.Fatalf("Failed to evict useless balances, want %v, got %d", 8, iterated)
+	}
+
+	for _, c := range cases {
+		ndb.setBalance(c.id, c.neg, c.balance)
+	}
+	clock.WaitForTimers(1)
+	clock.Run(time.Hour + time.Minute)
+	select {
+	case <-done:
+	case <-time.NewTimer(time.Second).C:
+		t.Fatalf("timeout")
+	}
+	if iterated != 16 {
+		t.Fatalf("Failed to evict useless balances, want %v, got %d", 16, iterated)
+	}
+}

les/clientpool.go

@@ -17,22 +17,16 @@
 package les
 
 import (
-	"bytes"
-	"encoding/binary"
 	"fmt"
-	"io"
 	"math"
 	"sync"
 	"time"
 
-	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/mclock"
 	"github.com/ethereum/go-ethereum/common/prque"
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p/enode"
-	"github.com/ethereum/go-ethereum/rlp"
-	lru "github.com/hashicorp/golang-lru"
 )
 
 const (
@@ -42,8 +36,6 @@ const (
 	tryActivatePeriod        = time.Second * 5  // periodically check whether inactive clients can be activated
 	dropInactiveCycles       = 2                // number of activation check periods after non-priority inactive peers are dropped
 	persistExpirationRefresh = time.Minute * 5  // refresh period of the token expiration persistence
-	posBalanceCacheLimit     = 8192             // the maximum number of cached items in positive balance queue
-	negBalanceCacheLimit     = 8192             // the maximum number of cached items in negative balance queue
 	freeRatioTC              = time.Hour        // time constant of token supply control based on free service availability
 
 	// activeBias is applied to already connected clients So that
@@ -72,11 +64,8 @@ const (
 // each client can have several minutes of connection time.
 //
 // Balances of disconnected clients are stored in nodeDB including positive balance
-// and negative banalce. Negative balance is transformed into a logarithmic form
+// and negative banalce. Boeth positive balance and negative balance will decrease
-// with a constantly shifting linear offset in order to implement an exponential
+// exponentially. If the balance is low enough, then the record will be dropped.
-// decrease. Besides nodeDB will have a background thread to check the negative
-// balance of disconnected client. If the balance is low enough, then the record
-// will be dropped.
 type clientPool struct {
 	ndb        *nodeDB
 	lock       sync.Mutex
@@ -138,7 +127,6 @@ type clientInfo struct {
 	queueIndex             int // position in activeQueue
 	balanceTracker         balanceTracker
 	posFactors, negFactors priceFactors
-	balanceMetaInfo        string
 }
 
 // connSetIndex callback updates clientInfo item index in activeQueue
@@ -212,7 +200,7 @@ func newClientPool(db ethdb.Database, minCap, freeClientCap uint64, clock mclock
 			stop = true
 		}
 		for i := 0; i < l; i++ {
-			pool.inactiveBalances.addExp(pool.ndb.getOrNewPB(ids[i]).value)
+			pool.inactiveBalances.addExp(pool.ndb.getOrNewBalance(ids[i].Bytes(), false).value)
 		}
 		if stop {
 			break
@@ -220,8 +208,14 @@ func newClientPool(db ethdb.Database, minCap, freeClientCap uint64, clock mclock
 	}
 	// If the negative balance of free client is low enough,
 	// delete this entry.
-	ndb.nbEvictCallBack = func(now mclock.AbsTime, b negBalance) bool {
+	ndb.evictCallBack = func(now mclock.AbsTime, neg bool, b tokenBalance) bool {
-		return b.value.value(pool.negExpiration(now)) < uint64(time.Second)
+		var expiration float64
+		if neg {
+			expiration = pool.negExpiration(now)
+		} else {
+			expiration = pool.posExpiration(now)
+		}
+		return b.value.value(expiration) < uint64(time.Second)
 	}
 	go func() {
 		for {
@@ -239,11 +233,9 @@ func newClientPool(db ethdb.Database, minCap, freeClientCap uint64, clock mclock
 		for {
 			select {
 			case <-clock.After(persistExpirationRefresh):
-				pool.lock.Lock()
 				now := pool.clock.Now()
 				posExp := pool.posExpiration(now)
 				negExp := pool.negExpiration(now)
-				pool.lock.Unlock()
 				pool.ndb.setExpiration(posExp, negExp)
 			case <-pool.stopCh:
 				return
@@ -412,21 +404,20 @@ func (f *clientPool) connect(peer clientPoolPeer, reqCapacity uint64) (uint64, e
 		log.Debug("Client already connected", "address", freeID, "id", peerIdToString(id))
 		return 0, fmt.Errorf("Client already connected   address = %s  id = %s", freeID, peerIdToString(id))
 	}
-	pb := f.ndb.getOrNewPB(id)
+	pb := f.ndb.getOrNewBalance(id.Bytes(), false)
-	nb := f.ndb.getOrNewNB(freeID)
+	nb := f.ndb.getOrNewBalance([]byte(freeID), true)
 	e := &clientInfo{
-		capacity:        reqCapacity,
+		capacity:    reqCapacity,
-		pool:            f,
+		pool:        f,
-		peer:            peer,
+		peer:        peer,
-		address:         freeID,
+		address:     freeID,
-		queueIndex:      -1,
+		queueIndex:  -1,
-		id:              id,
+		id:          id,
-		freeID:          freeID,
+		freeID:      freeID,
-		connectedAt:     f.clock.Now(),
+		connectedAt: f.clock.Now(),
-		priority:        pb.value.base != 0,
+		priority:    pb.value.base != 0,
-		posFactors:      f.defaultPosFactors,
+		posFactors:  f.defaultPosFactors,
-		negFactors:      f.defaultNegFactors,
+		negFactors:  f.defaultNegFactors,
-		balanceMetaInfo: pb.meta,
 	}
 	missing, capacity := f.capAvailable(id, freeID, reqCapacity, 0, true)
 	f.connectedMap[id] = e
@@ -460,7 +451,7 @@ func (f *clientPool) connect(peer clientPoolPeer, reqCapacity uint64) (uint64, e
 }
 
 // initBalanceTracker initializes the positive and negative balances and price factors
-func (f *clientPool) initBalanceTracker(bt *balanceTracker, pb posBalance, nb negBalance, capacity uint64, active bool) {
+func (f *clientPool) initBalanceTracker(bt *balanceTracker, pb tokenBalance, nb tokenBalance, capacity uint64, active bool) {
 	bt.exp = f
 	bt.init(f.clock, capacity)
 	bt.setBalance(pb.value, nb.value)
@@ -552,7 +543,7 @@ func (f *clientPool) capAvailable(id enode.ID, freeID string, capacity uint64, m
 				bt = &client.balanceTracker
 			} else {
 				bt = &balanceTracker{}
-				f.initBalanceTracker(bt, f.ndb.getOrNewPB(id), f.ndb.getOrNewNB(freeID), capacity, true)
+				f.initBalanceTracker(bt, f.ndb.getOrNewBalance(id.Bytes(), false), f.ndb.getOrNewBalance([]byte(freeID), true), capacity, true)
 			}
 			if capacity != f.freeClientCap && targetPriority >= 0 {
 				targetPriority = -1
@@ -687,15 +678,24 @@ func (f *clientPool) finalizeBalance(c *clientInfo, now mclock.AbsTime) {
 	f.inactiveBalances.addExp(pos)
 	f.activeBalances.subExp(pos)
 
-	pb, nb := f.ndb.getOrNewPB(c.id), f.ndb.getOrNewNB(c.address)
+	for index, value := range []expiredValue{pos, neg} {
-	pb.value = pos
+		var (
-	f.ndb.setPB(c.id, pb)
+			id         []byte
-
+			expiration float64
-	if neg.value(f.negExpiration(f.clock.Now())) > uint64(time.Second) {
+		)
-		nb.value = neg
+		neg := index == 1
-		f.ndb.setNB(c.address, nb)
+		if !neg {
-	} else {
+			id = c.id.Bytes()
-		f.ndb.delNB(c.address) // Negative balance is small enough, drop it directly.
+			expiration = f.posExpiration(f.clock.Now())
+		} else {
+			id = []byte(c.address)
+			expiration = f.negExpiration(f.clock.Now())
+		}
+		if value.value(expiration) > uint64(time.Second) {
+			f.ndb.setBalance(id, neg, tokenBalance{value: value})
+		} else {
+			f.ndb.delBalance(id, neg) // balance is small enough, drop it directly.
+		}
 	}
 }
 
@@ -721,9 +721,7 @@ func (f *clientPool) balanceExhausted(id enode.ID) {
 		c.balanceTracker.setCapacity(c.capacity)
 		c.peer.updateCapacity(c.capacity)
 	}
-	pb := f.ndb.getOrNewPB(id)
+	f.ndb.delBalance(id.Bytes(), false)
-	pb.value = expiredValue{}
-	f.ndb.setPB(id, pb)
 }
 
 // setactiveLimit sets the maximum number and total capacity of connected clients,
@@ -811,26 +809,26 @@ func zeroPriceFactors(bt *balanceTracker) {
 }
 
 // getPosBalance retrieves a single positive balance entry from cache or the database
-func (f *clientPool) getPosBalance(id enode.ID) posBalance {
+func (f *clientPool) getPosBalance(id enode.ID) tokenBalance {
 	f.lock.Lock()
 	defer f.lock.Unlock()
 
 	if c := f.connectedMap[id]; c != nil {
-		pb, _ := c.balanceTracker.getBalance(f.clock.Now())
+		value, _ := c.balanceTracker.getBalance(f.clock.Now())
-		return posBalance{value: pb, meta: c.balanceMetaInfo}
+		return tokenBalance{value: value}
 	} else {
-		return f.ndb.getOrNewPB(id)
+		return f.ndb.getOrNewBalance(id.Bytes(), false)
 	}
 }
 
 // addBalance updates the balance of a client (either overwrites it or adds to it).
 // It also updates the balance meta info string.
-func (f *clientPool) addBalance(id enode.ID, amount int64, meta string) (uint64, uint64, error) {
+func (f *clientPool) addBalance(id enode.ID, amount int64) (uint64, uint64, error) {
 	f.lock.Lock()
 	defer f.lock.Unlock()
 
 	now := f.clock.Now()
-	pb := f.ndb.getOrNewPB(id)
+	pb := f.ndb.getOrNewBalance(id.Bytes(), false)
 	var negBalance expiredValue
 	c := f.connectedMap[id]
 	if c != nil {
@@ -843,8 +841,7 @@ func (f *clientPool) addBalance(id enode.ID, amount int64, meta string) (uint64,
 		return oldValue, oldValue, errBalanceOverflow
 	}
 	pb.value.add(amount, posExp)
-	pb.meta = meta
+	f.ndb.setBalance(id.Bytes(), false, pb)
-	f.ndb.setPB(id, pb)
 	if c != nil {
 		c.balanceTracker.setBalance(pb.value, negBalance)
 		if c.active {
@@ -857,7 +854,6 @@ func (f *clientPool) addBalance(id enode.ID, amount int64, meta string) (uint64,
 				f.updateFreeRatio()
 				c.balanceTracker.addCallback(balanceCallbackZero, 0, func() { f.balanceExhausted(id) })
 			}
-			c.balanceMetaInfo = meta
 			f.activeBalances.subExp(oldBalance)
 			f.activeBalances.addExp(pb.value)
 		} else {
@@ -878,268 +874,3 @@ func (f *clientPool) addBalance(id enode.ID, amount int64, meta string) (uint64,
 	f.tryActivateClients()
 	return oldValue, pb.value.value(posExp), nil
 }
-
-// posBalance represents a recently accessed positive balance entry
-type posBalance struct {
-	value expiredValue
-	meta  string
-}
-
-// EncodeRLP implements rlp.Encoder
-func (e *posBalance) EncodeRLP(w io.Writer) error {
-	return rlp.Encode(w, []interface{}{e.value.base, e.value.exp, e.meta})
-}
-
-// DecodeRLP implements rlp.Decoder
-func (e *posBalance) DecodeRLP(s *rlp.Stream) error {
-	var entry struct {
-		ValueBase, ValueExp uint64
-		Meta                string
-	}
-	if err := s.Decode(&entry); err != nil {
-		return err
-	}
-	e.value = expiredValue{base: entry.ValueBase, exp: entry.ValueExp}
-	e.meta = entry.Meta
-	return nil
-}
-
-// negBalance represents a negative balance entry of a disconnected client
-type negBalance struct{ value expiredValue }
-
-// EncodeRLP implements rlp.Encoder
-func (e *negBalance) EncodeRLP(w io.Writer) error {
-	return rlp.Encode(w, []interface{}{e.value.base, e.value.exp})
-}
-
-// DecodeRLP implements rlp.Decoder
-func (e *negBalance) DecodeRLP(s *rlp.Stream) error {
-	var entry struct {
-		Base, Exp uint64
-	}
-	if err := s.Decode(&entry); err != nil {
-		return err
-	}
-	e.value = expiredValue{base: entry.Base, exp: entry.Exp}
-	return nil
-}
-
-const (
-	// nodeDBVersion is the version identifier of the node data in db
-	//
-	// Changelog:
-	// * Replace `lastTotal` with `meta` in positive balance: version 0=>1
-	nodeDBVersion = 1
-
-	// dbCleanupCycle is the cycle of db for useless data cleanup
-	dbCleanupCycle = time.Hour
-)
-
-var (
-	positiveBalancePrefix = []byte("pb:")         // dbVersion(uint16 big endian) + positiveBalancePrefix + id -> balance
-	negativeBalancePrefix = []byte("nb:")         // dbVersion(uint16 big endian) + negativeBalancePrefix + ip -> balance
-	expirationKey         = []byte("expiration:") // dbVersion(uint16 big endian) + expirationKey -> posExp, negExp
-)
-
-type nodeDB struct {
-	db              ethdb.Database
-	pcache          *lru.Cache
-	ncache          *lru.Cache
-	auxbuf          []byte                                // 37-byte auxiliary buffer for key encoding
-	verbuf          [2]byte                               // 2-byte auxiliary buffer for db version
-	nbEvictCallBack func(mclock.AbsTime, negBalance) bool // Callback to determine whether the negative balance can be evicted.
-	clock           mclock.Clock
-	closeCh         chan struct{}
-	cleanupHook     func() // Test hook used for testing
-}
-
-func newNodeDB(db ethdb.Database, clock mclock.Clock) *nodeDB {
-	pcache, _ := lru.New(posBalanceCacheLimit)
-	ncache, _ := lru.New(negBalanceCacheLimit)
-	ndb := &nodeDB{
-		db:      db,
-		pcache:  pcache,
-		ncache:  ncache,
-		auxbuf:  make([]byte, 37),
-		clock:   clock,
-		closeCh: make(chan struct{}),
-	}
-	binary.BigEndian.PutUint16(ndb.verbuf[:], uint16(nodeDBVersion))
-	go ndb.expirer()
-	return ndb
-}
-
-func (db *nodeDB) close() {
-	close(db.closeCh)
-}
-
-func (db *nodeDB) key(id []byte, neg bool) []byte {
-	prefix := positiveBalancePrefix
-	if neg {
-		prefix = negativeBalancePrefix
-	}
-	if len(prefix)+len(db.verbuf)+len(id) > len(db.auxbuf) {
-		db.auxbuf = append(db.auxbuf, make([]byte, len(prefix)+len(db.verbuf)+len(id)-len(db.auxbuf))...)
-	}
-	copy(db.auxbuf[:len(db.verbuf)], db.verbuf[:])
-	copy(db.auxbuf[len(db.verbuf):len(db.verbuf)+len(prefix)], prefix)
-	copy(db.auxbuf[len(prefix)+len(db.verbuf):len(prefix)+len(db.verbuf)+len(id)], id)
-	return db.auxbuf[:len(prefix)+len(db.verbuf)+len(id)]
-}
-
-func (db *nodeDB) getExpiration() (float64, float64) {
-	blob, err := db.db.Get(append(expirationKey, db.verbuf[:]...))
-	if err != nil || len(blob) != 16 {
-		return 0, 0
-	}
-	return math.Float64frombits(binary.BigEndian.Uint64(blob[:8])), math.Float64frombits(binary.BigEndian.Uint64(blob[8:16]))
-}
-
-func (db *nodeDB) setExpiration(pos, neg float64) {
-	binary.BigEndian.PutUint64(db.auxbuf[:8], math.Float64bits(pos))
-	binary.BigEndian.PutUint64(db.auxbuf[8:16], math.Float64bits(neg))
-	db.db.Put(append(expirationKey, db.verbuf[:]...), db.auxbuf[:16])
-}
-
-func (db *nodeDB) getOrNewPB(id enode.ID) posBalance {
-	key := db.key(id.Bytes(), false)
-	item, exist := db.pcache.Get(string(key))
-	if exist {
-		return item.(posBalance)
-	}
-	var balance posBalance
-	if enc, err := db.db.Get(key); err == nil {
-		if err := rlp.DecodeBytes(enc, &balance); err != nil {
-			log.Error("Failed to decode positive balance", "err", err)
-		}
-	}
-	db.pcache.Add(string(key), balance)
-	return balance
-}
-
-func (db *nodeDB) setPB(id enode.ID, b posBalance) {
-	if b.value.base == 0 && len(b.meta) == 0 {
-		db.delPB(id)
-		return
-	}
-	key := db.key(id.Bytes(), false)
-	enc, err := rlp.EncodeToBytes(&(b))
-	if err != nil {
-		log.Error("Failed to encode positive balance", "err", err)
-		return
-	}
-	db.db.Put(key, enc)
-	db.pcache.Add(string(key), b)
-}
-
-func (db *nodeDB) delPB(id enode.ID) {
-	key := db.key(id.Bytes(), false)
-	db.db.Delete(key)
-	db.pcache.Remove(string(key))
-}
-
-// getPosBalanceIDs returns a lexicographically ordered list of IDs of accounts
-// with a positive balance
-func (db *nodeDB) getPosBalanceIDs(start, stop enode.ID, maxCount int) (result []enode.ID) {
-	if maxCount <= 0 {
-		return
-	}
-	it := db.db.NewIteratorWithStart(db.key(start.Bytes(), false))
-	defer it.Release()
-	for i := len(stop[:]) - 1; i >= 0; i-- {
-		stop[i]--
-		if stop[i] != 255 {
-			break
-		}
-	}
-	stopKey := db.key(stop.Bytes(), false)
-	keyLen := len(stopKey)
-
-	for it.Next() {
-		var id enode.ID
-		if len(it.Key()) != keyLen || bytes.Compare(it.Key(), stopKey) == 1 {
-			return
-		}
-		copy(id[:], it.Key()[keyLen-len(id):])
-		result = append(result, id)
-		if len(result) == maxCount {
-			return
-		}
-	}
-	return
-}
-
-func (db *nodeDB) getOrNewNB(id string) negBalance {
-	key := db.key([]byte(id), true)
-	item, exist := db.ncache.Get(string(key))
-	if exist {
-		return item.(negBalance)
-	}
-	var balance negBalance
-	if enc, err := db.db.Get(key); err == nil {
-		if err := rlp.DecodeBytes(enc, &balance); err != nil {
-			log.Error("Failed to decode negative balance", "err", err)
-		}
-	}
-	db.ncache.Add(string(key), balance)
-	return balance
-}
-
-func (db *nodeDB) setNB(id string, b negBalance) {
-	key := db.key([]byte(id), true)
-	enc, err := rlp.EncodeToBytes(&(b))
-	if err != nil {
-		log.Error("Failed to encode negative balance", "err", err)
-		return
-	}
-	db.db.Put(key, enc)
-	db.ncache.Add(string(key), b)
-}
-
-func (db *nodeDB) delNB(id string) {
-	key := db.key([]byte(id), true)
-	db.db.Delete(key)
-	db.ncache.Remove(string(key))
-}
-
-func (db *nodeDB) expirer() {
-	for {
-		select {
-		case <-db.clock.After(dbCleanupCycle):
-			db.expireNodes()
-		case <-db.closeCh:
-			return
-		}
-	}
-}
-
-// expireNodes iterates the whole node db and checks whether the negative balance
-// entry can deleted.
-//
-// The rationale behind this is: server doesn't need to keep the negative balance
-// records if they are low enough.
-func (db *nodeDB) expireNodes() {
-	var (
-		visited int
-		deleted int
-		start   = time.Now()
-	)
-	iter := db.db.NewIteratorWithPrefix(append(db.verbuf[:], negativeBalancePrefix...))
-	for iter.Next() {
-		visited += 1
-		var balance negBalance
-		if err := rlp.DecodeBytes(iter.Value(), &balance); err != nil {
-			log.Error("Failed to decode negative balance", "err", err)
-			continue
-		}
-		if db.nbEvictCallBack != nil && db.nbEvictCallBack(db.clock.Now(), balance) {
-			deleted += 1
-			db.db.Delete(iter.Key())
-		}
-	}
-	// Invoke testing hook if it's not nil.
-	if db.cleanupHook != nil {
-		db.cleanupHook()
-	}
-	log.Debug("Expire nodes", "visited", visited, "deleted", deleted, "elapsed", common.PrettyDuration(time.Since(start)))
-}

les/clientpool_test.go

@@ -17,10 +17,8 @@
 package les
 
 import (
-	"bytes"
 	"fmt"
 	"math/rand"
-	"reflect"
 	"testing"
 	"time"
 
@@ -116,7 +114,7 @@ func testClientPool(t *testing.T, activeLimit, clientCount, paidCount int, rando
 			// give a positive balance to some of the peers
 			amount := testClientPoolTicks / 2 * int64(time.Second) // enough for half of the simulation period
 			for i := 0; i < paidCount; i++ {
-				pool.addBalance(newPoolTestPeer(i, disconnCh).ID(), amount, "")
+				pool.addBalance(newPoolTestPeer(i, disconnCh).ID(), amount)
 			}
 		}
 
@@ -185,7 +183,7 @@ func TestConnectPaidClient(t *testing.T) {
 	pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1})
 
 	// Add balance for an external client and mark it as paid client
-	pool.addBalance(newPoolTestPeer(0, nil).ID(), 1000, "")
+	pool.addBalance(newPoolTestPeer(0, nil).ID(), 1000)
 
 	if cap, _ := pool.connect(newPoolTestPeer(0, nil), 10); cap == 0 {
 		t.Fatalf("Failed to connect paid client")
@@ -203,7 +201,7 @@ func TestConnectPaidClientToSmallPool(t *testing.T) {
 	pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1})
 
 	// Add balance for an external client and mark it as paid client
-	pool.addBalance(newPoolTestPeer(0, nil).ID(), 1000, "")
+	pool.addBalance(newPoolTestPeer(0, nil).ID(), 1000)
 
 	// Connect a fat paid client to pool, should reject it.
 	if cap, _ := pool.connect(newPoolTestPeer(0, nil), 100); cap != 0 {
@@ -223,15 +221,15 @@ func TestConnectPaidClientToFullPool(t *testing.T) {
 	pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1})
 
 	for i := 0; i < 10; i++ {
-		pool.addBalance(newPoolTestPeer(i, nil).ID(), 1000000000, "")
+		pool.addBalance(newPoolTestPeer(i, nil).ID(), 1000000000)
 		pool.connect(newPoolTestPeer(i, nil), 1)
 	}
-	pool.addBalance(newPoolTestPeer(11, nil).ID(), 1000, "") // Add low balance to new paid client
+	pool.addBalance(newPoolTestPeer(11, nil).ID(), 1000) // Add low balance to new paid client
 	if cap, _ := pool.connect(newPoolTestPeer(11, nil), 1); cap != 0 {
 		t.Fatalf("Low balance paid client should be rejected")
 	}
 	clock.Run(time.Second)
-	pool.addBalance(newPoolTestPeer(12, nil).ID(), 1000000000*60*3+1, "") // Add high balance to new paid client
+	pool.addBalance(newPoolTestPeer(12, nil).ID(), 1000000000*60*3+1) // Add high balance to new paid client
 	if cap, _ := pool.connect(newPoolTestPeer(12, nil), 1); cap == 0 {
 		t.Fatalf("High balance paid client should be accepted")
 	}
@@ -250,7 +248,7 @@ func TestPaidClientKickedOut(t *testing.T) {
 	pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1})
 
 	for i := 0; i < 10; i++ {
-		pool.addBalance(newPoolTestPeer(i, kickedCh).ID(), 1000000000, "") // 1 second allowance
+		pool.addBalance(newPoolTestPeer(i, kickedCh).ID(), 1000000000) // 1 second allowance
 		pool.connect(newPoolTestPeer(i, kickedCh), 1)
 		clock.Run(time.Millisecond)
 	}
@@ -359,12 +357,12 @@ func TestPositiveBalanceCalculation(t *testing.T) {
 	pool.setLimits(10, uint64(10)) // Total capacity limit is 10
 	pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1})
 
-	pool.addBalance(newPoolTestPeer(0, kicked).ID(), int64(time.Minute*3), "")
+	pool.addBalance(newPoolTestPeer(0, kicked).ID(), int64(time.Minute*3))
 	pool.connect(newPoolTestPeer(0, kicked), 10)
 	clock.Run(time.Minute)
 
 	pool.disconnect(newPoolTestPeer(0, kicked))
-	pb := pool.ndb.getOrNewPB(newPoolTestPeer(0, kicked).ID())
+	pb := pool.ndb.getOrNewBalance(newPoolTestPeer(0, kicked).ID().Bytes(), false)
 	if pb.value != expval(uint64(time.Minute*2)) {
 		t.Fatalf("Positive balance mismatch, want %v, got %v", uint64(time.Minute*2), pb.value)
 	}
@@ -383,7 +381,7 @@ func TestDowngradePriorityClient(t *testing.T) {
 	pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1})
 
 	p := newPoolTestPeer(0, kicked)
-	pool.addBalance(p.ID(), int64(time.Minute), "")
+	pool.addBalance(p.ID(), int64(time.Minute))
 	p.cap, _ = pool.connect(p, 10)
 	if p.cap != 10 {
 		t.Fatalf("The capcacity of priority peer hasn't been updated, got: %d", p.cap)
@@ -394,13 +392,13 @@ func TestDowngradePriorityClient(t *testing.T) {
 	if p.cap != 1 {
 		t.Fatalf("The capcacity of peer should be downgraded, got: %d", p.cap)
 	}
-	pb := pool.ndb.getOrNewPB(newPoolTestPeer(0, kicked).ID())
+	pb := pool.ndb.getOrNewBalance(newPoolTestPeer(0, kicked).ID().Bytes(), false)
 	if pb.value.base != 0 {
 		t.Fatalf("Positive balance mismatch, want %v, got %v", 0, pb.value)
 	}
 
-	pool.addBalance(newPoolTestPeer(0, kicked).ID(), int64(time.Minute), "")
+	pool.addBalance(newPoolTestPeer(0, kicked).ID(), int64(time.Minute))
-	pb = pool.ndb.getOrNewPB(newPoolTestPeer(0, kicked).ID())
+	pb = pool.ndb.getOrNewBalance(newPoolTestPeer(0, kicked).ID().Bytes(), false)
 	if pb.value != expval(uint64(time.Minute)) {
 		t.Fatalf("Positive balance mismatch, want %v, got %v", uint64(time.Minute), pb.value)
 	}
@@ -423,8 +421,8 @@ func TestNegativeBalanceCalculation(t *testing.T) {
 
 	for i := 0; i < 10; i++ {
 		pool.disconnect(newPoolTestPeer(i, nil))
-		nb := pool.ndb.getOrNewNB(newPoolTestPeer(i, nil).freeClientId())
+		nb := pool.ndb.getOrNewBalance([]byte(newPoolTestPeer(i, nil).freeClientId()), true)
-		if nb.logValue != 0 {
+		if nb.value.base != 0 {
 			t.Fatalf("Short connection shouldn't be recorded")
 		}
 	}
@@ -443,108 +441,6 @@ func TestNegativeBalanceCalculation(t *testing.T) {
 	}
 }
 
-func TestNodeDB(t *testing.T) {
-	ndb := newNodeDB(rawdb.NewMemoryDatabase(), mclock.System{})
-	defer ndb.close()
-
-	if !bytes.Equal(ndb.verbuf[:], []byte{0x00, nodeDBVersion}) {
-		t.Fatalf("version buffer mismatch, want %v, got %v", []byte{0x00, nodeDBVersion}, ndb.verbuf)
-	}
-	var cases = []struct {
-		id       enode.ID
-		ip       string
-		balance  interface{}
-		positive bool
-	}{
-		{enode.ID{0x00, 0x01, 0x02}, "", posBalance{value: expval(100)}, true},
-		{enode.ID{0x00, 0x01, 0x02}, "", posBalance{value: expval(200)}, true},
-		{enode.ID{}, "127.0.0.1", negBalance{value: expval(100)}, false},
-		{enode.ID{}, "127.0.0.1", negBalance{value: expval(200)}, false},
-	}
-	for _, c := range cases {
-		if c.positive {
-			ndb.setPB(c.id, c.balance.(posBalance))
-			if pb := ndb.getOrNewPB(c.id); !reflect.DeepEqual(pb, c.balance.(posBalance)) {
-				t.Fatalf("Positive balance mismatch, want %v, got %v", c.balance.(posBalance), pb)
-			}
-		} else {
-			ndb.setNB(c.ip, c.balance.(negBalance))
-			if nb := ndb.getOrNewNB(c.ip); !reflect.DeepEqual(nb, c.balance.(negBalance)) {
-				t.Fatalf("Negative balance mismatch, want %v, got %v", c.balance.(negBalance), nb)
-			}
-		}
-	}
-	for _, c := range cases {
-		if c.positive {
-			ndb.delPB(c.id)
-			if pb := ndb.getOrNewPB(c.id); !reflect.DeepEqual(pb, posBalance{}) {
-				t.Fatalf("Positive balance mismatch, want %v, got %v", posBalance{}, pb)
-			}
-		} else {
-			ndb.delNB(c.ip)
-			if nb := ndb.getOrNewNB(c.ip); !reflect.DeepEqual(nb, negBalance{}) {
-				t.Fatalf("Negative balance mismatch, want %v, got %v", negBalance{}, nb)
-			}
-		}
-	}
-	ndb.setExpiration(100, 200)
-	if pos, neg := ndb.getExpiration(); pos != 100 || neg != 200 {
-		t.Fatalf("Expiration mismatch, want %v / %v, got %v / %v", 100, 200, pos, neg)
-	}
-}
-
-func TestNodeDBExpiration(t *testing.T) {
-	var (
-		iterated int
-		done     = make(chan struct{}, 1)
-	)
-	callback := func(now mclock.AbsTime, b negBalance) bool {
-		iterated += 1
-		return true
-	}
-	clock := &mclock.Simulated{}
-	ndb := newNodeDB(rawdb.NewMemoryDatabase(), clock)
-	defer ndb.close()
-	ndb.nbEvictCallBack = callback
-	ndb.cleanupHook = func() { done <- struct{}{} }
-
-	var cases = []struct {
-		ip      string
-		balance negBalance
-	}{
-		{"127.0.0.1", negBalance{value: expiredValue{base: 1}}},
-		{"127.0.0.2", negBalance{value: expiredValue{base: 1}}},
-		{"127.0.0.3", negBalance{value: expiredValue{base: 1}}},
-		{"127.0.0.4", negBalance{value: expiredValue{base: 1}}},
-	}
-	for _, c := range cases {
-		ndb.setNB(c.ip, c.balance)
-	}
-	clock.WaitForTimers(1)
-	clock.Run(time.Hour + time.Minute)
-	select {
-	case <-done:
-	case <-time.NewTimer(time.Second).C:
-		t.Fatalf("timeout")
-	}
-	if iterated != 4 {
-		t.Fatalf("Failed to evict useless negative balances, want %v, got %d", 4, iterated)
-	}
-	clock.WaitForTimers(1)
-	for _, c := range cases {
-		ndb.setNB(c.ip, c.balance)
-	}
-	clock.Run(time.Hour + time.Minute)
-	select {
-	case <-done:
-	case <-time.NewTimer(time.Second).C:
-		t.Fatalf("timeout")
-	}
-	if iterated != 8 {
-		t.Fatalf("Failed to evict useless negative balances, want %v, got %d", 4, iterated)
-	}
-}
-
 func TestInactiveClient(t *testing.T) {
 	var (
 		clock mclock.Simulated
@@ -557,8 +453,8 @@ func TestInactiveClient(t *testing.T) {
 	p1 := newPoolTestPeer(1, nil)
 	p2 := newPoolTestPeer(2, nil)
 	p3 := newPoolTestPeer(3, nil)
-	pool.addBalance(p1.ID(), 1000, "")
+	pool.addBalance(p1.ID(), 1000)
-	pool.addBalance(p3.ID(), 2000, "")
+	pool.addBalance(p3.ID(), 2000)
 	// p1: 1000  p2: 0  p3: 2000
 	p1.cap, _ = pool.connect(p1, 1)
 	if p1.cap != 1 {
@@ -575,7 +471,7 @@ func TestInactiveClient(t *testing.T) {
 	if p2.cap != 0 {
 		t.Fatalf("Failed to deactivate peer #2")
 	}
-	pool.addBalance(p2.ID(), 3000, "")
+	pool.addBalance(p2.ID(), 3000)
 	// p1: 1000  p2: 3000  p3: 2000
 	if p2.cap != 1 {
 		t.Fatalf("Failed to activate peer #2")
@@ -583,7 +479,7 @@ func TestInactiveClient(t *testing.T) {
 	if p1.cap != 0 {
 		t.Fatalf("Failed to deactivate peer #1")
 	}
-	pool.addBalance(p2.ID(), -2500, "")
+	pool.addBalance(p2.ID(), -2500)
 	// p1: 1000  p2: 500  p3: 2000
 	if p1.cap != 1 {
 		t.Fatalf("Failed to activate peer #1")
@@ -593,7 +489,7 @@ func TestInactiveClient(t *testing.T) {
 	}
 	pool.setDefaultFactors(priceFactors{1e-9, 0, 0}, priceFactors{1e-9, 0, 0})
 	p4 := newPoolTestPeer(4, nil)
-	pool.addBalance(p4.ID(), 1500, "")
+	pool.addBalance(p4.ID(), 1500)
 	// p1: 1000  p2: 500  p3: 2000  p4: 1500
 	p4.cap, _ = pool.connect(p4, 1)
 	if p4.cap != 1 {
@@ -616,7 +512,7 @@ func TestInactiveClient(t *testing.T) {
 	}
 	pool.disconnect(p2)
 	pool.disconnect(p4)
-	pool.addBalance(p1.ID(), -1000, "")
+	pool.addBalance(p1.ID(), -1000)
 	if p1.cap != 1 {
 		t.Fatalf("Should not deactivate peer #1")
 	}

les/tokensale.go

@@ -19,7 +19,6 @@ package les
 import (
 	"encoding/binary"
 	"fmt"
-	"io"
 	"math"
 	"strconv"
 	"sync"
@@ -41,8 +40,8 @@ const (
 // payment technologies
 type paymentReceiver interface {
 	info() keyValueList
-	receivePayment(from enode.ID, proofOfPayment, oldMeta []byte) (value uint64, newMeta []byte, err error)
+	receivePayment(from enode.ID, proofOfPayment []byte) (value uint64, err error)
-	requestPayment(from enode.ID, value uint64, meta []byte) uint64
+	requestPayment(from enode.ID, value uint64) uint64
 }
 
 // tokenSale handles client balance deposits, conversion to and from service tokens
@@ -401,10 +400,8 @@ func (t *tokenSale) connection(id enode.ID, freeID string, requestedCapacity uin
 	tokensMissing, availableCapacity, err = t.clientPool.setCapacityLocked(id, freeID, requestedCapacity, stayConnected, setCap)
 	pb := t.clientPool.getPosBalance(id)
 	tokenBalance = pb.value.value(t.clientPool.posExpiration(mclock.Now()))
-	var meta tokenSaleMeta
+	cb := t.clientPool.ndb.getCurrencyBalance(id)
-	if err := rlp.DecodeBytes([]byte(pb.meta), &meta); err == nil {
+	pcBalance = cb.amount
-		pcBalance = meta.pcBalance
-	}
 	if tokensMissing == 0 {
 		return
 	}
@@ -440,7 +437,7 @@ func (t *tokenSale) connection(id enode.ID, freeID string, requestedCapacity uin
 		if rec, ok := t.receivers[recID]; !ok || pcMissing == math.MaxUint64 {
 			paymentRequired[i] = math.MaxUint64
 		} else {
-			paymentRequired[i] = rec.requestPayment(id, pcMissing, meta.receiverMeta[recID])
+			paymentRequired[i] = rec.requestPayment(id, pcMissing)
 		}
 	}
 	return
@@ -451,24 +448,19 @@ func (t *tokenSale) deposit(id enode.ID, paymentModule string, proofOfPayment []
 	t.lock.Lock()
 	defer t.lock.Unlock()
 
-	pb := t.clientPool.getPosBalance(id)
+	cb := t.clientPool.ndb.getCurrencyBalance(id)
-	var meta tokenSaleMeta
+	pcBalance = cb.amount
-	if err := rlp.DecodeBytes([]byte(pb.meta), &meta); err == nil {
-		pcBalance = meta.pcBalance
-	}
-
 	pm := t.receivers[paymentModule]
 	if pm == nil {
 		return 0, pcBalance, fmt.Errorf("Unknown payment receiver '%s'", paymentModule)
 	}
-	pcValue, meta.receiverMeta[paymentModule], err = pm.receivePayment(id, proofOfPayment, meta.receiverMeta[paymentModule])
+	pcValue, err = pm.receivePayment(id, proofOfPayment)
 	if err != nil {
 		return 0, pcBalance, err
 	}
 	pcBalance += pcValue
-	meta.pcBalance = pcBalance
+	cb.amount = pcBalance
-	metaEnc, _ := rlp.EncodeToBytes(&meta)
+	t.clientPool.ndb.setCurrencyBalance(id, cb)
-	t.clientPool.addBalance(id, 0, string(metaEnc))
 	return
 }
 
@@ -491,10 +483,8 @@ func (t *tokenSale) buyTokens(id enode.ID, maxSpend, minReceive uint64, relative
 
 	pb := t.clientPool.getPosBalance(id)
 	tokenBalance = pb.value.value(t.clientPool.posExpiration(mclock.Now()))
-	var meta tokenSaleMeta
+	cb := t.clientPool.ndb.getCurrencyBalance(id)
-	if err := rlp.DecodeBytes([]byte(pb.meta), &meta); err == nil {
+	pcBalance = cb.amount
-		pcBalance = meta.pcBalance
-	}
 	if relative {
 		if pcBalance > maxSpend {
 			maxSpend = pcBalance - maxSpend
@@ -526,10 +516,10 @@ func (t *tokenSale) buyTokens(id enode.ID, maxSpend, minReceive uint64, relative
 	}
 	if success {
 		pcBalance -= spend
+		cb.amount = pcBalance
 		tokenBalance += receive
-		meta.pcBalance = pcBalance
+		t.clientPool.ndb.setCurrencyBalance(id, cb)
-		metaEnc, _ := rlp.EncodeToBytes(&meta)
+		t.clientPool.addBalance(id, int64(receive))
-		t.clientPool.addBalance(id, int64(receive), string(metaEnc))
 	}
 	return
 }
@@ -542,10 +532,8 @@ func (t *tokenSale) sellTokens(id enode.ID, maxSell, minRefund uint64, relative,
 
 	pb := t.clientPool.getPosBalance(id)
 	tokenBalance = pb.value.value(t.clientPool.posExpiration(mclock.Now()))
-	var meta tokenSaleMeta
+	cb := t.clientPool.ndb.getCurrencyBalance(id)
-	if err := rlp.DecodeBytes([]byte(pb.meta), &meta); err == nil {
+	pcBalance = cb.amount
-		pcBalance = meta.pcBalance
-	}
 	if relative {
 		if pcBalance < minRefund {
 			minRefund -= pcBalance
@@ -580,9 +568,9 @@ func (t *tokenSale) sellTokens(id enode.ID, maxSell, minRefund uint64, relative,
 	if success {
 		pcBalance += refund
 		tokenBalance -= sell
-		meta.pcBalance = pcBalance
+		cb.amount = pcBalance
-		metaEnc, _ := rlp.EncodeToBytes(&meta)
+		t.clientPool.ndb.setCurrencyBalance(id, cb)
-		t.clientPool.addBalance(id, -int64(sell), string(metaEnc))
+		t.clientPool.addBalance(id, -int64(sell))
 	}
 	return
 }
@@ -593,12 +581,8 @@ func (t *tokenSale) getBalance(id enode.ID) (pcBalance, tokenBalance uint64) {
 	defer t.lock.Unlock()
 
 	pb := t.clientPool.getPosBalance(id)
-	tokenBalance = pb.value.value(t.clientPool.posExpiration(mclock.Now()))
+	cb := t.clientPool.ndb.getCurrencyBalance(id)
-	var meta tokenSaleMeta
+	return pb.value.value(t.clientPool.posExpiration(mclock.Now())), cb.amount
-	if err := rlp.DecodeBytes([]byte(pb.meta), &meta); err == nil {
-		pcBalance = meta.pcBalance
-	}
-	return
 }
 
 // info returns general information about the server, including version info of the
@@ -626,65 +610,6 @@ func (t *tokenSale) receiverInfo(receiverIDs []string) []keyValueList {
 	return res
 }
 
-// tokenSaleMeta is the "meta" field used by the lespay token sale module. It is
-// attached to token balances and it includes the permanent balance of the client
-// nominated in the server's preferred currency and the meta fields provided by
-// the used payment receivers.
-type tokenSaleMeta struct {
-	pcBalance    uint64
-	receiverMeta map[string][]byte
-}
-
-// receiverMetaEnc is used for easy RLP encoding/decoding
-type receiverMetaEnc struct {
-	Id   string
-	Meta []byte
-}
-
-// tokenSaleMetaEnc is used for easy RLP encoding/decoding
-type tokenSaleMetaEnc struct {
-	Id        string
-	Version   uint
-	PcBalance uint64
-	Receivers []receiverMetaEnc
-}
-
-// EncodeRLP implements rlp.Encoder
-func (t *tokenSaleMeta) EncodeRLP(w io.Writer) error {
-	receivers := make([]receiverMetaEnc, len(t.receiverMeta))
-	i := 0
-	for id, meta := range t.receiverMeta {
-		receivers[i] = receiverMetaEnc{id, meta}
-		i++
-	}
-	return rlp.Encode(w, tokenSaleMetaEnc{
-		Id:        "tokenSale",
-		Version:   1,
-		PcBalance: t.pcBalance,
-		Receivers: receivers,
-	})
-}
-
-// DecodeRLP implements rlp.Decoder
-func (t *tokenSaleMeta) DecodeRLP(s *rlp.Stream) error {
-	if t.receiverMeta == nil {
-		t.receiverMeta = make(map[string][]byte)
-	}
-	var e tokenSaleMetaEnc
-	if err := s.Decode(&e); err != nil {
-		return err
-	}
-	if e.Id != "tokenSale" || e.Version != 1 {
-		return fmt.Errorf("Unknown balance meta format '%s' version %d", e.Id, e.Version)
-	}
-	t.receiverMeta = make(map[string][]byte)
-	t.pcBalance = e.PcBalance
-	for _, r := range e.Receivers {
-		t.receiverMeta[r.Id] = r.Meta
-	}
-	return nil
-}
-
 const (
 	tsInfo = iota
 	tsReceiverInfo
@@ -853,7 +778,7 @@ func (t testReceiver) info() keyValueList {
 
 // receivePayment implements paymentReceiver. proofOfPayment is a base 10 ascii number
 // which is credited to the sender's account without any further conditions.
-func (t testReceiver) receivePayment(from enode.ID, proofOfPayment, oldMeta []byte) (value uint64, newMeta []byte, err error) {
+func (t testReceiver) receivePayment(from enode.ID, proofOfPayment []byte) (value uint64, err error) {
 	if len(proofOfPayment) > 8 {
 		err = fmt.Errorf("proof of payment is too long; max 8 bytes long big endian integer expected")
 		return
@@ -865,6 +790,6 @@ func (t testReceiver) receivePayment(from enode.ID, proofOfPayment, oldMeta []by
 }
 
 // requestPayment implements paymentReceiver
-func (t testReceiver) requestPayment(from enode.ID, value uint64, meta []byte) uint64 {
+func (t testReceiver) requestPayment(from enode.ID, value uint64) uint64 {
 	return value
 }