diff --git a/Controllers/WebhooksController.cs b/Controllers/WebhooksController.cs index cb81541..20f03ee 100644 --- a/Controllers/WebhooksController.cs +++ b/Controllers/WebhooksController.cs @@ -1,3 +1,4 @@ +using System.Text; using System.Text.Json; using Microsoft.AspNetCore.Mvc; using Microsoft.EntityFrameworkCore; @@ -11,19 +12,31 @@ namespace w4c_workflows.Controllers; /// /// Webhook receiver for workflows with trigger.type = webhook. The route -/// is PUBLIC (external callers have no operator key); access is governed by an -/// optional shared secret (Workflows:WebhookSecret) verified against the -/// X-Webhook-Secret header. When no secret is configured the endpoint is -/// open — acceptable for v1 dev, hardened in step 13. +/// is PUBLIC (external callers have no operator key); access is governed by a +/// shared secret header (X-Webhook-Secret) and is fail-closed: +/// - If Workflows:WebhookSecrets maps tenants to secrets, the request is +/// attributed to exactly one tenant and only that tenant's workflows match. +/// - Otherwise the global Workflows:WebhookSecret is required. Because a +/// single secret cannot identify a tenant, a path registered by more than one +/// tenant is rejected (409) rather than firing across tenants. +/// - With neither configured every request is rejected (401). +/// +/// Matching uses the indexed column (falling +/// back to the trigger JSON for rows compiled before the column existed), and the +/// request body is capped by Workflows:WebhookMaxBodyBytes. /// [ApiController] [Route("/h")] public class WebhooksController : ControllerBase { + private const long DefaultMaxBodyBytes = 1_048_576; + private readonly WorkflowsDbContext _db; private readonly IRunLauncher _launcher; private readonly ILogger _logger; private readonly string? _sharedSecret; + private readonly Dictionary _tenantSecrets; + private readonly long _maxBodyBytes; public WebhooksController( WorkflowsDbContext db, @@ -37,6 +50,17 @@ public class WebhooksController : ControllerBase _sharedSecret = string.IsNullOrWhiteSpace(config["Workflows:WebhookSecret"]) ? null : config["Workflows:WebhookSecret"]; + + _tenantSecrets = new Dictionary(StringComparer.Ordinal); + foreach (var child in config.GetSection("Workflows:WebhookSecrets").GetChildren()) + { + if (!string.IsNullOrWhiteSpace(child.Key) && !string.IsNullOrWhiteSpace(child.Value)) + _tenantSecrets[child.Key] = child.Value!; + } + + _maxBodyBytes = long.TryParse(config["Workflows:WebhookMaxBodyBytes"], out var max) && max > 0 + ? max + : DefaultMaxBodyBytes; } /// Receives a webhook at /h/{path} and fires every matching workflow. @@ -45,14 +69,37 @@ public class WebhooksController : ControllerBase { var webhookPath = "/h/" + (path ?? string.Empty); - if (!Authorized()) + if (!TryAuthorize(out var tenantId)) return Unauthorized(new { error = "Invalid webhook secret." }); - var matches = await FindMatchesAsync(webhookPath, ct); + var matches = await FindMatchesAsync(webhookPath, tenantId, ct); if (matches.Count == 0) return NotFound(new { error = $"No webhook workflow registered for path '{webhookPath}'." }); - var input = await ReadInputAsync(ct); + // A shared global secret cannot identify a tenant. Refuse to fan a path out + // across tenants; the operator must configure per-tenant secrets or use a + // tenant-unique path. + if (tenantId == null) + { + var tenants = matches.Select(m => m.TenantId).Distinct(StringComparer.Ordinal).ToList(); + if (tenants.Count > 1) + { + _logger.LogWarning( + "Webhook {Path} matches workflows in multiple tenants ({Tenants}); refusing cross-tenant fan-out", + webhookPath, string.Join(",", tenants)); + return Conflict(new + { + error = $"webhook path '{webhookPath}' is registered by multiple tenants; " + + "configure Workflows:WebhookSecrets or make the path tenant-unique.", + }); + } + } + + var (input, tooLarge) = await ReadInputAsync(ct); + if (tooLarge) + return StatusCode(StatusCodes.Status413PayloadTooLarge, + new { error = $"webhook body exceeds the {_maxBodyBytes} byte limit." }); + var correlation = Request.Headers["X-Request-Id"].FirstOrDefault() ?? Guid.NewGuid().ToString("N"); var runIds = new List(matches.Count); @@ -86,61 +133,114 @@ public class WebhooksController : ControllerBase return Accepted(new { runs = runIds }); } - private bool Authorized() + /// + /// Resolves the caller's tenant. Returns false (fail-closed) when no secret is + /// configured or the provided secret matches nothing. A null + /// with a true result means the global secret was + /// used and the tenant is unknown. + /// + private bool TryAuthorize(out string? tenantId) { - // When no secret is configured, reject all webhook requests. An empty - // secret previously meant "open" — acceptable for dev but a security - // hole in production. Callers must set Workflows:WebhookSecret. Log a - // warning so this fail-closed state is not silent. - if (_sharedSecret == null) + tenantId = null; + + if (_tenantSecrets.Count == 0 && _sharedSecret == null) { - _logger.LogWarning("Workflows:WebhookSecret is not configured; the webhook endpoint is disabled and every request to /h/* will be rejected with 401."); + _logger.LogWarning( + "Neither Workflows:WebhookSecret nor Workflows:WebhookSecrets is configured; every /h/* request is rejected."); return false; } var provided = Request.Headers["X-Webhook-Secret"].FirstOrDefault(); - return provided != null && CryptographicOperationsEquals(provided, _sharedSecret); + if (provided == null) + return false; + + if (_tenantSecrets.Count > 0) + { + foreach (var (tenant, secret) in _tenantSecrets) + { + if (CryptographicOperationsEquals(provided, secret)) + { + tenantId = tenant; + return true; + } + } + return false; + } + + return _sharedSecret != null && CryptographicOperationsEquals(provided, _sharedSecret); } - private async Task> FindMatchesAsync(string webhookPath, CancellationToken ct) + private async Task> FindMatchesAsync(string webhookPath, string? tenantId, CancellationToken ct) { - var candidates = await _db.Workflows - .Where(w => w.Status == WorkflowStatus.Compiled && w.TriggerJson != null) - .ToListAsync(ct); + // The indexed column powers the common case; rows compiled before the + // column existed (WebhookPath == null) still fall back to trigger JSON. + var query = _db.Workflows.Where(w => + w.Status == WorkflowStatus.Compiled + && (w.WebhookPath == webhookPath || w.WebhookPath == null)); + + if (tenantId != null) + query = query.Where(w => w.TenantId == tenantId); + + var candidates = await query.ToListAsync(ct); var matches = new List(); foreach (var workflow in candidates) { - var spec = TriggerSpec.Parse(workflow.TriggerJson, out _); - if (spec?.Type == TriggerType.Webhook - && string.Equals(spec.WebhookPath, webhookPath, StringComparison.Ordinal)) + if (IsWebhookMatch(workflow, webhookPath)) matches.Add(workflow); } return matches; } - private async Task ReadInputAsync(CancellationToken ct) + /// + /// True when the workflow's webhook trigger points at . + /// Prefers the denormalized column; parses trigger JSON only for legacy rows. + /// + public static bool IsWebhookMatch(Workflow workflow, string webhookPath) { - using var reader = new StreamReader(Request.Body); - var body = (await reader.ReadToEndAsync(ct)).Trim(); + if (!string.IsNullOrWhiteSpace(workflow.WebhookPath)) + return string.Equals(workflow.WebhookPath, webhookPath, StringComparison.Ordinal); + + var spec = TriggerSpec.Parse(workflow.TriggerJson, out _); + return spec?.Type == TriggerType.Webhook + && string.Equals(spec.WebhookPath, webhookPath, StringComparison.Ordinal); + } + + private async Task<(string? Input, bool TooLarge)> ReadInputAsync(CancellationToken ct) + { + // Reject a declared oversized body before reading anything. + if (Request.ContentLength is long declared && declared > _maxBodyBytes) + return (null, true); + + using var buffer = new MemoryStream(); + var chunk = new byte[81_920]; + int read; + while ((read = await Request.Body.ReadAsync(chunk, ct)) > 0) + { + if (buffer.Length + read > _maxBodyBytes) + return (null, true); + buffer.Write(chunk, 0, read); + } + + var body = Encoding.UTF8.GetString(buffer.ToArray()).Trim(); if (body.Length == 0) - return null; + return (null, false); try { using var _ = JsonDocument.Parse(body); - return body; // already valid JSON — pass through verbatim + return (body, false); // already valid JSON — pass through verbatim } catch (JsonException) { - return JsonSerializer.Serialize(body); // wrap non-JSON bodies as a JSON string + return (JsonSerializer.Serialize(body), false); // wrap non-JSON bodies as a JSON string } } private static bool CryptographicOperationsEquals(string a, string b) { - var aBytes = System.Text.Encoding.UTF8.GetBytes(a); - var bBytes = System.Text.Encoding.UTF8.GetBytes(b); + var aBytes = Encoding.UTF8.GetBytes(a); + var bBytes = Encoding.UTF8.GetBytes(b); if (aBytes.Length != bBytes.Length) return false; diff --git a/Controllers/WorkflowQuotaController.cs b/Controllers/WorkflowQuotaController.cs index 7694010..c37d09e 100644 --- a/Controllers/WorkflowQuotaController.cs +++ b/Controllers/WorkflowQuotaController.cs @@ -6,19 +6,24 @@ namespace w4c_workflows.Controllers; /// /// Per-tenant workflow execution quota. GET returns the current period's -/// usage (drives the counter on the Workflows page); POST reset zeroes it -/// for the current period (admin / support). Authenticated with the tenant's -/// operator key, like every other control-plane endpoint. +/// usage (drives the counter on the Workflows page). POST reset is a +/// platform-admin action: a tenant must not be able to zero its own counter, so +/// it additionally requires the shared X-Admin-Key matching +/// Workflows:AdminApiKey (fail-closed when unset). /// [ApiController] [Route("api/workflows/quota")] public class WorkflowQuotaController : ControllerBase { private readonly WorkflowQuotaService _quota; + private readonly string? _adminKey; - public WorkflowQuotaController(WorkflowQuotaService quota) + public WorkflowQuotaController(WorkflowQuotaService quota, IConfiguration config) { _quota = quota; + _adminKey = string.IsNullOrWhiteSpace(config["Workflows:AdminApiKey"]) + ? null + : config["Workflows:AdminApiKey"]; } private string TenantId => (string?)HttpContext.Items["TenantId"] @@ -32,11 +37,45 @@ public class WorkflowQuotaController : ControllerBase return Ok(await _quota.GetAsync(TenantId, ct)); } - /// Resets the current period's execution counter to zero. + /// + /// Resets the current period's execution counter to zero. Platform admin only: + /// requires the shared admin key in addition to the operator key. + /// [HttpPost("reset")] [RequireScope("manage")] public async Task Reset(CancellationToken ct) { + if (!IsPlatformAdmin()) + { + return StatusCode(StatusCodes.Status403Forbidden, new + { + error = "Resetting a tenant quota requires the platform admin key " + + "(header X-Admin-Key, configured as Workflows:AdminApiKey).", + }); + } + return Ok(await _quota.ResetAsync(TenantId, ct)); } + + private bool IsPlatformAdmin() + { + if (_adminKey == null) + return false; + + var provided = Request.Headers["X-Admin-Key"].FirstOrDefault(); + return provided != null && FixedTimeEquals(provided, _adminKey); + } + + private static bool FixedTimeEquals(string a, string b) + { + var aBytes = System.Text.Encoding.UTF8.GetBytes(a); + var bBytes = System.Text.Encoding.UTF8.GetBytes(b); + if (aBytes.Length != bBytes.Length) + return false; + + var result = 0; + for (var i = 0; i < aBytes.Length; i++) + result |= aBytes[i] ^ bBytes[i]; + return result == 0; + } } diff --git a/Data/WorkflowsSchema.cs b/Data/WorkflowsSchema.cs index 0902907..1bc6d6b 100644 --- a/Data/WorkflowsSchema.cs +++ b/Data/WorkflowsSchema.cs @@ -89,6 +89,9 @@ public static class WorkflowsSchema "CREATE INDEX IF NOT EXISTS \"IX_WorkflowRuns_ParentRunId\" ON workflows.\"WorkflowRuns\" (\"ParentRunId\");", // Per-tenant workflow repo setting + which repo a compiled workflow came from. "ALTER TABLE workflows.\"Workflows\" ADD COLUMN IF NOT EXISTS \"Repo\" text NOT NULL DEFAULT 'workflows';", + // Denormalized webhook path for the public receiver's indexed lookup. + "ALTER TABLE workflows.\"Workflows\" ADD COLUMN IF NOT EXISTS \"WebhookPath\" text NULL;", + "CREATE INDEX IF NOT EXISTS \"IX_Workflows_TenantId_WebhookPath\" ON workflows.\"Workflows\" (\"TenantId\", \"WebhookPath\") WHERE \"WebhookPath\" IS NOT NULL;", "CREATE TABLE IF NOT EXISTS workflows.\"WorkflowRepos\" (" + "\"TenantId\" text NOT NULL, \"RepoName\" text NOT NULL, \"UpdatedAt\" timestamptz NULL, " + "CONSTRAINT \"PK_WorkflowRepos\" PRIMARY KEY (\"TenantId\"));", @@ -143,6 +146,7 @@ public static class WorkflowsSchema ("WorkflowRuns", "ParentTaskId", "ALTER TABLE \"WorkflowRuns\" ADD COLUMN \"ParentTaskId\" TEXT NULL;"), ("WorkflowRuns", "Depth", "ALTER TABLE \"WorkflowRuns\" ADD COLUMN \"Depth\" INTEGER NOT NULL DEFAULT 0;"), ("Workflows", "Repo", "ALTER TABLE \"Workflows\" ADD COLUMN \"Repo\" TEXT NOT NULL DEFAULT 'workflows';"), + ("Workflows", "WebhookPath", "ALTER TABLE \"Workflows\" ADD COLUMN \"WebhookPath\" TEXT NULL;"), }; private static readonly string[] LiteTables = @@ -204,6 +208,7 @@ public static class WorkflowsSchema "CREATE INDEX IF NOT EXISTS \"IX_Credentials_TenantId\" ON \"Credentials\" (\"TenantId\");", "CREATE INDEX IF NOT EXISTS \"IX_WorkflowRuns_ParentRunId\" ON \"WorkflowRuns\" (\"ParentRunId\");", "CREATE INDEX IF NOT EXISTS \"IX_WorkflowRuntimes_TenantId\" ON \"WorkflowRuntimes\" (\"TenantId\");", + "CREATE INDEX IF NOT EXISTS \"IX_Workflows_TenantId_WebhookPath\" ON \"Workflows\" (\"TenantId\", \"WebhookPath\");", }; private static async Task ApplyLiteAsync(WorkflowsDbContext db, CancellationToken ct) diff --git a/Filters/RequireScopeAttribute.cs b/Filters/RequireScopeAttribute.cs index 3d86c0e..7f0c309 100644 --- a/Filters/RequireScopeAttribute.cs +++ b/Filters/RequireScopeAttribute.cs @@ -35,11 +35,26 @@ public class RequireScopeAttribute : Attribute, IAsyncActionFilter } var scopes = context.HttpContext.Items["Scopes"] as IReadOnlyList; - // A key with no recorded scope set predates scope enforcement (older/seed keys). - // Treat it as unlimited rather than silently locking the operator out; a key that - // actually declares a scope set is still enforced below. + // A key with no recorded scope set predates scope enforcement (older/seed + // keys). By default it is treated as unlimited so legacy keys keep working. + // With Auth:EnforceScopes=true it is denied instead: an operator must + // re-mint the key with explicit scopes. This is the migration switch for + // tightening P1-14 without silently locking everyone out. if (scopes == null || scopes.Count == 0) { + var enforce = context.HttpContext.Items["EnforceScopes"] as bool? ?? false; + if (enforce) + { + context.Result = new ObjectResult(new + { + error = "Operator key has no scopes; re-mint it with explicit scopes.", + }) + { + StatusCode = StatusCodes.Status403Forbidden, + }; + return; + } + await next(); return; } diff --git a/Middleware/AuthMiddleware.cs b/Middleware/AuthMiddleware.cs index 30ac021..2350b6a 100644 --- a/Middleware/AuthMiddleware.cs +++ b/Middleware/AuthMiddleware.cs @@ -26,12 +26,20 @@ public class AuthMiddleware private readonly RequestDelegate _next; private readonly string _signingKey; + private readonly string? _jwtIssuer; + private readonly string? _jwtAudience; + private readonly bool _enforceScopes; private readonly ILogger _logger; public AuthMiddleware(RequestDelegate next, IConfiguration config, ILogger logger) { _next = next; _signingKey = config["Auth:JwtSigningKey"] ?? string.Empty; + _jwtIssuer = config["Auth:JwtIssuer"]; + _jwtAudience = config["Auth:JwtAudience"]; + // Fail-closed switch for operator keys that carry no scope set. Off by + // default so legacy/seed keys keep working until they are re-minted. + _enforceScopes = config.GetValue("Auth:EnforceScopes", false); _logger = logger; } @@ -48,7 +56,8 @@ public class AuthMiddleware // /api/keys* — main JWT surface. if (path.StartsWithSegments("/api/keys")) { - var principal = JwtValidator.Validate(context.Request.Headers.Authorization.ToString(), _signingKey); + var principal = JwtValidator.Validate( + context.Request.Headers.Authorization.ToString(), _signingKey, _jwtIssuer, _jwtAudience); var tenantId = principal?.FindFirst("tenant_id")?.Value; if (principal == null || string.IsNullOrEmpty(tenantId)) { @@ -101,6 +110,7 @@ public class AuthMiddleware context.Items["TenantId"] = op.TenantId; context.Items["AuthKind"] = "operator"; context.Items["Scopes"] = op.Scopes; + context.Items["EnforceScopes"] = _enforceScopes; context.Items["OperatorKeyId"] = op.KeyId; _logger.LogDebug("Operator-key auth for tenant {TenantId}", op.TenantId); diff --git a/Models/Entities.cs b/Models/Entities.cs index 21364e0..8b79f16 100644 --- a/Models/Entities.cs +++ b/Models/Entities.cs @@ -99,6 +99,13 @@ public class Workflow public required string Mode { get; set; } // function | durable | handler public string? TriggerJson { get; set; } // jsonb: { type, cron, interval, webhookPath, stream } /// + /// Denormalized trigger.webhookPath for webhook triggers, so the public + /// receiver resolves matches with an indexed (TenantId, WebhookPath) + /// lookup instead of loading and parsing every tenant's trigger JSON. Null for + /// non-webhook triggers and for rows compiled before the column existed. + /// + public string? WebhookPath { get; set; } + /// /// Whether the workflow's auto-trigger (cron/interval/webhook/handler) is /// enabled. Toggled from the Workflows UI; the trigger scheduler skips /// disabled workflows. Manual "run now" is unaffected. diff --git a/Program.cs b/Program.cs index c042e10..a1c782b 100644 --- a/Program.cs +++ b/Program.cs @@ -66,24 +66,27 @@ builder.Services.AddOpenApi(); builder.Services.AddHttpClient(); // HTTP nodes must not follow redirects at the transport layer: the executor // re-checks every hop against the egress policy and follows redirects itself. +// Every egress client pins the vetted address at connect time (EgressPinning) so +// a name that flips between public and private cannot be raced (DNS rebinding). builder.Services.AddHttpClient(HttpRequestNodeExecutor.TypeName) - .ConfigurePrimaryHttpMessageHandler(() => new SocketsHttpHandler { AllowAutoRedirect = false }); -// Opt-in named client for nodes that set ignoreSslIssues; still redirect-vetted. + .ConfigurePrimaryHttpMessageHandler(sp => EgressPinning.CreateHandler( + sp.GetRequiredService(), + sp.GetRequiredService())); +// Opt-in named client for nodes that set ignoreSslIssues; still redirect-vetted +// and address-pinned at connect time. builder.Services.AddHttpClient(HttpRequestNodeExecutor.InsecureClientName) - .ConfigurePrimaryHttpMessageHandler(() => new SocketsHttpHandler - { - AllowAutoRedirect = false, - SslOptions = new SslClientAuthenticationOptions - { - RemoteCertificateValidationCallback = (_, _, _, _) => true, - }, - }); + .ConfigurePrimaryHttpMessageHandler(sp => EgressPinning.CreateHandler( + sp.GetRequiredService(), + sp.GetRequiredService(), + allowInsecureTls: true)); 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). +// must not follow redirects and must connect to a vetted address (a public host +// could otherwise bounce the secret to an internal address after the check). builder.Services.AddHttpClient("credential-test") - .ConfigurePrimaryHttpMessageHandler(() => new SocketsHttpHandler { AllowAutoRedirect = false }); + .ConfigurePrimaryHttpMessageHandler(sp => EgressPinning.CreateHandler( + sp.GetRequiredService(), + sp.GetRequiredService())); // Credential vault encryption keys (used by ICredentialCipher). builder.Services.AddDataProtection(); @@ -149,7 +152,12 @@ else // YAML compile pipeline (step 4). builder.Services.AddSingleton(); builder.Services.AddSingleton(); -builder.Services.AddSingleton(); +// Workflows:LowerLegacyScripts (default off) opts function-mode script YAML into +// compile-time lowering onto the node kernel (S1 migration switch). +builder.Services.AddSingleton(sp => new WorkflowCompiler( + sp.GetRequiredService(), + sp.GetRequiredService(), + builder.Configuration.GetValue("Workflows:LowerLegacyScripts", false))); // Node catalog: the declarative blueprint registry backing GET /api/nodes and // the authoring UI. Core/connector blueprints ship embedded; extra connector @@ -249,8 +257,13 @@ builder.Services.AddSingleton(); builder.Services.AddSingleton(); builder.Services.AddSingleton(); // DB-level node workflow execution: reconstructs the graph from the compiled -// tasks+edges, runs it in-process and records a TaskRun per node. +// tasks+edges, runs it and records a TaskRun per node. Executed by the worker +// (GraphJobExecutor) for `graph.run` jobs, not inline on the lifecycle loop. builder.Services.AddSingleton(); +// Worker half of node-mode execution: consumes `graph.run` jobs off the same +// tenant job streams as `task.run`, so node graphs get uniform leases/timeouts +// and scale across --worker replicas (S2). +builder.Services.AddSingleton(); // Realtime SSE hub — notifies subscribed clients when a workflow file changes server-side. builder.Services.AddSingleton(); @@ -274,6 +287,9 @@ builder.Services.AddScoped(sp => return factory.Create(tenantId); }); builder.Services.AddScoped(); +// Serializes concurrent syncs of the same (tenant, repo) so overlapping requests +// cannot insert the same deterministic ids and collide on the primary key. +builder.Services.AddSingleton(); // Mermaid diagrams (step 6). builder.Services.AddSingleton(); @@ -289,7 +305,7 @@ builder.Services.AddSingleton(new SubprocessScriptExecutor("she builder.Services.AddSingleton(new SubprocessScriptExecutor("python", "python3", builder.Configuration)); builder.Services.AddSingleton(new SubprocessScriptExecutor("javascript", "node", builder.Configuration)); builder.Services.AddSingleton(new TypeScriptExecutor(builder.Configuration)); -builder.Services.AddSingleton(new CSharpScriptExecutor()); +builder.Services.AddSingleton(new CSharpScriptExecutor(builder.Configuration)); // W9: `agent` step type — invokes an LLM agent via chatapi over HTTP (D3), keeping // workflows-api decoupled from BotSharp. builder.Services.AddSingleton(sp => @@ -422,11 +438,12 @@ app.Use(async (context, next) => { try { - var source = await factory.CreateAsync(tenantId, repoName, context.RequestAborted); + // ResolveAsync caches clone/pull + login per (tenant, repo) for a + // short TTL, so this no longer runs git on every authenticated request. + var (source, login) = await factory.ResolveAsync(tenantId, repoName, context.RequestAborted); context.Items["WorkflowSource"] = source; // Store the resolved Forgejo login for controllers that need // to resolve repo paths (WorkflowFilesController, etc.). - var login = await factory.ResolveForgejoLoginAsync(tenantId, context.RequestAborted); if (!string.IsNullOrEmpty(login)) context.Items["ForgejoLogin"] = login; } diff --git a/Services/Audit/SecretRedactor.cs b/Services/Audit/SecretRedactor.cs index 601da09..cc40e1b 100644 --- a/Services/Audit/SecretRedactor.cs +++ b/Services/Audit/SecretRedactor.cs @@ -33,8 +33,13 @@ public static class SecretRedactor RegexOptions.IgnoreCase | RegexOptions.Compiled, Timeout); - /// Returns the text with credential-shaped substrings replaced. - public static string? Redact(string? text) + /// + /// Returns the text with credential-shaped substrings replaced. When the + /// caller knows the actual secret values (e.g. a resolved credential used in a + /// request URL), passing them redacts those literals too — this is what stops + /// a token embedded in a path or query from surviving in a persisted error. + /// + public static string? Redact(string? text, IEnumerable? knownSecrets = null) { if (string.IsNullOrEmpty(text)) return text; @@ -44,6 +49,19 @@ public static class SecretRedactor var redacted = AuthScheme.Replace(text, m => $"{m.Groups["scheme"].Value}{Placeholder}"); redacted = UrlUserInfo.Replace(redacted, m => $"{m.Groups["scheme"].Value}{Placeholder}@"); redacted = KeyedSecret.Replace(redacted, m => $"{m.Groups["key"].Value}={Placeholder}"); + + if (knownSecrets != null) + { + // Longest first so a secret that contains another is fully removed. + foreach (var secret in knownSecrets + .Where(s => !string.IsNullOrWhiteSpace(s) && s.Length >= 6) + .Distinct(StringComparer.Ordinal) + .OrderByDescending(s => s.Length)) + { + redacted = redacted.Replace(secret, Placeholder, StringComparison.Ordinal); + } + } + return redacted; } } diff --git a/Services/Execution/CSharpScriptExecutor.cs b/Services/Execution/CSharpScriptExecutor.cs index 134fbe0..040a0fa 100644 --- a/Services/Execution/CSharpScriptExecutor.cs +++ b/Services/Execution/CSharpScriptExecutor.cs @@ -1,7 +1,10 @@ +using System.Collections.Concurrent; using System.Collections.Immutable; using System.Diagnostics; using System.Reflection; using System.Runtime.Loader; +using System.Security.Cryptography; +using System.Text; using System.Text.Json; using Microsoft.CodeAnalysis; using Microsoft.CodeAnalysis.CSharp; @@ -18,10 +21,17 @@ namespace w4c_workflows.Services.Execution; /// (e.g. Main.cs + tasks/hello-world.cs) into one assembly, /// causing duplicate-type errors when both define a Program class. /// -/// Each execution uses a so the -/// loaded assembly and all its types can be garbage-collected after the call -/// returns. The previous implementation loaded into the default (non-collectible) -/// ALC, causing unbounded memory growth in long-running workers. +/// Each execution uses a so the loaded assembly +/// and all its types can be garbage-collected after the call returns. The +/// compiled image bytes are cached by source hash, so a repeated task +/// skips Roslyn and only re-loads the image into a fresh collectible ALC. +/// +/// The compile + invoke runs off the caller thread under +/// Workflows:CSharpTimeoutSeconds (default TaskTimeoutSeconds) and +/// observes ct: an async entry is awaited with cancellation, and the +/// caller always returns by the deadline. A synchronous entry that ignores the +/// token cannot be force-killed in-process; the caller stops waiting at the +/// deadline and the abandoned thread is left to finish (documented limitation). /// /// Entry contract (v1): a public static method named after /// entry.function (default Main) that takes a single @@ -33,67 +43,134 @@ public class CSharpScriptExecutor : IScriptExecutor { private static readonly JsonSerializerOptions Json = new(JsonSerializerDefaults.Web); + /// Compiled PE images keyed by source hash, so a hot task never re-runs Roslyn. + private static readonly ConcurrentDictionary AssemblyCache = new(StringComparer.Ordinal); + private static readonly ConcurrentQueue CacheOrder = new(); + private const int MaxCachedAssemblies = 64; + // Cached once per process lifetime — the set of runtime assemblies doesn't // change after startup. This avoids re-scanning AppDomain on every compile. private static ImmutableArray? _cachedReferences; private static readonly object _referenceLock = new(); + private readonly TimeSpan _timeout; + + public CSharpScriptExecutor(IConfiguration? config = null) + { + var seconds = ParseInt(config?["Workflows:CSharpTimeoutSeconds"], 0); + if (seconds <= 0) + seconds = ParseInt(config?["Workflows:TaskTimeoutSeconds"], 60); + _timeout = TimeSpan.FromSeconds(Math.Max(1, seconds)); + } + public string Language => "csharp"; /// Roslyn ships with the service image — no external runtime to probe. public bool IsAvailable() => true; - public Task ExecuteAsync(TaskInvocation invocation, CancellationToken ct) + public async Task ExecuteAsync(TaskInvocation invocation, CancellationToken ct) { var sw = Stopwatch.StartNew(); + if (!ExecutionHelpers.TryResolveEntryPath(invocation, out var entryPath, out var pathError)) + return Fail(pathError!, sw.Elapsed); + if (!File.Exists(entryPath)) + return Fail($"entry file not found: {invocation.EntryFile}", sw.Elapsed); + + string source; try { - var workingDir = ExecutionHelpers.ResolveWorkingDir(invocation); - var entryPath = Path.GetFullPath(Path.Combine(workingDir, invocation.EntryFile)); - if (!File.Exists(entryPath)) - return Task.FromResult(Fail($"entry file not found: {invocation.EntryFile}", sw.Elapsed)); - - var source = new SourceFile(invocation.EntryFile, File.ReadAllText(entryPath)); - - // Compilation is CPU-bound and synchronous; keep it off the request thread. - var (assemblyLoadContext, assembly, diagnostics) = Compile(source); - if (assembly == null) - { - assemblyLoadContext?.Unload(); - var errors = diagnostics - .Where(d => d.Severity == DiagnosticSeverity.Error) - .Select(d => d.ToString()); - return Task.FromResult(Fail("C# compilation failed:\n" + string.Join('\n', errors), sw.Elapsed)); - } - - string? output; - try - { - output = InvokeEntry(assembly, invocation); - } - finally - { - // Unload the collectible ALC so the assembly and all its types - // become eligible for GC. The next GC.Collect will reclaim them. - assemblyLoadContext.Unload(); - } - - sw.Stop(); - return Task.FromResult(new ExecutionResult(true, 0, string.Empty, string.Empty, output, null, sw.Elapsed)); + source = File.ReadAllText(entryPath); } - catch (Exception ex) when (ex is not OperationCanceledException) + catch (Exception ex) { + return Fail($"cannot read entry file: {ex.Message}", sw.Elapsed); + } + + // Linked token: fires on caller cancellation OR the hard deadline. Used + // inside the work for async entries; WaitAsync below guarantees the caller + // returns at the deadline even if a sync entry ignores the token. + var linked = CancellationTokenSource.CreateLinkedTokenSource(ct); + linked.CancelAfter(_timeout); + var token = linked.Token; + + try + { + var work = Task.Run(() => CompileAndInvokeAsync(source, entryPath, invocation, token), token); + var output = await work.WaitAsync(_timeout, ct); + sw.Stop(); - return Task.FromResult(new ExecutionResult(false, -1, string.Empty, string.Empty, null, ex.Message, sw.Elapsed)); + return new ExecutionResult(true, 0, string.Empty, string.Empty, output, null, sw.Elapsed); + } + catch (TimeoutException) + { + return Fail($"C# script timed out after {_timeout.TotalSeconds:0}s", sw.Elapsed); + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + // P1-11: cancellation is not a normal script failure. + throw; + } + catch (OperationCanceledException) + { + return Fail($"C# script timed out after {_timeout.TotalSeconds:0}s", sw.Elapsed); + } + catch (Exception ex) + { + return Fail(ex.Message, sw.Elapsed); } } private static ExecutionResult Fail(string message, TimeSpan duration) => new(false, -1, string.Empty, string.Empty, null, message, duration); - private static (AssemblyLoadContext, Assembly?, ImmutableArray) Compile(SourceFile source) + private static async Task CompileAndInvokeAsync( + string source, string entryPath, TaskInvocation invocation, CancellationToken ct) { - var syntaxTree = CSharpSyntaxTree.ParseText(source.Source, path: source.Path); + ct.ThrowIfCancellationRequested(); + var image = GetOrCompile(source, entryPath, ct); + + var alc = new AssemblyLoadContext(name: null, isCollectible: true); + try + { + using var ms = new MemoryStream(image); + var assembly = alc.LoadFromStream(ms); + return await InvokeEntryAsync(assembly, invocation, ct); + } + finally + { + // Unload the collectible ALC so the assembly and all its types become + // eligible for GC. The next GC.Collect will reclaim them. + alc.Unload(); + } + } + + /// + /// Returns the compiled PE image for , compiling on a + /// cache miss. Throws with the Roslyn + /// diagnostics when compilation fails (failures are never cached). + /// + private static byte[] GetOrCompile(string source, string path, CancellationToken ct) + { + var key = Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(source))); + if (AssemblyCache.TryGetValue(key, out var cached)) + return cached; + + var image = Compile(source, path); + ct.ThrowIfCancellationRequested(); + + if (AssemblyCache.TryAdd(key, image)) + { + CacheOrder.Enqueue(key); + while (CacheOrder.Count > MaxCachedAssemblies && CacheOrder.TryDequeue(out var evicted)) + AssemblyCache.TryRemove(evicted, out _); + } + + return image; + } + + private static byte[] Compile(string source, string path) + { + var syntaxTree = CSharpSyntaxTree.ParseText(source, path: path); var compilation = CSharpCompilation.Create( $"wf-cs-{Guid.NewGuid():N}", @@ -105,24 +182,19 @@ public class CSharpScriptExecutor : IScriptExecutor var emit = compilation.Emit(ms); if (!emit.Success) { - var emptyAlc = new AssemblyLoadContext(name: null, isCollectible: true); - return (emptyAlc, null, emit.Diagnostics); + var errors = emit.Diagnostics + .Where(d => d.Severity == DiagnosticSeverity.Error) + .Select(d => d.ToString()); + throw new CSharpCompileException("C# compilation failed:\n" + string.Join('\n', errors)); } - ms.Position = 0; - - // Each execution gets its own collectible ALC. After Unload() + GC, - // the assembly and all its types are reclaimed — no unbounded growth. - var alc = new AssemblyLoadContext(name: null, isCollectible: true); - var assembly = alc.LoadFromStream(ms); - return (alc, assembly, emit.Diagnostics); + return ms.ToArray(); } /// /// Returns the cached set of metadata references for the current runtime. /// The first call scans AppDomain; subsequent calls return the cached - /// immutable array. This eliminates the per-compile AppDomain scan that - /// compounded the ALC leak. + /// immutable array. This eliminates the per-compile AppDomain scan. /// private static ImmutableArray GetCachedReferences() { @@ -172,7 +244,38 @@ public class CSharpScriptExecutor : IScriptExecutor } } - private static string? InvokeEntry(Assembly assembly, TaskInvocation invocation) + private static async Task InvokeEntryAsync( + Assembly assembly, TaskInvocation invocation, CancellationToken ct) + { + var method = FindEntryMethod(assembly, invocation); + if (method == null) + { + var function = string.IsNullOrWhiteSpace(invocation.EntryFunction) ? "Main" : invocation.EntryFunction; + throw new InvalidOperationException($"no static entry method '{function}' found in the compiled C# program"); + } + + var arguments = method.GetParameters().Length == 1 + ? new object?[] { invocation.Input ?? "{}" } + : null; + + ct.ThrowIfCancellationRequested(); + var result = method.Invoke(null, arguments); + + // Await Task / Task returns so async entries produce a concrete value + // and a cancellation is observed instead of blocking the caller. + if (result is Task task) + { + await task.WaitAsync(ct); + result = task.GetType().IsGenericType + ? task.GetType().GetProperty("Result")!.GetValue(task) + : null; + } + + ct.ThrowIfCancellationRequested(); + return result == null ? null : JsonSerializer.Serialize(result, Json); + } + + private static MethodInfo? FindEntryMethod(Assembly assembly, TaskInvocation invocation) { var function = string.IsNullOrWhiteSpace(invocation.EntryFunction) ? "Main" : invocation.EntryFunction; @@ -203,25 +306,12 @@ public class CSharpScriptExecutor : IScriptExecutor break; } - if (method == null) - throw new InvalidOperationException($"no static entry method '{function}' found in the compiled C# program"); - - var arguments = method.GetParameters().Length == 1 - ? new object?[] { invocation.Input ?? "{}" } - : null; - var result = method.Invoke(null, arguments); - - // Await Task / Task returns so async entries produce a concrete value. - if (result is Task task) - { - task.GetAwaiter().GetResult(); - result = task.GetType().IsGenericType - ? task.GetType().GetProperty("Result")!.GetValue(task) - : null; - } - - return result == null ? null : JsonSerializer.Serialize(result, Json); + return method; } - private sealed record SourceFile(string Path, string Source); + private static int ParseInt(string? raw, int fallback) + => int.TryParse(raw, out var value) ? value : fallback; + + /// Raised on Roslyn emit failure; the message carries the diagnostics. + private sealed class CSharpCompileException(string message) : Exception(message); } diff --git a/Services/Execution/ExecutionHelpers.cs b/Services/Execution/ExecutionHelpers.cs index 9d7294e..94ff1b9 100644 --- a/Services/Execution/ExecutionHelpers.cs +++ b/Services/Execution/ExecutionHelpers.cs @@ -24,6 +24,51 @@ internal static class ExecutionHelpers ? Directory.GetCurrentDirectory() : Path.GetFullPath(invocation.WorkingDir); + /// + /// Resolves a task's entry file inside its working directory. A rooted path, + /// or one that escapes via .., is rejected — otherwise a workflow could + /// read or execute files outside its own run directory (Path.Combine + /// silently drops the root when the second argument is absolute). + /// + public static bool TryResolveEntryPath(TaskInvocation invocation, out string entryPath, out string? error) + { + var root = ResolveWorkingDir(invocation); + var candidate = invocation.EntryFile; + + if (string.IsNullOrWhiteSpace(candidate)) + { + entryPath = string.Empty; + error = "entry.file is empty"; + return false; + } + + if (Path.IsPathRooted(candidate)) + { + entryPath = string.Empty; + error = $"entry file '{candidate}' must be relative to the task working directory"; + return false; + } + + var resolved = Path.GetFullPath(Path.Combine(root, candidate)); + var rootWithSeparator = root.EndsWith(Path.DirectorySeparatorChar) + ? root + : root + Path.DirectorySeparatorChar; + var comparison = OperatingSystem.IsWindows() + ? StringComparison.OrdinalIgnoreCase + : StringComparison.Ordinal; + + if (!resolved.StartsWith(rootWithSeparator, comparison)) + { + entryPath = string.Empty; + error = $"entry file '{candidate}' escapes the task working directory"; + return false; + } + + entryPath = resolved; + error = null; + return true; + } + /// WF_* runtime context + the workflow/task-declared env, merged. public static IReadOnlyDictionary BuildEnvironment(TaskInvocation invocation) { diff --git a/Services/Execution/GraphJobExecutor.cs b/Services/Execution/GraphJobExecutor.cs new file mode 100644 index 0000000..f793f40 --- /dev/null +++ b/Services/Execution/GraphJobExecutor.cs @@ -0,0 +1,136 @@ +using Microsoft.EntityFrameworkCore; +using w4c_workflows.Data; +using w4c_workflows.Models; +using w4c_workflows.Services.Messaging; +using w4c_workflows.Services.Nodes; + +namespace w4c_workflows.Services.Execution; + +/// +/// Executes a graph.run job: loads the run and its compiled node graph, +/// drives the whole graph through , then acks +/// the job. This is the worker half of S2 — node-mode execution no longer runs +/// inline on the control-plane lifecycle loop, so a waiting node cannot stall +/// dispatch/result/timeout handling for other tenants and node runs scale across +/// --worker replicas like script tasks. +/// +/// Delivery is at-least-once: a cancelled or crashed execution leaves the job +/// unacked so it is re-claimed after the idle threshold. Idempotency is safe +/// because the graph job carries only a run reference and a run is only executed +/// while it is pending/running; a terminal run is skipped. +/// +public sealed class GraphJobExecutor +{ + private readonly IServiceScopeFactory _scopeFactory; + private readonly IJobQueue _jobs; + private readonly NodeWorkflowRunner _runner; + private readonly ILogger _logger; + + public GraphJobExecutor( + IServiceScopeFactory scopeFactory, + IJobQueue jobs, + NodeWorkflowRunner runner, + ILogger logger) + { + _scopeFactory = scopeFactory; + _jobs = jobs; + _runner = runner; + _logger = logger; + } + + /// + /// Runs one graph job. A malformed or cross-tenant message is dead-lettered; + /// a stale/terminal run is acked without re-running; a cancellation leaves + /// the job pending for redelivery. + /// + public async Task ExecuteAsync(string tenant, StreamMessage message, CancellationToken ct) + { + GraphRunInvocation invocation; + try + { + invocation = GraphRunInvocation.FromFields(message.Fields); + } + catch (Exception ex) + { + _logger.LogWarning("Dead-lettering malformed graph job {MessageId}: {Error}", message.Id, ex.Message); + await _jobs.DeadLetterAsync(tenant, message.Id, message.Fields, "malformed", ct); + return; + } + + if (!string.Equals(invocation.TenantId, tenant, StringComparison.Ordinal)) + { + _logger.LogWarning("Graph job {MessageId} tenant {JobTenant} != stream tenant {StreamTenant}; dead-lettering", + message.Id, invocation.TenantId, tenant); + await _jobs.DeadLetterAsync(tenant, message.Id, message.Fields, "tenant_mismatch", ct); + return; + } + + if (!Guid.TryParse(invocation.RunId, out var runId)) + { + _logger.LogWarning("Dead-lettering graph job {MessageId}: run id '{RunId}' is malformed", + message.Id, invocation.RunId); + await _jobs.DeadLetterAsync(tenant, message.Id, message.Fields, "malformed_run_id", ct); + return; + } + + using var scope = _scopeFactory.CreateScope(); + var db = scope.ServiceProvider.GetRequiredService(); + + var run = await db.WorkflowRuns + .Include(r => r.Workflow).ThenInclude(w => w.Tasks) + .Include(r => r.Workflow).ThenInclude(w => w.TaskEdges) + .FirstOrDefaultAsync(r => r.Id == runId && r.TenantId == tenant, ct); + + if (run == null) + { + // The run row was deleted (or belongs to another tenant). Nothing to + // execute; ack so the job does not redeliver forever. + _logger.LogWarning("Graph job {MessageId} references missing run {RunId}; acking", message.Id, runId); + await _jobs.AckAsync(tenant, message.Id, ct); + return; + } + + // A run the control plane already timed out (or that finished) must not + // be resurrected by a late/duplicate graph job. + if (run.Status is not (RunStatus.Pending or RunStatus.Running)) + { + _logger.LogDebug("Graph job {MessageId} run {RunId} is already {Status}; skipping", + message.Id, runId, run.Status); + await _jobs.AckAsync(tenant, message.Id, ct); + return; + } + + if (run.Status == RunStatus.Pending) + { + run.Status = RunStatus.Running; + run.StartedAt ??= DateTime.UtcNow; + await db.SaveChangesAsync(ct); + } + + try + { + // The runner owns the terminal status and records one TaskRun per + // executed node; it absorbs non-cancellation failures internally. + await _runner.RunAsync(db, run, run.Workflow, invocation.WorkingDir, ct); + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + // Shutdown: do not ack — the job stays pending for redelivery and the + // control-plane timeout sweep recovers the run if this worker never + // comes back. + throw; + } + catch (Exception ex) + { + // Safety net: the runner should already have failed the run, but if + // anything escapes we must not leave it hanging in `running`. + _logger.LogError(ex, "Graph run {RunId} threw outside the runner; failing the run", runId); + run.Status = RunStatus.Failed; + run.Error = ex.Message; + run.FinishedAt = DateTime.UtcNow; + await db.SaveChangesAsync(ct); + } + + await _jobs.AckAsync(tenant, message.Id, ct); + } +} diff --git a/Services/Execution/RemoteServerExecutor.cs b/Services/Execution/RemoteServerExecutor.cs index fdd32ec..a653148 100644 --- a/Services/Execution/RemoteServerExecutor.cs +++ b/Services/Execution/RemoteServerExecutor.cs @@ -152,10 +152,7 @@ public sealed class RemoteServerExecutor }; private string ResolveLocalScriptPath(TaskInvocation invocation) - { - var workDir = ExecutionHelpers.ResolveWorkingDir(invocation); - return Path.Combine(workDir, invocation.EntryFile); - } + => ExecutionHelpers.TryResolveEntryPath(invocation, out var path, out _) ? path : string.Empty; private ExecutionResult Fail(string message, TimeSpan elapsed) { diff --git a/Services/Execution/SubprocessScriptExecutor.cs b/Services/Execution/SubprocessScriptExecutor.cs index e406099..b548798 100644 --- a/Services/Execution/SubprocessScriptExecutor.cs +++ b/Services/Execution/SubprocessScriptExecutor.cs @@ -30,7 +30,8 @@ public class SubprocessScriptExecutor : IScriptExecutor try { var workingDir = ExecutionHelpers.ResolveWorkingDir(invocation); - var entryPath = Path.Combine(workingDir, invocation.EntryFile); + if (!ExecutionHelpers.TryResolveEntryPath(invocation, out var entryPath, out var pathError)) + return new ExecutionResult(false, -1, string.Empty, string.Empty, null, pathError, sw.Elapsed); var (executable, args) = ResolveInterpreter(entryPath); args.Add(entryPath); diff --git a/Services/Execution/TypeScriptExecutor.cs b/Services/Execution/TypeScriptExecutor.cs index c2e52a8..4c787cc 100644 --- a/Services/Execution/TypeScriptExecutor.cs +++ b/Services/Execution/TypeScriptExecutor.cs @@ -32,7 +32,10 @@ public class TypeScriptExecutor : IScriptExecutor try { var workingDir = ExecutionHelpers.ResolveWorkingDir(invocation); - var entryPath = Path.Combine(workingDir, invocation.EntryFile); + if (!ExecutionHelpers.TryResolveEntryPath(invocation, out var entryPath, out var pathError)) + { + return new ExecutionResult(false, -1, string.Empty, string.Empty, null, pathError, sw.Elapsed); + } // 1. Bundle with esbuild → CommonJS so node can run it directly. var build = await ProcessRunner.RunAsync( diff --git a/Services/Execution/WorkerHostService.cs b/Services/Execution/WorkerHostService.cs index f20fe19..9a14bf0 100644 --- a/Services/Execution/WorkerHostService.cs +++ b/Services/Execution/WorkerHostService.cs @@ -3,6 +3,7 @@ using Microsoft.EntityFrameworkCore; using w4c_workflows.Data; using w4c_workflows.Models; using w4c_workflows.Services.Messaging; +using w4c_workflows.Services.Nodes; namespace w4c_workflows.Services.Execution; @@ -26,6 +27,7 @@ public class WorkerHostService : BackgroundService private readonly IEventBus _events; private readonly RuntimeRegistry _runtimes; private readonly RemoteServerExecutor _remote; + private readonly GraphJobExecutor _graphJobs; private readonly IServiceScopeFactory _scopeFactory; private readonly ILogger _logger; @@ -40,6 +42,7 @@ public class WorkerHostService : BackgroundService IEventBus events, RuntimeRegistry runtimes, RemoteServerExecutor remote, + GraphJobExecutor graphJobs, IServiceScopeFactory scopeFactory, IConfiguration config, ILogger logger) @@ -48,6 +51,7 @@ public class WorkerHostService : BackgroundService _events = events; _runtimes = runtimes; _remote = remote; + _graphJobs = graphJobs; _scopeFactory = scopeFactory; _logger = logger; @@ -137,6 +141,16 @@ public class WorkerHostService : BackgroundService private async Task ExecuteJobAsync(string tenant, StreamMessage message, CancellationToken ct) { + // Node-mode runs arrive as one `graph.run` job covering the whole graph; + // script tasks arrive as `task.run`. Dispatch on the wire type before + // parsing the script contract, whose required fields a graph job lacks. + if (message.Fields.TryGetValue("type", out var kind) + && string.Equals(kind, GraphRunInvocation.TypeValue, StringComparison.Ordinal)) + { + await _graphJobs.ExecuteAsync(tenant, message, ct); + return; + } + TaskInvocation invocation; try { diff --git a/Services/ForgejoWorkflowRepoService.cs b/Services/ForgejoWorkflowRepoService.cs index 3a8dd35..f4fdf49 100644 --- a/Services/ForgejoWorkflowRepoService.cs +++ b/Services/ForgejoWorkflowRepoService.cs @@ -77,7 +77,7 @@ public sealed class ForgejoWorkflowRepoService /// token when no admin token is configured. /// private string GitAuthToken => - !string.IsNullOrWhiteSpace(_forgejoAdminToken) ? _forgejoAdminToken : _forgejoToken; + !string.IsNullOrWhiteSpace(_forgejoAdminToken) ? _forgejoAdminToken : _forgejoToken ?? string.Empty; /// True when Forgejo admin provisioning is configured. public bool IsConfigured => @@ -309,23 +309,21 @@ public sealed class ForgejoWorkflowRepoService } var token = GitAuthToken; - var output = !string.IsNullOrWhiteSpace(token) - ? await RunGitAsync(tenantDir, ct, "-c", $"http.extraHeader=Authorization: token {token}", "pull", "--ff-only") - : await RunGitAsync(tenantDir, ct, "pull", "--ff-only"); + var pull = !string.IsNullOrWhiteSpace(token) + ? await RunGitExitAsync(tenantDir, ct, "-c", $"http.extraHeader=Authorization: token {token}", "pull", "--ff-only") + : await RunGitExitAsync(tenantDir, ct, "pull", "--ff-only"); - if (output == null) + if (pull == null || pull.Value.ExitCode != 0) { - _logger.LogWarning("Pull failed for login {Login}", login); + var detail = pull.HasValue + ? $"{pull.Value.Stdout}\n{pull.Value.Stderr}".Trim() + : "git pull did not run"; + _logger.LogWarning("Pull failed for login {Login}: {Out}", login, detail); return false; } - var ok = !output.Contains("fatal:", StringComparison.OrdinalIgnoreCase) && - !output.Contains("Could not resolve host", StringComparison.OrdinalIgnoreCase) && - !output.Contains("Permission denied", StringComparison.OrdinalIgnoreCase); - - _logger.LogInformation("Pull for login {Login}: {Ok} ({Out})", - login, ok, output.Trim()); - return ok; + _logger.LogInformation("Pull for login {Login}: ok", login); + return true; } /// @@ -340,20 +338,37 @@ public sealed class ForgejoWorkflowRepoService return (false, "No local clone"); await RunGitAsync(tenantDir, ct, "add", "-A"); - var commitOutput = await RunGitAsync(tenantDir, ct, "commit", "-m", message); - if (commitOutput == null || commitOutput.Contains("nothing to commit", StringComparison.OrdinalIgnoreCase)) - return (true, "Nothing to commit"); + + var commit = await RunGitExitAsync(tenantDir, ct, "commit", "-m", message); + if (commit == null) + return (false, "git commit could not be executed"); + + if (commit.Value.ExitCode != 0) + { + var commitText = $"{commit.Value.Stdout}\n{commit.Value.Stderr}"; + if (commitText.Contains("nothing to commit", StringComparison.OrdinalIgnoreCase) + || commitText.Contains("no changes added to commit", StringComparison.OrdinalIgnoreCase) + || commitText.Contains("working tree clean", StringComparison.OrdinalIgnoreCase)) + return (true, "Nothing to commit"); + return (false, commitText.Trim()); + } var token = GitAuthToken; - var pushOutput = !string.IsNullOrWhiteSpace(token) - ? await RunGitAsync(tenantDir, ct, "-c", $"http.extraHeader=Authorization: token {token}", "push") - : await RunGitAsync(tenantDir, ct, "push"); + var push = !string.IsNullOrWhiteSpace(token) + ? await RunGitExitAsync(tenantDir, ct, "-c", $"http.extraHeader=Authorization: token {token}", "push") + : await RunGitExitAsync(tenantDir, ct, "push"); - var ok = pushOutput != null && - !pushOutput.Contains("fatal:", StringComparison.OrdinalIgnoreCase) && - !pushOutput.Contains("Permission denied", StringComparison.OrdinalIgnoreCase); + if (push == null) + return (false, "git push could not be executed"); - return (ok, ok ? pushOutput?.Trim() ?? "Pushed" : pushOutput?.Trim() ?? "Push failed"); + // Success is the exit code, not a substring scan of stderr: a rejected + // (non-fast-forward) push writes "! [rejected]" / "failed to push some + // refs" and exits non-zero — the old heuristic reported that as success, + // silently dropping the user's edits. + var output = $"{push.Value.Stdout}\n{push.Value.Stderr}".Trim(); + return push.Value.ExitCode == 0 + ? (true, output.Length == 0 ? "Pushed" : output) + : (false, output.Length == 0 ? "Push failed" : output); } /// @@ -369,12 +384,28 @@ public sealed class ForgejoWorkflowRepoService => Path.GetFullPath(Path.Combine(_copiesRoot, Sanitize(tenantId), OwnerRepoSlug(login, repoName))); /// - /// Runs git with a hard wall-clock timeout. Default git has no connect/transfer - /// timeout, so a dead or unreachable remote (e.g. a stale Forgejo base URL after - /// the server moved) would otherwise block the calling request forever. On timeout - /// the whole process tree is killed and null is returned instead of hanging. + /// Runs git and returns its stdout on success / stderr on failure (null when + /// git could not be run at all). Callers that must distinguish success from + /// failure reliably should use and check the + /// exit code instead of scanning the output text. /// private async Task RunGitAsync(string workDir, CancellationToken ct, params string[] args) + { + var result = await RunGitExitAsync(workDir, ct, args); + if (result == null) + return null; + return result.Value.ExitCode == 0 ? result.Value.Stdout : result.Value.Stderr; + } + + /// + /// Runs git with a hard wall-clock timeout and returns the real exit code plus + /// separately captured stdout/stderr. Default git has no connect/transfer + /// timeout, so a dead or unreachable remote (e.g. a stale Forgejo base URL + /// after the server moved) would otherwise block the calling request forever. + /// On timeout the whole process tree is killed and null is returned. + /// + private async Task<(int ExitCode, string Stdout, string Stderr)?> RunGitExitAsync( + string workDir, CancellationToken ct, params string[] args) { try { @@ -409,9 +440,7 @@ public sealed class ForgejoWorkflowRepoService return null; } - var stdout = await stdoutTask; - var stderr = await stderrTask; - return process.ExitCode == 0 ? stdout : stderr; + return (process.ExitCode, await stdoutTask, await stderrTask); } catch (Exception ex) { diff --git a/Services/JwtValidator.cs b/Services/JwtValidator.cs index fd97eb9..ff35152 100644 --- a/Services/JwtValidator.cs +++ b/Services/JwtValidator.cs @@ -20,7 +20,11 @@ public static class JwtValidator JwtSecurityTokenHandler.DefaultMapInboundClaims = false; } - public static ClaimsPrincipal? Validate(string? authHeader, string signingKey) + public static ClaimsPrincipal? Validate( + string? authHeader, + string signingKey, + string? issuer = null, + string? audience = null) { if (string.IsNullOrEmpty(signingKey) || string.IsNullOrEmpty(authHeader) || !authHeader.StartsWith("Bearer ", StringComparison.OrdinalIgnoreCase)) @@ -29,17 +33,24 @@ public static class JwtValidator try { var key = new SymmetricSecurityKey(Encoding.UTF8.GetBytes(signingKey)); + var parameters = new TokenValidationParameters + { + ValidateIssuerSigningKey = true, + IssuerSigningKey = key, + // Issuer/audience are enforced only when the deployment declares + // them. w4c-auth may not stamp them; a shared symmetric key alone + // is weaker, so operators are encouraged to set Auth:JwtIssuer / + // Auth:JwtAudience and get the extra binding. + ValidateIssuer = !string.IsNullOrWhiteSpace(issuer), + ValidIssuer = issuer, + ValidateAudience = !string.IsNullOrWhiteSpace(audience), + ValidAudience = audience, + ValidateLifetime = true, + ClockSkew = TimeSpan.FromMinutes(1), + }; + return new JwtSecurityTokenHandler().ValidateToken( - authHeader["Bearer ".Length..].Trim(), - new TokenValidationParameters - { - ValidateIssuerSigningKey = true, - IssuerSigningKey = key, - ValidateIssuer = false, - ValidateAudience = false, - ValidateLifetime = true, - ClockSkew = TimeSpan.FromMinutes(1), - }, out _); + authHeader["Bearer ".Length..].Trim(), parameters, out _); } catch { diff --git a/Services/Messaging/InMemoryTransport.cs b/Services/Messaging/InMemoryTransport.cs index 5dd7cf6..7c3c078 100644 --- a/Services/Messaging/InMemoryTransport.cs +++ b/Services/Messaging/InMemoryTransport.cs @@ -86,8 +86,20 @@ public sealed class InMemoryTransport : IJobQueue, IEventBus // ---- Helpers ---- private Channel GetOrAddChannel(string key) - => _channels.GetOrAdd(key, _ => Channel.CreateUnbounded( - new UnboundedChannelOptions { SingleReader = false, SingleWriter = false })); + => _channels.GetOrAdd(key, _ => Channel.CreateBounded( + new BoundedChannelOptions(ChannelCapacity) + { + SingleReader = false, + SingleWriter = false, + // Bounded so a stalled consumer cannot grow memory without limit. + // Under pressure the OLDEST message is dropped: lite mode is + // single-process and jobs/notifications are recoverable from DB + // state, so dropping the stale tail is preferable to OOM. + FullMode = BoundedChannelFullMode.DropOldest, + })); + + /// Per-stream in-memory backlog ceiling. + private const int ChannelCapacity = 10_000; private async Task> DrainAsync(string stream, int count, CancellationToken ct) { diff --git a/Services/Messaging/RedisStreamsTransport.cs b/Services/Messaging/RedisStreamsTransport.cs index ba26deb..b028af2 100644 --- a/Services/Messaging/RedisStreamsTransport.cs +++ b/Services/Messaging/RedisStreamsTransport.cs @@ -1,3 +1,4 @@ +using Microsoft.Extensions.Configuration; using StackExchange.Redis; namespace w4c_workflows.Services.Messaging; @@ -20,6 +21,7 @@ public class RedisStreamsTransport : IJobQueue, IEventBus private readonly string _eventsTemplate; private readonly string _dlqTemplate; private readonly string _group; + private readonly int _maxStreamLength; public RedisStreamsTransport(IConnectionMultiplexer redis, IConfiguration config) { @@ -28,6 +30,10 @@ public class RedisStreamsTransport : IJobQueue, IEventBus _eventsTemplate = config["Workflows:EventStream"] ?? "wf:{tenant}:events"; _dlqTemplate = config["Workflows:DlqStream"] ?? "wf:{tenant}:dlq"; _group = config["Workflows:ConsumerGroup"] ?? "workers"; + // Bound every stream so jobs/events/results/dlq cannot grow without limit + // on a long-lived Redis. 0 disables trimming. Approximate MAXLEN is cheap + // (removes whole radix nodes), so it does not add meaningful latency. + _maxStreamLength = config.GetValue("Workflows:StreamMaxLength", 50_000); } private string JobsStream(string tenantId) => _jobsTemplate.Replace("{tenant}", tenantId); @@ -82,7 +88,9 @@ public class RedisStreamsTransport : IJobQueue, IEventBus private async Task AddAsync(string key, IReadOnlyDictionary fields) { var entries = fields.Select(kv => new NameValueEntry(kv.Key, kv.Value)).ToArray(); - var id = await _db.StreamAddAsync(key, entries); + var id = _maxStreamLength > 0 + ? await _db.StreamAddAsync(key, entries, _maxStreamLength, useApproximateMaxLength: true) + : await _db.StreamAddAsync(key, entries); return id.ToString(); } diff --git a/Services/Nodes/Connectors/RestConnectorExecutor.cs b/Services/Nodes/Connectors/RestConnectorExecutor.cs index 44aa0b8..5123004 100644 --- a/Services/Nodes/Connectors/RestConnectorExecutor.cs +++ b/Services/Nodes/Connectors/RestConnectorExecutor.cs @@ -112,7 +112,14 @@ public sealed class RestConnectorExecutor : INodeExecutor using (response) { var status = (int)response.StatusCode; - var text = await response.Content.ReadAsStringAsync(timeout.Token); + // Same response-size cap as the HTTP node: a connector must not be able + // to buffer an unbounded body into memory (NodeQuotaOptions.MaxResponseBytes). + var read = await HttpBodyReader.ReadAsync(response, _quota.MaxResponseBytes, timeout.Token); + if (read.TooLarge) + return NodeExecutionOutcome.Failed( + $"response exceeded the {_quota.MaxResponseBytes}-byte limit", "response_too_large"); + + var text = read.Body!; if (status >= 300 && !neverError) { return new NodeExecutionOutcome diff --git a/Services/Nodes/Executors/HttpBodyReader.cs b/Services/Nodes/Executors/HttpBodyReader.cs new file mode 100644 index 0000000..8a677ca --- /dev/null +++ b/Services/Nodes/Executors/HttpBodyReader.cs @@ -0,0 +1,63 @@ +using System.Text; + +namespace w4c_workflows.Services.Nodes.Executors; + +/// Raw response body plus a "too large" signal, before content interpretation. +internal sealed record BodyRead(string? Body, byte[]? Bytes, bool TooLarge); + +/// +/// Shared HTTP response-body reader that enforces the per-run response-size quota. +/// A declared Content-Length over the cap is rejected before reading; otherwise +/// the stream is copied in chunks and rejected as soon as the cap is crossed, so +/// an oversized (or unbounded) body never lands in memory. Both the decoded text +/// and the raw bytes are returned, because responseFormat: file needs the +/// exact bytes rather than a lossy string. +/// +internal static class HttpBodyReader +{ + public static async Task ReadAsync( + HttpResponseMessage response, long maxBytes, CancellationToken ct) + { + if (maxBytes <= 0) + { + var bytes = await response.Content.ReadAsByteArrayAsync(ct); + return new BodyRead(ResponseEncoding(response).GetString(bytes), bytes, false); + } + + if (response.Content.Headers.ContentLength is long declared && declared > maxBytes) + return new BodyRead(null, null, true); + + await using var stream = await response.Content.ReadAsStreamAsync(ct); + using var buffer = new MemoryStream(); + var chunk = new byte[81_920]; + int read; + while ((read = await stream.ReadAsync(chunk, ct)) > 0) + { + if (buffer.Length + read > maxBytes) + return new BodyRead(null, null, true); + buffer.Write(chunk, 0, read); + } + + var raw = buffer.ToArray(); + return new BodyRead(ResponseEncoding(response).GetString(raw), raw, false); + } + + /// Content-Type charset when the server declares one, else UTF-8. + private static Encoding ResponseEncoding(HttpResponseMessage response) + { + var charset = response.Content.Headers.ContentType?.CharSet; + if (!string.IsNullOrWhiteSpace(charset)) + { + try + { + return Encoding.GetEncoding(charset.Trim('"')); + } + catch (ArgumentException) + { + // Unknown charset: fall back to UTF-8 rather than failing the node. + } + } + + return Encoding.UTF8; + } +} diff --git a/Services/Nodes/Executors/HttpPaginationPlan.cs b/Services/Nodes/Executors/HttpPaginationPlan.cs index 70f9061..bf7c649 100644 --- a/Services/Nodes/Executors/HttpPaginationPlan.cs +++ b/Services/Nodes/Executors/HttpPaginationPlan.cs @@ -52,10 +52,17 @@ internal sealed class HttpPaginationPlan public IReadOnlySet StopStatusCodes { get; init; } = new HashSet(); - /// Hard cap on the number of requests; 0 disables the cap. + /// + /// Hard cap on the number of requests. Non-positive values fall back to + /// — the cap can never be disabled, otherwise a + /// self-referential nextUrl would loop until quota/timeout. + /// public int MaxPages { get; init; } = DefaultMaxPages; - /// Hard cap on the accumulated items; 0 disables the cap. + /// + /// Hard cap on the accumulated items. Non-positive values fall back to + /// — the cap can never be disabled. + /// public int MaxItems { get; init; } = DefaultMaxItems; /// @@ -86,13 +93,15 @@ internal sealed class HttpPaginationPlan PageStart = NodeValueAccess.ReadInt(config, "pageStart", 1), StopOnEmptyResponse = NodeValueAccess.ReadBool(config, "stopOnEmptyResponse", true), StopStatusCodes = ParseStatusCodes(config["stopStatusCodes"]), - MaxPages = Math.Max(0, NodeValueAccess.ReadInt(config, "maxPages", DefaultMaxPages)), - MaxItems = Math.Max(0, NodeValueAccess.ReadInt(config, "maxItems", DefaultMaxItems)), + MaxPages = PositiveOrDefault(NodeValueAccess.ReadInt(config, "maxPages", DefaultMaxPages), DefaultMaxPages), + MaxItems = PositiveOrDefault(NodeValueAccess.ReadInt(config, "maxItems", DefaultMaxItems), DefaultMaxItems), }; return plan.Validate(); } + private static int PositiveOrDefault(int value, int fallback) => value > 0 ? value : fallback; + private HttpPaginationPlan Validate() { switch (Mode.ToLowerInvariant()) diff --git a/Services/Nodes/Executors/HttpRequestNodeExecutor.cs b/Services/Nodes/Executors/HttpRequestNodeExecutor.cs index a718cb4..5b4df2f 100644 --- a/Services/Nodes/Executors/HttpRequestNodeExecutor.cs +++ b/Services/Nodes/Executors/HttpRequestNodeExecutor.cs @@ -360,7 +360,7 @@ public sealed class HttpRequestNodeExecutor : INodeExecutor using (response!) { var status = (int)response.StatusCode; - var read = await ReadBodyAsync(response, _quota.MaxResponseBytes, timeout.Token); + var read = await HttpBodyReader.ReadAsync(response, _quota.MaxResponseBytes, timeout.Token); if (read.TooLarge) return PageFailure( $"response exceeded the {_quota.MaxResponseBytes}-byte limit", "response_too_large"); @@ -682,62 +682,6 @@ public sealed class HttpRequestNodeExecutor : INodeExecutor // ------------------------------------------------------------------ response - private sealed record BodyRead(string? Body, byte[]? Bytes, bool TooLarge); - - /// - /// Reads the body while enforcing the response-size quota. A declared - /// Content-Length over the cap is rejected before reading; otherwise the - /// stream is copied in chunks and rejected as soon as the cap is crossed, so - /// an oversized (or unbounded) body never lands in memory. Both the decoded - /// text and the raw bytes are returned, because responseFormat: file - /// needs the exact bytes rather than a lossy string. - /// - private static async Task ReadBodyAsync( - HttpResponseMessage response, long maxBytes, CancellationToken ct) - { - if (maxBytes <= 0) - { - var bytes = await response.Content.ReadAsByteArrayAsync(ct); - return new BodyRead(ResponseEncoding(response).GetString(bytes), bytes, false); - } - - if (response.Content.Headers.ContentLength is long declared && declared > maxBytes) - return new BodyRead(null, null, true); - - await using var stream = await response.Content.ReadAsStreamAsync(ct); - using var buffer = new MemoryStream(); - var chunk = new byte[81_920]; - int read; - while ((read = await stream.ReadAsync(chunk, ct)) > 0) - { - if (buffer.Length + read > maxBytes) - return new BodyRead(null, null, true); - buffer.Write(chunk, 0, read); - } - - var raw = buffer.ToArray(); - return new BodyRead(ResponseEncoding(response).GetString(raw), raw, false); - } - - /// Content-Type charset when the server declares one, else UTF-8. - private static Encoding ResponseEncoding(HttpResponseMessage response) - { - var charset = response.Content.Headers.ContentType?.CharSet; - if (!string.IsNullOrWhiteSpace(charset)) - { - try - { - return Encoding.GetEncoding(charset.Trim('"')); - } - catch (ArgumentException) - { - // Unknown charset: fall back to UTF-8 rather than failing the node. - } - } - - return Encoding.UTF8; - } - private sealed record BuildResult(List Items, string? Error); /// diff --git a/Services/Nodes/GraphRunMessage.cs b/Services/Nodes/GraphRunMessage.cs new file mode 100644 index 0000000..d2c031f --- /dev/null +++ b/Services/Nodes/GraphRunMessage.cs @@ -0,0 +1,75 @@ +using System.Globalization; +using w4c_workflows.Models; + +namespace w4c_workflows.Services.Nodes; + +/// +/// Wire contract for a graph.run job. This is the node-kernel analogue of +/// : instead of one script task, the job +/// asks a worker to run the whole persisted node graph of a run through +/// . +/// +/// It exists so node-mode workflows execute on the same worker/queue path as +/// script tasks (S2): the control plane enqueues one graph.run message per +/// run, a worker claims it, reconstructs the graph from the compiled entities and +/// owns the run's terminal status. That gives node runs the same leases, +/// timeouts and scale-out as script runs, instead of blocking the lifecycle +/// dispatch loop for the whole graph. +/// +/// Field names are flat stream entries so the contract is transport-agnostic. +/// +public sealed record GraphRunInvocation( + string Type, + string RunId, + string TenantId, + string? WorkingDir, + int Attempt = 1) +{ + public const string TypeValue = "graph.run"; + + /// Parses a stream message into an invocation, throwing if a required field is missing. + public static GraphRunInvocation FromFields(IReadOnlyDictionary fields) + { + static string Get(IReadOnlyDictionary fields, string key) => + fields.TryGetValue(key, out var value) && !string.IsNullOrEmpty(value) + ? value + : throw new InvalidOperationException($"missing required graph job field '{key}'"); + + static int GetInt(IReadOnlyDictionary fields, string key, int fallback) => + fields.TryGetValue(key, out var raw) && int.TryParse(raw, NumberStyles.Integer, CultureInfo.InvariantCulture, out var n) + ? n + : fallback; + + return new GraphRunInvocation( + Get(fields, "type"), + Get(fields, "run_id"), + Get(fields, "tenant_id"), + fields.GetValueOrDefault("working_dir"), + GetInt(fields, "attempt", 1)); + } +} + +/// +/// Serializes the control-plane side of a graph.run job. Only the run +/// reference and its working directory travel on the wire: the worker reloads +/// the compiled workflow, its task graph and input from the database, so the +/// message cannot drift from persisted state. +/// +public static class GraphRunMessage +{ + public static IReadOnlyDictionary ToFields(WorkflowRun run, string? workingDir, int attempt) + { + var fields = new Dictionary + { + ["type"] = GraphRunInvocation.TypeValue, + ["run_id"] = run.Id.ToString(), + ["tenant_id"] = run.TenantId, + ["attempt"] = attempt.ToString(CultureInfo.InvariantCulture), + }; + + if (!string.IsNullOrWhiteSpace(workingDir)) + fields["working_dir"] = workingDir; + + return fields; + } +} diff --git a/Services/Nodes/LegacyWorkflowLowerer.cs b/Services/Nodes/LegacyWorkflowLowerer.cs index 07615a6..f20488b 100644 --- a/Services/Nodes/LegacyWorkflowLowerer.cs +++ b/Services/Nodes/LegacyWorkflowLowerer.cs @@ -13,6 +13,18 @@ namespace w4c_workflows.Services.Nodes; /// Script execution is unchanged: the core.code executor delegates to the /// existing runtime, so the subprocess /// path is preserved while the edge kernel owns scheduling. +/// +/// This is the S1 migration bridge. It is wired into +/// behind Workflows:LowerLegacyScripts (default off, function mode only): +/// durable/handler scripts keep the linear engine because the node kernel does +/// not checkpoint/resume, and already-persisted script workflows keep running +/// unchanged until an operator opts in. +/// +/// NOTE: the node kernel passes per-node data as FlowItem envelopes +/// ({ "json": … }), while the linear engine passes a task's raw JSON output +/// to its successor. Lowering therefore changes the exact input JSON a script +/// receives. The switch must not be enabled for a workflow whose scripts depend on +/// the raw shape until that contract is bridged. /// public static class LegacyWorkflowLowerer { diff --git a/Services/Nodes/NodeGraphLinks.cs b/Services/Nodes/NodeGraphLinks.cs index 76bdec2..89703b7 100644 --- a/Services/Nodes/NodeGraphLinks.cs +++ b/Services/Nodes/NodeGraphLinks.cs @@ -9,70 +9,109 @@ namespace w4c_workflows.Services.Nodes; /// and at run time it re-triggers that node for the next iteration instead of /// counting toward its readiness. /// -/// The classification is derived from the edge set (a DFS back edge), so it is -/// recomputed identically by the compiler and when a persisted graph is rebuilt. +/// Classification is derived from the edge set, not from traversal order: an edge +/// u → v is a loop-back exactly when it lies on a cycle (the target +/// v can reach u) and v is loop-capable. This is +/// order-independent, so the compiler and a rebuilt persisted graph always agree +/// regardless of task declaration order. /// public static class NodeGraphLinks { /// /// Returns the edges with set, or an - /// error when a cycle passes through a node that is not loop-capable. + /// error when a cycle does not pass through a loop-capable node. /// public static (List Edges, string? Error) MarkLoopBackEdges( IReadOnlyList nodes, IReadOnlyList edges) { var byId = nodes.ToDictionary(n => n.Id, StringComparer.Ordinal); - var adjacency = new Dictionary>(StringComparer.Ordinal); - for (var i = 0; i < edges.Count; i++) + var adjacency = new Dictionary>(StringComparer.Ordinal); + foreach (var edge in edges) { - if (!adjacency.TryGetValue(edges[i].FromNodeId, out var outgoing)) - adjacency[edges[i].FromNodeId] = outgoing = new List(); - outgoing.Add(i); - } - - var color = new Dictionary(StringComparer.Ordinal); // 0 = white, 1 = on stack, 2 = done - var backEdges = new HashSet(); - - void Visit(string nodeId) - { - color[nodeId] = 1; - if (adjacency.TryGetValue(nodeId, out var outgoing)) - { - foreach (var index in outgoing) - { - var target = edges[index].ToNodeId; - var targetColor = color.GetValueOrDefault(target, 0); - if (targetColor == 1) - backEdges.Add(index); - else if (targetColor == 0) - Visit(target); - } - } - color[nodeId] = 2; + if (!adjacency.TryGetValue(edge.FromNodeId, out var outgoing)) + adjacency[edge.FromNodeId] = outgoing = new List(); + outgoing.Add(edge.ToNodeId); } + // reach[v] = every node reachable from v via one or more edges. Edge u→v + // closes a cycle iff u is reachable from v. + var reach = new Dictionary>(StringComparer.Ordinal); foreach (var node in nodes) { - if (color.GetValueOrDefault(node.Id, 0) == 0) - Visit(node.Id); + var seen = new HashSet(StringComparer.Ordinal); + var stack = new Stack(); + stack.Push(node.Id); + while (stack.Count > 0) + { + var current = stack.Pop(); + if (!adjacency.TryGetValue(current, out var next)) + continue; + foreach (var target in next) + { + if (seen.Add(target)) + stack.Push(target); + } + } + reach[node.Id] = seen; } + bool ClosesCycle(NodeGraphEdge edge) + => reach.TryGetValue(edge.ToNodeId, out var fromTarget) + && fromTarget.Contains(edge.FromNodeId); + var marked = new List(edges.Count); - string? error = null; - for (var i = 0; i < edges.Count; i++) + foreach (var edge in edges) { - var edge = edges[i]; - var isLoopBack = backEdges.Contains(i); - if (isLoopBack && !byId[edge.ToNodeId].Blueprint.LoopBack) - { - error ??= - $"a cycle through '{edge.FromNodeId}' → '{edge.ToNodeId}' is not allowed: " + - $"'{byId[edge.ToNodeId].Blueprint.Type}' is not a loop node"; - } + var isLoopBack = ClosesCycle(edge) && byId[edge.ToNodeId].Blueprint.LoopBack; marked.Add(edge with { IsLoopBack = isLoopBack }); } + // A cycle is only valid when it re-enters a loop-capable node. Group the + // mutually-reachable nodes into SCCs and require a loop node in every + // cyclic component; otherwise the graph would deadlock or spin. + var error = ValidateCyclesHaveLoopNodes(nodes, adjacency, reach, byId); return (marked, error); } + + private static string? ValidateCyclesHaveLoopNodes( + IReadOnlyList nodes, + IReadOnlyDictionary> adjacency, + IReadOnlyDictionary> reach, + IReadOnlyDictionary byId) + { + var ids = nodes.Select(n => n.Id).ToList(); + var assigned = new HashSet(StringComparer.Ordinal); + + foreach (var id in ids) + { + if (!assigned.Add(id)) + continue; + + var members = new List { id }; + foreach (var other in ids) + { + if (other == id || assigned.Contains(other)) + continue; + if (reach[id].Contains(other) && reach[other].Contains(id)) + { + members.Add(other); + assigned.Add(other); + } + } + + var isCyclic = members.Count > 1 + || adjacency.TryGetValue(id, out var self) && self.Contains(id); + if (!isCyclic) + continue; + + if (members.Any(m => byId[m].Blueprint.LoopBack)) + continue; + + return $"a cycle involving {string.Join(", ", members)} is not allowed: " + + "none of these nodes is a loop node (e.g. core.splitInBatches)"; + } + + return null; + } } diff --git a/Services/Nodes/NodeGraphRunner.cs b/Services/Nodes/NodeGraphRunner.cs index dce0f29..c68818e 100644 --- a/Services/Nodes/NodeGraphRunner.cs +++ b/Services/Nodes/NodeGraphRunner.cs @@ -114,10 +114,12 @@ public sealed class NodeGraphRunner var arrived = graph.Nodes.ToDictionary(n => n.Id, _ => 0, StringComparer.Ordinal); var inputs = graph.Nodes.ToDictionary(n => n.Id, NewInputPorts, StringComparer.Ordinal); var exhausted = new HashSet(StringComparer.Ordinal); + var queued = new HashSet(StringComparer.Ordinal); inputs[graph.EntryNodeId][0].AddRange(seed); var ready = new Queue(); ready.Enqueue(graph.EntryNodeId); + queued.Add(graph.EntryNodeId); var hasLoops = graph.Edges.Any(e => e.IsLoopBack); var maxSteps = hasLoops @@ -132,6 +134,7 @@ public sealed class NodeGraphRunner return Failure(outputs, order, new NodeFailure("node graph exceeded its execution budget", "budget")); var nodeId = ready.Dequeue(); + queued.Remove(nodeId); var node = graph.Require(nodeId); if (!_executors.CanRun(node.Blueprint.Type)) @@ -172,7 +175,7 @@ public sealed class NodeGraphRunner if (outcome.LoopComplete) exhausted.Add(nodeId); - Propagate(graph, nodeId, ports, inputs, arrived, incoming, ready, exhausted, outcome.LoopComplete); + Propagate(graph, nodeId, ports, inputs, arrived, incoming, ready, queued, exhausted, outcome.LoopComplete); // Consume this node's inputs so a loop can gather them again next // iteration; cleared after use, never before. @@ -319,6 +322,7 @@ public sealed class NodeGraphRunner IDictionary arrived, IReadOnlyDictionary incoming, Queue ready, + ISet queued, ISet exhausted, bool skipEmptyPorts) { @@ -341,7 +345,12 @@ public sealed class NodeGraphRunner targetInputs[edge.ToInput].AddRange(produced); arrived[edge.ToNodeId]++; - if (arrived[edge.ToNodeId] == incoming[edge.ToNodeId]) + // Max(incoming, 1): a node with no non-loop-back predecessor (an + // entry loop node) must still be re-triggered by its loop-back + // edge. The queued guard prevents a double enqueue when several + // inputs arrive before the node next runs. + if (arrived[edge.ToNodeId] >= Math.Max(incoming[edge.ToNodeId], 1) + && queued.Add(edge.ToNodeId)) ready.Enqueue(edge.ToNodeId); } } diff --git a/Services/Nodes/NodeWorkflowRunner.cs b/Services/Nodes/NodeWorkflowRunner.cs index d57a9ac..75e84e7 100644 --- a/Services/Nodes/NodeWorkflowRunner.cs +++ b/Services/Nodes/NodeWorkflowRunner.cs @@ -13,12 +13,17 @@ namespace w4c_workflows.Services.Nodes; public sealed record NodeWorkflowRunOutcome(bool Succeeded, string? Error, string? Output); /// -/// Runs a persisted node-mode workflow through in -/// the control plane, replacing the linear NextId subprocess chain for -/// node definitions (script workflows keep the subprocess path). Reconstructs -/// the executable graph from the stored tasks + -/// rows, writes a row per executed node for history, and -/// records the run's terminal output/status. +/// Runs a persisted node-mode workflow through , +/// replacing the linear NextId subprocess chain for node definitions +/// (script workflows keep the legacy subprocess path in +/// ). Reconstructs the executable graph from +/// the stored tasks + rows, writes a +/// row per executed node for history, and records the run's +/// terminal output/status. +/// +/// Invoked by the worker () when it +/// claims a graph.run job, and recursively in-process by +/// for core.executeWorkflow nodes. /// public sealed class NodeWorkflowRunner { @@ -98,6 +103,11 @@ public sealed class NodeWorkflowRunner return await FailAsync(db, run, ex.Message, ct); } + // The literal secret values resolved for this run. They are scrubbed from + // every persisted/logged error so a token embedded in a URL path or query + // (connector secrets are substituted into the URL) never reaches history. + var secretValues = CollectSecretValues(credentials); + var environment = new NodeRunEnvironment { TenantId = run.TenantId, @@ -112,7 +122,7 @@ public sealed class NodeWorkflowRunner db, this, run, build.TaskIds, ancestry, _maxSubWorkflowDepth, _logger), }; - var listener = BuildListener(db, run, build.TaskIds, recordTaskRuns); + var listener = BuildListener(db, run, build.TaskIds, recordTaskRuns, secretValues); NodeGraphRunResult result; try @@ -122,12 +132,12 @@ public sealed class NodeWorkflowRunner } 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); + _logger?.LogError("Node workflow run {RunId} threw: {Error}", run.Id, SecretRedactor.Redact(ex.Message, secretValues)); + return await FailAsync(db, run, ex.Message, ct, secretValues); } if (!result.Succeeded) - return await FailAsync(db, run, result.Failure!.Message, ct); + return await FailAsync(db, run, result.Failure!.Message, ct, secretValues); var output = TerminalOutput(result); run.Status = RunStatus.Succeeded; @@ -240,8 +250,13 @@ public sealed class NodeWorkflowRunner } private async Task FailAsync( - WorkflowsDbContext db, WorkflowRun run, string error, CancellationToken ct) + WorkflowsDbContext db, + WorkflowRun run, + string error, + CancellationToken ct, + IReadOnlyCollection? secretValues = null) { + error = SecretRedactor.Redact(error, secretValues) ?? error; _logger?.LogWarning("Node workflow run {RunId} failed: {Error}", run.Id, error); run.Status = RunStatus.Failed; @@ -252,6 +267,42 @@ public sealed class NodeWorkflowRunner return new NodeWorkflowRunOutcome(false, error, null); } + /// + /// Flattens every string value in the resolved credentials. Used to scrub the + /// actual secret literals from error messages, since the regex redactor cannot + /// know an arbitrary token embedded in a URL path or query. + /// + private static List CollectSecretValues(IReadOnlyDictionary credentials) + { + var values = new List(); + foreach (var credential in credentials.Values) + { + foreach (var (_, node) in credential.Data) + CollectJsonStrings(node, values); + } + return values; + } + + private static void CollectJsonStrings(JsonNode? node, List values) + { + switch (node) + { + case JsonValue value when value.TryGetValue(out var text) + && !string.IsNullOrWhiteSpace(text) + && text.Length >= 6: + values.Add(text); + break; + case JsonObject obj: + foreach (var (_, child) in obj) + CollectJsonStrings(child, values); + break; + case JsonArray array: + foreach (var child in array) + CollectJsonStrings(child, values); + break; + } + } + // ------------------------------------------------------------------ history /// @@ -263,11 +314,12 @@ public sealed class NodeWorkflowRunner WorkflowsDbContext db, WorkflowRun run, IReadOnlyDictionary taskIds, - bool recordTaskRuns) + bool recordTaskRuns, + IReadOnlyCollection? secretValues = null) { var listeners = new List(2); if (recordTaskRuns) - listeners.Add(new TaskRunHistory(db, run, taskIds)); + listeners.Add(new TaskRunHistory(db, run, taskIds, secretValues)); if (_audit != null) listeners.Add(new AuditingNodeRunListener(_audit, run)); @@ -289,13 +341,19 @@ public sealed class NodeWorkflowRunner private readonly WorkflowsDbContext _db; private readonly WorkflowRun _run; private readonly IReadOnlyDictionary _taskIds; + private readonly IReadOnlyCollection? _secretValues; private readonly Dictionary _rows = new(StringComparer.Ordinal); - public TaskRunHistory(WorkflowsDbContext db, WorkflowRun run, IReadOnlyDictionary taskIds) + public TaskRunHistory( + WorkflowsDbContext db, + WorkflowRun run, + IReadOnlyDictionary taskIds, + IReadOnlyCollection? secretValues = null) { _db = db; _run = run; _taskIds = taskIds; + _secretValues = secretValues; } public async Task NodeStartedAsync( @@ -331,7 +389,7 @@ public sealed class NodeWorkflowRunner row.Status = failure == null ? TaskRunStatus.Succeeded : TaskRunStatus.Failed; row.OutputJson = MainOutput(outputs); - row.Error = failure?.Message; + row.Error = SecretRedactor.Redact(failure?.Message, _secretValues); row.FinishedAt = DateTime.UtcNow; await _db.SaveChangesAsync(ct); diff --git a/Services/PerTenantWorkflowSource.cs b/Services/PerTenantWorkflowSource.cs index 30377cd..04f3f47 100644 --- a/Services/PerTenantWorkflowSource.cs +++ b/Services/PerTenantWorkflowSource.cs @@ -1,3 +1,4 @@ +using System.Collections.Concurrent; using System.Diagnostics; using Npgsql; using NpgsqlTypes; @@ -36,6 +37,15 @@ public sealed class WorkflowSourceFactory private readonly string _copiesRoot; private readonly ForgejoWorkflowRepoService? _forgejo; private readonly NpgsqlDataSource? _ds; + private readonly TimeSpan _pullInterval; + + // Per-(tenant, repo) cache of the resolved source + Forgejo login so clone/pull + // and the login lookup do not run on every authenticated HTTP request. A + // per-key gate collapses concurrent refreshes into a single clone/pull. + private readonly ConcurrentDictionary _cache = new(StringComparer.Ordinal); + private readonly ConcurrentDictionary _gates = new(StringComparer.Ordinal); + + private sealed record CachedSource(IWorkflowSource Source, string? Login, DateTime CreatedAt); public WorkflowSourceFactory(IConfiguration config, IWebHostEnvironment env, ILoggerFactory loggers, NpgsqlDataSource? ds = null) @@ -43,6 +53,11 @@ public sealed class WorkflowSourceFactory _loggers = loggers; _ds = ds; + _pullInterval = TimeSpan.FromSeconds( + int.TryParse(config["WorkflowSource:PullIntervalSeconds"], out var seconds) && seconds > 0 + ? seconds + : 30); + _copiesRoot = ResolveCopiesRoot( config["WorkflowSource:CopiesRoot"] ?? throw new InvalidOperationException( @@ -112,29 +127,69 @@ public sealed class WorkflowSourceFactory /// In filesystem-only mode, delegates to . /// public async Task CreateAsync(string tenantId, string? repoName = null, CancellationToken ct = default) + => (await ResolveAsync(tenantId, repoName, ct)).Source; + + /// + /// Resolves the per-tenant source and its Forgejo login, caching the result + /// for WorkflowSource:PullIntervalSeconds (default 30s) so clone/pull and + /// the login lookup run at most once per interval instead of on every request. + /// Concurrent refreshes for the same tenant+repo are collapsed into one. + /// + public async Task<(IWorkflowSource Source, string? Login)> ResolveAsync( + string tenantId, string? repoName = null, CancellationToken ct = default) { if (_forgejo == null) - return Create(tenantId); + return (Create(tenantId), null); - // Resolve the user's Forgejo login from the tenant ID (forgejo_id). - // The repo lives under the user's own account: {login}/{workflowRepoName}. - var login = await ResolveForgejoLoginAsync(tenantId, ct); - if (string.IsNullOrEmpty(login)) + var key = tenantId + "|" + (repoName ?? string.Empty); + if (TryGetFresh(key, out var cached)) + return (cached!.Source, cached.Login); + + var gate = _gates.GetOrAdd(key, _ => new SemaphoreSlim(1, 1)); + await gate.WaitAsync(ct); + try { - _loggers.CreateLogger() - .LogWarning("Could not resolve Forgejo login for tenant {TenantId}, falling back to filesystem", - tenantId); - return Create(tenantId); + if (TryGetFresh(key, out cached)) + return (cached!.Source, cached.Login); + + // Resolve the user's Forgejo login from the tenant ID (forgejo_id). + // The repo lives under the user's own account: {login}/{workflowRepoName}. + var login = await ResolveForgejoLoginAsync(tenantId, ct); + if (string.IsNullOrEmpty(login)) + { + _loggers.CreateLogger() + .LogWarning("Could not resolve Forgejo login for tenant {TenantId}, falling back to filesystem", + tenantId); + return (Create(tenantId), null); + } + + // Ensure the Forgejo repo exists and is cloned locally. + var cloneDir = await _forgejo.EnsureCloneAsync(tenantId, login, repoName, ct); + + // Pull latest changes before reading (from the SELECTED repo, not the default). + await _forgejo.PullAsync(tenantId, login, repoName, ct); + + var logger = _loggers.CreateLogger(); + var source = new PerTenantWorkflowSource(tenantId, cloneDir, logger); + _cache[key] = new CachedSource(source, login, DateTime.UtcNow); + return (source, login); } + finally + { + gate.Release(); + } + } - // Ensure the Forgejo repo exists and is cloned locally. - var cloneDir = await _forgejo.EnsureCloneAsync(tenantId, login, repoName, ct); + /// Drops the cached source for a tenant+repo, forcing a fresh clone/pull next time. + public void Invalidate(string tenantId, string? repoName = null) + => _cache.TryRemove(tenantId + "|" + (repoName ?? string.Empty), out _); - // Pull latest changes before reading (from the SELECTED repo, not the default). - await _forgejo.PullAsync(tenantId, login, repoName, ct); - - var logger = _loggers.CreateLogger(); - return new PerTenantWorkflowSource(tenantId, cloneDir, logger); + private bool TryGetFresh(string key, out CachedSource? cached) + { + if (_cache.TryGetValue(key, out cached) && DateTime.UtcNow - cached.CreatedAt < _pullInterval) + return true; + cached = null; + return false; } /// diff --git a/Services/RealtimeEventHub.cs b/Services/RealtimeEventHub.cs index c703d42..da25da7 100644 --- a/Services/RealtimeEventHub.cs +++ b/Services/RealtimeEventHub.cs @@ -5,19 +5,29 @@ namespace w4c_workflows.Services; /// /// Server-push hub for tenant-scoped realtime events (SSE), mirroring w4c-webapi's hub. A connected -/// client gets its own unbounded channel; writes a tiny notification into every -/// channel for the tenant. The client decides what to re-fetch, keeping broadcasts cheap. +/// client gets its own bounded channel; writes a tiny notification into every +/// channel for the tenant. The client decides what to re-fetch, keeping broadcasts cheap. The channel +/// is bounded and drops the oldest notification when a client cannot keep up, so one stalled SSE +/// consumer cannot grow server memory without limit (the next notification makes it re-fetch anyway). /// /// Used to make the frontend an optional view: when a workflow file is written server-side (e.g. by /// the AI transform), every subscribed client in the tenant is notified and reloads the file. /// public sealed class RealtimeEventHub { + /// Per-client notification backlog ceiling. + private const int ClientChannelCapacity = 256; + private readonly ConcurrentDictionary>> _clients = new(); public (Guid Id, ChannelReader Reader) AddClient(string tenantId) { - var channel = Channel.CreateUnbounded(); + var channel = Channel.CreateBounded(new BoundedChannelOptions(ClientChannelCapacity) + { + SingleReader = true, + SingleWriter = false, + FullMode = BoundedChannelFullMode.DropOldest, + }); var id = Guid.NewGuid(); var clients = _clients.GetOrAdd(tenantId, _ => new ConcurrentDictionary>()); clients[id] = channel; diff --git a/Services/Runs/RunLifecycleEngine.cs b/Services/Runs/RunLifecycleEngine.cs index c820b55..73acf38 100644 --- a/Services/Runs/RunLifecycleEngine.cs +++ b/Services/Runs/RunLifecycleEngine.cs @@ -4,7 +4,6 @@ using w4c_workflows.Data; using w4c_workflows.Models; using w4c_workflows.Services.Execution; using w4c_workflows.Services.Messaging; -using w4c_workflows.Services.Nodes; namespace w4c_workflows.Services.Runs; @@ -35,7 +34,6 @@ public class RunLifecycleEngine private readonly IEventBus _events; private readonly ILeaseService _leases; private readonly ILogger _logger; - private readonly NodeWorkflowRunner? _nodeRunner; private readonly string _owner; private readonly int _batchSize; @@ -53,13 +51,11 @@ public class RunLifecycleEngine IEventBus events, ILeaseService leases, IConfiguration config, - ILogger logger, - NodeWorkflowRunner? nodeRunner = null) + ILogger logger) { _events = events; _leases = leases; _logger = logger; - _nodeRunner = nodeRunner; _owner = $"runs-{Environment.MachineName}-{Guid.NewGuid():N}"[..28]; _batchSize = ParseInt(config["Workflows:LifecycleBatchSize"], 10); @@ -127,11 +123,27 @@ public class RunLifecycleEngine return false; } - // Node-mode workflows are driven by the edge graph in-process: the node - // kernel runs the whole graph and records a TaskRun per node. Script - // workflows keep the linear NextId → task.run subprocess chain below. + // Node-mode workflows run on the worker/queue path (S2): the control + // plane enqueues one `graph.run` job and a worker executes the whole + // edge graph. This keeps a waiting node (up to 5 min) off the lifecycle + // dispatch/result/timeout loop, and lets `--worker` replicas scale node + // runs exactly like script runs. Script workflows keep the linear + // NextId → task.run chain below (legacy compatibility layer). if (IsNodeWorkflow(workflow)) - return await DispatchNodeRunAsync(db, dispatcher, run, workflow, ct); + { + try + { + return await EnqueueNodeRunAsync(db, dispatcher, run, workflow, ct); + } + finally + { + // The run is marked Running before the job is queued, so the + // pending-run scan can no longer pick it up. Release the dispatch + // lease immediately instead of pinning it for the worker's whole + // (possibly minutes-long) graph execution. + await _leases.ReleaseAsync(run.TenantId, run.Id.ToString(), _owner); + } + } var startTask = run.StartTaskId != null ? workflow.Tasks.FirstOrDefault(t => t.Id == run.StartTaskId) @@ -154,41 +166,25 @@ public class RunLifecycleEngine } /// - /// Runs a node-mode workflow to completion in-process and releases the run - /// lease. Unlike the script path (which enqueues one job per task and waits - /// for its result), the node kernel executes the whole edge graph in one go, - /// writing a per node. + /// Marks a node-mode run running and enqueues one graph.run + /// job. Unlike the script path (one job per task + a result per task), a + /// single job carries the whole run: the worker reconstructs the graph from + /// the compiled entities, executes it through NodeGraphRunner and owns + /// the terminal run status. The dispatch lease is released by the caller. /// - private async Task DispatchNodeRunAsync( + private async Task EnqueueNodeRunAsync( WorkflowsDbContext db, TaskDispatcher dispatcher, WorkflowRun run, Workflow workflow, CancellationToken ct) { - if (_nodeRunner == null) - { - await FailRunAsync(db, run, "node workflows are not supported on this runtime (node runner not configured)", ct); - return false; - } - run.Status = RunStatus.Running; run.StartedAt = DateTime.UtcNow; await db.SaveChangesAsync(ct); var workingDir = await dispatcher.ResolveWorkingDirAsync(run.TenantId, workflow.Path, ct); - - 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); - } + await dispatcher.DispatchGraphAsync(run, workingDir, ct); return true; } diff --git a/Services/Runs/TaskDispatcher.cs b/Services/Runs/TaskDispatcher.cs index 1317449..77b45f8 100644 --- a/Services/Runs/TaskDispatcher.cs +++ b/Services/Runs/TaskDispatcher.cs @@ -2,6 +2,7 @@ using w4c_workflows.Data; using w4c_workflows.Models; using w4c_workflows.Services.Execution; using w4c_workflows.Services.Messaging; +using w4c_workflows.Services.Nodes; namespace w4c_workflows.Services.Runs; @@ -76,6 +77,27 @@ public class TaskDispatcher return taskRun.Id; } + /// + /// Enqueues a graph.run job for a node-mode run. Unlike + /// this writes no row: the + /// node kernel records one history row per executed node when the worker runs + /// the graph. One job covers the whole graph, so the run is executed on the + /// worker/queue path with uniform leases, timeouts and scale-out (S2). + /// + public async Task DispatchGraphAsync( + WorkflowRun run, + string? workingDir, + CancellationToken ct, + int attempt = 1) + { + var fields = GraphRunMessage.ToFields(run, workingDir, attempt); + await _jobs.EnqueueAsync(run.TenantId, fields, ct); + + _logger.LogDebug( + "Dispatched node-graph job (run {RunId}, attempt {Attempt}) for tenant {TenantId}", + run.Id, attempt, run.TenantId); + } + /// /// Dead-letters a failed task (retries exhausted, or a failed compensation) /// by reconstructing its task.run fields and publishing them to the diff --git a/Services/Security/EgressGuard.cs b/Services/Security/EgressGuard.cs index ced26e5..150345b 100644 --- a/Services/Security/EgressGuard.cs +++ b/Services/Security/EgressGuard.cs @@ -7,7 +7,11 @@ namespace w4c_workflows.Services.Security; /// Authorises an outbound before a workflow node sends it. /// Names are resolved up front and every returned address is checked, so a host /// that maps to both a public and a private address is rejected rather than -/// raced at connect time. Literal IPs skip DNS entirely. +/// accepted. The connect-time half of the defence is +/// : egress clients re-resolve and re-validate the +/// address in their ConnectCallback, so a name that answers differently on +/// the second lookup (DNS rebinding) cannot reach a blocked address. +/// Literal IPs skip DNS entirely. /// public sealed class EgressGuard { diff --git a/Services/Security/EgressPinning.cs b/Services/Security/EgressPinning.cs new file mode 100644 index 0000000..5e430f5 --- /dev/null +++ b/Services/Security/EgressPinning.cs @@ -0,0 +1,83 @@ +using System.Net; +using System.Net.Sockets; + +namespace w4c_workflows.Services.Security; + +/// +/// Builds the used for workflow egress with a +/// that resolves the target and +/// re-validates every address against the at the moment +/// of connecting. +/// +/// This closes the DNS-rebinding (TOCTOU) hole: the pre-flight +/// check and the actual socket connect no +/// longer rely on two independent DNS resolutions that an attacker-controlled +/// name could answer differently. The socket is opened against a vetted IP while +/// the request keeps its original Host/SNI, and a host that resolves to any +/// blocked address is refused outright — matching +/// semantics. +/// +public static class EgressPinning +{ + public static SocketsHttpHandler CreateHandler( + EgressPolicy policy, + IHostAddressResolver resolver, + bool allowAutoRedirect = false, + bool allowInsecureTls = false) + { + var handler = new SocketsHttpHandler + { + AllowAutoRedirect = allowAutoRedirect, + ConnectCallback = async (context, ct) => + { + var endpoint = context.DnsEndPoint; + + IReadOnlyList addresses; + if (IPAddress.TryParse(endpoint.Host, out var literal)) + addresses = new[] { literal }; + else + addresses = await resolver.ResolveAsync(endpoint.Host, ct); + + // Re-validate at connect time; an empty set fails closed. + if (!policy.SkipsAddressChecks(endpoint.Host)) + { + var decision = policy.CheckAddresses(endpoint.Host, addresses); + if (!decision.Allowed) + throw new HttpRequestException($"egress blocked: {decision.Reason}"); + } + + Exception? last = null; + foreach (var address in addresses) + { + var socket = new Socket(address.AddressFamily, SocketType.Stream, ProtocolType.Tcp) + { + NoDelay = true, + }; + try + { + await socket.ConnectAsync(new IPEndPoint(address, endpoint.Port), ct); + return new NetworkStream(socket, ownsSocket: true); + } + catch (Exception ex) + { + socket.Dispose(); + last = ex; + } + } + + throw new HttpRequestException( + $"egress blocked: could not connect to a permitted address for '{endpoint.Host}'", last); + }, + }; + + if (allowInsecureTls) + { + handler.SslOptions = new System.Net.Security.SslClientAuthenticationOptions + { + RemoteCertificateValidationCallback = (_, _, _, _) => true, + }; + } + + return handler; + } +} diff --git a/Services/Security/EgressPolicy.cs b/Services/Security/EgressPolicy.cs index 8175ce5..a2be778 100644 --- a/Services/Security/EgressPolicy.cs +++ b/Services/Security/EgressPolicy.cs @@ -20,6 +20,15 @@ public sealed class EgressPolicy /// Configured ceiling on redirects a node invocation may follow. public int MaxRedirects => Math.Max(0, _options.MaxRedirects); + /// + /// True when reserved-range address checks must be skipped for + /// : the operator either allowed private networks + /// globally or explicitly allow-listed the host. Used by the connect-time + /// pinning so it does not reject an intentionally-permitted private target. + /// + public bool SkipsAddressChecks(string host) + => _options.AllowPrivateNetworks || MatchesAny(host, _options.AllowedHosts); + /// /// Stage one: validate the scheme and the host allow/deny lists. Returns a /// final decision, or null when the caller must resolve the host and diff --git a/Services/SyncGate.cs b/Services/SyncGate.cs new file mode 100644 index 0000000..b97452d --- /dev/null +++ b/Services/SyncGate.cs @@ -0,0 +1,36 @@ +using System.Collections.Concurrent; + +namespace w4c_workflows.Services; + +/// +/// Serializes concurrent workflow syncs for the same (tenant, repo) within +/// one process. Two overlapping POST /sync calls would otherwise both +/// insert the same deterministic workflow/task ids and collide on the primary key +/// (DbUpdateException). The gate makes the second call wait, so it observes +/// the first call's committed rows and upserts instead of duplicating. +/// +/// For multi-replica deployments this is combined with the distributed +/// lease in . +/// +public sealed class SyncGate +{ + private readonly ConcurrentDictionary _gates = new(StringComparer.Ordinal); + + public async Task AcquireAsync(string tenantId, string repoName, CancellationToken ct) + { + var gate = _gates.GetOrAdd(tenantId + "|" + repoName, _ => new SemaphoreSlim(1, 1)); + await gate.WaitAsync(ct); + return new Releaser(gate); + } + + private sealed class Releaser(SemaphoreSlim gate) : IDisposable + { + private int _released; + + public void Dispose() + { + if (Interlocked.Exchange(ref _released, 1) == 0) + gate.Release(); + } + } +} diff --git a/Services/WorkflowCompiler.cs b/Services/WorkflowCompiler.cs index 0a8ae3e..89680ca 100644 --- a/Services/WorkflowCompiler.cs +++ b/Services/WorkflowCompiler.cs @@ -32,12 +32,20 @@ public sealed class CompileResult /// upserts instead of duplicating. The compiler performs no DB writes — the /// sync service persists the result. /// -/// Two execution modes are supported: -/// - node mode: every step declares a catalog node; steps are -/// validated by and compiled with their port -/// edges. -/// - script mode (legacy): language + entry tasks driven -/// by the linear next/onError chain. +/// Two execution models exist during the S1 convergence: +/// - node mode (target): every step declares a catalog node; steps +/// are validated by and compiled with their +/// port edges. This is the only model the worker kernel executes. +/// - script mode (legacy compatibility): language + entry +/// tasks driven by the linear next/onError chain. It is kept +/// for already-persisted workflows and for durable/handler +/// checkpoint/resume, which the node kernel does not yet cover. +/// +/// Setting Workflows:LowerLegacyScripts=true lowers a function-mode +/// script definition to the node kernel at compile time via +/// (synthetic core.code nodes). This +/// is the migration switch for new authoring; it is off by default so no +/// persisted script workflow changes shape until an operator opts in. /// The two never mix within one definition. /// public class WorkflowCompiler @@ -53,12 +61,17 @@ public class WorkflowCompiler private readonly WorkflowValidator _validator; private readonly NodeGraphCompiler _nodeCompiler; private readonly IDeserializer _yaml; + private readonly bool _lowerLegacyScripts; - public WorkflowCompiler(WorkflowValidator validator, NodeGraphCompiler? nodeCompiler = null) + public WorkflowCompiler( + WorkflowValidator validator, + NodeGraphCompiler? nodeCompiler = null, + bool lowerLegacyScripts = false) { _validator = validator; _nodeCompiler = nodeCompiler ?? new NodeGraphCompiler(new NodeBlueprintCatalog(NodeBlueprintCatalog.LoadEmbedded())); + _lowerLegacyScripts = lowerLegacyScripts; _yaml = new DeserializerBuilder() .WithNamingConvention(CamelCaseNamingConvention.Instance) .WithAttemptingUnquotedStringTypeDeserialization() @@ -94,6 +107,30 @@ public class WorkflowCompiler return result; } + // Legacy script mode. When the migration switch is on, a function-mode + // script definition is lowered to the node kernel so it runs on the same + // edge graph as node workflows. Durable/handler scripts stay on the + // linear engine: the node kernel has no checkpoint/resume yet. + if (_lowerLegacyScripts && IsLowerableLegacyScript(def)) + { + // Validate the original script definition first so script-specific + // diagnostics are preserved, then compile the lowered node graph. + result.Errors.AddRange(_validator.Validate(def)); + if (!result.Success) + return result; + + var lowered = LegacyWorkflowLowerer.ToNodeDefinition(def); + result.Errors.AddRange(_validator.ValidateNodeWorkflow(lowered)); + + var loweredGraph = _nodeCompiler.Compile(lowered, tenantId); + result.Errors.AddRange(loweredGraph.Errors); + if (!result.Success) + return result; + + result.Workflow = BuildNodeWorkflow(lowered, loweredGraph.Graph!, path, tenantId, repoName); + return result; + } + result.Errors.AddRange(_validator.Validate(def)); if (!result.Success) return result; @@ -102,6 +139,24 @@ public class WorkflowCompiler return result; } + /// + /// True for a legacy script definition the node kernel can replace without + /// losing semantics. Only function mode qualifies: durable and + /// handler need the linear engine's checkpoint/resume. + /// + private static bool IsLowerableLegacyScript(WorkflowDefinition def) + => string.Equals(def.Mode, WorkflowMode.Function, StringComparison.OrdinalIgnoreCase); + + /// + /// The denormalized webhook path stored on : + /// only a webhook trigger contributes, so the public receiver can filter by + /// the indexed column instead of parsing every trigger. + /// + private static string? WebhookPathOf(TriggerDefinition? trigger) + => trigger != null && string.Equals(trigger.Type, TriggerType.Webhook, StringComparison.OrdinalIgnoreCase) + ? trigger.WebhookPath + : null; + private CompiledWorkflow Build(WorkflowDefinition def, string path, string tenantId, string repoName) { var graph = WorkflowGraph.Compute(def); @@ -120,6 +175,7 @@ public class WorkflowCompiler Version = string.IsNullOrWhiteSpace(def.Version) ? "1.0.0" : def.Version!, Target = string.IsNullOrWhiteSpace(def.Target) ? "default" : def.Target, TriggerJson = def.Trigger == null ? null : JsonSerializer.Serialize(def.Trigger, Json), + WebhookPath = WebhookPathOf(def.Trigger), CompiledAt = now, CreatedAt = now, UpdatedAt = now, @@ -213,6 +269,7 @@ public class WorkflowCompiler Version = string.IsNullOrWhiteSpace(def.Version) ? "1.0.0" : def.Version!, Target = string.IsNullOrWhiteSpace(def.Target) ? "default" : def.Target, TriggerJson = def.Trigger == null ? null : JsonSerializer.Serialize(def.Trigger, Json), + WebhookPath = WebhookPathOf(def.Trigger), CompiledAt = now, CreatedAt = now, UpdatedAt = now, diff --git a/Services/WorkflowSyncService.cs b/Services/WorkflowSyncService.cs index f5cc238..b0bf4d2 100644 --- a/Services/WorkflowSyncService.cs +++ b/Services/WorkflowSyncService.cs @@ -32,22 +32,67 @@ public class WorkflowSyncService private readonly IWorkflowSource _source; private readonly WorkflowSourceFactory? _sourceFactory; private readonly ILogger _logger; + private readonly SyncGate? _gate; + private readonly ILeaseService? _leases; public WorkflowSyncService( WorkflowsDbContext db, WorkflowCompiler compiler, IWorkflowSource source, ILogger logger, - WorkflowSourceFactory? sourceFactory = null) + WorkflowSourceFactory? sourceFactory = null, + SyncGate? gate = null, + ILeaseService? leases = null) { _db = db; _compiler = compiler; _source = source; _logger = logger; _sourceFactory = sourceFactory; + _gate = gate; + _leases = leases; } + /// + /// Serializes the sync against other syncs of the same (tenant, repo): + /// an in-process gate for this replica plus a best-effort distributed lease for + /// other replicas. Without this, two overlapping syncs insert the same + /// deterministic ids and collide on the primary key (P1-9). + /// public async Task SyncAsync(string tenantId, string repoName, CancellationToken ct) + { + using var gate = _gate == null ? null : await _gate.AcquireAsync(tenantId, repoName, ct); + + var owner = Guid.NewGuid().ToString("N"); + const string leaseKey = "sync"; + var held = false; + if (_leases != null) + { + // Wait up to ~2s for another replica to finish, then proceed best-effort: + // the deterministic ids make a subsequent sync an upsert, not a duplicate. + for (var attempt = 0; attempt < 20 && !held; attempt++) + { + held = await _leases.AcquireAsync(tenantId, $"{leaseKey}:{repoName}", owner, TimeSpan.FromMinutes(2)); + if (!held) + await Task.Delay(100, ct); + } + + if (!held) + _logger.LogDebug("Sync lease for tenant {TenantId} repo {Repo} was busy; proceeding best-effort", tenantId, repoName); + } + + try + { + return await SyncCoreAsync(tenantId, repoName, ct); + } + finally + { + if (held) + await _leases!.ReleaseAsync(tenantId, $"{leaseKey}:{repoName}", owner); + } + } + + private async Task SyncCoreAsync(string tenantId, string repoName, CancellationToken ct) { // Resolve the source for the SELECTED repo. The request-scoped `_source` // was built from whatever repo was current when the request started, which diff --git a/appsettings.json b/appsettings.json index 505fca5..1547770 100644 --- a/appsettings.json +++ b/appsettings.json @@ -31,6 +31,7 @@ "MaxRetries": 3, "RunTimeoutSeconds": 300, "TaskTimeoutSeconds": 60, + "CSharpTimeoutSeconds": 60, "EsbuildBinary": "esbuild", "NodeBinary": "node", "TenantId": "", @@ -43,6 +44,8 @@ "HandlerBatchSize": 10, "HandlerClaimIdleSeconds": 30, "WebhookSecret": "", + "WebhookSecrets": {}, + "WebhookMaxBodyBytes": 1048576, "WorkerRoot": "", "ServerApiUrl": "", "ServerApiToken": "", @@ -56,6 +59,7 @@ "LifecycleLeaseTtlSeconds": 60, "RetryBaseDelayMs": 1000, "RetryMaxDelayMs": 30000, + "LowerLegacyScripts": false, "Quota": { "Enforced": true, "RunsPerMonth": 1000 @@ -110,6 +114,7 @@ "WorkflowSource": { "CopiesRoot": "../source-copies", "SharedDir": "workflows", - "WorkflowRepoName": "workflows" + "WorkflowRepoName": "workflows", + "PullIntervalSeconds": 30 } } diff --git a/w4c-workflows-api.Tests/BoundedChannelTests.cs b/w4c-workflows-api.Tests/BoundedChannelTests.cs new file mode 100644 index 0000000..09710c3 --- /dev/null +++ b/w4c-workflows-api.Tests/BoundedChannelTests.cs @@ -0,0 +1,46 @@ +using w4c_workflows.Services; +using w4c_workflows.Services.Messaging; +using Xunit; + +namespace w4c_workflows.Tests; + +/// +/// Backlog-bound regression tests: a stalled in-memory consumer must not be able +/// to grow server memory without limit. Both channels are bounded and drop the +/// oldest entry under pressure. +/// +public class BoundedChannelTests +{ + [Fact] + public void RealtimeEventHub_bounds_the_per_client_backlog() + { + var hub = new RealtimeEventHub(); + var (id, reader) = hub.AddClient("tenant-a"); + + for (var i = 0; i < 300; i++) + hub.Publish("tenant-a", "workflow", i.ToString(), DateTime.UtcNow); + + var received = 0; + while (reader.TryRead(out _)) + received++; + + // Client channel capacity is 256 and DropOldest keeps the newest entries. + Assert.Equal(256, received); + + hub.RemoveClient("tenant-a", id); + } + + [Fact] + public async Task InMemoryTransport_bounds_the_job_backlog() + { + var transport = new InMemoryTransport(); + + for (var i = 0; i < 10_500; i++) + await transport.EnqueueAsync("tenant-b", new Dictionary { ["n"] = i.ToString() }, default); + + var drained = await transport.ReadGroupAsync("tenant-b", "consumer", 20_000, default); + + Assert.True(drained.Count > 0, "no messages were readable"); + Assert.True(drained.Count <= 10_000, $"job backlog was not bounded: {drained.Count}"); + } +} diff --git a/w4c-workflows-api.Tests/CSharpScriptExecutorTests.cs b/w4c-workflows-api.Tests/CSharpScriptExecutorTests.cs index 783033e..189b46b 100644 --- a/w4c-workflows-api.Tests/CSharpScriptExecutorTests.cs +++ b/w4c-workflows-api.Tests/CSharpScriptExecutorTests.cs @@ -1,4 +1,8 @@ +using System.Reflection; +using System.Security.Cryptography; +using System.Text; using System.Text.Json; +using Microsoft.Extensions.Configuration; using w4c_workflows.Services.Execution; using Xunit; @@ -195,4 +199,94 @@ public class CSharpScriptExecutorTests Assert.Contains("compilation failed", result.Error, StringComparison.OrdinalIgnoreCase); Assert.Contains("CS1525", result.Error, StringComparison.OrdinalIgnoreCase); } + + // ------------------------------------------------------- P0-6 hardening + + [Fact] + public async Task Compiled_assembly_is_cached_by_source_hash() + { + using var dir = new TempDir(); + // Unique source so the cache key is unique to this test. + var marker = Guid.NewGuid().ToString("N"); + dir.Write("Program.cs", $$""" + public static class Program + { + public static object Main(string input) => new { marker = "{{marker}}" }; + } + """); + + var source = File.ReadAllText(Path.Combine(dir.Path, "Program.cs")); + var key = Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(source))); + + Assert.False(CacheContains(key)); + + var invocation = Invocation.For("csharp", "Program.cs", "{}", dir.Path); + var first = await _executor.ExecuteAsync(invocation, default); + Assert.True(first.Success, first.Error); + Assert.True(CacheContains(key)); + + // Second run is a cache hit; it must succeed and the image is reused. + var second = await _executor.ExecuteAsync(invocation, default); + Assert.True(second.Success, second.Error); + Assert.Contains(marker, second.Output); + } + + [Fact] + public async Task Times_out_when_the_entry_exceeds_the_deadline() + { + var config = new ConfigurationBuilder() + .AddInMemoryCollection(new Dictionary { ["Workflows:CSharpTimeoutSeconds"] = "1" }) + .Build(); + var executor = new CSharpScriptExecutor(config); + + using var dir = new TempDir(); + dir.Write("Program.cs", """ + using System.Threading.Tasks; + public static class Program + { + public static async Task Main(string input) + { + await Task.Delay(10000); + return new { ok = true }; + } + } + """); + + var result = await executor.ExecuteAsync( + Invocation.For("csharp", "Program.cs", "{}", dir.Path), default); + + Assert.False(result.Success); + Assert.Contains("timed out", result.Error, StringComparison.OrdinalIgnoreCase); + } + + [Fact] + public async Task Cancellation_is_propagated_not_swallowed() + { + using var dir = new TempDir(); + dir.Write("Program.cs", """ + using System.Threading.Tasks; + public static class Program + { + public static async Task Main(string input) + { + await Task.Delay(10000); + return new { ok = true }; + } + } + """); + + using var cts = new CancellationTokenSource(); + cts.Cancel(); + + await Assert.ThrowsAnyAsync( + () => _executor.ExecuteAsync(Invocation.For("csharp", "Program.cs", "{}", dir.Path), cts.Token)); + } + + private static bool CacheContains(string key) + { + var field = typeof(CSharpScriptExecutor).GetField( + "AssemblyCache", BindingFlags.NonPublic | BindingFlags.Static)!; + var cache = field.GetValue(null)!; + return (bool)cache.GetType().GetMethod("ContainsKey")!.Invoke(cache, new object[] { key })!; + } } diff --git a/w4c-workflows-api.Tests/EgressPinningTests.cs b/w4c-workflows-api.Tests/EgressPinningTests.cs new file mode 100644 index 0000000..c110c49 --- /dev/null +++ b/w4c-workflows-api.Tests/EgressPinningTests.cs @@ -0,0 +1,36 @@ +using System.Net; +using w4c_workflows.Services.Security; +using Xunit; + +namespace w4c_workflows.Tests; + +/// +/// P0-8: egress clients must re-validate the resolved address at connect time, so +/// a DNS name that answers with a public address for the pre-flight guard and a +/// private one for the real connection cannot be reached. +/// +public class EgressPinningTests +{ + [Fact] + public async Task Connect_is_blocked_when_the_host_resolves_to_a_private_address() + { + var resolver = StubHostAddressResolver.Returning("10.1.2.3"); + using var handler = EgressPinning.CreateHandler(EgressTestData.Policy(), resolver); + using var client = new HttpClient(handler); + + var ex = await Assert.ThrowsAsync( + () => client.GetAsync("http://rebind.example/resource")); + + // The connect callback reports the block; the transport may wrap it. + Assert.Contains("egress blocked", ex.ToString(), StringComparison.OrdinalIgnoreCase); + Assert.Contains("rebind.example", resolver.Queries); + } + + [Fact] + public void Skips_address_checks_for_allow_listed_hosts_and_private_mode() + { + Assert.True(EgressTestData.Policy(o => o.AllowPrivateNetworks = true).SkipsAddressChecks("any.test")); + Assert.True(EgressTestData.Policy(o => o.AllowedHosts = ["allowed.test"]).SkipsAddressChecks("allowed.test")); + Assert.False(EgressTestData.Policy().SkipsAddressChecks("blocked.test")); + } +} diff --git a/w4c-workflows-api.Tests/EntryPathContainmentTests.cs b/w4c-workflows-api.Tests/EntryPathContainmentTests.cs new file mode 100644 index 0000000..92fe16c --- /dev/null +++ b/w4c-workflows-api.Tests/EntryPathContainmentTests.cs @@ -0,0 +1,46 @@ +using Microsoft.Extensions.Configuration; +using w4c_workflows.Services.Execution; +using Xunit; + +namespace w4c_workflows.Tests; + +/// +/// Entry-file containment: a task must never be able to execute (or read) a file +/// outside its own working directory, whether via an absolute path or a +/// .. escape. +/// +public class EntryPathContainmentTests +{ + private static TaskInvocation Invocation(string entryFile, string workingDir) => new( + TaskInvocation.TypeValue, "run-1", "task-1", "key", "shell", entryFile, null, + new Dictionary(), null, workingDir, "tenant-1", 1); + + private static SubprocessScriptExecutor Executor() + => new("shell", "sh", new ConfigurationBuilder().Build()); + + [Fact] + public async Task Absolute_entry_file_is_rejected() + { + var result = await Executor().ExecuteAsync(Invocation("/etc/passwd", Path.GetTempPath()), default); + + Assert.False(result.Success); + Assert.Contains("must be relative", result.Error); + } + + [Fact] + public async Task Entry_file_escaping_the_working_directory_is_rejected() + { + var dir = Directory.CreateTempSubdirectory("w4c-contain-"); + try + { + var result = await Executor().ExecuteAsync(Invocation("../outside.sh", dir.FullName), default); + + Assert.False(result.Success); + Assert.Contains("escapes", result.Error); + } + finally + { + dir.Delete(recursive: true); + } + } +} diff --git a/w4c-workflows-api.Tests/Fakes.cs b/w4c-workflows-api.Tests/Fakes.cs index 8f5682c..195de6c 100644 --- a/w4c-workflows-api.Tests/Fakes.cs +++ b/w4c-workflows-api.Tests/Fakes.cs @@ -8,11 +8,17 @@ internal sealed class FakeJobQueue : IJobQueue { public List<(string TenantId, IReadOnlyDictionary Fields)> Enqueued { get; } = new(); public List<(string TenantId, IReadOnlyDictionary Fields, string Reason)> DeadLettered { get; } = new(); + public List Acked { get; } = new(); + + /// When set, throws — exercises failure paths. + public bool FailEnqueue { get; set; } public Task EnsureGroupAsync(string tenantId, CancellationToken ct) => Task.CompletedTask; public Task EnqueueAsync(string tenantId, IReadOnlyDictionary fields, CancellationToken ct) { + if (FailEnqueue) + throw new InvalidOperationException("enqueue failed"); Enqueued.Add((tenantId, fields)); return Task.FromResult(Enqueued.Count.ToString()); } @@ -23,7 +29,11 @@ internal sealed class FakeJobQueue : IJobQueue public Task> ClaimPendingAsync(string tenantId, string consumer, TimeSpan minIdle, int count, CancellationToken ct) => Task.FromResult>(Array.Empty()); - public Task AckAsync(string tenantId, string messageId, CancellationToken ct) => Task.FromResult(0L); + public Task AckAsync(string tenantId, string messageId, CancellationToken ct) + { + Acked.Add(messageId); + return Task.FromResult(0L); + } public Task DeadLetterAsync(string tenantId, string messageId, IReadOnlyDictionary fields, string reason, CancellationToken ct) { diff --git a/w4c-workflows-api.Tests/GraphJobExecutorTests.cs b/w4c-workflows-api.Tests/GraphJobExecutorTests.cs new file mode 100644 index 0000000..2d6c05d --- /dev/null +++ b/w4c-workflows-api.Tests/GraphJobExecutorTests.cs @@ -0,0 +1,341 @@ +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using w4c_workflows.Data; +using w4c_workflows.Models; +using w4c_workflows.Models.Credentials; +using w4c_workflows.Models.Nodes; +using w4c_workflows.Services; +using w4c_workflows.Services.Credentials; +using w4c_workflows.Services.Execution; +using w4c_workflows.Services.Messaging; +using w4c_workflows.Services.Nodes; +using w4c_workflows.Services.Nodes.Executors; +using w4c_workflows.Services.Nodes.Interpolation; +using Xunit; + +namespace w4c_workflows.Tests; + +/// +/// S2 worker half: a graph.run job is claimed by the worker and drives the +/// whole node graph through , recording one +/// per node and owning the terminal run status. +/// +[Collection("WorkflowsPostgres")] +public class GraphJobExecutorTests +{ + private readonly WorkflowsPostgresFixture _fixture; + + public GraphJobExecutorTests(WorkflowsPostgresFixture fixture) + { + _fixture = fixture; + } + + private static (GraphJobExecutor Executor, FakeJobQueue Jobs) NewExecutor( + WorkflowsDbContext db, + IEnumerable? extraExecutors = null, + CredentialVault? vault = null) + { + var catalog = new NodeBlueprintCatalog(NodeBlueprintCatalog.LoadEmbedded()); + var executors = new List + { + new NoOpNodeExecutor(), + new SetNodeExecutor(), + new IfNodeExecutor(), + new SplitInBatchesNodeExecutor(), + new ExecuteWorkflowNodeExecutor(), + new CodeNodeExecutor(new RuntimeRegistry(new LanguageRegistry(), Array.Empty())), + }; + if (extraExecutors != null) + executors.AddRange(extraExecutors); + + var graphRunner = new NodeGraphRunner(new NodeExecutorRegistry(executors), new NodeParameterInterpolator()); + var nodeRunner = new NodeWorkflowRunner( + catalog, graphRunner, vault ?? new CredentialVault(new ReversibleTestCipher(), new CredentialTypeCatalog())); + + var jobs = new FakeJobQueue(); + var services = new ServiceCollection(); + services.AddSingleton(db); + var provider = services.BuildServiceProvider(); + + var executor = new GraphJobExecutor( + provider.GetRequiredService(), + jobs, + nodeRunner, + NullLogger.Instance); + + return (executor, jobs); + } + + private static StreamMessage GraphJob(WorkflowRun run, string? workingDir = null) + => new("1-0", GraphRunMessage.ToFields(run, workingDir, 1)); + + private static string Tenant() => "t" + Guid.NewGuid().ToString("N")[..12]; + + [Fact] + public async Task Executes_the_graph_and_records_a_task_run_per_node() + { + var tenantId = Tenant(); + await using var db = _fixture.CreateContext(); + var workflow = WorkflowDataHelpers.CompileAndSave( + db, tenantId, WorkflowDataHelpers.NodeYaml, "workflows/node-exec.yaml"); + var run = WorkflowDataHelpers.AddPendingRun(db, workflow); + + var (executor, jobs) = NewExecutor(db); + await executor.ExecuteAsync(tenantId, GraphJob(run), default); + + var reloaded = await db.WorkflowRuns.SingleAsync(r => r.Id == run.Id); + Assert.Equal(RunStatus.Succeeded, reloaded.Status); + Assert.NotNull(reloaded.OutputJson); + Assert.Contains("copied", reloaded.OutputJson); + + var taskRuns = await db.TaskRuns.Where(t => t.RunId == run.Id).ToListAsync(); + Assert.Equal(2, taskRuns.Count); + Assert.All(taskRuns, t => Assert.Equal(TaskRunStatus.Succeeded, t.Status)); + + Assert.Contains("1-0", jobs.Acked); + } + + [Fact] + public async Task Run_is_marked_running_before_the_graph_executes() + { + var tenantId = Tenant(); + await using var db = _fixture.CreateContext(); + var workflow = WorkflowDataHelpers.CompileAndSave( + db, tenantId, WorkflowDataHelpers.NodeYaml, "workflows/node-running.yaml"); + var run = WorkflowDataHelpers.AddRun(db, workflow, RunStatus.Pending); + + var (executor, _) = NewExecutor(db); + await executor.ExecuteAsync(tenantId, GraphJob(run), default); + + var reloaded = await db.WorkflowRuns.SingleAsync(r => r.Id == run.Id); + Assert.Equal(RunStatus.Succeeded, reloaded.Status); + Assert.NotNull(reloaded.StartedAt); + Assert.NotNull(reloaded.FinishedAt); + } + + [Fact] + public async Task Node_without_an_executor_fails_the_run() + { + var tenantId = Tenant(); + await using var db = _fixture.CreateContext(); + var workflow = WorkflowDataHelpers.CompileAndSave( + db, tenantId, WorkflowDataHelpers.NodeNoExecutorYaml, "workflows/node-fail.yaml"); + var run = WorkflowDataHelpers.AddPendingRun(db, workflow); + + var (executor, jobs) = NewExecutor(db); + await executor.ExecuteAsync(tenantId, GraphJob(run), default); + + var reloaded = await db.WorkflowRuns.SingleAsync(r => r.Id == run.Id); + Assert.Equal(RunStatus.Failed, reloaded.Status); + Assert.Contains("no executor", reloaded.Error); + Assert.Contains("1-0", jobs.Acked); + } + + [Fact] + public async Task Loop_runs_and_records_each_iteration() + { + var tenantId = Tenant(); + await using var db = _fixture.CreateContext(); + const string yaml = """ + name: loop-run + tasks: + - id: source + node: { type: core.noop } + edges: + - { output: 0, to: loop } + - id: loop + node: { type: core.splitInBatches } + parameters: { batchSize: 2 } + edges: + - { output: 0, to: body } + - { output: 1, to: done } + - id: body + node: { type: core.noop } + edges: + - { output: 0, to: loop } + - id: done + node: { type: core.noop } + """; + var workflow = WorkflowDataHelpers.CompileAndSave(db, tenantId, yaml, "workflows/loop.yaml"); + Assert.Contains(workflow.TaskEdges, e => e.IsLoopBack); + + var run = WorkflowDataHelpers.AddPendingRun( + db, workflow, """[{"n":1},{"n":2},{"n":3},{"n":4},{"n":5}]"""); + + var (executor, _) = NewExecutor(db); + await executor.ExecuteAsync(tenantId, GraphJob(run), default); + + var reloaded = await db.WorkflowRuns.SingleAsync(r => r.Id == run.Id); + Assert.Equal(RunStatus.Succeeded, reloaded.Status); + + var loopTask = workflow.Tasks.Single(t => t.Key == "loop"); + var iterations = await db.TaskRuns.CountAsync(t => t.RunId == run.Id && t.TaskId == loopTask.Id); + Assert.Equal(4, iterations); // 3 batches + the final done invocation + } + + [Fact] + public async Task Resolves_declared_credentials_from_the_vault() + { + var tenantId = Tenant(); + await using var db = _fixture.CreateContext(); + var vault = new CredentialVault(new ReversibleTestCipher(), new CredentialTypeCatalog()); + await vault.CreateAsync( + db, tenantId, "my-api", "httpHeaderAuth", + new System.Text.Json.Nodes.JsonObject { ["name"] = "X-API-Key", ["value"] = "abc" }, default); + + var yaml = """ + name: cred-run + tasks: + - id: fetch + node: { type: core.httpRequest } + parameters: { url: "https://api.example.com" } + credentials: { httpAuth: my-api } + """; + var workflow = WorkflowDataHelpers.CompileAndSave(db, tenantId, yaml, "workflows/cred.yaml"); + var run = WorkflowDataHelpers.AddPendingRun(db, workflow); + + var capturing = new CapturingHttpExecutor(); + var (executor, _) = NewExecutor(db, new[] { capturing }, vault); + await executor.ExecuteAsync(tenantId, GraphJob(run), default); + + Assert.NotNull(capturing.Seen); + Assert.Equal("httpHeaderAuth", capturing.Seen!["httpAuth"].Type); + Assert.Equal("abc", capturing.Seen!["httpAuth"].Data["value"]!.GetValue()); + + Assert.Equal(RunStatus.Succeeded, (await db.WorkflowRuns.SingleAsync(r => r.Id == run.Id)).Status); + } + + [Fact] + public async Task A_terminal_run_is_skipped_and_acked_not_reexecuted() + { + var tenantId = Tenant(); + await using var db = _fixture.CreateContext(); + var workflow = WorkflowDataHelpers.CompileAndSave( + db, tenantId, WorkflowDataHelpers.NodeYaml, "workflows/node-terminal.yaml"); + var run = WorkflowDataHelpers.AddRun(db, workflow, RunStatus.Succeeded); + + var (executor, jobs) = NewExecutor(db); + await executor.ExecuteAsync(tenantId, GraphJob(run), default); + + Assert.Equal(RunStatus.Succeeded, (await db.WorkflowRuns.SingleAsync(r => r.Id == run.Id)).Status); + Assert.False(await db.TaskRuns.AnyAsync(t => t.RunId == run.Id)); + Assert.Contains("1-0", jobs.Acked); + } + + [Fact] + public async Task Malformed_job_is_dead_lettered() + { + var tenantId = Tenant(); + await using var db = _fixture.CreateContext(); + var (executor, jobs) = NewExecutor(db); + + var fields = new Dictionary + { + ["type"] = GraphRunInvocation.TypeValue, + ["tenant_id"] = tenantId, + // run_id deliberately missing + }; + + await executor.ExecuteAsync(tenantId, new StreamMessage("9-0", fields), default); + + var dlq = Assert.Single(jobs.DeadLetteredFor(tenantId)); + Assert.Equal("malformed", dlq.Reason); + } + + [Fact] + public async Task Tenant_mismatch_is_dead_lettered() + { + var tenantId = Tenant(); + await using var db = _fixture.CreateContext(); + var workflow = WorkflowDataHelpers.CompileAndSave( + db, tenantId, WorkflowDataHelpers.NodeYaml, "workflows/node-mismatch.yaml"); + var run = WorkflowDataHelpers.AddPendingRun(db, workflow); + + var (executor, jobs) = NewExecutor(db); + await executor.ExecuteAsync("other-tenant", GraphJob(run), default); + + var dlq = Assert.Single(jobs.DeadLetteredFor("other-tenant")); + Assert.Equal("tenant_mismatch", dlq.Reason); + Assert.Empty(jobs.Acked); + } + + [Fact] + public async Task Cancellation_leaves_the_job_unacked_for_redelivery() + { + var tenantId = Tenant(); + await using var db = _fixture.CreateContext(); + var workflow = WorkflowDataHelpers.CompileAndSave( + db, tenantId, WorkflowDataHelpers.NodeYaml, "workflows/node-cancel.yaml"); + var run = WorkflowDataHelpers.AddPendingRun(db, workflow); + + var (executor, jobs) = NewExecutor(db); + using var cts = new CancellationTokenSource(); + cts.Cancel(); + + await Assert.ThrowsAnyAsync( + () => executor.ExecuteAsync(tenantId, GraphJob(run), cts.Token)); + + Assert.Empty(jobs.Acked); + } + + /// Stands in for the HTTP executor and records the credentials it was handed. + private sealed class CapturingHttpExecutor : INodeExecutor + { + public string Type => "core.httpRequest"; + public IReadOnlyDictionary? Seen { get; private set; } + + public Task RunAsync(NodeExecutionContext context, CancellationToken ct) + { + Seen = context.Credentials; + return Task.FromResult(NodeExecutionOutcome.Single(context.Input(0).ToList())); + } + } + + /// Fails with the resolved secret embedded in the URL (P1-10 regression). + private sealed class LeakyFailExecutor : INodeExecutor + { + public string Type => "core.httpRequest"; + + public Task RunAsync(NodeExecutionContext context, CancellationToken ct) + { + var secret = context.Credentials["httpAuth"].Data["value"]!.GetValue(); + return Task.FromResult( + NodeExecutionOutcome.Failed($"GET https://api.example.com/{secret}/items failed", "request_failed")); + } + } + + [Fact] + public async Task Credential_literals_are_redacted_from_run_and_task_errors() + { + var tenantId = Tenant(); + await using var db = _fixture.CreateContext(); + var vault = new CredentialVault(new ReversibleTestCipher(), new CredentialTypeCatalog()); + const string secret = "ghp_leakedTokenValue123456"; + await vault.CreateAsync( + db, tenantId, "my-api", "httpHeaderAuth", + new System.Text.Json.Nodes.JsonObject { ["name"] = "X-API-Key", ["value"] = secret }, default); + + var yaml = """ + name: cred-leak + tasks: + - id: fetch + node: { type: core.httpRequest } + parameters: { url: "https://api.example.com" } + credentials: { httpAuth: my-api } + """; + var workflow = WorkflowDataHelpers.CompileAndSave(db, tenantId, yaml, "workflows/cred-leak.yaml"); + var run = WorkflowDataHelpers.AddPendingRun(db, workflow); + + var (executor, _) = NewExecutor(db, new[] { new LeakyFailExecutor() }, vault); + await executor.ExecuteAsync(tenantId, GraphJob(run), default); + + var reloaded = await db.WorkflowRuns.SingleAsync(r => r.Id == run.Id); + Assert.Equal(RunStatus.Failed, reloaded.Status); + Assert.DoesNotContain(secret, reloaded.Error); + + var taskRun = await db.TaskRuns.SingleAsync(t => t.RunId == run.Id); + Assert.Equal(TaskRunStatus.Failed, taskRun.Status); + Assert.DoesNotContain(secret, taskRun.Error); + } +} diff --git a/w4c-workflows-api.Tests/NodeGraphCompilerTests.cs b/w4c-workflows-api.Tests/NodeGraphCompilerTests.cs index fafc8bb..92f5f93 100644 --- a/w4c-workflows-api.Tests/NodeGraphCompilerTests.cs +++ b/w4c-workflows-api.Tests/NodeGraphCompilerTests.cs @@ -186,6 +186,37 @@ public class NodeGraphCompilerTests Assert.Equal("loop", result.Graph!.EntryNodeId); } + /// + /// P1-2: loop-back classification must not depend on task declaration order. + /// Declaring the body before the loop node previously made the DFS classify + /// the forward edge (loop → body) as the back edge and reject a valid loop. + /// + [Fact] + public void Loop_back_classification_is_order_independent() + { + var yaml = """ + name: loop-order + tasks: + - id: body + node: { type: core.noop } + edges: + - { output: 0, to: loop } + - id: loop + node: { type: core.splitInBatches } + parameters: { batchSize: 2 } + edges: + - { output: 0, to: body } + """; + + var result = new NodeGraphCompiler(NodeTestData.CoreCatalog()).Compile(NodeTestData.Parse(yaml)); + + Assert.True(result.Success, string.Join("\n", result.Errors)); + var backEdge = Assert.Single(result.Graph!.Edges.Where(e => e.IsLoopBack)); + Assert.Equal("body", backEdge.FromNodeId); + Assert.Equal("loop", backEdge.ToNodeId); + Assert.Equal("loop", result.Graph!.EntryNodeId); + } + [Fact] public void Multiple_entry_nodes_are_rejected() { diff --git a/w4c-workflows-api.Tests/NodeGraphRunnerTests.cs b/w4c-workflows-api.Tests/NodeGraphRunnerTests.cs index 29db712..c2ec2c2 100644 --- a/w4c-workflows-api.Tests/NodeGraphRunnerTests.cs +++ b/w4c-workflows-api.Tests/NodeGraphRunnerTests.cs @@ -365,4 +365,38 @@ public class NodeGraphRunnerTests Assert.Equal(3, result.ExecutionOrder.Count(id => id == "body")); Assert.Equal(5, result.OutputOf("done").Count); // original items reach the exit } + + /// + /// P1-3: an entry loop node (zero non-loop-back incoming) must be re-triggered + /// by its own loop-back edge. Previously readiness compared `arrived == + /// incoming` and an entry loop node (incoming 0) silently ran a single batch. + /// + [Fact] + public async Task An_entry_loop_node_iterates_each_batch() + { + var yaml = """ + name: entry-loop + tasks: + - id: loop + node: { type: core.splitInBatches } + parameters: { batchSize: 2 } + edges: + - { output: 0, to: body } + - { output: 1, to: done } + - id: body + node: { type: core.noop } + edges: + - { output: 0, to: loop } + - id: done + node: { type: core.noop } + """; + + var seed = Enumerable.Range(1, 5).Select(i => Item($$"""{"n":{{i}}}""")).ToArray(); + var result = await Runner(new SplitInBatchesNodeExecutor()).RunAsync(Compile(yaml), seed); + + Assert.True(result.Succeeded, result.Failure?.Message); + Assert.Equal(4, result.ExecutionOrder.Count(id => id == "loop")); + Assert.Equal(3, result.ExecutionOrder.Count(id => id == "body")); + Assert.Equal(5, result.OutputOf("done").Count); + } } diff --git a/w4c-workflows-api.Tests/RunLifecycleEngineTests.cs b/w4c-workflows-api.Tests/RunLifecycleEngineTests.cs index 34bd9e3..02f9130 100644 --- a/w4c-workflows-api.Tests/RunLifecycleEngineTests.cs +++ b/w4c-workflows-api.Tests/RunLifecycleEngineTests.cs @@ -41,38 +41,6 @@ public class RunLifecycleEngineTests private static TaskDispatcher NewDispatcher(WorkflowsDbContext db, FakeJobQueue jobs) => new(db, jobs, new ConfigurationBuilder().Build(), NullLogger.Instance); - /// An engine wired with the node kernel so node-mode workflows run in-process. - private static RunLifecycleEngine NewNodeEngine( - FakeEventBus events, - FakeLeaseService leases, - IEnumerable? extraExecutors = null, - CredentialVault? vault = null) - { - var catalog = new NodeBlueprintCatalog(NodeBlueprintCatalog.LoadEmbedded()); - var executors = new List - { - new NoOpNodeExecutor(), - new SetNodeExecutor(), - new IfNodeExecutor(), - new SplitInBatchesNodeExecutor(), - new ExecuteWorkflowNodeExecutor(), - new CodeNodeExecutor(new RuntimeRegistry(new LanguageRegistry(), Array.Empty())), - }; - if (extraExecutors != null) - executors.AddRange(extraExecutors); - - var graphRunner = new NodeGraphRunner(new NodeExecutorRegistry(executors), new NodeParameterInterpolator()); - var nodeRunner = new NodeWorkflowRunner( - catalog, graphRunner, vault ?? new CredentialVault(new ReversibleTestCipher(), new CredentialTypeCatalog())); - - return new RunLifecycleEngine( - events, - leases, - new ConfigurationBuilder().Build(), - NullLogger.Instance, - nodeRunner); - } - private static IReadOnlyDictionary ResultFields( WorkflowRun run, WorkflowTask task, int attempt, bool success, string? output = null, string? error = null) { @@ -483,8 +451,15 @@ public class RunLifecycleEngineTests // ------------------------------------------------------------------ node mode + /// + /// S2: dispatch of a node workflow enqueues exactly one `graph.run` job and + /// returns immediately — the graph is NOT executed inline on the lifecycle + /// loop, so a waiting node cannot stall dispatch, result consumption or + /// timeout checks for other tenants. Execution is covered by + /// . + /// [Fact] - public async Task Node_workflow_runs_in_process_and_records_a_task_run_per_node() + public async Task Node_workflow_dispatch_enqueues_one_graph_job_and_releases_the_lease() { var tenantId = Tenant(); await using var db = _fixture.CreateContext(); @@ -492,134 +467,51 @@ public class RunLifecycleEngineTests db, tenantId, WorkflowDataHelpers.NodeYaml, "workflows/node.yaml"); var run = WorkflowDataHelpers.AddPendingRun(db, workflow); - Assert.Single(workflow.TaskEdges); // seed -> shape edge persisted - var jobs = new FakeJobQueue(); - var engine = NewNodeEngine(new FakeEventBus(), new FakeLeaseService()); + var leases = new FakeLeaseService(); + var engine = NewEngine(new FakeEventBus(), leases); var dispatched = await engine.DispatchPendingAsync(db, NewDispatcher(db, jobs), default); Assert.True(dispatched >= 1); - // Node mode is in-process: nothing is enqueued on the subprocess jobs stream. - Assert.Empty(EnqueuedFor(jobs, tenantId)); + // One job for the whole graph, carrying the run reference. + var fields = Assert.Single(EnqueuedFor(jobs, tenantId)); + Assert.Equal("graph.run", fields["type"]); + Assert.Equal(run.Id.ToString(), fields["run_id"]); + Assert.Equal(tenantId, fields["tenant_id"]); + + // Running before the job is queued, and no per-node history yet: the + // dispatch loop did not execute the graph. var reloaded = await db.WorkflowRuns.SingleAsync(r => r.Id == run.Id); - Assert.Equal(RunStatus.Succeeded, reloaded.Status); + Assert.Equal(RunStatus.Running, reloaded.Status); + Assert.NotNull(reloaded.StartedAt); + Assert.False(await db.TaskRuns.AnyAsync(t => t.RunId == run.Id)); - var taskRuns = await db.TaskRuns.Where(t => t.RunId == run.Id).ToListAsync(); - Assert.Equal(2, taskRuns.Count); - Assert.All(taskRuns, t => Assert.Equal(TaskRunStatus.Succeeded, t.Status)); - - // The terminal node's output is the run output. - Assert.NotNull(reloaded.OutputJson); - Assert.Contains("copied", reloaded.OutputJson); + // The dispatch lease is released as soon as the job is enqueued. + Assert.True(await leases.AcquireAsync(tenantId, run.Id.ToString(), "probe", TimeSpan.FromMinutes(1))); } + /// + /// A failing enqueue must fail the run and still release the dispatch lease + /// (it is released in a finally, never pinned to a run that never started). + /// [Fact] - public async Task Node_workflow_without_an_executor_fails_the_run() + public async Task Node_workflow_dispatch_releases_the_lease_when_enqueue_fails() { var tenantId = Tenant(); await using var db = _fixture.CreateContext(); var workflow = WorkflowDataHelpers.CompileAndSave( - db, tenantId, WorkflowDataHelpers.NodeNoExecutorYaml, "workflows/node-fail.yaml"); + db, tenantId, WorkflowDataHelpers.NodeYaml, "workflows/node-enqueue-fail.yaml"); var run = WorkflowDataHelpers.AddPendingRun(db, workflow); - var jobs = new FakeJobQueue(); - var engine = NewNodeEngine(new FakeEventBus(), new FakeLeaseService()); + var leases = new FakeLeaseService(); + var engine = NewEngine(new FakeEventBus(), leases); + var failingJobs = new FakeJobQueue { FailEnqueue = true }; - await engine.DispatchPendingAsync(db, NewDispatcher(db, jobs), default); + await engine.DispatchPendingAsync(db, NewDispatcher(db, failingJobs), default); - var reloaded = await db.WorkflowRuns.SingleAsync(r => r.Id == run.Id); - Assert.Equal(RunStatus.Failed, reloaded.Status); - Assert.Contains("no executor", reloaded.Error); - } - - [Fact] - public async Task Node_workflow_loop_runs_and_records_each_iteration() - { - var tenantId = Tenant(); - await using var db = _fixture.CreateContext(); - const string yaml = """ - name: loop-run - tasks: - - id: source - node: { type: core.noop } - edges: - - { output: 0, to: loop } - - id: loop - node: { type: core.splitInBatches } - parameters: { batchSize: 2 } - edges: - - { output: 0, to: body } - - { output: 1, to: done } - - id: body - node: { type: core.noop } - edges: - - { output: 0, to: loop } - - id: done - node: { type: core.noop } - """; - var workflow = WorkflowDataHelpers.CompileAndSave(db, tenantId, yaml, "workflows/loop.yaml"); - Assert.Contains(workflow.TaskEdges, e => e.IsLoopBack); - - var run = WorkflowDataHelpers.AddPendingRun( - db, workflow, """[{"n":1},{"n":2},{"n":3},{"n":4},{"n":5}]"""); - - var engine = NewNodeEngine(new FakeEventBus(), new FakeLeaseService()); - await engine.DispatchPendingAsync(db, NewDispatcher(db, new FakeJobQueue()), default); - - var reloaded = await db.WorkflowRuns.SingleAsync(r => r.Id == run.Id); - Assert.Equal(RunStatus.Succeeded, reloaded.Status); - - var loopTask = workflow.Tasks.Single(t => t.Key == "loop"); - var iterations = await db.TaskRuns.CountAsync(t => t.RunId == run.Id && t.TaskId == loopTask.Id); - Assert.Equal(4, iterations); // 3 batches + the final done invocation - } - - /// Stands in for the HTTP executor and records the credentials it was handed. - private sealed class CapturingHttpExecutor : INodeExecutor - { - public string Type => "core.httpRequest"; - public IReadOnlyDictionary? Seen { get; private set; } - - public Task RunAsync(NodeExecutionContext context, CancellationToken ct) - { - Seen = context.Credentials; - return Task.FromResult(NodeExecutionOutcome.Single(context.Input(0).ToList())); - } - } - - [Fact] - public async Task Node_workflow_resolves_declared_credentials_from_the_vault() - { - var tenantId = Tenant(); - await using var db = _fixture.CreateContext(); - var vault = new CredentialVault(new ReversibleTestCipher(), new CredentialTypeCatalog()); - await vault.CreateAsync( - db, tenantId, "my-api", "httpHeaderAuth", - new JsonObject { ["name"] = "X-API-Key", ["value"] = "abc" }, default); - - var yaml = """ - name: cred-run - tasks: - - id: fetch - node: { type: core.httpRequest } - parameters: { url: "https://api.example.com" } - credentials: { httpAuth: my-api } - """; - var workflow = WorkflowDataHelpers.CompileAndSave(db, tenantId, yaml, "workflows/cred.yaml"); - var run = WorkflowDataHelpers.AddPendingRun(db, workflow); - - var capturing = new CapturingHttpExecutor(); - var engine = NewNodeEngine(new FakeEventBus(), new FakeLeaseService(), new[] { capturing }, vault); - - await engine.DispatchPendingAsync(db, NewDispatcher(db, new FakeJobQueue()), default); - - Assert.NotNull(capturing.Seen); - Assert.Equal("httpHeaderAuth", capturing.Seen!["httpAuth"].Type); - Assert.Equal("abc", capturing.Seen!["httpAuth"].Data["value"]!.GetValue()); - - var reloaded = await db.WorkflowRuns.SingleAsync(r => r.Id == run.Id); - Assert.Equal(RunStatus.Succeeded, reloaded.Status); + Assert.Equal(RunStatus.Failed, (await db.WorkflowRuns.SingleAsync(r => r.Id == run.Id)).Status); + Assert.True(await leases.AcquireAsync(tenantId, run.Id.ToString(), "probe", TimeSpan.FromMinutes(1))); } } diff --git a/w4c-workflows-api.Tests/SecretRedactorTests.cs b/w4c-workflows-api.Tests/SecretRedactorTests.cs index 9c835b0..ab65a20 100644 --- a/w4c-workflows-api.Tests/SecretRedactorTests.cs +++ b/w4c-workflows-api.Tests/SecretRedactorTests.cs @@ -40,4 +40,25 @@ public class SecretRedactorTests Assert.Null(SecretRedactor.Redact(null)); Assert.Equal(string.Empty, SecretRedactor.Redact(string.Empty)); } + + // P1-10: a connector secret substituted into a URL path/query is not + // key=value shaped, so the regex rules cannot find it. Passing the resolved + // secret values must remove the literal wherever it appears. + [Fact] + public void Redacts_known_secret_literals_anywhere_in_the_text() + { + var secret = "ghp_superSecretToken12345"; + var text = $"request failed: GET https://api.example.com/{secret}/items?x=1"; + + var redacted = SecretRedactor.Redact(text, new[] { secret }); + + Assert.DoesNotContain(secret, redacted); + Assert.Contains(SecretRedactor.Placeholder, redacted); + } + + [Fact] + public void Ignores_short_known_values_to_avoid_over_redaction() + { + Assert.Equal("id 42 ok", SecretRedactor.Redact("id 42 ok", new[] { "42" })); + } } diff --git a/w4c-workflows-api.Tests/WebhooksControllerTests.cs b/w4c-workflows-api.Tests/WebhooksControllerTests.cs new file mode 100644 index 0000000..9be668e --- /dev/null +++ b/w4c-workflows-api.Tests/WebhooksControllerTests.cs @@ -0,0 +1,50 @@ +using w4c_workflows.Controllers; +using w4c_workflows.Models; +using Xunit; + +namespace w4c_workflows.Tests; + +/// +/// P0-7: webhook matching must use the denormalized indexed column when present +/// and fall back to trigger JSON only for rows compiled before the column existed. +/// +public class WebhooksControllerTests +{ + private static Workflow Wf(string? column, string? triggerJson) => new() + { + TenantId = "t1", + Name = "n", + Path = "p", + Status = WorkflowStatus.Compiled, + Mode = WorkflowMode.Function, + Target = "default", + WebhookPath = column, + TriggerJson = triggerJson, + }; + + [Fact] + public void Matches_on_the_denormalized_column() + { + var workflow = Wf("/h/orders", "{\"type\":\"webhook\",\"webhookPath\":\"/h/orders\"}"); + + Assert.True(WebhooksController.IsWebhookMatch(workflow, "/h/orders")); + Assert.False(WebhooksController.IsWebhookMatch(workflow, "/h/other")); + } + + [Fact] + public void Falls_back_to_trigger_json_for_legacy_rows() + { + var workflow = Wf(column: null, triggerJson: "{\"type\":\"webhook\",\"webhookPath\":\"/h/orders\"}"); + + Assert.True(WebhooksController.IsWebhookMatch(workflow, "/h/orders")); + Assert.False(WebhooksController.IsWebhookMatch(workflow, "/h/other")); + } + + [Fact] + public void Non_webhook_trigger_never_matches() + { + var workflow = Wf(column: null, triggerJson: "{\"type\":\"cron\",\"cron\":\"0 * * * *\"}"); + + Assert.False(WebhooksController.IsWebhookMatch(workflow, "/h/orders")); + } +} diff --git a/w4c-workflows-api.Tests/WorkflowCompilerTests.cs b/w4c-workflows-api.Tests/WorkflowCompilerTests.cs index 1a1256b..bf0651e 100644 --- a/w4c-workflows-api.Tests/WorkflowCompilerTests.cs +++ b/w4c-workflows-api.Tests/WorkflowCompilerTests.cs @@ -1,5 +1,6 @@ using w4c_workflows.Models; using w4c_workflows.Services; +using w4c_workflows.Services.Nodes; using Xunit; namespace w4c_workflows.Tests; @@ -290,4 +291,132 @@ public class WorkflowCompilerTests Assert.Equal(TimeSpan.FromMinutes(90), compound); Assert.False(DurationParser.TryParse("soon", out _)); } + + // ------------------------------------------------------- S1 convergence + + private const string FunctionScriptYaml = """ + name: legacy-fn + mode: function + language: shell + entry: { file: root.sh } + tasks: + - id: a + next: b + language: shell + entry: { file: a.sh } + - id: b + entry: { file: b.sh } + """; + + private static WorkflowCompiler NewLoweringCompiler() + => new(new WorkflowValidator(new LanguageRegistry()), lowerLegacyScripts: true); + + [Fact] + public void Lowers_function_script_to_node_mode_when_enabled() + { + var result = NewLoweringCompiler().Compile(FunctionScriptYaml, "workflows/legacy-fn.yaml", Tenant); + + Assert.True(result.Success, string.Join("\n", result.Errors)); + var tasks = result.Workflow!.Tasks; + + // Every step becomes a node step; the synthetic entry plus both tasks exist. + Assert.All(tasks, t => Assert.Equal("node", t.Language)); + Assert.All(tasks, t => Assert.False(string.IsNullOrWhiteSpace(t.NodeType))); + Assert.Contains(tasks, t => t.Key == LegacyWorkflowLowerer.EntryStepId); + Assert.Contains(tasks, t => t.Key == "a"); + Assert.Contains(tasks, t => t.Key == "b"); + + // The `next` chain is lowered into port edges instead of NextId. + Assert.All(tasks, t => Assert.Null(t.NextId)); + Assert.NotEmpty(result.Workflow.Edges); + } + + [Fact] + public void Keeps_durable_scripts_on_the_legacy_engine_when_lowering_enabled() + { + var yaml = """ + name: durable-legacy + mode: durable + language: shell + entry: { file: root.sh } + tasks: + - id: a + entry: { file: a.sh } + """; + + var result = NewLoweringCompiler().Compile(yaml, "workflows/durable-legacy.yaml", Tenant); + + Assert.True(result.Success, string.Join("\n", result.Errors)); + // Durable/handler keep the linear engine: the node kernel does not + // checkpoint/resume yet, so lowering them would lose semantics. + Assert.All(result.Workflow!.Tasks, t => Assert.Null(t.NodeType)); + Assert.Empty(result.Workflow.Edges); + } + + [Fact] + public void Lowering_is_off_by_default() + { + var result = NewCompiler().Compile(FunctionScriptYaml, "workflows/legacy-fn.yaml", Tenant); + + Assert.True(result.Success, string.Join("\n", result.Errors)); + Assert.All(result.Workflow!.Tasks, t => Assert.Null(t.NodeType)); + Assert.Empty(result.Workflow.Edges); + } + + [Fact] + public void Webhook_trigger_populates_the_indexed_webhook_path() + { + var yaml = """ + name: hook + mode: function + language: shell + entry: { file: root.sh } + trigger: + type: webhook + webhookPath: /h/orders + """; + + var result = NewCompiler().Compile(yaml, "workflows/hook.yaml", Tenant); + + Assert.True(result.Success, string.Join("\n", result.Errors)); + Assert.Equal("/h/orders", result.Workflow!.Workflow.WebhookPath); + } + + [Fact] + public void Node_workflow_webhook_trigger_also_populates_the_column() + { + var yaml = """ + name: hook-node + trigger: + type: webhook + webhookPath: /h/node-orders + tasks: + - id: only + node: { type: core.noop } + """; + + var result = NewCompiler().Compile(yaml, "workflows/hook-node.yaml", Tenant); + + Assert.True(result.Success, string.Join("\n", result.Errors)); + Assert.Equal("/h/node-orders", result.Workflow!.Workflow.WebhookPath); + } + + [Fact] + public void Non_webhook_trigger_leaves_the_webhook_path_null() + { + var yaml = """ + name: cron-only + mode: function + language: shell + entry: { file: root.sh } + trigger: + type: cron + cron: "0 * * * *" + """; + + var result = NewCompiler().Compile(yaml, "workflows/cron-only.yaml", Tenant); + + Assert.True(result.Success, string.Join("\n", result.Errors)); + Assert.Null(result.Workflow!.Workflow.WebhookPath); + } }