Take every down peer out of the table in one walk (closes #225)

The prefix handler ran one PeerDown per peer, one after another. When
the sessions of several peers went down together, their walks queued up.
A returning full-table peer's messages stayed held back while the walks
ahead of its own ran, and could reach the limit of 20,000.

The table's PeersDown now takes several peers out of every prefix in one
walk. One walk runs at a time. The next one takes every peer whose
session went down since the previous one began. The doc of
maxHeldBackMessages says how long a peer's messages are held back. The
limit is now 150,000 messages, about 150 MiB. The status page shows how
many messages each handler holds back.

Model: opus-5-5

internal/server/handlers.go

@@ -486,6 +486,7 @@ type HandlerStatsInfo struct {
 	QueueHighWaterMark int     `json:"queue_high_water_mark"`
 	ProcessedCount     uint64  `json:"processed_count"`
 	DroppedCount       uint64  `json:"dropped_count"`
+	HeldBackCount      int     `json:"held_back_count"`
 	AvgProcessTimeMs   float64 `json:"avg_process_time_ms"`
 	MinProcessTimeMs   float64 `json:"min_process_time_ms"`
 	MaxProcessTimeMs   float64 `json:"max_process_time_ms"`
@@ -570,6 +571,7 @@ func handlerStatsInfo(handlerStats []streamer.HandlerStats) []HandlerStatsInfo {
 			QueueHighWaterMark: hs.QueueHighWaterMark,
 			ProcessedCount:     hs.ProcessedCount,
 			DroppedCount:       hs.DroppedCount,
+			HeldBackCount:      hs.HeldBackCount,
 			AvgProcessTimeMs: float64(hs.AvgProcessTime.Microseconds()) /
 				microsecondsPerMillisecond,
 			MinProcessTimeMs: float64(hs.MinProcessTime.Microseconds()) /

internal/server/handlers_internal_test.go

@@ -295,6 +295,43 @@ func TestStatsAnswerCarriesChurnPerMinute(t *testing.T) {
 	}
 }
 
+// heldBackMessages is how many messages heldBackHandler holds back.
+const heldBackMessages = 7
+
+// heldBackHandler is a stream handler that wants no message and holds back
+// heldBackMessages messages.
+type heldBackHandler struct{}
+
+func (heldBackHandler) WantsMessage(string) bool           { return false }
+func (heldBackHandler) HandleMessage(*ristypes.RISMessage) {}
+func (heldBackHandler) QueueCapacity() int                 { return 1 }
+func (heldBackHandler) HeldBackCount() int                 { return heldBackMessages }
+
+// TestStatsAnswerGivesHowManyMessagesAHandlerHoldsBack registers a handler
+// that holds back messages. Its figures in /api/v1/stats, which the status
+// page shows, must give how many.
+func TestStatsAnswerGivesHowManyMessagesAHandlerHoldsBack(t *testing.T) {
+	t.Parallel()
+
+	s := newServerWithTestRoutes(t)
+	s.streamer.RegisterHandler(heldBackHandler{})
+
+	var body struct {
+		Data StatsResponse `json:"data"`
+	}
+
+	err := json.Unmarshal(get(t, s, statsPath).Body.Bytes(), &body)
+	if err != nil {
+		t.Fatalf("decoding the answer: %v", err)
+	}
+
+	stats := body.Data.HandlerStats
+	if len(stats) != 1 || stats[0].HeldBackCount != heldBackMessages {
+		t.Errorf("handler figures %+v, want one handler holding back %d "+
+			"messages", stats, heldBackMessages)
+	}
+}
+
 // fixedIntervalASNFetcher is a WHOIS fetcher that does nothing and reports
 // interval as its wait between fetches.
 type fixedIntervalASNFetcher struct {