diff --git a/Controllers/CredentialsController.cs b/Controllers/CredentialsController.cs index c45c76f..3e8c342 100644 --- a/Controllers/CredentialsController.cs +++ b/Controllers/CredentialsController.cs @@ -5,6 +5,7 @@ using w4c_workflows.Filters; using w4c_workflows.Models; using w4c_workflows.Models.Credentials; using w4c_workflows.Services.Credentials; +using w4c_workflows.Services.Security; namespace w4c_workflows.Controllers; @@ -21,6 +22,7 @@ public class CredentialsController : ControllerBase private readonly CredentialVault _vault; private readonly CredentialTypeCatalog _types; private readonly IHttpClientFactory _http; + private readonly EgressGuard _egress; private readonly ILogger _logger; public CredentialsController( @@ -28,12 +30,14 @@ public class CredentialsController : ControllerBase CredentialVault vault, CredentialTypeCatalog types, IHttpClientFactory http, + EgressGuard egress, ILogger logger) { _db = db; _vault = vault; _types = types; _http = http; + _egress = egress; _logger = logger; } @@ -136,6 +140,15 @@ public class CredentialsController : ControllerBase if (!Uri.TryCreate(request.Url, UriKind.Absolute, out var uri)) return Ok(new CredentialTestResult(false, null, $"'{request.Url}' is not a valid absolute URL")); + // The probe URL is caller-supplied and the request carries a decrypted + // credential, so it must never be allowed to reach a private/loopback/ + // metadata address. Vet it exactly like a workflow HTTP node; the named + // client is registered without automatic redirects so a public URL + // cannot bounce the credential to an internal host. + var egress = await _egress.AuthorizeAsync(uri, ct); + if (!egress.Allowed) + return Ok(new CredentialTestResult(false, null, $"probe blocked by egress policy: {egress.Reason}")); + using var httpRequest = new HttpRequestMessage(HttpMethod.Get, uri); var injectionError = CredentialInjector.Apply(new CredentialData(entity.Type, data), _types, httpRequest); if (injectionError != null) diff --git a/Controllers/HealthController.cs b/Controllers/HealthController.cs index c012eb2..7e1574d 100644 --- a/Controllers/HealthController.cs +++ b/Controllers/HealthController.cs @@ -9,9 +9,11 @@ namespace w4c_workflows.Controllers; public class HealthController : ControllerBase { private readonly WorkflowsDbContext _db; - private readonly IConnectionMultiplexer _redis; + private readonly IConnectionMultiplexer? _redis; - public HealthController(WorkflowsDbContext db, IConnectionMultiplexer redis) + // Redis is absent in Lite mode (in-memory transport). The dependency is + // optional so liveness/readiness still resolve instead of failing DI. + public HealthController(WorkflowsDbContext db, IConnectionMultiplexer? redis = null) { _db = db; _redis = redis; @@ -39,14 +41,16 @@ public class HealthController : ControllerBase try { - checks["redis"] = _redis.IsConnected ? "ok" : "disconnected"; + checks["redis"] = _redis == null + ? "not_configured" + : (_redis.IsConnected ? "ok" : "disconnected"); } catch (Exception ex) { checks["redis"] = $"error: {ex.Message}"; } - var healthy = checks.Values.All(v => v == "ok"); + var healthy = checks.Values.All(v => v is "ok" or "not_configured"); return healthy ? Ok(new { status = "ok", checks }) : StatusCode(503, new { status = "degraded", checks }); } } diff --git a/Program.cs b/Program.cs index 1354414..c042e10 100644 --- a/Program.cs +++ b/Program.cs @@ -79,6 +79,11 @@ builder.Services.AddHttpClient(HttpRequestNodeExecutor.InsecureClientName) }, }); builder.Services.AddHttpContextAccessor(); +// Credential probe client: caller-supplied URL + decrypted credential, so it +// must not follow redirects (a public host could otherwise bounce the secret to +// an internal address after the egress check). +builder.Services.AddHttpClient("credential-test") + .ConfigurePrimaryHttpMessageHandler(() => new SocketsHttpHandler { AllowAutoRedirect = false }); // Credential vault encryption keys (used by ICredentialCipher). builder.Services.AddDataProtection(); @@ -449,7 +454,11 @@ try } catch (Exception ex) { - Log.Logger.Error(ex, "Failed to apply EF Core migrations on startup"); + // Serving traffic with an unusable/partial schema produces silent, hard-to- + // trace failures in every request. Fail fast so the orchestrator restarts or + // surfaces the real error instead. + Log.Logger.Fatal(ex, "Failed to apply EF Core migrations on startup; refusing to start"); + throw; } // Scalar API reference + OpenAPI document are exposed under a dedicated diff --git a/Services/Execution/CSharpScriptExecutor.cs b/Services/Execution/CSharpScriptExecutor.cs index 6af1e0c..134fbe0 100644 --- a/Services/Execution/CSharpScriptExecutor.cs +++ b/Services/Execution/CSharpScriptExecutor.cs @@ -81,7 +81,7 @@ public class CSharpScriptExecutor : IScriptExecutor sw.Stop(); return Task.FromResult(new ExecutionResult(true, 0, string.Empty, string.Empty, output, null, sw.Elapsed)); } - catch (Exception ex) + catch (Exception ex) when (ex is not OperationCanceledException) { sw.Stop(); return Task.FromResult(new ExecutionResult(false, -1, string.Empty, string.Empty, null, ex.Message, sw.Elapsed)); diff --git a/Services/Execution/ProcessRunner.cs b/Services/Execution/ProcessRunner.cs index 561496d..9ffd1e1 100644 --- a/Services/Execution/ProcessRunner.cs +++ b/Services/Execution/ProcessRunner.cs @@ -1,4 +1,5 @@ using System.Diagnostics; +using System.Text; namespace w4c_workflows.Services.Execution; @@ -8,10 +9,14 @@ public sealed record ProcessOutput(int ExitCode, string Stdout, string Stderr); /// /// Runs a subprocess with stdin/stdout/stderr redirection, an optional JSON /// stdin payload, extra environment variables, and a hard timeout. On timeout -/// the whole process tree is killed. +/// or external cancellation the whole process tree is always killed, and the +/// captured output is bounded so a chatty script cannot exhaust memory. /// public static class ProcessRunner { + /// Per-stream capture ceiling. Extra output is drained and discarded. + internal const int MaxCapturedChars = 1024 * 1024; + public static async Task RunAsync( string executable, IReadOnlyList args, @@ -44,9 +49,10 @@ public static class ProcessRunner throw new InvalidOperationException($"failed to start '{executable}'"); // Start draining stdout/stderr before writing stdin so a chatty child - // never blocks on a full pipe. - var stdoutTask = process.StandardOutput.ReadToEndAsync(); - var stderrTask = process.StandardError.ReadToEndAsync(); + // never blocks on a full pipe. Both readers drain without bound but keep + // only MaxCapturedChars in memory. + var stdoutTask = ReadCappedAsync(process.StandardOutput, MaxCapturedChars); + var stderrTask = ReadCappedAsync(process.StandardError, MaxCapturedChars); if (!string.IsNullOrEmpty(stdin)) { @@ -80,17 +86,63 @@ public static class ProcessRunner } catch (OperationCanceledException) when (!ct.IsCancellationRequested) { - try { process.Kill(entireProcessTree: true); } catch { /* already gone */ } - await stdoutTask; - await stderrTask; throw new TimeoutException($"process exceeded the {timeout.TotalSeconds:0}s time limit"); } + finally + { + // The wait throws on timeout AND on external cancellation; without + // this the child tree survived as an orphan whenever a run was + // cancelled or the host shut down. Reap it unconditionally, then + // drain the readers so they are never left unobserved. + if (!process.HasExited) + { + try { process.Kill(entireProcessTree: true); } catch { /* already gone */ } + // Reap the killed child so it does not linger as a zombie; bounded + // so a pathological kill cannot wedge the runner. + try { await process.WaitForExitAsync(CancellationToken.None).WaitAsync(TimeSpan.FromSeconds(5)); } + catch { /* best effort */ } + } + + await IgnoreFailureAsync(stdoutTask); + await IgnoreFailureAsync(stderrTask); + } var stdout = await stdoutTask; var stderr = await stderrTask; return new ProcessOutput(process.ExitCode, stdout, stderr); } + /// + /// Reads to the end while capturing at most + /// characters. The remainder is still drained so + /// the child never blocks on a full pipe, but it is not retained. + /// + public static async Task ReadCappedAsync(TextReader reader, int maxChars, CancellationToken ct = default) + { + var buffer = new char[8192]; + var captured = new StringBuilder(Math.Min(maxChars, 64 * 1024)); + + while (true) + { + var read = await reader.ReadAsync(buffer.AsMemory(0, buffer.Length), ct); + if (read == 0) + break; + + if (captured.Length >= maxChars) + continue; // over the cap — keep draining, discard + + var take = Math.Min(read, maxChars - captured.Length); + captured.Append(buffer, 0, take); + } + + return captured.ToString(); + } + + private static async Task IgnoreFailureAsync(Task task) + { + try { await task; } catch { /* best-effort drain */ } + } + /// /// Resolves an executable name against PATH (or a literal path) without /// invoking it — a cheap availability probe. diff --git a/Services/Execution/SubprocessScriptExecutor.cs b/Services/Execution/SubprocessScriptExecutor.cs index e62870d..e406099 100644 --- a/Services/Execution/SubprocessScriptExecutor.cs +++ b/Services/Execution/SubprocessScriptExecutor.cs @@ -46,7 +46,7 @@ public class SubprocessScriptExecutor : IScriptExecutor return ExecutionHelpers.ToResult(output, sw.Elapsed); } - catch (Exception ex) + catch (Exception ex) when (ex is not OperationCanceledException) { sw.Stop(); return new ExecutionResult(false, -1, string.Empty, string.Empty, null, ex.Message, sw.Elapsed); diff --git a/Services/Execution/TypeScriptExecutor.cs b/Services/Execution/TypeScriptExecutor.cs index fe9467b..c2e52a8 100644 --- a/Services/Execution/TypeScriptExecutor.cs +++ b/Services/Execution/TypeScriptExecutor.cs @@ -63,7 +63,7 @@ public class TypeScriptExecutor : IScriptExecutor return ExecutionHelpers.ToResult(run, sw.Elapsed); } - catch (Exception ex) + catch (Exception ex) when (ex is not OperationCanceledException) { sw.Stop(); return new ExecutionResult(false, -1, string.Empty, string.Empty, null, ex.Message, sw.Elapsed); diff --git a/Services/Execution/WorkerHostService.cs b/Services/Execution/WorkerHostService.cs index 2e37a99..f20fe19 100644 --- a/Services/Execution/WorkerHostService.cs +++ b/Services/Execution/WorkerHostService.cs @@ -196,7 +196,7 @@ public class WorkerHostService : BackgroundService { result = await executor.ExecuteAsync(invocation, ct); } - catch (Exception ex) + catch (Exception ex) when (ex is not OperationCanceledException) { result = new ExecutionResult(false, -1, string.Empty, string.Empty, null, ex.Message, sw.Elapsed); } diff --git a/Services/Nodes/Executors/CodeNodeExecutor.cs b/Services/Nodes/Executors/CodeNodeExecutor.cs index 1ed7b4a..4dfdc68 100644 --- a/Services/Nodes/Executors/CodeNodeExecutor.cs +++ b/Services/Nodes/Executors/CodeNodeExecutor.cs @@ -63,7 +63,7 @@ public sealed class CodeNodeExecutor : INodeExecutor { result = await executor.ExecuteAsync(invocation, ct); } - catch (Exception ex) + catch (Exception ex) when (ex is not OperationCanceledException) { return NodeExecutionOutcome.Failed(ex.Message, "script_failed"); } diff --git a/Services/Nodes/NodeWorkflowRunner.cs b/Services/Nodes/NodeWorkflowRunner.cs index 80a574e..d57a9ac 100644 --- a/Services/Nodes/NodeWorkflowRunner.cs +++ b/Services/Nodes/NodeWorkflowRunner.cs @@ -120,7 +120,7 @@ public sealed class NodeWorkflowRunner result = await _runner.RunAsync( build.Graph, FlowItemJson.Parse(run.InputJson), environment, listener, ct); } - catch (Exception ex) + catch (Exception ex) when (ex is not OperationCanceledException) { _logger?.LogError(ex, "Node workflow run {RunId} threw", run.Id); return await FailAsync(db, run, ex.Message, ct); diff --git a/Services/Runs/RunLifecycleEngine.cs b/Services/Runs/RunLifecycleEngine.cs index 1bb0e91..c820b55 100644 --- a/Services/Runs/RunLifecycleEngine.cs +++ b/Services/Runs/RunLifecycleEngine.cs @@ -96,6 +96,12 @@ public class RunLifecycleEngine if (await DispatchRunAsync(db, dispatcher, run, ct)) dispatched++; } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + // Shutdown mid-dispatch. A cancelled run did not fail, so do not + // mark it failed; leave it for the timeout sweep / next replica. + throw; + } catch (Exception ex) { _logger.LogError(ex, "Failed to dispatch run {RunId} (tenant {TenantId})", run.Id, run.TenantId); @@ -171,11 +177,19 @@ public class RunLifecycleEngine await db.SaveChangesAsync(ct); var workingDir = await dispatcher.ResolveWorkingDirAsync(run.TenantId, workflow.Path, ct); - await _nodeRunner.RunAsync(db, run, workflow, workingDir, ct); - // The runner owns the terminal run status; the dispatch lease is released - // here for both success and failure so a finished run is immediately claimable. - await _leases.ReleaseAsync(run.TenantId, run.Id.ToString(), _owner); + try + { + await _nodeRunner.RunAsync(db, run, workflow, workingDir, ct); + } + finally + { + // The runner owns the terminal run status; the dispatch lease is + // released here for success, failure AND cancellation so a cancelled + // run does not pin the lease for its whole TTL. + await _leases.ReleaseAsync(run.TenantId, run.Id.ToString(), _owner); + } + return true; } @@ -358,18 +372,33 @@ public class RunLifecycleEngine foreach (var message in messages) { ct.ThrowIfCancellationRequested(); + + // Ack only when the result was actually applied. Acking in a + // finally block turned a transient failure (e.g. a DB error) + // into permanent data loss: the message was removed while the + // run stayed stuck until its timeout. Leaving it pending lets + // the next tick re-claim and retry it (at-least-once). + var applied = false; try { await ProcessResultAsync(db, dispatcher, store, tenant, message, ct); + applied = true; + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + throw; // shutdown in progress — do not ack, do not count } catch (Exception ex) { - _logger.LogError(ex, "Failed to process result {MessageId} for tenant {TenantId}", message.Id, tenant); - } - finally - { - await _events.AckAsync(tenant, Streams.Results, message.Id, ct); + _logger.LogError(ex, + "Failed to process result {MessageId} for tenant {TenantId}; leaving it pending for redelivery", + message.Id, tenant); } + + if (!applied) + continue; + + await _events.AckAsync(tenant, Streams.Results, message.Id, ct); processed++; } } diff --git a/Services/Triggers/TriggerScheduler.cs b/Services/Triggers/TriggerScheduler.cs index 43dfa74..5bf4397 100644 --- a/Services/Triggers/TriggerScheduler.cs +++ b/Services/Triggers/TriggerScheduler.cs @@ -47,27 +47,46 @@ public class TriggerScheduler : BackgroundService _logger.LogInformation("TriggerScheduler started (poll {Interval}s)", _pollSeconds); using var timer = new PeriodicTimer(TimeSpan.FromSeconds(_pollSeconds)); + // First pass after a short warm-up so the app can serve initial requests. try { - // First pass after a short warm-up so the app can serve initial requests. await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken); - await EvaluateAllAsync(stoppingToken); - - while (await timer.WaitForNextTickAsync(stoppingToken)) - { - await EvaluateAllAsync(stoppingToken); - } } catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { - // Graceful shutdown — the host is stopping. + return; // graceful shutdown during warm-up } - catch (Exception ex) + + while (!stoppingToken.IsCancellationRequested) { - // A transient env/DB error must never take the whole control plane - // down (HostOptions.BackgroundServiceExceptionBehavior defaults to - // StopHost). Log and keep the scheduler alive for the next tick. - _logger.LogError(ex, "TriggerScheduler loop failed; will retry on next tick"); + // The try/catch must stay INSIDE the loop. If it wrapped the loop, a + // single throwing tick would end ExecuteAsync and silently kill every + // cron/interval trigger for the lifetime of the process (the host is + // configured with BackgroundServiceExceptionBehavior.Ignore). + try + { + await EvaluateAllAsync(stoppingToken); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + break; // graceful shutdown — the host is stopping + } + catch (Exception ex) + { + // A transient env/DB error must never take the whole control plane + // down. Log and keep the scheduler alive for the next tick. + _logger.LogError(ex, "TriggerScheduler tick failed; will retry on next tick"); + } + + try + { + if (!await timer.WaitForNextTickAsync(stoppingToken)) + break; + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + break; + } } } diff --git a/w4c-workflows-api.Tests/ProcessRunnerTests.cs b/w4c-workflows-api.Tests/ProcessRunnerTests.cs new file mode 100644 index 0000000..f8eb15d --- /dev/null +++ b/w4c-workflows-api.Tests/ProcessRunnerTests.cs @@ -0,0 +1,97 @@ +using System.Diagnostics; +using w4c_workflows.Services.Execution; +using Xunit; + +namespace w4c_workflows.Tests; + +/// +/// Regression tests for subprocess capture and reaping. The runner must bound +/// memory for chatty children and must never leave a child tree alive when the +/// caller's token is cancelled. +/// +public class ProcessRunnerTests +{ + [Fact] + public async Task ReadCappedAsync_returns_everything_under_the_cap() + { + using var reader = new StringReader("hello world"); + + var captured = await ProcessRunner.ReadCappedAsync(reader, 100); + + Assert.Equal("hello world", captured); + } + + [Fact] + public async Task ReadCappedAsync_captures_at_most_the_cap_but_drains_the_reader() + { + // 10x the cap: the reader must be fully drained (so a child never blocks + // on a full pipe) while only the first `cap` characters are retained. + using var reader = new StringReader(new string('a', 1_000)); + + var captured = await ProcessRunner.ReadCappedAsync(reader, 100); + + Assert.Equal(100, captured.Length); + Assert.Equal(new string('a', 100), captured); + } + + [Fact] + public async Task RunAsync_throws_timeout_when_the_child_exceeds_the_limit() + { + if (!OperatingSystem.IsLinux()) + return; + + var ex = await Assert.ThrowsAsync(() => ProcessRunner.RunAsync( + "/bin/sh", new[] { "-c", "sleep 30" }, Path.GetTempPath(), + null, null, TimeSpan.FromMilliseconds(300), default)); + + Assert.Contains("time limit", ex.Message); + } + + [Fact] + public async Task RunAsync_kills_the_child_when_cancelled_externally() + { + if (!OperatingSystem.IsLinux()) + return; + + var pidFile = Path.Combine(Path.GetTempPath(), $"w4c-proc-{Guid.NewGuid():N}.pid"); + using var cts = new CancellationTokenSource(); + + var run = ProcessRunner.RunAsync( + "/bin/sh", + new[] { "-c", $"echo $$ > '{pidFile}'; exec sleep 30" }, + Path.GetTempPath(), + null, null, TimeSpan.FromSeconds(30), cts.Token); + + var deadline = DateTime.UtcNow.AddSeconds(5); + while (!File.Exists(pidFile) && DateTime.UtcNow < deadline) + await Task.Delay(20); + Assert.True(File.Exists(pidFile), "child never reported its pid"); + + cts.Cancel(); + await Assert.ThrowsAnyAsync(() => run); + + var pid = int.Parse(File.ReadAllText(pidFile).Trim()); + var gone = false; + for (var i = 0; i < 100 && !gone; i++) + { + gone = !IsAlive(pid); + if (!gone) + await Task.Delay(50); + } + + Assert.True(gone, $"child process {pid} survived external cancellation"); + } + + private static bool IsAlive(int pid) + { + try + { + using var process = Process.GetProcessById(pid); + return !process.HasExited; + } + catch (ArgumentException) + { + return false; + } + } +}