diff --git a/README.md b/README.md index 3e5a2e7..05ca2a5 100644 --- a/README.md +++ b/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. diff --git a/internal/database/counts_internal_test.go b/internal/database/counts_internal_test.go index bbe5577..76f52be 100644 --- a/internal/database/counts_internal_test.go +++ b/internal/database/counts_internal_test.go @@ -2,6 +2,7 @@ package database import ( "context" + "fmt" "slices" "sync" "testing" @@ -494,6 +495,124 @@ func TestDeleteLiveRoutesOfPeer(t *testing.T) { checkOneLiveRoutePerIPVersion(t, db) } +// TestDeleteLiveRoutesOfPeerDeletesInChunks gives a peer one IPv4 route more +// than a chunk holds, two of them for prefixes another peer also has. One chunk +// deletes peerRouteChunkSize of the peer's routes and commits, leaving the last +// one. The routes are written again, and DeleteLiveRoutesOfPeer deletes all of +// them, a chunk and then the last one. After each step the route count and +// prefix distribution held in memory and those read from the route table are +// the ones wanted. +func TestDeleteLiveRoutesOfPeerDeletesInChunks(t *testing.T) { + t.Parallel() + + const ( + peerIP = "192.0.2.1" + otherPeerIP = "192.0.2.2" + // How many of the peer's prefixes the other peer also has, the first + // ones in prefix order. + sharedPrefixes = 2 + ) + + db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New()) + if err != nil { + t.Fatalf("failed to create database: %v", err) + } + defer func() { _ = db.Close() }() + + ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + + peerRoutes := make([]*LiveRoute, 0, peerRouteChunkSize+1) + otherPeerRoutes := make([]*LiveRoute, 0, sharedPrefixes) + + for i := range peerRouteChunkSize + 1 { + prefix := fmt.Sprintf("10.%d.%d.0/24", i/256, i%256) + + route := mkV4Route(t, prefix, 64500, ts) + route.PeerIP = peerIP + peerRoutes = append(peerRoutes, route) + + if i < sharedPrefixes { + route := mkV4Route(t, prefix, 64500, ts) + route.PeerIP = otherPeerIP + otherPeerRoutes = append(otherPeerRoutes, route) + } + } + + err = db.UpsertLiveRouteBatch(slices.Concat(peerRoutes, otherPeerRoutes)) + if err != nil { + t.Fatalf("UpsertLiveRouteBatch: %v", err) + } + + deletedV4, deletedV6, err := db.deletePeerRouteChunk(peerIP) + if err != nil { + t.Fatalf("deletePeerRouteChunk: %v", err) + } + + if deletedV4 != peerRouteChunkSize || deletedV6 != 0 { + t.Errorf("deletePeerRouteChunk deleted (%d, %d) routes, want (%d, 0)", + deletedV4, deletedV6, peerRouteChunkSize) + } + + // The shared prefixes went with the chunk, so each route left has a + // prefix of its own. + checkV4RoutesOfOwnPrefixes(t, db, "after one chunk", 1+sharedPrefixes) + + err = db.UpsertLiveRouteBatch(peerRoutes) + if err != nil { + t.Fatalf("UpsertLiveRouteBatch: %v", err) + } + + deleted, err := db.DeleteLiveRoutesOfPeer(peerIP) + if err != nil { + t.Fatalf("DeleteLiveRoutesOfPeer: %v", err) + } + + if deleted != len(peerRoutes) { + t.Errorf("DeleteLiveRoutesOfPeer deleted %d routes, want %d", + deleted, len(peerRoutes)) + } + + checkV4RoutesOfOwnPrefixes(t, db, "after DeleteLiveRoutesOfPeer", + sharedPrefixes) +} + +// checkV4RoutesOfOwnPrefixes checks that the database has wantRoutes IPv4 live +// routes, each for a /24 no other live route has, both as counted in memory and +// as read from the route table. when says which step of the test checks. +func checkV4RoutesOfOwnPrefixes( + t *testing.T, db *Database, when string, wantRoutes int, +) { + t.Helper() + + ctx := context.Background() + + stats, err := db.GetStatsContext(ctx) + if err != nil { + t.Fatalf("GetStatsContext: %v", err) + } + + tableRoutes, _, err := db.GetLiveRouteCountsContext(ctx) + if err != nil { + t.Fatalf("GetLiveRouteCountsContext: %v", err) + } + + if stats.IPv4Routes != wantRoutes || tableRoutes != wantRoutes { + t.Errorf("%s: IPv4 live routes: %d in memory, %d in the table, want %d", + when, stats.IPv4Routes, tableRoutes, wantRoutes) + } + + queryDistribution, _, err := db.GetPrefixDistributionContext(ctx) + if err != nil { + t.Fatalf("GetPrefixDistributionContext: %v", err) + } + + want := []PrefixDistribution{{MaskLength: 24, Count: wantRoutes}} + assertDistribution(t, when+": IPv4 distribution", + stats.IPv4PrefixDistribution, want) + assertDistribution(t, when+": IPv4 distribution query", + queryDistribution, want) +} + // checkOneLiveRoutePerIPVersion checks that the database has one IPv4 live // route, for a /24, and one IPv6 live route, for a /32, both as counted in // memory and as read from the route tables. diff --git a/internal/database/database.go b/internal/database/database.go index c4832bb..3d4e89e 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -658,48 +658,71 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error { return nil } -// Deleting all live routes of a peer. The first query of each table selects the -// mask length of each of the peer's routes whose prefix no other peer has a -// live route for: the delete takes those prefixes out of the prefix -// distribution. -const ( - peerOnlyRouteMaskLengthsV4SQL = `SELECT r.mask_length - FROM live_routes_v4 AS r WHERE r.peer_ip = ? - AND NOT EXISTS (SELECT 1 FROM live_routes_v4 AS o - WHERE o.prefix = r.prefix AND o.peer_ip != r.peer_ip)` - deletePeerLiveRoutesV4SQL = `DELETE FROM live_routes_v4 WHERE peer_ip = ?` +// peerRouteChunkSize is the most routes of a peer that one transaction of +// DeleteLiveRoutesOfPeer deletes from each route table. A peer can have a +// million routes; deleted in one transaction, they would hold the database lock +// for most of a minute. +const peerRouteChunkSize = 10000 - peerOnlyRouteMaskLengthsV6SQL = `SELECT r.mask_length +// Deleting a chunk of a peer's live routes. The first query of each table +// selects the peer's first routes in prefix order, through the +// UNIQUE(peer_ip, prefix) key, each with its mask length and whether another +// peer has a live route for its prefix: the delete takes the prefixes no other +// peer has out of the prefix distribution. The second query deletes the peer's +// routes up to the last prefix the first one selected. +const ( + peerRouteChunkV4SQL = `SELECT r.prefix, r.mask_length, + EXISTS (SELECT 1 FROM live_routes_v4 AS o + WHERE o.prefix = r.prefix AND o.peer_ip != r.peer_ip) + FROM live_routes_v4 AS r WHERE r.peer_ip = ? + ORDER BY r.prefix LIMIT ?` + deletePeerRouteChunkV4SQL = `DELETE FROM live_routes_v4 + WHERE peer_ip = ? AND prefix <= ?` + + peerRouteChunkV6SQL = `SELECT r.prefix, r.mask_length, + EXISTS (SELECT 1 FROM live_routes_v6 AS o + WHERE o.prefix = r.prefix AND o.peer_ip != r.peer_ip) FROM live_routes_v6 AS r WHERE r.peer_ip = ? - AND NOT EXISTS (SELECT 1 FROM live_routes_v6 AS o - WHERE o.prefix = r.prefix AND o.peer_ip != r.peer_ip)` - deletePeerLiveRoutesV6SQL = `DELETE FROM live_routes_v6 WHERE peer_ip = ?` + ORDER BY r.prefix LIMIT ?` + deletePeerRouteChunkV6SQL = `DELETE FROM live_routes_v6 + WHERE peer_ip = ? AND prefix <= ?` ) -// deletePeerRoutes deletes all live routes of peerIP from one table, given that -// table's two statements above. It returns how many routes it deleted and the -// mask length of each prefix it left with no live route. +// deletePeerRoutes deletes the first peerRouteChunkSize live routes of peerIP, +// or as many as it has, from one table, given that table's two statements +// above. It returns how many routes it deleted and the mask length of each +// prefix it left with no live route. func deletePeerRoutes( - ctx context.Context, tx *loggingTx, - onlyRouteMaskLengthsSQL, deleteSQL, peerIP string, + ctx context.Context, tx *loggingTx, chunkSQL, deleteSQL, peerIP string, ) (int64, []int, error) { - rows, err := tx.QueryContext(ctx, onlyRouteMaskLengthsSQL, peerIP) + rows, err := tx.QueryContext(ctx, chunkSQL, peerIP, peerRouteChunkSize) if err != nil { return 0, nil, err } defer func() { _ = rows.Close() }() - var goneMaskLengths []int + var ( + selected int + lastPrefix string + goneMaskLengths []int + ) for rows.Next() { - var maskLength int + var ( + maskLength int + otherPeerHasARoute bool + ) - err := rows.Scan(&maskLength) + err := rows.Scan(&lastPrefix, &maskLength, &otherPeerHasARoute) if err != nil { return 0, nil, err } - goneMaskLengths = append(goneMaskLengths, maskLength) + selected++ + + if !otherPeerHasARoute { + goneMaskLengths = append(goneMaskLengths, maskLength) + } } err = rows.Err() @@ -707,7 +730,11 @@ func deletePeerRoutes( return 0, nil, err } - res, err := tx.ExecContext(ctx, deleteSQL, peerIP) + if selected == 0 { + return 0, nil, nil + } + + res, err := tx.ExecContext(ctx, deleteSQL, peerIP, lastPrefix) if err != nil { return 0, nil, err } @@ -721,43 +748,26 @@ func deletePeerRoutes( } // DeleteLiveRoutesOfPeer deletes every live route of the peer, IPv4 and IPv6, -// in one transaction, as when the peer's BGP session goes down. It returns how -// many routes it deleted. +// as when the peer's BGP session goes down. It deletes them a chunk at a time +// with deletePeerRouteChunk, each chunk in a transaction of its own, and +// releases the database lock between chunks so other writes go on. It returns +// how many routes it deleted; after an error, the chunks it deleted before stay +// deleted. func (d *Database) DeleteLiveRoutesOfPeer(peerIP string) (int, error) { - d.lock("DeleteLiveRoutesOfPeer") - defer d.unlock() + total := 0 - tx, err := d.beginTx() - if err != nil { - return 0, fmt.Errorf("failed to begin transaction: %w", err) + for { + deletedV4, deletedV6, err := d.deletePeerRouteChunk(peerIP) + if err != nil { + return total, err + } + + total += deletedV4 + deletedV6 + + if deletedV4 < peerRouteChunkSize && deletedV6 < peerRouteChunkSize { + return total, nil + } } - defer d.rollback(tx) - - ctx := context.Background() - - deletedV4, goneMaskLengthsV4, err := deletePeerRoutes(ctx, tx, - peerOnlyRouteMaskLengthsV4SQL, deletePeerLiveRoutesV4SQL, peerIP) - if err != nil { - return 0, fmt.Errorf("failed to delete the IPv4 routes of peer %s: %w", - peerIP, err) - } - - deletedV6, goneMaskLengthsV6, err := deletePeerRoutes(ctx, tx, - peerOnlyRouteMaskLengthsV6SQL, deletePeerLiveRoutesV6SQL, peerIP) - if err != nil { - return 0, fmt.Errorf("failed to delete the IPv6 routes of peer %s: %w", - peerIP, err) - } - - err = tx.Commit() - if err != nil { - return 0, fmt.Errorf("failed to commit transaction: %w", err) - } - - d.counts.addRoutes(-int(deletedV4), -int(deletedV6)) - d.counts.addToDistribution(goneMaskLengthsV4, goneMaskLengthsV6, -1) - - return int(deletedV4 + deletedV6), nil } // insertPrefixesInTable inserts the prefixes, all of one IP version, that the @@ -2332,6 +2342,46 @@ func (d *Database) rollback(tx *loggingTx) { } } +// deletePeerRouteChunk deletes, in one transaction, the first +// peerRouteChunkSize IPv4 and the first peerRouteChunkSize IPv6 live routes of +// the peer, or as many as it has. It returns how many of each it deleted. +func (d *Database) deletePeerRouteChunk(peerIP string) (int, int, error) { + d.lock("DeleteLiveRoutesOfPeer") + defer d.unlock() + + tx, err := d.beginTx() + if err != nil { + return 0, 0, fmt.Errorf("failed to begin transaction: %w", err) + } + defer d.rollback(tx) + + ctx := context.Background() + + deletedV4, goneMaskLengthsV4, err := deletePeerRoutes(ctx, tx, + peerRouteChunkV4SQL, deletePeerRouteChunkV4SQL, peerIP) + if err != nil { + return 0, 0, fmt.Errorf("failed to delete the IPv4 routes of peer %s: %w", + peerIP, err) + } + + deletedV6, goneMaskLengthsV6, err := deletePeerRoutes(ctx, tx, + peerRouteChunkV6SQL, deletePeerRouteChunkV6SQL, peerIP) + if err != nil { + return 0, 0, fmt.Errorf("failed to delete the IPv6 routes of peer %s: %w", + peerIP, err) + } + + err = tx.Commit() + if err != nil { + return 0, 0, fmt.Errorf("failed to commit transaction: %w", err) + } + + d.counts.addRoutes(-int(deletedV4), -int(deletedV6)) + d.counts.addToDistribution(goneMaskLengthsV4, goneMaskLengthsV6, -1) + + return int(deletedV4), int(deletedV6), nil +} + // routeTimestampRange returns the earliest and latest last_updated across both // live route tables, or nil values when both tables are empty. Each query reads // one row from an end of the last_updated index rather than scanning the tables. diff --git a/internal/routewatch/prefixhandler.go b/internal/routewatch/prefixhandler.go index c3ae80e..4467e2a 100644 --- a/internal/routewatch/prefixhandler.go +++ b/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 handler holds at most + // prefixBatchSize updates, about 6 MiB, that a flush kept for the peers + // whose route delete failed (held). prefixBatchMaxSize = 250000 // prefixBatchTimeout is the maximum time to wait before flushing a batch @@ -48,13 +50,21 @@ type PrefixHandler struct { logger *logger.Logger metrics *metrics.Tracker - // Batching. mu guards batch and peersDown only: the flush goroutine takes - // both under mu and writes them after releasing mu, so HandleMessage - // waits for a database write only while batch is at prefixBatchMaxSize. + // Batching. mu guards batch, held and peersDown only: the flush goroutine + // takes them under mu and writes them after releasing mu, so + // HandleMessage waits for a database write only while batch is at + // prefixBatchMaxSize. mu sync.Mutex batch []prefixUpdate - // peersDown holds the peers whose BGP session went down since the batch - // was taken. + // held holds the updates that the peers whose route delete failed in the + // last flush sent after their session went down, for the next flush to + // write after it retries the delete. They are not in batch, so they do + // not count toward a full batch: a delete that keeps failing is retried + // at the pace of the flush timer. + held []prefixUpdate + // peersDown holds, each once, the peers whose BGP session went down + // since the batch was taken, and those whose route delete failed in the + // last flush. peersDown []string // batchTaken is signalled when the flush goroutine takes batch; its L is // &mu. @@ -200,17 +210,21 @@ func (h *PrefixHandler) Stop() { // peerSessionDown notes that the peer's BGP session went down, so that the next // flush deletes all of the peer's live routes before it writes the batch. The -// peer's updates still in the batch came before the down, which replaces them, -// so they are dropped; updates the peer sends later stay in the batch and are -// written after the delete. +// peer's updates still in the batch or held came before the down, which +// replaces them, so they are dropped; updates the peer sends later stay in the +// batch and are written after the delete. A peer already in peersDown is not +// added again: one delete removes all of its routes. func (h *PrefixHandler) peerSessionDown(peer string) { h.mu.Lock() defer h.mu.Unlock() - h.batch = slices.DeleteFunc(h.batch, func(update prefixUpdate) bool { - return update.peer == peer - }) - h.peersDown = append(h.peersDown, peer) + isPeers := func(update prefixUpdate) bool { return update.peer == peer } + h.batch = slices.DeleteFunc(h.batch, isPeers) + h.held = slices.DeleteFunc(h.held, isPeers) + + if !slices.Contains(h.peersDown, peer) { + h.peersDown = append(h.peersDown, peer) + } } // flushLoop runs in a goroutine, the only one that writes the batch: it @@ -238,18 +252,23 @@ func (h *PrefixHandler) flushLoop() { } } -// takeBatch returns the pending updates and the peers whose session went down -// since the last flush, and leaves an empty batch and no peers in their place -// for HandleMessage to fill. +// takeBatch returns the pending updates, the held ones first, and the peers in +// peersDown, and leaves an empty batch, no held updates and no peers in their +// place for HandleMessage to fill. func (h *PrefixHandler) takeBatch() ([]prefixUpdate, []string) { h.mu.Lock() defer h.mu.Unlock() - if len(h.batch) == 0 && len(h.peersDown) == 0 { + if len(h.held) == 0 && len(h.batch) == 0 && len(h.peersDown) == 0 { return nil, nil } batch, peersDown := h.batch, h.peersDown + if len(h.held) > 0 { + batch = slices.Concat(h.held, h.batch) + } + + h.held = nil h.batch = make([]prefixUpdate, 0, prefixBatchSize) h.peersDown = nil h.batchTaken.Broadcast() @@ -269,8 +288,12 @@ func (h *PrefixHandler) flushBatch() { batchSize := len(batch) // The routes of a peer whose session went down are deleted first: the - // batch holds only the updates the peer sent after the down. - h.deleteLiveRoutesOfPeers(peersDown) + // batch holds only the updates the peer sent after the down. A peer whose + // delete failed is kept, with those updates, for the next flush to retry. + failedPeers := h.deleteLiveRoutesOfPeers(peersDown) + if len(failedPeers) > 0 { + batch = h.keepForNextFlush(batch, failedPeers) + } // For each prefix and peer, keep only the last update the feed delivered. // Updates from different peers for the same prefix are all kept. Arrival @@ -344,15 +367,21 @@ func (h *PrefixHandler) flushBatch() { } // deleteLiveRoutesOfPeers deletes all live routes of each of peers, the peers -// whose BGP session went down. Most downs come from peers that have no routes, -// so only a delete that removed routes is logged. -func (h *PrefixHandler) deleteLiveRoutesOfPeers(peers []string) { +// whose BGP session went down, and returns those whose delete failed. Most +// downs come from peers that have no routes, so only a delete that removed +// routes is logged. +func (h *PrefixHandler) deleteLiveRoutesOfPeers(peers []string) []string { + var failed []string + for _, peer := range peers { deleted, err := h.db.DeleteLiveRoutesOfPeer(peer) if err != nil { h.logger.Error( - "Failed to delete the live routes of a peer whose session went down", - "peer", peer, "error", err) + "Failed to delete the live routes of a peer whose session "+ + "went down; the next flush retries", + "peer", peer, "routes_deleted", deleted, "error", err) + + failed = append(failed, peer) continue } @@ -363,6 +392,63 @@ func (h *PrefixHandler) deleteLiveRoutesOfPeers(peers []string) { "peer", peer, "routes", deleted) } } + + return failed +} + +// keepForNextFlush puts the peers whose route delete failed back in peersDown, +// and their updates from batch in held, so that the next flush retries each +// delete before it writes the peer's updates; written now, they would be +// deleted by the retry. The updates of a peer whose session went down again +// since the batch was taken came before that down, so they are dropped, as +// peerSessionDown drops them, and the peer is already in peersDown. It returns +// the updates of the other peers, for this flush to write. +// +// While a delete keeps failing, the held 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 { + var kept, toWrite []prefixUpdate + + for _, update := range batch { + if slices.Contains(failedPeers, update.peer) { + kept = append(kept, update) + } else { + toWrite = append(toWrite, update) + } + } + + h.mu.Lock() + defer h.mu.Unlock() + + 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) + + // A copy, so that the dropped updates are freed. + kept = slices.Clone(kept[len(kept)-prefixBatchSize:]) + } + + // takeBatch left held empty, and only this flush fills it. + h.held = kept + + for _, peer := range failedPeers { + if !slices.Contains(h.peersDown, peer) { + h.peersDown = append(h.peersDown, peer) + } + } + + return toWrite } // writeBatch writes the collected route upserts, route deletions and prefix diff --git a/internal/routewatch/prefixhandler_internal_test.go b/internal/routewatch/prefixhandler_internal_test.go index e6a9ad8..0fb851a 100644 --- a/internal/routewatch/prefixhandler_internal_test.go +++ b/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 {