diff --git a/internal/routewatch/app.go b/internal/routewatch/app.go index a4d1d5b..1a4e925 100644 --- a/internal/routewatch/app.go +++ b/internal/routewatch/app.go @@ -116,17 +116,18 @@ func (rw *RouteWatch) Run(ctx context.Context) error { return err } + // Start ASN WHOIS fetcher for background updates. The HTTP handlers read + // the server's fetcher without a lock, so it is set before they run. + rw.asnFetcher = NewASNFetcher(rw.db, rw.logger.Logger) + rw.asnFetcher.Start(ctx) + rw.server.SetASNFetcher(rw.asnFetcher) + // Start HTTP server err = rw.server.Start(ctx) if err != nil { return err } - // Start ASN WHOIS fetcher for background updates - rw.asnFetcher = NewASNFetcher(rw.db, rw.logger.Logger) - rw.asnFetcher.Start(ctx) - rw.server.SetASNFetcher(rw.asnFetcher) - // Start database maintenance goroutine rw.dbMaintainer = NewDBMaintainer(rw.db, rw.logger.Logger) rw.dbMaintainer.Start(ctx) diff --git a/internal/server/server.go b/internal/server/server.go index cd60c3f..0d97002 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -139,7 +139,8 @@ func (s *Server) Stop(ctx context.Context) error { return nil } -// SetASNFetcher sets the ASN WHOIS fetcher for on-demand lookups. +// SetASNFetcher sets the ASN WHOIS fetcher for on-demand lookups. Call it +// before Start: the handlers read the fetcher without a lock. func (s *Server) SetASNFetcher(fetcher ASNFetcher) { s.asnFetcher = fetcher } diff --git a/internal/streamer/streamer.go b/internal/streamer/streamer.go index 2df25f5..5beaf95 100644 --- a/internal/streamer/streamer.go +++ b/internal/streamer/streamer.go @@ -106,12 +106,12 @@ type RawMessageHandler func(line string) // handlerMetrics tracks performance metrics for a handler type handlerMetrics struct { processedCount uint64 // Total messages processed - droppedCount uint64 // Total messages dropped + droppedCount atomic.Uint64 // Total messages dropped; not under mu totalTime time.Duration // Total processing time (for average calculation) minTime time.Duration // Minimum processing time maxTime time.Duration // Maximum processing time queueHighWaterMark int // Maximum queue length seen - mu sync.Mutex // Protects the metrics + mu sync.Mutex // Protects the other metrics } // handlerInfo wraps a handler with its queue and metrics @@ -371,7 +371,7 @@ func (s *Streamer) GetHandlerStats() []HandlerStats { QueueCapacity: cap(info.queue), QueueHighWaterMark: info.metrics.queueHighWaterMark, ProcessedCount: info.metrics.processedCount, - DroppedCount: info.metrics.droppedCount, + DroppedCount: info.metrics.droppedCount.Load(), MinProcessTime: info.metrics.minTime, MaxProcessTime: info.metrics.maxTime, } @@ -488,7 +488,7 @@ func (s *Streamer) logMetrics() { "queue_len", len(info.queue), "queue_cap", cap(info.queue), "processed", info.metrics.processedCount, - "dropped", info.metrics.droppedCount, + "dropped", info.metrics.droppedCount.Load(), "avg_time", avgTime, "min_time", info.metrics.minTime, "max_time", info.metrics.maxTime, @@ -858,7 +858,7 @@ func (s *Streamer) dispatchMessage( } if !s.queueMessage(info, msg) { - atomic.AddUint64(&info.metrics.droppedCount, 1) + info.metrics.droppedCount.Add(1) s.totalDropped.Add(1) continue diff --git a/internal/streamer/streamer_internal_test.go b/internal/streamer/streamer_internal_test.go index 20f33a5..e124caa 100644 --- a/internal/streamer/streamer_internal_test.go +++ b/internal/streamer/streamer_internal_test.go @@ -621,6 +621,68 @@ func TestStreamReadsLongLine(t *testing.T) { } } +// blockedHandler wants UPDATE messages, has room on its queue for one, and +// does not return from HandleMessage until release is closed. +type blockedHandler struct { + release chan struct{} +} + +func (blockedHandler) WantsMessage(messageType string) bool { + return messageType == ristypes.MessageTypeUpdate +} + +func (blockedHandler) QueueCapacity() int { return 1 } + +func (h blockedHandler) HandleMessage(*ristypes.RISMessage) { <-h.release } + +// TestHandlerStatsReadWhileMessagesAreDropped has the server send a few UPDATE +// lines to a blockedHandler, so the read loop drops all but the one the +// handler holds and the one on its queue, while the test reads the handler's +// statistics. The drop count must reach the number dropped, and the race +// detector must find the count read and written safely. +func TestHandlerStatsReadWhileMessagesAreDropped(t *testing.T) { + t.Parallel() + + const ( + lines = 5 + wantDropped = lines - 2 + maxWait = 5 * time.Second + ) + + srv := httptest.NewServer(http.HandlerFunc( + func(w http.ResponseWriter, r *http.Request) { + _, _ = io.WriteString(w, strings.Repeat(updateLine, lines)) + + flush(w) + <-r.Context().Done() + })) + defer srv.Close() + + s := New(logger.New(), metrics.New()) + s.url = srv.URL + handler := blockedHandler{release: make(chan struct{})} + s.RegisterHandler(handler) + + err := s.Start(t.Context()) + if err != nil { + t.Fatalf("Start returned error: %v", err) + } + defer s.Stop(t.Context()) + defer close(handler.release) + + // The test polls instead of waiting for a signal from the read loop: a + // signal would order the drops before the reads here, and the race + // detector would have nothing to report. + deadline := time.Now().Add(maxWait) + + for s.GetHandlerStats()[0].DroppedCount < wantDropped { + if time.Now().After(deadline) { + t.Fatalf("fewer than %d messages counted as dropped within %s", + wantDropped, maxWait) + } + } +} + // TestPeerCountCountsEveryPeerHeardFrom has the server send an UPDATE from // stateDownPeer and a KEEPALIVE from another peer, with no OPEN from either, // and then a message that ends stateDownPeer's session: a NOTIFICATION, or a