agent: worker.Subscribe: start at most one log stream per task

Track task IDs for which a subscription has already started a log
stream.

A task may be present in the initial snapshot and also be delivered by
the watcher after it is registered. Use a shared helper for both paths
to ensure that only one log stream is started for each task.

Signed-off-by: Sebastiaan van Stijn <github@gone.nl>

Author: thaJeztah

agent/worker.go

@@ -641,11 +641,24 @@ func (w *worker) Subscribe(ctx context.Context, subscription *api.SubscriptionMe
 	}
 	w.mu.RUnlock()
 	var wg sync.WaitGroup
-	for _, tm := range taskManagers {
+
+	// A task may be present in the initial snapshot and also be delivered by
+	// the watcher. Start at most one log stream per task for this subscription.
+	started := make(map[string]struct{})
+	startLogs := func(tm *taskManager) {
+		taskID := tm.task.ID
+		if _, ok := started[taskID]; ok {
+			return
+		}
+		started[taskID] = struct{}{}
+
 		wg.Go(func() {
 			tm.Logs(ctx, options, publisher)
 		})
 	}
+	for _, tm := range taskManagers {
+		startLogs(tm)
+	}
 
 	// In follow mode, watch for new matching tasks until the subscription
 	// context is cancelled.
@@ -660,9 +673,7 @@ func (w *worker) Subscribe(ctx context.Context, subscription *api.SubscriptionMe
 				continue
 			}
 
-			wg.Go(func() {
-				tm.Logs(ctx, options, publisher)
-			})
+			startLogs(tm)
 		}
 	}