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 a9d0a6b..762562a 100644 --- a/TODO.md +++ b/TODO.md @@ -35,6 +35,16 @@ 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. It holds back the down peer's later + messages until that peer's `PeerDown` returns. Then it 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. + `PeerDown` lets the table's lock go after every 10,000 prefixes, so the IP + lookup and the pages no longer wait for the whole visit. When `Stop` runs past + its deadline, it logs how many held-back messages were not applied (closes + #216) - 2026-11-02: the AS handler never waits for a database write. Its batch holds one entry per AS, with the earliest and the latest time of its updates. So the batch no longer grows with the feed's rate. It used to hold one entry for each 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..3792efe 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 @@ -202,7 +207,10 @@ 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. +// visits every prefix in the table. It lets the table's lock go after every +// walkBatch prefixes, so that waiting reads and updates get in. An +// announcement by the peer made meanwhile 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 +220,8 @@ func (t *Table) PeerDown(peer string) { return } + visited := 0 + for _, f := range []*family{&t.ipv4, &t.ipv6} { for _, prefixes := range f.byLength { for prefix, origins := range prefixes { @@ -219,6 +229,14 @@ 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 { + t.mu.Unlock() + t.mu.Lock() + } } } } 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..8e20e4a 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,14 @@ 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 over a second on a full table. So the handler runs PeerDown on +// a separate goroutine, one peer at a time, and goes on taking messages +// meanwhile. The handler holds back the down peer's later messages until +// PeerDown returns. Then it 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 +81,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 +101,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 +179,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 +210,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 +309,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 +369,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 it 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..e65a7e6 100644 --- a/internal/routewatch/prefixhandler_table_internal_test.go +++ b/internal/routewatch/prefixhandler_table_internal_test.go @@ -9,6 +9,7 @@ import ( "reflect" "slices" "strings" + "sync" "testing" "time" @@ -101,6 +102,167 @@ 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 the down peer's own messages: an announcement, its +// session going down again, and 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}}, + }) +} + +// TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes holds back many +// messages of a peer whose session went down. Each names a prefix that does +// not parse, so the handler logs an error as it applies it. 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 @@ -200,7 +362,8 @@ 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. +// checks its routing table once the PeerDowns it started have returned, with +// no flush. func checkTableCases(t *testing.T, tests []tableCase) { t.Helper() @@ -213,17 +376,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 +672,7 @@ func TestARestartReadsTheRoutingTableBack(t *testing.T) { h.HandleMessage(msg) } + h.peerDowns.Wait() h.flushBatch() var logs bytes.Buffer @@ -536,6 +711,7 @@ func TestARestartReadsTheRoutingTableBack(t *testing.T) { restarted.HandleMessage(msg) } + h.peerDowns.Wait() restarted.Stop(t.Context()) read, err := db.ReadRouteTable(t.Context())