Effects client reconciles pending effects: drop stale wait deadlines and seen keys, emit completion for vanished runs; run_not_active is terminal

Services/EffectsService.cs

@@ -21,7 +21,7 @@ public sealed class EffectsService
 
     private readonly RudderClient _client;
     private readonly object _gate = new();
-    private readonly HashSet<(string RunId, string NodeId)> _seen = new();
+    private readonly Dictionary<(string RunId, string NodeId), string> _seen = new();
     private readonly Dictionary<(string RunId, string NodeId), DateTimeOffset> _waitDeadlines = new();
 
     private bool _refreshDue;
@@ -99,8 +99,9 @@ public sealed class EffectsService
                     continue;
 
                 var key = (effect.RunId, effect.NodeId);
-                if (!_seen.Add(key))
+                if (_seen.ContainsKey(key))
                     continue;
+                _seen[key] = effect.ScenarioId;
 
                 if (string.Equals(effect.Type, EffectTypes.Wait, StringComparison.Ordinal))
                 {
@@ -215,7 +216,12 @@ public sealed class EffectsService
                 "/sdk/v1/scenarios/pending",
                 CancellationToken.None).ConfigureAwait(false);
 
-            Ingest(response?.Effects);
+            if (response != null)
+            {
+                Ingest(response.Effects);
+                Reconcile(response.Effects);
+            }
+
             lock (_gate)
                 _nextHeartbeat = _client.Clock.UtcNow + HeartbeatInterval;
         }
@@ -280,7 +286,7 @@ public sealed class EffectsService
         lock (_gate)
         {
             var toRemove = new List<(string RunId, string NodeId)>();
-            foreach (var key in _seen)
+            foreach (var key in _seen.Keys)
             {
                 if (key.RunId == runId)
                     toRemove.Add(key);
@@ -296,6 +302,45 @@ public sealed class EffectsService
         Emit(OnScenarioFailed, new ScenarioFailedEffect(runId, scenarioId, nodeId, exception));
     }
 
+    private void Reconcile(IEnumerable<PendingEffect>? effects)
+    {
+        var incoming = new HashSet<(string RunId, string NodeId)>();
+        if (effects != null)
+        {
+            foreach (var effect in effects)
+            {
+                if (effect == null || string.IsNullOrEmpty(effect.RunId) || string.IsNullOrEmpty(effect.NodeId))
+                    continue;
+
+                incoming.Add((effect.RunId, effect.NodeId));
+            }
+        }
+
+        var finished = new Dictionary<string, string>();
+        lock (_gate)
+        {
+            var stale = new List<(string RunId, string NodeId)>();
+            foreach (var key in _seen.Keys)
+            {
+                if (!incoming.Contains(key))
+                    stale.Add(key);
+            }
+
+            foreach (var key in stale)
+            {
+                finished[key.RunId] = _seen[key];
+                _seen.Remove(key);
+                _waitDeadlines.Remove(key);
+            }
+
+            foreach (var key in _seen.Keys)
+                finished.Remove(key.RunId);
+        }
+
+        foreach (var entry in finished)
+            Emit(OnScenarioCompleted, new ScenarioCompletedEffect(entry.Key, entry.Value));
+    }
+
     private void ForgetWait(string runId, string nodeId)
     {
         lock (_gate)
@@ -347,7 +392,9 @@ public sealed class EffectsService
             return true;
 
         return ex is RudderApiException api
-            && (api.Code == RudderErrorCodes.UnknownRun || api.Code == RudderErrorCodes.RunExpired);
+            && (api.Code == RudderErrorCodes.UnknownRun
+                || api.Code == RudderErrorCodes.RunExpired
+                || api.Code == RudderErrorCodes.RunNotActive);
     }
 }