diff --git a/TODO.md b/TODO.md index f805dcb..7e6de2c 100644 --- a/TODO.md +++ b/TODO.md @@ -33,6 +33,9 @@ tested. # Completed Steps +- 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) - 2026-10-27: the prefix handler applies each update to the routing table in memory, and writes the table's changes to the database every second. A write takes at most 20,000 changed pairs, and the rest follow in the next writes. A diff --git a/internal/routewatch/batchwrite_internal_test.go b/internal/routewatch/batchwrite_internal_test.go index 42c16bc..c14ac79 100644 --- a/internal/routewatch/batchwrite_internal_test.go +++ b/internal/routewatch/batchwrite_internal_test.go @@ -317,7 +317,9 @@ func stopWhileAWriteIsBlocked( // TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline also checks the // routing table's changes: the flush whose delete is held wrote those of the -// fill, so Stop must log the one of the extra prefix as not written. +// fill, so Stop must log the one of the extra prefix as not written. The flush +// timer must not fire. A flush it started before the fill would be the one +// held, and it would not have written the fill's change. func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) { t.Parallel() @@ -326,8 +328,8 @@ func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) { var logs bytes.Buffer store := newBlockingStore() - h := NewPrefixHandler(store, newErrorLogger(&logs), - routetable.New(routetable.Params{})) + h := newPrefixHandler(store, newErrorLogger(&logs), + routetable.New(routetable.Params{}), longerThanAnyTest) fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize) // The peer's session goes down first, so the write held is the delete of // its routes, and Stop must log that peer down too. @@ -351,6 +353,31 @@ func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) { } } +// TestPrefixHandlerSkipsAFullBatchAfterTheStopDeadline hands the prefix 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 TestPrefixHandlerSkipsAFullBatchAfterTheStopDeadline(t *testing.T) { + t.Parallel() + + const extraPrefix = "198.51.100.0/24" + + store := newBlockingStore() + h := NewPrefixHandler(store, logger.New(), routetable.New(routetable.Params{})) + fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize) + full := slices.Repeat([]string{extraPrefix}, prefixBatchSize) + stopWhileAWriteIsBlocked(t, store, h, h.done, + []*ristypes.RISMessage{prefixAnnouncement(fill)}, + prefixAnnouncement(full)) + + store.mu.Lock() + defer store.mu.Unlock() + + if store.prefixes[extraPrefix] { + t.Errorf("%s was written after the stop deadline", extraPrefix) + } +} + // tableWriteBlockingStore is a blockingStore whose writes of the routing // table's changes also wait until the test calls releaseWrites. type tableWriteBlockingStore struct { diff --git a/internal/routewatch/prefixhandler.go b/internal/routewatch/prefixhandler.go index c2764cf..e02f424 100644 --- a/internal/routewatch/prefixhandler.go +++ b/internal/routewatch/prefixhandler.go @@ -130,6 +130,16 @@ func (u prefixUpdate) announcesRoute() bool { // the updates to table, the routing table read from db at startup. func NewPrefixHandler( db database.Store, logger *logger.Logger, table *routetable.Table, +) *PrefixHandler { + return newPrefixHandler(db, logger, table, prefixBatchTimeout) +} + +// newPrefixHandler is NewPrefixHandler with the flush timer's interval, +// batchTimeout, in place of prefixBatchTimeout. A test that the timer must not +// disturb gives an interval longer than the test runs. +func newPrefixHandler( + db database.Store, logger *logger.Logger, table *routetable.Table, + batchTimeout time.Duration, ) *PrefixHandler { h := &PrefixHandler{ db: db, @@ -144,7 +154,7 @@ func NewPrefixHandler( h.batchTaken.L = &h.mu // Start the flush timer goroutine - go h.flushLoop() + go h.flushLoop(batchTimeout) return h } @@ -315,17 +325,32 @@ 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, periodically, and once more on Stop. -func (h *PrefixHandler) flushLoop() { +// 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(prefixBatchTimeout) + 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) >= prefixBatchTimeout { + if time.Since(h.lastFlush) >= batchTimeout { h.flushBatch() } case <-h.batchFullCh: diff --git a/internal/routewatch/prefixhandler_internal_test.go b/internal/routewatch/prefixhandler_internal_test.go index 45fb3b1..53c4fb6 100644 --- a/internal/routewatch/prefixhandler_internal_test.go +++ b/internal/routewatch/prefixhandler_internal_test.go @@ -980,6 +980,11 @@ 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. +const longerThanAnyTest = time.Hour + // prefixHandlerWithoutTimer returns a prefix handler that writes to store and // applies the updates to an empty routing table. It is built without // NewPrefixHandler's flush goroutine, so the test flushes itself. diff --git a/internal/routewatch/prefixhandler_table_internal_test.go b/internal/routewatch/prefixhandler_table_internal_test.go index 4b64db3..dc8a7cd 100644 --- a/internal/routewatch/prefixhandler_table_internal_test.go +++ b/internal/routewatch/prefixhandler_table_internal_test.go @@ -341,15 +341,14 @@ func TestPrefixHandlerWritesManyTableChangesOverSeveralFlushes(t *testing.T) { } // TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten stops a prefix handler, -// with no deadline, whose routing table holds more changed pairs than two -// writes take. Its last flush writes one write's worth, so Stop must log the -// pairs that no write took. +// with no deadline, whose routing table holds more changed pairs than one write +// takes. Its last flush writes one write's worth, so Stop must log the pairs +// that no write took. The flush timer must not fire, or its writes could leave +// no pairs for Stop to log. func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) { t.Parallel() - // More than two writes take, so that pairs are left even when the flush - // timer writes once before Stop. - const pairs = 2*maxTableChangesPerWrite + 1 + const pairs = maxTableChangesPerWrite + 1 var logs bytes.Buffer @@ -357,7 +356,7 @@ func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) { table := routetable.New(routetable.Params{}) announceManyPrefixes(table, pairs) - h := NewPrefixHandler(store, newErrorLogger(&logs), table) + h := newPrefixHandler(store, newErrorLogger(&logs), table, longerThanAnyTest) h.Stop(t.Context()) written := 0