Stop the streamer without sending on closed queues (closes #34)

Stopping the daemon while the RIS Live feed was flowing could panic with "send on closed channel" and skip the rest of the shutdown. Stop cancels the stream and closes the handler queues under the streamer's write lock, but the read loop checked for a stop only before parsing each line. It now checks again under the read lock it already takes just before handing a message to the queues, so it never sends to a closed queue. Stop also clears its cancel function and returns early when there is none, so a second call no longer closes the queues again. A test forces both cases.

Behaviour change: Stop before Start now does nothing.

Model: opus-5-5

internal/streamer/streamer_test.go

@@ -2,6 +2,8 @@ package streamer
 
 import (
 	"context"
+	"errors"
+	"io"
 	"net/http"
 	"net/http/httptest"
 	"runtime"
@@ -10,6 +12,7 @@ import (
 
 	"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) {
@@ -75,6 +78,45 @@ func TestStreamDoesNotLeakTickersAcrossReconnects(t *testing.T) {
 	}
 }
 
+// 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) {
+	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
+
+	if err := s.stream(ctx); !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 {