Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ebb7bf4d71 |
@@ -35,15 +35,6 @@ tested.
|
|||||||
|
|
||||||
# Completed Steps
|
# Completed Steps
|
||||||
|
|
||||||
- 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 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)
|
|
||||||
- 2026-10-27: README "Rationale" says why RouteWatch exists: to answer our other
|
- 2026-10-27: README "Rationale" says why RouteWatch exists: to answer our other
|
||||||
services' questions about IP addresses, from its own routing table and the
|
services' questions about IP addresses, from its own routing table and the
|
||||||
WHOIS details it keeps. It still says what RIS Live does not give (closes
|
WHOIS details it keeps. It still says what RIS Live does not give (closes
|
||||||
|
|||||||
@@ -509,6 +509,8 @@ func announcementFeed(prefixes []string) string {
|
|||||||
func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) {
|
func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
|
const maxWait = 10 * time.Second
|
||||||
|
|
||||||
prefixes := []string{"192.0.2.0/24", "198.51.100.0/24", "203.0.113.0/24"}
|
prefixes := []string{"192.0.2.0/24", "198.51.100.0/24", "203.0.113.0/24"}
|
||||||
feed := announcementFeed(prefixes)
|
feed := announcementFeed(prefixes)
|
||||||
|
|
||||||
@@ -554,9 +556,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(maxWait):
|
||||||
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), maxWait)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -569,8 +571,8 @@ func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) {
|
|||||||
|
|
||||||
select {
|
select {
|
||||||
case <-shutDown:
|
case <-shutDown:
|
||||||
case <-time.After(stepTimeout):
|
case <-time.After(maxWait):
|
||||||
t.Fatalf("Shutdown did not return within %s", stepTimeout)
|
t.Fatalf("Shutdown did not return within %s", maxWait)
|
||||||
}
|
}
|
||||||
|
|
||||||
store.mu.Lock()
|
store.mu.Lock()
|
||||||
@@ -607,16 +609,21 @@ 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 within its deadline
|
||||||
// the HTTP server stopped, and must neither checkpoint nor close the database:
|
// plus a small margin, with the HTTP server stopped, and must neither
|
||||||
// the real one would wait for the write.
|
// checkpoint nor close 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
|
||||||
|
// margin is how long Shutdown may take after its deadline, to stop
|
||||||
|
// the HTTP server.
|
||||||
|
margin = 2 * time.Second
|
||||||
|
)
|
||||||
|
|
||||||
blocking := newBlockingStore()
|
blocking := newBlockingStore()
|
||||||
defer blocking.releaseWrites()
|
defer blocking.releaseWrites()
|
||||||
@@ -660,9 +667,9 @@ func TestShutdownDoesNotWaitForABlockedWrite(t *testing.T) {
|
|||||||
|
|
||||||
select {
|
select {
|
||||||
case <-shutDown:
|
case <-shutDown:
|
||||||
case <-time.After(deadline + stepTimeout):
|
case <-time.After(deadline + margin):
|
||||||
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", margin)
|
||||||
}
|
}
|
||||||
|
|
||||||
var dialer net.Dialer
|
var dialer net.Dialer
|
||||||
@@ -737,6 +744,8 @@ func (s *closeOrderStore) record(call string) {
|
|||||||
func TestShutdownCheckpointsAndClosesTheDatabaseLast(t *testing.T) {
|
func TestShutdownCheckpointsAndClosesTheDatabaseLast(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
|
const deadline = 10 * time.Second
|
||||||
|
|
||||||
store := &closeOrderStore{Store: newMockStore()}
|
store := &closeOrderStore{Store: newMockStore()}
|
||||||
log := logger.New()
|
log := logger.New()
|
||||||
cfg := &config.Config{}
|
cfg := &config.Config{}
|
||||||
@@ -754,7 +763,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(), deadline)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
|
|
||||||
rw.Shutdown(ctx)
|
rw.Shutdown(ctx)
|
||||||
|
|||||||
@@ -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()
|
||||||
|
|||||||
@@ -19,15 +19,9 @@ 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 HandleMessage to
|
||||||
// HandleMessage returning, a write starting or Shutdown returning, before
|
// return, or for a write to start, before it fails.
|
||||||
// it fails. Only a step that never comes should reach it. On a loaded
|
handleMessageTimeout = 2 * time.Second
|
||||||
// host, under the race detector, HandleMessage takes seconds to apply a
|
|
||||||
// full prefix batch. The helpers below bound every wait with it, Stop
|
|
||||||
// included. A test that hangs in one of them waits out at most two of
|
|
||||||
// these bounds, its step's and Stop's, and then fails with its own
|
|
||||||
// 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 +198,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 +210,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 +240,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 +261,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 +275,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 +288,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 +315,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 +345,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 +361,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 +380,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 +597,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 +618,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 +714,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 +736,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 +755,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 +763,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 +796,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 +809,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 +822,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")
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|||||||
@@ -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()
|
||||||
|
|||||||
@@ -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,13 +30,12 @@ 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
|
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// tableWriteTimeout is how long a test waits for the flush timer to write the
|
||||||
|
// routing table's changes: a few ticks of it.
|
||||||
|
const tableWriteTimeout = 5 * prefixBatchTimeout
|
||||||
|
|
||||||
// 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
|
||||||
// the prefix handler, and for each prefix the peers that carry it with each
|
// the prefix handler, and for each prefix the peers that carry it with each
|
||||||
// origin AS after them.
|
// origin AS after them.
|
||||||
@@ -277,9 +276,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(tableWriteTimeout):
|
||||||
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)
|
tableWriteTimeout)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -376,32 +375,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 +415,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 +427,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
|
||||||
|
|||||||
@@ -6,7 +6,6 @@ import (
|
|||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
"testing/synctest"
|
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"git.eeqj.de/sneak/routewatch/internal/database"
|
"git.eeqj.de/sneak/routewatch/internal/database"
|
||||||
@@ -342,66 +341,6 @@ func TestStartedServerWorksOutTheCards(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestStartedServerWorksOutTheCardsUntilStop starts a server with a context
|
|
||||||
// that ends as soon as Start returns. It runs in a synctest bubble, whose clock
|
|
||||||
// is fake: time.Sleep there moves the clock on and returns at once. The server
|
|
||||||
// must work the cards out again a minute later, and no more after Stop. Each
|
|
||||||
// time it works them out, it keeps a new copy. Before the fix the end of
|
|
||||||
// Start's context stopped the goroutine that works them out.
|
|
||||||
func TestStartedServerWorksOutTheCardsUntilStop(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
// The server is made outside the bubble. The streamer's metrics start a
|
|
||||||
// goroutine that never returns, and a bubble waits for each of its
|
|
||||||
// goroutines to return.
|
|
||||||
s := newServerWithTable(t, newCardsTable())
|
|
||||||
|
|
||||||
synctest.Test(t, func(t *testing.T) {
|
|
||||||
ctx, cancel := context.WithCancel(t.Context())
|
|
||||||
|
|
||||||
err := s.Start(ctx)
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("Start: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
cancel()
|
|
||||||
|
|
||||||
// The bubble's clock moves only while each of its goroutines waits
|
|
||||||
// on the bubble, and the goroutine that serves HTTP waits on the
|
|
||||||
// network. The test sends no request, so it closes the HTTP server.
|
|
||||||
_ = s.srv.Close()
|
|
||||||
|
|
||||||
synctest.Wait()
|
|
||||||
|
|
||||||
started := s.cards.Load()
|
|
||||||
if started == nil {
|
|
||||||
t.Fatal("the server did not work out the cards when it started")
|
|
||||||
}
|
|
||||||
|
|
||||||
time.Sleep(homeCardsRefreshInterval)
|
|
||||||
synctest.Wait()
|
|
||||||
|
|
||||||
if s.cards.Load() == started {
|
|
||||||
t.Error("the server did not work out the cards again a minute " +
|
|
||||||
"after its start context ended")
|
|
||||||
}
|
|
||||||
|
|
||||||
err = s.Stop(t.Context())
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("Stop: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
stopped := s.cards.Load()
|
|
||||||
|
|
||||||
time.Sleep(homeCardsRefreshInterval)
|
|
||||||
synctest.Wait()
|
|
||||||
|
|
||||||
if s.cards.Load() != stopped {
|
|
||||||
t.Error("the server worked out the cards again after Stop")
|
|
||||||
}
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
// waitUntil fails the test unless done reports true within maxWait.
|
// waitUntil fails the test unless done reports true within maxWait.
|
||||||
func waitUntil(t *testing.T, maxWait time.Duration, done func() bool) {
|
func waitUntil(t *testing.T, maxWait time.Duration, done func() bool) {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
|
|||||||
@@ -82,12 +82,6 @@ type Server struct {
|
|||||||
// cards is the copy of the home page's cards that the pages serve, nil
|
// cards is the copy of the home page's cards that the pages serve, nil
|
||||||
// until the server first works them out
|
// until the server first works them out
|
||||||
cards atomic.Pointer[homeCards]
|
cards atomic.Pointer[homeCards]
|
||||||
|
|
||||||
// stopCards ends the goroutine that works out the cards, and
|
|
||||||
// cardsStopped is closed once it has returned. Both are nil until Start
|
|
||||||
// starts the goroutine.
|
|
||||||
stopCards context.CancelFunc
|
|
||||||
cardsStopped chan struct{}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// New creates a new HTTP server that reads routes from table, the routing
|
// New creates a new HTTP server that reads routes from table, the routing
|
||||||
@@ -117,8 +111,7 @@ func New(
|
|||||||
// Start listens on the configured port and serves HTTP in the background.
|
// Start listens on the configured port and serves HTTP in the background.
|
||||||
// It returns an error if it cannot listen, for example because the port is
|
// It returns an error if it cannot listen, for example because the port is
|
||||||
// already in use. ctx bounds opening the port, not serving. Once listening,
|
// already in use. ctx bounds opening the port, not serving. Once listening,
|
||||||
// the server works out the home page's cards every minute until Stop: the end
|
// the server works out the home page's cards every minute until ctx ends.
|
||||||
// of ctx does not stop it.
|
|
||||||
func (s *Server) Start(ctx context.Context) error {
|
func (s *Server) Start(ctx context.Context) error {
|
||||||
const (
|
const (
|
||||||
network = "tcp"
|
network = "tcp"
|
||||||
@@ -155,22 +148,16 @@ func (s *Server) Start(ctx context.Context) error {
|
|||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
ctx, s.stopCards = context.WithCancel(context.WithoutCancel(ctx))
|
|
||||||
s.cardsStopped = make(chan struct{})
|
|
||||||
|
|
||||||
go func() {
|
go func() {
|
||||||
defer close(s.cardsStopped)
|
|
||||||
|
|
||||||
s.keepHomeCardsFresh(ctx, homeCardsRefreshInterval)
|
s.keepHomeCardsFresh(ctx, homeCardsRefreshInterval)
|
||||||
}()
|
}()
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// Stop stops the goroutine that works out the home page's cards, and waits
|
// Stop stops the HTTP server. It waits, until ctx ends, for the requests in
|
||||||
// until it has returned or ctx ends. It then stops the HTTP server. It waits,
|
// progress to finish, and then closes the connections of those still running
|
||||||
// until ctx ends, for the requests in progress to finish, and then closes the
|
// and returns ctx's error.
|
||||||
// connections of those still running and returns ctx's error.
|
|
||||||
func (s *Server) Stop(ctx context.Context) error {
|
func (s *Server) Stop(ctx context.Context) error {
|
||||||
if s.srv == nil {
|
if s.srv == nil {
|
||||||
return nil
|
return nil
|
||||||
@@ -178,16 +165,6 @@ func (s *Server) Stop(ctx context.Context) error {
|
|||||||
|
|
||||||
s.logger.Info("Stopping HTTP server")
|
s.logger.Info("Stopping HTTP server")
|
||||||
|
|
||||||
// Start returns before it starts the goroutine when it cannot listen.
|
|
||||||
if s.stopCards != nil {
|
|
||||||
s.stopCards()
|
|
||||||
|
|
||||||
select {
|
|
||||||
case <-s.cardsStopped:
|
|
||||||
case <-ctx.Done():
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
err := s.srv.Shutdown(ctx)
|
err := s.srv.Shutdown(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return errors.Join(err, s.srv.Close())
|
return errors.Join(err, s.srv.Close())
|
||||||
|
|||||||
Reference in New Issue
Block a user