1 Commits
Author SHA1 Message Date
clawbot 44c2ce9261 Wait up to 30 seconds for each step the routewatch tests expect (closes #213)
check / check (push) Canceled after 0s
Five prefix handler tests gave HandleMessage two seconds to apply a full
batch of 25,000 prefixes. Under the race detector on a loaded host that
takes longer, so they failed although nothing waited. Among them were
TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline,
TestPrefixHandlerSkipsAFullBatchAfterTheStopDeadline and
TestPrefixHandlerHandlesMessagesWhileAWriteIsBlocked.

handleMessageTimeout is now 30 seconds, and every bound a test in the
package puts on a step it expects uses it. Only a step that never comes
reaches it. A test that hangs still fails with its own message before go
test's 90-second timeout.

Model: opus-5-5
2026-10-27 22:22:01 +00:00
14 changed files with 134 additions and 230 deletions
+1 -3
View File
@@ -38,9 +38,7 @@ tested.
- 2026-10-27: the tests of `internal/routewatch` no longer fail on a loaded - 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 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 batch, which under the race detector can take longer. A test now waits up to
15 seconds for each step it expects. The tests of the handlers' limits give 30 seconds for each step it expects (closes #213)
the handlers small ones, so the package no longer runs past its 90-second
limit (closes #213, #215)
- 2026-10-27: the server works out the home page's cards every minute until it - 2026-10-27: the server works out the home page's cards every minute until it
stops. It used to work them out only once, at startup: the end of the context stops. It used to work them out only once, at startup: the end of the context
that `Start` is given stopped the refreshes (closes #211) that `Start` is given stopped the refreshes (closes #211)
@@ -554,9 +554,9 @@ func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) {
for range cap(linesSeen) { for range cap(linesSeen) {
select { select {
case <-linesSeen: case <-linesSeen:
case <-time.After(stepTimeout): case <-time.After(handleMessageTimeout):
t.Fatalf("the streamer did not read %d lines within %s", t.Fatalf("the streamer did not read %d lines within %s",
cap(linesSeen), stepTimeout) cap(linesSeen), handleMessageTimeout)
} }
} }
@@ -569,8 +569,8 @@ func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) {
select { select {
case <-shutDown: case <-shutDown:
case <-time.After(stepTimeout): case <-time.After(handleMessageTimeout):
t.Fatalf("Shutdown did not return within %s", stepTimeout) t.Fatalf("Shutdown did not return within %s", handleMessageTimeout)
} }
store.mu.Lock() store.mu.Lock()
@@ -607,13 +607,13 @@ func freePort(t *testing.T) int {
return addr.Port return addr.Port
} }
// TestShutdownDoesNotWaitForABlockedWrite starts the HTTP server and has the // TestShutdownReturnsByItsDeadlineWhileAWriteIsBlocked starts the HTTP server
// AS handler start a write that the store holds until the test ends, past // and has the AS handler start a write that the store holds until the test
// Shutdown's deadline. Shutdown must return without waiting for the write, with // ends, past Shutdown's deadline. Shutdown must return without waiting for the
// the HTTP server stopped, and must neither checkpoint nor close the database: // write, with the HTTP server stopped, and must neither checkpoint nor close
// the real one would wait for the write. // the database: the real one would wait for the write.
// Before the fix Shutdown waited for the write. // Before the fix Shutdown waited for the write.
func TestShutdownDoesNotWaitForABlockedWrite(t *testing.T) { func TestShutdownReturnsByItsDeadlineWhileAWriteIsBlocked(t *testing.T) {
t.Parallel() t.Parallel()
const deadline = 100 * time.Millisecond const deadline = 100 * time.Millisecond
@@ -660,9 +660,9 @@ func TestShutdownDoesNotWaitForABlockedWrite(t *testing.T) {
select { select {
case <-shutDown: case <-shutDown:
case <-time.After(deadline + stepTimeout): case <-time.After(deadline + handleMessageTimeout):
t.Fatalf("Shutdown did not return within %s of its deadline while "+ t.Fatalf("Shutdown did not return within %s of its deadline while "+
"a write was blocked", stepTimeout) "a write was blocked", handleMessageTimeout)
} }
var dialer net.Dialer var dialer net.Dialer
@@ -754,7 +754,7 @@ func TestShutdownCheckpointsAndClosesTheDatabaseLast(t *testing.T) {
Path: ristypes.ASPath{Sequence: []int{testASNA}}, Path: ristypes.ASPath{Sequence: []int{testASNA}},
}) })
ctx, cancel := context.WithTimeout(t.Context(), stepTimeout) ctx, cancel := context.WithTimeout(t.Context(), handleMessageTimeout)
defer cancel() defer cancel()
rw.Shutdown(ctx) rw.Shutdown(ctx)
+9 -14
View File
@@ -40,13 +40,9 @@ type ASHandler struct {
db database.Store db database.Store
logger *logger.Logger 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 // Batching. mu guards batch and writing only: the flush goroutine takes
// the batch under mu and writes it after releasing mu, so HandleMessage // the batch under mu and writes it after releasing mu, so HandleMessage
// waits for a database write only while batch is at batchMaxSize. // waits for a database write only while batch is at asnBatchMaxSize.
mu sync.Mutex mu sync.Mutex
batch []asnOp batch []asnOp
// writing is how many operations the flush goroutine took from batch // writing is how many operations the flush goroutine took from batch
@@ -83,13 +79,12 @@ func newASHandler(
db database.Store, logger *logger.Logger, batchTimeout time.Duration, db database.Store, logger *logger.Logger, batchTimeout time.Duration,
) *ASHandler { ) *ASHandler {
h := &ASHandler{ h := &ASHandler{
db: db, db: db,
logger: logger, logger: logger,
batchMaxSize: asnBatchMaxSize, batch: make([]asnOp, 0, asnBatchSize),
batch: make([]asnOp, 0, asnBatchSize), batchFullCh: make(chan struct{}, 1),
batchFullCh: make(chan struct{}, 1), stopCh: make(chan context.Context, 1),
stopCh: make(chan context.Context, 1), done: make(chan struct{}),
done: make(chan struct{}),
} }
h.batchTaken.L = &h.mu h.batchTaken.L = &h.mu
@@ -132,7 +127,7 @@ func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) {
// While the batch is at its maximum size, wait until the flush goroutine // While the batch is at its maximum size, wait until the flush goroutine
// takes it. Meanwhile the handler takes nothing off its queue, which // takes it. Meanwhile the handler takes nothing off its queue, which
// fills, and the streamer drops messages for it. // fills, and the streamer drops messages for it.
for len(h.batch) >= h.batchMaxSize { for len(h.batch) >= asnBatchMaxSize {
h.batchTaken.Wait() h.batchTaken.Wait()
} }
@@ -156,7 +151,7 @@ func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) {
// A full batch wakes the flush goroutine. If it is writing, the wake-up // 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 // waits in batchFullCh and the batch keeps growing until the write ends
// or the batch reaches batchMaxSize. // or the batch reaches asnBatchMaxSize.
if len(h.batch) >= asnBatchSize { if len(h.batch) >= asnBatchSize {
select { select {
case h.batchFullCh <- struct{}{}: case h.batchFullCh <- struct{}{}:
@@ -19,7 +19,7 @@ func TestASHandlerStoresTheEarliestAndLatestTimesOfANewAS(t *testing.T) {
db := newTestDatabase(t) db := newTestDatabase(t)
// Built without NewASHandler's flush goroutine, so the test flushes itself. // 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()}
first := peerAnnouncement(testPeerIP, testV4Prefix) first := peerAnnouncement(testPeerIP, testV4Prefix)
second := peerAnnouncement(testPeerIP, testV4Prefix) second := peerAnnouncement(testPeerIP, testV4Prefix)
@@ -54,7 +54,7 @@ func TestASHandlerRecordsNothingFromAPathWithASZero(t *testing.T) {
store := newMockStore() store := newMockStore()
// Built without NewASHandler's flush goroutine, so the test flushes itself. // 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()}
msg := peerAnnouncement(testPeerIP, testV4Prefix) msg := peerAnnouncement(testPeerIP, testV4Prefix)
msg.Path = decodePath(t, testASNA, 0, testASNB) msg.Path = decodePath(t, testASNA, 0, testASNB)
@@ -224,7 +224,7 @@ func TestASNFetcherStopReturnsByItsDeadline(t *testing.T) {
select { select {
case <-lookupStarted: case <-lookupStarted:
case <-time.After(stepTimeout): case <-time.After(handleMessageTimeout):
t.Fatal("the fetcher did not start the lookup") t.Fatal("the fetcher did not start the lookup")
} }
@@ -240,22 +240,22 @@ func TestASNFetcherStopReturnsByItsDeadline(t *testing.T) {
select { select {
case <-f.done: case <-f.done:
case <-time.After(stepTimeout): case <-time.After(handleMessageTimeout):
t.Fatal("the fetcher goroutine did not return after the lookup ended") t.Fatal("the fetcher goroutine did not return after the lookup ended")
} }
} }
// TestRecordKeepsOnlyTheStatsWindow records a success and a failure every // TestRecordKeepsOnlyTheStatsWindow records a success and a failure every
// second, by a clock the test sets, for three times statsWindow, and never // second, by a clock the test sets, far longer than statsWindow, and never
// asks for the stats. The fetcher must keep only the times within the last // asks for the stats. The fetcher must keep only the times within the last
// statsWindow. // statsWindow.
func TestRecordKeepsOnlyTheStatsWindow(t *testing.T) { func TestRecordKeepsOnlyTheStatsWindow(t *testing.T) {
t.Parallel() t.Parallel()
const ( const (
records = 100_000
step = time.Second step = time.Second
want = int(statsWindow / step) want = int(statsWindow / step)
records = 3 * want
) )
clock := time.Now() clock := time.Now()
+31 -67
View File
@@ -19,15 +19,13 @@ import (
) )
const ( const (
// stepTimeout is how long a test waits for a step it expects, such as // handleMessageTimeout is how long a test waits for a step it expects,
// HandleMessage returning, a write starting or Shutdown returning, before // such as HandleMessage returning or a write starting, before it fails.
// it fails. Only a step that never comes should reach it. On a loaded // Only a step that never comes should reach it. On a loaded host, under
// host, under the race detector, HandleMessage takes seconds to apply a // the race detector, HandleMessage takes seconds to apply a full prefix
// full prefix batch. The helpers below bound every wait with it, Stop // batch. A test that hangs still fails with its own message before go
// included. A test that hangs in one of them waits out at most two of // test's 90-second timeout.
// these bounds, its step's and Stop's, and then fails with its own handleMessageTimeout = 30 * time.Second
// message. That leaves time before go test's 90-second timeout.
stepTimeout = 15 * time.Second
// stillWaitingAfter is how long a test watches a HandleMessage that must // stillWaitingAfter is how long a test watches a HandleMessage that must
// wait. One that does not wait returns well within it. // wait. One that does not wait returns well within it.
@@ -204,7 +202,7 @@ func lastLogged(t *testing.T, logs *bytes.Buffer) stopLog {
return logged return logged
} }
// returnsInTime reports whether f returns within stepTimeout. // returnsInTime reports whether f returns within handleMessageTimeout.
func returnsInTime(f func()) bool { func returnsInTime(f func()) bool {
done := make(chan struct{}) done := make(chan struct{})
@@ -216,22 +214,11 @@ func returnsInTime(f func()) bool {
select { select {
case <-done: case <-done:
return true return true
case <-time.After(stepTimeout): case <-time.After(handleMessageTimeout):
return false return false
} }
} }
// stopInTime stops h, with no deadline, and fails the test if Stop does not
// return within stepTimeout. It only marks the test failed, so it can be
// deferred.
func stopInTime(t *testing.T, h interface{ Stop(ctx context.Context) }) {
t.Helper()
if !returnsInTime(func() { h.Stop(t.Context()) }) {
t.Errorf("Stop did not return within %s", stepTimeout)
}
}
// startBlockedWrite hands h the messages in fill and waits until h starts to // 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 // 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 // their batch, and the peering handler on its timer. store holds the write
@@ -257,7 +244,7 @@ func startBlockedWrite(
select { select {
case <-store.writeStarted: case <-store.writeStarted:
case <-time.After(stepTimeout): case <-time.After(handleMessageTimeout):
t.Fatal("the handler did not start to write the messages") t.Fatal("the handler did not start to write the messages")
} }
} }
@@ -278,7 +265,7 @@ func handleWhileAWriteIsBlocked(
) { ) {
t.Helper() t.Helper()
defer stopInTime(t, h) defer h.Stop(t.Context())
defer store.releaseWrites() defer store.releaseWrites()
startBlockedWrite(t, store, h, fill) startBlockedWrite(t, store, h, fill)
@@ -292,9 +279,7 @@ func handleWhileAWriteIsBlocked(
// while store holds the write, hands h the messages in toMax, which bring its // while store holds the write, hands h the messages in toMax, which bring its
// next batch to its maximum size, and the message extra. HandleMessage for // next batch to its maximum size, and the message extra. HandleMessage for
// extra must wait until the test lets the write finish, and return after. Then // extra must wait until the test lets the write finish, and return after. Then
// it stops h, which writes what it holds. The tests lower h's maximum batch // it stops h, which writes what it holds.
// size to one batch, so toMax is one batch: the real maximum of ten batches is
// slow to reach under the race detector.
func handleAtMaxBatchSize( func handleAtMaxBatchSize(
t *testing.T, t *testing.T,
store *blockingStore, store *blockingStore,
@@ -307,18 +292,13 @@ func handleAtMaxBatchSize(
) { ) {
t.Helper() t.Helper()
defer stopInTime(t, h) defer h.Stop(t.Context())
defer store.releaseWrites() defer store.releaseWrites()
startBlockedWrite(t, store, h, fill) startBlockedWrite(t, store, h, fill)
if !returnsInTime(func() { for _, msg := range toMax {
for _, msg := range toMax { h.HandleMessage(msg)
h.HandleMessage(msg)
}
}) {
t.Fatal("HandleMessage did not return while the batch grew to its " +
"maximum size")
} }
returned := make(chan struct{}) returned := make(chan struct{})
@@ -339,7 +319,7 @@ func handleAtMaxBatchSize(
select { select {
case <-returned: case <-returned:
case <-time.After(stepTimeout): case <-time.After(handleMessageTimeout):
t.Fatal("HandleMessage did not return after the database write finished") t.Fatal("HandleMessage did not return after the database write finished")
} }
} }
@@ -369,12 +349,8 @@ func stopWhileAWriteIsBlocked(
startBlockedWrite(t, store, h, fill) startBlockedWrite(t, store, h, fill)
if !returnsInTime(func() { for _, msg := range extra {
for _, msg := range extra { h.HandleMessage(msg)
h.HandleMessage(msg)
}
}) {
t.Fatal("HandleMessage did not return while a database write was blocked")
} }
ctx, cancel := context.WithTimeout(t.Context(), stopDeadline) ctx, cancel := context.WithTimeout(t.Context(), stopDeadline)
@@ -389,7 +365,7 @@ func stopWhileAWriteIsBlocked(
select { select {
case <-done: case <-done:
case <-time.After(stepTimeout): case <-time.After(handleMessageTimeout):
t.Fatal("the flush goroutine did not return after the write finished") t.Fatal("the flush goroutine did not return after the write finished")
} }
} }
@@ -408,8 +384,7 @@ func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
store := newBlockingStore() store := newBlockingStore()
h := newPrefixHandler(store, newErrorLogger(&logs), h := newPrefixHandler(store, newErrorLogger(&logs),
routetable.New(routetable.Params{}), longerThanAnyTest, routetable.New(routetable.Params{}), longerThanAnyTest)
maxTableChangesPerWrite)
fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize) fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
stopWhileAWriteIsBlocked(t, store, h, h.done, stopWhileAWriteIsBlocked(t, store, h, h.done,
[]*ristypes.RISMessage{prefixAnnouncement(fill)}, []*ristypes.RISMessage{prefixAnnouncement(fill)},
@@ -626,7 +601,7 @@ func TestPeeringHandlerStopCountsTheWriteInProgress(t *testing.T) {
select { select {
case <-h.done: case <-h.done:
case <-time.After(stepTimeout): case <-time.After(handleMessageTimeout):
t.Fatal("the processing goroutine did not return after the write " + t.Fatal("the processing goroutine did not return after the write " +
"finished") "finished")
} }
@@ -647,7 +622,7 @@ func TestPeeringHandlerStartsNoWriteAfterTheStopDeadline(t *testing.T) {
const processInterval = 10 * time.Millisecond const processInterval = 10 * time.Millisecond
store := newBlockingStore() store := newBlockingStore()
h := newPeeringHandler(store, logger.New(), processInterval, maxTrackedPaths) h := newPeeringHandler(store, logger.New(), processInterval)
stopWhileAWriteIsBlocked(t, store, h, h.done, stopWhileAWriteIsBlocked(t, store, h, h.done,
[]*ristypes.RISMessage{ []*ristypes.RISMessage{
{Path: ristypes.ASPath{Sequence: []int{testASNA, testASNB}}}, {Path: ristypes.ASPath{Sequence: []int{testASNA, testASNB}}},
@@ -743,11 +718,9 @@ func TestPrefixHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
const extraPrefix = "198.51.100.0/24" const extraPrefix = "198.51.100.0/24"
store := newBlockingStore() store := newBlockingStore()
h := newPrefixHandler(store, logger.New(), routetable.New(routetable.Params{}), h := NewPrefixHandler(store, logger.New(), routetable.New(routetable.Params{}))
longerThanAnyTest, maxTableChangesPerWrite)
h.batchMaxSize = prefixBatchSize
fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize) fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
toMax := slices.Repeat([]string{testV4Prefix}, prefixBatchSize) toMax := slices.Repeat([]string{testV4Prefix}, prefixBatchMaxSize)
handleAtMaxBatchSize(t, store, h, handleAtMaxBatchSize(t, store, h,
[]*ristypes.RISMessage{prefixAnnouncement(fill)}, []*ristypes.RISMessage{prefixAnnouncement(fill)},
[]*ristypes.RISMessage{prefixAnnouncement(toMax)}, []*ristypes.RISMessage{prefixAnnouncement(toMax)},
@@ -767,9 +740,8 @@ func TestASHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
store := newBlockingStore() store := newBlockingStore()
h := newASHandler(store, logger.New(), longerThanAnyTest) h := newASHandler(store, logger.New(), longerThanAnyTest)
h.batchMaxSize = asnBatchSize
fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)} fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
toMax := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)} toMax := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchMaxSize)}
handleAtMaxBatchSize(t, store, h, handleAtMaxBatchSize(t, store, h,
[]*ristypes.RISMessage{{Path: fill}}, []*ristypes.RISMessage{{Path: fill}},
[]*ristypes.RISMessage{{Path: toMax}}, []*ristypes.RISMessage{{Path: toMax}},
@@ -787,7 +759,7 @@ func TestASHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
// TestPeerHandlerWaitsAtTheMaximumBatchSize must not have its flush timer fire. // TestPeerHandlerWaitsAtTheMaximumBatchSize must not have its flush timer fire.
// A timer flush during the fill would take only part of it. The batch would // 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 // then reach its maximum size before the test's last message, and the test
// would fail. // would hang.
func TestPeerHandlerWaitsAtTheMaximumBatchSize(t *testing.T) { func TestPeerHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
t.Parallel() t.Parallel()
@@ -795,11 +767,10 @@ func TestPeerHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
store := newBlockingStore() store := newBlockingStore()
h := newPeerHandler(store, logger.New(), longerThanAnyTest) h := newPeerHandler(store, logger.New(), longerThanAnyTest)
h.batchMaxSize = peerBatchSize
message := []*ristypes.RISMessage{{Peer: testPeerIP}} message := []*ristypes.RISMessage{{Peer: testPeerIP}}
handleAtMaxBatchSize(t, store, h, handleAtMaxBatchSize(t, store, h,
slices.Repeat(message, peerBatchSize), slices.Repeat(message, peerBatchSize),
slices.Repeat(message, peerBatchSize), slices.Repeat(message, peerBatchMaxSize),
&ristypes.RISMessage{Peer: otherPeerIP}) &ristypes.RISMessage{Peer: otherPeerIP})
store.mu.Lock() store.mu.Lock()
@@ -829,9 +800,8 @@ func TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull(t *testing.T) {
batchSize: prefixBatchSize, batchSize: prefixBatchSize,
handle: func(n int) chan struct{} { handle: func(n int) chan struct{} {
h := &PrefixHandler{ h := &PrefixHandler{
batchMaxSize: prefixBatchMaxSize, batchFullCh: make(chan struct{}, 1),
batchFullCh: make(chan struct{}, 1), table: routetable.New(routetable.Params{}),
table: routetable.New(routetable.Params{}),
} }
h.HandleMessage(prefixAnnouncement( h.HandleMessage(prefixAnnouncement(
slices.Repeat([]string{testV4Prefix}, n))) slices.Repeat([]string{testV4Prefix}, n)))
@@ -843,10 +813,7 @@ func TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull(t *testing.T) {
name: "AS", name: "AS",
batchSize: asnBatchSize, batchSize: asnBatchSize,
handle: func(n int) chan struct{} { handle: func(n int) chan struct{} {
h := &ASHandler{ h := &ASHandler{batchFullCh: make(chan struct{}, 1)}
batchMaxSize: asnBatchMaxSize,
batchFullCh: make(chan struct{}, 1),
}
// The origin AS and each AS in the path are one update each. // The origin AS and each AS in the path are one update each.
h.HandleMessage(&ristypes.RISMessage{ h.HandleMessage(&ristypes.RISMessage{
Path: ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, n-1)}, Path: ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, n-1)},
@@ -859,10 +826,7 @@ func TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull(t *testing.T) {
name: "peer", name: "peer",
batchSize: peerBatchSize, batchSize: peerBatchSize,
handle: func(n int) chan struct{} { handle: func(n int) chan struct{} {
h := &PeerHandler{ h := &PeerHandler{batchFullCh: make(chan struct{}, 1)}
batchMaxSize: peerBatchMaxSize,
batchFullCh: make(chan struct{}, 1),
}
for range n { for range n {
h.HandleMessage(&ristypes.RISMessage{Peer: testPeerIP}) h.HandleMessage(&ristypes.RISMessage{Peer: testPeerIP})
} }
@@ -97,7 +97,7 @@ func TestDBMaintainerStopReturnsByItsDeadline(t *testing.T) {
select { select {
case <-store.started: case <-store.started:
case <-time.After(stepTimeout): case <-time.After(handleMessageTimeout):
t.Fatal("the maintainer did not start the checkpoint") t.Fatal("the maintainer did not start the checkpoint")
} }
@@ -113,7 +113,7 @@ func TestDBMaintainerStopReturnsByItsDeadline(t *testing.T) {
select { select {
case <-m.done: case <-m.done:
case <-time.After(stepTimeout): case <-time.After(handleMessageTimeout):
t.Fatal("the maintenance goroutine did not return after the " + t.Fatal("the maintenance goroutine did not return after the " +
"checkpoint ended") "checkpoint ended")
} }
+9 -14
View File
@@ -43,14 +43,10 @@ type PeerHandler struct {
db database.Store db database.Store
logger *logger.Logger logger *logger.Logger
// batchMaxSize is peerBatchMaxSize. Only HandleMessage reads it, so a test
// may lower it before it hands the handler a message.
batchMaxSize int
// Batching. mu guards peerBatch and writing only: the flush goroutine // Batching. mu guards peerBatch and writing only: the flush goroutine
// takes the batch under mu and writes it after releasing mu, so // takes the batch under mu and writes it after releasing mu, so
// HandleMessage waits for a database write only while peerBatch is at // HandleMessage waits for a database write only while peerBatch is at
// batchMaxSize. // peerBatchMaxSize.
mu sync.Mutex mu sync.Mutex
peerBatch []peerUpdate peerBatch []peerUpdate
// writing is how many updates the flush goroutine took from peerBatch // writing is how many updates the flush goroutine took from peerBatch
@@ -88,13 +84,12 @@ func newPeerHandler(
db database.Store, logger *logger.Logger, batchTimeout time.Duration, db database.Store, logger *logger.Logger, batchTimeout time.Duration,
) *PeerHandler { ) *PeerHandler {
h := &PeerHandler{ h := &PeerHandler{
db: db, db: db,
logger: logger, logger: logger,
batchMaxSize: peerBatchMaxSize, peerBatch: make([]peerUpdate, 0, peerBatchSize),
peerBatch: make([]peerUpdate, 0, peerBatchSize), batchFullCh: make(chan struct{}, 1),
batchFullCh: make(chan struct{}, 1), stopCh: make(chan context.Context, 1),
stopCh: make(chan context.Context, 1), done: make(chan struct{}),
done: make(chan struct{}),
} }
h.batchTaken.L = &h.mu h.batchTaken.L = &h.mu
@@ -140,7 +135,7 @@ func (h *PeerHandler) HandleMessage(msg *ristypes.RISMessage) {
// While the batch is at its maximum size, wait until the flush goroutine // While the batch is at its maximum size, wait until the flush goroutine
// takes it. Meanwhile the handler takes nothing off its queue, which // takes it. Meanwhile the handler takes nothing off its queue, which
// fills, and the streamer drops messages for it. // fills, and the streamer drops messages for it.
for len(h.peerBatch) >= h.batchMaxSize { for len(h.peerBatch) >= peerBatchMaxSize {
h.batchTaken.Wait() h.batchTaken.Wait()
} }
@@ -154,7 +149,7 @@ func (h *PeerHandler) HandleMessage(msg *ristypes.RISMessage) {
// A full batch wakes the flush goroutine. If it is writing, the wake-up // 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 // waits in batchFullCh and the batch keeps growing until the write ends
// or the batch reaches batchMaxSize. // or the batch reaches peerBatchMaxSize.
if len(h.peerBatch) >= peerBatchSize { if len(h.peerBatch) >= peerBatchSize {
select { select {
case h.batchFullCh <- struct{}{}: case h.batchFullCh <- struct{}{}:
@@ -35,7 +35,7 @@ func TestPeerHandlerWritesTheEarliestAndLatestTimesOfANewPeer(t *testing.T) {
// Built without NewPeerHandler's flush goroutine, so the test flushes // Built without NewPeerHandler's flush goroutine, so the test flushes
// itself. // itself.
h := &PeerHandler{db: store, logger: logger.New(), batchMaxSize: peerBatchMaxSize} h := &PeerHandler{db: store, logger: logger.New()}
first := peerAnnouncement(testPeerIP, testV4Prefix) first := peerAnnouncement(testPeerIP, testV4Prefix)
second := peerAnnouncement(testPeerIP, testV4Prefix) second := peerAnnouncement(testPeerIP, testV4Prefix)
+9 -16
View File
@@ -42,10 +42,6 @@ type PeeringHandler struct {
db database.Store db database.Store
logger *logger.Logger logger *logger.Logger
// maxPaths is the most paths asPaths holds: maxTrackedPaths, except in
// tests that give a smaller one.
maxPaths int
// In-memory AS path tracking: each path, JSON-encoded, with the earliest // In-memory AS path tracking: each path, JSON-encoded, with the earliest
// and the latest time an update carried it. // and the latest time an update carried it.
mu sync.Mutex mu sync.Mutex
@@ -66,24 +62,21 @@ type PeeringHandler struct {
// periodically processes accumulated AS paths into peering records. The // periodically processes accumulated AS paths into peering records. The
// handler begins processing immediately upon creation. // handler begins processing immediately upon creation.
func NewPeeringHandler(db database.Store, logger *logger.Logger) *PeeringHandler { func NewPeeringHandler(db database.Store, logger *logger.Logger) *PeeringHandler {
return newPeeringHandler(db, logger, peeringProcessInterval, maxTrackedPaths) return newPeeringHandler(db, logger, peeringProcessInterval)
} }
// newPeeringHandler is NewPeeringHandler with the processing timer's interval, // newPeeringHandler is NewPeeringHandler with the processing timer's interval,
// processInterval, in place of peeringProcessInterval, and the most paths it // processInterval, in place of peeringProcessInterval. A test gives a short
// holds, maxPaths, in place of maxTrackedPaths. A test gives a short interval to // one to have the timer start a write.
// have the timer start a write, and a small maxPaths to fill the map quickly.
func newPeeringHandler( func newPeeringHandler(
db database.Store, logger *logger.Logger, processInterval time.Duration, db database.Store, logger *logger.Logger, processInterval time.Duration,
maxPaths int,
) *PeeringHandler { ) *PeeringHandler {
h := &PeeringHandler{ h := &PeeringHandler{
db: db, db: db,
logger: logger, logger: logger,
asPaths: make(map[string]database.SeenTimes), asPaths: make(map[string]database.SeenTimes),
maxPaths: maxPaths, stopCh: make(chan context.Context, 1),
stopCh: make(chan context.Context, 1), done: make(chan struct{}),
done: make(chan struct{}),
} }
// Start the periodic processing goroutine // Start the periodic processing goroutine
@@ -138,7 +131,7 @@ func (h *PeeringHandler) HandleMessage(msg *ristypes.RISMessage) {
defer h.mu.Unlock() defer h.mu.Unlock()
seen, exists := h.asPaths[key] seen, exists := h.asPaths[key]
if !exists && len(h.asPaths) >= h.maxPaths { if !exists && len(h.asPaths) >= maxTrackedPaths {
// Map is full; drop this new path and count it. // Map is full; drop this new path and count it.
h.droppedPaths++ h.droppedPaths++
@@ -5,6 +5,7 @@ import (
"fmt" "fmt"
"maps" "maps"
"slices" "slices"
"strconv"
"sync" "sync"
"testing" "testing"
"time" "time"
@@ -19,10 +20,6 @@ const (
testASNB = 64501 testASNB = 64501
testASNC = 64502 testASNC = 64502
testASND = 64503 testASND = 64503
// testMaxTrackedPaths is a limit of the path map that a test fills
// quickly, in place of maxTrackedPaths.
testMaxTrackedPaths = 10
) )
// recordingStore wraps mockStore to count every RecordPeering call, so a // recordingStore wraps mockStore to count every RecordPeering call, so a
@@ -55,10 +52,9 @@ func (r *recordingStore) callCount() int {
// processing goroutine, so tests drive processing explicitly. // processing goroutine, so tests drive processing explicitly.
func newTestHandler(db database.Store) *PeeringHandler { func newTestHandler(db database.Store) *PeeringHandler {
return &PeeringHandler{ return &PeeringHandler{
db: db, db: db,
logger: logger.New(), logger: logger.New(),
asPaths: make(map[string]database.SeenTimes), asPaths: make(map[string]database.SeenTimes),
maxPaths: maxTrackedPaths,
} }
} }
@@ -248,28 +244,25 @@ func TestPeeringHandlerStopWritesTheRemainingPaths(t *testing.T) {
// TestPeeringHandlerCapDropsAndCounts verifies that a full map drops new paths // TestPeeringHandlerCapDropsAndCounts verifies that a full map drops new paths
// and counts them, while a path already tracked is refreshed rather than // and counts them, while a path already tracked is refreshed rather than
// dropped. The handler's limit is small, and its timer does not fire. // dropped.
func TestPeeringHandlerCapDropsAndCounts(t *testing.T) { func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
t.Parallel() t.Parallel()
store := &recordingStore{mockStore: newMockStore()} store := &recordingStore{mockStore: newMockStore()}
h := newPeeringHandler(store, logger.New(), longerThanAnyTest, h := newTestHandler(store)
testMaxTrackedPaths)
defer stopInTime(t, h)
now := time.Now().UTC() now := time.Now().UTC()
// Fill the map to exactly its limit, including one real path key so the // Fill the map to exactly maxTrackedPaths, including one real path key so
// "already tracked" branch can be exercised. The other paths are of one // the "already tracked" branch can be exercised. The filler keys are never
// AS, so they carry no peering. // processed in this test, so their contents do not matter.
existing := pathKey(t, testASNA, testASNB) existing := pathKey(t, testASNA, testASNB)
h.mu.Lock() h.mu.Lock()
h.asPaths[existing] = database.SeenTimes{First: now, Last: now} h.asPaths[existing] = database.SeenTimes{First: now, Last: now}
for asn := 1; len(h.asPaths) < testMaxTrackedPaths; asn++ { for i := 0; len(h.asPaths) < maxTrackedPaths; i++ {
h.asPaths[pathKey(t, asn)] = database.SeenTimes{First: now, Last: now} h.asPaths[strconv.Itoa(i)] = database.SeenTimes{First: now, Last: now}
} }
h.mu.Unlock() h.mu.Unlock()
@@ -277,9 +270,8 @@ func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
handle(h, now.Add(time.Second), testASNA, testASNC) handle(h, now.Add(time.Second), testASNA, testASNC)
tracked, dropped := snapshot(h) tracked, dropped := snapshot(h)
if tracked != testMaxTrackedPaths { if tracked != maxTrackedPaths {
t.Fatalf("want map size %d after drop, got %d", testMaxTrackedPaths, t.Fatalf("want map size %d after drop, got %d", maxTrackedPaths, tracked)
tracked)
} }
if dropped != 1 { if dropped != 1 {
@@ -291,9 +283,8 @@ func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
handle(h, refreshed, testASNA, testASNB) handle(h, refreshed, testASNA, testASNB)
tracked, dropped = snapshot(h) tracked, dropped = snapshot(h)
if tracked != testMaxTrackedPaths { if tracked != maxTrackedPaths {
t.Fatalf("want map size %d after refresh, got %d", testMaxTrackedPaths, t.Fatalf("want map size %d after refresh, got %d", maxTrackedPaths, tracked)
tracked)
} }
if dropped != 1 { if dropped != 1 {
@@ -312,21 +303,17 @@ func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
// TestPeeringHandlerRunResetsTheDroppedCount fills the map, has the handler // TestPeeringHandlerRunResetsTheDroppedCount fills the map, has the handler
// drop one path, and runs the processing. The count of dropped paths is the // drop one path, and runs the processing. The count of dropped paths is the
// run's own, so it is 0 again after the run. The handler's limit is small, and // run's own, so it is 0 again after the run.
// its timer does not fire.
func TestPeeringHandlerRunResetsTheDroppedCount(t *testing.T) { func TestPeeringHandlerRunResetsTheDroppedCount(t *testing.T) {
t.Parallel() t.Parallel()
h := newPeeringHandler(newMockStore(), logger.New(), longerThanAnyTest, h := newTestHandler(newMockStore())
testMaxTrackedPaths)
defer stopInTime(t, h)
now := time.Now().UTC() now := time.Now().UTC()
// Paths of one AS carry no peering, so the run records nothing. // Paths of one AS carry no peering, so the run records nothing.
h.mu.Lock() h.mu.Lock()
for asn := 1; len(h.asPaths) < testMaxTrackedPaths; asn++ { for asn := 1; len(h.asPaths) < maxTrackedPaths; asn++ {
h.asPaths[pathKey(t, asn)] = database.SeenTimes{First: now, Last: now} h.asPaths[pathKey(t, asn)] = database.SeenTimes{First: now, Last: now}
} }
h.mu.Unlock() h.mu.Unlock()
+19 -31
View File
@@ -60,17 +60,10 @@ type PrefixHandler struct {
metrics *metrics.Tracker metrics *metrics.Tracker
table *routetable.Table table *routetable.Table
// batchMaxSize is prefixBatchMaxSize. Only HandleMessage reads it, so a
// test may lower it before it hands the handler a message.
batchMaxSize int
// tableChangesPerWrite is maxTableChangesPerWrite, except in tests that
// give a smaller one.
tableChangesPerWrite int
// Batching. mu guards batch, writing and writingTableChanges only: the // Batching. mu guards batch, writing and writingTableChanges only: the
// flush goroutine takes them under mu and writes them after releasing // flush goroutine takes them under mu and writes them after releasing
// mu, so HandleMessage waits for a database write only while batch is at // mu, so HandleMessage waits for a database write only while batch is at
// batchMaxSize. // prefixBatchMaxSize.
mu sync.Mutex mu sync.Mutex
batch []seenPrefix batch []seenPrefix
// writing is how many updates the flush goroutine took and has not // writing is how many updates the flush goroutine took and has not
@@ -126,30 +119,24 @@ func (u prefixUpdate) announcesRoute() bool {
func NewPrefixHandler( func NewPrefixHandler(
db database.Store, logger *logger.Logger, table *routetable.Table, db database.Store, logger *logger.Logger, table *routetable.Table,
) *PrefixHandler { ) *PrefixHandler {
return newPrefixHandler(db, logger, table, prefixBatchTimeout, return newPrefixHandler(db, logger, table, prefixBatchTimeout)
maxTableChangesPerWrite)
} }
// newPrefixHandler is NewPrefixHandler with the flush timer's interval, // newPrefixHandler is NewPrefixHandler with the flush timer's interval,
// batchTimeout, in place of prefixBatchTimeout, and tableChangesPerWrite in // batchTimeout, in place of prefixBatchTimeout. A test that the timer must not
// place of maxTableChangesPerWrite. A test that the timer must not disturb // disturb gives an interval longer than the test runs.
// gives an interval longer than the test runs. A test of writes split over
// several flushes gives a small tableChangesPerWrite, so that it exceeds it
// with few changes.
func newPrefixHandler( func newPrefixHandler(
db database.Store, logger *logger.Logger, table *routetable.Table, db database.Store, logger *logger.Logger, table *routetable.Table,
batchTimeout time.Duration, tableChangesPerWrite int, batchTimeout time.Duration,
) *PrefixHandler { ) *PrefixHandler {
h := &PrefixHandler{ h := &PrefixHandler{
db: db, db: db,
logger: logger, logger: logger,
table: table, table: table,
batch: make([]seenPrefix, 0, prefixBatchSize), batch: make([]seenPrefix, 0, prefixBatchSize),
batchMaxSize: prefixBatchMaxSize, batchFullCh: make(chan struct{}, 1),
tableChangesPerWrite: tableChangesPerWrite, stopCh: make(chan context.Context, 1),
batchFullCh: make(chan struct{}, 1), done: make(chan struct{}),
stopCh: make(chan context.Context, 1),
done: make(chan struct{}),
} }
h.batchTaken.L = &h.mu h.batchTaken.L = &h.mu
@@ -204,7 +191,7 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
// While the batch is at its maximum size, wait until the flush goroutine // While the batch is at its maximum size, wait until the flush goroutine
// takes it. Meanwhile the handler takes nothing off its queue, which // takes it. Meanwhile the handler takes nothing off its queue, which
// fills, and the streamer drops messages for it. // fills, and the streamer drops messages for it.
for len(h.batch) >= h.batchMaxSize { for len(h.batch) >= prefixBatchMaxSize {
h.batchTaken.Wait() h.batchTaken.Wait()
} }
@@ -252,7 +239,7 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
// A full batch wakes the flush goroutine. If it is writing, the wake-up // 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 // waits in batchFullCh and the batch keeps growing until the write ends
// or the batch reaches batchMaxSize. // or the batch reaches prefixBatchMaxSize.
if len(h.batch) >= prefixBatchSize { if len(h.batch) >= prefixBatchSize {
select { select {
case h.batchFullCh <- struct{}{}: case h.batchFullCh <- struct{}{}:
@@ -263,7 +250,7 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
// Stop stops the flush goroutine, which makes one last flush before it returns, // Stop stops the flush goroutine, which makes one last flush before it returns,
// and waits for it until ctx ends. That flush writes at most // and waits for it until ctx ends. That flush writes at most
// tableChangesPerWrite of the routing table's changes, and the prefixes of // maxTableChangesPerWrite of the routing table's changes, and the prefixes of
// the updates the handler holds. Once ctx has ended the flush goroutine skips // the updates the handler holds. Once ctx has ended the flush goroutine skips
// that last flush, and a write still running is lost when the process exits. A // that last flush, and a write still running is lost when the process exits. A
// failed write of the table's changes is not retried either. Stop logs how many // failed write of the table's changes is not retried either. Stop logs how many
@@ -391,13 +378,14 @@ func (h *PrefixHandler) applyToTable(update prefixUpdate) {
}) })
} }
// writeTableChanges writes at most tableChangesPerWrite changes of the routing // writeTableChanges writes at most maxTableChangesPerWrite changes of the
// table in memory, and every peer the table numbered since the last write. // routing table in memory, and every peer the table numbered since the last
// write.
// When the write fails, they go back to the table, so that a later write writes // When the write fails, they go back to the table, so that a later write writes
// them again, each pair with its content then. // them again, each pair with its content then.
func (h *PrefixHandler) writeTableChanges() { func (h *PrefixHandler) writeTableChanges() {
h.mu.Lock() h.mu.Lock()
changes, newPeers := h.table.TakeChanges(h.tableChangesPerWrite) changes, newPeers := h.table.TakeChanges(maxTableChangesPerWrite)
h.writingTableChanges = len(changes) h.writingTableChanges = len(changes)
h.mu.Unlock() h.mu.Unlock()
@@ -237,10 +237,9 @@ func TestPrefixHandlerStopLogsNothingWhenItWroteEverything(t *testing.T) {
} }
} }
// longerThanAnyTest is a timer interval that no test lasts. A test gives it to // longerThanAnyTest is a flush timer interval that no test lasts. A test whose
// newPrefixHandler, newASHandler, newPeerHandler or newPeeringHandler when the // prefix, AS or peer handler must flush only when its batch is full and on Stop
// handler must write only when its batch is full, when the test says, or on // gives it to newPrefixHandler, newASHandler or newPeerHandler.
// Stop.
const longerThanAnyTest = time.Hour const longerThanAnyTest = time.Hour
// prefixHandlerWithoutTimer returns a prefix handler that writes to store and // prefixHandlerWithoutTimer returns a prefix handler that writes to store and
@@ -248,11 +247,9 @@ const longerThanAnyTest = time.Hour
// NewPrefixHandler's flush goroutine, so the test flushes itself. // NewPrefixHandler's flush goroutine, so the test flushes itself.
func prefixHandlerWithoutTimer(store database.Store) *PrefixHandler { func prefixHandlerWithoutTimer(store database.Store) *PrefixHandler {
return &PrefixHandler{ return &PrefixHandler{
db: store, db: store,
logger: logger.New(), logger: logger.New(),
table: routetable.New(routetable.Params{}), table: routetable.New(routetable.Params{}),
batchMaxSize: prefixBatchMaxSize,
tableChangesPerWrite: maxTableChangesPerWrite,
} }
} }
@@ -30,11 +30,6 @@ const (
// firstManyPrefix is the first prefix announceManyPrefixes announces. // firstManyPrefix is the first prefix announceManyPrefixes announces.
firstManyPrefix = "10.0.0.0/32" firstManyPrefix = "10.0.0.0/32"
// testMaxTableChangesPerWrite is a limit of one write of the routing
// table's changes that a test exceeds quickly, in place of
// maxTableChangesPerWrite.
testMaxTableChangesPerWrite = 10
) )
// tableCase is one test of the routing table in memory: the messages handed to // tableCase is one test of the routing table in memory: the messages handed to
@@ -277,9 +272,9 @@ func TestPrefixHandlerWritesTheTableChangesEachSecond(t *testing.T) {
t.Fatalf("wrote %+v, want the pair of %s and AS%d with "+ t.Fatalf("wrote %+v, want the pair of %s and AS%d with "+
"removed %t", changes, prefix, testASNB, step.wantRemoved) "removed %t", changes, prefix, testASNB, step.wantRemoved)
} }
case <-time.After(stepTimeout): case <-time.After(handleMessageTimeout):
t.Fatalf("the routing table's change was not written within %s", t.Fatalf("the routing table's change was not written within %s",
stepTimeout) handleMessageTimeout)
} }
} }
} }
@@ -376,32 +371,25 @@ func (s *tableWriteSizesStore) WriteRouteChanges(
// takes, then flushes three times. The first write must hold as many pairs as // takes, then flushes three times. The first write must hold as many pairs as
// one write takes, and the peer. The second must hold the other pairs, and the // one write takes, and the peer. The second must hold the other pairs, and the
// third flush must write nothing. The database must then hold the table in // third flush must write nothing. The database must then hold the table in
// memory. The handler's limit is small, and its timer does not fire. // memory.
func TestPrefixHandlerWritesManyTableChangesOverSeveralFlushes(t *testing.T) { func TestPrefixHandlerWritesManyTableChangesOverSeveralFlushes(t *testing.T) {
t.Parallel() t.Parallel()
const ( const (
pairs = testMaxTableChangesPerWrite * 3 / 2 pairs = maxTableChangesPerWrite * 3 / 2
flushes = 3 flushes = 3
) )
db := newTestDatabase(t) db := newTestDatabase(t)
store := &tableWriteSizesStore{Database: db} store := &tableWriteSizesStore{Database: db}
table := routetable.New(routetable.Params{}) h := prefixHandlerWithoutTimer(store)
announceManyPrefixes(table, pairs) announceManyPrefixes(h.table, pairs)
h := newPrefixHandler(store, logger.New(), table, longerThanAnyTest,
testMaxTableChangesPerWrite)
defer stopInTime(t, h)
for range flushes { for range flushes {
h.flushBatch() h.flushBatch()
} }
wantChanges := []int{ wantChanges := []int{maxTableChangesPerWrite, pairs - maxTableChangesPerWrite}
testMaxTableChangesPerWrite, pairs - testMaxTableChangesPerWrite,
}
wantNewPeers := []int{1, 0} wantNewPeers := []int{1, 0}
if !slices.Equal(store.changes, wantChanges) || if !slices.Equal(store.changes, wantChanges) ||
@@ -423,11 +411,11 @@ func TestPrefixHandlerWritesManyTableChangesOverSeveralFlushes(t *testing.T) {
// takes. Its last flush writes one write's worth, so Stop must log the pairs // takes. Its last flush writes one write's worth, so Stop must log the pairs
// that no write took, and must not say that the stop deadline passed. The // that no write took, and must not say that the stop deadline passed. The
// flush timer must not fire, or its writes could leave no pairs for Stop to // flush timer must not fire, or its writes could leave no pairs for Stop to
// log. The handler's limit of one write is small. // log.
func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) { func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) {
t.Parallel() t.Parallel()
const pairs = testMaxTableChangesPerWrite + 1 const pairs = maxTableChangesPerWrite + 1
var logs bytes.Buffer var logs bytes.Buffer
@@ -435,8 +423,7 @@ func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) {
table := routetable.New(routetable.Params{}) table := routetable.New(routetable.Params{})
announceManyPrefixes(table, pairs) announceManyPrefixes(table, pairs)
h := newPrefixHandler(store, newErrorLogger(&logs), table, longerThanAnyTest, h := newPrefixHandler(store, newErrorLogger(&logs), table, longerThanAnyTest)
testMaxTableChangesPerWrite)
h.Stop(t.Context()) h.Stop(t.Context())
written := 0 written := 0