diff --git a/README.md b/README.md index 25764fe..c97ee45 100644 --- a/README.md +++ b/README.md @@ -444,7 +444,9 @@ the routing table, and everything else from the database. - An announcement gives the peer a route to the prefix. - A withdrawal removes the peer's route. So does an announcement whose AS path is empty or holds AS 0 anywhere. - - A peer whose BGP session goes down leaves every prefix. + - A peer whose BGP session goes down leaves every prefix. The handler + holds back the peer's later updates until the peer has left every + prefix, then applies them in order. Every second it writes the changed prefix and origin AS pairs to the `routes` table. A write takes at most 20,000 pairs, and the rest follow @@ -580,8 +582,8 @@ Budget: - Go heap: a 2.5 GiB soft limit (`GOMEMLIMIT=2560MiB`, set in the image). - The routing table takes about 700 bytes per prefix and origin AS. A full table has about 1.4 million of these pairs, so it takes about 1 GB. - - The rest of the heap takes up to about 0.5 GiB. Most of it is the - handlers' batches and queues, described below. + - The rest of the heap takes up to about 0.5 GiB. Most of it is what the + handlers hold, described below. - The remaining 1 GiB is room for the garbage collector. With much less, it would run almost all the time. - SQLite: 656 MiB of page cache. @@ -648,6 +650,12 @@ What happens at each limit: as above. With the batches being written, the three handlers' batches take at most about 0.1 GiB of the Go heap. +- Held-back messages: the prefix handler holds back a peer's messages after the + peer's session goes down. It applies them once the peer has left every prefix. + It holds at most 20,000 such messages (`maxHeldBackMessages`), about 20 MiB. + At that number it takes nothing more off its queue until it has applied one of + them. + With `DEBUG=routewatch` the daemon logs a `System stats` line every 60 seconds with the goroutine count and Go memory figures. diff --git a/TODO.md b/TODO.md index 8caaa8e..aa18f0e 100644 --- a/TODO.md +++ b/TODO.md @@ -35,6 +35,15 @@ tested. # Completed Steps +- 2026-11-02: a peer's session going down no longer stops the prefix handler. + The handler runs the table's `PeerDown` on a separate goroutine, one peer at a + time, and goes on taking messages. `PeerDown` holds the table's lock for + 10,000 prefixes at a time, then lets the lock go for as long again. The + handler holds back the down peer's later messages until that peer's `PeerDown` + returns. Then the handler applies them in order, one at a time, and takes new + messages in between. It holds back at most 20,000 messages. At that number, + `HandleMessage` waits until one of them is applied. When `Stop` runs past its + deadline, it logs how many held-back messages were not applied (closes #216) - 2026-11-02: RouteWatch pauses an RIR's or NIC.br's WHOIS server for an hour after the server refuses a query. A refused or reset connection is a refusal. So is an answer with an `%ERROR:201:` or `%ERROR:202:` line. Lookups at the diff --git a/internal/routetable/overview.go b/internal/routetable/overview.go index b593c54..3b5bbca 100644 --- a/internal/routetable/overview.go +++ b/internal/routetable/overview.go @@ -10,10 +10,6 @@ import ( "git.eeqj.de/sneak/routewatch/internal/ristypes" ) -// walkBatch is how many prefixes the walk over the whole table reads in one -// hold of the read lock. The updates waiting for the lock get it in between. -const walkBatch = 10000 - // Overview holds figures about the whole table. Working them out takes a walk // over every prefix, so a caller asks for them now and then and keeps them. type Overview struct { diff --git a/internal/routetable/routetable.go b/internal/routetable/routetable.go index 2a737c5..b083b6c 100644 --- a/internal/routetable/routetable.go +++ b/internal/routetable/routetable.go @@ -27,6 +27,11 @@ const ( // kept routes in order: each of its nodes holds up to 63 times. keptTimesDegree = 32 + // walkBatch is how many prefixes a walk over the whole table visits in + // one hold of the lock. Overview and PeerDown each walk the whole table. + // Callers waiting for the lock get it between batches. + walkBatch = 10000 + ipv4Bits = 32 ipv6Bits = 128 @@ -201,8 +206,13 @@ func (t *Table) Withdraw(prefix netip.Prefix, peer string) { } // PeerDown applies a peer's session going down: the peer leaves every peer -// set, and every route of it is dropped. When the peer carries any prefix, it -// visits every prefix in the table. +// set, and every route of the peer is dropped. When the peer carries any +// prefix, PeerDown visits every prefix in the table, in batches of walkBatch. +// After each batch, it lets the table's lock go for as long as it held the +// lock for the batch. During a walk over many batches, reads and updates can +// then take the lock about half of the time. An announcement by the peer +// applied during the walk may be dropped too. So the caller applies the peer's +// later updates only once PeerDown has returned. func (t *Table) PeerDown(peer string) { t.mu.Lock() defer t.mu.Unlock() @@ -212,6 +222,9 @@ func (t *Table) PeerDown(peer string) { return } + visited := 0 + batchStart := time.Now() + for _, f := range []*family{&t.ipv4, &t.ipv6} { for _, prefixes := range f.byLength { for prefix, origins := range prefixes { @@ -219,6 +232,19 @@ func (t *Table) PeerDown(peer string) { if o != nil { t.leave(prefix, o, number) } + + // Go allows the map to change before the next step of + // the range over it. + visited++ + if visited%walkBatch == 0 { + held := time.Since(batchStart) + + t.mu.Unlock() + time.Sleep(held) + t.mu.Lock() + + batchStart = time.Now() + } } } } diff --git a/internal/routetable/routetable_internal_test.go b/internal/routetable/routetable_internal_test.go new file mode 100644 index 0000000..64dde1a --- /dev/null +++ b/internal/routetable/routetable_internal_test.go @@ -0,0 +1,69 @@ +package routetable + +import ( + "net/netip" + "runtime" + "testing" + "time" +) + +// TestPeerDownLetsReadsIn reads the table's counts again and again while a peer +// that carries more than two batches of prefixes goes down. PeerDown lets the +// lock go after every walkBatch prefixes, so a read must get in while the peer +// still carries some of its prefixes but no longer all. +func TestPeerDownLetsReadsIn(t *testing.T) { + t.Parallel() + + const ( + asn = 1000 + peer = "192.0.2.1" + // prefixCount is more prefixes than two of walkBatch. + prefixCount = 2*walkBatch + 1 + ) + + routeTime := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + table := New(Params{}) + addr := netip.MustParseAddr("1.0.0.0") + + for range prefixCount { + prefix := netip.PrefixFrom(addr, addr.BitLen()) + table.Announce(prefix, asn, Route{Peer: peer, Time: routeTime}) + + addr = addr.Next() + } + + // Hold a read lock until PeerDown waits for the lock, so that the first + // read below starts once PeerDown holds it. TryRLock fails once a writer + // waits. + table.mu.RLock() + + peerDownDone := make(chan struct{}) + + go func() { + defer close(peerDownDone) + + table.PeerDown(peer) + }() + + for table.mu.TryRLock() { + table.mu.RUnlock() + runtime.Gosched() + } + + table.mu.RUnlock() + + for { + routes := table.Counts().IPv4Routes + if routes > 0 && routes < prefixCount { + break + } + + select { + case <-peerDownDone: + t.Fatal("no read got in while PeerDown ran") + default: + } + } + + <-peerDownDone +} diff --git a/internal/routewatch/batchwrite_internal_test.go b/internal/routewatch/batchwrite_internal_test.go index c8c6dce..ecacc8d 100644 --- a/internal/routewatch/batchwrite_internal_test.go +++ b/internal/routewatch/batchwrite_internal_test.go @@ -5,6 +5,7 @@ import ( "context" "encoding/json" "fmt" + "io" "log/slog" "slices" "sync" @@ -177,13 +178,14 @@ type stopLog struct { Msg string `json:"msg"` Updates int `json:"updates"` TableChanges int `json:"table_changes"` + HeldMessages int `json:"held_messages"` Paths int `json:"paths"` BatchEntries int `json:"batch_entries"` } // newErrorLogger returns a logger that writes only errors, as JSON lines, to // logs. -func newErrorLogger(logs *bytes.Buffer) *logger.Logger { +func newErrorLogger(logs io.Writer) *logger.Logger { return &logger.Logger{Logger: slog.New(slog.NewJSONHandler(logs, &slog.HandlerOptions{Level: slog.LevelError}))} } diff --git a/internal/routewatch/prefixhandler.go b/internal/routewatch/prefixhandler.go index 1de36c2..666edf3 100644 --- a/internal/routewatch/prefixhandler.go +++ b/internal/routewatch/prefixhandler.go @@ -44,6 +44,12 @@ const ( // most one pair. So the million pairs that a full-table peer's session // going down changes are written in about a minute. maxTableChangesPerWrite = 20000 + + // maxHeldBackMessages is the most messages the handler holds back from + // peers whose PeerDown has not returned. A message takes about 1 KiB, so + // that many take about 20 MiB (README "Memory"). At this number + // HandleMessage waits until one of them is applied. + maxHeldBackMessages = 20000 ) // PrefixHandler keeps the routing table in memory current from the feed. It @@ -52,6 +58,15 @@ const ( // does an announcement whose AS path is empty or holds AS 0 anywhere. A peer // whose BGP session goes down leaves every prefix. // +// The table's PeerDown applies a session going down. It visits every prefix, +// which takes seconds on a full table. So the handler runs PeerDown on +// a separate goroutine, one peer at a time, and goes on taking messages +// meanwhile. PeerDown lets the table's lock go about half of the time, so the +// handler applies other peers' messages while PeerDown runs. The handler holds +// back the down peer's later messages until PeerDown returns. Then the handler +// applies them in order, one at a time, and takes new messages in between. It +// holds back at most maxHeldBackMessages messages. +// // Each flush writes the table's changes to the database. It also stores each // prefix the updates name with the time it was first seen. type PrefixHandler struct { @@ -67,7 +82,7 @@ type PrefixHandler struct { // give a smaller one. tableChangesPerWrite int - // Batching. mu guards batch, writing and writingTableChanges only: the + // Batching. mu guards batch, writing and writingTableChanges: 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 // batchMaxSize. @@ -87,6 +102,21 @@ type PrefixHandler struct { // Stop does not wait for a write in progress to hand it over. stopCh chan context.Context done chan struct{} // closed when the flush goroutine returns + + // Peers whose session went down. peerDownMu lets one PeerDown run at a + // time, so the handler's own updates wait for the table's lock behind + // at most one. heldBack, under mu, holds each such peer's later + // messages, in order, until its PeerDown has returned and they are + // applied. A peer with no such messages yet has an empty entry. + // heldCount, under mu, is how many messages heldBack holds. + // heldTaken is signalled each time a held-back message is taken out to + // be applied; its L is &mu. peerDowns counts the peerDown goroutines + // that have not returned. + peerDownMu sync.Mutex + heldBack map[string][]*ristypes.RISMessage + heldCount int + heldTaken sync.Cond + peerDowns sync.WaitGroup } // seenPrefix is what the batch keeps of an update: the prefix it names and its @@ -150,8 +180,10 @@ func newPrefixHandler( batchFullCh: make(chan struct{}, 1), stopCh: make(chan context.Context, 1), done: make(chan struct{}), + heldBack: make(map[string][]*ristypes.RISMessage), } h.batchTaken.L = &h.mu + h.heldTaken.L = &h.mu // Start the flush timer goroutine go flushLoop(batchTimeout, h.batchFullCh, h.stopCh, h.done, h.flushBatch) @@ -179,11 +211,91 @@ func (h *PrefixHandler) QueueCapacity() int { return prefixHandlerQueueSize } -// HandleMessage processes a message to track prefix information +// HandleMessage processes a message to track prefix information. A message +// from a peer whose PeerDown has not returned is held back until it has. func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) { + h.mu.Lock() + defer h.mu.Unlock() + + // While maxHeldBackMessages messages are held back, wait until one of them + // is taken out to be applied. Meanwhile the handler takes nothing off its + // queue, as at batchMaxSize. + for h.heldCount >= maxHeldBackMessages { + h.heldTaken.Wait() + } + + held, goingDown := h.heldBack[msg.Peer] + if goingDown { + h.heldBack[msg.Peer] = append(held, msg) + h.heldCount++ + + return + } + + h.applyMessage(msg) +} + +// Stop waits for every PeerDown the handler started and for the messages held +// back meanwhile to be applied. It gives up this wait once ctx ends. Then Stop +// stops the flush goroutine and waits for it until ctx ends. The flush +// goroutine makes one last flush before it returns. That flush writes at most +// tableChangesPerWrite of the routing table's changes, and the prefixes of the +// updates the handler holds. Once ctx has ended the flush goroutine skips that +// last flush, and a write still running is lost when the process exits. A +// failed write of the table's changes is not retried either. Stop logs how +// many updates and table changes are not written. It also logs how many +// held-back messages are not applied. The caller calls Stop once its last +// HandleMessage has returned. +func (h *PrefixHandler) Stop(ctx context.Context) { + peerDownsDone := make(chan struct{}) + + go func() { + h.peerDowns.Wait() + close(peerDownsDone) + }() + + select { + case <-peerDownsDone: + case <-ctx.Done(): + } + + h.stopCh <- ctx + + select { + case <-h.done: + case <-ctx.Done(): + } + + h.mu.Lock() + unwritten := h.writing + len(h.batch) + tableChanges := h.writingTableChanges + h.table.ChangeCount() + heldMessages := h.heldCount + h.mu.Unlock() + + if unwritten == 0 && tableChanges == 0 && heldMessages == 0 { + return + } + + if ctx.Err() != nil { + h.logger.Error("Stop deadline passed before the prefix handler wrote "+ + "its updates", "updates", unwritten, "table_changes", tableChanges, + "held_messages", heldMessages) + + return + } + + h.logger.Error("The prefix handler stopped before it wrote everything", + "updates", unwritten, "table_changes", tableChanges, + "held_messages", heldMessages) +} + +// applyMessage applies msg. The caller holds mu. A peer's session going down +// starts the peer's PeerDown. An update is applied to the routing table, and +// its prefixes join the batch. +func (h *PrefixHandler) applyMessage(msg *ristypes.RISMessage) { if msg.Type == ristypes.MessageTypeState { if msg.State == ristypes.StateDown { - h.table.PeerDown(msg.Peer) + h.startPeerDown(msg.Peer) } return @@ -198,9 +310,6 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) { // the path after the rest of it. path := slices.Concat(msg.Path.Sequence, msg.Path.Set) - h.mu.Lock() - defer h.mu.Unlock() - // While the batch is at its maximum size, wait until the flush goroutine // takes it. Meanwhile the handler takes nothing off its queue, which // fills, and the streamer drops messages for it. @@ -261,39 +370,58 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) { } } -// Stop stops the flush goroutine, which makes one last flush before it returns, -// and waits for it until ctx ends. That flush writes at most -// tableChangesPerWrite of the routing table's changes, and the prefixes of -// the updates the handler holds. Once ctx has ended the flush goroutine skips -// that last flush, and a write still running is lost when the process exits. A -// failed write of the table's changes is not retried either. Stop logs how many -// updates and how many of the table's changes are not written. -func (h *PrefixHandler) Stop(ctx context.Context) { - h.stopCh <- ctx - - select { - case <-h.done: - case <-ctx.Done(): +// startPeerDown starts peerDown for peer on a new goroutine. The caller holds +// mu. The peer's messages are held back until peerDown has applied them. +// Messages held back for an earlier PeerDown of the peer, and not yet applied, +// stay held back. +func (h *PrefixHandler) startPeerDown(peer string) { + _, goingDown := h.heldBack[peer] + if !goingDown { + h.heldBack[peer] = []*ristypes.RISMessage{} } - h.mu.Lock() - unwritten := h.writing + len(h.batch) - tableChanges := h.writingTableChanges + h.table.ChangeCount() - h.mu.Unlock() + h.peerDowns.Add(1) - if unwritten == 0 && tableChanges == 0 { - return + go h.peerDown(peer) +} + +// peerDown runs the table's PeerDown for peer, once no other PeerDown is +// running. Then peerDown applies the messages held back from the peer, in +// order. It takes mu for one message at a time, so HandleMessage takes new +// messages in between. The peer's new messages join the end of the held-back +// ones. A held-back message that says the peer's session went down again +// starts a new peerDown, which applies the rest. +func (h *PrefixHandler) peerDown(peer string) { + defer h.peerDowns.Done() + + h.peerDownMu.Lock() + h.table.PeerDown(peer) + h.peerDownMu.Unlock() + + for { + h.mu.Lock() + + held := h.heldBack[peer] + if len(held) == 0 { + delete(h.heldBack, peer) + h.mu.Unlock() + + return + } + + msg := held[0] + held[0] = nil // so that the message is freed once applied + h.heldBack[peer] = held[1:] + h.heldCount-- + h.heldTaken.Broadcast() + h.applyMessage(msg) + h.mu.Unlock() + + if msg.Type == ristypes.MessageTypeState && + msg.State == ristypes.StateDown { + return + } } - - if ctx.Err() != nil { - h.logger.Error("Stop deadline passed before the prefix handler wrote "+ - "its updates", "updates", unwritten, "table_changes", tableChanges) - - return - } - - h.logger.Error("The prefix handler stopped before it wrote everything", - "updates", unwritten, "table_changes", tableChanges) } // takeBatch returns the pending updates and leaves an empty batch in their diff --git a/internal/routewatch/prefixhandler_internal_test.go b/internal/routewatch/prefixhandler_internal_test.go index a81d346..10c190f 100644 --- a/internal/routewatch/prefixhandler_internal_test.go +++ b/internal/routewatch/prefixhandler_internal_test.go @@ -237,6 +237,83 @@ func TestPrefixHandlerStopLogsNothingWhenItWroteEverything(t *testing.T) { } } +// TestPrefixHandlerStopLogsTheMessagesHeldBack stops a prefix handler with a +// deadline that has passed, while a peer's PeerDown is held up and one message +// of the peer is held back. Stop must log that message as not applied. +func TestPrefixHandlerStopLogsTheMessagesHeldBack(t *testing.T) { + t.Parallel() + + var logs bytes.Buffer + + h := newPrefixHandler(newMockStore(), newErrorLogger(&logs), + routetable.New(routetable.Params{}), longerThanAnyTest, + maxTableChangesPerWrite) + + h.peerDownMu.Lock() + defer h.peerDownMu.Unlock() + + h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix)) + h.HandleMessage(sessionDown()) + h.HandleMessage(peerAnnouncement(testPeerIP, testV6Prefix)) + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + h.Stop(ctx) + + logged := lastLogged(t, &logs) + if logged.HeldMessages != 1 { + t.Errorf("Stop logged %d held messages not applied, want 1", + logged.HeldMessages) + } +} + +// TestPrefixHandlerStopWaitsForTheMessagesHeldBack stops a prefix handler +// while a peer's PeerDown is held up and an announcement of the peer is held +// back. Stop must wait. Once the PeerDown is let go and Stop returns, the +// announcement must be in the database. +func TestPrefixHandlerStopWaitsForTheMessagesHeldBack(t *testing.T) { + t.Parallel() + + db := newTestDatabase(t) + h := newPrefixHandler(db, logger.New(), routetable.New(routetable.Params{}), + longerThanAnyTest, maxTableChangesPerWrite) + + h.peerDownMu.Lock() + h.HandleMessage(sessionDown()) + h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix)) + + stopped := make(chan struct{}) + + go func() { + h.Stop(t.Context()) + close(stopped) + }() + + select { + case <-stopped: + t.Error("Stop returned while a PeerDown was held up") + case <-time.After(stillWaitingAfter): + } + + h.peerDownMu.Unlock() + + select { + case <-stopped: + case <-time.After(stepTimeout): + t.Fatalf("Stop did not return within %s of the PeerDown's release", + stepTimeout) + } + + read, err := db.ReadRouteTable(t.Context()) + if err != nil { + t.Fatalf("ReadRouteTable: %v", err) + } + + checkTableOrigins(t, read, map[string]map[int][]string{ + testV4Prefix: {testASNB: {testPeerIP}}, + }) +} + // longerThanAnyTest is a timer interval that no test lasts. A test gives it to // newPrefixHandler, newASHandler, newPeerHandler or newPeeringHandler when the // handler must write only when its batch is full, when the test says, or on @@ -253,6 +330,7 @@ func prefixHandlerWithoutTimer(store database.Store) *PrefixHandler { table: routetable.New(routetable.Params{}), batchMaxSize: prefixBatchMaxSize, tableChangesPerWrite: maxTableChangesPerWrite, + heldBack: make(map[string][]*ristypes.RISMessage), } } diff --git a/internal/routewatch/prefixhandler_table_internal_test.go b/internal/routewatch/prefixhandler_table_internal_test.go index e144842..fa94967 100644 --- a/internal/routewatch/prefixhandler_table_internal_test.go +++ b/internal/routewatch/prefixhandler_table_internal_test.go @@ -7,8 +7,10 @@ import ( "log/slog" "net/netip" "reflect" + "runtime" "slices" "strings" + "sync" "testing" "time" @@ -101,6 +103,228 @@ func TestPrefixHandlerAppliesUpdatesToTheRoutingTable(t *testing.T) { }) } +// TestPrefixHandlerTakesMessagesWhileAPeerGoesDown holds up the PeerDown of a +// peer whose session went down, as another peer's long PeerDown would. The +// handler must go on taking messages. It applies another peer's announcement +// at once. It holds back these messages of the down peer, in this order: +// - an announcement; +// - its session going down again; +// - a second announcement. +// +// Once the PeerDowns have returned, the down peer must carry only the prefix +// of the second announcement. +func TestPrefixHandlerTakesMessagesWhileAPeerGoesDown(t *testing.T) { + t.Parallel() + + h := prefixHandlerWithoutTimer(newMockStore()) + h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix)) + + h.peerDownMu.Lock() + + handled := returnsInTime(func() { + for _, msg := range []*ristypes.RISMessage{ + sessionDown(), + peerAnnouncement(secondPeerIP, otherV4Prefix), + peerAnnouncement(testPeerIP, testV6Prefix), + sessionDown(), + peerAnnouncement(testPeerIP, setV4Prefix), + } { + h.HandleMessage(msg) + } + }) + if !handled { + t.Fatalf("the handler did not take the messages after the session "+ + "went down within %s", stepTimeout) + } + + checkTableOrigins(t, h.table, map[string]map[int][]string{ + testV4Prefix: {testASNB: {testPeerIP}}, + otherV4Prefix: {testASNB: {secondPeerIP}}, + testV6Prefix: {}, + setV4Prefix: {}, + }) + + h.peerDownMu.Unlock() + + if !returnsInTime(h.peerDowns.Wait) { + t.Fatalf("the PeerDowns did not return within %s", stepTimeout) + } + + checkTableOrigins(t, h.table, map[string]map[int][]string{ + testV4Prefix: {}, + otherV4Prefix: {testASNB: {secondPeerIP}}, + testV6Prefix: {}, + setV4Prefix: {testASNB: {testPeerIP}}, + }) +} + +// TestPrefixHandlerAppliesMessagesWhilePeerDownRuns gives a peer 50,000 +// prefixes and hands the prefix handler that peer's session going down. The +// table's PeerDown then visits the prefixes in five batches. Once PeerDown has +// begun, the handler is handed 1,000 announcements of another peer. The +// handler must apply them all before PeerDown ends, while the down peer still +// carries some of its prefixes. +func TestPrefixHandlerAppliesMessagesWhilePeerDownRuns(t *testing.T) { + t.Parallel() + + const ( + // downPeerPrefixes is five times the 10,000 prefixes PeerDown visits + // in one hold of the table's lock. + downPeerPrefixes = 50000 + // otherPeerMessages is far more messages than the handler applies + // while PeerDown runs, if PeerDown takes the lock back at once after + // each batch. The handler then applies only a few for each batch. + otherPeerMessages = 1000 + // firstOtherAddr is the address of the other peer's first prefix. + firstOtherAddr = "2001:db8:2::" + ) + + h := prefixHandlerWithoutTimer(newMockStore()) + announceManyPrefixes(h.table, downPeerPrefixes) + + h.HandleMessage(sessionDown()) + + // Wait until PeerDown has begun: the down peer has left some of its + // prefixes. The down peer's prefixes are IPv4 and the other peer's are + // IPv6, so the counts tell them apart. + for h.table.Counts().IPv4Routes == downPeerPrefixes { + runtime.Gosched() + } + + // Each announcement names the /128 of the next address. + addr := netip.MustParseAddr(firstOtherAddr) + for range otherPeerMessages { + h.HandleMessage(peerAnnouncement(secondPeerIP, + netip.PrefixFrom(addr, addr.BitLen()).String())) + + addr = addr.Next() + } + + counts := h.table.Counts() + if counts.IPv6Routes != otherPeerMessages { + t.Errorf("the other peer carries %d prefixes, want %d", + counts.IPv6Routes, otherPeerMessages) + } + + if counts.IPv4Routes == 0 { + t.Errorf("PeerDown ended before the handler applied the other peer's "+ + "%d announcements", otherPeerMessages) + } + + if !returnsInTime(h.peerDowns.Wait) { + t.Fatalf("the PeerDown did not return within %s", stepTimeout) + } +} + +// TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes holds back many +// messages of a peer whose session went down. Each names a prefix that does +// not parse, so applying it logs an error. Each log write takes a +// millisecond. While the handler applies the held-back messages, it must take +// another peer's announcement. +func TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes(t *testing.T) { + t.Parallel() + + const ( + // heldMessages take the handler at least a second to apply. + heldMessages = 1000 + badPrefix = "not a prefix" + ) + + logs := &slowWriter{firstWrite: make(chan struct{})} + h := newPrefixHandler(newMockStore(), newErrorLogger(logs), + routetable.New(routetable.Params{}), longerThanAnyTest, + maxTableChangesPerWrite) + + defer stopInTime(t, h) + + h.peerDownMu.Lock() + h.HandleMessage(sessionDown()) + + for range heldMessages { + h.HandleMessage(peerAnnouncement(testPeerIP, badPrefix)) + } + + h.peerDownMu.Unlock() + + select { + case <-logs.firstWrite: + case <-time.After(stepTimeout): + t.Fatal("the handler did not start to apply the held-back messages") + } + + if !returnsInTime(func() { + h.HandleMessage(peerAnnouncement(secondPeerIP, otherV4Prefix)) + }) { + t.Fatalf("HandleMessage did not return within %s", stepTimeout) + } + + h.mu.Lock() + left := len(h.heldBack[testPeerIP]) + h.mu.Unlock() + + if left == 0 { + t.Error("the handler took another peer's announcement only once it " + + "had applied every held-back message") + } +} + +// TestPrefixHandlerWaitsAtTheMaximumHeldBack holds up the PeerDown of a peer +// whose session went down. It hands the handler maxHeldBackMessages messages +// of that peer. HandleMessage must then wait, even for another peer's +// message, until the PeerDown returns and a held-back message is applied. +func TestPrefixHandlerWaitsAtTheMaximumHeldBack(t *testing.T) { + t.Parallel() + + h := newPrefixHandler(newMockStore(), logger.New(), + routetable.New(routetable.Params{}), longerThanAnyTest, + maxTableChangesPerWrite) + defer stopInTime(t, h) + + h.peerDownMu.Lock() + h.HandleMessage(sessionDown()) + + held := peerAnnouncement(testPeerIP, testV6Prefix) + for range maxHeldBackMessages { + h.HandleMessage(held) + } + + returned := make(chan struct{}) + + go func() { + h.HandleMessage(peerAnnouncement(secondPeerIP, otherV4Prefix)) + close(returned) + }() + + select { + case <-returned: + t.Error("HandleMessage returned while maxHeldBackMessages messages " + + "were held back") + case <-time.After(stillWaitingAfter): + } + + h.peerDownMu.Unlock() + + select { + case <-returned: + case <-time.After(stepTimeout): + t.Fatal("HandleMessage did not return after the PeerDown returned") + } +} + +// slowWriter takes a millisecond over each write and drops what it is given. +// It closes firstWrite at the first write. +type slowWriter struct { + firstWrite chan struct{} + once sync.Once +} + +func (w *slowWriter) Write(p []byte) (int, error) { + w.once.Do(func() { close(w.firstWrite) }) + time.Sleep(time.Millisecond) + + return len(p), nil +} + // TestPrefixHandlerAppliesAPathWithoutAnOriginToTheRoutingTable hands the // prefix handler announcements whose AS path gives no origin. One whose path // is empty or holds AS 0 removes the peer's route, as a withdrawal does; one @@ -199,8 +423,9 @@ func TestPrefixHandlerKeepsThePathOfAPathEndingInASet(t *testing.T) { } } -// checkTableCases hands the messages of each case to a new prefix handler and -// checks its routing table right after, with no flush. +// checkTableCases hands the messages of each case to a new prefix handler. It +// checks the handler's routing table once the handler's PeerDowns have +// returned, with no flush. func checkTableCases(t *testing.T, tests []tableCase) { t.Helper() @@ -213,17 +438,28 @@ func checkTableCases(t *testing.T, tests []tableCase) { h.HandleMessage(msg) } - for prefix, want := range tt.want { - got := tableOrigins(h.table, prefix) - if !reflect.DeepEqual(got, want) { - t.Errorf("peers of %s by origin AS = %v, want %v", - prefix, got, want) - } - } + h.peerDowns.Wait() + checkTableOrigins(t, h.table, tt.want) }) } } +// checkTableOrigins checks, for each prefix in want, the peers that carry it +// in table with each origin AS. +func checkTableOrigins( + t *testing.T, table *routetable.Table, want map[string]map[int][]string, +) { + t.Helper() + + for prefix, wantOrigins := range want { + got := tableOrigins(table, prefix) + if !reflect.DeepEqual(got, wantOrigins) { + t.Errorf("peers of %s by origin AS = %v, want %v", + prefix, got, wantOrigins) + } + } +} + // tableWritesStore is a real database that hands each write of the routing // table's changes to written once it is made. type tableWritesStore struct { @@ -498,6 +734,7 @@ func TestARestartReadsTheRoutingTableBack(t *testing.T) { h.HandleMessage(msg) } + h.peerDowns.Wait() h.flushBatch() var logs bytes.Buffer @@ -536,6 +773,7 @@ func TestARestartReadsTheRoutingTableBack(t *testing.T) { restarted.HandleMessage(msg) } + h.peerDowns.Wait() restarted.Stop(t.Context()) read, err := db.ReadRouteTable(t.Context())