Re-vendor the shared files from sneak/prompts at 0653579 (closes #52)

Takes .golangci.yml, .dockerignore, .gitignore, .editorconfig, the
workflow and REPO_POLICIES.md from sneak/prompts 0653579, with this
repository's own entries after the shared content. Lint (golangci-lint
v2.14.0) and test are Dockerfile phases that script/lint and script/test
build alone with a new CACHEBUST value, keeping Go's and golangci-lint's
caches; the image build depends on both. make fmt and make fmt-check also
format the markdown with prettier, pinned in yarn.lock. The agent
guidance moves to AGENTS.md. The Go code is fixed for what the new
configuration finds, and the AS and prefix detail reads now return the
rows' error after their loops.

Model: opus-5-5

internal/streamer/streamer.go

@@ -1,5 +1,6 @@
-// Package streamer implements an HTTP client that connects to the RIPE RIS Live streaming API,
-// parses BGP UPDATE messages from the JSON stream, and dispatches them to registered handlers.
+// Package streamer implements an HTTP client that connects to the RIPE RIS Live
+// streaming API, parses BGP UPDATE messages from the JSON stream, and
+// dispatches them to registered handlers.
 package streamer
 
 import (
@@ -25,50 +26,60 @@ import (
 // countingReader wraps an io.Reader and counts bytes read
 type countingReader struct {
 	reader io.Reader
-	count  int64
+	count  atomic.Int64
 }
 
 // Read implements io.Reader and counts bytes
 func (c *countingReader) Read(p []byte) (int, error) {
 	n, err := c.reader.Read(p)
-	atomic.AddInt64(&c.count, int64(n))
+	c.count.Add(int64(n))
 
 	return n, err
 }
 
 // Count returns the total bytes read
 func (c *countingReader) Count() int64 {
-	return atomic.LoadInt64(&c.count)
+	return c.count.Load()
 }
 
 // Configuration constants for the RIS Live streamer.
 const (
 	risLiveURL = "https://ris-live.ripe.net/v1/stream/?format=json&" +
 		"client=https%3A%2F%2Fgit.eeqj.de%2Fsneak%2Froutewatch"
-	metricsWindowSize     = 60 // seconds for rolling average
-	metricsUpdateRate     = time.Second
-	minBackoffDelay       = 5 * time.Second
-	maxBackoffDelay       = 320 * time.Second
-	streamIdleTimeout     = 60 * time.Second // Reconnect when no line arrives for this long
-	metricsLogInterval    = 10 * time.Second
-	bytesPerKB            = 1024
-	bytesPerMB            = 1024 * 1024
-	maxLineSize           = bytesPerMB // Well above an extended BGP message (RFC 8654) in hex plus its prefixes
-	maxConcurrentHandlers = 800        // Maximum number of concurrent message handlers
+	metricsWindowSize = 60 // seconds for rolling average
+	metricsUpdateRate = time.Second
+	minBackoffDelay   = 5 * time.Second
+	maxBackoffDelay   = 320 * time.Second
+	// Reconnect when no line arrives for this long
+	streamIdleTimeout  = 60 * time.Second
+	metricsLogInterval = 10 * time.Second
+	bytesPerKB         = 1024
+	bytesPerMB         = 1024 * 1024
+	// Well above an extended BGP message (RFC 8654) in hex plus its prefixes
+	maxLineSize           = bytesPerMB
+	maxConcurrentHandlers = 800 // Maximum number of concurrent message handlers
 
 	// Backpressure constants
 	backpressureThreshold = 0.5 // Start dropping at 50% queue utilization
 	backpressureSlope     = 2.0 // Slope for linear drop probability increase
 )
 
-// errStreamEnded is returned by stream when RIS Live ends the response.
-var errStreamEnded = errors.New("RIS Live ended the stream")
+var (
+	errAlreadyRunning   = errors.New("streamer already running")
+	errUnexpectedStatus = errors.New("unexpected status code")
+	// errStreamEnded is returned by stream when RIS Live ends the response.
+	errStreamEnded = errors.New("RIS Live ended the stream")
+	// errStreamIdle is returned by stream when no line arrived for the idle
+	// timeout.
+	errStreamIdle = errors.New("no line received")
+)
 
 // MessageHandler defines the interface for processing RIS messages.
 // Implementations must specify which message types they want to receive,
 // how to process messages, and their desired queue capacity.
 type MessageHandler interface {
-	// WantsMessage returns true if this handler wants to process messages of the given type.
+	// WantsMessage returns true if this handler wants to process messages of
+	// the given type.
 	WantsMessage(messageType string) bool
 
 	// HandleMessage processes a RIS message. This method is called from a dedicated
@@ -114,18 +125,20 @@ type handlerInfo struct {
 // backoff, dispatches messages to registered handlers via per-handler queues, and
 // implements backpressure to prevent queue overflow during high traffic periods.
 type Streamer struct {
-	logger       *logger.Logger
-	client       *http.Client
-	url          string
-	idleTimeout  time.Duration // Drop the connection when no line arrives for this long
-	minBackoff   time.Duration // First wait before reconnecting; doubles up to maxBackoffDelay
+	logger *logger.Logger
+	client *http.Client
+	url    string
+	// Drop the connection when no line arrives for this long
+	idleTimeout time.Duration
+	// First wait before reconnecting; doubles up to maxBackoffDelay
+	minBackoff   time.Duration
 	handlers     []*handlerInfo
 	rawHandler   RawMessageHandler
 	mu           sync.RWMutex
 	cancel       context.CancelFunc
 	running      bool
 	metrics      *metrics.Tracker
-	totalDropped uint64          // Total dropped messages across all handlers
+	totalDropped atomic.Uint64   // Total dropped messages across all handlers
 	random       *rand.Rand      // Random number generator for backpressure drops
 	bgpPeers     map[string]bool // Track active BGP peers by peer IP
 	bgpPeersMu   sync.RWMutex    // Protects bgpPeers map
@@ -133,7 +146,8 @@ type Streamer struct {
 
 // New creates a new Streamer instance configured to connect to the RIS Live API.
 // The logger is used for structured logging of connection events and errors.
-// The metrics tracker is used to record message counts, bytes received, and connection status.
+// The metrics tracker is used to record message counts, bytes received, and
+// connection status.
 func New(logger *logger.Logger, metrics *metrics.Tracker) *Streamer {
 	return &Streamer{
 		logger:      logger,
@@ -167,7 +181,8 @@ func (s *Streamer) RegisterHandler(handler MessageHandler) {
 		handler: handler,
 		queue:   make(chan *ristypes.RISMessage, handler.QueueCapacity()),
 		metrics: handlerMetrics{
-			minTime: time.Duration(math.MaxInt64), // Initialize to max so first value sets the floor
+			// Initialize to max so first value sets the floor
+			minTime: time.Duration(math.MaxInt64),
 		},
 	}
 
@@ -185,22 +200,24 @@ func (s *Streamer) RegisterHandler(handler MessageHandler) {
 func (s *Streamer) RegisterRawHandler(handler RawMessageHandler) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
+
 	s.rawHandler = handler
 }
 
 // Start begins streaming BGP updates from the RIS Live API in a background goroutine.
 // It starts worker goroutines for each registered handler and manages automatic
 // reconnection with exponential backoff on connection failures.
-// Returns an error if the streamer is already running.
-func (s *Streamer) Start() error {
+// Returns an error if the streamer is already running. It streams until Stop:
+// the end of ctx does not stop it.
+func (s *Streamer) Start(ctx context.Context) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
 	if s.running {
-		return fmt.Errorf("streamer already running")
+		return errAlreadyRunning
 	}
 
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx, cancel := context.WithCancel(context.WithoutCancel(ctx))
 	s.cancel = cancel
 	s.running = true
 
@@ -230,44 +247,21 @@ func (s *Streamer) Stop() {
 
 		return
 	}
+
 	s.cancel()
 	s.cancel = nil
 	// Close all handler queues to signal workers to stop
 	for _, info := range s.handlers {
 		close(info.queue)
 	}
+
 	s.running = false
 	s.mu.Unlock()
 	s.metrics.SetConnected(false)
 }
 
-// runHandlerWorker processes messages for a specific handler
-func (s *Streamer) runHandlerWorker(info *handlerInfo) {
-	for msg := range info.queue {
-		start := time.Now()
-		info.handler.HandleMessage(msg)
-		elapsed := time.Since(start)
-
-		// Update metrics
-		info.metrics.mu.Lock()
-		info.metrics.processedCount++
-		info.metrics.totalTime += elapsed
-
-		// Update min time
-		if elapsed < info.metrics.minTime {
-			info.metrics.minTime = elapsed
-		}
-
-		// Update max time
-		if elapsed > info.metrics.maxTime {
-			info.metrics.maxTime = elapsed
-		}
-		info.metrics.mu.Unlock()
-	}
-}
-
-// IsRunning reports whether the streamer is currently connected and processing messages.
-// This is safe to call concurrently from multiple goroutines.
+// IsRunning reports whether the streamer is currently connected and processing
+// messages. This is safe to call concurrently from multiple goroutines.
 func (s *Streamer) IsRunning() bool {
 	s.mu.RLock()
 	defer s.mu.RUnlock()
@@ -328,12 +322,15 @@ func (s *Streamer) GetHandlerStats() []HandlerStats {
 		// Calculate average time
 		if info.metrics.processedCount > 0 {
 			processedCount := info.metrics.processedCount
+
 			const maxInt64 = 1<<63 - 1
+
 			if processedCount > maxInt64 {
 				processedCount = maxInt64
 			}
-			//nolint:gosec // processedCount is explicitly bounded above
-			hs.AvgProcessTime = info.metrics.totalTime / time.Duration(processedCount)
+
+			hs.AvgProcessTime = info.metrics.totalTime /
+				time.Duration(processedCount)
 		}
 
 		info.metrics.mu.Unlock()
@@ -344,11 +341,38 @@ func (s *Streamer) GetHandlerStats() []HandlerStats {
 	return stats
 }
 
-// GetDroppedMessages returns the total number of messages dropped across all handlers
-// due to queue overflow or backpressure. This counter is monotonically increasing
-// and is safe to call concurrently.
+// GetDroppedMessages returns the total number of messages dropped across all
+// handlers due to queue overflow or backpressure. This counter is monotonically
+// increasing and is safe to call concurrently.
 func (s *Streamer) GetDroppedMessages() uint64 {
-	return atomic.LoadUint64(&s.totalDropped)
+	return s.totalDropped.Load()
+}
+
+// runHandlerWorker processes messages for a specific handler
+func (s *Streamer) runHandlerWorker(info *handlerInfo) {
+	for msg := range info.queue {
+		start := time.Now()
+
+		info.handler.HandleMessage(msg)
+
+		elapsed := time.Since(start)
+
+		// Update metrics
+		info.metrics.mu.Lock()
+		info.metrics.processedCount++
+		info.metrics.totalTime += elapsed
+
+		// Update min time
+		if elapsed < info.metrics.minTime {
+			info.metrics.minTime = elapsed
+		}
+
+		// Update max time
+		if elapsed > info.metrics.maxTime {
+			info.metrics.maxTime = elapsed
+		}
+		info.metrics.mu.Unlock()
+	}
 }
 
 // logMetrics logs the current streaming statistics
@@ -357,7 +381,8 @@ func (s *Streamer) logMetrics() {
 	uptime := time.Since(metrics.ConnectedSince)
 
 	const bitsPerMegabit = 1000000
-	totalDropped := atomic.LoadUint64(&s.totalDropped)
+
+	totalDropped := s.totalDropped.Load()
 
 	s.logger.Info(
 		"Stream statistics",
@@ -383,16 +408,19 @@ func (s *Streamer) logMetrics() {
 
 	// Log per-handler statistics
 	s.mu.RLock()
+
 	for i, info := range s.handlers {
 		info.metrics.mu.Lock()
 		if info.metrics.processedCount > 0 {
 			// Safe conversion: processedCount is bounded by maxInt64
 			processedCount := info.metrics.processedCount
+
 			const maxInt64 = 1<<63 - 1
+
 			if processedCount > maxInt64 {
 				processedCount = maxInt64
 			}
-			//nolint:gosec // processedCount is explicitly bounded above
+
 			avgTime := info.metrics.totalTime / time.Duration(processedCount)
 			s.logger.Info(
 				"Handler statistics",
@@ -409,6 +437,7 @@ func (s *Streamer) logMetrics() {
 		}
 		info.metrics.mu.Unlock()
 	}
+
 	s.mu.RUnlock()
 }
 
@@ -417,7 +446,8 @@ func (s *Streamer) updateMetrics(messageBytes int) {
 	s.metrics.RecordMessage(int64(messageBytes))
 }
 
-// streamWithReconnect handles streaming with automatic reconnection and exponential backoff
+// streamWithReconnect handles streaming with automatic reconnection and
+// exponential backoff
 func (s *Streamer) streamWithReconnect(ctx context.Context) {
 	backoffDelay := s.minBackoff
 	consecutiveFailures := 0
@@ -455,6 +485,7 @@ func (s *Streamer) streamWithReconnect(ctx context.Context) {
 		if streamDuration > 30*time.Second {
 			s.logger.Info("Resetting backoff delay due to successful connection",
 				"stream_duration", streamDuration)
+
 			backoffDelay = s.minBackoff
 			consecutiveFailures = 0
 		} else {
@@ -494,7 +525,7 @@ func (s *Streamer) stream(ctx context.Context) error {
 	idleTimer := time.AfterFunc(s.idleTimeout, connCancel)
 	defer idleTimer.Stop()
 
-	req, err := http.NewRequestWithContext(connCtx, "GET", s.url, nil)
+	req, err := http.NewRequestWithContext(connCtx, http.MethodGet, s.url, nil)
 	if err != nil {
 		return fmt.Errorf("failed to create request: %w", err)
 	}
@@ -507,13 +538,14 @@ func (s *Streamer) stream(ctx context.Context) error {
 		return fmt.Errorf("failed to connect to RIS Live: %w", err)
 	}
 	defer func() {
-		if err := resp.Body.Close(); err != nil {
+		err := resp.Body.Close()
+		if err != nil {
 			s.logger.Error("Failed to close response body", "error", err)
 		}
 	}()
 
 	if resp.StatusCode != http.StatusOK {
-		return fmt.Errorf("unexpected status code: %d", resp.StatusCode)
+		return fmt.Errorf("%w: %d", errUnexpectedStatus, resp.StatusCode)
 	}
 
 	// Wrap body with counting reader to track actual wire bytes
@@ -527,7 +559,9 @@ func (s *Streamer) stream(ctx context.Context) error {
 			return fmt.Errorf("failed to create gzip reader: %w", err)
 		}
 		defer func() { _ = gzReader.Close() }()
+
 		reader = gzReader
+
 		s.logger.Info("Connected to RIS Live stream", "compression", "gzip")
 	} else {
 		s.logger.Info("Connected to RIS Live stream", "compression", "none")
@@ -535,9 +569,6 @@ func (s *Streamer) stream(ctx context.Context) error {
 
 	s.metrics.SetConnected(true)
 
-	// Track wire bytes for metrics updates
-	var lastWireBytes int64
-
 	// Start metrics logging goroutine
 	metricsTicker := time.NewTicker(metricsLogInterval)
 	defer metricsTicker.Stop()
@@ -554,25 +585,44 @@ func (s *Streamer) stream(ctx context.Context) error {
 	}()
 
 	// Wire byte update ticker - update metrics with actual wire bytes periodically
+	go s.recordWireBytes(connCtx, wireCounter)
+
+	return s.readStream(ctx, connCtx, reader, idleTimer)
+}
+
+// recordWireBytes adds the bytes wireCounter has counted to the metrics once a
+// second until ctx is cancelled.
+func (s *Streamer) recordWireBytes(
+	ctx context.Context, wireCounter *countingReader,
+) {
 	wireUpdateTicker := time.NewTicker(time.Second)
 	defer wireUpdateTicker.Stop()
 
-	go func() {
-		for {
-			select {
-			case <-wireUpdateTicker.C:
-				currentBytes := wireCounter.Count()
-				delta := currentBytes - lastWireBytes
-				if delta > 0 {
-					s.metrics.RecordWireBytes(delta)
-					lastWireBytes = currentBytes
-				}
-			case <-connCtx.Done():
-				return
-			}
-		}
-	}()
+	var lastWireBytes int64
 
+	for {
+		select {
+		case <-wireUpdateTicker.C:
+			currentBytes := wireCounter.Count()
+
+			delta := currentBytes - lastWireBytes
+			if delta > 0 {
+				s.metrics.RecordWireBytes(delta)
+
+				lastWireBytes = currentBytes
+			}
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+// readStream hands each line of the stream to processLine until the stream
+// ends, a line fails, or ctx is cancelled. Each line restarts idleTimer, which
+// cancels connCtx when it fires.
+func (s *Streamer) readStream(
+	ctx, connCtx context.Context, reader io.Reader, idleTimer *time.Timer,
+) error {
 	scanner := bufio.NewScanner(reader)
 	scanner.Buffer(nil, maxLineSize)
 
@@ -592,129 +642,10 @@ func (s *Streamer) stream(ctx context.Context) error {
 			continue
 		}
 
-		// Update metrics with decompressed message size
-		s.updateMetrics(len(line))
-
-		// Call raw handler if registered
-		s.mu.RLock()
-		rawHandler := s.rawHandler
-		s.mu.RUnlock()
-
-		if rawHandler != nil {
-			// Call raw handler synchronously to preserve order
-			rawHandler(string(line))
+		err := s.processLine(ctx, line)
+		if err != nil {
+			return err
 		}
-
-		// Parse the message first
-		var wrapper ristypes.RISLiveMessage
-		if err := json.Unmarshal(line, &wrapper); err != nil {
-			// Log the error and return to trigger reconnection
-			s.logger.Error("Failed to parse JSON",
-				"error", err,
-				"line", string(line),
-				"line_length", len(line))
-
-			return fmt.Errorf("JSON parse error: %w", err)
-		}
-
-		// Check if it's a ris_message wrapper
-		if wrapper.Type != "ris_message" {
-			s.logger.Error("Unexpected wrapper type",
-				"type", wrapper.Type,
-				"line", string(line),
-			)
-
-			continue
-		}
-
-		// Get the actual message
-		msg := wrapper.Data
-
-		// Parse the timestamp
-		msg.ParsedTimestamp = time.Unix(int64(msg.Timestamp), 0).UTC()
-
-		// Process based on message type
-		switch msg.Type {
-		case ristypes.MessageTypeUpdate:
-			// Process BGP UPDATE messages
-			// Will be dispatched to handlers
-		case "RIS_PEER_STATE":
-			// RIS peer state messages - silently ignore
-			continue
-		case "KEEPALIVE":
-			// BGP keepalive messages - silently process
-			continue
-		case "OPEN":
-			// BGP open messages - track peer as active
-			s.bgpPeersMu.Lock()
-			s.bgpPeers[msg.Peer] = true
-			peerCount := len(s.bgpPeers)
-			s.bgpPeersMu.Unlock()
-			s.metrics.SetBGPPeerCount(peerCount)
-
-			s.logger.Info("BGP session opened",
-				"peer", msg.Peer,
-				"peer_asn", msg.PeerASN,
-				"total_peers", peerCount,
-			)
-
-			continue
-		case "NOTIFICATION":
-			// BGP notification messages (session closed)
-			s.bgpPeersMu.Lock()
-			delete(s.bgpPeers, msg.Peer)
-			peerCount := len(s.bgpPeers)
-			s.bgpPeersMu.Unlock()
-			s.metrics.SetBGPPeerCount(peerCount)
-
-			s.logger.Warn("BGP notification",
-				"peer", msg.Peer,
-				"peer_asn", msg.PeerASN,
-				"total_peers", peerCount,
-			)
-
-			continue
-		case ristypes.MessageTypeState:
-			// The BGP session with the peer changed state, for example went
-			// down. Dispatched to the handlers that want it.
-		default:
-			s.logger.Warn("Unknown message type, skipping",
-				"type", msg.Type,
-			)
-
-			continue
-		}
-
-		// Dispatch to interested handlers. Stop cancels ctx and closes the
-		// queues under the write lock, so if ctx is cancelled here, under the
-		// read lock, the queues are closed and must not be sent to.
-		s.mu.RLock()
-		if ctx.Err() != nil {
-			s.mu.RUnlock()
-
-			return ctx.Err()
-		}
-		for _, info := range s.handlers {
-			if !info.handler.WantsMessage(msg.Type) {
-				continue
-			}
-
-			if !s.queueMessage(info, &msg) {
-				atomic.AddUint64(&info.metrics.droppedCount, 1)
-				atomic.AddUint64(&s.totalDropped, 1)
-
-				continue
-			}
-
-			// Update high water mark if needed
-			queueLen := len(info.queue)
-			info.metrics.mu.Lock()
-			if queueLen > info.metrics.queueHighWaterMark {
-				info.metrics.queueHighWaterMark = queueLen
-			}
-			info.metrics.mu.Unlock()
-		}
-		s.mu.RUnlock()
 	}
 
 	if ctx.Err() != nil {
@@ -722,16 +653,161 @@ func (s *Streamer) stream(ctx context.Context) error {
 	}
 
 	if connCtx.Err() != nil {
-		return fmt.Errorf("no line received for %s", s.idleTimeout)
+		return fmt.Errorf("%w for %s", errStreamIdle, s.idleTimeout)
 	}
 
-	if err := scanner.Err(); err != nil {
+	err := scanner.Err()
+	if err != nil {
 		return fmt.Errorf("scanner error: %w", err)
 	}
 
 	return errStreamEnded
 }
 
+// processLine passes one line of the stream to the raw handler, parses it,
+// and dispatches an UPDATE or STATE message to the handlers. An error ends the
+// connection.
+func (s *Streamer) processLine(ctx context.Context, line []byte) error {
+	// Update metrics with decompressed message size
+	s.updateMetrics(len(line))
+
+	// Call raw handler if registered
+	s.mu.RLock()
+	rawHandler := s.rawHandler
+	s.mu.RUnlock()
+
+	if rawHandler != nil {
+		// Call raw handler synchronously to preserve order
+		rawHandler(string(line))
+	}
+
+	// Parse the message first
+	var wrapper ristypes.RISLiveMessage
+
+	err := json.Unmarshal(line, &wrapper)
+	if err != nil {
+		// Log the error and return to trigger reconnection
+		s.logger.Error("Failed to parse JSON",
+			"error", err,
+			"line", string(line),
+			"line_length", len(line))
+
+		return fmt.Errorf("JSON parse error: %w", err)
+	}
+
+	// Check if it's a ris_message wrapper
+	if wrapper.Type != "ris_message" {
+		s.logger.Error("Unexpected wrapper type",
+			"type", wrapper.Type,
+			"line", string(line),
+		)
+
+		return nil
+	}
+
+	// Get the actual message
+	msg := wrapper.Data
+
+	// Parse the timestamp
+	msg.ParsedTimestamp = time.Unix(int64(msg.Timestamp), 0).UTC()
+
+	// Process based on message type
+	switch msg.Type {
+	case ristypes.MessageTypeUpdate:
+		// Process BGP UPDATE messages
+		// Will be dispatched to handlers
+	case "RIS_PEER_STATE":
+		// RIS peer state messages - silently ignore
+		return nil
+	case "KEEPALIVE":
+		// BGP keepalive messages - silently process
+		return nil
+	case "OPEN":
+		// BGP open messages - track peer as active
+		s.bgpPeersMu.Lock()
+		s.bgpPeers[msg.Peer] = true
+		peerCount := len(s.bgpPeers)
+		s.bgpPeersMu.Unlock()
+		s.metrics.SetBGPPeerCount(peerCount)
+
+		s.logger.Info("BGP session opened",
+			"peer", msg.Peer,
+			"peer_asn", msg.PeerASN,
+			"total_peers", peerCount,
+		)
+
+		return nil
+	case "NOTIFICATION":
+		// BGP notification messages (session closed)
+		s.bgpPeersMu.Lock()
+		delete(s.bgpPeers, msg.Peer)
+		peerCount := len(s.bgpPeers)
+		s.bgpPeersMu.Unlock()
+		s.metrics.SetBGPPeerCount(peerCount)
+
+		s.logger.Warn("BGP notification",
+			"peer", msg.Peer,
+			"peer_asn", msg.PeerASN,
+			"total_peers", peerCount,
+		)
+
+		return nil
+	case ristypes.MessageTypeState:
+		// The BGP session with the peer changed state, for example went
+		// down. Dispatched to the handlers that want it.
+	default:
+		s.logger.Warn("Unknown message type, skipping",
+			"type", msg.Type,
+		)
+
+		return nil
+	}
+
+	return s.dispatchMessage(ctx, &msg)
+}
+
+// dispatchMessage queues msg for each handler that wants it, unless
+// queueMessage drops it for that handler.
+func (s *Streamer) dispatchMessage(
+	ctx context.Context, msg *ristypes.RISMessage,
+) error {
+	// Stop cancels ctx and closes the queues under the write lock, so if ctx
+	// is cancelled here, under the read lock, the queues are closed and must
+	// not be sent to.
+	s.mu.RLock()
+
+	if ctx.Err() != nil {
+		s.mu.RUnlock()
+
+		return ctx.Err()
+	}
+
+	for _, info := range s.handlers {
+		if !info.handler.WantsMessage(msg.Type) {
+			continue
+		}
+
+		if !s.queueMessage(info, msg) {
+			atomic.AddUint64(&info.metrics.droppedCount, 1)
+			s.totalDropped.Add(1)
+
+			continue
+		}
+
+		// Update high water mark if needed
+		queueLen := len(info.queue)
+		info.metrics.mu.Lock()
+		if queueLen > info.metrics.queueHighWaterMark {
+			info.metrics.queueHighWaterMark = queueLen
+		}
+		info.metrics.mu.Unlock()
+	}
+
+	s.mu.RUnlock()
+
+	return nil
+}
+
 // queueMessage puts msg on the handler's queue and reports false if it dropped
 // msg instead. It never waits for room, so a full queue holds up neither the
 // read loop nor Stop.
@@ -746,7 +822,9 @@ func (s *Streamer) stream(ctx context.Context) error {
 //
 // The held list stays small: a STATE message is a few hundred bytes and the
 // feed sends about ten a second, so a queue full for a minute holds about 600.
-func (s *Streamer) queueMessage(info *handlerInfo, msg *ristypes.RISMessage) bool {
+func (s *Streamer) queueMessage(
+	info *handlerInfo, msg *ristypes.RISMessage,
+) bool {
 	for len(info.heldState) > 0 && trySend(info.queue, info.heldState[0]) {
 		info.heldState = info.heldState[1:]
 	}
@@ -776,7 +854,8 @@ func trySend(queue chan *ristypes.RISMessage, msg *ristypes.RISMessage) bool {
 	}
 }
 
-// shouldDropForBackpressure determines if a message should be dropped based on queue utilization
+// shouldDropForBackpressure determines if a message should be dropped based on
+// queue utilization
 func (s *Streamer) shouldDropForBackpressure(info *handlerInfo) bool {
 	// Calculate queue utilization
 	queueLen := len(info.queue)