Read the drop counter atomically and set the WHOIS fetcher before serving (closes #115)
check / check (push) Canceled after 0s

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
This commit was merged in pull request #139.
This commit is contained in:
2026-10-20 03:18:00 +02:00
parent f489321e3a
commit 2f238236bd
4 changed files with 75 additions and 11 deletions
+5 -5
View File
@@ -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