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)

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
 }

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

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