From df8e793e1f2cbe968e9fc54b2bea43dca92292d3 Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Thu, 18 Apr 2024 00:02:49 +0200 Subject: [PATCH] p2p/discover: new revalidation --- p2p/discover/common.go | 35 ++ p2p/discover/lookup.go | 4 +- p2p/discover/node.go | 11 +- p2p/discover/table.go | 638 ++++++++++++++------------------ p2p/discover/table_reval.go | 204 ++++++++++ p2p/discover/table_test.go | 132 +++---- p2p/discover/table_util_test.go | 60 ++- p2p/discover/v4_udp.go | 2 +- p2p/discover/v5_udp.go | 2 +- 9 files changed, 628 insertions(+), 460 deletions(-) create mode 100644 p2p/discover/table_reval.go diff --git a/p2p/discover/common.go b/p2p/discover/common.go index 1f763904bb..8b024231b7 100644 --- a/p2p/discover/common.go +++ b/p2p/discover/common.go @@ -18,7 +18,11 @@ package discover import ( "crypto/ecdsa" + crand "crypto/rand" + "encoding/binary" + "math/rand" "net" + "sync/atomic" "time" "github.com/ethereum/go-ethereum/common/mclock" @@ -92,3 +96,34 @@ type ReadPacket struct { Data []byte Addr *net.UDPAddr } + +type randomSource interface { + Intn(int) int + Int63n(int64) int64 +} + +// reseedingRandom is a random number generator that tracks when it was last re-seeded. +type reseedingRandom struct { + cur atomic.Pointer[rand.Rand] + lastSeed mclock.AbsTime +} + +func (r *reseedingRandom) nextReseedTime() mclock.AbsTime { + return r.lastSeed.Add(10 * time.Minute) +} + +func (r *reseedingRandom) seed(now mclock.AbsTime) { + var b [8]byte + crand.Read(b[:]) + seed := binary.BigEndian.Uint64(b[:]) + r.cur.Store(rand.New(rand.NewSource(int64(seed)))) + r.lastSeed = now +} + +func (r *reseedingRandom) Intn(n int) int { + return r.cur.Load().Intn(n) +} + +func (r *reseedingRandom) Int63n(n int64) int64 { + return r.cur.Load().Int63n(n) +} diff --git a/p2p/discover/lookup.go b/p2p/discover/lookup.go index b8d97b44e1..df6cddc833 100644 --- a/p2p/discover/lookup.go +++ b/p2p/discover/lookup.go @@ -152,9 +152,9 @@ func (it *lookup) query(n *node, reply chan<- []*node) { // Remove the node from the local table if it fails to return anything useful too // many times, but only if there are enough other nodes in the bucket. dropped := false - if fails >= maxFindnodeFailures && it.tab.bucketLen(n.ID()) >= bucketSize/2 { + if fails >= maxFindnodeFailures { dropped = true - it.tab.delete(n) + it.tab.trackFindFailure(n) } it.tab.log.Trace("FINDNODE failed", "id", n.ID(), "failcount", fails, "dropped", dropped, "err", err) } else if fails > 0 { diff --git a/p2p/discover/node.go b/p2p/discover/node.go index 079fed8852..17280a5878 100644 --- a/p2p/discover/node.go +++ b/p2p/discover/node.go @@ -38,9 +38,10 @@ type BucketNode struct { // node represents a host on the network. // The fields of Node may not be modified. type node struct { - enode.Node - addedAt time.Time // time when the node was added to the table - livenessChecks uint // how often liveness was checked + *enode.Node + addedAt time.Time // time when the node was added to the table + livenessChecks uint // how often liveness was checked + isValidatedLive bool } type encPubkey [64]byte @@ -71,7 +72,7 @@ func (e encPubkey) id() enode.ID { } func wrapNode(n *enode.Node) *node { - return &node{Node: *n} + return &node{Node: n} } func wrapNodes(ns []*enode.Node) []*node { @@ -83,7 +84,7 @@ func wrapNodes(ns []*enode.Node) []*node { } func unwrapNode(n *node) *enode.Node { - return &n.Node + return n.Node } func unwrapNodes(ns []*node) []*enode.Node { diff --git a/p2p/discover/table.go b/p2p/discover/table.go index 46e6401448..6e2bf65341 100644 --- a/p2p/discover/table.go +++ b/p2p/discover/table.go @@ -24,16 +24,15 @@ package discover import ( "context" - crand "crypto/rand" - "encoding/binary" "fmt" - mrand "math/rand" "net" + "slices" "sort" "sync" "time" "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/mclock" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/metrics" "github.com/ethereum/go-ethereum/p2p/enode" @@ -55,21 +54,21 @@ const ( bucketIPLimit, bucketSubnet = 2, 24 // at most 2 addresses from the same /24 tableIPLimit, tableSubnet = 10, 24 - copyNodesInterval = 30 * time.Second - seedMinTableTime = 5 * time.Minute - seedCount = 30 - seedMaxAge = 5 * 24 * time.Hour + seedMinTableTime = 5 * time.Minute + seedCount = 30 + seedMaxAge = 5 * 24 * time.Hour ) // Table is the 'node table', a Kademlia-like index of neighbor nodes. The table keeps // itself up-to-date by verifying the liveness of neighbors and requesting their node // records when announcements of a new record version are received. type Table struct { - mutex sync.Mutex // protects buckets, bucket content, nursery, rand - buckets [nBuckets]*bucket // index of known nodes by distance - nursery []*node // bootstrap nodes - rand *mrand.Rand // source of randomness, periodically reseeded - ips netutil.DistinctNetSet + mutex sync.Mutex // protects buckets, bucket content, nursery, rand + buckets [nBuckets]*bucket // index of known nodes by distance + nursery []*node // bootstrap nodes + rand *reseedingRandom // source of randomness, periodically reseeded + ips netutil.DistinctNetSet + revalidation tableRevalidation db *enode.DB // database of known nodes net transport @@ -77,10 +76,14 @@ type Table struct { log log.Logger // loop channels - refreshReq chan chan struct{} - initDone chan struct{} - closeReq chan struct{} - closed chan struct{} + refreshReq chan chan struct{} + revalidateResp chan revalidationResponse + addNodeCh chan addNodeRequest + addNodeHandled chan struct{} + findFailureCh chan *node + initDone chan struct{} + closeReq chan struct{} + closed chan struct{} nodeAddedHook func(*bucket, *node) nodeRemovedHook func(*bucket, *node) @@ -104,22 +107,28 @@ type bucket struct { index int } +type addNodeRequest struct { + node *node + isLive bool +} + func newTable(t transport, db *enode.DB, cfg Config) (*Table, error) { cfg = cfg.withDefaults() tab := &Table{ - net: t, - db: db, - cfg: cfg, - log: cfg.Log, - refreshReq: make(chan chan struct{}), - initDone: make(chan struct{}), - closeReq: make(chan struct{}), - closed: make(chan struct{}), - rand: mrand.New(mrand.NewSource(0)), - ips: netutil.DistinctNetSet{Subnet: tableSubnet, Limit: tableIPLimit}, - } - if err := tab.setFallbackNodes(cfg.Bootnodes); err != nil { - return nil, err + net: t, + db: db, + cfg: cfg, + log: cfg.Log, + refreshReq: make(chan chan struct{}), + revalidateResp: make(chan revalidationResponse), + addNodeCh: make(chan addNodeRequest), + addNodeHandled: make(chan struct{}), + findFailureCh: make(chan *node), + initDone: make(chan struct{}), + closeReq: make(chan struct{}), + closed: make(chan struct{}), + rand: new(reseedingRandom), + ips: netutil.DistinctNetSet{Subnet: tableSubnet, Limit: tableIPLimit}, } for i := range tab.buckets { tab.buckets[i] = &bucket{ @@ -127,26 +136,23 @@ func newTable(t transport, db *enode.DB, cfg Config) (*Table, error) { ips: netutil.DistinctNetSet{Subnet: bucketSubnet, Limit: bucketIPLimit}, } } - tab.seedRand() + tab.rand.seed(cfg.Clock.Now()) + tab.revalidation.init(&cfg) + + // initial table content + if err := tab.setFallbackNodes(cfg.Bootnodes); err != nil { + return nil, err + } tab.loadSeedNodes() return tab, nil } -func newMeteredTable(t transport, db *enode.DB, cfg Config) (*Table, error) { - tab, err := newTable(t, db, cfg) - if err != nil { - return nil, err +func (tab *Table) trackFindFailure(n *node) { + select { + case tab.findFailureCh <- n: + case <-tab.closed: } - if metrics.Enabled { - tab.nodeAddedHook = func(b *bucket, n *node) { - bucketsCounter[b.index].Inc(1) - } - tab.nodeRemovedHook = func(b *bucket, n *node) { - bucketsCounter[b.index].Dec(1) - } - } - return tab, nil } // Nodes returns all nodes contained in the table. @@ -172,15 +178,6 @@ func (tab *Table) self() *enode.Node { return tab.net.Self() } -func (tab *Table) seedRand() { - var b [8]byte - crand.Read(b[:]) - - tab.mutex.Lock() - tab.rand.Seed(int64(binary.BigEndian.Uint64(b[:]))) - tab.mutex.Unlock() -} - // getNode returns the node with the given ID or nil if it isn't in the table. func (tab *Table) getNode(id enode.ID) *enode.Node { tab.mutex.Lock() @@ -240,52 +237,157 @@ func (tab *Table) refresh() <-chan struct{} { return done } -// loop schedules runs of doRefresh, doRevalidate and copyLiveNodes. +// findnodeByID returns the n nodes in the table that are closest to the given id. +// This is used by the FINDNODE/v4 handler. +// +// The preferLive parameter says whether the caller wants liveness-checked results. If +// preferLive is true and the table contains any verified nodes, the result will not +// contain unverified nodes. However, if there are no verified nodes at all, the result +// will contain unverified nodes. +func (tab *Table) findnodeByID(target enode.ID, nresults int, preferLive bool) *nodesByDistance { + tab.mutex.Lock() + defer tab.mutex.Unlock() + + // Scan all buckets. There might be a better way to do this, but there aren't that many + // buckets, so this solution should be fine. The worst-case complexity of this loop + // is O(tab.len() * nresults). + nodes := &nodesByDistance{target: target} + liveNodes := &nodesByDistance{target: target} + for _, b := range &tab.buckets { + for _, n := range b.entries { + nodes.push(n, nresults) + if preferLive && n.isValidatedLive { + liveNodes.push(n, nresults) + } + } + } + + if preferLive && len(liveNodes.entries) > 0 { + return liveNodes + } + return nodes +} + +// appendLiveNodes adds nodes at the given distance to the result slice. +// This is used by the FINDNODE/v5 handler. +func (tab *Table) appendLiveNodes(dist uint, result []*enode.Node) []*enode.Node { + if dist > 256 { + return result + } + if dist == 0 { + return append(result, tab.self()) + } + + tab.mutex.Lock() + defer tab.mutex.Unlock() + for _, n := range tab.bucketAtDistance(int(dist)).entries { + if n.isValidatedLive { + result = append(result, n.Node) + } + } + return result +} + +// len returns the number of nodes in the table. +func (tab *Table) len() (n int) { + tab.mutex.Lock() + defer tab.mutex.Unlock() + + for _, b := range &tab.buckets { + n += len(b.entries) + } + return n +} + +// addSeenNode adds a node which may not be live. If the bucket has space available, +// adding the node succeeds immediately. Otherwise, the node is added to the replacements +// list. +// +// The caller must not hold tab.mutex. +func (tab *Table) addSeenNode(n *node) { + req := addNodeRequest{node: n, isLive: false} + select { + case tab.addNodeCh <- req: + <-tab.addNodeHandled + case <-tab.closed: + } +} + +// addVerifiedNode adds a node whose existence has been verified recently. If the bucket +// has no space, the node is added to the replacements list. +// +// There is an additional safety measure: if the table is still initializing the node +// is not added. This prevents an attack where the table could be filled by just sending +// ping repeatedly. +// +// The caller must not hold tab.mutex. +func (tab *Table) addVerifiedNode(n *node) { + req := addNodeRequest{node: n, isLive: true} + select { + case tab.addNodeCh <- req: + <-tab.addNodeHandled + case <-tab.closed: + } +} + +// loop is the main loop of Table. func (tab *Table) loop() { var ( - revalidate = time.NewTimer(tab.nextRevalidateTime()) - refresh = time.NewTimer(tab.nextRefreshTime()) - copyNodes = time.NewTicker(copyNodesInterval) - refreshDone = make(chan struct{}) // where doRefresh reports completion - revalidateDone chan struct{} // where doRevalidate reports completion - waiting = []chan struct{}{tab.initDone} // holds waiting callers while doRefresh runs + refresh = time.NewTimer(tab.nextRefreshTime()) + refreshDone = make(chan struct{}) // where doRefresh reports completion + waiting = []chan struct{}{tab.initDone} // holds waiting callers while doRefresh runs + revalTimer = mclock.NewAlarm(tab.cfg.Clock) + reseedRandTimer = mclock.NewAlarm(tab.cfg.Clock) ) defer refresh.Stop() - defer revalidate.Stop() - defer copyNodes.Stop() + defer revalTimer.Stop() + defer reseedRandTimer.Stop() // Start initial refresh. go tab.doRefresh(refreshDone) loop: for { + reseedRandTimer.Schedule(tab.rand.nextReseedTime()) + revalTimer.Schedule(tab.revalidation.nextTime()) + select { + case <-reseedRandTimer.C(): + tab.rand.seed(tab.cfg.Clock.Now()) + + case <-revalTimer.C(): + tab.revalidation.run(tab, mclock.Now()) + + case r := <-tab.revalidateResp: + tab.revalidation.handleResponse(tab, r) + + case addreq := <-tab.addNodeCh: + tab.handleAddNode(addreq) + tab.addNodeHandled <- struct{}{} + + case <-tab.findFailureCh: + // TODO: handle failure by potentially dropping node + case <-refresh.C: - tab.seedRand() if refreshDone == nil { refreshDone = make(chan struct{}) go tab.doRefresh(refreshDone) } + case req := <-tab.refreshReq: waiting = append(waiting, req) if refreshDone == nil { refreshDone = make(chan struct{}) go tab.doRefresh(refreshDone) } + case <-refreshDone: for _, ch := range waiting { close(ch) } waiting, refreshDone = nil, nil refresh.Reset(tab.nextRefreshTime()) - case <-revalidate.C: - revalidateDone = make(chan struct{}) - go tab.doRevalidate(revalidateDone) - case <-revalidateDone: - revalidate.Reset(tab.nextRevalidateTime()) - revalidateDone = nil - case <-copyNodes.C: - go tab.copyLiveNodes() + case <-tab.closeReq: break loop } @@ -297,9 +399,6 @@ loop: for _, ch := range waiting { close(ch) } - if revalidateDone != nil { - <-revalidateDone - } close(tab.closed) } @@ -340,165 +439,11 @@ func (tab *Table) loadSeedNodes() { } } -// doRevalidate checks that the last node in a random bucket is still live and replaces or -// deletes the node if it isn't. -func (tab *Table) doRevalidate(done chan<- struct{}) { - defer func() { done <- struct{}{} }() - - last, bi := tab.nodeToRevalidate() - if last == nil { - // No non-empty bucket found. - return - } - - // Ping the selected node and wait for a pong. - remoteSeq, err := tab.net.ping(unwrapNode(last)) - - // Also fetch record if the node replied and returned a higher sequence number. - if last.Seq() < remoteSeq { - n, err := tab.net.RequestENR(unwrapNode(last)) - if err != nil { - tab.log.Debug("ENR request failed", "id", last.ID(), "addr", last.addr(), "err", err) - } else { - last = &node{Node: *n, addedAt: last.addedAt, livenessChecks: last.livenessChecks} - } - } - - tab.mutex.Lock() - defer tab.mutex.Unlock() - b := tab.buckets[bi] - if err == nil { - // The node responded, move it to the front. - last.livenessChecks++ - tab.log.Debug("Revalidated node", "b", bi, "id", last.ID(), "checks", last.livenessChecks) - tab.bumpInBucket(b, last) - return - } - // No reply received, pick a replacement or delete the node if there aren't - // any replacements. - if r := tab.replace(b, last); r != nil { - tab.log.Debug("Replaced dead node", "b", bi, "id", last.ID(), "ip", last.IP(), "checks", last.livenessChecks, "r", r.ID(), "rip", r.IP()) - } else { - tab.log.Debug("Removed dead node", "b", bi, "id", last.ID(), "ip", last.IP(), "checks", last.livenessChecks) - } -} - -// nodeToRevalidate returns the last node in a random, non-empty bucket. -func (tab *Table) nodeToRevalidate() (n *node, bi int) { - tab.mutex.Lock() - defer tab.mutex.Unlock() - - for _, bi = range tab.rand.Perm(len(tab.buckets)) { - b := tab.buckets[bi] - if len(b.entries) > 0 { - last := b.entries[len(b.entries)-1] - return last, bi - } - } - return nil, 0 -} - -func (tab *Table) nextRevalidateTime() time.Duration { - tab.mutex.Lock() - defer tab.mutex.Unlock() - - return time.Duration(tab.rand.Int63n(int64(tab.cfg.PingInterval))) -} - func (tab *Table) nextRefreshTime() time.Duration { - tab.mutex.Lock() - defer tab.mutex.Unlock() - half := tab.cfg.RefreshInterval / 2 return half + time.Duration(tab.rand.Int63n(int64(half))) } -// copyLiveNodes adds nodes from the table to the database if they have been in the table -// longer than seedMinTableTime. -func (tab *Table) copyLiveNodes() { - tab.mutex.Lock() - defer tab.mutex.Unlock() - - now := time.Now() - for _, b := range &tab.buckets { - for _, n := range b.entries { - if n.livenessChecks > 0 && now.Sub(n.addedAt) >= seedMinTableTime { - tab.db.UpdateNode(unwrapNode(n)) - } - } - } -} - -// findnodeByID returns the n nodes in the table that are closest to the given id. -// This is used by the FINDNODE/v4 handler. -// -// The preferLive parameter says whether the caller wants liveness-checked results. If -// preferLive is true and the table contains any verified nodes, the result will not -// contain unverified nodes. However, if there are no verified nodes at all, the result -// will contain unverified nodes. -func (tab *Table) findnodeByID(target enode.ID, nresults int, preferLive bool) *nodesByDistance { - tab.mutex.Lock() - defer tab.mutex.Unlock() - - // Scan all buckets. There might be a better way to do this, but there aren't that many - // buckets, so this solution should be fine. The worst-case complexity of this loop - // is O(tab.len() * nresults). - nodes := &nodesByDistance{target: target} - liveNodes := &nodesByDistance{target: target} - for _, b := range &tab.buckets { - for _, n := range b.entries { - nodes.push(n, nresults) - if preferLive && n.livenessChecks > 0 { - liveNodes.push(n, nresults) - } - } - } - - if preferLive && len(liveNodes.entries) > 0 { - return liveNodes - } - return nodes -} - -// appendLiveNodes adds nodes at the given distance to the result slice. -func (tab *Table) appendLiveNodes(dist uint, result []*enode.Node) []*enode.Node { - if dist > 256 { - return result - } - if dist == 0 { - return append(result, tab.self()) - } - - tab.mutex.Lock() - defer tab.mutex.Unlock() - for _, n := range tab.bucketAtDistance(int(dist)).entries { - if n.livenessChecks >= 1 { - node := n.Node // avoid handing out pointer to struct field - result = append(result, &node) - } - } - return result -} - -// len returns the number of nodes in the table. -func (tab *Table) len() (n int) { - tab.mutex.Lock() - defer tab.mutex.Unlock() - - for _, b := range &tab.buckets { - n += len(b.entries) - } - return n -} - -// bucketLen returns the number of nodes in the bucket for the given ID. -func (tab *Table) bucketLen(id enode.ID) int { - tab.mutex.Lock() - defer tab.mutex.Unlock() - - return len(tab.bucket(id).entries) -} - // bucket returns the bucket for the given node ID hash. func (tab *Table) bucket(id enode.ID) *bucket { d := enode.LogDist(tab.self().ID(), id) @@ -512,95 +457,6 @@ func (tab *Table) bucketAtDistance(d int) *bucket { return tab.buckets[d-bucketMinDistance-1] } -// addSeenNode adds a node which may or may not be live to the end of a bucket. If the -// bucket has space available, adding the node succeeds immediately. Otherwise, the node is -// added to the replacements list. -// -// The caller must not hold tab.mutex. -func (tab *Table) addSeenNode(n *node) { - if n.ID() == tab.self().ID() { - return - } - - tab.mutex.Lock() - defer tab.mutex.Unlock() - b := tab.bucket(n.ID()) - if contains(b.entries, n.ID()) { - // Already in bucket, don't add. - return - } - if len(b.entries) >= bucketSize { - // Bucket full, maybe add as replacement. - tab.addReplacement(b, n) - return - } - if !tab.addIP(b, n.IP()) { - // Can't add: IP limit reached. - return - } - - // Add to end of bucket: - b.entries = append(b.entries, n) - b.replacements = deleteNode(b.replacements, n) - n.addedAt = time.Now() - - if tab.nodeAddedHook != nil { - tab.nodeAddedHook(b, n) - } -} - -// addVerifiedNode adds a node whose existence has been verified recently to the front of a -// bucket. If the node is already in the bucket, it is moved to the front. If the bucket -// has no space, the node is added to the replacements list. -// -// There is an additional safety measure: if the table is still initializing the node -// is not added. This prevents an attack where the table could be filled by just sending -// ping repeatedly. -// -// The caller must not hold tab.mutex. -func (tab *Table) addVerifiedNode(n *node) { - if !tab.isInitDone() { - return - } - if n.ID() == tab.self().ID() { - return - } - - tab.mutex.Lock() - defer tab.mutex.Unlock() - b := tab.bucket(n.ID()) - if tab.bumpInBucket(b, n) { - // Already in bucket, moved to front. - return - } - if len(b.entries) >= bucketSize { - // Bucket full, maybe add as replacement. - tab.addReplacement(b, n) - return - } - if !tab.addIP(b, n.IP()) { - // Can't add: IP limit reached. - return - } - - // Add to front of bucket. - b.entries, _ = pushNode(b.entries, n, bucketSize) - b.replacements = deleteNode(b.replacements, n) - n.addedAt = time.Now() - - if tab.nodeAddedHook != nil { - tab.nodeAddedHook(b, n) - } -} - -// delete removes an entry from the node table. It is used to evacuate dead nodes. -func (tab *Table) delete(node *node) { - tab.mutex.Lock() - defer tab.mutex.Unlock() - - tab.deleteInBucket(tab.bucket(node.ID()), node) -} - func (tab *Table) addIP(b *bucket, ip net.IP) bool { if len(ip) == 0 { return false // Nodes without IP cannot be added. @@ -628,11 +484,43 @@ func (tab *Table) removeIP(b *bucket, ip net.IP) { b.ips.Remove(ip) } +func (tab *Table) handleAddNode(req addNodeRequest) { + if req.node.ID() == tab.self().ID() { + return + } + if !req.isLive && !tab.isInitDone() { + return + } + + tab.mutex.Lock() + defer tab.mutex.Unlock() + b := tab.bucket(req.node.ID()) + if tab.bumpInBucket(b, req.node.Node) { + // Already in bucket, update record. + return + } + if len(b.entries) >= bucketSize { + // Bucket full, maybe add as replacement. + tab.addReplacement(b, req.node) + return + } + if !tab.addIP(b, req.node.IP()) { + // Can't add: IP limit reached. + return + } + + // Add to bucket. + b.entries, _ = pushNode(b.entries, req.node, bucketSize) + b.replacements = deleteNode(b.replacements, req.node) + req.node.addedAt = time.Now() + tab.nodeAdded(b, req.node) +} + +// addReplacement adds n to the replacement cache of bucket b. func (tab *Table) addReplacement(b *bucket, n *node) { - for _, e := range b.replacements { - if e.ID() == n.ID() { - return // already in list - } + if contains(b.replacements, n.ID()) { + // TODO: update ENR + return } if !tab.addIP(b, n.IP()) { return @@ -644,60 +532,76 @@ func (tab *Table) addReplacement(b *bucket, n *node) { } } -// replace removes n from the replacement list and replaces 'last' with it if it is the -// last entry in the bucket. If 'last' isn't the last entry, it has either been replaced -// with someone else or became active. -func (tab *Table) replace(b *bucket, last *node) *node { - if len(b.entries) == 0 || b.entries[len(b.entries)-1].ID() != last.ID() { - // Entry has moved, don't replace it. - return nil +func (tab *Table) nodeAdded(b *bucket, n *node) { + tab.revalidation.nodeAdded(tab, n) + if tab.nodeAddedHook != nil { + tab.nodeAddedHook(b, n) } - // Still the last entry. - if len(b.replacements) == 0 { - tab.deleteInBucket(b, last) - return nil + if metrics.Enabled { + bucketsCounter[b.index].Inc(1) } - r := b.replacements[tab.rand.Intn(len(b.replacements))] - b.replacements = deleteNode(b.replacements, r) - b.entries[len(b.entries)-1] = r - tab.removeIP(b, last.IP()) - return r } -// bumpInBucket moves the given node to the front of the bucket entry list -// if it is contained in that list. -func (tab *Table) bumpInBucket(b *bucket, n *node) bool { - for i := range b.entries { - if b.entries[i].ID() == n.ID() { - if !n.IP().Equal(b.entries[i].IP()) { - // Endpoint has changed, ensure that the new IP fits into table limits. - tab.removeIP(b, b.entries[i].IP()) - if !tab.addIP(b, n.IP()) { - // It doesn't, put the previous one back. - tab.addIP(b, b.entries[i].IP()) - return false - } - } - // Move it to the front. - copy(b.entries[1:], b.entries[:i]) - b.entries[0] = n - return true - } - } - return false -} - -func (tab *Table) deleteInBucket(b *bucket, n *node) { - // Check if the node is actually in the bucket so the removed hook - // isn't called multiple times for the same node. - if !contains(b.entries, n.ID()) { - return - } - b.entries = deleteNode(b.entries, n) - tab.removeIP(b, n.IP()) +func (tab *Table) nodeRemoved(b *bucket, n *node) { + tab.revalidation.nodeRemoved(n) if tab.nodeRemovedHook != nil { tab.nodeRemovedHook(b, n) } + if metrics.Enabled { + bucketsCounter[b.index].Dec(1) + } +} + +// deleteInBucket removes node n from the table. +// If there are replacement nodes in the bucket, the node is replaced. +func (tab *Table) deleteInBucket(b *bucket, id enode.ID) *node { + index := slices.IndexFunc(b.entries, func(e *node) bool { return e.ID() == id }) + if index == -1 { + // Entry has been removed already, don't replace it. + return nil + } + + // Remove the node. + n := b.entries[index] + b.entries = slices.Delete(b.entries, index, index+1) + tab.removeIP(b, n.IP()) + tab.nodeRemoved(b, n) + + // Add replacement. + if len(b.replacements) == 0 { + tab.log.Debug("Removed dead node", "b", b.index, "id", n.ID(), "ip", n.IP(), "checks", n.livenessChecks) + return nil + } + rindex := tab.rand.Intn(len(b.replacements)) + rep := b.replacements[rindex] + b.replacements = slices.Delete(b.replacements, rindex, rindex+1) + b.entries = append(b.entries, rep) + tab.nodeAdded(b, rep) + + tab.log.Debug("Replaced dead node", "b", b.index, "id", n.ID(), "ip", n.IP(), "checks", n.livenessChecks, "r", rep.ID(), "rip", rep.IP()) + return rep +} + +// bumpInBucket updates the node record of n in the bucket. +func (tab *Table) bumpInBucket(b *bucket, newRecord *enode.Node) bool { + i := slices.IndexFunc(b.entries, func(elem *node) bool { + return elem.ID() == newRecord.ID() + }) + if i == -1 { + return false + } + + if !newRecord.IP().Equal(b.entries[i].IP()) { + // Endpoint has changed, ensure that the new IP fits into table limits. + tab.removeIP(b, b.entries[i].IP()) + if !tab.addIP(b, newRecord.IP()) { + // It doesn't, put the previous one back. + tab.addIP(b, b.entries[i].IP()) + return false + } + } + b.entries[i].Node = newRecord + return true } func contains(ns []*node, id enode.ID) bool { diff --git a/p2p/discover/table_reval.go b/p2p/discover/table_reval.go new file mode 100644 index 0000000000..d4bb92a844 --- /dev/null +++ b/p2p/discover/table_reval.go @@ -0,0 +1,204 @@ +// Copyright 2024 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 . + +package discover + +import ( + "fmt" + "slices" + "time" + + "github.com/ethereum/go-ethereum/common/mclock" + "github.com/ethereum/go-ethereum/p2p/enode" +) + +const never = ^mclock.AbsTime(0) + +type tableRevalidation struct { + newNodes revalidationQueue + nodes revalidationQueue + activeReq map[enode.ID]struct{} +} + +type revalidationResponse struct { + n *node + didRespond bool + isNewNode bool + newRecord *enode.Node +} + +func (tr *tableRevalidation) init(cfg *Config) { + tr.activeReq = make(map[enode.ID]struct{}) + tr.newNodes.nextTime = never + tr.newNodes.interval = cfg.PingInterval + tr.nodes.nextTime = never + tr.nodes.interval = cfg.PingInterval +} + +// nodeAdded is called when the table receives a new node. +func (tr *tableRevalidation) nodeAdded(tab *Table, n *node) { + tr.newNodes.push(n, tab.rand) +} + +// nodeRemoved is called when a node was removed from the table. +func (tr *tableRevalidation) nodeRemoved(n *node) { + wasnew := tr.newNodes.remove(n) + if !wasnew { + tr.nodes.remove(n) + } +} + +// nextTime returns the next time run() should be invoked. +// The Table main loop uses this to schedule a timer. +func (tr *tableRevalidation) nextTime() mclock.AbsTime { + return min(tr.newNodes.nextTime, tr.nodes.nextTime) +} + +// run performs node revalidation. +func (tr *tableRevalidation) run(tab *Table, now mclock.AbsTime) { + if n := tr.newNodes.get(now, tab.rand, tr.activeReq); n != nil { + tr.startRequest(tab, n, true) + tr.newNodes.schedule(tab.rand) + } + if n := tr.nodes.get(now, tab.rand, tr.activeReq); n != nil { + tr.startRequest(tab, n, false) + tr.nodes.schedule(tab.rand) + } +} + +// startRequest spawns a revalidation request for node n. +func (tr *tableRevalidation) startRequest(tab *Table, n *node, newNode bool) { + if _, ok := tr.activeReq[n.ID()]; ok { + panic("duplicate startRequest") + } + tr.activeReq[n.ID()] = struct{}{} + resp := revalidationResponse{n: n, isNewNode: newNode} + + go func() { + // Ping the selected node and wait for a pong response. + remoteSeq, err := tab.net.ping(unwrapNode(n)) + resp.didRespond = err == nil + + // Also fetch record if the node replied and returned a higher sequence number. + if remoteSeq > n.Seq() { + newrec, err := tab.net.RequestENR(unwrapNode(n)) + if err != nil { + tab.log.Debug("ENR request failed", "id", n.ID(), "addr", n.addr(), "err", err) + } else { + resp.newRecord = newrec + } + } + + select { + case tab.revalidateResp <- resp: + case <-tab.closed: + } + }() +} + +// handleResponse processes the result of a revalidation request. +func (tr *tableRevalidation) handleResponse(tab *Table, resp revalidationResponse) { + n := resp.n + b := tab.bucket(n.ID()) + delete(tr.activeReq, n.ID()) + + tab.mutex.Lock() + defer tab.mutex.Unlock() + + if !resp.didRespond { + // Revalidation failed. + n.livenessChecks /= 3 + if n.livenessChecks == 0 || resp.isNewNode { + tab.deleteInBucket(b, n.ID()) + } + return + } + + // The node responded. + n.livenessChecks++ + n.isValidatedLive = true + tab.log.Debug("Revalidated node", "b", b.index, "id", n.ID(), "checks", n.livenessChecks) + if resp.newRecord != nil { + updated := tab.bumpInBucket(b, resp.newRecord) + if updated { + // If the node changed its advertised endpoint, the updated ENR is not served + // until it has been revalidated. + n.isValidatedLive = false + } + } + + // Move node over to main queue after first validation. + if resp.isNewNode { + tr.newNodes.remove(n) + tr.nodes.push(n, tab.rand) + } + + // Store potential seeds in database. + if n.isValidatedLive && n.livenessChecks > 5 { + tab.db.UpdateNode(resp.n.Node) + } +} + +// revalidationQueue holds a list nodes and the next revalidation time. +type revalidationQueue struct { + nodes []*node + nextTime mclock.AbsTime + interval time.Duration +} + +// get returns a random node from the queue. Nodes in the 'exclude' map are not returned. +func (rq *revalidationQueue) get(now mclock.AbsTime, rand randomSource, exclude map[enode.ID]struct{}) *node { + if now < rq.nextTime || len(rq.nodes) == 0 { + return nil + } + for i := 0; i < len(rq.nodes)*3; i++ { + n := rq.nodes[rand.Intn(len(rq.nodes))] + _, excluded := exclude[n.ID()] + if !excluded { + return n + } + } + return nil +} + +func (rq *revalidationQueue) push(n *node, rand randomSource) { + rq.nodes = append(rq.nodes, n) + if rq.nextTime == never { + rq.schedule(rand) + } +} + +func (rq *revalidationQueue) schedule(rand randomSource) { + rq.nextTime = mclock.AbsTime(rand.Int63n(int64(rq.interval))) +} + +func (rq *revalidationQueue) remove(n *node) bool { + i := slices.Index(rq.nodes, n) + if i == -1 { + return false + } + rq.nodes = slices.Delete(rq.nodes, i, i+1) + if len(rq.nodes) == 0 { + rq.nextTime = never + } + return true +} + +func printIDs(list []*node) { + for i, n := range list { + fmt.Println(" - ", i, n.ID()) + } +} diff --git a/p2p/discover/table_test.go b/p2p/discover/table_test.go index 3ba3422251..9796d576e5 100644 --- a/p2p/discover/table_test.go +++ b/p2p/discover/table_test.go @@ -20,14 +20,16 @@ import ( "crypto/ecdsa" "fmt" "math/rand" - "net" "reflect" "testing" "testing/quick" "time" + "github.com/ethereum/go-ethereum/common/mclock" "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/internal/testlog" + "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/p2p/enr" "github.com/ethereum/go-ethereum/p2p/netutil" @@ -49,29 +51,34 @@ func TestTable_pingReplace(t *testing.T) { } func testPingReplace(t *testing.T, newNodeIsResponding, lastInBucketIsResponding bool) { + simclock := new(mclock.Simulated) transport := newPingRecorder() - tab, db := newTestTable(transport) + tab, db := newTestTable(transport, Config{ + Clock: simclock, + Log: testlog.Logger(t, log.LevelTrace), + }) defer db.Close() defer tab.close() <-tab.initDone // Fill up the sender's bucket. - pingKey, _ := crypto.HexToECDSA("45a915e4d060149eb4365960e6a7a45f334393093061116b197e3240065ff2d8") - pingSender := wrapNode(enode.NewV4(&pingKey.PublicKey, net.IP{127, 0, 0, 1}, 99, 99)) - last := fillBucket(tab, pingSender) + replacementNodeKey, _ := crypto.HexToECDSA("45a915e4d060149eb4365960e6a7a45f334393093061116b197e3240065ff2d8") + replacementNode := wrapNode(enode.NewV4(&replacementNodeKey.PublicKey, net.IP{127, 0, 0, 1}, 99, 99)) + last := fillBucket(tab, replacementNode) - // Add the sender as if it just pinged us. Revalidate should replace the last node in - // its bucket if it is unresponsive. Revalidate again to ensure that + // Add the sender as if it just pinged us. The revalidation process should replace + // this node in the bucket if it is unresponsive. transport.dead[last.ID()] = !lastInBucketIsResponding - transport.dead[pingSender.ID()] = !newNodeIsResponding - tab.addSeenNode(pingSender) - tab.doRevalidate(make(chan struct{}, 1)) - tab.doRevalidate(make(chan struct{}, 1)) + transport.dead[replacementNode.ID()] = !newNodeIsResponding + tab.addSeenNode(replacementNode) - if !transport.pinged[last.ID()] { - // Oldest node in bucket is pinged to see whether it is still alive. - t.Error("table did not ping last node in bucket") + // Wait until the last node was pinged. + waitForRevalidationPing(t, transport, tab, last.ID()) + + // If a replacement is expected, we also need to wait until the replacement node was pinged. + if !lastInBucketIsResponding { + waitForRevalidationPing(t, transport, tab, replacementNode.ID()) } tab.mutex.Lock() @@ -80,69 +87,41 @@ func testPingReplace(t *testing.T, newNodeIsResponding, lastInBucketIsResponding if !lastInBucketIsResponding && !newNodeIsResponding { wantSize-- } - if l := len(tab.bucket(pingSender.ID()).entries); l != wantSize { + bucket := tab.bucket(replacementNode.ID()) + if l := len(bucket.entries); l != wantSize { t.Errorf("wrong bucket size after bond: got %d, want %d", l, wantSize) } - if found := contains(tab.bucket(pingSender.ID()).entries, last.ID()); found != lastInBucketIsResponding { - t.Errorf("last entry found: %t, want: %t", found, lastInBucketIsResponding) + if ok := contains(bucket.entries, last.ID()); ok != lastInBucketIsResponding { + t.Errorf("last entry found: %t, want: %t", ok, lastInBucketIsResponding) } wantNewEntry := newNodeIsResponding && !lastInBucketIsResponding - if found := contains(tab.bucket(pingSender.ID()).entries, pingSender.ID()); found != wantNewEntry { - t.Errorf("new entry found: %t, want: %t", found, wantNewEntry) + if ok := contains(bucket.entries, replacementNode.ID()); ok != wantNewEntry { + t.Errorf("new entry found: %t, want: %t", ok, wantNewEntry) } } -func TestBucket_bumpNoDuplicates(t *testing.T) { - t.Parallel() - cfg := &quick.Config{ - MaxCount: 1000, - Rand: rand.New(rand.NewSource(time.Now().Unix())), - Values: func(args []reflect.Value, rand *rand.Rand) { - // generate a random list of nodes. this will be the content of the bucket. - n := rand.Intn(bucketSize-1) + 1 - nodes := make([]*node, n) - for i := range nodes { - nodes[i] = nodeAtDistance(enode.ID{}, 200, intIP(200)) - } - args[0] = reflect.ValueOf(nodes) - // generate random bump positions. - bumps := make([]int, rand.Intn(100)) - for i := range bumps { - bumps[i] = rand.Intn(len(nodes)) - } - args[1] = reflect.ValueOf(bumps) - }, - } - - prop := func(nodes []*node, bumps []int) (ok bool) { - tab, db := newTestTable(newPingRecorder()) - defer db.Close() - defer tab.close() - - b := &bucket{entries: make([]*node, len(nodes))} - copy(b.entries, nodes) - for i, pos := range bumps { - tab.bumpInBucket(b, b.entries[pos]) - if hasDuplicates(b.entries) { - t.Logf("bucket has duplicates after %d/%d bumps:", i+1, len(bumps)) - for _, n := range b.entries { - t.Logf(" %p", n) - } - return false - } +// waitForRevalidationPing waits until a PING message is sent to a node with the given id. +func waitForRevalidationPing(t *testing.T, transport *pingRecorder, tab *Table, id enode.ID) *enode.Node { + simclock := tab.cfg.Clock.(*mclock.Simulated) + maxAttempts := tab.len() * 5 + for i := 0; i < maxAttempts; i++ { + simclock.Run(tab.cfg.PingInterval) + p := transport.waitPing(500 * time.Millisecond) + if p == nil { + t.Fatal("Table did not send any revalidation ping") + } + if id == (enode.ID{}) || p.ID() == id { + return p } - checkIPLimitInvariant(t, tab) - return true - } - if err := quick.Check(prop, cfg); err != nil { - t.Error(err) } + t.Fatalf("Table did not ping node %v (%d attempts)", id, maxAttempts) + return nil } // This checks that the table-wide IP limit is applied correctly. func TestTable_IPLimit(t *testing.T) { transport := newPingRecorder() - tab, db := newTestTable(transport) + tab, db := newTestTable(transport, Config{}) defer db.Close() defer tab.close() @@ -159,7 +138,7 @@ func TestTable_IPLimit(t *testing.T) { // This checks that the per-bucket IP limit is applied correctly. func TestTable_BucketIPLimit(t *testing.T) { transport := newPingRecorder() - tab, db := newTestTable(transport) + tab, db := newTestTable(transport, Config{}) defer db.Close() defer tab.close() @@ -196,7 +175,7 @@ func TestTable_findnodeByID(t *testing.T) { test := func(test *closeTest) bool { // for any node table, Target and N transport := newPingRecorder() - tab, db := newTestTable(transport) + tab, db := newTestTable(transport, Config{}) defer db.Close() defer tab.close() fillTable(tab, test.All, true) @@ -271,7 +250,7 @@ func (*closeTest) Generate(rand *rand.Rand, size int) reflect.Value { } func TestTable_addVerifiedNode(t *testing.T) { - tab, db := newTestTable(newPingRecorder()) + tab, db := newTestTable(newPingRecorder(), Config{}) <-tab.initDone defer db.Close() defer tab.close() @@ -281,11 +260,12 @@ func TestTable_addVerifiedNode(t *testing.T) { n2 := nodeAtDistance(tab.self().ID(), 256, net.IP{88, 77, 66, 2}) tab.addSeenNode(n1) tab.addSeenNode(n2) + bucket := tab.bucket(n1.ID()) // Verify bucket content: bcontent := []*node{n1, n2} - if !reflect.DeepEqual(tab.bucket(n1.ID()).entries, bcontent) { - t.Fatalf("wrong bucket content: %v", tab.bucket(n1.ID()).entries) + if !reflect.DeepEqual(unwrapNodes(bucket.entries), unwrapNodes(bcontent)) { + t.Fatalf("wrong bucket content: %v", bucket.entries) } // Add a changed version of n2. @@ -295,15 +275,15 @@ func TestTable_addVerifiedNode(t *testing.T) { tab.addVerifiedNode(newn2) // Check that bucket is updated correctly. - newBcontent := []*node{newn2, n1} - if !reflect.DeepEqual(tab.bucket(n1.ID()).entries, newBcontent) { - t.Fatalf("wrong bucket content after update: %v", tab.bucket(n1.ID()).entries) + newBcontent := []*node{n1, newn2} + if !reflect.DeepEqual(unwrapNodes(bucket.entries), unwrapNodes(newBcontent)) { + t.Fatalf("wrong bucket content after update: %v", bucket.entries) } checkIPLimitInvariant(t, tab) } func TestTable_addSeenNode(t *testing.T) { - tab, db := newTestTable(newPingRecorder()) + tab, db := newTestTable(newPingRecorder(), Config{}) <-tab.initDone defer db.Close() defer tab.close() @@ -337,7 +317,10 @@ func TestTable_addSeenNode(t *testing.T) { // announces a new sequence number, the new record should be pulled. func TestTable_revalidateSyncRecord(t *testing.T) { transport := newPingRecorder() - tab, db := newTestTable(transport) + tab, db := newTestTable(transport, Config{ + Clock: new(mclock.Simulated), + Log: testlog.Logger(t, log.LevelTrace), + }) <-tab.initDone defer db.Close() defer tab.close() @@ -354,7 +337,8 @@ func TestTable_revalidateSyncRecord(t *testing.T) { n2 := enode.SignNull(&r, id) transport.updateRecord(n2) - tab.doRevalidate(make(chan struct{}, 1)) + waitForRevalidationPing(t, transport, tab, n2.ID()) + intable := tab.getNode(id) if !reflect.DeepEqual(intable, n2) { t.Fatalf("table contains old record with seq %d, want seq %d", intable.Seq(), n2.Seq()) diff --git a/p2p/discover/table_util_test.go b/p2p/discover/table_util_test.go index f5d4d39bdb..5eda147949 100644 --- a/p2p/discover/table_util_test.go +++ b/p2p/discover/table_util_test.go @@ -26,6 +26,8 @@ import ( "net" "slices" "sync" + "sync/atomic" + "time" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/p2p/enode" @@ -40,8 +42,7 @@ func init() { nullNode = enode.SignNull(&r, enode.ID{}) } -func newTestTable(t transport) (*Table, *enode.DB) { - cfg := Config{} +func newTestTable(t transport, cfg Config) (*Table, *enode.DB) { db, _ := enode.OpenDB("") tab, _ := newTable(t, db, cfg) go tab.loop() @@ -102,7 +103,9 @@ func fillBucket(tab *Table, n *node) (last *node) { ld := enode.LogDist(tab.self().ID(), n.ID()) b := tab.bucket(n.ID()) for len(b.entries) < bucketSize { - b.entries = append(b.entries, nodeAtDistance(tab.self().ID(), ld, intIP(ld))) + node := nodeAtDistance(tab.self().ID(), ld, intIP(ld)) + b.entries = append(b.entries, node) + tab.nodeAdded(b, node) } return b.entries[bucketSize-1] } @@ -119,10 +122,12 @@ func fillTable(tab *Table, nodes []*node, setLive bool) { } type pingRecorder struct { - mu sync.Mutex - dead, pinged map[enode.ID]bool - records map[enode.ID]*enode.Node - n *enode.Node + mu sync.Mutex + cond *sync.Cond + dead map[enode.ID]bool + records map[enode.ID]*enode.Node + pinged []*enode.Node + n *enode.Node } func newPingRecorder() *pingRecorder { @@ -130,12 +135,13 @@ func newPingRecorder() *pingRecorder { r.Set(enr.IP{0, 0, 0, 0}) n := enode.SignNull(&r, enode.ID{}) - return &pingRecorder{ + t := &pingRecorder{ dead: make(map[enode.ID]bool), - pinged: make(map[enode.ID]bool), records: make(map[enode.ID]*enode.Node), n: n, } + t.cond = sync.NewCond(&t.mu) + return t } // updateRecord updates a node record. Future calls to ping and @@ -151,12 +157,46 @@ func (t *pingRecorder) Self() *enode.Node { return nullNode } func (t *pingRecorder) lookupSelf() []*enode.Node { return nil } func (t *pingRecorder) lookupRandom() []*enode.Node { return nil } +func (t *pingRecorder) wasPinged(id enode.ID) bool { + t.mu.Lock() + defer t.mu.Unlock() + return slices.ContainsFunc(t.pinged, func(n *enode.Node) bool { return n.ID() == id }) +} + +func (t *pingRecorder) waitPing(timeout time.Duration) *enode.Node { + t.mu.Lock() + defer t.mu.Unlock() + + // Wake up the loop on timeout. + var timedout atomic.Bool + timer := time.AfterFunc(timeout, func() { + timedout.Store(true) + t.cond.Broadcast() + }) + defer timer.Stop() + + // Wait for a ping. + for { + if timedout.Load() { + return nil + } + if len(t.pinged) > 0 { + n := t.pinged[0] + t.pinged = append(t.pinged[:0], t.pinged[1:]...) + return n + } + t.cond.Wait() + } +} + // ping simulates a ping request. func (t *pingRecorder) ping(n *enode.Node) (seq uint64, err error) { t.mu.Lock() defer t.mu.Unlock() - t.pinged[n.ID()] = true + t.pinged = append(t.pinged, n) + t.cond.Broadcast() + if t.dead[n.ID()] { return 0, errTimeout } diff --git a/p2p/discover/v4_udp.go b/p2p/discover/v4_udp.go index c2de12133c..a6121d1923 100644 --- a/p2p/discover/v4_udp.go +++ b/p2p/discover/v4_udp.go @@ -142,7 +142,7 @@ func ListenV4(c UDPConn, ln *enode.LocalNode, cfg Config) (*UDPv4, error) { log: cfg.Log, } - tab, err := newMeteredTable(t, ln.Database(), cfg) + tab, err := newTable(t, ln.Database(), cfg) if err != nil { return nil, err } diff --git a/p2p/discover/v5_udp.go b/p2p/discover/v5_udp.go index 20a8bccd05..c6e65bfca2 100644 --- a/p2p/discover/v5_udp.go +++ b/p2p/discover/v5_udp.go @@ -175,7 +175,7 @@ func newUDPv5(conn UDPConn, ln *enode.LocalNode, cfg Config) (*UDPv5, error) { cancelCloseCtx: cancelCloseCtx, } t.talk = newTalkSystem(t) - tab, err := newMeteredTable(t, t.db, cfg) + tab, err := newTable(t, t.db, cfg) if err != nil { return nil, err }