rudder/rudder-csharp-sdk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit

Browse Source
This commit is contained in:
rudder
2026-08-12 14:04:55 +03:00
commit 8cf738d9f0
124 changed files with 5433 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Services/Scenarios/ScenarioService.Runtime.cs
+341
View File
@@ -0,0 +1,341 @@
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;
/// <summary>
/// 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.
/// </summary>
public sealed class TransientBoundaryException : Exception
{
/// <summary>Creates the exception.</summary>
public TransientBoundaryException(string message, Exception inner)
: base(message, inner)
{
}
/// <summary>Creates the exception wrapping the transport error.</summary>
public TransientBoundaryException(Exception inner)
: base($"Transient boundary error: {inner.Message}", inner)
{
}
}
public sealed partial class ScenarioService
{
/// <summary>
/// Returns true for exceptions that may succeed on retry
/// (timeout / network failures).
/// </summary>
private static bool IsTransientException(Exception ex)
{
return ex is OperationCanceledException || ex is RudderNetworkException;
}
private IReadOnlyList<PlanRun> StartPlans(IEnumerable<ExecutionPlan>? plans)
{
var started = new List<PlanRun>();
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;
}
/// <summary>
/// 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.
/// </summary>
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<bool> 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<HandleScenarioCallbackRequest, HandleScenarioCallbackResponse>(
"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<GetScenarioRunRequest, GetScenarioRunResponse>(
"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<PlanEdge> 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<BoundaryNode> 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);
}