From 73793e68e0a5a3fff31f9550901e2ad8acaa375b Mon Sep 17 00:00:00 2001 From: sneak Date: Wed, 14 Oct 2026 15:16:57 +0000 Subject: [PATCH] Write each handler batch from its flush goroutine, outside the mutex (closes #65) The AS, peer and prefix handlers wrote their batch to the database while holding the mutex HandleMessage needs, and HandleMessage wrote a full batch itself. For as long as a write took, the handler took nothing off its queue, and the streamer dropped updates once the queue was half full. Now only the flush goroutine writes: it takes the batch under the mutex, releases it, then writes. A full batch wakes the flush goroutine instead. Stop still writes what the handler holds. While a write runs, the next batch has no size limit; the README memory section says so. Model: opus-5-5 --- README.md | 4 + internal/routewatch/ashandler.go | 76 ++++++--- internal/routewatch/batchwrite_test.go | 192 ++++++++++++++++++++++ internal/routewatch/peerhandler.go | 76 ++++++--- internal/routewatch/prefixhandler.go | 74 ++++++--- internal/routewatch/prefixhandler_test.go | 13 +- 6 files changed, 345 insertions(+), 90 deletions(-) create mode 100644 internal/routewatch/batchwrite_test.go diff --git a/README.md b/README.md index 05615d5..648621b 100644 --- a/README.md +++ b/README.md @@ -233,6 +233,10 @@ What happens at each limit: keeps running. - Handler queues: each of the four handler queues holds at most 20,000 messages. When a queue fills, the streamer drops messages instead of blocking. +- Handler batches: while the AS, peer or prefix handler writes a batch to the + database, the updates that arrive collect in its next batch, which has no size + limit. A database that stays slower than the feed grows memory instead of + dropping updates. With `DEBUG=routewatch` the daemon logs a `System stats` line every 60 seconds with the goroutine count and Go memory figures. diff --git a/internal/routewatch/ashandler.go b/internal/routewatch/ashandler.go index 3eae6f3..86376db 100644 --- a/internal/routewatch/ashandler.go +++ b/internal/routewatch/ashandler.go @@ -32,12 +32,15 @@ type ASHandler struct { db database.Store logger *logger.Logger - // Batching - mu sync.Mutex - batch []asnOp - lastFlush time.Time - stopCh chan struct{} - wg sync.WaitGroup + // Batching. mu guards batch only: the flush goroutine takes the batch + // under mu and writes it after releasing mu, so HandleMessage never + // waits for a database write. + mu sync.Mutex + batch []asnOp + batchFullCh chan struct{} // wakes the flush goroutine when batch is full + lastFlush time.Time // used only by the flush goroutine + stopCh chan struct{} + wg sync.WaitGroup } type asnOp struct { @@ -52,11 +55,12 @@ type asnOp struct { // background goroutine is terminated. func NewASHandler(db database.Store, logger *logger.Logger) *ASHandler { h := &ASHandler{ - db: db, - logger: logger, - batch: make([]asnOp, 0, asnBatchSize), - lastFlush: time.Now(), - stopCh: make(chan struct{}), + db: db, + logger: logger, + batch: make([]asnOp, 0, asnBatchSize), + batchFullCh: make(chan struct{}, 1), + lastFlush: time.Now(), + stopCh: make(chan struct{}), } // Start the flush timer goroutine @@ -85,8 +89,8 @@ func (h *ASHandler) QueueCapacity() int { // 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 (last element in the path) and all transit ASNs are recorded -// with their associated timestamps. The batch is automatically flushed when -// it reaches the configured size threshold. +// with their associated timestamps. When the batch reaches the configured size +// threshold, the flush goroutine is woken to write it. func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) { // Use the pre-parsed timestamp timestamp := msg.ParsedTimestamp @@ -118,13 +122,18 @@ func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) { } } - // Check if we need to flush + // 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. if len(h.batch) >= asnBatchSize { - h.flushBatchLocked() + select { + case h.batchFullCh <- struct{}{}: + default: // already woken + } } } -// flushLoop runs in a goroutine and periodically flushes batches +// flushLoop runs in a goroutine, the only one that writes the batch: it +// flushes when the batch is full, periodically, and once more on Stop. func (h *ASHandler) flushLoop() { defer h.wg.Done() ticker := time.NewTicker(asnBatchTimeout) @@ -133,31 +142,46 @@ func (h *ASHandler) flushLoop() { for { select { case <-ticker.C: - h.mu.Lock() if time.Since(h.lastFlush) >= asnBatchTimeout { - h.flushBatchLocked() + h.flushBatch() } - h.mu.Unlock() + case <-h.batchFullCh: + h.flushBatch() case <-h.stopCh: // Final flush - h.mu.Lock() - h.flushBatchLocked() - h.mu.Unlock() + h.flushBatch() return } } } -// flushBatchLocked flushes the ASN batch to the database (must be called with mutex held) -func (h *ASHandler) flushBatchLocked() { +// takeBatch returns the pending operations and leaves an empty batch in their +// place for HandleMessage to fill. +func (h *ASHandler) takeBatch() []asnOp { + h.mu.Lock() + defer h.mu.Unlock() + if len(h.batch) == 0 { + return nil + } + batch := h.batch + h.batch = make([]asnOp, 0, asnBatchSize) + + return batch +} + +// flushBatch takes the pending batch and writes it to the database, without +// holding the mutex during the write. +func (h *ASHandler) flushBatch() { + batch := h.takeBatch() + if len(batch) == 0 { return } // Process ASNs first (deduped) asnMap := make(map[int]time.Time) - for _, op := range h.batch { + for _, op := range batch { if existing, ok := asnMap[op.number]; !ok || op.timestamp.After(existing) { asnMap[op.number] = op.timestamp } @@ -168,8 +192,6 @@ func (h *ASHandler) flushBatchLocked() { h.logger.Error("Failed to process ASN batch", "error", err, "count", len(asnMap)) } - // Clear batch - h.batch = h.batch[:0] h.lastFlush = time.Now() } diff --git a/internal/routewatch/batchwrite_test.go b/internal/routewatch/batchwrite_test.go new file mode 100644 index 0000000..4eefa7b --- /dev/null +++ b/internal/routewatch/batchwrite_test.go @@ -0,0 +1,192 @@ +package routewatch + +import ( + "slices" + "sync" + "testing" + "time" + + "git.eeqj.de/sneak/routewatch/internal/database" + "git.eeqj.de/sneak/routewatch/internal/logger" + "git.eeqj.de/sneak/routewatch/internal/ristypes" +) + +// handleMessageTimeout is how long a test waits for HandleMessage to return, +// or for a write to start, before it fails. +const handleMessageTimeout = 2 * time.Second + +// blockingStore is a mockStore whose batch writes for the AS, peer and prefix +// handlers wait until the test closes release. writeStarted is closed when the +// first of them starts. It records what each write was given. +type blockingStore struct { + *mockStore + + release chan struct{} + writeStarted chan struct{} + startOnce sync.Once + + mu sync.Mutex + prefixes map[string]bool + asns map[int]bool + peers map[string]bool +} + +func newBlockingStore() *blockingStore { + return &blockingStore{ + mockStore: newMockStore(), + release: make(chan struct{}), + writeStarted: make(chan struct{}), + prefixes: make(map[string]bool), + asns: make(map[int]bool), + peers: make(map[string]bool), + } +} + +func (s *blockingStore) waitForRelease() { + s.startOnce.Do(func() { close(s.writeStarted) }) + <-s.release +} + +func (s *blockingStore) UpsertLiveRouteBatch(routes []*database.LiveRoute) error { + s.waitForRelease() + s.mu.Lock() + defer s.mu.Unlock() + for _, route := range routes { + s.prefixes[route.Prefix] = true + } + + return nil +} + +func (s *blockingStore) GetOrCreateASNBatch(asns map[int]time.Time) error { + s.waitForRelease() + s.mu.Lock() + defer s.mu.Unlock() + for asn := range asns { + s.asns[asn] = true + } + + return nil +} + +func (s *blockingStore) UpdatePeerBatch(peers map[string]database.PeerUpdate) error { + s.waitForRelease() + s.mu.Lock() + defer s.mu.Unlock() + for peerIP := range peers { + s.peers[peerIP] = true + } + + return nil +} + +// returnsInTime reports whether f returns within handleMessageTimeout. +func returnsInTime(f func()) bool { + done := make(chan struct{}) + go func() { + f() + close(done) + }() + + select { + case <-done: + return true + case <-time.After(handleMessageTimeout): + return false + } +} + +// handleWhileAWriteIsBlocked hands h the messages in fill, which fill its +// batch so that the handler writes it to store, waits until the write starts, +// and hands h the message extra while store holds the write. HandleMessage +// must return each time. Then it lets the writes finish and stops h, which +// writes what it holds. +func handleWhileAWriteIsBlocked( + t *testing.T, + store *blockingStore, + h interface { + HandleMessage(*ristypes.RISMessage) + Stop() + }, + fill []*ristypes.RISMessage, + extra *ristypes.RISMessage, +) { + t.Helper() + defer h.Stop() + defer close(store.release) + + if !returnsInTime(func() { + for _, msg := range fill { + h.HandleMessage(msg) + } + }) { + t.Fatal("HandleMessage did not return after filling the batch: it waits for the database write") + } + + select { + case <-store.writeStarted: + case <-time.After(handleMessageTimeout): + t.Fatal("the handler did not write its full batch") + } + + if !returnsInTime(func() { h.HandleMessage(extra) }) { + t.Fatal("HandleMessage did not return while a database write was blocked") + } +} + +func TestPrefixHandlerHandlesMessagesWhileAWriteIsBlocked(t *testing.T) { + const ( + fillPrefix = "192.0.2.0/24" + extraPrefix = "198.51.100.0/24" + ) + + announce := func(prefixes []string) *ristypes.RISMessage { + return &ristypes.RISMessage{ + Peer: testPeerIP, + Path: ristypes.ASPath{testASNA}, + Announcements: []ristypes.RISAnnouncement{{NextHop: testPeerIP, Prefixes: prefixes}}, + } + } + + store := newBlockingStore() + h := NewPrefixHandler(store, logger.New()) + handleWhileAWriteIsBlocked(t, store, h, + []*ristypes.RISMessage{announce(slices.Repeat([]string{fillPrefix}, prefixBatchSize))}, + announce([]string{extraPrefix})) + + store.mu.Lock() + defer store.mu.Unlock() + if !store.prefixes[extraPrefix] { + t.Errorf("Stop did not write %s, handled while a write was blocked", extraPrefix) + } +} + +func TestASHandlerHandlesMessagesWhileAWriteIsBlocked(t *testing.T) { + store := newBlockingStore() + h := NewASHandler(store, logger.New()) + handleWhileAWriteIsBlocked(t, store, h, + []*ristypes.RISMessage{{Path: slices.Repeat(ristypes.ASPath{testASNA}, asnBatchSize)}}, + &ristypes.RISMessage{Path: ristypes.ASPath{testASNB}}) + + store.mu.Lock() + defer store.mu.Unlock() + if !store.asns[testASNB] { + t.Errorf("Stop did not write AS%d, handled while a write was blocked", testASNB) + } +} + +func TestPeerHandlerHandlesMessagesWhileAWriteIsBlocked(t *testing.T) { + const otherPeerIP = "2001:db8:ffff::2" + + store := newBlockingStore() + h := NewPeerHandler(store, logger.New()) + handleWhileAWriteIsBlocked(t, store, h, + slices.Repeat([]*ristypes.RISMessage{{Peer: testPeerIP}}, peerBatchSize), + &ristypes.RISMessage{Peer: otherPeerIP}) + + store.mu.Lock() + defer store.mu.Unlock() + if !store.peers[otherPeerIP] { + t.Errorf("Stop did not write peer %s, handled while a write was blocked", otherPeerIP) + } +} diff --git a/internal/routewatch/peerhandler.go b/internal/routewatch/peerhandler.go index 2f7bf70..3d00991 100644 --- a/internal/routewatch/peerhandler.go +++ b/internal/routewatch/peerhandler.go @@ -35,12 +35,15 @@ type PeerHandler struct { db database.Store logger *logger.Logger - // Batching - mu sync.Mutex - peerBatch []peerUpdate - lastFlush time.Time - stopCh chan struct{} - wg sync.WaitGroup + // Batching. mu guards peerBatch only: the flush goroutine takes the batch + // under mu and writes it after releasing mu, so HandleMessage never + // waits for a database write. + mu sync.Mutex + peerBatch []peerUpdate + batchFullCh chan struct{} // wakes the flush goroutine when peerBatch is full + lastFlush time.Time // used only by the flush goroutine + stopCh chan struct{} + wg sync.WaitGroup } type peerUpdate struct { @@ -56,11 +59,12 @@ type peerUpdate struct { // should be stopped by calling Stop when it is no longer needed. func NewPeerHandler(db database.Store, logger *logger.Logger) *PeerHandler { h := &PeerHandler{ - db: db, - logger: logger, - peerBatch: make([]peerUpdate, 0, peerBatchSize), - lastFlush: time.Now(), - stopCh: make(chan struct{}), + db: db, + logger: logger, + peerBatch: make([]peerUpdate, 0, peerBatchSize), + batchFullCh: make(chan struct{}, 1), + lastFlush: time.Now(), + stopCh: make(chan struct{}), } // Start the flush timer goroutine @@ -87,8 +91,8 @@ func (h *PeerHandler) QueueCapacity() int { // HandleMessage processes a RIS message to track peer information. // It extracts the peer IP address and ASN from the message and adds -// the update to an internal batch. When the batch reaches peerBatchSize -// or the batch timeout expires, the batch is flushed to the database. +// the update to an internal batch. The flush goroutine writes the batch to the +// database when it reaches peerBatchSize or the batch timeout expires. func (h *PeerHandler) HandleMessage(msg *ristypes.RISMessage) { // Parse peer ASN from string peerASN := 0 @@ -109,13 +113,18 @@ func (h *PeerHandler) HandleMessage(msg *ristypes.RISMessage) { timestamp: msg.ParsedTimestamp, }) - // Check if we need to flush + // 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. if len(h.peerBatch) >= peerBatchSize { - h.flushBatchLocked() + select { + case h.batchFullCh <- struct{}{}: + default: // already woken + } } } -// flushLoop runs in a goroutine and periodically flushes batches +// flushLoop runs in a goroutine, the only one that writes the batch: it +// flushes when the batch is full, periodically, and once more on Stop. func (h *PeerHandler) flushLoop() { defer h.wg.Done() ticker := time.NewTicker(peerBatchTimeout) @@ -124,31 +133,46 @@ func (h *PeerHandler) flushLoop() { for { select { case <-ticker.C: - h.mu.Lock() if time.Since(h.lastFlush) >= peerBatchTimeout { - h.flushBatchLocked() + h.flushBatch() } - h.mu.Unlock() + case <-h.batchFullCh: + h.flushBatch() case <-h.stopCh: // Final flush - h.mu.Lock() - h.flushBatchLocked() - h.mu.Unlock() + h.flushBatch() return } } } -// flushBatchLocked flushes the peer batch to the database (must be called with mutex held) -func (h *PeerHandler) flushBatchLocked() { +// takeBatch returns the pending updates and leaves an empty batch in their +// place for HandleMessage to fill. +func (h *PeerHandler) takeBatch() []peerUpdate { + h.mu.Lock() + defer h.mu.Unlock() + if len(h.peerBatch) == 0 { + return nil + } + batch := h.peerBatch + h.peerBatch = make([]peerUpdate, 0, peerBatchSize) + + return batch +} + +// flushBatch takes the pending batch and writes it to the database, without +// holding the mutex during the write. +func (h *PeerHandler) flushBatch() { + batch := h.takeBatch() + if len(batch) == 0 { return } // Deduplicate by peer IP, keeping the latest update for each peer peerMap := make(map[string]peerUpdate) - for _, update := range h.peerBatch { + for _, update := range batch { if existing, ok := peerMap[update.peerIP]; !ok || update.timestamp.After(existing.timestamp) { peerMap[update.peerIP] = update } @@ -170,8 +194,6 @@ func (h *PeerHandler) flushBatchLocked() { h.logger.Error("Failed to process peer batch", "error", err, "count", len(dbPeerMap)) } - // Clear batch - h.peerBatch = h.peerBatch[:0] h.lastFlush = time.Now() } diff --git a/internal/routewatch/prefixhandler.go b/internal/routewatch/prefixhandler.go index 59f3991..952bb16 100644 --- a/internal/routewatch/prefixhandler.go +++ b/internal/routewatch/prefixhandler.go @@ -40,12 +40,15 @@ type PrefixHandler struct { logger *logger.Logger metrics *metrics.Tracker - // Batching - mu sync.Mutex - batch []prefixUpdate - lastFlush time.Time - stopCh chan struct{} - wg sync.WaitGroup + // Batching. mu guards batch only: the flush goroutine takes the batch + // under mu and writes it after releasing mu, so HandleMessage never + // waits for a database write. + mu sync.Mutex + batch []prefixUpdate + batchFullCh chan struct{} // wakes the flush goroutine when batch is full + lastFlush time.Time // used only by the flush goroutine + stopCh chan struct{} + wg sync.WaitGroup } type prefixUpdate struct { @@ -60,11 +63,12 @@ type prefixUpdate struct { // NewPrefixHandler creates a new batched prefix tracking handler func NewPrefixHandler(db database.Store, logger *logger.Logger) *PrefixHandler { h := &PrefixHandler{ - db: db, - logger: logger, - batch: make([]prefixUpdate, 0, prefixBatchSize), - lastFlush: time.Now(), - stopCh: make(chan struct{}), + db: db, + logger: logger, + batch: make([]prefixUpdate, 0, prefixBatchSize), + batchFullCh: make(chan struct{}, 1), + lastFlush: time.Now(), + stopCh: make(chan struct{}), } // Start the flush timer goroutine @@ -140,13 +144,18 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) { } } - // Check if we need to flush + // 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. if len(h.batch) >= prefixBatchSize { - h.flushBatchLocked() + select { + case h.batchFullCh <- struct{}{}: + default: // already woken + } } } -// flushLoop runs in a goroutine and periodically flushes batches +// flushLoop runs in a goroutine, the only one that writes the batch: it +// flushes when the batch is full, periodically, and once more on Stop. func (h *PrefixHandler) flushLoop() { defer h.wg.Done() ticker := time.NewTicker(prefixBatchTimeout) @@ -155,30 +164,45 @@ func (h *PrefixHandler) flushLoop() { for { select { case <-ticker.C: - h.mu.Lock() if time.Since(h.lastFlush) >= prefixBatchTimeout { - h.flushBatchLocked() + h.flushBatch() } - h.mu.Unlock() + case <-h.batchFullCh: + h.flushBatch() case <-h.stopCh: // Final flush - h.mu.Lock() - h.flushBatchLocked() - h.mu.Unlock() + h.flushBatch() return } } } -// flushBatchLocked flushes the prefix batch to the database (must be called with mutex held) -func (h *PrefixHandler) flushBatchLocked() { +// takeBatch returns the pending updates and leaves an empty batch in their +// place for HandleMessage to fill. +func (h *PrefixHandler) takeBatch() []prefixUpdate { + h.mu.Lock() + defer h.mu.Unlock() + if len(h.batch) == 0 { + return nil + } + batch := h.batch + h.batch = make([]prefixUpdate, 0, prefixBatchSize) + + return batch +} + +// flushBatch takes the pending batch and writes it to the database, without +// holding the mutex during the write. +func (h *PrefixHandler) flushBatch() { + batch := h.takeBatch() + if len(batch) == 0 { return } startTime := time.Now() - batchSize := len(h.batch) + batchSize := len(batch) // For each prefix and peer, keep only the last update the feed delivered. // Updates from different peers for the same prefix are all kept. Arrival @@ -186,7 +210,7 @@ func (h *PrefixHandler) flushBatchLocked() { // updates within one second carry the same one. type prefixAndPeer struct{ prefix, peer string } lastUpdates := make(map[prefixAndPeer]prefixUpdate) - for _, update := range h.batch { + for _, update := range batch { lastUpdates[prefixAndPeer{update.prefix, update.peer}] = update } @@ -261,8 +285,6 @@ func (h *PrefixHandler) flushBatchLocked() { "duration_ms", elapsed.Milliseconds(), ) - // Clear batch - h.batch = h.batch[:0] h.lastFlush = time.Now() } diff --git a/internal/routewatch/prefixhandler_test.go b/internal/routewatch/prefixhandler_test.go index fcfa9a3..c94ff5f 100644 --- a/internal/routewatch/prefixhandler_test.go +++ b/internal/routewatch/prefixhandler_test.go @@ -30,11 +30,6 @@ func TestPrefixHandlerStoresPrefixesTheIPLookupFinds(t *testing.T) { // Built without NewPrefixHandler's flush timer, so the test flushes itself. h := &PrefixHandler{db: db, logger: logger.New()} - flush := func() { - h.mu.Lock() - defer h.mu.Unlock() - h.flushBatchLocked() - } ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) h.HandleMessage(&ristypes.RISMessage{ @@ -46,7 +41,7 @@ func TestPrefixHandlerStoresPrefixesTheIPLookupFinds(t *testing.T) { Prefixes: []string{"2001:db8:1::/48", "192.0.2.0/24"}, }}, }) - flush() + h.flushBatch() for ip, want := range map[string]string{ "2001:db8:1::1": "2001:db8:1::/48", @@ -66,7 +61,7 @@ func TestPrefixHandlerStoresPrefixesTheIPLookupFinds(t *testing.T) { ParsedTimestamp: ts.Add(time.Minute), Withdrawals: []string{"2001:db8:1:0:0:0:0:0/48", "192.0.2.0/24"}, }) - flush() + h.flushBatch() for _, ip := range []string{"2001:db8:1::1", "192.0.2.1"} { if info, err := db.GetIPInfo(ip); !errors.Is(err, database.ErrNoRoute) { @@ -164,9 +159,7 @@ func TestPrefixHandlerAppliesEachPeersLastUpdateInABatch(t *testing.T) { for _, msg := range batch { h.HandleMessage(msg) } - h.mu.Lock() - h.flushBatchLocked() - h.mu.Unlock() + h.flushBatch() } ctx := context.Background()