diff --git a/README.md b/README.md index 5f003b7..257b439 100644 --- a/README.md +++ b/README.md @@ -429,10 +429,12 @@ the routing table, and everything else from the database. - Parses BGP UPDATE messages from JSON stream - Dispatches to registered handlers based on message type - Each handler has its own queue with backpressure handling - - Handlers batch writes for efficiency: the AS, peer and prefix handlers - write 30,000, 10,000 and 25,000 updates at a time, or after 2, 2 and 1 - seconds; the peering handler writes the peerings it has collected every 30 - seconds + - Handlers batch writes for efficiency. The AS, peer and prefix handlers + write a batch when it reaches a size, or after a time: + - the AS handler at 30,000 ASes, or after 2 seconds; + - the peer handler at 10,000 updates, or after 2 seconds; + - the prefix handler at 25,000 updates, or after 1 second. + - The peering handler writes the peerings it has collected every 30 seconds 4. **Handler Details** - **ASHandler**: Tracks all ASNs seen in AS paths @@ -632,16 +634,18 @@ What happens at each limit: `STATE` message, which it holds, in feed order, until there is room: a few hundred bytes each, about ten a second. - Handler batches: while the AS, peer or prefix handler writes a batch to the - database, the updates that arrive collect in its next batch. That batch stops - at ten times the batch size: - - 300,000 updates for the AS handler (`asnBatchMaxSize`, about 10 MiB); - - 100,000 for the peer handler (`peerBatchMaxSize`, about 10 MiB); - - 250,000 for the prefix handler (`prefixBatchMaxSize`, about 15 MiB). + database, the updates that arrive collect in its next batch. + - The AS handler's batch holds one entry per AS, with the earliest and the + latest time of its updates. About 80,000 ASes appear in the feed, so the + batch takes at most about 10 MiB, however fast the feed runs. + - The peer and prefix handlers' batches stop at ten times their batch size: + 100,000 updates for the peer handler (`peerBatchMaxSize`, about 10 MiB), + and 250,000 for the prefix handler (`prefixBatchMaxSize`, about 15 MiB). + At that size the handler takes nothing more off its queue until the write + ends. The queue then fills, and the streamer drops messages as above. - At that size the handler takes nothing more off its queue until the write - ends. The queue then fills, and the streamer drops messages as above. With - the batches being written, the three handlers' batches take at most about - 0.1 GiB of the Go heap. + With the batches being written, the three handlers' batches take at most + about 0.1 GiB of the Go heap. 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 86d7452..ef69485 100644 --- a/TODO.md +++ b/TODO.md @@ -35,6 +35,11 @@ tested. # Completed Steps +- 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 it + no longer grows with the feed's rate. It used to hold one entry for each AS of + each path and to wait once it held 300,000. In a burst of the feed its queue + then filled, and the streamer dropped messages for it (closes #217) - 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/routewatch/app_integration_internal_test.go b/internal/routewatch/app_integration_internal_test.go index 8c1d10e..c145e9b 100644 --- a/internal/routewatch/app_integration_internal_test.go +++ b/internal/routewatch/app_integration_internal_test.go @@ -644,9 +644,8 @@ func TestShutdownDoesNotWaitForABlockedWrite(t *testing.T) { } rw.dbHandler = NewASHandler(store, log) - fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)} startBlockedWrite(t, blocking, rw.dbHandler, - []*ristypes.RISMessage{{Path: fill}}) + []*ristypes.RISMessage{{Path: pathOfASes(asnBatchSize)}}) ctx, cancel := context.WithTimeout(t.Context(), deadline) defer cancel() diff --git a/internal/routewatch/ashandler.go b/internal/routewatch/ashandler.go index fa8faa6..cc5613c 100644 --- a/internal/routewatch/ashandler.go +++ b/internal/routewatch/ashandler.go @@ -18,43 +18,34 @@ const ( // (asnBatchTimeout), so a queue smaller than asnBatchSize is fine. asHandlerQueueSize = 20000 - // asnBatchSize is the number of ASN operations to batch together + // asnBatchSize is the number of ASes in a full batch, which wakes the + // flush goroutine to write it. asnBatchSize = 30000 - // asnBatchMaxSize is the most ASN operations the batch holds while the - // flush goroutine writes the previous one: ten batches, about 10 MiB - // (README "Memory"). At this size HandleMessage waits until the flush - // goroutine takes the batch. - asnBatchMaxSize = 300000 - // asnBatchTimeout is the maximum time to wait before flushing a batch // DO NOT reduce this timeout - larger batches are more efficient asnBatchTimeout = 2 * time.Second ) -// ASHandler processes Autonomous System Number (ASN) information extracted from -// BGP UPDATE messages. It uses batched database operations to efficiently store -// ASN data, collecting operations into batches that are flushed either when the -// batch reaches a size threshold or after a timeout period. +// ASHandler records every AS in the AS paths of BGP UPDATE messages. It +// collects, for each AS, the earliest and the latest time of its updates in a +// batch. The flush goroutine writes the batch to the database when it holds +// asnBatchSize ASes, and every asnBatchTimeout. type ASHandler struct { db database.Store logger *logger.Logger - // batchMaxSize is asnBatchMaxSize. Only HandleMessage reads it, so a test - // may lower it before it hands the handler a message. - batchMaxSize int - // Batching. mu guards batch and writing only: the flush goroutine takes // the batch under mu and writes it after releasing mu, so HandleMessage - // waits for a database write only while batch is at batchMaxSize. - mu sync.Mutex - batch []asnOp - // writing is how many operations the flush goroutine took from batch - // and has not finished writing. - writing int - // batchTaken is signalled when the flush goroutine takes batch; its L is - // &mu. - batchTaken sync.Cond + // never waits for a database write. + mu sync.Mutex + // batch holds the earliest and the latest time of the updates for each AS + // since the flush goroutine last took it. It holds one entry per AS, + // however many updates arrive. + batch map[int]database.SeenTimes + // writing is how many ASes the flush goroutine took from batch and has + // not finished writing. + writing int batchFullCh chan struct{} // wakes the flush goroutine when batch is full // stopCh hands Stop's context to the flush goroutine. It holds one, so // Stop does not wait for a write in progress to hand it over. @@ -62,16 +53,11 @@ type ASHandler struct { done chan struct{} // closed when the flush goroutine returns } -type asnOp struct { - number int - timestamp time.Time -} - // NewASHandler creates and returns a new ASHandler instance. It initializes // the batching system and starts a background goroutine that periodically -// flushes accumulated ASN operations to the database. The caller must call -// Stop when finished to ensure all pending operations are flushed and the -// background goroutine is terminated. +// writes the collected ASes to the database. The caller must call Stop when +// finished to ensure all pending ASes are written and the background goroutine +// is terminated. func NewASHandler(db database.Store, logger *logger.Logger) *ASHandler { return newASHandler(db, logger, asnBatchTimeout) } @@ -83,15 +69,13 @@ func newASHandler( db database.Store, logger *logger.Logger, batchTimeout time.Duration, ) *ASHandler { h := &ASHandler{ - db: db, - logger: logger, - batchMaxSize: asnBatchMaxSize, - batch: make([]asnOp, 0, asnBatchSize), - batchFullCh: make(chan struct{}, 1), - stopCh: make(chan context.Context, 1), - done: make(chan struct{}), + db: db, + logger: logger, + batch: make(map[int]database.SeenTimes), + batchFullCh: make(chan struct{}, 1), + stopCh: make(chan context.Context, 1), + done: make(chan struct{}), } - h.batchTaken.L = &h.mu // Start the flush timer goroutine go flushLoop(batchTimeout, h.batchFullCh, h.stopCh, h.done, h.flushBatch) @@ -115,48 +99,25 @@ func (h *ASHandler) QueueCapacity() int { return asHandlerQueueSize } -// HandleMessage processes a RIS Live BGP message by extracting all ASNs from -// the AS path and queuing them for batch insertion into the database. The -// origin ASN and all transit ASNs are recorded with their associated -// timestamps. When the batch reaches the configured size threshold, the flush -// goroutine is woken to write it. +// HandleMessage adds the time of msg to the batch entry of each AS in its AS +// path, the origin AS and the members of an AS set included. When the batch +// holds asnBatchSize ASes, it wakes the flush goroutine to write it. func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) { - // Use the pre-parsed timestamp - timestamp := msg.ParsedTimestamp - - originASN := msg.Path.Origin() - 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. - for len(h.batch) >= h.batchMaxSize { - h.batchTaken.Wait() - } - - // Queue origin ASN operation - if originASN > 0 { - h.batch = append(h.batch, asnOp{ - number: originASN, - timestamp: timestamp, - }) - } - - // Also track all ASNs in the path, the members of an AS set included + // The origin AS is the last AS of the sequence, so the loop adds it too. for _, asn := range slices.Concat(msg.Path.Sequence, msg.Path.Set) { if asn > 0 { - h.batch = append(h.batch, asnOp{ - number: asn, - timestamp: timestamp, - }) + seen := h.batch[asn] + seen.Add(msg.ParsedTimestamp) + h.batch[asn] = seen } } // A full batch wakes the flush goroutine. If it is writing, the wake-up - // waits in batchFullCh and the batch keeps growing until the write ends - // or the batch reaches batchMaxSize. + // waits in batchFullCh and the batch keeps taking updates until the write + // ends. if len(h.batch) >= asnBatchSize { select { case h.batchFullCh <- struct{}{}: @@ -165,10 +126,10 @@ func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) { } } -// Stop stops the flush goroutine, which writes the operations the handler holds -// before it returns, and waits for it until ctx ends. Once ctx has ended the -// flush goroutine skips that last write, and a write still running is lost -// when the process exits, so Stop logs how many operations are not written. +// Stop stops the flush goroutine, which writes the ASes the handler holds before +// it returns, and waits for it until ctx ends. Once ctx has ended the flush +// goroutine skips that last write, and a write still running is lost when the +// process exits, so Stop logs how many ASes are not written. func (h *ASHandler) Stop(ctx context.Context) { h.stopCh <- ctx @@ -183,14 +144,14 @@ func (h *ASHandler) Stop(ctx context.Context) { if unwritten > 0 { h.logger.Error("Stop deadline passed before the AS handler wrote "+ - "its updates", "updates", unwritten) + "its ASes", "asns", unwritten) } } -// takeBatch returns the pending operations and leaves an empty batch in their -// place for HandleMessage to fill. They count in writing until flushBatch has -// written them. -func (h *ASHandler) takeBatch() []asnOp { +// takeBatch returns the pending ASes and leaves an empty batch in their place +// for HandleMessage to fill. They count in writing until flushBatch has written +// them. +func (h *ASHandler) takeBatch() map[int]database.SeenTimes { h.mu.Lock() defer h.mu.Unlock() @@ -200,8 +161,7 @@ func (h *ASHandler) takeBatch() []asnOp { batch := h.batch h.writing = len(batch) - h.batch = make([]asnOp, 0, asnBatchSize) - h.batchTaken.Broadcast() + h.batch = make(map[int]database.SeenTimes) return batch } @@ -214,21 +174,11 @@ func (h *ASHandler) flushBatch() { return } - // The earliest and the latest time the batch saw each AS. The batch holds - // several updates for an AS, from many peers, so a new AS is stored first - // seen at the earliest of them and last seen at the latest. - asns := make(map[int]database.SeenTimes) - for _, op := range batch { - seen := asns[op.number] - seen.Add(op.timestamp) - asns[op.number] = seen - } - // Process all ASNs in a single batch transaction - err := h.db.GetOrCreateASNBatch(asns) + err := h.db.GetOrCreateASNBatch(batch) if err != nil { h.logger.Error("Failed to process ASN batch", - "error", err, "count", len(asns)) + "error", err, "count", len(batch)) } h.mu.Lock() diff --git a/internal/routewatch/ashandler_internal_test.go b/internal/routewatch/ashandler_internal_test.go index dffadc6..ce15c5f 100644 --- a/internal/routewatch/ashandler_internal_test.go +++ b/internal/routewatch/ashandler_internal_test.go @@ -5,28 +5,43 @@ import ( "slices" "testing" + "git.eeqj.de/sneak/routewatch/internal/database" "git.eeqj.de/sneak/routewatch/internal/logger" + "git.eeqj.de/sneak/routewatch/internal/ristypes" ) +// pathOfASes returns an AS path of n different ASes: AS 1 to AS n. With n up to +// asnBatchSize, none of them is testASNA to testASND. +func pathOfASes(n int) ristypes.ASPath { + sequence := make([]int, n) + for i := range sequence { + sequence[i] = i + 1 + } + + return ristypes.ASPath{Sequence: sequence} +} + // TestASHandlerStoresTheEarliestAndLatestTimesOfANewAS handles two updates -// whose path carries an AS the database does not hold yet, the second 5 -// seconds after the first, in one batch into a real database. The AS is stored -// first seen at the time of the first update and last seen at the time of the -// second. +// whose path carries an AS the database does not hold yet, in one batch into a +// real database. The second is 5 seconds after the first, and the handler gets +// it first. The AS is stored first seen at the time of the first update and +// last seen at the time of the second. func TestASHandlerStoresTheEarliestAndLatestTimesOfANewAS(t *testing.T) { t.Parallel() db := newTestDatabase(t) // Built without NewASHandler's flush goroutine, so the test flushes itself. - h := &ASHandler{db: db, logger: logger.New(), batchMaxSize: asnBatchMaxSize} + h := &ASHandler{ + db: db, logger: logger.New(), batch: make(map[int]database.SeenTimes), + } first := peerAnnouncement(testPeerIP, testV4Prefix) second := peerAnnouncement(testPeerIP, testV4Prefix) second.ParsedTimestamp = first.ParsedTimestamp.Add(secondUpdateAfter) - h.HandleMessage(first) h.HandleMessage(second) + h.HandleMessage(first) h.flushBatch() asn, _, err := db.GetAS(t.Context(), testASNA) @@ -54,7 +69,9 @@ func TestASHandlerRecordsNothingFromAPathWithASZero(t *testing.T) { store := newMockStore() // Built without NewASHandler's flush goroutine, so the test flushes itself. - h := &ASHandler{db: store, logger: logger.New(), batchMaxSize: asnBatchMaxSize} + h := &ASHandler{ + db: store, logger: logger.New(), batch: make(map[int]database.SeenTimes), + } msg := peerAnnouncement(testPeerIP, testV4Prefix) msg.Path = decodePath(t, testASNA, 0, testASNB) diff --git a/internal/routewatch/batchwrite_internal_test.go b/internal/routewatch/batchwrite_internal_test.go index 0f29c84..016a5c9 100644 --- a/internal/routewatch/batchwrite_internal_test.go +++ b/internal/routewatch/batchwrite_internal_test.go @@ -178,6 +178,7 @@ type stopLog struct { Updates int `json:"updates"` TableChanges int `json:"table_changes"` Paths int `json:"paths"` + ASNs int `json:"asns"` } // newErrorLogger returns a logger that writes only errors, as JSON lines, to @@ -495,20 +496,20 @@ func TestPrefixHandlerStopCountsTheTableChangesOfAWriteInProgress(t *testing.T) } } +// TestASHandlerSkipsItsLastWriteAfterTheStopDeadline also checks what Stop +// logs: the asnBatchSize ASes of the held write and testASNB, handled since, +// are not written. func TestASHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) { t.Parallel() - // Each message gives an update for its origin AS and one for each AS - // in its path: fill asnBatchSize+1, extra 2. - const wantUpdates = asnBatchSize + 1 + 2 + const wantASNs = asnBatchSize + 1 var logs bytes.Buffer store := newBlockingStore() h := newASHandler(store, newErrorLogger(&logs), longerThanAnyTest) - fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)} stopWhileAWriteIsBlocked(t, store, h, h.done, - []*ristypes.RISMessage{{Path: fill}}, + []*ristypes.RISMessage{{Path: pathOfASes(asnBatchSize)}}, &ristypes.RISMessage{Path: ristypes.ASPath{Sequence: []int{testASNB}}}) store.mu.Lock() @@ -518,26 +519,27 @@ func TestASHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) { t.Errorf("AS%d was written after the stop deadline", testASNB) } - if logged := lastLogged(t, &logs); logged.Updates != wantUpdates { - t.Errorf("Stop logged %d updates not written, want %d", - logged.Updates, wantUpdates) + if logged := lastLogged(t, &logs); logged.ASNs != wantASNs { + t.Errorf("Stop logged %d ASes not written, want %d", + logged.ASNs, wantASNs) } } // TestASHandlerSkipsAFullBatchAfterTheStopDeadline hands the AS handler a full // batch while a write is held, so the batch's wake-up of the flush goroutine -// waits. The stop deadline passes meanwhile. When the write ends, the flush -// goroutine must stop instead of writing the batch. +// waits. The full batch holds the fill's ASes again, and testASNB. The stop +// deadline passes meanwhile. When the write ends, the flush goroutine must stop +// instead of writing the batch. func TestASHandlerSkipsAFullBatchAfterTheStopDeadline(t *testing.T) { t.Parallel() store := newBlockingStore() h := newASHandler(store, logger.New(), longerThanAnyTest) - fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)} - full := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNB}, asnBatchSize)} + fill := &ristypes.RISMessage{Path: pathOfASes(asnBatchSize)} stopWhileAWriteIsBlocked(t, store, h, h.done, - []*ristypes.RISMessage{{Path: fill}}, - &ristypes.RISMessage{Path: full}) + []*ristypes.RISMessage{fill}, + fill, + &ristypes.RISMessage{Path: ristypes.ASPath{Sequence: []int{testASNB}}}) store.mu.Lock() defer store.mu.Unlock() @@ -703,9 +705,8 @@ func TestASHandlerHandlesMessagesWhileAWriteIsBlocked(t *testing.T) { store := newBlockingStore() h := newASHandler(store, logger.New(), longerThanAnyTest) - fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)} handleWhileAWriteIsBlocked(t, store, h, - []*ristypes.RISMessage{{Path: fill}}, + []*ristypes.RISMessage{{Path: pathOfASes(asnBatchSize)}}, &ristypes.RISMessage{Path: ristypes.ASPath{Sequence: []int{testASNB}}}) store.mu.Lock() @@ -762,25 +763,41 @@ func TestPrefixHandlerWaitsAtTheMaximumBatchSize(t *testing.T) { } } -func TestASHandlerWaitsAtTheMaximumBatchSize(t *testing.T) { +// TestASHandlerNeverWaitsForAWrite starts a write that the store holds. Then it +// hands the AS handler twenty messages whose paths each hold one AS +// asnBatchSize times. HandleMessage must return each time, and the batch must +// hold one entry, for that AS. +func TestASHandlerNeverWaitsForAWrite(t *testing.T) { t.Parallel() + const messages = 20 + store := newBlockingStore() h := newASHandler(store, logger.New(), longerThanAnyTest) - h.batchMaxSize = asnBatchSize - fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)} - toMax := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)} - handleAtMaxBatchSize(t, store, h, - []*ristypes.RISMessage{{Path: fill}}, - []*ristypes.RISMessage{{Path: toMax}}, - &ristypes.RISMessage{Path: ristypes.ASPath{Sequence: []int{testASNB}}}) - store.mu.Lock() - defer store.mu.Unlock() + defer stopInTime(t, h) + defer store.releaseWrites() - if !store.asns[testASNB] { - t.Errorf("Stop did not write AS%d, handled after the batch reached its "+ - "maximum size", testASNB) + startBlockedWrite(t, store, h, + []*ristypes.RISMessage{{Path: pathOfASes(asnBatchSize)}}) + + oneAS := &ristypes.RISMessage{ + Path: ristypes.ASPath{Sequence: slices.Repeat([]int{testASNB}, asnBatchSize)}, + } + + if !returnsInTime(func() { + for range messages { + h.HandleMessage(oneAS) + } + }) { + t.Fatal("HandleMessage did not return while a database write was blocked") + } + + h.mu.Lock() + defer h.mu.Unlock() + + if len(h.batch) != 1 { + t.Errorf("the batch holds %d entries for one AS, want 1", len(h.batch)) } } @@ -844,13 +861,11 @@ func TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull(t *testing.T) { batchSize: asnBatchSize, handle: func(n int) chan struct{} { h := &ASHandler{ - batchMaxSize: asnBatchMaxSize, - batchFullCh: make(chan struct{}, 1), + batch: make(map[int]database.SeenTimes), + batchFullCh: make(chan struct{}, 1), } - // The origin AS and each AS in the path are one update each. - h.HandleMessage(&ristypes.RISMessage{ - Path: ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, n-1)}, - }) + // Each of the path's n ASes is one entry of the batch. + h.HandleMessage(&ristypes.RISMessage{Path: pathOfASes(n)}) return h.batchFullCh },