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, once, with the updates it sent after the down,
for the next flush, which retries the delete before it writes them; each
failure is logged. At most the last 25,000 of those updates are kept,
outside the batch, so they do not start a flush. 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

internal/routewatch/prefixhandler_internal_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net/netip"
 	"slices"
 	"sync"
 	"testing"
@@ -560,6 +561,211 @@ func TestPrefixHandlerHandlesADownWhileAFlushWrites(t *testing.T) {
 	})
 }
 
+// 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
+// live routes fails, deleting nothing. deletes counts the deletes asked for.
+type firstPeerDeleteFailsStore struct {
+	*database.Database
+
+	// whileFailing, if not nil, runs during the delete that fails.
+	whileFailing func()
+	deletes      int
+}
+
+// DeleteLiveRoutesOfPeer fails the first call and deletes the peer's routes on
+// every later one.
+func (s *firstPeerDeleteFailsStore) DeleteLiveRoutesOfPeer(
+	peerIP string,
+) (int, error) {
+	s.deletes++
+	if s.deletes > 1 {
+		return s.Database.DeleteLiveRoutesOfPeer(peerIP)
+	}
+
+	if s.whileFailing != nil {
+		s.whileFailing()
+	}
+
+	return 0, errTestPeerDelete
+}
+
+// TestPrefixHandlerRetriesAFailedDeleteOfAPeersRoutes runs, through the prefix
+// handler into a database whose first delete of a peer's routes fails, one flush
+// after another: a peer's announcement, then its session going down and a new
+// announcement from the peer, then nothing. The failed delete must be retried
+// on the next flush, and the announcement the peer sent after the down written
+// only after that retry, so it is kept. In the other cases the peer's session
+// goes down again, and the peer announces another prefix, while the delete
+// fails or after the flush whose delete failed: the announcement sent between
+// the two downs must not be written, and the next flush must delete the peer's
+// routes once, not once for the retry and once for the down.
+func TestPrefixHandlerRetriesAFailedDeleteOfAPeersRoutes(t *testing.T) {
+	t.Parallel()
+
+	const (
+		beforeDown      = "192.0.2.0/24"
+		afterDown       = "198.51.100.0/24"
+		afterSecondDown = "203.0.113.0/24"
+		maskLength      = 24
+		// The delete that fails and its retry, which is also the delete for
+		// a second down.
+		wantDeletes = 2
+	)
+
+	tests := []struct {
+		name string
+		// Handled while the delete fails, if not nil.
+		whileFailing []*ristypes.RISMessage
+		// Handled after the flush whose delete fails, if not nil.
+		afterFailing []*ristypes.RISMessage
+		// The peers left with a live route for each prefix.
+		wantPeers map[string][]string
+	}{
+		{
+			name: "the delete fails once",
+			wantPeers: map[string][]string{
+				beforeDown: nil, afterDown: {testPeerIP}, afterSecondDown: nil,
+			},
+		},
+		{
+			name: "the session goes down again while the delete fails",
+			whileFailing: []*ristypes.RISMessage{
+				sessionDown(), peerAnnouncement(testPeerIP, afterSecondDown),
+			},
+			wantPeers: map[string][]string{
+				beforeDown: nil, afterDown: nil, afterSecondDown: {testPeerIP},
+			},
+		},
+		{
+			name: "the session goes down again after the delete failed",
+			afterFailing: []*ristypes.RISMessage{
+				sessionDown(), peerAnnouncement(testPeerIP, afterSecondDown),
+			},
+			wantPeers: map[string][]string{
+				beforeDown: nil, afterDown: nil, afterSecondDown: {testPeerIP},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			db := newTestDatabase(t)
+			store := &firstPeerDeleteFailsStore{Database: db}
+
+			// Built without NewPrefixHandler's flush timer, so the test
+			// flushes itself.
+			h := &PrefixHandler{db: store, logger: logger.New()}
+			store.whileFailing = func() {
+				for _, msg := range tt.whileFailing {
+					h.HandleMessage(msg)
+				}
+			}
+
+			h.HandleMessage(peerAnnouncement(testPeerIP, beforeDown))
+			h.flushBatch()
+			h.HandleMessage(sessionDown())
+			h.HandleMessage(peerAnnouncement(testPeerIP, afterDown))
+			h.flushBatch() // the delete fails
+
+			for _, msg := range tt.afterFailing {
+				h.HandleMessage(msg)
+			}
+
+			h.flushBatch()
+
+			if store.deletes != wantDeletes {
+				t.Errorf("the store was asked for %d deletes of the peer's "+
+					"routes, want %d", store.deletes, wantDeletes)
+			}
+
+			assertPeers(t, db, tt.wantPeers)
+			assertCounts(t, db, ipv4Version, 1, []database.PrefixDistribution{
+				{MaskLength: maskLength, Count: 1},
+			})
+		})
+	}
+}
+
+// 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 handler must hold only the peer's last
+// prefixBatchSize updates, not every update since the down, and the other
+// peer's route must be written. After the last flush, one more update must not
+// wake the flush goroutine: held updates do not count toward a full batch.
+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.held))
+		for _, update := range h.held {
+			kept = append(kept, update.prefix)
+		}
+
+		if !slices.Equal(kept, prefixes) {
+			t.Fatalf("after flush %d the handler holds %d updates, not just "+
+				"the peer's last %d", flush+1, len(kept), len(prefixes))
+		}
+
+		assertPeers(t, db, map[string][]string{otherPeersPrefix: {otherPeerIP}})
+	}
+
+	// A new channel, so that a wake-up in it comes from the update below.
+	h.batchFullCh = make(chan struct{}, 1)
+	h.HandleMessage(peerAnnouncement(otherPeerIP, testV4Prefix))
+
+	if len(h.batchFullCh) > 0 {
+		t.Error("one update after a flush that held a batch of updates " +
+			"woke the flush goroutine")
+	}
+}
+
 // newTestDatabase opens a database in a new temporary directory. It is closed
 // when the test ends.
 func newTestDatabase(t *testing.T) *database.Database {