p2p/discover: evict unresponsive nodes topdisc

p2p/discover/table.go

@@ -87,6 +87,7 @@ type Table struct {
 	closed          chan struct{}
 
 	nodeFeed        event.FeedOf[*enode.Node]
+	removedFeed     event.FeedOf[enode.ID]
 	nodeAddedHook   func(*bucket, *tableNode)
 	nodeRemovedHook func(*bucket, *tableNode)
 }
@@ -625,6 +626,7 @@ func (tab *Table) nodeAdded(b *bucket, n *tableNode) {
 
 func (tab *Table) nodeRemoved(b *bucket, n *tableNode) {
 	tab.revalidation.nodeRemoved(n)
+	tab.removedFeed.Send(n.ID())
 	if tab.nodeRemovedHook != nil {
 		tab.nodeRemovedHook(b, n)
 	}
@@ -749,6 +751,12 @@ func (tab *Table) subscribeNodes(ch chan *enode.Node) event.Subscription {
 	return tab.nodeFeed.Subscribe(ch)
 }
 
+// subscribeRemovedNodes subscribes to node removal events. The ID of each node
+// removed from the table (e.g. on failed revalidation) is sent on the channel.
+func (tab *Table) subscribeRemovedNodes(ch chan enode.ID) event.Subscription {
+	return tab.removedFeed.Subscribe(ch)
+}
+
 // allNodes returns all live nodes in the table.
 func (tab *Table) allNodes() []*enode.Node {
 	tab.mutex.Lock()

p2p/discover/topicindex/registration.go

@@ -214,6 +214,21 @@ func (r *Registration) AddNodes(src *enode.Node, nodes []*enode.Node) {
 	}
 }
 
+// RemoveNode drops registration attempts of a node evicted as dead
+func (r *Registration) RemoveNode(id enode.ID) {
+	b := &r.buckets[r.bucketIndex(id)]
+	att, ok := b.att[id]
+	if !ok {
+		return
+	}
+	//An in-flight attempt (index == -2) is left for its pending response to remove.
+	if att.index == -2 {
+		return
+	}
+	r.removeAttempt(att, "evicted")
+	r.refillAttempts(b)
+}
+
 func (r *Registration) setAttemptState(att *RegAttempt, state RegAttemptState) {
 	att.bucket.count[att.State]--
 	att.bucket.count[state]++

p2p/discover/topicindex/registration_test.go

@@ -317,6 +317,54 @@ func TestRegistrationExpiry(t *testing.T) {
 	}
 }
 
+// TestRegistrationRemoveNode checks that RemoveNode drops a parked attempt but
+// leaves an in-flight one for its pending response.
+func TestRegistrationRemoveNode(t *testing.T) {
+	cfg := testConfig(t)
+	r := NewRegistration(topic1, cfg)
+
+	// Waiting attempt: removed.
+	waiting := nodesAtDistance(enode.ID(r.Topic()), 30, 1)
+	r.AddNodes(nil, waiting)
+	if att := r.Update(); att == nil || att.State != Waiting {
+		t.Fatal("no waiting attempt scheduled")
+	}
+	r.RemoveNode(waiting[0].ID())
+	if r.buckets[r.bucketIndex(waiting[0].ID())].att[waiting[0].ID()] != nil {
+		t.Fatal("waiting attempt not removed")
+	}
+
+	// Registered attempt (ad not yet expired): removed before expiry.
+	registered := nodesAtDistance(enode.ID(r.Topic()), 40, 1)
+	r.AddNodes(nil, registered)
+	att := r.Update()
+	if att == nil {
+		t.Fatal("no attempt scheduled")
+	}
+	r.StartRequest(att)
+	r.HandleRegistered(att, cfg.AdLifetime)
+	r.RemoveNode(registered[0].ID())
+	if r.buckets[r.bucketIndex(registered[0].ID())].att[registered[0].ID()] != nil {
+		t.Fatal("registered attempt not removed before ad expiry")
+	}
+
+	// Unknown node: no-op.
+	r.RemoveNode(enode.ID{42})
+
+	// In-flight attempt: left in place, the pending response handles it.
+	inflight := nodesAtDistance(enode.ID(r.Topic()), 50, 1)
+	r.AddNodes(nil, inflight)
+	att = r.Update()
+	if att == nil {
+		t.Fatal("no attempt scheduled")
+	}
+	r.StartRequest(att)
+	r.RemoveNode(inflight[0].ID())
+	if r.buckets[r.bucketIndex(inflight[0].ID())].att[inflight[0].ID()] == nil {
+		t.Fatal("in-flight attempt removed; must be left for response handling")
+	}
+}
+
 // TestRegistrationHandleTicketResponseDropAboveBudget verifies that a
 // registrar quoting a wait time above RegAttemptTimeout causes the attempt
 // to be dropped, freeing the bucket slot for another registrar. Without

p2p/discover/topicindex/search.go

@@ -53,7 +53,7 @@ type Search struct {
 type searchBucket struct {
 	dist        int
 	new         map[enode.ID]*enode.Node
-	asked       map[enode.ID]struct{}
+	asked       map[enode.ID]*enode.Node
 	numRequests int
 
 	ips netutil.DistinctNetSet
@@ -74,7 +74,7 @@ func NewSearch(topic TopicID, cfg Config) *Search {
 		s.buckets[i] = searchBucket{
 			dist:  dist,
 			new:   make(map[enode.ID]*enode.Node, cfg.SearchBucketSize),
-			asked: make(map[enode.ID]struct{}, cfg.SearchBucketSize),
+			asked: make(map[enode.ID]*enode.Node, cfg.SearchBucketSize),
 			ips: netutil.DistinctNetSet{
 				Subnet: searchBucketSubnet,
 				Limit:  searchBucketIPLimit,
@@ -155,6 +155,31 @@ func (s *Search) AddNodes(src *enode.Node, nodes []*enode.Node) {
 	}
 }
 
+// HandleErrorResponse drops a failed node from the table, freeing its slot and
+// IP-limit entry.
+func (s *Search) HandleErrorResponse(from *enode.Node, err error) {
+	s.log.Debug("Topic query failed", "id", from.ID(), "err", err)
+	s.removeNode(from.ID())
+}
+
+// removeNode drops a node from the search table. The node is removed from both
+// the unasked ('new') and asked sets of its bucket, and its IP-limit entry is
+// released regardless of which set it was in.
+func (s *Search) removeNode(id enode.ID) {
+	b := s.bucket(id)
+	n, ok := b.new[id]
+	if !ok {
+		n, ok = b.asked[id]
+	}
+	if ok {
+		if ip := n.IP(); ip != nil && !netutil.IsLAN(ip) {
+			b.ips.Remove(ip)
+		}
+	}
+	delete(b.new, id)
+	delete(b.asked, id)
+}
+
 // QueryTarget returns a random node to which a topic query should be sent.
 // Random nodes are collected from buckets progressively: only buckets with unasked nodes
 // that have received at least one response, plus the next unqueried bucket
@@ -241,6 +266,6 @@ func (b *searchBucket) count() int {
 }
 
 func (b *searchBucket) setAsked(n *enode.Node) {
-	b.asked[n.ID()] = struct{}{}
+	b.asked[n.ID()] = n
 	delete(b.new, n.ID())
 }

p2p/discover/topicindex/search_test.go

@@ -17,6 +17,7 @@
 package topicindex
 
 import (
+	"errors"
 	"testing"
 
 	"github.com/ethereum/go-ethereum/p2p/enode"
@@ -169,6 +170,88 @@ func TestSearchAddNodesOnePerBucketRule(t *testing.T) {
 	}
 }
 
+// TestSearchHandleErrorResponse checks that a failed topic query drops the
+// queried node from the table and frees its bucket slot and IP-limit entry.
+func TestSearchHandleErrorResponse(t *testing.T) {
+	config := testConfig(t)
+	s := NewSearch(topic1, config)
+
+	// Two nodes in bucket 0 (logdist 256), one in bucket 5 (logdist 251).
+	far := nodesAtDistanceFrom(enode.ID(topic1), 256, 2, 1)
+	mid := nodesAtDistanceFrom(enode.ID(topic1), 251, 1, 10)
+	s.AddNodes(nil, far)
+	s.AddNodes(nil, mid)
+
+	// The query to far[0] fails: the node must leave the table entirely.
+	s.HandleErrorResponse(far[0], errors.New("timeout"))
+	if s.buckets[0].contains(far[0].ID()) {
+		t.Fatal("failed node still present in its bucket")
+	}
+	if got := s.buckets[0].count(); got != 1 {
+		t.Fatalf("bucket[0] count is %d after eviction, want 1", got)
+	}
+
+	// The IP-limit slot is freed: a replacement in the same /24 is accepted.
+	replacement := nodeAtDistance(enode.ID(topic1), 256, far[0].IP())
+	s.AddNodes(nil, []*enode.Node{replacement})
+	if !s.buckets[0].contains(replacement.ID()) {
+		t.Fatal("replacement with the failed node's IP was not admitted")
+	}
+
+	// The failure did not count as a response: bucket 0 still has candidates
+	// and no response yet, so it keeps gating the walk.
+	if n := s.QueryTarget(); n == nil || !s.buckets[0].contains(n.ID()) {
+		t.Fatalf("QueryTarget should keep picking from unwarmed bucket[0], got %v", n)
+	}
+
+	// Failing all of bucket 0 empties it; the empty bucket no longer blocks
+	// the walk, so QueryTarget advances to bucket 5.
+	s.HandleErrorResponse(far[1], errors.New("timeout"))
+	s.HandleErrorResponse(replacement, errors.New("timeout"))
+	target := s.QueryTarget()
+	if target == nil {
+		t.Fatal("QueryTarget returned nil after bucket[0] failed out; want bucket[5] node")
+	}
+	if !s.buckets[5].contains(target.ID()) {
+		t.Fatalf("QueryTarget returned %v, want the bucket[5] node", target.ID())
+	}
+
+	// When every node has failed, the search is done and rolls over.
+	s.HandleErrorResponse(mid[0], errors.New("timeout"))
+	if !s.IsDone() {
+		t.Fatal("IsDone should report true once every node has failed out")
+	}
+}
+
+// TestSearchRemoveAskedNodeFreesIP verifies that removing a node that has moved
+// to the 'asked' set (it responded before being removed) still releases its
+// IP-limit entry, so a same-/24 replacement is admitted.
+func TestSearchRemoveAskedNodeFreesIP(t *testing.T) {
+	config := testConfig(t)
+	s := NewSearch(topic1, config)
+
+	nodes := nodesAtDistanceFrom(enode.ID(topic1), 256, 1, 1)
+	n := nodes[0]
+	s.AddNodes(nil, nodes)
+
+	// Move n into the 'asked' set by recording a query response from it.
+	s.AddQueryResults(n, nil)
+	if _, inAsked := s.buckets[0].asked[n.ID()]; !inAsked {
+		t.Fatal("node was not moved to the asked set")
+	}
+
+	// Removing it now must free the IP-limit slot even though it is in 'asked'.
+	s.HandleErrorResponse(n, errors.New("timeout"))
+	if s.buckets[0].contains(n.ID()) {
+		t.Fatal("removed asked-node still present")
+	}
+	replacement := nodeAtDistance(enode.ID(topic1), 256, n.IP())
+	s.AddNodes(nil, []*enode.Node{replacement})
+	if !s.buckets[0].contains(replacement.ID()) {
+		t.Fatal("replacement with the asked node's IP was not admitted (IP slot leaked)")
+	}
+}
+
 // This checks (de)queueing of topic search results: results come out of the
 // buffer in the order they were received.
 func TestSearchResultsTracking(t *testing.T) {

p2p/discover/topicindex/topictable.go

@@ -194,6 +194,21 @@ func (tab *TopicTable) add(n *enode.Node, topic TopicID) *topicTableEntry {
 	return reg
 }
 
+// RemoveNode removes all registrations advertised by the given node, across all
+// topics. It is used to evict ads pointing at a node that has become
+// unresponsive.
+func (tab *TopicTable) RemoveNode(id enode.ID) {
+	for e := tab.all.Front(); e != nil; {
+		next := e.Next()
+		reg := e.Value.(*topicTableEntry)
+		if reg.node.ID() == id {
+			tab.remove(reg)
+			tab.wt.removeReg(reg)
+		}
+		e = next
+	}
+}
+
 func (tab *TopicTable) remove(reg *topicTableEntry) {
 	tab.all.Remove(reg.allElem)
 	topicList := tab.reg[reg.topic]