Keep a peer-down until its route delete succeeds (closes #101)

When the delete of a peer's live routes after its session went down
failed, the down was dropped and the routes stayed. The prefix handler
now keeps such a peer, with the updates it sent after the down, for the
next flush, which retries the delete before it writes them; each
failure is logged. While the delete keeps failing, only the last 25,000
of those updates are kept, and the older ones are dropped and logged.
The delete runs in chunks of at most 10,000 routes per table, taken in
prefix order through the peer's key, each chunk in its own transaction,
so the database lock is released between chunks and a failure undoes
at most one chunk.

Model: opus-5-5

README.md

@@ -318,8 +318,14 @@ What happens at each limit:
   (`peerBatchMaxSize`, about 10 MiB) and 250,000 for the prefix handler
   (`prefixBatchMaxSize`, about 60 MiB). At that size the handler takes nothing
   more off its queue until the write ends, so the queue fills and the streamer
-  drops messages as above. With the batches being written, the three handlers'
-  batches take at most about 0.25 GiB of the Go heap's 1.5 GiB.
+  drops messages as above. When the delete of a peer's live routes after its
+  session went down fails, the prefix handler keeps the updates the peer sent
+  after the down for the next write, which retries the delete first; on top of
+  its 250,000 it keeps at most the last 25,000 of them (`prefixBatchSize`, about
+  6 MiB) and drops and logs the older ones, so while the delete keeps failing
+  the other peers' updates still come off the queue. With the batches being
+  written, the three handlers' batches take at most about 0.25 GiB of the Go
+  heap's 1.5 GiB.
 
 With `DEBUG=routewatch` the daemon logs a `System stats` line every 60 seconds
 with the goroutine count and Go memory figures.

internal/routewatch/prefixhandler.go

@@ -28,7 +28,9 @@ const (
 	// prefixBatchMaxSize is the most prefix updates the batch holds while the
 	// flush goroutine writes the previous one: ten batches, about 60 MiB
 	// (README "Memory"). At this size HandleMessage waits until the flush
-	// goroutine takes the batch.
+	// goroutine takes the batch. On top of these, the batch holds at most
+	// prefixBatchSize updates, about 6 MiB, that a flush kept for the peers
+	// whose route delete failed (keepForNextFlush).
 	prefixBatchMaxSize = 250000
 
 	// prefixBatchTimeout is the maximum time to wait before flushing a batch
@@ -384,6 +386,13 @@ func (h *PrefixHandler) deleteLiveRoutesOfPeers(peers []string) []string {
 // session went down again since the batch was taken came before that down, so
 // they are dropped, as peerSessionDown drops them. It returns the updates of the
 // other peers, for this flush to write.
+//
+// While a delete keeps failing, the kept updates would grow with every flush,
+// so only the last prefixBatchSize of them are kept and the older ones are
+// dropped. This never leaves in place a route the peer withdrew: a prefix with
+// a kept update keeps its last one, and the retry deletes all of the peer's
+// routes, so a prefix whose updates were all dropped has no route from the peer
+// until the peer announces it again.
 func (h *PrefixHandler) keepForNextFlush(
 	batch []prefixUpdate, failedPeers []string,
 ) []prefixUpdate {
@@ -403,6 +412,15 @@ func (h *PrefixHandler) keepForNextFlush(
 	kept = slices.DeleteFunc(kept, func(update prefixUpdate) bool {
 		return slices.Contains(h.peersDown, update.peer)
 	})
+
+	if len(kept) > prefixBatchSize {
+		h.logger.Error(
+			"Dropped the oldest updates of peers whose route delete failed",
+			"peers", failedPeers, "dropped", len(kept)-prefixBatchSize)
+
+		kept = kept[len(kept)-prefixBatchSize:]
+	}
+
 	h.batch = slices.Concat(kept, h.batch)
 	h.peersDown = slices.Concat(failedPeers, h.peersDown)
 

internal/routewatch/prefixhandler_internal_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net/netip"
 	"slices"
 	"sync"
 	"testing"
@@ -560,8 +561,8 @@ func TestPrefixHandlerHandlesADownWhileAFlushWrites(t *testing.T) {
 	})
 }
 
-// errTestPeerDelete is the error firstPeerDeleteFailsStore's first delete
-// returns.
+// errTestPeerDelete is the error a test store's failing delete of a peer's
+// routes returns.
 var errTestPeerDelete = errors.New("test: the delete of a peer's routes failed")
 
 // firstPeerDeleteFailsStore is a real database whose first delete of a peer's
@@ -678,6 +679,71 @@ func TestPrefixHandlerRetriesAFailedDeleteOfAPeersRoutes(t *testing.T) {
 	}
 }
 
+// peerDeleteAlwaysFailsStore is a real database whose every delete of a peer's
+// live routes fails, deleting nothing.
+type peerDeleteAlwaysFailsStore struct {
+	*database.Database
+}
+
+// DeleteLiveRoutesOfPeer fails.
+func (peerDeleteAlwaysFailsStore) DeleteLiveRoutesOfPeer(string) (int, error) {
+	return 0, errTestPeerDelete
+}
+
+// TestPrefixHandlerKeepsAtMostABatchOfUpdatesWhileADeleteKeepsFailing runs,
+// through the prefix handler into a database whose every delete of a peer's
+// routes fails, a peer's session going down and then three flushes. Before each
+// flush the peer announces prefixBatchSize new prefixes and another peer
+// announces one. After each flush the batch must hold only the peer's last
+// prefixBatchSize updates, not every update since the down, and the other
+// peer's route must be written.
+func TestPrefixHandlerKeepsAtMostABatchOfUpdatesWhileADeleteKeepsFailing(
+	t *testing.T,
+) {
+	t.Parallel()
+
+	const otherPeerIP = "2001:db8:ffff::2"
+
+	// One prefix the other peer announces before each flush.
+	otherPeersPrefixes := []string{testV4Prefix, testV6Prefix, "2001:db8:2::/48"}
+
+	db := newTestDatabase(t)
+
+	// Built without NewPrefixHandler's flush timer, so the test flushes itself.
+	h := &PrefixHandler{
+		db: peerDeleteAlwaysFailsStore{Database: db}, logger: logger.New(),
+	}
+	h.HandleMessage(sessionDown())
+
+	// The peer announces a /32 for each address from 10.0.0.0 on, each
+	// address once.
+	addr := netip.MustParseAddr("10.0.0.0")
+
+	for flush, otherPeersPrefix := range otherPeersPrefixes {
+		prefixes := make([]string, prefixBatchSize)
+		for i := range prefixes {
+			prefixes[i] = netip.PrefixFrom(addr, addr.BitLen()).String()
+			addr = addr.Next()
+		}
+
+		h.HandleMessage(peerAnnouncement(testPeerIP, prefixes...))
+		h.HandleMessage(peerAnnouncement(otherPeerIP, otherPeersPrefix))
+		h.flushBatch()
+
+		kept := make([]string, 0, len(h.batch))
+		for _, update := range h.batch {
+			kept = append(kept, update.prefix)
+		}
+
+		if !slices.Equal(kept, prefixes) {
+			t.Fatalf("after flush %d the batch holds %d updates, not just "+
+				"the peer's last %d", flush+1, len(kept), len(prefixes))
+		}
+
+		assertPeers(t, db, map[string][]string{otherPeersPrefix: {otherPeerIP}})
+	}
+}
+
 // newTestDatabase opens a database in a new temporary directory. It is closed
 // when the test ends.
 func newTestDatabase(t *testing.T) *database.Database {