1// Copyright 2020 The go-ethereum Authors
2// This file is part of the go-ethereum library.
3//
4// The go-ethereum library is free software: you can redistribute it and/or modify
5// it under the terms of the GNU Lesser General Public License as published by
6// the Free Software Foundation, either version 3 of the License, or
7// (at your option) any later version.
8//
9// The go-ethereum library is distributed in the hope that it will be useful,
10// but WITHOUT ANY WARRANTY; without even the implied warranty of
11// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12// GNU Lesser General Public License for more details.
13//
14// You should have received a copy of the GNU Lesser General Public License
15// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
16
17package server
18
19import (
20	"bytes"
21	"encoding/binary"
22	"time"
23
24	"github.com/ethereum/go-ethereum/common"
25	"github.com/ethereum/go-ethereum/common/mclock"
26	"github.com/ethereum/go-ethereum/ethdb"
27	"github.com/ethereum/go-ethereum/les/utils"
28	"github.com/ethereum/go-ethereum/log"
29	"github.com/ethereum/go-ethereum/p2p/enode"
30	"github.com/ethereum/go-ethereum/rlp"
31	lru "github.com/hashicorp/golang-lru"
32)
33
34const (
35	balanceCacheLimit = 8192 // the maximum number of cached items in service token balance queue
36
37	// nodeDBVersion is the version identifier of the node data in db
38	//
39	// Changelog:
40	// Version 0 => 1
41	// * Replace `lastTotal` with `meta` in positive balance: version 0=>1
42	//
43	// Version 1 => 2
44	// * Positive Balance and negative balance is changed:
45	// * Cumulative time is replaced with expiration
46	nodeDBVersion = 2
47
48	// dbCleanupCycle is the cycle of db for useless data cleanup
49	dbCleanupCycle = time.Hour
50)
51
52var (
53	positiveBalancePrefix = []byte("pb:")         // dbVersion(uint16 big endian) + positiveBalancePrefix + id -> balance
54	negativeBalancePrefix = []byte("nb:")         // dbVersion(uint16 big endian) + negativeBalancePrefix + ip -> balance
55	expirationKey         = []byte("expiration:") // dbVersion(uint16 big endian) + expirationKey -> posExp, negExp
56)
57
58type nodeDB struct {
59	db            ethdb.KeyValueStore
60	cache         *lru.Cache
61	auxbuf        []byte                                              // 37-byte auxiliary buffer for key encoding
62	verbuf        [2]byte                                             // 2-byte auxiliary buffer for db version
63	evictCallBack func(mclock.AbsTime, bool, utils.ExpiredValue) bool // Callback to determine whether the balance can be evicted.
64	clock         mclock.Clock
65	closeCh       chan struct{}
66	cleanupHook   func() // Test hook used for testing
67}
68
69func newNodeDB(db ethdb.KeyValueStore, clock mclock.Clock) *nodeDB {
70	cache, _ := lru.New(balanceCacheLimit)
71	ndb := &nodeDB{
72		db:      db,
73		cache:   cache,
74		auxbuf:  make([]byte, 37),
75		clock:   clock,
76		closeCh: make(chan struct{}),
77	}
78	binary.BigEndian.PutUint16(ndb.verbuf[:], uint16(nodeDBVersion))
79	go ndb.expirer()
80	return ndb
81}
82
83func (db *nodeDB) close() {
84	close(db.closeCh)
85}
86
87func (db *nodeDB) getPrefix(neg bool) []byte {
88	prefix := positiveBalancePrefix
89	if neg {
90		prefix = negativeBalancePrefix
91	}
92	return append(db.verbuf[:], prefix...)
93}
94
95func (db *nodeDB) key(id []byte, neg bool) []byte {
96	prefix := positiveBalancePrefix
97	if neg {
98		prefix = negativeBalancePrefix
99	}
100	if len(prefix)+len(db.verbuf)+len(id) > len(db.auxbuf) {
101		db.auxbuf = append(db.auxbuf, make([]byte, len(prefix)+len(db.verbuf)+len(id)-len(db.auxbuf))...)
102	}
103	copy(db.auxbuf[:len(db.verbuf)], db.verbuf[:])
104	copy(db.auxbuf[len(db.verbuf):len(db.verbuf)+len(prefix)], prefix)
105	copy(db.auxbuf[len(prefix)+len(db.verbuf):len(prefix)+len(db.verbuf)+len(id)], id)
106	return db.auxbuf[:len(prefix)+len(db.verbuf)+len(id)]
107}
108
109func (db *nodeDB) getExpiration() (utils.Fixed64, utils.Fixed64) {
110	blob, err := db.db.Get(append(db.verbuf[:], expirationKey...))
111	if err != nil || len(blob) != 16 {
112		return 0, 0
113	}
114	return utils.Fixed64(binary.BigEndian.Uint64(blob[:8])), utils.Fixed64(binary.BigEndian.Uint64(blob[8:16]))
115}
116
117func (db *nodeDB) setExpiration(pos, neg utils.Fixed64) {
118	var buff [16]byte
119	binary.BigEndian.PutUint64(buff[:8], uint64(pos))
120	binary.BigEndian.PutUint64(buff[8:16], uint64(neg))
121	db.db.Put(append(db.verbuf[:], expirationKey...), buff[:16])
122}
123
124func (db *nodeDB) getOrNewBalance(id []byte, neg bool) utils.ExpiredValue {
125	key := db.key(id, neg)
126	item, exist := db.cache.Get(string(key))
127	if exist {
128		return item.(utils.ExpiredValue)
129	}
130	var b utils.ExpiredValue
131	enc, err := db.db.Get(key)
132	if err != nil || len(enc) == 0 {
133		return b
134	}
135	if err := rlp.DecodeBytes(enc, &b); err != nil {
136		log.Crit("Failed to decode positive balance", "err", err)
137	}
138	db.cache.Add(string(key), b)
139	return b
140}
141
142func (db *nodeDB) setBalance(id []byte, neg bool, b utils.ExpiredValue) {
143	key := db.key(id, neg)
144	enc, err := rlp.EncodeToBytes(&(b))
145	if err != nil {
146		log.Crit("Failed to encode positive balance", "err", err)
147	}
148	db.db.Put(key, enc)
149	db.cache.Add(string(key), b)
150}
151
152func (db *nodeDB) delBalance(id []byte, neg bool) {
153	key := db.key(id, neg)
154	db.db.Delete(key)
155	db.cache.Remove(string(key))
156}
157
158// getPosBalanceIDs returns a lexicographically ordered list of IDs of accounts
159// with a positive balance
160func (db *nodeDB) getPosBalanceIDs(start, stop enode.ID, maxCount int) (result []enode.ID) {
161	if maxCount <= 0 {
162		return
163	}
164	prefix := db.getPrefix(false)
165	keylen := len(prefix) + len(enode.ID{})
166
167	it := db.db.NewIterator(prefix, start.Bytes())
168	defer it.Release()
169
170	for it.Next() {
171		var id enode.ID
172		if len(it.Key()) != keylen {
173			return
174		}
175		copy(id[:], it.Key()[keylen-len(id):])
176		if bytes.Compare(id.Bytes(), stop.Bytes()) >= 0 {
177			return
178		}
179		result = append(result, id)
180		if len(result) == maxCount {
181			return
182		}
183	}
184	return
185}
186
187// forEachBalance iterates all balances and passes values to callback.
188func (db *nodeDB) forEachBalance(neg bool, callback func(id enode.ID, balance utils.ExpiredValue) bool) {
189	prefix := db.getPrefix(neg)
190	keylen := len(prefix) + len(enode.ID{})
191
192	it := db.db.NewIterator(prefix, nil)
193	defer it.Release()
194
195	for it.Next() {
196		var id enode.ID
197		if len(it.Key()) != keylen {
198			return
199		}
200		copy(id[:], it.Key()[keylen-len(id):])
201
202		var b utils.ExpiredValue
203		if err := rlp.DecodeBytes(it.Value(), &b); err != nil {
204			continue
205		}
206		if !callback(id, b) {
207			return
208		}
209	}
210}
211
212func (db *nodeDB) expirer() {
213	for {
214		select {
215		case <-db.clock.After(dbCleanupCycle):
216			db.expireNodes()
217		case <-db.closeCh:
218			return
219		}
220	}
221}
222
223// expireNodes iterates the whole node db and checks whether the
224// token balances can be deleted.
225func (db *nodeDB) expireNodes() {
226	var (
227		visited int
228		deleted int
229		start   = time.Now()
230	)
231	for _, neg := range []bool{false, true} {
232		iter := db.db.NewIterator(db.getPrefix(neg), nil)
233		for iter.Next() {
234			visited++
235			var balance utils.ExpiredValue
236			if err := rlp.DecodeBytes(iter.Value(), &balance); err != nil {
237				log.Crit("Failed to decode negative balance", "err", err)
238			}
239			if db.evictCallBack != nil && db.evictCallBack(db.clock.Now(), neg, balance) {
240				deleted++
241				db.db.Delete(iter.Key())
242			}
243		}
244	}
245	// Invoke testing hook if it's not nil.
246	if db.cleanupHook != nil {
247		db.cleanupHook()
248	}
249	log.Debug("Expire nodes", "visited", visited, "deleted", deleted, "elapsed", common.PrettyDuration(time.Since(start)))
250}
251