Skip to content

Commit c89cec5

Browse files
lokesh755lokesh755
committed
Background steps engine
1 parent 77d6014 commit c89cec5

9 files changed

Lines changed: 1291 additions & 16 deletions

File tree

‎src/Runner.Common/JobServerQueue.cs‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -837,6 +837,15 @@ private List<TimelineRecord> MergeTimelineRecords(List<TimelineRecord> timelineR
837837
timelineRecord.Variables[variable.Key] = variable.Value.Clone();
838838
}
839839
}
840+
841+
// Merge background step metadata
842+
if (rec.IsBackground)
843+
{
844+
timelineRecord.IsBackground = rec.IsBackground;
845+
}
846+
timelineRecord.BackgroundControlType = rec.BackgroundControlType ?? timelineRecord.BackgroundControlType;
847+
timelineRecord.BackgroundControlStepIds = rec.BackgroundControlStepIds ?? timelineRecord.BackgroundControlStepIds;
848+
timelineRecord.ParallelGroupId = rec.ParallelGroupId ?? timelineRecord.ParallelGroupId;
840849
}
841850
else
842851
{

‎src/Runner.Worker/BackgroundStepContext.cs‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
using System;
2+
using System.Threading;
3+
using System.Threading.Tasks;
4+
5+
namespace GitHub.Runner.Worker
6+
{
7+
/// <summary>
8+
/// Tracks a background step's execution state.
9+
/// </summary>
10+
internal sealed class BackgroundStepContext
11+
{
12+
public string StepId { get; }
13+
public IStep Step { get; }
14+
public Task ExecutionTask { get; set; }
15+
public CancellationTokenSource Cts { get; set; }
16+
public GitHub.DistributedTask.WebApi.TaskResult? Result { get; set; }
17+
public bool IsCompleted => ExecutionTask?.IsCompleted ?? false;
18+
public string ExternalId => Step.ExecutionContext.Id.ToString("N");
19+
20+
public BackgroundStepContext(string stepId, IStep step)
21+
{
22+
StepId = stepId;
23+
Step = step;
24+
}
25+
}
26+
}

‎src/Runner.Worker/BackgroundStepCoordinator.cs‎

Lines changed: 338 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,338 @@
1+
using System;
2+
using System.Collections.Generic;
3+
using System.Linq;
4+
using System.Threading;
5+
using System.Threading.Tasks;
6+
using GitHub.DistributedTask.Pipelines.ContextData;
7+
using GitHub.DistributedTask.WebApi;
8+
using GitHub.Runner.Common;
9+
using GitHub.Runner.Common.Util;
10+
using GitHub.Runner.Sdk;
11+
12+
namespace GitHub.Runner.Worker
13+
{
14+
/// <summary>
15+
/// Coordinates background step execution, waiting, cancellation, and deferred state.
16+
/// Extracted from StepsRunner so the main step loop stays clean.
17+
/// </summary>
18+
public sealed class BackgroundStepCoordinator
19+
{
20+
private const int DefaultMaxBackgroundSteps = 10;
21+
private readonly Dictionary<string, BackgroundStepContext> _backgroundSteps = new();
22+
private readonly HashSet<string> _waitedStepIds = new();
23+
private SemaphoreSlim _backgroundSlotSemaphore;
24+
private readonly IHostContext _hostContext;
25+
private readonly ITraceWriter _trace;
26+
27+
public BackgroundStepCoordinator(IHostContext hostContext, ITraceWriter trace)
28+
{
29+
_hostContext = hostContext;
30+
_trace = trace;
31+
_backgroundSlotSemaphore = new SemaphoreSlim(DefaultMaxBackgroundSteps);
32+
}
33+
34+
/// <summary>
35+
/// Reset per-job state. Call at the start of each job.
36+
/// </summary>
37+
public void Initialize(int maxConcurrent)
38+
{
39+
_backgroundSteps.Clear();
40+
_waitedStepIds.Clear();
41+
var max = maxConcurrent > 0 ? maxConcurrent : DefaultMaxBackgroundSteps;
42+
_backgroundSlotSemaphore = new SemaphoreSlim(max);
43+
}
44+
45+
public bool HasActiveBackgroundSteps => _backgroundSteps.Count > 0;
46+
47+
public bool HasUnwaitedSteps => _backgroundSteps.Keys.Any(id => !_waitedStepIds.Contains(id));
48+
49+
public IReadOnlyList<string> UnwaitedStepIds =>
50+
_backgroundSteps.Keys.Where(id => !_waitedStepIds.Contains(id)).ToList();
51+
52+
// -----------------------------------------------------------------
53+
// Starting background steps
54+
// -----------------------------------------------------------------
55+
56+
/// <summary>
57+
/// Prepare and launch a background step. Does not block the caller.
58+
/// </summary>
59+
public void StartBackgroundStep(IStep step, CancellationToken jobCancellationToken)
60+
{
61+
var stepId = step.ExecutionContext?.ContextName ?? step.DisplayName;
62+
63+
// Isolate GitHubContext so concurrent steps don't overwrite each other's GITHUB_OUTPUT paths
64+
if (step.ExecutionContext.ExpressionValues.TryGetValue("github", out var ghCtx) && ghCtx is GitHubContext sharedGitHub)
65+
{
66+
step.ExecutionContext.ExpressionValues["github"] = sharedGitHub.ShallowCopy();
67+
}
68+
69+
// Defer outputs, env, path, outcome — flushed at wait/wait-all
70+
step.ExecutionContext.DeferredOutputs = new Dictionary<string, string>();
71+
step.ExecutionContext.DeferredEnvironmentVariables = new Dictionary<string, string>(StringComparer.OrdinalIgnoreCase);
72+
step.ExecutionContext.DeferredPrependPath = new List<string>();
73+
step.ExecutionContext.DeferOutcomeConclusion = true;
74+
75+
var bgCts = CancellationTokenSource.CreateLinkedTokenSource(jobCancellationToken);
76+
77+
// Evaluate timeout on the main thread (needs expression context)
78+
var timeoutMinutes = 0;
79+
try
80+
{
81+
var templateEvaluator = step.ExecutionContext.ToPipelineTemplateEvaluator();
82+
timeoutMinutes = templateEvaluator.EvaluateStepTimeout(step.Timeout, step.ExecutionContext.ExpressionValues, step.ExecutionContext.ExpressionFunctions);
83+
}
84+
catch (Exception ex)
85+
{
86+
_trace.Info($"Error determining timeout for background step '{stepId}': {ex.Message}");
87+
}
88+
89+
var bgContext = new BackgroundStepContext(stepId, step) { Cts = bgCts };
90+
bgContext.ExecutionTask = ExecuteBackgroundStepCoreAsync(bgContext, step, bgCts, stepId, timeoutMinutes);
91+
92+
_backgroundSteps[stepId] = bgContext;
93+
_trace.Info($"Background step '{stepId}' queued (slot will be acquired asynchronously).");
94+
}
95+
96+
private async Task ExecuteBackgroundStepCoreAsync(
97+
BackgroundStepContext bgContext, IStep step, CancellationTokenSource bgCts,
98+
string stepId, int timeoutMinutes)
99+
{
100+
_trace.Info($"Background step '{stepId}' waiting for slot.");
101+
await _backgroundSlotSemaphore.WaitAsync(bgCts.Token);
102+
_trace.Info($"Background step '{stepId}' acquired slot.");
103+
104+
step.ExecutionContext.Start();
105+
106+
if (timeoutMinutes > 0)
107+
{
108+
step.ExecutionContext.SetTimeout(TimeSpan.FromMinutes(timeoutMinutes));
109+
}
110+
111+
// When the linked CTS fires (job cancellation or explicit cancel),
112+
// immediately signal the step's process to terminate (SIGTERM).
113+
// Without this, the process continues until the task framework
114+
// propagates the cancellation, causing a noticeable delay.
115+
using var cancelReg = bgCts.Token.Register(() =>
116+
{
117+
_trace.Info($"Background step '{stepId}': cancellation signalled, sending CancelToken to process.");
118+
step.ExecutionContext.CancelToken();
119+
});
120+
121+
try
122+
{
123+
await step.RunAsync();
124+
bgContext.Result = step.ExecutionContext.Result ?? TaskResult.Succeeded;
125+
}
126+
catch (OperationCanceledException) when (bgCts.Token.IsCancellationRequested)
127+
{
128+
bgContext.Result = TaskResult.Canceled;
129+
}
130+
catch (OperationCanceledException) when (step.ExecutionContext.CancellationToken.IsCancellationRequested)
131+
{
132+
_trace.Info($"Background step '{stepId}' timed out after {timeoutMinutes} minutes.");
133+
step.ExecutionContext.Error($"The background step '{step.DisplayName}' has timed out after {timeoutMinutes} minutes.");
134+
bgContext.Result = TaskResult.Failed;
135+
}
136+
catch (Exception ex)
137+
{
138+
_trace.Info($"Background step '{stepId}' failed: {ex.Message}");
139+
step.ExecutionContext.Error(ex);
140+
bgContext.Result = TaskResult.Failed;
141+
}
142+
finally
143+
{
144+
_backgroundSlotSemaphore.Release();
145+
146+
if (step.ExecutionContext.CommandResult != null)
147+
{
148+
bgContext.Result = TaskResultUtil.MergeTaskResults(
149+
bgContext.Result, step.ExecutionContext.CommandResult.Value);
150+
}
151+
152+
step.ExecutionContext.Result = bgContext.Result;
153+
step.ExecutionContext.ApplyContinueOnError(step.ContinueOnError);
154+
bgContext.Result = step.ExecutionContext.Result;
155+
156+
step.ExecutionContext.Complete(bgContext.Result);
157+
_trace.Info($"Background step '{stepId}' completed with result: {bgContext.Result}");
158+
}
159+
}
160+
161+
// -----------------------------------------------------------------
162+
// Wait
163+
// -----------------------------------------------------------------
164+
165+
/// <summary>
166+
/// Wait for specific background steps by ID. Flushes deferred state and
167+
/// returns the aggregate result.
168+
/// </summary>
169+
public async Task<TaskResult> WaitForStepsAsync(string[] stepIds, CancellationToken cancellationToken)
170+
{
171+
var ids = stepIds ?? Array.Empty<string>();
172+
173+
await WaitForStepTasksAsync(ids, cancellationToken);
174+
175+
foreach (var id in ids)
176+
{
177+
FlushDeferredState(id);
178+
_waitedStepIds.Add(id);
179+
}
180+
181+
return GetWaitResult(ids);
182+
}
183+
184+
/// <summary>
185+
/// Wait for all not-yet-waited background steps. Returns the aggregate result.
186+
/// </summary>
187+
public async Task<TaskResult> WaitForAllAsync(CancellationToken cancellationToken)
188+
{
189+
var remaining = _backgroundSteps.Keys.Where(id => !_waitedStepIds.Contains(id)).ToList();
190+
191+
await WaitForStepTasksAsync(remaining, cancellationToken);
192+
193+
foreach (var id in remaining)
194+
{
195+
FlushDeferredState(id);
196+
_waitedStepIds.Add(id);
197+
}
198+
199+
return GetWaitResult(remaining);
200+
}
201+
202+
// -----------------------------------------------------------------
203+
// Cancel
204+
// -----------------------------------------------------------------
205+
206+
/// <summary>
207+
/// Cancel specific background steps by ID. Flushes deferred state after cancellation.
208+
/// </summary>
209+
public async Task CancelStepsAsync(string[] cancelStepIds)
210+
{
211+
if (cancelStepIds == null || cancelStepIds.Length == 0) return;
212+
213+
var idsToCancel = cancelStepIds
214+
.Where(id => _backgroundSteps.ContainsKey(id) && !_backgroundSteps[id].IsCompleted)
215+
.ToArray();
216+
217+
if (idsToCancel.Length > 0)
218+
{
219+
_trace.Info($"Cancelling {idsToCancel.Length} background step(s): {string.Join(", ", idsToCancel)}");
220+
await CancelWithGracePeriodAsync(idsToCancel);
221+
}
222+
223+
foreach (var id in cancelStepIds)
224+
{
225+
FlushDeferredState(id);
226+
}
227+
}
228+
229+
// -----------------------------------------------------------------
230+
// Failure propagation
231+
// -----------------------------------------------------------------
232+
233+
/// <summary>
234+
/// Propagate any unhandled background step failures to the job result.
235+
/// </summary>
236+
public void PropagateFailures(IExecutionContext jobContext)
237+
{
238+
foreach (var bgCtx in _backgroundSteps.Values)
239+
{
240+
if (bgCtx.Result == TaskResult.Failed)
241+
{
242+
_trace.Info($"Propagating failure from background step '{bgCtx.StepId}' to job result.");
243+
jobContext.Result = TaskResultUtil.MergeTaskResults(jobContext.Result, TaskResult.Failed);
244+
jobContext.JobContext.Status = jobContext.Result?.ToActionResult();
245+
break;
246+
}
247+
}
248+
}
249+
250+
// -----------------------------------------------------------------
251+
// Private helpers
252+
// -----------------------------------------------------------------
253+
254+
private async Task WaitForStepTasksAsync(IEnumerable<string> stepIds, CancellationToken cancellationToken)
255+
{
256+
var ids = stepIds.ToList();
257+
var tasks = new List<Task>();
258+
259+
foreach (var stepId in ids)
260+
{
261+
if (_backgroundSteps.TryGetValue(stepId, out var bgCtx) && !bgCtx.IsCompleted)
262+
{
263+
tasks.Add(bgCtx.ExecutionTask);
264+
}
265+
else if (!_backgroundSteps.ContainsKey(stepId))
266+
{
267+
_trace.Info($"Wait references unknown background step: {stepId}");
268+
}
269+
}
270+
271+
if (tasks.Count > 0)
272+
{
273+
_trace.Info($"Waiting for {tasks.Count} background step(s)...");
274+
try
275+
{
276+
await Task.WhenAll(tasks).WaitAsync(cancellationToken);
277+
}
278+
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
279+
{
280+
_trace.Info("Wait interrupted by job cancellation — cancelling background steps.");
281+
await CancelWithGracePeriodAsync(ids);
282+
}
283+
}
284+
}
285+
286+
private async Task CancelWithGracePeriodAsync(IEnumerable<string> stepIds, double graceSeconds = 7.5)
287+
{
288+
var tasks = new List<Task>();
289+
foreach (var stepId in stepIds)
290+
{
291+
if (_backgroundSteps.TryGetValue(stepId, out var bgCtx) && !bgCtx.IsCompleted)
292+
{
293+
bgCtx.Step.ExecutionContext.CancelToken();
294+
bgCtx.Cts.Cancel();
295+
tasks.Add(bgCtx.ExecutionTask);
296+
}
297+
}
298+
299+
if (tasks.Count > 0)
300+
{
301+
try
302+
{
303+
await Task.WhenAll(tasks).WaitAsync(TimeSpan.FromSeconds(graceSeconds));
304+
}
305+
catch (TimeoutException)
306+
{
307+
_trace.Info($"Some background steps did not terminate within {graceSeconds}s grace period.");
308+
}
309+
}
310+
}
311+
312+
private void FlushDeferredState(string stepId)
313+
{
314+
if (_backgroundSteps.TryGetValue(stepId, out var bgCtx))
315+
{
316+
bgCtx.Step.ExecutionContext.FlushDeferredOutputs();
317+
bgCtx.Step.ExecutionContext.FlushDeferredEnvironment();
318+
bgCtx.Step.ExecutionContext.FlushDeferredOutcomeConclusion();
319+
_trace.Info($"Flushed deferred state for background step '{stepId}'.");
320+
}
321+
}
322+
323+
private TaskResult GetWaitResult(IEnumerable<string> stepIds)
324+
{
325+
if (stepIds == null) return TaskResult.Succeeded;
326+
327+
foreach (var stepId in stepIds)
328+
{
329+
if (_backgroundSteps.TryGetValue(stepId, out var bgCtx) && bgCtx.Result == TaskResult.Failed)
330+
{
331+
_trace.Info($"Background step '{stepId}' failed.");
332+
return TaskResult.Failed;
333+
}
334+
}
335+
return TaskResult.Succeeded;
336+
}
337+
}
338+
}

0 commit comments

Comments
 (0)