routewatch/internal/streamer/streamer_internal_test.go

package streamer

import (
	"context"
	"errors"
	"io"
	"net/http"
	"net/http/httptest"
	"runtime"
	"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")
	}
}

// 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 err != nil {
		t.Fatalf("warm-up stream returned error: %v", err)
	}

	s.client.CloseIdleConnections()

	baseline := settledGoroutineCount()

	const reconnects = 20
	for range reconnects {
		err := s.stream(context.Background())
		if err != nil {
			t.Fatalf("stream returned error: %v", err)
		}
	}

	s.client.CloseIdleConnections()

	if !waitForGoroutines(baseline) {
		t.Fatalf("goroutines did not return to baseline %d after %d reconnects, got %d",
			baseline, reconnects, runtime.NumGoroutine())
	}
}

// updateHandler wants UPDATE messages and does nothing with them.
type updateHandler struct{}

func (updateHandler) WantsMessage(messageType string) bool {
	return messageType == "UPDATE"
}

func (updateHandler) HandleMessage(*ristypes.RISMessage) {}

func (updateHandler) QueueCapacity() int { return 10 }

// 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()

	const line = `{"type":"ris_message","data":{"type":"UPDATE",` +
		`"peer":"192.0.2.1","peer_asn":"64496","timestamp":1700000000}}` + "\n"

	srv := httptest.NewServer(http.HandlerFunc(
		func(w http.ResponseWriter, _ *http.Request) {
			_, _ = io.WriteString(w, line)
		}))
	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()
}

// 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
}