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_internal_test.go

@@ -0,0 +1,332 @@
+package streamer
+
+import (
+	"context"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"runtime"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"git.eeqj.de/sneak/routewatch/internal/logger"
+	"git.eeqj.de/sneak/routewatch/internal/metrics"
+	"git.eeqj.de/sneak/routewatch/internal/ristypes"
+)
+
+func TestNewStreamer(t *testing.T) {
+	t.Parallel()
+
+	logger := logger.New()
+	metricsTracker := metrics.New()
+	s := New(logger, metricsTracker)
+
+	if s == nil {
+		t.Fatal("New() returned nil")
+	}
+
+	if s.logger != logger {
+		t.Error("logger not set correctly")
+	}
+
+	if s.client == nil {
+		t.Error("HTTP client not initialized")
+	}
+
+	if s.handlers == nil {
+		t.Error("handlers slice not initialized")
+	}
+
+	if s.metrics != metricsTracker {
+		t.Error("metrics tracker not set correctly")
+	}
+
+	if s.idleTimeout != streamIdleTimeout {
+		t.Errorf("idleTimeout = %s, want %s", s.idleTimeout, streamIdleTimeout)
+	}
+
+	if s.minBackoff != minBackoffDelay {
+		t.Errorf("minBackoff = %s, want %s", s.minBackoff, minBackoffDelay)
+	}
+}
+
+// TestStreamDoesNotLeakTickersAcrossReconnects drives many short-lived
+// connections (each stream call is one reconnect cycle) and asserts the
+// goroutine count returns to its starting value. Each connection starts two
+// ticker goroutines; before the fix they lived until the streamer's lifetime
+// context was cancelled, so every reconnect leaked two.
+//
+//nolint:paralleltest // counts goroutines, so no other test may run alongside
+func TestStreamDoesNotLeakTickersAcrossReconnects(t *testing.T) {
+	// The handler returns immediately, so the response body is empty and each
+	// stream call ends at once, standing in for a dropped connection.
+	srv := httptest.NewServer(http.HandlerFunc(
+		func(_ http.ResponseWriter, _ *http.Request) {}))
+	defer srv.Close()
+
+	s := New(logger.New(), metrics.New())
+	s.url = srv.URL
+
+	// One warm-up connection so any persistent HTTP transport goroutine exists
+	// before we take the baseline.
+	err := s.stream(context.Background())
+	if !errors.Is(err, errStreamEnded) {
+		t.Fatalf("warm-up stream returned %v, want %v", err, errStreamEnded)
+	}
+
+	s.client.CloseIdleConnections()
+
+	baseline := settledGoroutineCount()
+
+	const reconnects = 20
+	for range reconnects {
+		err := s.stream(context.Background())
+		if !errors.Is(err, errStreamEnded) {
+			t.Fatalf("stream returned %v, want %v", err, errStreamEnded)
+		}
+	}
+
+	s.client.CloseIdleConnections()
+
+	if !waitForGoroutines(baseline) {
+		t.Fatalf("goroutines did not return to baseline %d after %d reconnects, got %d",
+			baseline, reconnects, runtime.NumGoroutine())
+	}
+}
+
+// updateLine is one UPDATE message as RIS Live sends it.
+const updateLine = `{"type":"ris_message","data":{"type":"UPDATE",` +
+	`"peer":"192.0.2.1","peer_asn":"64496","timestamp":1700000000}}` + "\n"
+
+// updateHandler wants UPDATE messages. For each one it handles it sends on
+// handled if there is room, and otherwise does nothing; a nil handled is
+// never sent on.
+type updateHandler struct {
+	handled chan struct{}
+}
+
+func (updateHandler) WantsMessage(messageType string) bool {
+	return messageType == "UPDATE"
+}
+
+func (updateHandler) QueueCapacity() int { return 10 }
+
+func (h updateHandler) HandleMessage(*ristypes.RISMessage) {
+	select {
+	case h.handled <- struct{}{}:
+	default:
+	}
+}
+
+// flush sends what w has buffered to the client. The writers httptest servers
+// hand to handlers can always do that.
+func flush(w http.ResponseWriter) {
+	flusher, ok := w.(http.Flusher)
+	if ok {
+		flusher.Flush()
+	}
+}
+
+// TestStopBeforeMessageReachesQueues stops the streamer after the read loop
+// has checked for cancellation but before it hands the message to the handler
+// queues. That is the gap a stop from another goroutine can land in, and it
+// used to end in "send on closed channel". The raw handler runs in that gap on
+// the read loop itself, so calling Stop from it hits the gap every time.
+func TestStopBeforeMessageReachesQueues(t *testing.T) {
+	t.Parallel()
+
+	srv := httptest.NewServer(http.HandlerFunc(
+		func(w http.ResponseWriter, _ *http.Request) {
+			_, _ = io.WriteString(w, updateLine)
+		}))
+	defer srv.Close()
+
+	s := New(logger.New(), metrics.New())
+	s.url = srv.URL
+	s.RegisterHandler(updateHandler{})
+	s.RegisterRawHandler(func(string) { s.Stop() })
+
+	// Start would run the stream in the background, where the test cannot
+	// wait for it. Setting cancel as Start does lets Stop cancel the stream
+	// run here instead.
+	ctx, cancel := context.WithCancel(context.Background())
+	s.cancel = cancel
+
+	err := s.stream(ctx)
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("stream returned %v, want %v", err, context.Canceled)
+	}
+
+	// A second Stop must not close the queues again.
+	s.Stop()
+}
+
+// TestReconnectWhenStreamEnds has the server end the first response
+// normally, as when RIS Live restarts. The streamer must treat that as a
+// disconnect and connect again.
+func TestReconnectWhenStreamEnds(t *testing.T) {
+	t.Parallel()
+
+	endResponse := func(http.ResponseWriter, *http.Request) {}
+	checkReconnect(t, endResponse, streamIdleTimeout)
+}
+
+// TestReconnectWhenStreamGoesIdle has the server hold the first response
+// open without sending anything. The streamer must drop it once the idle
+// timeout passes and connect again.
+func TestReconnectWhenStreamGoesIdle(t *testing.T) {
+	t.Parallel()
+
+	const idleTimeout = 50 * time.Millisecond
+
+	holdResponse := func(w http.ResponseWriter, r *http.Request) {
+		flush(w)
+		<-r.Context().Done()
+	}
+	checkReconnect(t, holdResponse, idleTimeout)
+}
+
+// TestReconnectWhenServerNeverAnswers has the server accept the first
+// request and never answer it, not even with headers. The streamer must give
+// up on it once the idle timeout passes and connect again.
+func TestReconnectWhenServerNeverAnswers(t *testing.T) {
+	t.Parallel()
+
+	const idleTimeout = 50 * time.Millisecond
+
+	neverAnswer := func(_ http.ResponseWriter, r *http.Request) {
+		<-r.Context().Done()
+	}
+	checkReconnect(t, neverAnswer, idleTimeout)
+}
+
+// TestStreamNotDroppedWhileLinesArrive has the server send lines well within
+// the idle timeout for several times the timeout, then end the response. Each
+// line must start the idle timeout over, so the stream lasts until the server
+// ends it instead of being dropped as idle.
+func TestStreamNotDroppedWhileLinesArrive(t *testing.T) {
+	t.Parallel()
+
+	const (
+		idleTimeout  = 100 * time.Millisecond
+		lineInterval = idleTimeout / 10
+		sendFor      = 5 * idleTimeout
+	)
+
+	srv := httptest.NewServer(http.HandlerFunc(
+		func(w http.ResponseWriter, _ *http.Request) {
+			deadline := time.Now().Add(sendFor)
+			for time.Now().Before(deadline) {
+				_, _ = io.WriteString(w, updateLine)
+
+				flush(w)
+				time.Sleep(lineInterval)
+			}
+		}))
+	defer srv.Close()
+
+	s := New(logger.New(), metrics.New())
+	s.url = srv.URL
+	s.idleTimeout = idleTimeout
+
+	err := s.stream(context.Background())
+	if !errors.Is(err, errStreamEnded) {
+		t.Fatalf("stream returned %v, want %v", err, errStreamEnded)
+	}
+}
+
+// checkReconnect runs a streamer against a fake RIS Live server. The first
+// response is served by firstResponse; every later one carries one UPDATE
+// line and then stays open. The streamer must come back for a second
+// response, must not be marked connected when it does, and must handle the
+// UPDATE in it.
+func checkReconnect(
+	t *testing.T, firstResponse http.HandlerFunc, idleTimeout time.Duration,
+) {
+	t.Helper()
+
+	const maxWait = 5 * time.Second
+
+	tracker := metrics.New()
+	connectedAtReconnect := make(chan bool, 1)
+
+	var requests atomic.Int32
+
+	srv := httptest.NewServer(http.HandlerFunc(
+		func(w http.ResponseWriter, r *http.Request) {
+			if requests.Add(1) == 1 {
+				firstResponse(w, r)
+
+				return
+			}
+
+			select {
+			case connectedAtReconnect <- tracker.IsConnected():
+			default:
+			}
+
+			_, _ = io.WriteString(w, updateLine)
+
+			flush(w)
+			<-r.Context().Done()
+		}))
+	defer srv.Close()
+
+	s := New(logger.New(), tracker)
+	s.url = srv.URL
+	s.idleTimeout = idleTimeout
+	s.minBackoff = time.Millisecond
+	handler := updateHandler{handled: make(chan struct{}, 1)}
+	s.RegisterHandler(handler)
+
+	err := s.Start(t.Context())
+	if err != nil {
+		t.Fatalf("Start returned error: %v", err)
+	}
+	defer s.Stop()
+
+	select {
+	case <-handler.handled:
+	case <-time.After(maxWait):
+		t.Fatalf("no message from a second response handled within %s", maxWait)
+	}
+
+	if <-connectedAtReconnect {
+		t.Error("streamer was still marked connected when it came back " +
+			"for a second response")
+	}
+}
+
+// settledGoroutineCount lets transient goroutines finish, then reports the
+// current count.
+func settledGoroutineCount() int {
+	prev := runtime.NumGoroutine()
+
+	for range 20 {
+		time.Sleep(10 * time.Millisecond)
+
+		cur := runtime.NumGoroutine()
+		if cur == prev {
+			return cur
+		}
+
+		prev = cur
+	}
+
+	return prev
+}
+
+// waitForGoroutines waits until the goroutine count drops to target or below.
+func waitForGoroutines(target int) bool {
+	for range 100 {
+		if runtime.NumGoroutine() <= target {
+			return true
+		}
+
+		time.Sleep(10 * time.Millisecond)
+	}
+
+	return false
+}