using System; using System.Collections.Generic; using System.Linq; using System.Threading; using System.Threading.Tasks; using RudderSdk.Core.Abstractions; using RudderSdk.Core.Models; using RudderSdk.Core.Models.Scenarios; namespace RudderSdk.Core; /// /// Thrown when a boundary HTTP call fails with a transient error /// (network failure, timeout, or server 5xx). The caller should NOT advance /// the run; the node stays active and the handle stays pending for retry. /// public sealed class TransientBoundaryException : Exception { /// Creates the exception. public TransientBoundaryException(string message, Exception inner) : base(message, inner) { } /// Creates the exception wrapping the transport error. public TransientBoundaryException(Exception inner) : base($"Transient boundary error: {inner.Message}", inner) { } } public sealed partial class ScenarioService { /// /// Returns true for exceptions that may succeed on retry /// (timeout / network failures). /// private static bool IsTransientException(Exception ex) { return ex is OperationCanceledException || ex is RudderNetworkException; } private IReadOnlyList StartPlans(IEnumerable? plans) { var started = new List(); if (plans == null) return started; foreach (var plan in plans) { var run = StartPlan(plan); if (run != null) started.Add(ToPlanRun(run)); } return started; } private RuntimeRun? StartPlan(ExecutionPlan? plan) { if (plan?.Nodes == null || plan.Nodes.Count == 0) return null; // Dedup: server returned same runId — skip without restarting. if (!string.IsNullOrEmpty(plan.RunId) && _runs.ContainsKey(plan.RunId)) return null; if (plan.BoundaryNodes?.Count > 0 && string.IsNullOrEmpty(plan.RunId)) throw new InvalidOperationException("ExecutionPlan has boundaryNodes but missing RunId."); var runId = plan.RunId ?? Guid.NewGuid().ToString("N"); var startNode = FindNode(plan, plan.StartNodeId) ?? plan.Nodes[0]; var run = new RuntimeRun(runId, plan); _runs[run.RunId] = run; ActivateNode(run, startNode.Id); Persist(); return run; } /// /// Replaces an existing run's plan with a server-provided continuation (same RunId): /// the previous segment is done, the new segment's start node becomes active. /// Idempotent: if the continuation was already applied, nothing is re-dispatched. /// private void ReplaceRun(ExecutionPlan? plan, string? fallbackRunId = null) { if (plan?.Nodes == null || plan.Nodes.Count == 0) return; var runId = plan.RunId ?? fallbackRunId; if (string.IsNullOrEmpty(runId)) return; var startNode = FindNode(plan, plan.StartNodeId) ?? plan.Nodes[0]; if (startNode == null) return; if (_runs.TryGetValue(runId, out var existing) && existing.ActiveNodes.ContainsKey(startNode.Id)) return; // continuation already applied (idempotent callback echo) var run = new RuntimeRun(runId, plan); _runs[runId] = run; ActivateNode(run, startNode.Id); Persist(); } private void ActivateNode(RuntimeRun run, string nodeId, ActiveNodeState? restoredState = null) { var node = FindNode(run.Plan, nodeId); if (node == null) return; var state = restoredState ?? new ActiveNodeState { NodeId = nodeId }; run.ActiveNodes[nodeId] = state; DispatchActiveNode(run, state, restored: restoredState != null); } internal Task CompleteNodeAsync( string runId, string nodeId, string handle, CancellationToken cancellationToken = default) { return CompleteNodeAsync(runId, nodeId, handle, continueOnBoundary: true, cancellationToken); } private async Task CompleteNodeAsync( string runId, string nodeId, string handle, bool continueOnBoundary, CancellationToken cancellationToken = default) { if (!_runs.TryGetValue(runId, out var run)) return; var key = CompletedHandleKey(nodeId, handle); if (run.CompletedHandles.Contains(key)) return; // idempotent try { var continuedOnBoundary = false; if (continueOnBoundary) { try { continuedOnBoundary = await ContinueBoundaryAsync( run, nodeId, handle, cancellationToken).ConfigureAwait(false); } catch (TransientBoundaryException) { // Transient error — don't advance the run. // Node stays active, handle stays pending for retry on reconnect. return; } } if (continuedOnBoundary) { // The boundary may have replaced or removed the run (continuation plan // or reconcile). The transferred run owns the state — touching the stale // object here would complete or delete the new run. if (!_runs.TryGetValue(runId, out var currentRun) || !ReferenceEquals(currentRun, run)) return; } // Only now — after the server has confirmed — mark the handle and node. run.CompletedHandles.Add(key); run.ActiveNodes.Remove(nodeId); Persist(); if (!continuedOnBoundary) { foreach (var edge in MatchingEdges(run.Plan, nodeId, handle)) ActivateNode(run, edge.Target); } CheckRunCompleted(run); Persist(); } catch (Exception ex) { FailRun(run, nodeId, ex); } } private async Task ContinueBoundaryAsync( RuntimeRun run, string nodeId, string handle, CancellationToken cancellationToken) { var boundaries = MatchingBoundaryNodes(run.Plan, nodeId, handle).ToList(); if (boundaries.Count == 0) return false; foreach (var boundary in boundaries) { try { var response = await _client.SendAsync( "POST", "/sdk/v1/scenarios/callback", new HandleScenarioCallbackRequest { ScenarioId = run.Plan.ScenarioId, NodeId = boundary.SourceNodeId, Handle = boundary.SourceHandle, RunId = run.RunId }, cancellationToken).ConfigureAwait(false); if (response?.Plan != null) { // Continuation of the current run (server keeps the RunId) replaces // the run's plan; fresh/terminal plans start as new runs. if (_runs.ContainsKey(response.Plan.RunId ?? string.Empty)) ReplaceRun(response.Plan); else StartPlan(response.Plan); } } catch (Exception ex) { // Boundary call failed — try to reconcile with server. var reconciled = false; try { var reconcile = await _client.SendAsync( "POST", "/sdk/v1/scenarios/run", new GetScenarioRunRequest { RunId = run.RunId }, cancellationToken).ConfigureAwait(false); if (reconcile?.Status == "unknown_run" || reconcile?.Status == "expired") { _runs.Remove(run.RunId); Persist(); reconciled = true; } if (reconcile?.Plan != null) { ReplaceRun(reconcile.Plan, run.RunId); reconciled = true; } } catch { // Reconciliation also failed. } if (!reconciled) { // Reconcile did not resolve — distinguish transient from terminal. if (IsTransientException(ex)) throw new TransientBoundaryException(ex); // Terminal error — let the caller fail the run. throw; } // If reconciled, the boundary was handled (run corrected or removed). // Fall through to continue to the next boundary. } } return true; } private void CheckRunCompleted(RuntimeRun run) { if (run.ActiveNodes.Count > 0) return; _runs.Remove(run.RunId); Persist(); OnScenarioCompleted?.Invoke(ToPlanRun(run)); } private void FailRun(RuntimeRun run, string nodeId, Exception ex) { _client.Options.Logger?.Log(RudderLogLevel.Error, $"[Rudder] Scenario run {run.RunId} failed at node {nodeId}: {ex.Message}"); _runs.Remove(run.RunId); Persist(); OnScenarioFailed?.Invoke(new ScenarioFailedEvent(ToPlanRun(run), nodeId, ex)); } private static PlanRun ToPlanRun(RuntimeRun run) { return new PlanRun( run.RunId, run.Plan.PlanId, run.Plan.ScenarioId, run.Plan.UserId, run.ActiveNodes.Keys.ToList(), run.Plan); } private static ExecutionPlanNode? FindNode(ExecutionPlan? plan, string? nodeId) { if (plan?.Nodes == null || plan.Nodes.Count == 0) return null; if (!string.IsNullOrEmpty(nodeId)) { foreach (var node in plan.Nodes) { if (node.Id == nodeId) return node; } } return null; } private static IEnumerable MatchingEdges(ExecutionPlan plan, string sourceNodeId, string sourceHandle) { if (plan?.Edges == null) yield break; foreach (var edge in plan.Edges) { if (edge.Source == sourceNodeId && string.Equals(edge.SourceHandle ?? string.Empty, sourceHandle ?? string.Empty, StringComparison.Ordinal)) yield return edge; } } private static IEnumerable MatchingBoundaryNodes(ExecutionPlan plan, string sourceNodeId, string sourceHandle) { if (plan?.BoundaryNodes == null) yield break; foreach (var boundary in plan.BoundaryNodes) { if (boundary.SourceNodeId == sourceNodeId && string.Equals(boundary.SourceHandle ?? string.Empty, sourceHandle ?? string.Empty, StringComparison.Ordinal)) yield return boundary; } } private static string CompletedHandleKey(string nodeId, string? handle) => nodeId + ":" + (handle ?? string.Empty); }