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

Takes .golangci.yml, .dockerignore, .gitignore, .editorconfig, the
workflow and REPO_POLICIES.md from sneak/prompts dd4027b, keeping this
repository's own entries after the shared content. Lint (golangci-lint
v2.14.0) and test become Dockerfile phases that script/lint and
script/test build alone and uncached; the image build depends on both,
and script/cibuild bootstraps, runs script/check, then builds the image.
The agent guidance file moves to AGENTS.md. The Go code is fixed for
what the new configuration finds; reads that iterate rows now also
report an iteration error.

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 (
@@ -7,6 +8,7 @@ import (
 	"compress/gzip"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"math"
@@ -24,20 +26,20 @@ 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.
@@ -58,11 +60,17 @@ const (
 	backpressureSlope     = 2.0 // Slope for linear drop probability increase
 )
 
+var (
+	errAlreadyRunning   = errors.New("streamer already running")
+	errUnexpectedStatus = errors.New("unexpected status code")
+)
+
 // 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
@@ -113,7 +121,7 @@ type Streamer struct {
 	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
@@ -121,7 +129,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,
@@ -153,7 +162,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),
 		},
 	}
 
@@ -171,22 +181,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
 
@@ -216,44 +228,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()
@@ -314,12 +303,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()
@@ -330,11 +322,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
@@ -343,7 +362,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",
@@ -369,16 +389,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",
@@ -395,6 +418,7 @@ func (s *Streamer) logMetrics() {
 		}
 		info.metrics.mu.Unlock()
 	}
+
 	s.mu.RUnlock()
 }
 
@@ -403,7 +427,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 := minBackoffDelay
 	consecutiveFailures := 0
@@ -444,6 +469,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 = minBackoffDelay
 			consecutiveFailures = 0
 		} else {
@@ -477,7 +503,7 @@ func (s *Streamer) stream(ctx context.Context) error {
 	connCtx, connCancel := context.WithCancel(ctx)
 	defer connCancel()
 
-	req, err := http.NewRequestWithContext(ctx, "GET", s.url, nil)
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, s.url, nil)
 	if err != nil {
 		return fmt.Errorf("failed to create request: %w", err)
 	}
@@ -490,13 +516,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
@@ -510,7 +537,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")
@@ -518,9 +547,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()
@@ -537,25 +563,41 @@ 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, reader)
+}
+
+// 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.
+func (s *Streamer) readStream(ctx context.Context, reader io.Reader) error {
 	scanner := bufio.NewScanner(reader)
 
 	for scanner.Scan() {
@@ -572,149 +614,176 @@ 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 "UPDATE":
-			// 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 "STATE":
-			// Peer state changes - silently ignore
-			continue
-		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
-			}
-
-			// Check if we should drop due to backpressure
-			if s.shouldDropForBackpressure(info) {
-				atomic.AddUint64(&info.metrics.droppedCount, 1)
-				atomic.AddUint64(&s.totalDropped, 1)
-
-				continue
-			}
-
-			// Try to queue the message
-			select {
-			case info.queue <- &msg:
-				// Message queued successfully
-				// 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()
-			default:
-				// Queue is full, drop the message
-				atomic.AddUint64(&info.metrics.droppedCount, 1)
-				atomic.AddUint64(&s.totalDropped, 1)
-			}
-		}
-		s.mu.RUnlock()
 	}
 
-	if err := scanner.Err(); err != nil {
+	err := scanner.Err()
+	if err != nil {
 		return fmt.Errorf("scanner error: %w", err)
 	}
 
 	return nil
 }
 
-// shouldDropForBackpressure determines if a message should be dropped based on queue utilization
+// processLine passes one line of the stream to the raw handler, parses it,
+// and dispatches an UPDATE 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 "UPDATE":
+		// 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 "STATE":
+		// Peer state changes - silently ignore
+		return nil
+	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, dropping it for
+// a handler whose queue is full or under backpressure.
+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
+		}
+
+		// Check if we should drop due to backpressure
+		if s.shouldDropForBackpressure(info) {
+			atomic.AddUint64(&info.metrics.droppedCount, 1)
+			s.totalDropped.Add(1)
+
+			continue
+		}
+
+		// Try to queue the message
+		select {
+		case info.queue <- msg:
+			// Message queued successfully
+			// 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()
+		default:
+			// Queue is full, drop the message
+			atomic.AddUint64(&info.metrics.droppedCount, 1)
+			s.totalDropped.Add(1)
+		}
+	}
+
+	s.mu.RUnlock()
+
+	return nil
+}
+
+// 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)