From a9a9cab82e16f7699fce32c20911d1dd6537987d Mon Sep 17 00:00:00 2001 From: Zacgoose <107489668+Zacgoose@users.noreply.github.com> Date: Sat, 10 Oct 2026 01:14:09 +0800 Subject: [PATCH] fix(powershell): abandon and replace a worker whose pipeline cannot be stopped - Stop a cancelled pipeline from its own thread and give it Worker:StopGraceSeconds (default 30) to unwind - A pipeline still running after the grace is abandoned: the caller gets its timeout and the pool replaces the worker - A sequential run swaps out a pinned worker that hung - /healthz reports workers.hung and workers.hungTotal - e2e: hung-worker section (run-e2e.ps1 -HungOnly) driving PerfHang and PerfBgHang --- Services/Configuration/WorkerSettings.cs | 8 + Services/Hosting/Endpoints/HealthEndpoint.cs | 1 + Services/Orchestration/OrchestratorService.cs | 8 + Services/PowerShellHost/PowerShellWorker.cs | 170 +++++++++--- .../PowerShellHost/PowerShellWorkerPool.cs | 113 +++++--- .../PowerShellHost/WorkerHungException.cs | 12 + appsettings.example.jsonc | 5 + docs/configuration.md | 5 + .../API/Modules/PerfApi/PerfApi.psd1 | 2 +- .../API/Modules/PerfApi/PerfApi.psm1 | 44 ++++ perf-harness/scripts/run-e2e-hung-workers.ps1 | 92 +++++++ perf-harness/scripts/run-e2e.ps1 | 15 +- tests/Craft.Tests/HungWorkerRecycleTests.cs | 249 ++++++++++++++++++ tests/Craft.Tests/OrchestrationHarness.cs | 19 +- .../CraftPoolFixture/CraftPoolFixture.psd1 | 2 +- .../CraftPoolFixture/CraftPoolFixture.psm1 | 7 + 16 files changed, 665 insertions(+), 87 deletions(-) create mode 100644 Services/PowerShellHost/WorkerHungException.cs create mode 100644 perf-harness/scripts/run-e2e-hung-workers.ps1 create mode 100644 tests/Craft.Tests/HungWorkerRecycleTests.cs diff --git a/Services/Configuration/WorkerSettings.cs b/Services/Configuration/WorkerSettings.cs index f7299e0..e6c9ca3 100644 --- a/Services/Configuration/WorkerSettings.cs +++ b/Services/Configuration/WorkerSettings.cs @@ -92,6 +92,14 @@ public class WorkerSettings /// public int BgTimeoutSeconds { get; set; } + /// + /// How long a timed-out or cancelled invocation gets to unwind after PowerShell is asked to stop. + /// A pipeline blocked inside a .NET call (a lock, a sync wait, a read with no timeout) never sees a + /// stop request; past this grace its worker is abandoned and replaced, so the pool does not shrink + /// to nothing. Values below 1 use the default of 30 seconds. + /// + public int StopGraceSeconds { get; set; } = 30; + /// /// How long an incoming HTTP request waits for a free PowerShell runspace when every worker in the /// HTTP pool is already busy, before it is shed with 503 "Server busy, please retry". diff --git a/Services/Hosting/Endpoints/HealthEndpoint.cs b/Services/Hosting/Endpoints/HealthEndpoint.cs index 67b7295..bf23b8a 100644 --- a/Services/Hosting/Endpoints/HealthEndpoint.cs +++ b/Services/Hosting/Endpoints/HealthEndpoint.cs @@ -48,6 +48,7 @@ public static WebApplication MapCraftHealthEndpoint( status = (httpReady && bgReady && storageReady) ? "ready" : "starting", roles = new { frontend = roles.Frontend, http = roles.Http, background = roles.Background }, ready = new { http = httpReady, background = bgReady, storage = storageReady }, + workers = new { hung = pool.HungWorkers, hungTotal = pool.HungWorkersTotal }, }); }); diff --git a/Services/Orchestration/OrchestratorService.cs b/Services/Orchestration/OrchestratorService.cs index 838dec4..e5507e9 100644 --- a/Services/Orchestration/OrchestratorService.cs +++ b/Services/Orchestration/OrchestratorService.cs @@ -527,6 +527,14 @@ void StepDone(string status, string? error, WorkStore.ClaimedTask? next) => break; } + if (worker is { IsHung: true }) + { + // A step timed out on a worker that could not be stopped. Hand it back for replacement. + ReclaimSequentialWorker(worker, faulted: true); + worker = null; + worker = CheckoutSequentialWorker(jobCt); + } + var parameters = nextPayload ?? await _store.GetPayloadAsync(header.RunKey, step.Seq, jobCt); nextPayload = null; WorkStore.Finish finish; diff --git a/Services/PowerShellHost/PowerShellWorker.cs b/Services/PowerShellHost/PowerShellWorker.cs index f57ba9c..a840f06 100644 --- a/Services/PowerShellHost/PowerShellWorker.cs +++ b/Services/PowerShellHost/PowerShellWorker.cs @@ -315,12 +315,14 @@ public void RestoreExportedModuleNames(ExportedModuleState state) /// When a cancellation token is provided and fires, the PowerShell pipeline is /// stopped via . The resulting PipelineStoppedException /// is normalized to an so timeout callers can - /// distinguish a cancelled request from a genuine script failure. + /// distinguish a cancelled request from a genuine script failure. A pipeline that does not + /// stop within throws and leaves + /// this worker . /// public async Task> InvokeAsync(string functionName, Dictionary parameters, CancellationToken ct = default) { - CancellationTokenRegistration? registration = null; + ThrowIfHung(functionName); var prof = DispatchProfiler.Enabled; long buildTicks = 0, runTicks = 0, copyTicks = 0; try @@ -330,19 +332,10 @@ public async Task> InvokeAsync(string functionName, Diction _pwsh.AddCommand(functionName); foreach (var p in parameters) _pwsh.AddParameter(p.Key, p.Value); - - // Register cancellation callback to stop the PS pipeline - if (ct.CanBeCanceled) - registration = ct.Register(() => _pwsh.Stop()); if (prof) buildTicks = System.Diagnostics.Stopwatch.GetTimestamp() - bStart; var rStart = prof ? System.Diagnostics.Stopwatch.GetTimestamp() : 0; - // A fresh output collection per invocation: after an invocation that threw, the PowerShell object's - // own output buffer comes back null from the next EndInvoke, silently dropping that call's output. - using var outputs = new PSDataCollection(); - var asyncResult = _pwsh.BeginInvoke(null, outputs); - await Task.Factory.FromAsync(asyncResult, _pwsh.EndInvoke); - ct.ThrowIfCancellationRequested(); + using var outputs = await RunPipelineAsync(functionName, ct); if (prof) runTicks = System.Diagnostics.Stopwatch.GetTimestamp() - rStart; var cpStart = prof ? System.Diagnostics.Stopwatch.GetTimestamp() : 0; @@ -352,18 +345,21 @@ public async Task> InvokeAsync(string functionName, Diction } catch (PipelineStoppedException) when (ct.IsCancellationRequested) { - // ct.Register(_pwsh.Stop) stopped the pipeline mid-invoke, so EndInvoke threw - // PipelineStoppedException rather than reaching ThrowIfCancellationRequested below. - // Normalize to OperationCanceledException so timeout callers return 504, not 500. + // The stop landed mid-invoke, so EndInvoke threw PipelineStoppedException rather than + // reaching ThrowIfCancellationRequested. Normalize to OperationCanceledException so + // timeout callers return 504, not 500. throw new OperationCanceledException(ct); } finally { - registration?.Dispose(); - var cStart = prof ? System.Diagnostics.Stopwatch.GetTimestamp() : 0; - Cleanup(); - if (prof) DispatchProfiler.RecordInvokeDetail(buildTicks, runTicks, copyTicks, - System.Diagnostics.Stopwatch.GetTimestamp() - cStart); + // A hung worker's pipeline is still running, and the worker never runs again. + if (!IsHung) + { + var cStart = prof ? System.Diagnostics.Stopwatch.GetTimestamp() : 0; + Cleanup(); + if (prof) DispatchProfiler.RecordInvokeDetail(buildTicks, runTicks, copyTicks, + System.Diagnostics.Stopwatch.GetTimestamp() - cStart); + } } } @@ -373,7 +369,7 @@ public async Task> InvokeAsync(string functionName, Diction public async Task> InvokeScriptAsync(ScriptBlock scriptBlock, Dictionary? parameters = null, CancellationToken ct = default) { - CancellationTokenRegistration? registration = null; + ThrowIfHung("script"); try { StampOperationContext(); @@ -382,30 +378,117 @@ public async Task> InvokeScriptAsync(ScriptBlock scriptBloc foreach (var p in parameters) _pwsh.AddParameter(p.Key, p.Value); - if (ct.CanBeCanceled) - registration = ct.Register(() => _pwsh.Stop()); - - // A fresh output collection per invocation; see InvokeAsync. - using var outputs = new PSDataCollection(); - var asyncResult = _pwsh.BeginInvoke(null, outputs); - await Task.Factory.FromAsync(asyncResult, _pwsh.EndInvoke); - - ct.ThrowIfCancellationRequested(); + using var outputs = await RunPipelineAsync("script", ct); return new Collection(outputs.ReadAll()); } catch (PipelineStoppedException) when (ct.IsCancellationRequested) { - // See InvokeAsync: a cancellation-triggered _pwsh.Stop() surfaces as - // PipelineStoppedException; normalize it to OperationCanceledException. + // See InvokeAsync. throw new OperationCanceledException(ct); } finally { - registration?.Dispose(); - Cleanup(); + if (!IsHung) Cleanup(); } } + /// + /// How long a cancelled invocation gets to unwind after the stop request before the worker is + /// abandoned. See . + /// + public TimeSpan StopGrace { get; set; } = TimeSpan.FromSeconds(30); + + /// + /// True once an invocation failed to stop within . Its pipeline thread is + /// blocked in a call PowerShell cannot interrupt; the worker never runs again and the pool replaces it. + /// + public bool IsHung => _hungRun != null; + + /// The abandoned pipeline of a hung worker; completes if the blocked call ever returns. + internal Task? HungRun => _hungRun; + + private volatile Task? _hungRun; + private int _disposed; + + private void ThrowIfHung(string function) + { + if (IsHung) + throw new WorkerHungException(Id, $"Worker W{Id} is hung and cannot run {function}"); + } + + /// + /// Run the queued commands. On cancellation the stop runs on a thread of its own, because + /// blocks until the pipeline unwinds — which a pipeline blocked in .NET + /// never does. Called from the cancellation callback, as it used to be, it never returned, so the + /// timeout was never reported and the worker never came back to the pool. + /// + private async Task> RunPipelineAsync(string label, CancellationToken ct) + { + // A fresh output collection per invocation: after an invocation that threw, the PowerShell object's + // own output buffer comes back null from the next EndInvoke, silently dropping that call's output. + var outputs = new PSDataCollection(); + var run = Task.Factory.FromAsync(_pwsh.BeginInvoke(null, outputs), _pwsh.EndInvoke); + Task? stop = null; + try + { + if (ct.CanBeCanceled && !run.IsCompleted) + { + var cancelled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using (ct.Register(() => cancelled.TrySetResult())) + await Task.WhenAny(run, cancelled.Task); + + if (!run.IsCompleted) + { + stop = Task.Factory.StartNew(StopPipeline, CancellationToken.None, + TaskCreationOptions.LongRunning, TaskScheduler.Default); + if (await Task.WhenAny(run, Task.Delay(StopGrace, CancellationToken.None)) != run) + { + Abandon(run, outputs, label); + throw new WorkerHungException(Id, + $"Worker W{Id} running {label} did not stop within {StopGrace.TotalSeconds:0}s of being cancelled and was abandoned"); + } + } + } + + await run; + ct.ThrowIfCancellationRequested(); + return outputs; + } + catch (Exception ex) when (ex is not WorkerHungException) + { + outputs.Dispose(); + throw; + } + finally + { + // The pipeline has finished, so a stop still in flight returns at once. Wait for it, or it could + // land on the cleanup pipelines that run next. + if (stop != null && !IsHung) await stop; + } + } + + private void StopPipeline() + { + try { _pwsh.Stop(); } + catch (Exception ex) { _logger.LogDebug(ex, "Worker{Id}: pipeline stop threw", Id); } + } + + private void Abandon(Task run, PSDataCollection outputs, string label) + { + var abandonedAt = System.Diagnostics.Stopwatch.GetTimestamp(); + _hungRun = run; + _logger.LogError( + "[Pool] W{Id} {Function} did not stop within {Grace:0}s of being cancelled: it is blocked in a call " + + "PowerShell cannot interrupt. Abandoning the worker.", Id, label, StopGrace.TotalSeconds); + _ = run.ContinueWith(t => + { + _ = t.Exception; + outputs.Dispose(); + _logger.LogWarning("[Pool] Abandoned W{Id} {Function} returned {Seconds:0}s after it was abandoned", + Id, label, System.Diagnostics.Stopwatch.GetElapsedTime(abandonedAt).TotalSeconds); + }, CancellationToken.None, TaskContinuationOptions.None, TaskScheduler.Default); + } + public PSDataStreams Streams => _pwsh.Streams; /// @@ -656,6 +739,22 @@ public ExportedModuleState ExportModuleState() public void Dispose() { + // Nothing here owns unmanaged resources directly, but suppressing finalization keeps a + // derived type that adds a finalizer from having to re-implement IDisposable to do it. + GC.SuppressFinalize(this); + if (_hungRun is { IsCompleted: false } hung) + { + // Closing the runspace would wait on the blocked pipeline, possibly forever. + _ = hung.ContinueWith(_ => DisposeCore(), CancellationToken.None, + TaskContinuationOptions.None, TaskScheduler.Default); + return; + } + DisposeCore(); + } + + private void DisposeCore() + { + if (Interlocked.Exchange(ref _disposed, 1) == 1) return; // PowerShell.Create(iss) ASSIGNS the runspace rather than creating it lazily, and an assigned // runspace is caller-owned — _pwsh.Dispose() does not close it. Left open, the runspace keeps // its ReuseThread pipeline thread alive, and a live thread roots the entire session state @@ -664,8 +763,5 @@ public void Dispose() var runspace = _pwsh.Runspace; _pwsh.Dispose(); runspace?.Dispose(); - // Nothing else here owns unmanaged resources directly, but suppressing finalization keeps a - // derived type that adds a finalizer from having to re-implement IDisposable to do it. - GC.SuppressFinalize(this); } } diff --git a/Services/PowerShellHost/PowerShellWorkerPool.cs b/Services/PowerShellHost/PowerShellWorkerPool.cs index 61cc143..77a793d 100644 --- a/Services/PowerShellHost/PowerShellWorkerPool.cs +++ b/Services/PowerShellHost/PowerShellWorkerPool.cs @@ -34,6 +34,12 @@ public class PowerShellWorkerPool : IDisposable public int BgAvailable => _bgPool.Count; public int HttpPoolSize => _httpPoolSize; public int BgPoolSize => _bgPoolSize; + /// Workers abandoned as hung whose blocked call has not returned yet. + public int HungWorkers => Volatile.Read(ref _hungNow); + /// Workers abandoned as hung since startup. + public long HungWorkersTotal => Interlocked.Read(ref _hungTotal); + private int _hungNow; + private long _hungTotal; public PowerShellWorkerPool(ScriptRepository repo, ILogger logger, IConfiguration config, CraftSettings settings) { @@ -168,7 +174,7 @@ private void InitializeWithSharedBase(System.Diagnostics.Stopwatch sw, // ── Base worker: parse shared modules ────────────────────────────── var baseISS = BuildISSForModules(sharedModules); - var baseWorker = new PowerShellWorker(Interlocked.Increment(ref _nextId), baseISS, _logger); + var baseWorker = NewWorker(baseISS); baseWorker.Initialize(_repo, _apiBasePath, _settings); var baseMs = sw.ElapsedMilliseconds; @@ -184,7 +190,7 @@ private void InitializeWithSharedBase(System.Diagnostics.Stopwatch sw, // ── HTTP first-worker (cloned base + HTTP-only modules) ──────────── var httpISS = BuildClonedISSWithModules(baseState, httpOnlyModules); - var firstHttpWorker = new PowerShellWorker(Interlocked.Increment(ref _nextId), httpISS, _logger); + var firstHttpWorker = NewWorker(httpISS); InitializeClonedWorker(firstHttpWorker, baseState); var httpState = firstHttpWorker.ExportModuleState(); @@ -210,7 +216,7 @@ private void InitializeWithSharedBase(System.Diagnostics.Stopwatch sw, var clonedHttpISS = BuildClonedISS(_httpClonedState); var httpRemaining = new List(); for (int i = 1; i < _httpPoolSize; i++) - httpRemaining.Add(new PowerShellWorker(Interlocked.Increment(ref _nextId), clonedHttpISS, _logger)); + httpRemaining.Add(NewWorker(clonedHttpISS)); Parallel.ForEach(httpRemaining, w => InitializeClonedWorker(w, _httpClonedState!)); foreach (var w in httpRemaining) @@ -223,7 +229,7 @@ private void InitializeWithSharedBase(System.Diagnostics.Stopwatch sw, // ── BG first-worker (cloned base + BG-only modules) ──────────────── var bgISS = BuildClonedISSWithModules(baseState, bgOnlyModules); - var firstBgWorker = new PowerShellWorker(Interlocked.Increment(ref _nextId), bgISS, _logger); + var firstBgWorker = NewWorker(bgISS); InitializeClonedWorker(firstBgWorker, baseState); var bgState = firstBgWorker.ExportModuleState(); @@ -245,7 +251,7 @@ private void InitializeWithSharedBase(System.Diagnostics.Stopwatch sw, var clonedBgISS = BuildClonedISS(_bgClonedState); var bgRemaining = new List(); for (int i = 1; i < _bgPoolSize; i++) - bgRemaining.Add(new PowerShellWorker(Interlocked.Increment(ref _nextId), clonedBgISS, _logger)); + bgRemaining.Add(NewWorker(clonedBgISS)); Parallel.ForEach(bgRemaining, w => InitializeClonedWorker(w, _bgClonedState!)); foreach (var w in bgRemaining) @@ -279,7 +285,7 @@ private void InitializeSimple(System.Diagnostics.Stopwatch sw, bool separateModu { // First HTTP worker: full module import var fullHttpISS = BuildISS(isHttp: true); - var firstWorker = new PowerShellWorker(Interlocked.Increment(ref _nextId), fullHttpISS, _logger); + var firstWorker = NewWorker(fullHttpISS); firstWorker.Initialize(_repo, _apiBasePath, _settings); _httpClonedState = firstWorker.ExportModuleState(); @@ -302,7 +308,7 @@ private void InitializeSimple(System.Diagnostics.Stopwatch sw, bool separateModu var clonedHttpISS = BuildClonedISS(_httpClonedState); var httpRemaining = new List(); for (int i = 1; i < _httpPoolSize; i++) - httpRemaining.Add(new PowerShellWorker(Interlocked.Increment(ref _nextId), clonedHttpISS, _logger)); + httpRemaining.Add(NewWorker(clonedHttpISS)); Parallel.ForEach(httpRemaining, w => InitializeClonedWorker(w, _httpClonedState!)); foreach (var w in httpRemaining) @@ -329,7 +335,7 @@ private void InitializeSimple(System.Diagnostics.Stopwatch sw, bool separateModu { var bgSw = System.Diagnostics.Stopwatch.StartNew(); var fullBgISS = BuildISS(isHttp: false); - var firstBgWorker = new PowerShellWorker(Interlocked.Increment(ref _nextId), fullBgISS, _logger); + var firstBgWorker = NewWorker(fullBgISS); firstBgWorker.Initialize(_repo, _apiBasePath, _settings); _bgClonedState = firstBgWorker.ExportModuleState(); @@ -351,7 +357,7 @@ private void InitializeSimple(System.Diagnostics.Stopwatch sw, bool separateModu var bgStart = separateModuleLists ? 1 : 0; var bgRemaining = new List(); for (int i = bgStart; i < _bgPoolSize; i++) - bgRemaining.Add(new PowerShellWorker(Interlocked.Increment(ref _nextId), clonedBgISS, _logger)); + bgRemaining.Add(NewWorker(clonedBgISS)); if (bgRemaining.Count > 0) { @@ -365,7 +371,7 @@ private void InitializeSimple(System.Diagnostics.Stopwatch sw, bool separateModu // Background-only node — no HTTP worker exists to clone from; build the BG base directly. var bgSw = System.Diagnostics.Stopwatch.StartNew(); var fullBgISS = BuildISS(isHttp: false); - var firstBgWorker = new PowerShellWorker(Interlocked.Increment(ref _nextId), fullBgISS, _logger); + var firstBgWorker = NewWorker(fullBgISS); firstBgWorker.Initialize(_repo, _apiBasePath, _settings); _bgClonedState = firstBgWorker.ExportModuleState(); @@ -382,7 +388,7 @@ private void InitializeSimple(System.Diagnostics.Stopwatch sw, bool separateModu var clonedBgISS = BuildClonedISS(_bgClonedState); var bgRemaining = new List(); for (int i = 1; i < _bgPoolSize; i++) - bgRemaining.Add(new PowerShellWorker(Interlocked.Increment(ref _nextId), clonedBgISS, _logger)); + bgRemaining.Add(NewWorker(clonedBgISS)); Parallel.ForEach(bgRemaining, w => InitializeClonedWorker(w, _bgClonedState!)); foreach (var w in bgRemaining) @@ -504,6 +510,19 @@ public void Reclaim(PowerShellWorker worker, bool isHttp, bool faulted = false) worker.CheckoutTimestamp = 0; worker.InvocationCount++; + if (worker.HungRun is { } hungRun) + { + _workerFaults.TryRemove(worker.Id, out _); + Interlocked.Increment(ref _hungTotal); + var stuck = Interlocked.Increment(ref _hungNow); + _ = hungRun.ContinueWith(_ => Interlocked.Decrement(ref _hungNow), CancellationToken.None, + TaskContinuationOptions.None, TaskScheduler.Default); + _logger.LogError("[Pool] Replacing hung W{Id} ({Type}); {Stuck} abandoned worker(s) still blocked", + worker.Id, isHttp ? "HTTP" : "BG", stuck); + Replace(worker, isHttp); + return; + } + bool needsReplace = false; if (faulted) @@ -532,43 +551,53 @@ public void Reclaim(PowerShellWorker worker, bool isHttp, bool faulted = false) if (needsReplace) { - var oldId = worker.Id; - WorkerMetricsBridge.DeregisterWorker(oldId); - worker.Dispose(); - - // Build the replacement off the calling thread so the dispatch loop is not held - // for the 6-13s ISS rebuild + Initialize() cost. Capacity briefly drops by 1 (the - // pool is short one worker until this Task completes), but the thread that just - // finished a task returns to the dispatch loop immediately. - var ish = isHttp; - _ = Task.Run(() => - { - try - { - var cloned = ish ? _httpClonedState : _bgClonedState; - var iss = cloned != null ? BuildClonedISS(cloned) : BuildISS(isHttp: ish); - var fresh = new PowerShellWorker(Interlocked.Increment(ref _nextId), iss, _logger); - if (cloned != null) - InitializeClonedWorker(fresh, cloned); - else - fresh.Initialize(_repo, _apiBasePath, _settings); - WorkerMetricsBridge.RegisterWorker(fresh.Id, ish); - if (ish) _httpPool.Add(fresh); else _bgPool.Add(fresh); - _logger.LogInformation("[Pool] Replaced W{OldId} → W{NewId} ({Type}) (background recycle)", - oldId, fresh.Id, ish ? "HTTP" : "BG"); - } - catch (Exception ex) - { - _logger.LogError(ex, "[Pool] Background recycle failed for W{OldId} ({Type}); pool capacity reduced by 1", - oldId, ish ? "HTTP" : "BG"); - } - }); + Replace(worker, isHttp); return; } if (isHttp) _httpPool.Add(worker); else _bgPool.Add(worker); } + private void Replace(PowerShellWorker worker, bool isHttp) + { + var oldId = worker.Id; + WorkerMetricsBridge.DeregisterWorker(oldId); + worker.Dispose(); + + // Build the replacement off the calling thread so the dispatch loop is not held + // for the 6-13s ISS rebuild + Initialize() cost. Capacity briefly drops by 1 (the + // pool is short one worker until this Task completes), but the thread that just + // finished a task returns to the dispatch loop immediately. + _ = Task.Run(() => + { + try + { + var cloned = isHttp ? _httpClonedState : _bgClonedState; + var iss = cloned != null ? BuildClonedISS(cloned) : BuildISS(isHttp: isHttp); + var fresh = NewWorker(iss); + if (cloned != null) + InitializeClonedWorker(fresh, cloned); + else + fresh.Initialize(_repo, _apiBasePath, _settings); + WorkerMetricsBridge.RegisterWorker(fresh.Id, isHttp); + if (isHttp) _httpPool.Add(fresh); else _bgPool.Add(fresh); + _logger.LogInformation("[Pool] Replaced W{OldId} → W{NewId} ({Type}) (background recycle)", + oldId, fresh.Id, isHttp ? "HTTP" : "BG"); + } + catch (Exception ex) + { + _logger.LogError(ex, "[Pool] Background recycle failed for W{OldId} ({Type}); pool capacity reduced by 1", + oldId, isHttp ? "HTTP" : "BG"); + } + }); + } + + private PowerShellWorker NewWorker(InitialSessionState iss) => + new(Interlocked.Increment(ref _nextId), iss, _logger) + { + StopGrace = TimeSpan.FromSeconds(_settings.Worker.StopGraceSeconds > 0 ? _settings.Worker.StopGraceSeconds : 30), + }; + private InitialSessionState BuildISS(bool isHttp) { var allowList = isHttp ? _settings.Worker.HttpModules : _settings.Worker.BgModules; diff --git a/Services/PowerShellHost/WorkerHungException.cs b/Services/PowerShellHost/WorkerHungException.cs new file mode 100644 index 0000000..ef4acc2 --- /dev/null +++ b/Services/PowerShellHost/WorkerHungException.cs @@ -0,0 +1,12 @@ +namespace Craft.PowerShellHost; + +/// +/// A cancelled invocation that did not stop within its worker's stop grace, or a call on a worker already +/// in that state. Derives from so every caller that reports a +/// cancelled invocation as a timeout keeps doing so; the worker is left +/// and the pool replaces it when it is reclaimed. +/// +public sealed class WorkerHungException(int workerId, string message) : OperationCanceledException(message) +{ + public int WorkerId { get; } = workerId; +} diff --git a/appsettings.example.jsonc b/appsettings.example.jsonc index 521e0e8..a3ef7fa 100644 --- a/appsettings.example.jsonc +++ b/appsettings.example.jsonc @@ -232,6 +232,11 @@ // 0 = no timeout (default). Recommended: 600-3600 for background jobs // "BgTimeoutSeconds": 0, + // Grace (seconds) a timed-out or cancelled pipeline gets to stop. A pipeline blocked inside a .NET + // call never sees the stop; past this grace its worker is abandoned and replaced, and the caller + // gets the usual timeout. /healthz reports workers.hung / workers.hungTotal. + // "StopGraceSeconds": 30, + // How long an HTTP request waits for a free runspace when the pool is saturated before it is // shed with 503 "Server busy, please retry". A load-shedding bound, NOT a capacity knob — it // does not add throughput, it only decides how long callers wait before a spurious 503. Widen diff --git a/docs/configuration.md b/docs/configuration.md index da7ec5b..44737a2 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -193,6 +193,11 @@ Controls the PowerShell runspace pools that execute all scripts. // 0 = no timeout (default). Recommended: 600-3600 for background jobs. "BgTimeoutSeconds": 0, + // Grace (seconds) a timed-out or cancelled pipeline gets to stop. A pipeline blocked inside a .NET + // call never sees the stop; past this grace its worker is abandoned and replaced, and the caller + // gets the usual timeout. /healthz reports workers.hung / workers.hungTotal. + "StopGraceSeconds": 30, + // Extra env vars injected into every runspace. // Use "{ApiBasePath}" as a placeholder — replaced with the resolved API directory. "EnvVars": { diff --git a/perf-harness/api-harness/API/Modules/PerfApi/PerfApi.psd1 b/perf-harness/api-harness/API/Modules/PerfApi/PerfApi.psd1 index ef1a48f..9d10b6a 100644 --- a/perf-harness/api-harness/API/Modules/PerfApi/PerfApi.psd1 +++ b/perf-harness/api-harness/API/Modules/PerfApi/PerfApi.psd1 @@ -5,7 +5,7 @@ Author = 'CRAFT perf-harness' Description = 'Synthetic HTTP endpoints for load-testing CRAFT in http-only mode. Not for production.' PowerShellVersion = '7.2' - FunctionsToExport = @('Invoke-PerfPing', 'Invoke-PerfEcho', 'Invoke-PerfCpu', 'Invoke-PerfSleep', 'Invoke-PerfJson', 'Invoke-PerfFile', 'Invoke-PerfBgEnqueue', 'Push-PerfBg', 'Push-PerfBgLeaf', 'Invoke-PerfManyRuns', 'Push-PerfHold', 'Invoke-PerfThreads', 'Invoke-PerfTableOp', 'Push-PerfCheck', 'Invoke-PerfCheckCounts', 'Push-PerfSeq', 'Invoke-PerfSeqResult', 'Push-PerfSeqWorker', 'Invoke-PerfSeqWorkerEnqueue', 'Invoke-PerfSeqWorkerResult', 'Invoke-PerfThreadBreakdown', 'Invoke-PerfSeedRuns', 'Invoke-ListPerf', 'Invoke-PerfWhoami', 'Invoke-PerfTimerTick', 'Invoke-PerfTimerCount', 'Invoke-PerfPublish', 'Invoke-PerfWatchRun', 'Invoke-PerfAllocation', 'Invoke-PerfRuns', 'Push-PerfE2E', 'Push-PerfE2EPost', 'Invoke-PerfE2EStart', 'Invoke-PerfE2EState', 'Invoke-PerfE2ERuns', 'Invoke-PerfE2EBridge', 'Invoke-PerfE2ELegacy') + FunctionsToExport = @('Invoke-PerfPing', 'Invoke-PerfEcho', 'Invoke-PerfCpu', 'Invoke-PerfSleep', 'Invoke-PerfJson', 'Invoke-PerfFile', 'Invoke-PerfBgEnqueue', 'Push-PerfBg', 'Push-PerfBgLeaf', 'Invoke-PerfManyRuns', 'Push-PerfHold', 'Invoke-PerfThreads', 'Invoke-PerfTableOp', 'Push-PerfCheck', 'Invoke-PerfCheckCounts', 'Push-PerfSeq', 'Invoke-PerfSeqResult', 'Push-PerfSeqWorker', 'Invoke-PerfSeqWorkerEnqueue', 'Invoke-PerfSeqWorkerResult', 'Invoke-PerfThreadBreakdown', 'Invoke-PerfSeedRuns', 'Invoke-ListPerf', 'Invoke-PerfWhoami', 'Invoke-PerfTimerTick', 'Invoke-PerfTimerCount', 'Invoke-PerfPublish', 'Invoke-PerfWatchRun', 'Invoke-PerfAllocation', 'Invoke-PerfRuns', 'Push-PerfE2E', 'Push-PerfE2EPost', 'Invoke-PerfE2EStart', 'Invoke-PerfE2EState', 'Invoke-PerfE2ERuns', 'Invoke-PerfE2EBridge', 'Invoke-PerfE2ELegacy', 'Invoke-PerfHang', 'Invoke-PerfBgHangEnqueue', 'Push-PerfBgHang', 'Invoke-PerfHangRelease') CmdletsToExport = @() VariablesToExport = @() AliasesToExport = @() diff --git a/perf-harness/api-harness/API/Modules/PerfApi/PerfApi.psm1 b/perf-harness/api-harness/API/Modules/PerfApi/PerfApi.psm1 index 109486b..61d17ef 100644 --- a/perf-harness/api-harness/API/Modules/PerfApi/PerfApi.psm1 +++ b/perf-harness/api-harness/API/Modules/PerfApi/PerfApi.psm1 @@ -866,3 +866,47 @@ function Invoke-PerfE2ELegacy { return @{ StatusCode = 500; Body = @{ ok = $false; error = "$_" } } } } + +# Hung-worker drivers (run-e2e-hung-workers.ps1). Each call blocks its worker in a .NET wait that PowerShell's stop +# request cannot interrupt, the shape of a stuck lock or a read with no timeout. PerfHangRelease frees them all. +function New-PerfHangGate { + $gate = [System.Threading.ManualResetEventSlim]::new($false) + [Craft.Services.PowerShellRunnerService]::GetSharedCache('E2E:Hang')[[guid]::NewGuid().ToString('N')] = $gate + $gate +} + +function Invoke-PerfHang { + param($Request, $TriggerMetadata) + $null = (New-PerfHangGate).Wait() + return @{ StatusCode = 200; Body = @{ ok = $true; endpoint = 'PerfHang' } } +} + +# Query: n=N hung background tasks (default 1). +function Invoke-PerfBgHangEnqueue { + param($Request, $TriggerMetadata) + $n = 1; if ($Request.Query.n) { $n = [int]$Request.Query.n } + $batch = @(for ($i = 0; $i -lt $n; $i++) { @{ FunctionName = 'PerfBgHang'; idx = $i } }) + $run = Start-CraftOrchestrator -InputObject @{ + OrchestratorName = "PerfBgHang-$([guid]::NewGuid().ToString('N').Substring(0, 8))" + Batch = $batch + } + return @{ StatusCode = 200; Body = @{ ok = $true; enqueued = $n; run = $run } } +} + +function Push-PerfBgHang { + param($Item) + $null = (New-PerfHangGate).Wait() + return @{ ok = $true; idx = $Item.idx } +} + +function Invoke-PerfHangRelease { + param($Request, $TriggerMetadata) + $cache = [Craft.Services.PowerShellRunnerService]::GetSharedCache('E2E:Hang') + $released = 0 + foreach ($key in @($cache.Keys)) { + $cache[$key].Set() + $cache.Remove($key) + $released++ + } + return @{ StatusCode = 200; Body = @{ ok = $true; released = $released } } +} diff --git a/perf-harness/scripts/run-e2e-hung-workers.ps1 b/perf-harness/scripts/run-e2e-hung-workers.ps1 new file mode 100644 index 0000000..43b5846 --- /dev/null +++ b/perf-harness/scripts/run-e2e-hung-workers.ps1 @@ -0,0 +1,92 @@ +<# +.SYNOPSIS + Hung-worker section of run-e2e.ps1 (dot-sourced; uses its Add-Result/Info/Json and the Azurite network). + +.DESCRIPTION + A pipeline blocked inside a .NET call never sees PowerShell's stop request. Before workers were recycled, a + timeout on such a pipeline blocked forever in PowerShell.Stop(): the worker never came back, and once every + worker had hit one the instance shed every HTTP request with 503 and ran no background work, while idle. + + A throwaway container with two HTTP and two BG workers and short timeouts is driven into exactly that state + with PerfHang / PerfBgHang (each blocks in ManualResetEventSlim.Wait), then must keep serving: every hung + request returns 504, the pool answers again, background work still drains, /healthz counts the abandoned + workers, and releasing the blocked calls lets them unwind. +#> + +Info "hung-workers: launching a container with 2 HTTP + 2 BG workers and short timeouts ..." +$hungName = 'craft-e2e-hung' +$hungBase = 'http://127.0.0.1:5403' +$hungApi = (Resolve-Path (Join-Path $root 'api-harness/API')).Path +$hungConn = 'DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://azurite:10000/devstoreaccount1;QueueEndpoint=http://azurite:10001/devstoreaccount1;TableEndpoint=http://azurite:10002/devstoreaccount1;' +docker rm -f $hungName 2>&1 | Out-Null +docker run -d --name $hungName --network craft-e2e-aznet -p '5403:8080' ` + -v "${hungApi}:/app/API:ro" ` + -e ASPNETCORE_ENVIRONMENT=Production -e CRAFT_SERVE_API=true -e CRAFT_RUN_BACKGROUND=true ` + -e App__Scripts__HttpModules__0=PerfApi -e App__ReadinessMode=Immediate -e App__Setup__Enabled=false ` + -e App__Worker__IgnoreSkuProfiles=true -e App__Worker__HttpPoolSize=2 -e App__Worker__BgPoolSize=2 ` + -e App__Worker__HttpTimeoutSeconds=4 -e App__Worker__BgTimeoutSeconds=4 -e App__Worker__StopGraceSeconds=2 ` + -e App__Worker__HttpQueueTimeoutSeconds=60 ` + -e App__Orchestrator__TablePrefix=E2EHung -e BackgroundBaseConcurrency=2 -e BackgroundHttpPressureThreshold=0 ` + -e App__RateLimit__Enabled=false -e CRAFT_LOG_LEVEL=Information ` + -e "AzureWebJobsStorage=$hungConn" ` + $SutImage 2>&1 | Out-Null +try { + $up = $false; $dl = (Get-Date).AddSeconds(180) + while ((Get-Date) -lt $dl) { + if ((Json "$hungBase/API/PerfPing").ok -eq $true) { $up = $true; break } + Start-Sleep -Seconds 2 + } + Add-Result 'hung-workers' 'container-ready' $up '-' "PerfPing on the 2+2 worker container" + + # ── Background: hang every BG worker, then queue ordinary work behind it ───── + $hangRun = Json "$hungBase/API/PerfBgHangEnqueue?n=2" + Start-Sleep -Seconds 1 + $work = Json "$hungBase/API/PerfBgEnqueue?n=10&taskms=0" + $workName = "$($work.run)" + $sw = [Diagnostics.Stopwatch]::StartNew(); $done = $null; $dl = (Get-Date).AddSeconds(120) + while ((Get-Date) -lt $dl) { + $done = @((Json "$hungBase/API/PerfRuns").runs | Where-Object { $_.name -like '*PerfBg-*' -and [int]$_.completed -eq 10 }) + if ($done.Count) { break } + Start-Sleep -Seconds 1 + } + $sw.Stop() + Add-Result 'hung-workers' 'bg-drains-past-hang' ([bool]$done.Count) ("{0:N0}s" -f $sw.Elapsed.TotalSeconds) "10 tasks queued behind 2 hung ones on 2 workers (hang run $($hangRun.run), work run $workName)" + + # ── HTTP: more hung requests than workers ───────────────────────────────────── + # Three at once against two workers: the third only gets a worker if the first two are replaced. + $sw = [Diagnostics.Stopwatch]::StartNew() + $codes = 1..3 | ForEach-Object -ThrottleLimit 3 -Parallel { + $r = try { Invoke-WebRequest "$using:hungBase/API/PerfHang" -TimeoutSec 90 -SkipHttpErrorCheck } catch { $null } + if ($r) { [int]$r.StatusCode } else { 0 } + } + $sw.Stop() + $codes = @($codes | Sort-Object) + Add-Result 'hung-workers' 'http-hang-504' (@($codes | Where-Object { $_ -eq 504 }).Count -eq 3) ("{0:N0}s" -f $sw.Elapsed.TotalSeconds) "3 hung requests on 2 workers -> $($codes -join ',') (0 = no answer in 90s)" + + $ping = $null; $dl = (Get-Date).AddSeconds(60) + while ((Get-Date) -lt $dl) { + $ping = try { Invoke-WebRequest "$hungBase/API/PerfPing" -TimeoutSec 20 -SkipHttpErrorCheck } catch { $null } + if ($ping -and $ping.StatusCode -eq 200) { break } + Start-Sleep -Seconds 1 + } + Add-Result 'hung-workers' 'http-recovers' ($ping -and $ping.StatusCode -eq 200) '-' "PerfPing after the hangs -> HTTP $(if ($ping) { $ping.StatusCode } else { 'none' })" + + $h = Json "$hungBase/healthz" + $hungTotal = [int]$h.workers.hungTotal + Add-Result 'hung-workers' 'health-counts-hung' ($hungTotal -ge 5) '-' "healthz workers.hungTotal=$hungTotal hung=$($h.workers.hung) (expect >= 5)" + + # ── Release: the abandoned calls return and the workers are let go ─────────── + $rel = Json "$hungBase/API/PerfHangRelease" + $h = $null; $dl = (Get-Date).AddSeconds(60) + while ((Get-Date) -lt $dl) { + $h = Json "$hungBase/healthz" + if ($h -and [int]$h.workers.hung -eq 0) { break } + Start-Sleep -Seconds 1 + } + Add-Result 'hung-workers' 'abandoned-unwind' ($h -and $null -ne $h.workers -and [int]$h.workers.hung -eq 0) '-' "released $($rel.released) blocked calls -> workers.hung=$($h.workers.hung)" +} +catch { Add-Result 'hung-workers' 'flow' $false '-' $_.Exception.Message } +finally { + if ($KeepUp) { Info "hung-workers: leaving $hungName up (-KeepUp)" } + else { docker rm -f $hungName 2>&1 | Out-Null } +} diff --git a/perf-harness/scripts/run-e2e.ps1 b/perf-harness/scripts/run-e2e.ps1 index def0022..adf91d8 100644 --- a/perf-harness/scripts/run-e2e.ps1 +++ b/perf-harness/scripts/run-e2e.ps1 @@ -22,7 +22,9 @@ param( [switch]$KeepUp, # Run only the orchestration section (skips the platform checks), optionally only some of its groups. [switch]$OrchOnly, - [string[]]$OrchChecks + [string[]]$OrchChecks, + # Run only the hung-worker section (run-e2e-hung-workers.ps1). + [switch]$HungOnly ) $ErrorActionPreference = 'Stop' @@ -91,7 +93,7 @@ try { Add-Result 'storage' 'azurite-ready' ($h.ready.storage -eq $true) '-' "storageReady=$($h.ready.storage)" $suiteSw = [Diagnostics.Stopwatch]::StartNew() - if (-not $OrchOnly) { + if (-not $OrchOnly -and -not $HungOnly) { # ── API dispatch ──────────────────────────────────────────────────────────── $sw = [Diagnostics.Stopwatch]::StartNew(); $ping = Api '/API/PerfPing'; $sw.Stop() Add-Result 'api' 'dispatch-ping' ($ping.ok -eq $true) ("{0}ms" -f $sw.ElapsedMilliseconds) "endpoint=$($ping.endpoint)" @@ -288,9 +290,14 @@ try { } + # -- Hung workers (run-e2e-hung-workers.ps1) ---------------------------------------- + if (-not $OrchOnly) { . (Join-Path $here 'run-e2e-hung-workers.ps1') } + # -- Orchestration engine (run-e2e-orchestration.ps1) ------------------------------ - if ($ready) { . (Join-Path $here 'run-e2e-orchestration.ps1') } - else { Add-Result 'orchestration' 'not-run' $false '-' 'SUT never became ready' } + if (-not $HungOnly) { + if ($ready) { . (Join-Path $here 'run-e2e-orchestration.ps1') } + else { Add-Result 'orchestration' 'not-run' $false '-' 'SUT never became ready' } + } # ── Summary ───────────────────────────────────────────────────────────────── $fail = @($results | Where-Object { -not $_.pass }) diff --git a/tests/Craft.Tests/HungWorkerRecycleTests.cs b/tests/Craft.Tests/HungWorkerRecycleTests.cs new file mode 100644 index 0000000..6a01b69 --- /dev/null +++ b/tests/Craft.Tests/HungWorkerRecycleTests.cs @@ -0,0 +1,249 @@ +using System.Collections; +using System.Diagnostics; +using System.Management.Automation; +using System.Management.Automation.Runspaces; +using Craft.Configuration; +using Craft.PowerShellHost; +using Craft.Services; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.Logging.Abstractions; + +namespace Craft.Tests; + +/// +/// A pipeline blocked inside a .NET call (a lock, a sync wait, a read with no timeout) never sees PowerShell's stop +/// request. The timeout used to call from the cancellation callback, which blocks until +/// the pipeline unwinds, so the timeout was never reported and the worker never went back to the pool. Once every +/// worker had hit such a call the instance served nothing: HTTP shed every request with 503 and background work +/// sat at "N running" for hours, with the process otherwise idle. +/// +/// A worker that does not stop within its grace is now abandoned (), the caller +/// gets a timeout, and the pool replaces the worker. +/// +/// +public sealed class HungWorkerRecycleTests : IDisposable +{ + private readonly List _gates = []; + + public void Dispose() + { + // Release every blocked pipeline so the abandoned runspaces can unwind and close. + foreach (var gate in _gates) gate.Set(); + } + + private ManualResetEventSlim NewGate() + { + var gate = new ManualResetEventSlim(false); + _gates.Add(gate); + return gate; + } + + private static PowerShellWorker NewWorker(TimeSpan grace) + { + var iss = InitialSessionState.CreateDefault2(); + iss.Commands.Add(new SessionStateFunctionEntry("Block", "param($Gate) $null = $Gate.Wait(); 'unblocked'")); + var worker = new PowerShellWorker(7, iss, NullLogger.Instance) { StopGrace = grace }; + worker.Runspace.ThreadOptions = PSThreadOptions.ReuseThread; + if (worker.Runspace.RunspaceStateInfo.State == RunspaceState.BeforeOpen) worker.Runspace.Open(); + return worker; + } + + /// Hang a worker the way production did: cancel an invocation blocked in a .NET wait. + private static async Task HungWorkerAsync(ManualResetEventSlim gate) + { + var worker = NewWorker(TimeSpan.FromMilliseconds(300)); + using var cts = new CancellationTokenSource(100); + await Assert.ThrowsAsync(() => worker.InvokeAsync("Block", new() { ["Gate"] = gate }, cts.Token)); + return worker; + } + + // ── The worker ────────────────────────────────────────────────────────────────────────────── + + [Fact] + public async Task ACancelledInvocationBlockedInDotNet_ReturnsAfterTheGrace() + { + var worker = NewWorker(TimeSpan.FromSeconds(1)); + using var cts = new CancellationTokenSource(500); + var sw = Stopwatch.StartNew(); + + var invoke = Task.Run(() => worker.InvokeAsync("Block", new() { ["Gate"] = NewGate() }, cts.Token)); + var returned = await Task.WhenAny(invoke, Task.Delay(TimeSpan.FromSeconds(15))) == invoke; + + Assert.True(returned, "InvokeAsync never returned after its token was cancelled"); + await Assert.ThrowsAsync(() => invoke); + Assert.True(sw.Elapsed < TimeSpan.FromSeconds(8), $"took {sw.Elapsed.TotalSeconds:0.0}s"); + Assert.True(worker.IsHung); + } + + [Fact] + public async Task TheTimeoutIsStillAnOperationCanceledException() + { + // Every runner catch site maps OperationCanceledException to its timeout (504, TimeoutException). + var worker = NewWorker(TimeSpan.FromMilliseconds(300)); + using var cts = new CancellationTokenSource(100); + + var ex = await Assert.ThrowsAnyAsync( + () => worker.InvokeAsync("Block", new() { ["Gate"] = NewGate() }, cts.Token)); + + Assert.IsType(ex); + } + + [Fact] + public async Task AHungWorkerRefusesFurtherCalls() + { + var worker = await HungWorkerAsync(NewGate()); + + await Assert.ThrowsAsync(() => worker.InvokeAsync("Block", new())); + await Assert.ThrowsAsync(() => worker.InvokeScriptAsync(ScriptBlock.Create("'x'"))); + } + + [Fact] + public async Task AScriptBlockedInDotNet_IsAbandonedToo() + { + var worker = NewWorker(TimeSpan.FromMilliseconds(300)); + using var cts = new CancellationTokenSource(100); + + var key = NewSharedGate(); + + await Assert.ThrowsAsync(() => worker.InvokeScriptAsync(ScriptBlock.Create( + $"$null = [Craft.Services.PowerShellRunnerService]::GetSharedCache('CraftPoolFixture')['{key}'].Wait()"), ct: cts.Token)); + + Assert.True(worker.IsHung); + } + + [Fact] + public async Task AStoppableInvocation_IsCancelledWithoutAbandoningTheWorker() + { + // Start-Sleep honours the stop request, so this must stay an ordinary cancellation and the worker usable. + using var worker = NewWorker(TimeSpan.FromSeconds(5)); + using var cts = new CancellationTokenSource(200); + + var ex = await Assert.ThrowsAnyAsync( + () => worker.InvokeScriptAsync(ScriptBlock.Create("Start-Sleep -Seconds 30"), ct: cts.Token)); + + Assert.IsNotType(ex); + Assert.False(worker.IsHung); + Assert.Equal("after", Assert.Single(await worker.InvokeScriptAsync(ScriptBlock.Create("'after'"))).ToString()); + } + + [Fact] + public async Task DisposingAHungWorkerDoesNotBlock_AndClosesItOnceTheCallReturns() + { + var gate = NewGate(); + var worker = await HungWorkerAsync(gate); + var runspace = worker.Runspace; + + var dispose = Task.Run(worker.Dispose); + Assert.True(await Task.WhenAny(dispose, Task.Delay(TimeSpan.FromSeconds(5))) == dispose, "Dispose blocked on the hung pipeline"); + Assert.NotEqual(RunspaceState.Closed, runspace.RunspaceStateInfo.State); + + gate.Set(); + var deadline = Environment.TickCount64 + 10_000; + while (runspace.RunspaceStateInfo.State != RunspaceState.Closed && Environment.TickCount64 < deadline) + await Task.Delay(50); + Assert.Equal(RunspaceState.Closed, runspace.RunspaceStateInfo.State); + } + + // ── A sequential run's pinned worker ──────────────────────────────────────────────────────── + + [Fact] + public async Task ASequentialRunSwapsOutAWorkerThatHung_AndRunsTheRemainingStepsOnAFreshOne() + { + // A sequential run pins one worker for every step. Kept after a step hung, it would fail every step after. + await using var h = await OrchestrationHarness.CreateAsync(); + var hung = NewWorker(TimeSpan.FromMilliseconds(300)); + using var fresh = NewWorker(TimeSpan.FromSeconds(5)); + var workers = new Queue([hung, fresh]); + h.Svc.NextWorker = () => workers.Dequeue(); + var gate = NewGate(); + h.Svc.BeforeRun = async t => + { + if (FakeOrchestrator.IdOf(t) != "s0") return; + using var cts = new CancellationTokenSource(100); + await Assert.ThrowsAsync(() => hung.InvokeAsync("Block", new() { ["Gate"] = gate }, cts.Token)); + throw new TimeoutException("step timed out"); + }; + Assert.True(await h.Start("SeqHung", OrchestrationHarness.Batch(3, "s"), sequential: true)); + + Assert.True(await h.DriveUntilFinished("SeqHung")); + Assert.Equal(new PowerShellWorker?[] { hung, fresh, fresh }, h.Svc.RanOn.ToArray()); + Assert.Equal(new (PowerShellWorker?, bool)[] { (hung, true), (fresh, false) }, h.Svc.Reclaimed.ToArray()); + var run = (await h.Store.GetRunByNameAsync("SeqHung"))!; + Assert.Equal((1, 3), (run.Failed, run.Done)); + } + + // ── The pool, through the runner ──────────────────────────────────────────────────────────── + + private static (PowerShellWorkerPool Pool, PowerShellRunnerService Runner) StartPool() + { + var settings = new CraftSettings(); + settings.Worker.HttpPoolSize = 1; + settings.Worker.BgPoolSize = 1; + settings.Worker.HttpTimeoutSeconds = 1; + settings.Worker.BgTimeoutSeconds = 1; + settings.Worker.StopGraceSeconds = 1; + settings.Worker.HttpQueueTimeoutSeconds = 30; + + var repo = new ScriptRepository(NullLogger.Instance, settings); + var pool = new PowerShellWorkerPool(repo, NullLogger.Instance, + new ConfigurationBuilder().Build(), settings); + pool.Initialize(); + Assert.True(pool.WaitForBgReady(TimeSpan.FromMinutes(2)), "pool never became ready"); + return (pool, new PowerShellRunnerService(NullLogger.Instance, pool, repo, settings)); + } + + private string NewSharedGate() + { + var key = Guid.NewGuid().ToString("N"); + PowerShellRunnerService.GetSharedCache("CraftPoolFixture")[key] = NewGate(); + return key; + } + + private static async Task WaitFor(Func condition, string what) + { + var deadline = Environment.TickCount64 + 60_000; + while (!condition() && Environment.TickCount64 < deadline) await Task.Delay(50); + Assert.True(condition(), what); + } + + [Fact] + public async Task AHungBackgroundJobTimesOut_AndThePoolKeepsServing() + { + var (pool, runner) = StartPool(); + using var _ = pool; + + // A pool of one: without the replacement, the second call would wait for that worker forever. + for (var i = 0; i < 2; i++) + { + var call = runner.ExecuteScript("Wait-CraftPoolFixtureGate", new() { ["GateKey"] = NewSharedGate() }); + Assert.True(await Task.WhenAny(call, Task.Delay(TimeSpan.FromSeconds(30))) == call, $"hung job {i} never timed out"); + await Assert.ThrowsAsync(() => call); + } + + var output = await runner.ExecuteScriptWithOutput("Get-CraftPoolFixtureThing", new() { ["TenantFilter"] = "after" }) + .WaitAsync(TimeSpan.FromSeconds(60)); + Assert.Equal("thing:after", output); + Assert.Equal(2, pool.HungWorkersTotal); + Assert.Equal(2, pool.HungWorkers); + + foreach (var gate in _gates) gate.Set(); + await WaitFor(() => pool.HungWorkers == 0, "the abandoned workers were never counted as unwound"); + } + + [Fact] + public async Task AHungHttpRequestReturns504_AndTheNextRequestGetsAWorker() + { + var (pool, runner) = StartPool(); + using var _ = pool; + + for (var i = 0; i < 2; i++) + { + var call = runner.ExecuteHttpEndpoint("Wait-CraftPoolFixtureGate", new Hashtable { ["GateKey"] = NewSharedGate() }); + Assert.True(await Task.WhenAny(call, Task.Delay(TimeSpan.FromSeconds(30))) == call, $"hung request {i} never returned"); + Assert.Equal(504, (await call).StatusCode); + } + + await WaitFor(() => pool.HttpAvailable == pool.HttpPoolSize, "the HTTP pool never got its worker back"); + Assert.Equal(2, pool.HungWorkersTotal); + } +} diff --git a/tests/Craft.Tests/OrchestrationHarness.cs b/tests/Craft.Tests/OrchestrationHarness.cs index c8a281a..a877944 100644 --- a/tests/Craft.Tests/OrchestrationHarness.cs +++ b/tests/Craft.Tests/OrchestrationHarness.cs @@ -60,6 +60,7 @@ internal override async Task RunScriptAsync(string path, Dictionary seen; seen = _maxActive) if (Interlocked.CompareExchange(ref _maxActive, now, seen) == seen) break; @@ -76,8 +77,22 @@ internal override async Task RunScriptAsync(string path, Dictionary Interlocked.Increment(ref Reclaims); + /// The worker a sequential driver checks out; null unless a test supplies one. + public Func? NextWorker; + public readonly ConcurrentQueue RanOn = new(); + public readonly ConcurrentQueue<(PowerShellWorker? Worker, bool Faulted)> Reclaimed = new(); + + internal override PowerShellWorker? CheckoutSequentialWorker(CancellationToken ct) + { + Interlocked.Increment(ref Checkouts); + return NextWorker?.Invoke(); + } + + internal override void ReclaimSequentialWorker(PowerShellWorker? worker, bool faulted) + { + Interlocked.Increment(ref Reclaims); + Reclaimed.Enqueue((worker, faulted)); + } } /// Store, pump and JobManager wired as in the host, over an in-memory table store, driven by hand. diff --git a/tests/Craft.Tests/TestApi/Modules/CraftPoolFixture/CraftPoolFixture.psd1 b/tests/Craft.Tests/TestApi/Modules/CraftPoolFixture/CraftPoolFixture.psd1 index deffb26..342eea8 100644 --- a/tests/Craft.Tests/TestApi/Modules/CraftPoolFixture/CraftPoolFixture.psd1 +++ b/tests/Craft.Tests/TestApi/Modules/CraftPoolFixture/CraftPoolFixture.psd1 @@ -3,7 +3,7 @@ GUID = 'c9eb2266-3d9f-4e78-969b-38c45d44d0fd' Author = 'Craft.Tests' RootModule = 'CraftPoolFixture.psm1' - FunctionsToExport = @('Get-CraftPoolFixtureThing') + FunctionsToExport = @('Get-CraftPoolFixtureThing', 'Wait-CraftPoolFixtureGate') CmdletsToExport = @() VariablesToExport = @() AliasesToExport = @() diff --git a/tests/Craft.Tests/TestApi/Modules/CraftPoolFixture/CraftPoolFixture.psm1 b/tests/Craft.Tests/TestApi/Modules/CraftPoolFixture/CraftPoolFixture.psm1 index 9495af8..7d51bb7 100644 --- a/tests/Craft.Tests/TestApi/Modules/CraftPoolFixture/CraftPoolFixture.psm1 +++ b/tests/Craft.Tests/TestApi/Modules/CraftPoolFixture/CraftPoolFixture.psm1 @@ -3,3 +3,10 @@ function Get-CraftPoolFixtureThing { param([string]$TenantFilter) "thing:$TenantFilter" } + +function Wait-CraftPoolFixtureGate { + # Blocks in a .NET wait, which a pipeline stop request cannot interrupt (as a stuck lock or socket read would). + param($Request, $TriggerMetadata, [string]$GateKey) + if (-not $GateKey) { $GateKey = $Request.GateKey } + $null = [Craft.Services.PowerShellRunnerService]::GetSharedCache('CraftPoolFixture')[$GateKey].Wait() +}