diff --git a/TODO.md b/TODO.md index 7e6de2c..59637e2 100644 --- a/TODO.md +++ b/TODO.md @@ -33,6 +33,11 @@ tested. # Completed Steps +- 2026-10-27: once the stop deadline has passed, the AS, peer and peering + handlers start no write either. A full batch or a tick of the timer that + waited while a write ran past the deadline could still start one. The flush + goroutines of the prefix, AS and peer handlers now run the same code (closes + #189) - 2026-10-27: once the stop deadline has passed, the prefix handler starts no flush. A batch that filled while a write ran past the deadline could still be written after it (closes #188) diff --git a/internal/routewatch/ashandler.go b/internal/routewatch/ashandler.go index d82b9f8..cee6699 100644 --- a/internal/routewatch/ashandler.go +++ b/internal/routewatch/ashandler.go @@ -52,7 +52,6 @@ type ASHandler struct { // &mu. batchTaken sync.Cond batchFullCh chan struct{} // wakes the flush goroutine when batch is full - lastFlush time.Time // used only by the flush goroutine // 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. stopCh chan context.Context @@ -70,19 +69,27 @@ type asnOp struct { // Stop when finished to ensure all pending operations are flushed and the // background goroutine is terminated. func NewASHandler(db database.Store, logger *logger.Logger) *ASHandler { + return newASHandler(db, logger, asnBatchTimeout) +} + +// newASHandler is NewASHandler with the flush timer's interval, batchTimeout, +// in place of asnBatchTimeout. A test that the timer must not disturb gives an +// interval longer than the test runs. +func newASHandler( + db database.Store, logger *logger.Logger, batchTimeout time.Duration, +) *ASHandler { h := &ASHandler{ db: db, logger: logger, batch: make([]asnOp, 0, asnBatchSize), batchFullCh: make(chan struct{}, 1), - lastFlush: time.Now(), stopCh: make(chan context.Context, 1), done: make(chan struct{}), } h.batchTaken.L = &h.mu // Start the flush timer goroutine - go h.flushLoop() + go flushLoop(batchTimeout, h.batchFullCh, h.stopCh, h.done, h.flushBatch) return h } @@ -175,33 +182,6 @@ func (h *ASHandler) Stop(ctx context.Context) { } } -// 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 close(h.done) - - ticker := time.NewTicker(asnBatchTimeout) - defer ticker.Stop() - - for { - select { - case <-ticker.C: - if time.Since(h.lastFlush) >= asnBatchTimeout { - h.flushBatch() - } - case <-h.batchFullCh: - h.flushBatch() - case ctx := <-h.stopCh: - // Final flush, unless the stop deadline has passed - if ctx.Err() == nil { - h.flushBatch() - } - - return - } - } -} - // 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. @@ -222,11 +202,11 @@ func (h *ASHandler) takeBatch() []asnOp { } // flushBatch takes the pending batch and writes it to the database, without -// holding the mutex during the write. -func (h *ASHandler) flushBatch() { +// holding the mutex during the write. It reports whether there was a batch. +func (h *ASHandler) flushBatch() bool { batch := h.takeBatch() if len(batch) == 0 { - return + return false } // The earliest and the latest time the batch saw each AS. The batch holds @@ -250,5 +230,5 @@ func (h *ASHandler) flushBatch() { h.writing = 0 h.mu.Unlock() - h.lastFlush = time.Now() + return true } diff --git a/internal/routewatch/batchwrite.go b/internal/routewatch/batchwrite.go new file mode 100644 index 0000000..53ef900 --- /dev/null +++ b/internal/routewatch/batchwrite.go @@ -0,0 +1,64 @@ +package routewatch + +import ( + "context" + "time" +) + +// flushLoop is the flush goroutine of the prefix, AS and peer handlers. It is +// the only goroutine that writes a handler's batch. flushBatch takes the batch, +// writes it, and reports whether there was one. +// +// flushLoop calls flushBatch when batchFullCh wakes it. It also calls it every +// batchTimeout, unless it wrote a batch less than batchTimeout ago. When stopCh +// hands it Stop's context, it calls flushBatch once more, unless the stop +// deadline has passed. Then it closes done and returns. +func flushLoop( + batchTimeout time.Duration, + batchFullCh <-chan struct{}, + stopCh <-chan context.Context, + done chan<- struct{}, + flushBatch func() bool, +) { + defer close(done) + + ticker := time.NewTicker(batchTimeout) + defer ticker.Stop() + + lastFlush := time.Now() + + for { + // Stop goes first. A write can run past the stop deadline while a full + // batch's wake-up or a tick waits beside Stop's context. The select + // below picks one of them at random, but past the deadline no flush + // may start. + select { + case ctx := <-stopCh: + // Final flush, unless the stop deadline has passed + if ctx.Err() == nil { + flushBatch() + } + + return + default: + } + + select { + case <-ticker.C: + if time.Since(lastFlush) >= batchTimeout && flushBatch() { + lastFlush = time.Now() + } + case <-batchFullCh: + if flushBatch() { + lastFlush = time.Now() + } + case ctx := <-stopCh: + // Final flush, unless the stop deadline has passed + if ctx.Err() == nil { + flushBatch() + } + + return + } + } +} diff --git a/internal/routewatch/batchwrite_internal_test.go b/internal/routewatch/batchwrite_internal_test.go index c14ac79..f659a0f 100644 --- a/internal/routewatch/batchwrite_internal_test.go +++ b/internal/routewatch/batchwrite_internal_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "fmt" "log/slog" "slices" "sync" @@ -168,9 +169,10 @@ func returnsInTime(f func()) bool { } } -// startBlockedWrite hands h the messages in fill, which fill its batch so that -// the handler writes it to store, and waits until that write starts. store -// holds it until the test calls releaseWrites. +// startBlockedWrite hands h the messages in fill and waits until h starts to +// write them to store. The prefix, AS and peer handlers start when fill fills +// their batch, and the peering handler on its timer. store holds the write +// until the test calls releaseWrites. func startBlockedWrite( t *testing.T, store *blockingStore, @@ -193,7 +195,7 @@ func startBlockedWrite( select { case <-store.writeStarted: case <-time.After(handleMessageTimeout): - t.Fatal("the handler did not write its full batch") + t.Fatal("the handler did not start to write the messages") } } @@ -273,7 +275,7 @@ func handleAtMaxBatchSize( } // stopWhileAWriteIsBlocked starts a blocked write with the messages in fill, -// hands h the message extra, and stops h with a deadline that passes while +// hands h the messages in extra, and stops h with a deadline that passes while // store holds the write. Stop must return all the same. Then it lets the write // finish and waits until done, closed when h's flush goroutine returns. Past // the deadline that goroutine skips its last write, so extra is not written, @@ -287,7 +289,7 @@ func stopWhileAWriteIsBlocked( }, done <-chan struct{}, fill []*ristypes.RISMessage, - extra *ristypes.RISMessage, + extra ...*ristypes.RISMessage, ) { t.Helper() @@ -296,7 +298,10 @@ func stopWhileAWriteIsBlocked( defer store.releaseWrites() startBlockedWrite(t, store, h, fill) - h.HandleMessage(extra) + + for _, msg := range extra { + h.HandleMessage(msg) + } ctx, cancel := context.WithTimeout(t.Context(), stopDeadline) defer cancel() @@ -428,7 +433,7 @@ func TestASHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) { var logs bytes.Buffer store := newBlockingStore() - h := NewASHandler(store, newErrorLogger(&logs)) + 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}}, @@ -447,6 +452,29 @@ func TestASHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) { } } +// 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. +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)} + stopWhileAWriteIsBlocked(t, store, h, h.done, + []*ristypes.RISMessage{{Path: fill}}, + &ristypes.RISMessage{Path: full}) + + store.mu.Lock() + defer store.mu.Unlock() + + if store.asns[testASNB] { + t.Errorf("AS%d was written after the stop deadline", testASNB) + } +} + func TestPeerHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) { t.Parallel() @@ -455,7 +483,7 @@ func TestPeerHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) { var logs bytes.Buffer store := newBlockingStore() - h := NewPeerHandler(store, newErrorLogger(&logs)) + h := newPeerHandler(store, newErrorLogger(&logs), longerThanAnyTest) stopWhileAWriteIsBlocked(t, store, h, h.done, slices.Repeat([]*ristypes.RISMessage{{Peer: testPeerIP}}, peerBatchSize), &ristypes.RISMessage{Peer: otherPeerIP}) @@ -473,6 +501,30 @@ func TestPeerHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) { } } +// TestPeerHandlerSkipsAFullBatchAfterTheStopDeadline hands the peer 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. +func TestPeerHandlerSkipsAFullBatchAfterTheStopDeadline(t *testing.T) { + t.Parallel() + + const otherPeerIP = "2001:db8:ffff::2" + + store := newBlockingStore() + h := newPeerHandler(store, logger.New(), longerThanAnyTest) + stopWhileAWriteIsBlocked(t, store, h, h.done, + slices.Repeat([]*ristypes.RISMessage{{Peer: testPeerIP}}, peerBatchSize), + slices.Repeat([]*ristypes.RISMessage{{Peer: otherPeerIP}}, + peerBatchSize)...) + + store.mu.Lock() + defer store.mu.Unlock() + + if store.peers[otherPeerIP] { + t.Errorf("peer %s was written after the stop deadline", otherPeerIP) + } +} + // TestPeeringHandlerStopCountsTheWriteInProgress hands a running peering // handler one AS path and stops it with a deadline that passes while store // holds Stop's last write, of that path's peering. Stop must log that path as @@ -512,6 +564,35 @@ func TestPeeringHandlerStopCountsTheWriteInProgress(t *testing.T) { } } +// TestPeeringHandlerStartsNoWriteAfterTheStopDeadline has the peering handler's +// timer start a write that store holds, and hands the handler one more AS path +// meanwhile. The stop deadline passes, and a tick of the timer waits. When the +// write ends, the processing goroutine must stop instead of writing the path. +func TestPeeringHandlerStartsNoWriteAfterTheStopDeadline(t *testing.T) { + t.Parallel() + + // The timer fires many times while Stop waits for its deadline. + const processInterval = 10 * time.Millisecond + + store := newBlockingStore() + h := newPeeringHandler(store, logger.New(), processInterval) + stopWhileAWriteIsBlocked(t, store, h, h.done, + []*ristypes.RISMessage{ + {Path: ristypes.ASPath{Sequence: []int{testASNA, testASNB}}}, + }, + &ristypes.RISMessage{ + Path: ristypes.ASPath{Sequence: []int{testASNC, testASND}}, + }) + + store.mockStore.mu.Lock() + defer store.mockStore.mu.Unlock() + + if store.Peerings[fmt.Sprintf("%d_%d", testASNC, testASND)] { + t.Errorf("the peering of AS%d and AS%d was written after the stop "+ + "deadline", testASNC, testASND) + } +} + // prefixAnnouncement returns a message in which testPeerIP announces prefixes // with origin AS testASNA. func prefixAnnouncement(prefixes []string) *ristypes.RISMessage { @@ -549,7 +630,7 @@ func TestASHandlerHandlesMessagesWhileAWriteIsBlocked(t *testing.T) { t.Parallel() store := newBlockingStore() - h := NewASHandler(store, logger.New()) + h := newASHandler(store, logger.New(), longerThanAnyTest) fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)} handleWhileAWriteIsBlocked(t, store, h, []*ristypes.RISMessage{{Path: fill}}, @@ -570,7 +651,7 @@ func TestPeerHandlerHandlesMessagesWhileAWriteIsBlocked(t *testing.T) { const otherPeerIP = "2001:db8:ffff::2" store := newBlockingStore() - h := NewPeerHandler(store, logger.New()) + h := newPeerHandler(store, logger.New(), longerThanAnyTest) handleWhileAWriteIsBlocked(t, store, h, slices.Repeat([]*ristypes.RISMessage{{Peer: testPeerIP}}, peerBatchSize), &ristypes.RISMessage{Peer: otherPeerIP}) @@ -611,7 +692,7 @@ func TestASHandlerWaitsAtTheMaximumBatchSize(t *testing.T) { t.Parallel() store := newBlockingStore() - h := NewASHandler(store, logger.New()) + h := newASHandler(store, logger.New(), longerThanAnyTest) fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)} toMax := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchMaxSize)} handleAtMaxBatchSize(t, store, h, @@ -628,13 +709,17 @@ func TestASHandlerWaitsAtTheMaximumBatchSize(t *testing.T) { } } +// TestPeerHandlerWaitsAtTheMaximumBatchSize must not have its flush timer fire. +// A timer flush during the fill would take only part of it. The batch would +// then reach its maximum size before the test's last message, and the test +// would hang. func TestPeerHandlerWaitsAtTheMaximumBatchSize(t *testing.T) { t.Parallel() const otherPeerIP = "2001:db8:ffff::2" store := newBlockingStore() - h := NewPeerHandler(store, logger.New()) + h := newPeerHandler(store, logger.New(), longerThanAnyTest) message := []*ristypes.RISMessage{{Peer: testPeerIP}} handleAtMaxBatchSize(t, store, h, slices.Repeat(message, peerBatchSize), diff --git a/internal/routewatch/peerhandler.go b/internal/routewatch/peerhandler.go index 6fb3278..5800474 100644 --- a/internal/routewatch/peerhandler.go +++ b/internal/routewatch/peerhandler.go @@ -56,7 +56,6 @@ type PeerHandler struct { // is &mu. batchTaken sync.Cond batchFullCh chan struct{} // wakes the flush goroutine when peerBatch is full - lastFlush time.Time // used only by the flush goroutine // 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. stopCh chan context.Context @@ -75,19 +74,27 @@ type peerUpdate struct { // periodically flushes accumulated peer updates to the database. The handler // should be stopped by calling Stop when it is no longer needed. func NewPeerHandler(db database.Store, logger *logger.Logger) *PeerHandler { + return newPeerHandler(db, logger, peerBatchTimeout) +} + +// newPeerHandler is NewPeerHandler with the flush timer's interval, +// batchTimeout, in place of peerBatchTimeout. A test that the timer must not +// disturb gives an interval longer than the test runs. +func newPeerHandler( + db database.Store, logger *logger.Logger, batchTimeout time.Duration, +) *PeerHandler { h := &PeerHandler{ db: db, logger: logger, peerBatch: make([]peerUpdate, 0, peerBatchSize), batchFullCh: make(chan struct{}, 1), - lastFlush: time.Now(), stopCh: make(chan context.Context, 1), done: make(chan struct{}), } h.batchTaken.L = &h.mu // Start the flush timer goroutine - go h.flushLoop() + go flushLoop(batchTimeout, h.batchFullCh, h.stopCh, h.done, h.flushBatch) return h } @@ -173,33 +180,6 @@ func (h *PeerHandler) Stop(ctx context.Context) { } } -// 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 close(h.done) - - ticker := time.NewTicker(peerBatchTimeout) - defer ticker.Stop() - - for { - select { - case <-ticker.C: - if time.Since(h.lastFlush) >= peerBatchTimeout { - h.flushBatch() - } - case <-h.batchFullCh: - h.flushBatch() - case ctx := <-h.stopCh: - // Final flush, unless the stop deadline has passed - if ctx.Err() == nil { - h.flushBatch() - } - - return - } - } -} - // takeBatch returns the pending updates and leaves an empty batch in their // place for HandleMessage to fill. They count in writing until flushBatch has // written them. @@ -220,11 +200,11 @@ func (h *PeerHandler) takeBatch() []peerUpdate { } // flushBatch takes the pending batch and writes it to the database, without -// holding the mutex during the write. -func (h *PeerHandler) flushBatch() { +// holding the mutex during the write. It reports whether there was a batch. +func (h *PeerHandler) flushBatch() bool { batch := h.takeBatch() if len(batch) == 0 { - return + return false } // The earliest and the latest time the batch saw each peer, with the peer @@ -255,5 +235,5 @@ func (h *PeerHandler) flushBatch() { h.writing = 0 h.mu.Unlock() - h.lastFlush = time.Now() + return true } diff --git a/internal/routewatch/peeringhandler.go b/internal/routewatch/peeringhandler.go index 8e5f801..046a210 100644 --- a/internal/routewatch/peeringhandler.go +++ b/internal/routewatch/peeringhandler.go @@ -62,6 +62,15 @@ type PeeringHandler struct { // periodically processes accumulated AS paths into peering records. The // handler begins processing immediately upon creation. func NewPeeringHandler(db database.Store, logger *logger.Logger) *PeeringHandler { + return newPeeringHandler(db, logger, peeringProcessInterval) +} + +// newPeeringHandler is NewPeeringHandler with the processing timer's interval, +// processInterval, in place of peeringProcessInterval. A test gives a short +// one to have the timer start a write. +func newPeeringHandler( + db database.Store, logger *logger.Logger, processInterval time.Duration, +) *PeeringHandler { h := &PeeringHandler{ db: db, logger: logger, @@ -71,7 +80,7 @@ func NewPeeringHandler(db database.Store, logger *logger.Logger) *PeeringHandler } // Start the periodic processing goroutine - go h.processLoop() + go h.processLoop(processInterval) return h } @@ -163,15 +172,29 @@ func (h *PeeringHandler) Stop(ctx context.Context) { } } -// processLoop runs periodically to process AS paths into peerings, and once +// processLoop processes AS paths into peerings every processInterval, and once // more on Stop. -func (h *PeeringHandler) processLoop() { +func (h *PeeringHandler) processLoop(processInterval time.Duration) { defer close(h.done) - ticker := time.NewTicker(peeringProcessInterval) + ticker := time.NewTicker(processInterval) defer ticker.Stop() for { + // Stop goes first. A write can run past the stop deadline while a + // tick waits beside Stop's context. The select below picks one of + // them at random, but past the deadline no write may start. + select { + case ctx := <-h.stopCh: + // Final processing, unless the stop deadline has passed + if ctx.Err() == nil { + h.processPeerings() + } + + return + default: + } + select { case <-ticker.C: h.processPeerings() diff --git a/internal/routewatch/prefixhandler.go b/internal/routewatch/prefixhandler.go index e02f424..6835699 100644 --- a/internal/routewatch/prefixhandler.go +++ b/internal/routewatch/prefixhandler.go @@ -96,7 +96,6 @@ type PrefixHandler struct { // &mu. batchTaken sync.Cond batchFullCh chan struct{} // wakes the flush goroutine when batch is full - lastFlush time.Time // used only by the flush goroutine // 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. stopCh chan context.Context @@ -147,14 +146,13 @@ func newPrefixHandler( table: table, batch: make([]prefixUpdate, 0, prefixBatchSize), batchFullCh: make(chan struct{}, 1), - lastFlush: time.Now(), stopCh: make(chan context.Context, 1), done: make(chan struct{}), } h.batchTaken.L = &h.mu // Start the flush timer goroutine - go h.flushLoop(batchTimeout) + go flushLoop(batchTimeout, h.batchFullCh, h.stopCh, h.done, h.flushBatch) return h } @@ -324,48 +322,6 @@ func (h *PrefixHandler) peerSessionDown(peer string) { } } -// flushLoop runs in a goroutine, the only one that writes the batch: it -// flushes when the batch is full, every batchTimeout, and once more on Stop. -func (h *PrefixHandler) flushLoop(batchTimeout time.Duration) { - defer close(h.done) - - ticker := time.NewTicker(batchTimeout) - defer ticker.Stop() - - for { - // Stop goes first. A write can run past the stop deadline while a full - // batch's wake-up or a tick waits beside Stop's context. The select - // below picks one of them at random, but past the deadline no flush - // may start. - select { - case ctx := <-h.stopCh: - // Final flush, unless the stop deadline has passed - if ctx.Err() == nil { - h.flushBatch() - } - - return - default: - } - - select { - case <-ticker.C: - if time.Since(h.lastFlush) >= batchTimeout { - h.flushBatch() - } - case <-h.batchFullCh: - h.flushBatch() - case ctx := <-h.stopCh: - // Final flush, unless the stop deadline has passed - if ctx.Err() == nil { - h.flushBatch() - } - - return - } - } -} - // takeBatch returns the pending updates, the held ones first, and the peers in // peersDown, and leaves an empty batch, no held updates and no peers in their // place for HandleMessage to fill. The updates count in writing until @@ -396,12 +352,13 @@ func (h *PrefixHandler) takeBatch() ([]prefixUpdate, []string) { // flushBatch writes the routing table's changes, then takes the pending batch // and writes it to the database, without holding the mutex during the writes. -func (h *PrefixHandler) flushBatch() { +// It reports whether there was a batch, or a peer down to delete the routes of. +func (h *PrefixHandler) flushBatch() bool { h.writeTableChanges() batch, peersDown := h.takeBatch() if len(batch) == 0 && len(peersDown) == 0 { - return + return false } startTime := time.Now() @@ -485,7 +442,7 @@ func (h *PrefixHandler) flushBatch() { "duration_ms", elapsed.Milliseconds(), ) - h.lastFlush = time.Now() + return true } // applyToTable applies update to the routing table in memory: as an diff --git a/internal/routewatch/prefixhandler_internal_test.go b/internal/routewatch/prefixhandler_internal_test.go index 53c4fb6..2e95267 100644 --- a/internal/routewatch/prefixhandler_internal_test.go +++ b/internal/routewatch/prefixhandler_internal_test.go @@ -981,8 +981,8 @@ func TestPrefixHandlerStopLogsNothingAfterADeleteThatSucceeds(t *testing.T) { } // longerThanAnyTest is a flush timer interval that no test lasts. A test whose -// prefix handler must flush only when its batch is full and on Stop gives it to -// newPrefixHandler. +// prefix, AS or peer handler must flush only when its batch is full and on Stop +// gives it to newPrefixHandler, newASHandler or newPeerHandler. const longerThanAnyTest = time.Hour // prefixHandlerWithoutTimer returns a prefix handler that writes to store and