Complete fix for goroutine lifecycle race condition in Subscribe

Address comprehensive goroutine lifecycle management issue:

Problem:
The previous fix with buffered channel only prevented blocking, but didn't
prevent goroutine leaks. When the main function returns with an error or
due to context cancellation BEFORE the goroutine enters its main loop, the
goroutine would continue running indefinitely until the original context
expires (could be much later).

Complete Solution:
1. Added stop channel (chan struct{}) to signal goroutine termination
2. Goroutine checks stop channel during initialization (both error and success paths)
3. Goroutine checks stop channel as first case in main select loop
4. Main function closes stop channel when returning with error or ctx.Done()
5. Main function does NOT close stop on success - goroutine continues normally

This ensures:
- No goroutine leaks when Subscribe returns early with error
- No goroutine leaks when context is cancelled during initialization
- Clean resource cleanup via defer statements
- Goroutine runs normally when initialization succeeds

Verified with race detector running 3 consecutive times.

Author: Mat001

pkg/syncer/pubsub/redis_streams.go

@@ -100,9 +100,11 @@ func (r *RedisStreams) Subscribe(ctx context.Context, channel string) (chan stri
 
 	ch := make(chan string)
 	ready := make(chan error, 1) // Signal when consumer group is ready
+	stop := make(chan struct{})   // Signal to stop goroutine
 
 	go func() {
 		defer close(ch)
+		defer close(ready) // Ensure ready is closed
 
 		batchSize := r.getBatchSize()
 		flushTicker := time.NewTicker(r.getFlushInterval())
@@ -123,20 +125,23 @@ func (r *RedisStreams) Subscribe(ctx context.Context, channel string) (chan stri
 			log.Error().Err(err).Str("stream", streamName).Str("group", consumerGroup).Msg("Failed to create consumer group")
 			select {
 			case ready <- err: // Signal initialization failure
-			case <-ctx.Done(): // Main function already returned
+			case <-stop: // Main function signaled stop
 			}
 			return
 		}
 
 		// Signal that consumer group is ready
 		select {
 		case ready <- nil:
-		case <-ctx.Done(): // Main function already returned
+		case <-stop: // Main function signaled stop
 			return
 		}
 
 		for {
 			select {
+			case <-stop:
+				// Main function requested stop
+				return
 			case <-ctx.Done():
 				// Send any remaining batch before closing
 				if len(batch) > 0 {
@@ -239,10 +244,13 @@ func (r *RedisStreams) Subscribe(ctx context.Context, channel string) (chan stri
 	select {
 	case err := <-ready:
 		if err != nil {
+			close(stop) // Signal goroutine to stop on initialization error
 			return nil, err
 		}
+		// Success - goroutine continues running
 		return ch, nil
 	case <-ctx.Done():
+		close(stop) // Signal goroutine to stop on context cancellation
 		return nil, ctx.Err()
 	}
 }