p2p/discover/database.go

   1 // Contains the node database, storing previously seen nodes and any collected
   2 // metadata about them for QoS purposes.
   3
   4 package discover
   5
   6 import (
   7         "bytes"
   8         "crypto/rand"
   9         "encoding/binary"
  10         "fmt"
  11         "os"
  12         "sync"
  13         "time"
  14
  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"
  23
  24         "github.com/vapor/crypto"
  25 )
  26
  27 var (
  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.
  31 )
  32
  33 // nodeDB stores all nodes we know about.
  34 type nodeDB struct {
  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
  39 }
  40
  41 // Schema layout for the node database
  42 var (
  43         nodeDBVersionKey = []byte("version") // Version of the database to flush if changes
  44         nodeDBItemPrefix = []byte("n:")      // Identifier to prefix node entries with
  45
  46         nodeDBDiscoverRoot          = ":discover"
  47         nodeDBDiscoverPing          = nodeDBDiscoverRoot + ":lastping"
  48         nodeDBDiscoverPong          = nodeDBDiscoverRoot + ":lastpong"
  49         nodeDBDiscoverFindFails     = nodeDBDiscoverRoot + ":findfail"
  50         nodeDBDiscoverLocalEndpoint = nodeDBDiscoverRoot + ":localendpoint"
  51         nodeDBTopicRegTickets       = ":tickets"
  52 )
  53
  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) {
  58         if path == "" {
  59                 return newMemoryNodeDB(self)
  60         }
  61         return newPersistentNodeDB(path, version, self)
  62 }
  63
  64 // newMemoryNodeDB creates a new in-memory node database without a persistent
  65 // backend.
  66 func newMemoryNodeDB(self NodeID) (*nodeDB, error) {
  67         db, err := leveldb.Open(storage.NewMemStorage(), nil)
  68         if err != nil {
  69                 return nil, err
  70         }
  71         return &nodeDB{
  72                 lvl:  db,
  73                 self: self,
  74                 quit: make(chan struct{}),
  75         }, nil
  76 }
  77
  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)
  85         }
  86         if err != nil {
  87                 return nil, err
  88         }
  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))]
  93
  94         blob, err := db.Get(nodeDBVersionKey, nil)
  95         switch err {
  96         case leveldb.ErrNotFound:
  97                 // Version not found (i.e. empty cache), insert it
  98                 if err := db.Put(nodeDBVersionKey, currentVer, nil); err != nil {
  99                         db.Close()
 100                         return nil, err
 101                 }
 102
 103         case nil:
 104                 // Version present, flush if different
 105                 if !bytes.Equal(blob, currentVer) {
 106                         db.Close()
 107                         if err = os.RemoveAll(path); err != nil {
 108                                 return nil, err
 109                         }
 110                         return newPersistentNodeDB(path, version, self)
 111                 }
 112         }
 113         return &nodeDB{
 114                 lvl:  db,
 115                 self: self,
 116                 quit: make(chan struct{}),
 117         }, nil
 118 }
 119
 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[:]) {
 124                 return []byte(field)
 125         }
 126         return append(nodeDBItemPrefix, append(id[:], field...)...)
 127 }
 128
 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)
 134         }
 135         // Otherwise split the id and field
 136         item := key[len(nodeDBItemPrefix):]
 137         copy(id[:], item[:len(id)])
 138         field = string(item[len(id):])
 139
 140         return id, field
 141 }
 142
 143 // fetchInt64 retrieves an integer instance associated with a particular
 144 // database key.
 145 func (db *nodeDB) fetchInt64(key []byte) int64 {
 146         blob, err := db.lvl.Get(key, nil)
 147         if err != nil {
 148                 return 0
 149         }
 150         val, read := binary.Varint(blob)
 151         if read <= 0 {
 152                 return 0
 153         }
 154         return val
 155 }
 156
 157 // storeInt64 update a specific database entry to the current time instance as a
 158 // unix timestamp.
 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)
 163 }
 164
 165 //func (db *nodeDB) storeRLP(key []byte, val interface{}) error {
 166 //      blob, err := wire.WriteBinary(val,)
 167 //      if err != nil {
 168 //              return err
 169 //      }
 170 //      return db.lvl.Put(key, blob, nil)
 171 //}
 172
 173 //func (db *nodeDB) fetchRLP(key []byte, val interface{}) error {
 174 //      blob, err := db.lvl.Get(key, nil)
 175 //      if err != nil {
 176 //              return err
 177 //      }
 178 //      err = rlp.DecodeBytes(blob, val)
 179 //      if err != nil {
 180 //              log.Warn(fmt.Sprintf("key %x (%T) %v", key, val, err))
 181 //      }
 182 //      return err
 183 //}
 184
 185 // node retrieves a node with a given id from the database.
 186 func (db *nodeDB) node(id NodeID) *Node {
 187         var node Node
 188         //if err := db.fetchRLP(makeKey(id, nodeDBDiscoverRoot), &node); err != nil {
 189         //      return nil
 190         //}
 191         node.sha = crypto.Sha256Hash(node.ID[:])
 192         return &node
 193 }
 194
 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)
 198 //}
 199
 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)
 203         for deleter.Next() {
 204                 if err := db.lvl.Delete(deleter.Key(), nil); err != nil {
 205                         return err
 206                 }
 207         }
 208         return nil
 209 }
 210
 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.
 214 //
 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() })
 222 }
 223
 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)
 228         defer tick.Stop()
 229         for {
 230                 select {
 231                 case <-tick.C:
 232                         if err := db.expireNodes(); err != nil {
 233                                 log.Error(fmt.Sprintf("Failed to expire nodedb items: %v", err))
 234                         }
 235                 case <-db.quit:
 236                         return
 237                 }
 238         }
 239 }
 240
 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)
 245
 246         // Find discovered nodes that are older than the allowance
 247         it := db.lvl.NewIterator(nil, nil)
 248         defer it.Release()
 249
 250         for it.Next() {
 251                 // Skip the item if not a discovery node
 252                 id, field := splitKey(it.Key())
 253                 if field != nodeDBDiscoverRoot {
 254                         continue
 255                 }
 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) {
 259                                 continue
 260                         }
 261                 }
 262                 // Otherwise delete all associated information
 263                 db.deleteNode(id)
 264         }
 265         return nil
 266 }
 267
 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)
 272 }
 273
 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())
 277 }
 278
 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)
 282 }
 283
 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())
 287 }
 288
 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)))
 292 }
 293
 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))
 297 }
 298
 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 {
 304 //              return nil
 305 //      }
 306 //      return &ep
 307 //}
 308
 309 //func (db *nodeDB) updateLocalEndpoint(id NodeID, ep rpcEndpoint) error {
 310 //      return db.storeRLP(makeKey(id, nodeDBDiscoverLocalEndpoint), &ep)
 311 //}
 312
 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 {
 316         var (
 317                 now   = time.Now()
 318                 nodes = make([]*Node, 0, n)
 319                 it    = db.lvl.NewIterator(nil, nil)
 320                 id    NodeID
 321         )
 322         defer it.Release()
 323
 324 seek:
 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.
 329                 ctr := id[0]
 330                 rand.Read(id[:])
 331                 id[0] = ctr + id[0]%16
 332                 it.Seek(makeKey(id, nodeDBDiscoverRoot))
 333
 334                 n := nextNode(it)
 335                 if n == nil {
 336                         id[0] = 0
 337                         continue seek // iterator exhausted
 338                 }
 339                 if n.ID == db.self {
 340                         continue seek
 341                 }
 342                 if now.Sub(db.lastPong(n.ID)) > maxAge {
 343                         continue seek
 344                 }
 345                 for i := range nodes {
 346                         if nodes[i].ID == n.ID {
 347                                 continue seek // duplicate
 348                         }
 349                 }
 350                 nodes = append(nodes, n)
 351         }
 352         return nodes
 353 }
 354
 355 func (db *nodeDB) fetchTopicRegTickets(id NodeID) (issued, used uint32) {
 356         key := makeKey(id, nodeDBTopicRegTickets)
 357         blob, _ := db.lvl.Get(key, nil)
 358         if len(blob) != 8 {
 359                 return 0, 0
 360         }
 361         issued = binary.BigEndian.Uint32(blob[0:4])
 362         used = binary.BigEndian.Uint32(blob[4:8])
 363         return
 364 }
 365
 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)
 372 }
 373
 374 // reads the next node record from the iterator, skipping over other
 375 // database entries.
 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 {
 380                         continue
 381                 }
 382                 var n Node
 383                 if err := wire.ReadBinaryBytes(it.Value(), &n); err != nil {
 384                         log.Error("invalid node:", id, err)
 385                         continue
 386                 }
 387                 //if err := rlp.DecodeBytes(it.Value(), &n); err != nil {
 388                 //      log.Warn(fmt.Sprintf("invalid node %x: %v", id, err))
 389                 //      continue
 390                 //}
 391                 return &n
 392         }
 393         return nil
 394 }
 395
 396 // close flushes and closes the database files.
 397 func (db *nodeDB) close() {
 398         close(db.quit)
 399         db.lvl.Close()
 400 }