From 2f238236bd2230107f71bb56a0453e732b3a82d0 Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Tue, 20 Oct 2026 03:18:00 +0200 Subject: [PATCH] Read the drop counter atomically and set the WHOIS fetcher before serving (closes #115) The read loop added to a handler's drop count atomically while the statistics endpoint and the metrics log read it as a plain field, a data race whenever messages were dropped. The count is now an atomic.Uint64, read with Load. A test reads the handler statistics while the read loop drops messages for a handler that never finishes one. Run created the WHOIS fetcher and gave it to the HTTP server after the server had started, while request handlers could read it. The fetcher is now set before the server starts. Model: opus-5-5 --- internal/routewatch/app.go | 11 ++-- internal/server/server.go | 3 +- internal/streamer/streamer.go | 10 ++-- internal/streamer/streamer_internal_test.go | 62 +++++++++++++++++++++ 4 files changed, 75 insertions(+), 11 deletions(-) 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