loop/server: fix race condition in SSE stream iterators
Fix panic 'send on closed channel' when a client disconnects from SSE stream.
The issue occurred because iterators were created at the beginning of
handleSSEStream with defer iterator.Close(), but used in separate goroutines.
When the client disconnected, the iterators were closed while goroutines
were still using them.
Move iterator creation into the goroutines that use them to ensure their
lifecycle matches the goroutines' lifecycle, preventing the race condition.
Fixes #108
diff --git a/loop/server/loophttp.go b/loop/server/loophttp.go
index 534452b..541014e 100644
--- a/loop/server/loophttp.go
+++ b/loop/server/loophttp.go
@@ -1086,14 +1086,6 @@
// Create a context for the SSE stream
ctx := r.Context()
- // Create an iterator to receive new messages as they arrive
- iterator := s.agent.NewIterator(ctx, fromIndex) // Start from the requested index
- defer iterator.Close()
-
- // Create an iterator to receive state transitions
- stateIterator := s.agent.NewStateTransitionIterator(ctx)
- defer stateIterator.Close()
-
// Setup heartbeat timer
heartbeatTicker := time.NewTicker(45 * time.Second)
defer heartbeatTicker.Stop()
@@ -1106,6 +1098,9 @@
// Start a goroutine to read messages without blocking the heartbeat
go func() {
+ // Create an iterator to receive new messages as they arrive
+ iterator := s.agent.NewIterator(ctx, fromIndex) // Start from the requested index
+ defer iterator.Close()
defer close(messageChan)
for {
// This can block, but it's in its own goroutine
@@ -1128,6 +1123,9 @@
// Start a goroutine to read state transitions
go func() {
+ // Create an iterator to receive state transitions
+ stateIterator := s.agent.NewStateTransitionIterator(ctx)
+ defer stateIterator.Close()
defer close(stateChan)
for {
// This can block, but it's in its own goroutine