1 // Contains the node database, storing previously seen nodes and any collected
2 // metadata about them for QoS purposes.
15 log "github.com/sirupsen/logrus"
16 "github.com/syndtr/goleveldb/leveldb"
17 "github.com/syndtr/goleveldb/leveldb/errors"
18 "github.com/syndtr/goleveldb/leveldb/iterator"
19 "github.com/syndtr/goleveldb/leveldb/opt"
20 "github.com/syndtr/goleveldb/leveldb/storage"
21 "github.com/syndtr/goleveldb/leveldb/util"
22 "github.com/tendermint/go-wire"
24 "github.com/vapor/crypto"
28 nodeDBNilNodeID = NodeID{} // Special node ID to use as a nil element.
29 nodeDBNodeExpiration = 24 * time.Hour // Time after which an unseen node should be dropped.
30 nodeDBCleanupCycle = time.Hour // Time period for running the expiration task.
33 // nodeDB stores all nodes we know about.
35 lvl *leveldb.DB // Interface to the database itself
36 self NodeID // Own node id to prevent adding it into the database
37 runner sync.Once // Ensures we can start at most one expirer
38 quit chan struct{} // Channel to signal the expiring thread to stop
41 // Schema layout for the node database
43 nodeDBVersionKey = []byte("version") // Version of the database to flush if changes
44 nodeDBItemPrefix = []byte("n:") // Identifier to prefix node entries with
46 nodeDBDiscoverRoot = ":discover"
47 nodeDBDiscoverPing = nodeDBDiscoverRoot + ":lastping"
48 nodeDBDiscoverPong = nodeDBDiscoverRoot + ":lastpong"
49 nodeDBDiscoverFindFails = nodeDBDiscoverRoot + ":findfail"
50 nodeDBDiscoverLocalEndpoint = nodeDBDiscoverRoot + ":localendpoint"
51 nodeDBTopicRegTickets = ":tickets"
54 // newNodeDB creates a new node database for storing and retrieving infos about
55 // known peers in the network. If no path is given, an in-memory, temporary
56 // database is constructed.
57 func newNodeDB(path string, version int, self NodeID) (*nodeDB, error) {
59 return newMemoryNodeDB(self)
61 return newPersistentNodeDB(path, version, self)
64 // newMemoryNodeDB creates a new in-memory node database without a persistent
66 func newMemoryNodeDB(self NodeID) (*nodeDB, error) {
67 db, err := leveldb.Open(storage.NewMemStorage(), nil)
74 quit: make(chan struct{}),
78 // newPersistentNodeDB creates/opens a leveldb backed persistent node database,
79 // also flushing its contents in case of a version mismatch.
80 func newPersistentNodeDB(path string, version int, self NodeID) (*nodeDB, error) {
81 opts := &opt.Options{OpenFilesCacheCapacity: 5}
82 db, err := leveldb.OpenFile(path, opts)
83 if _, iscorrupted := err.(*errors.ErrCorrupted); iscorrupted {
84 db, err = leveldb.RecoverFile(path, nil)
89 // The nodes contained in the cache correspond to a certain protocol version.
90 // Flush all nodes if the version doesn't match.
91 currentVer := make([]byte, binary.MaxVarintLen64)
92 currentVer = currentVer[:binary.PutVarint(currentVer, int64(version))]
94 blob, err := db.Get(nodeDBVersionKey, nil)
96 case leveldb.ErrNotFound:
97 // Version not found (i.e. empty cache), insert it
98 if err := db.Put(nodeDBVersionKey, currentVer, nil); err != nil {
104 // Version present, flush if different
105 if !bytes.Equal(blob, currentVer) {
107 if err = os.RemoveAll(path); err != nil {
110 return newPersistentNodeDB(path, version, self)
116 quit: make(chan struct{}),
120 // makeKey generates the leveldb key-blob from a node id and its particular
121 // field of interest.
122 func makeKey(id NodeID, field string) []byte {
123 if bytes.Equal(id[:], nodeDBNilNodeID[:]) {
126 return append(nodeDBItemPrefix, append(id[:], field...)...)
129 // splitKey tries to split a database key into a node id and a field part.
130 func splitKey(key []byte) (id NodeID, field string) {
131 // If the key is not of a node, return it plainly
132 if !bytes.HasPrefix(key, nodeDBItemPrefix) {
133 return NodeID{}, string(key)
135 // Otherwise split the id and field
136 item := key[len(nodeDBItemPrefix):]
137 copy(id[:], item[:len(id)])
138 field = string(item[len(id):])
143 // fetchInt64 retrieves an integer instance associated with a particular
145 func (db *nodeDB) fetchInt64(key []byte) int64 {
146 blob, err := db.lvl.Get(key, nil)
150 val, read := binary.Varint(blob)
157 // storeInt64 update a specific database entry to the current time instance as a
159 func (db *nodeDB) storeInt64(key []byte, n int64) error {
160 blob := make([]byte, binary.MaxVarintLen64)
161 blob = blob[:binary.PutVarint(blob, n)]
162 return db.lvl.Put(key, blob, nil)
165 //func (db *nodeDB) storeRLP(key []byte, val interface{}) error {
166 // blob, err := wire.WriteBinary(val,)
170 // return db.lvl.Put(key, blob, nil)
173 //func (db *nodeDB) fetchRLP(key []byte, val interface{}) error {
174 // blob, err := db.lvl.Get(key, nil)
178 // err = rlp.DecodeBytes(blob, val)
180 // log.Warn(fmt.Sprintf("key %x (%T) %v", key, val, err))
185 // node retrieves a node with a given id from the database.
186 func (db *nodeDB) node(id NodeID) *Node {
188 //if err := db.fetchRLP(makeKey(id, nodeDBDiscoverRoot), &node); err != nil {
191 node.sha = crypto.Sha256Hash(node.ID[:])
195 // updateNode inserts - potentially overwriting - a node into the peer database.
196 //func (db *nodeDB) updateNode(node *Node) error {
197 // return db.storeRLP(makeKey(node.ID, nodeDBDiscoverRoot), node)
200 // deleteNode deletes all information/keys associated with a node.
201 func (db *nodeDB) deleteNode(id NodeID) error {
202 deleter := db.lvl.NewIterator(util.BytesPrefix(makeKey(id, "")), nil)
204 if err := db.lvl.Delete(deleter.Key(), nil); err != nil {
211 // ensureExpirer is a small helper method ensuring that the data expiration
212 // mechanism is running. If the expiration goroutine is already running, this
213 // method simply returns.
215 // The goal is to start the data evacuation only after the network successfully
216 // bootstrapped itself (to prevent dumping potentially useful seed nodes). Since
217 // it would require significant overhead to exactly trace the first successful
218 // convergence, it's simpler to "ensure" the correct state when an appropriate
219 // condition occurs (i.e. a successful bonding), and discard further events.
220 func (db *nodeDB) ensureExpirer() {
221 db.runner.Do(func() { go db.expirer() })
224 // expirer should be started in a go routine, and is responsible for looping ad
225 // infinitum and dropping stale data from the database.
226 func (db *nodeDB) expirer() {
227 tick := time.NewTicker(nodeDBCleanupCycle)
232 if err := db.expireNodes(); err != nil {
233 log.Error(fmt.Sprintf("Failed to expire nodedb items: %v", err))
241 // expireNodes iterates over the database and deletes all nodes that have not
242 // been seen (i.e. received a pong from) for some allotted time.
243 func (db *nodeDB) expireNodes() error {
244 threshold := time.Now().Add(-nodeDBNodeExpiration)
246 // Find discovered nodes that are older than the allowance
247 it := db.lvl.NewIterator(nil, nil)
251 // Skip the item if not a discovery node
252 id, field := splitKey(it.Key())
253 if field != nodeDBDiscoverRoot {
256 // Skip the node if not expired yet (and not self)
257 if !bytes.Equal(id[:], db.self[:]) {
258 if seen := db.lastPong(id); seen.After(threshold) {
262 // Otherwise delete all associated information
268 // lastPing retrieves the time of the last ping packet send to a remote node,
269 // requesting binding.
270 func (db *nodeDB) lastPing(id NodeID) time.Time {
271 return time.Unix(db.fetchInt64(makeKey(id, nodeDBDiscoverPing)), 0)
274 // updateLastPing updates the last time we tried contacting a remote node.
275 func (db *nodeDB) updateLastPing(id NodeID, instance time.Time) error {
276 return db.storeInt64(makeKey(id, nodeDBDiscoverPing), instance.Unix())
279 // lastPong retrieves the time of the last successful contact from remote node.
280 func (db *nodeDB) lastPong(id NodeID) time.Time {
281 return time.Unix(db.fetchInt64(makeKey(id, nodeDBDiscoverPong)), 0)
284 // updateLastPong updates the last time a remote node successfully contacted.
285 func (db *nodeDB) updateLastPong(id NodeID, instance time.Time) error {
286 return db.storeInt64(makeKey(id, nodeDBDiscoverPong), instance.Unix())
289 // findFails retrieves the number of findnode failures since bonding.
290 func (db *nodeDB) findFails(id NodeID) int {
291 return int(db.fetchInt64(makeKey(id, nodeDBDiscoverFindFails)))
294 // updateFindFails updates the number of findnode failures since bonding.
295 func (db *nodeDB) updateFindFails(id NodeID, fails int) error {
296 return db.storeInt64(makeKey(id, nodeDBDiscoverFindFails), int64(fails))
299 // localEndpoint returns the last local endpoint communicated to the
300 // given remote node.
301 //func (db *nodeDB) localEndpoint(id NodeID) *rpcEndpoint {
302 // var ep rpcEndpoint
303 // if err := db.fetchRLP(makeKey(id, nodeDBDiscoverLocalEndpoint), &ep); err != nil {
309 //func (db *nodeDB) updateLocalEndpoint(id NodeID, ep rpcEndpoint) error {
310 // return db.storeRLP(makeKey(id, nodeDBDiscoverLocalEndpoint), &ep)
313 // querySeeds retrieves random nodes to be used as potential seed nodes
314 // for bootstrapping.
315 func (db *nodeDB) querySeeds(n int, maxAge time.Duration) []*Node {
318 nodes = make([]*Node, 0, n)
319 it = db.lvl.NewIterator(nil, nil)
325 for seeks := 0; len(nodes) < n && seeks < n*5; seeks++ {
326 // Seek to a random entry. The first byte is incremented by a
327 // random amount each time in order to increase the likelihood
328 // of hitting all existing nodes in very small databases.
331 id[0] = ctr + id[0]%16
332 it.Seek(makeKey(id, nodeDBDiscoverRoot))
337 continue seek // iterator exhausted
342 if now.Sub(db.lastPong(n.ID)) > maxAge {
345 for i := range nodes {
346 if nodes[i].ID == n.ID {
347 continue seek // duplicate
350 nodes = append(nodes, n)
355 func (db *nodeDB) fetchTopicRegTickets(id NodeID) (issued, used uint32) {
356 key := makeKey(id, nodeDBTopicRegTickets)
357 blob, _ := db.lvl.Get(key, nil)
361 issued = binary.BigEndian.Uint32(blob[0:4])
362 used = binary.BigEndian.Uint32(blob[4:8])
366 func (db *nodeDB) updateTopicRegTickets(id NodeID, issued, used uint32) error {
367 key := makeKey(id, nodeDBTopicRegTickets)
368 blob := make([]byte, 8)
369 binary.BigEndian.PutUint32(blob[0:4], issued)
370 binary.BigEndian.PutUint32(blob[4:8], used)
371 return db.lvl.Put(key, blob, nil)
374 // reads the next node record from the iterator, skipping over other
376 func nextNode(it iterator.Iterator) *Node {
377 for end := false; !end; end = !it.Next() {
378 id, field := splitKey(it.Key())
379 if field != nodeDBDiscoverRoot {
383 if err := wire.ReadBinaryBytes(it.Value(), &n); err != nil {
384 log.Error("invalid node:", id, err)
387 //if err := rlp.DecodeBytes(it.Value(), &n); err != nil {
388 // log.Warn(fmt.Sprintf("invalid node %x: %v", id, err))
396 // close flushes and closes the database files.
397 func (db *nodeDB) close() {