Compare commits
1
Commits
930eb39e09
..
next
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
38b43549b9 |
@@ -38,9 +38,9 @@ tested.
|
||||
- 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
|
||||
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 peering handler's path
|
||||
limit and of the prefix handler's limits give the handler small limits, so the
|
||||
package no longer runs past its 90-second limit (closes #213, #215)
|
||||
15 seconds for each step it expects. The tests of the handlers' limits give
|
||||
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
|
||||
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)
|
||||
|
||||
@@ -40,9 +40,13 @@ type ASHandler struct {
|
||||
db database.Store
|
||||
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
|
||||
// the batch under mu and writes it after releasing mu, so HandleMessage
|
||||
// waits for a database write only while batch is at asnBatchMaxSize.
|
||||
// waits for a database write only while batch is at batchMaxSize.
|
||||
mu sync.Mutex
|
||||
batch []asnOp
|
||||
// writing is how many operations the flush goroutine took from batch
|
||||
@@ -79,12 +83,13 @@ 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),
|
||||
stopCh: make(chan context.Context, 1),
|
||||
done: make(chan struct{}),
|
||||
db: db,
|
||||
logger: logger,
|
||||
batchMaxSize: asnBatchMaxSize,
|
||||
batch: make([]asnOp, 0, asnBatchSize),
|
||||
batchFullCh: make(chan struct{}, 1),
|
||||
stopCh: make(chan context.Context, 1),
|
||||
done: make(chan struct{}),
|
||||
}
|
||||
h.batchTaken.L = &h.mu
|
||||
|
||||
@@ -127,7 +132,7 @@ func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) {
|
||||
// While the batch is at its maximum size, wait until the flush goroutine
|
||||
// takes it. Meanwhile the handler takes nothing off its queue, which
|
||||
// fills, and the streamer drops messages for it.
|
||||
for len(h.batch) >= asnBatchMaxSize {
|
||||
for len(h.batch) >= h.batchMaxSize {
|
||||
h.batchTaken.Wait()
|
||||
}
|
||||
|
||||
@@ -151,7 +156,7 @@ func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) {
|
||||
|
||||
// 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
|
||||
// or the batch reaches asnBatchMaxSize.
|
||||
// or the batch reaches batchMaxSize.
|
||||
if len(h.batch) >= asnBatchSize {
|
||||
select {
|
||||
case h.batchFullCh <- struct{}{}:
|
||||
|
||||
@@ -19,7 +19,7 @@ func TestASHandlerStoresTheEarliestAndLatestTimesOfANewAS(t *testing.T) {
|
||||
db := newTestDatabase(t)
|
||||
|
||||
// Built without NewASHandler's flush goroutine, so the test flushes itself.
|
||||
h := &ASHandler{db: db, logger: logger.New()}
|
||||
h := &ASHandler{db: db, logger: logger.New(), batchMaxSize: asnBatchMaxSize}
|
||||
|
||||
first := peerAnnouncement(testPeerIP, testV4Prefix)
|
||||
second := peerAnnouncement(testPeerIP, testV4Prefix)
|
||||
@@ -54,7 +54,7 @@ func TestASHandlerRecordsNothingFromAPathWithASZero(t *testing.T) {
|
||||
store := newMockStore()
|
||||
|
||||
// Built without NewASHandler's flush goroutine, so the test flushes itself.
|
||||
h := &ASHandler{db: store, logger: logger.New()}
|
||||
h := &ASHandler{db: store, logger: logger.New(), batchMaxSize: asnBatchMaxSize}
|
||||
|
||||
msg := peerAnnouncement(testPeerIP, testV4Prefix)
|
||||
msg.Path = decodePath(t, testASNA, 0, testASNB)
|
||||
|
||||
@@ -246,16 +246,16 @@ func TestASNFetcherStopReturnsByItsDeadline(t *testing.T) {
|
||||
}
|
||||
|
||||
// TestRecordKeepsOnlyTheStatsWindow records a success and a failure every
|
||||
// second, by a clock the test sets, far longer than statsWindow, and never
|
||||
// second, by a clock the test sets, for three times statsWindow, and never
|
||||
// asks for the stats. The fetcher must keep only the times within the last
|
||||
// statsWindow.
|
||||
func TestRecordKeepsOnlyTheStatsWindow(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const (
|
||||
records = 100_000
|
||||
step = time.Second
|
||||
want = int(statsWindow / step)
|
||||
records = 3 * want
|
||||
)
|
||||
|
||||
clock := time.Now()
|
||||
|
||||
@@ -292,7 +292,9 @@ func handleWhileAWriteIsBlocked(
|
||||
// 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
|
||||
// extra must wait until the test lets the write finish, and return after. Then
|
||||
// it stops h, which writes what it holds.
|
||||
// it stops h, which writes what it holds. The tests lower h's maximum batch
|
||||
// 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(
|
||||
t *testing.T,
|
||||
store *blockingStore,
|
||||
@@ -407,7 +409,7 @@ func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
|
||||
store := newBlockingStore()
|
||||
h := newPrefixHandler(store, newErrorLogger(&logs),
|
||||
routetable.New(routetable.Params{}), longerThanAnyTest,
|
||||
prefixBatchMaxSize, maxTableChangesPerWrite)
|
||||
maxTableChangesPerWrite)
|
||||
fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
|
||||
stopWhileAWriteIsBlocked(t, store, h, h.done,
|
||||
[]*ristypes.RISMessage{prefixAnnouncement(fill)},
|
||||
@@ -735,10 +737,6 @@ func TestPeerHandlerHandlesMessagesWhileAWriteIsBlocked(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestPrefixHandlerWaitsAtTheMaximumBatchSize gives the prefix handler a
|
||||
// maximum batch size of one batch, so the test reaches it with one batch more.
|
||||
// Each update is applied to the routing table, which makes the real maximum of
|
||||
// ten batches slow to reach under the race detector.
|
||||
func TestPrefixHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -746,7 +744,8 @@ func TestPrefixHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
|
||||
|
||||
store := newBlockingStore()
|
||||
h := newPrefixHandler(store, logger.New(), routetable.New(routetable.Params{}),
|
||||
longerThanAnyTest, prefixBatchSize, maxTableChangesPerWrite)
|
||||
longerThanAnyTest, maxTableChangesPerWrite)
|
||||
h.batchMaxSize = prefixBatchSize
|
||||
fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
|
||||
toMax := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
|
||||
handleAtMaxBatchSize(t, store, h,
|
||||
@@ -768,8 +767,9 @@ func TestASHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
|
||||
|
||||
store := newBlockingStore()
|
||||
h := newASHandler(store, logger.New(), longerThanAnyTest)
|
||||
h.batchMaxSize = asnBatchSize
|
||||
fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
|
||||
toMax := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchMaxSize)}
|
||||
toMax := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
|
||||
handleAtMaxBatchSize(t, store, h,
|
||||
[]*ristypes.RISMessage{{Path: fill}},
|
||||
[]*ristypes.RISMessage{{Path: toMax}},
|
||||
@@ -787,7 +787,7 @@ 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.
|
||||
// would fail.
|
||||
func TestPeerHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -795,10 +795,11 @@ func TestPeerHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
|
||||
|
||||
store := newBlockingStore()
|
||||
h := newPeerHandler(store, logger.New(), longerThanAnyTest)
|
||||
h.batchMaxSize = peerBatchSize
|
||||
message := []*ristypes.RISMessage{{Peer: testPeerIP}}
|
||||
handleAtMaxBatchSize(t, store, h,
|
||||
slices.Repeat(message, peerBatchSize),
|
||||
slices.Repeat(message, peerBatchMaxSize),
|
||||
slices.Repeat(message, peerBatchSize),
|
||||
&ristypes.RISMessage{Peer: otherPeerIP})
|
||||
|
||||
store.mu.Lock()
|
||||
@@ -842,7 +843,10 @@ func TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull(t *testing.T) {
|
||||
name: "AS",
|
||||
batchSize: asnBatchSize,
|
||||
handle: func(n int) chan struct{} {
|
||||
h := &ASHandler{batchFullCh: make(chan struct{}, 1)}
|
||||
h := &ASHandler{
|
||||
batchMaxSize: asnBatchMaxSize,
|
||||
batchFullCh: make(chan struct{}, 1),
|
||||
}
|
||||
// The origin AS and each AS in the path are one update each.
|
||||
h.HandleMessage(&ristypes.RISMessage{
|
||||
Path: ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, n-1)},
|
||||
@@ -855,7 +859,10 @@ func TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull(t *testing.T) {
|
||||
name: "peer",
|
||||
batchSize: peerBatchSize,
|
||||
handle: func(n int) chan struct{} {
|
||||
h := &PeerHandler{batchFullCh: make(chan struct{}, 1)}
|
||||
h := &PeerHandler{
|
||||
batchMaxSize: peerBatchMaxSize,
|
||||
batchFullCh: make(chan struct{}, 1),
|
||||
}
|
||||
for range n {
|
||||
h.HandleMessage(&ristypes.RISMessage{Peer: testPeerIP})
|
||||
}
|
||||
|
||||
@@ -43,10 +43,14 @@ type PeerHandler struct {
|
||||
db database.Store
|
||||
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
|
||||
// takes the batch under mu and writes it after releasing mu, so
|
||||
// HandleMessage waits for a database write only while peerBatch is at
|
||||
// peerBatchMaxSize.
|
||||
// batchMaxSize.
|
||||
mu sync.Mutex
|
||||
peerBatch []peerUpdate
|
||||
// writing is how many updates the flush goroutine took from peerBatch
|
||||
@@ -84,12 +88,13 @@ 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),
|
||||
stopCh: make(chan context.Context, 1),
|
||||
done: make(chan struct{}),
|
||||
db: db,
|
||||
logger: logger,
|
||||
batchMaxSize: peerBatchMaxSize,
|
||||
peerBatch: make([]peerUpdate, 0, peerBatchSize),
|
||||
batchFullCh: make(chan struct{}, 1),
|
||||
stopCh: make(chan context.Context, 1),
|
||||
done: make(chan struct{}),
|
||||
}
|
||||
h.batchTaken.L = &h.mu
|
||||
|
||||
@@ -135,7 +140,7 @@ func (h *PeerHandler) HandleMessage(msg *ristypes.RISMessage) {
|
||||
// While the batch is at its maximum size, wait until the flush goroutine
|
||||
// takes it. Meanwhile the handler takes nothing off its queue, which
|
||||
// fills, and the streamer drops messages for it.
|
||||
for len(h.peerBatch) >= peerBatchMaxSize {
|
||||
for len(h.peerBatch) >= h.batchMaxSize {
|
||||
h.batchTaken.Wait()
|
||||
}
|
||||
|
||||
@@ -149,7 +154,7 @@ func (h *PeerHandler) HandleMessage(msg *ristypes.RISMessage) {
|
||||
|
||||
// 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
|
||||
// or the batch reaches peerBatchMaxSize.
|
||||
// or the batch reaches batchMaxSize.
|
||||
if len(h.peerBatch) >= peerBatchSize {
|
||||
select {
|
||||
case h.batchFullCh <- struct{}{}:
|
||||
|
||||
@@ -35,7 +35,7 @@ func TestPeerHandlerWritesTheEarliestAndLatestTimesOfANewPeer(t *testing.T) {
|
||||
|
||||
// Built without NewPeerHandler's flush goroutine, so the test flushes
|
||||
// itself.
|
||||
h := &PeerHandler{db: store, logger: logger.New()}
|
||||
h := &PeerHandler{db: store, logger: logger.New(), batchMaxSize: peerBatchMaxSize}
|
||||
|
||||
first := peerAnnouncement(testPeerIP, testV4Prefix)
|
||||
second := peerAnnouncement(testPeerIP, testV4Prefix)
|
||||
|
||||
@@ -60,9 +60,11 @@ type PrefixHandler struct {
|
||||
metrics *metrics.Tracker
|
||||
table *routetable.Table
|
||||
|
||||
// batchMaxSize is prefixBatchMaxSize, and tableChangesPerWrite is
|
||||
// maxTableChangesPerWrite, except in tests that give smaller ones.
|
||||
batchMaxSize int
|
||||
// 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
|
||||
@@ -125,24 +127,25 @@ func NewPrefixHandler(
|
||||
db database.Store, logger *logger.Logger, table *routetable.Table,
|
||||
) *PrefixHandler {
|
||||
return newPrefixHandler(db, logger, table, prefixBatchTimeout,
|
||||
prefixBatchMaxSize, maxTableChangesPerWrite)
|
||||
maxTableChangesPerWrite)
|
||||
}
|
||||
|
||||
// newPrefixHandler is NewPrefixHandler with the flush timer's interval,
|
||||
// batchTimeout, in place of prefixBatchTimeout. batchMaxSize and
|
||||
// tableChangesPerWrite replace prefixBatchMaxSize and maxTableChangesPerWrite.
|
||||
// A test that the timer must not disturb gives an interval longer than the test
|
||||
// runs. A test of either limit gives a smaller one, so that it reaches it fast.
|
||||
// batchTimeout, in place of prefixBatchTimeout, and tableChangesPerWrite in
|
||||
// place of maxTableChangesPerWrite. A test that the timer must not disturb
|
||||
// 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(
|
||||
db database.Store, logger *logger.Logger, table *routetable.Table,
|
||||
batchTimeout time.Duration, batchMaxSize, tableChangesPerWrite int,
|
||||
batchTimeout time.Duration, tableChangesPerWrite int,
|
||||
) *PrefixHandler {
|
||||
h := &PrefixHandler{
|
||||
db: db,
|
||||
logger: logger,
|
||||
table: table,
|
||||
batch: make([]seenPrefix, 0, prefixBatchSize),
|
||||
batchMaxSize: batchMaxSize,
|
||||
batchMaxSize: prefixBatchMaxSize,
|
||||
tableChangesPerWrite: tableChangesPerWrite,
|
||||
batchFullCh: make(chan struct{}, 1),
|
||||
stopCh: make(chan context.Context, 1),
|
||||
|
||||
@@ -391,7 +391,7 @@ func TestPrefixHandlerWritesManyTableChangesOverSeveralFlushes(t *testing.T) {
|
||||
announceManyPrefixes(table, pairs)
|
||||
|
||||
h := newPrefixHandler(store, logger.New(), table, longerThanAnyTest,
|
||||
prefixBatchMaxSize, testMaxTableChangesPerWrite)
|
||||
testMaxTableChangesPerWrite)
|
||||
|
||||
defer stopInTime(t, h)
|
||||
|
||||
@@ -436,7 +436,7 @@ func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) {
|
||||
announceManyPrefixes(table, pairs)
|
||||
|
||||
h := newPrefixHandler(store, newErrorLogger(&logs), table, longerThanAnyTest,
|
||||
prefixBatchMaxSize, testMaxTableChangesPerWrite)
|
||||
testMaxTableChangesPerWrite)
|
||||
h.Stop(t.Context())
|
||||
|
||||
written := 0
|
||||
|
||||
Reference in New Issue
Block a user