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..d6c369a 100644 --- a/internal/routewatch/batchwrite_internal_test.go +++ b/internal/routewatch/batchwrite_internal_test.go @@ -351,6 +351,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..d68d593 100644 --- a/internal/routewatch/prefixhandler.go +++ b/internal/routewatch/prefixhandler.go @@ -323,6 +323,21 @@ func (h *PrefixHandler) flushLoop() { 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 {