diff --git a/README.md b/README.md index 5f003b7..3ed449a 100644 --- a/README.md +++ b/README.md @@ -442,7 +442,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 it has left them all, 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 diff --git a/TODO.md b/TODO.md index 86d7452..75671ff 100644 --- a/TODO.md +++ b/TODO.md @@ -35,6 +35,13 @@ 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 goroutine of its own, one peer at + a time, and goes on taking messages. It holds back the down peer's later + messages until its `PeerDown` returns, then applies them in order. `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. A stop past its deadline logs the + held-back messages it did not apply (closes #216) - 2026-10-27: the tests of `internal/routewatch` no longer fail on a loaded host. Five of them gave `HandleMessage` two seconds to apply a full prefix batch, which under the race detector can take longer. A test now waits up to 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..6cdccf3 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's walk and PeerDown's. The others + // waiting for the lock get it in between. + 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 lock go after every walkBatch +// prefixes, so the reads and updates waiting for it get in. An announcement of +// 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 0f29c84..8be7aaa 100644 --- a/internal/routewatch/batchwrite_internal_test.go +++ b/internal/routewatch/batchwrite_internal_test.go @@ -177,6 +177,7 @@ 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"` } diff --git a/internal/routewatch/prefixhandler.go b/internal/routewatch/prefixhandler.go index 1de36c2..0010b1d 100644 --- a/internal/routewatch/prefixhandler.go +++ b/internal/routewatch/prefixhandler.go @@ -52,6 +52,12 @@ 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 it on a +// goroutine of its own, one peer at a time, and goes on taking messages +// meanwhile. It holds back the peer's later messages until PeerDown returns, +// then applies them in order. +// // 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 +73,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 +93,16 @@ 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. + // peerDowns counts the peerDown goroutines that have not returned. + peerDownMu sync.Mutex + heldBack map[string][]*ristypes.RISMessage + peerDowns sync.WaitGroup } // seenPrefix is what the batch keeps of an update: the prefix it names and its @@ -150,6 +166,7 @@ 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 @@ -179,11 +196,86 @@ 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() + + held, goingDown := h.heldBack[msg.Peer] + if goingDown { + h.heldBack[msg.Peer] = append(held, msg) + + return + } + + h.applyMessage(msg) +} + +// Stop waits until ctx ends for every PeerDown the handler started, and for +// the messages it held back meanwhile to be applied. Then it 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, and 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 := 0 + for _, held := range h.heldBack { + heldMessages += len(held) + } + 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, with mu held. 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 +290,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 +350,50 @@ 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 runs peerDown for peer on a goroutine of its own, with mu +// held. The peer's messages are held back until it returns. Those held back +// for an earlier PeerDown of the peer and not yet applied stay held. +func (h *PrefixHandler) startPeerDown(peer string) { + _, goingDown := h.heldBack[peer] + if !goingDown { + h.heldBack[peer] = []*ristypes.RISMessage{} } + h.peerDowns.Add(1) + + go h.peerDown(peer) +} + +// peerDown runs the table's PeerDown for peer once no other peer's runs. Then +// it applies the messages held back from the peer, in order. When the peer's +// session went down again among them, the PeerDown that starts applies the +// rest. +func (h *PrefixHandler) peerDown(peer string) { + defer h.peerDowns.Done() + + h.peerDownMu.Lock() + h.table.PeerDown(peer) + h.peerDownMu.Unlock() + h.mu.Lock() - unwritten := h.writing + len(h.batch) - tableChanges := h.writingTableChanges + h.table.ChangeCount() - h.mu.Unlock() + defer h.mu.Unlock() - if unwritten == 0 && tableChanges == 0 { - return + // applyMessage lets mu go while it waits for the batch. HandleMessage + // then adds the peer's new messages after these. + for len(h.heldBack[peer]) > 0 { + msg := h.heldBack[peer][0] + h.heldBack[peer] = h.heldBack[peer][1:] + h.applyMessage(msg) + + if msg.Type == ristypes.MessageTypeState && + msg.State == ristypes.StateDown { + // applyMessage started the next PeerDown, which applies the + // rest. + 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) + delete(h.heldBack, peer) } // 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..f4c0859 100644 --- a/internal/routewatch/prefixhandler_internal_test.go +++ b/internal/routewatch/prefixhandler_internal_test.go @@ -237,6 +237,36 @@ 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) + } +} + // 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 +283,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..d1475d4 100644 --- a/internal/routewatch/prefixhandler_table_internal_test.go +++ b/internal/routewatch/prefixhandler_table_internal_test.go @@ -101,6 +101,58 @@ 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}}, + }) +} + // 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 +252,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 +266,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 +562,7 @@ func TestARestartReadsTheRoutingTableBack(t *testing.T) { h.HandleMessage(msg) } + h.peerDowns.Wait() h.flushBatch() var logs bytes.Buffer @@ -536,6 +601,7 @@ func TestARestartReadsTheRoutingTableBack(t *testing.T) { restarted.HandleMessage(msg) } + h.peerDowns.Wait() restarted.Stop(t.Context()) read, err := db.ReadRouteTable(t.Context())