SemianiakaVY/w4c-workflows-api
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

enchancements

Browse source
This commit is contained in:
Vitali sharp8n 2026-09-13 11:35:17 +03:00
parent b977619f94
commit c33e34796f
51 changed files with 2331 additions and 506 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

160
Controllers/WebhooksController.cs
View file

@ -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;
/// <summary>
/// Webhook receiver for workflows with <c>trigger.type = webhook</c>. The route
/// is PUBLIC (external callers have no operator key); access is governed by an
/// optional shared secret (<c>Workflows:WebhookSecret</c>) verified against the
/// <c>X-Webhook-Secret</c> 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 (<c>X-Webhook-Secret</c>) and is fail-closed:
/// - If <c>Workflows:WebhookSecrets</c> maps tenants to secrets, the request is
/// attributed to exactly one tenant and only that tenant's workflows match.
/// - Otherwise the global <c>Workflows:WebhookSecret</c> 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 <see cref="Workflow.WebhookPath"/> column (falling
/// back to the trigger JSON for rows compiled before the column existed), and the
/// request body is capped by <c>Workflows:WebhookMaxBodyBytes</c>.
/// </summary>
[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<WebhooksController> _logger;
private readonly string? _sharedSecret;
private readonly Dictionary<string, string> _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<string, string>(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;
}
/// <summary>Receives a webhook at <c>/h/{path}</c> and fires every matching workflow.</summary>
@ -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<Guid>(matches.Count);
@ -86,61 +133,114 @@ public class WebhooksController : ControllerBase
return Accepted(new { runs = runIds });
}
private bool Authorized()
/// <summary>
/// Resolves the caller's tenant. Returns false (fail-closed) when no secret is
/// configured or the provided secret matches nothing. A null
/// <paramref name="tenantId"/> with a true result means the global secret was
/// used and the tenant is unknown.
/// </summary>
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<List<Workflow>> FindMatchesAsync(string webhookPath, CancellationToken ct)
private async Task<List<Workflow>> 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<Workflow>();
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<string?> ReadInputAsync(CancellationToken ct)
/// <summary>
/// True when the workflow's webhook trigger points at <paramref name="webhookPath"/>.
/// Prefers the denormalized column; parses trigger JSON only for legacy rows.
/// </summary>
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;

49
Controllers/WorkflowQuotaController.cs
View file

@ -6,19 +6,24 @@ namespace w4c_workflows.Controllers;
/// <summary>
/// Per-tenant workflow execution quota. <c>GET</c> returns the current period's
/// usage (drives the counter on the Workflows page); <c>POST reset</c> 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). <c>POST reset</c> is a
/// platform-admin action: a tenant must not be able to zero its own counter, so
/// it additionally requires the shared <c>X-Admin-Key</c> matching
/// <c>Workflows:AdminApiKey</c> (fail-closed when unset).
/// </summary>
[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));
}
/// <summary>Resets the current period's execution counter to zero.</summary>
/// <summary>
/// Resets the current period's execution counter to zero. Platform admin only:
/// requires the shared admin key in addition to the operator key.
/// </summary>
[HttpPost("reset")]
[RequireScope("manage")]
public async Task<IActionResult> 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;
}
}

5
Data/WorkflowsSchema.cs
View file

@ -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)

21
Filters/RequireScopeAttribute.cs
View file

@ -35,11 +35,26 @@ public class RequireScopeAttribute : Attribute, IAsyncActionFilter
}
var scopes = context.HttpContext.Items["Scopes"] as IReadOnlyList<string>;
// 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;
}

12
Middleware/AuthMiddleware.cs
View file

@ -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<AuthMiddleware> _logger;
public AuthMiddleware(RequestDelegate next, IConfiguration config, ILogger<AuthMiddleware> 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);

7
Models/Entities.cs
View file

@ -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 }
/// <summary>
/// Denormalized <c>trigger.webhookPath</c> for webhook triggers, so the public
/// receiver resolves matches with an indexed <c>(TenantId, WebhookPath)</c>
/// lookup instead of loading and parsing every tenant's trigger JSON. Null for
/// non-webhook triggers and for rows compiled before the column existed.
/// </summary>
public string? WebhookPath { get; set; }
/// <summary>
/// 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.

53
Program.cs
View file

@ -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<EgressPolicy>(),
sp.GetRequiredService<IHostAddressResolver>()));
// 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<EgressPolicy>(),
sp.GetRequiredService<IHostAddressResolver>(),
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<EgressPolicy>(),
sp.GetRequiredService<IHostAddressResolver>()));
// Credential vault encryption keys (used by ICredentialCipher).
builder.Services.AddDataProtection();
@ -149,7 +152,12 @@ else
// YAML compile pipeline (step 4).
builder.Services.AddSingleton<LanguageRegistry>();
builder.Services.AddSingleton<WorkflowValidator>();
builder.Services.AddSingleton<WorkflowCompiler>();
// 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<WorkflowValidator>(),
sp.GetRequiredService<NodeGraphCompiler>(),
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<CredentialTypeCatalog>();
builder.Services.AddSingleton<ICredentialCipher, DataProtectionCredentialCipher>();
builder.Services.AddSingleton<CredentialVault>();
// 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<NodeWorkflowRunner>();
// 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<GraphJobExecutor>();
// Realtime SSE hub — notifies subscribed clients when a workflow file changes server-side.
builder.Services.AddSingleton<RealtimeEventHub>();
@ -274,6 +287,9 @@ builder.Services.AddScoped<IWorkflowSource>(sp =>
return factory.Create(tenantId);
});
builder.Services.AddScoped<WorkflowSyncService>();
// 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<SyncGate>();
// Mermaid diagrams (step 6).
builder.Services.AddSingleton<MermaidGeneratorService>();
@ -289,7 +305,7 @@ builder.Services.AddSingleton<IScriptExecutor>(new SubprocessScriptExecutor("she
builder.Services.AddSingleton<IScriptExecutor>(new SubprocessScriptExecutor("python", "python3", builder.Configuration));
builder.Services.AddSingleton<IScriptExecutor>(new SubprocessScriptExecutor("javascript", "node", builder.Configuration));
builder.Services.AddSingleton<IScriptExecutor>(new TypeScriptExecutor(builder.Configuration));
builder.Services.AddSingleton<IScriptExecutor>(new CSharpScriptExecutor());
builder.Services.AddSingleton<IScriptExecutor>(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<IScriptExecutor>(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;
}

22
Services/Audit/SecretRedactor.cs
View file

@ -33,8 +33,13 @@ public static class SecretRedactor
RegexOptions.IgnoreCase | RegexOptions.Compiled,
Timeout);
/// <summary>Returns the text with credential-shaped substrings replaced.</summary>
public static string? Redact(string? text)
/// <summary>
/// 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.
/// </summary>
public static string? Redact(string? text, IEnumerable<string>? 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;
}
}

234
Services/Execution/CSharpScriptExecutor.cs
View file

@ -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. <c>Main.cs</c> + <c>tasks/hello-world.cs</c>) into one assembly,
/// causing duplicate-type errors when both define a <c>Program</c> class.
///
/// Each execution uses a <see cref="CollectibleAssemblyLoadContext"/> 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 <see cref="AssemblyLoadContext"/> so the loaded assembly
/// and all its types can be garbage-collected after the call returns. The
/// compiled <em>image bytes</em> 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
/// <c>Workflows:CSharpTimeoutSeconds</c> (default <c>TaskTimeoutSeconds</c>) and
/// observes <c>ct</c>: 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 <c>public static</c> method named after
/// <c>entry.function</c> (default <c>Main</c>) that takes a single
@ -33,67 +43,134 @@ public class CSharpScriptExecutor : IScriptExecutor
{
private static readonly JsonSerializerOptions Json = new(JsonSerializerDefaults.Web);
/// <summary>Compiled PE images keyed by source hash, so a hot task never re-runs Roslyn.</summary>
private static readonly ConcurrentDictionary<string, byte[]> AssemblyCache = new(StringComparer.Ordinal);
private static readonly ConcurrentQueue<string> 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<MetadataReference>? _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";
/// <summary>Roslyn ships with the service image — no external runtime to probe.</summary>
public bool IsAvailable() => true;
public Task<ExecutionResult> ExecuteAsync(TaskInvocation invocation, CancellationToken ct)
public async Task<ExecutionResult> 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<Diagnostic>) Compile(SourceFile source)
private static async Task<string?> 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();
}
}
/// <summary>
/// Returns the compiled PE image for <paramref name="source"/>, compiling on a
/// cache miss. Throws <see cref="CSharpCompileException"/> with the Roslyn
/// diagnostics when compilation fails (failures are never cached).
/// </summary>
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();
}
/// <summary>
/// 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.
/// </summary>
private static ImmutableArray<MetadataReference> GetCachedReferences()
{
@ -172,7 +244,38 @@ public class CSharpScriptExecutor : IScriptExecutor
}
}
private static string? InvokeEntry(Assembly assembly, TaskInvocation invocation)
private static async Task<string?> 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<T> 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<T> 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;
/// <summary>Raised on Roslyn emit failure; the message carries the diagnostics.</summary>
private sealed class CSharpCompileException(string message) : Exception(message);
}

45
Services/Execution/ExecutionHelpers.cs
View file

@ -24,6 +24,51 @@ internal static class ExecutionHelpers
? Directory.GetCurrentDirectory()
: Path.GetFullPath(invocation.WorkingDir);
/// <summary>
/// Resolves a task's entry file inside its working directory. A rooted path,
/// or one that escapes via <c>..</c>, is rejected — otherwise a workflow could
/// read or execute files outside its own run directory (<c>Path.Combine</c>
/// silently drops the root when the second argument is absolute).
/// </summary>
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;
}
/// <summary>WF_* runtime context + the workflow/task-declared env, merged.</summary>
public static IReadOnlyDictionary<string, string> BuildEnvironment(TaskInvocation invocation)
{

136
Services/Execution/GraphJobExecutor.cs Normal file
View file

@ -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;
/// <summary>
/// Executes a <c>graph.run</c> job: loads the run and its compiled node graph,
/// drives the whole graph through <see cref="NodeWorkflowRunner"/>, 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
/// <c>--worker</c> 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 <c>pending</c>/<c>running</c>; a terminal run is skipped.
/// </summary>
public sealed class GraphJobExecutor
{
private readonly IServiceScopeFactory _scopeFactory;
private readonly IJobQueue _jobs;
private readonly NodeWorkflowRunner _runner;
private readonly ILogger<GraphJobExecutor> _logger;
public GraphJobExecutor(
IServiceScopeFactory scopeFactory,
IJobQueue jobs,
NodeWorkflowRunner runner,
ILogger<GraphJobExecutor> logger)
{
_scopeFactory = scopeFactory;
_jobs = jobs;
_runner = runner;
_logger = logger;
}
/// <summary>
/// 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.
/// </summary>
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<WorkflowsDbContext>();
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);
}
}

5
Services/Execution/RemoteServerExecutor.cs
View file

@ -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)
{

3
Services/Execution/SubprocessScriptExecutor.cs
View file

@ -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);

5
Services/Execution/TypeScriptExecutor.cs
View file

@ -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(

14
Services/Execution/WorkerHostService.cs
View file

@ -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<WorkerHostService> _logger;
@ -40,6 +42,7 @@ public class WorkerHostService : BackgroundService
IEventBus events,
RuntimeRegistry runtimes,
RemoteServerExecutor remote,
GraphJobExecutor graphJobs,
IServiceScopeFactory scopeFactory,
IConfiguration config,
ILogger<WorkerHostService> 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
{

89
Services/ForgejoWorkflowRepoService.cs
View file

@ -77,7 +77,7 @@ public sealed class ForgejoWorkflowRepoService
/// token when no admin token is configured.
/// </summary>
private string GitAuthToken =>
!string.IsNullOrWhiteSpace(_forgejoAdminToken) ? _forgejoAdminToken : _forgejoToken;
!string.IsNullOrWhiteSpace(_forgejoAdminToken) ? _forgejoAdminToken : _forgejoToken ?? string.Empty;
/// <summary>True when Forgejo admin provisioning is configured.</summary>
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;
}
/// <summary>
@ -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);
}
/// <summary>
@ -369,12 +384,28 @@ public sealed class ForgejoWorkflowRepoService
=> Path.GetFullPath(Path.Combine(_copiesRoot, Sanitize(tenantId), OwnerRepoSlug(login, repoName)));
/// <summary>
/// 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 <see cref="RunGitExitAsync"/> and check the
/// exit code instead of scanning the output text.
/// </summary>
private async Task<string?> 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;
}
/// <summary>
/// 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.
/// </summary>
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)
{

33
Services/JwtValidator.cs
View file

@ -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
{

16
Services/Messaging/InMemoryTransport.cs
View file

@ -86,8 +86,20 @@ public sealed class InMemoryTransport : IJobQueue, IEventBus
// ---- Helpers ----
private Channel<StreamMessage> GetOrAddChannel(string key)
=> _channels.GetOrAdd(key, _ => Channel.CreateUnbounded<StreamMessage>(
new UnboundedChannelOptions { SingleReader = false, SingleWriter = false }));
=> _channels.GetOrAdd(key, _ => Channel.CreateBounded<StreamMessage>(
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,
}));
/// <summary>Per-stream in-memory backlog ceiling.</summary>
private const int ChannelCapacity = 10_000;
private async Task<IReadOnlyList<StreamMessage>> DrainAsync(string stream, int count, CancellationToken ct)
{

10
Services/Messaging/RedisStreamsTransport.cs
View file

@ -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<string> AddAsync(string key, IReadOnlyDictionary<string, string> 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();
}

9
Services/Nodes/Connectors/RestConnectorExecutor.cs
View file

@ -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

63
Services/Nodes/Executors/HttpBodyReader.cs Normal file
View file

@ -0,0 +1,63 @@
using System.Text;
namespace w4c_workflows.Services.Nodes.Executors;
/// <summary>Raw response body plus a "too large" signal, before content interpretation.</summary>
internal sealed record BodyRead(string? Body, byte[]? Bytes, bool TooLarge);
/// <summary>
/// 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 <c>responseFormat: file</c> needs the
/// exact bytes rather than a lossy string.
/// </summary>
internal static class HttpBodyReader
{
public static async Task<BodyRead> 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);
}
/// <summary>Content-Type charset when the server declares one, else UTF-8.</summary>
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;
}
}

17
Services/Nodes/Executors/HttpPaginationPlan.cs
View file

@ -52,10 +52,17 @@ internal sealed class HttpPaginationPlan
public IReadOnlySet<int> StopStatusCodes { get; init; } = new HashSet<int>();
/// <summary>Hard cap on the number of requests; 0 disables the cap.</summary>
/// <summary>
/// Hard cap on the number of requests. Non-positive values fall back to
/// <see cref="DefaultMaxPages"/> — the cap can never be disabled, otherwise a
/// self-referential <c>nextUrl</c> would loop until quota/timeout.
/// </summary>
public int MaxPages { get; init; } = DefaultMaxPages;
/// <summary>Hard cap on the accumulated items; 0 disables the cap.</summary>
/// <summary>
/// Hard cap on the accumulated items. Non-positive values fall back to
/// <see cref="DefaultMaxItems"/> — the cap can never be disabled.
/// </summary>
public int MaxItems { get; init; } = DefaultMaxItems;
/// <summary>
@ -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())

58
Services/Nodes/Executors/HttpRequestNodeExecutor.cs
View file

@ -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);
/// <summary>
/// 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 <c>responseFormat: file</c>
/// needs the exact bytes rather than a lossy string.
/// </summary>
private static async Task<BodyRead> 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);
}
/// <summary>Content-Type charset when the server declares one, else UTF-8.</summary>
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<FlowItem> Items, string? Error);
/// <summary>

75
Services/Nodes/GraphRunMessage.cs Normal file
View file

@ -0,0 +1,75 @@
using System.Globalization;
using w4c_workflows.Models;
namespace w4c_workflows.Services.Nodes;
/// <summary>
/// Wire contract for a <c>graph.run</c> job. This is the node-kernel analogue of
/// <see cref="Execution.TaskInvocation"/>: instead of one script task, the job
/// asks a worker to run the whole persisted node graph of a run through
/// <see cref="NodeWorkflowRunner"/>.
///
/// It exists so node-mode workflows execute on the same worker/queue path as
/// script tasks (S2): the control plane enqueues one <c>graph.run</c> 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.
/// </summary>
public sealed record GraphRunInvocation(
string Type,
string RunId,
string TenantId,
string? WorkingDir,
int Attempt = 1)
{
public const string TypeValue = "graph.run";
/// <summary>Parses a stream message into an invocation, throwing if a required field is missing.</summary>
public static GraphRunInvocation FromFields(IReadOnlyDictionary<string, string> fields)
{
static string Get(IReadOnlyDictionary<string, string> 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<string, string> 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));
}
}
/// <summary>
/// Serializes the control-plane side of a <c>graph.run</c> 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.
/// </summary>
public static class GraphRunMessage
{
public static IReadOnlyDictionary<string, string> ToFields(WorkflowRun run, string? workingDir, int attempt)
{
var fields = new Dictionary<string, string>
{
["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;
}
}

12
Services/Nodes/LegacyWorkflowLowerer.cs
View file

@ -13,6 +13,18 @@ namespace w4c_workflows.Services.Nodes;
/// Script execution is unchanged: the <c>core.code</c> executor delegates to the
/// existing <see cref="Execution.IScriptExecutor"/> runtime, so the subprocess
/// path is preserved while the edge kernel owns scheduling.
///
/// This is the S1 migration bridge. It is wired into <see cref="WorkflowCompiler"/>
/// behind <c>Workflows:LowerLegacyScripts</c> (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 <c>FlowItem</c> envelopes
/// (<c>{ "json": … }</c>), 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.
/// </summary>
public static class LegacyWorkflowLowerer
{

121
Services/Nodes/NodeGraphLinks.cs
View file

@ -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
/// <c>u → v</c> is a loop-back exactly when it lies on a cycle (the target
/// <c>v</c> can reach <c>u</c>) and <c>v</c> is loop-capable. This is
/// order-independent, so the compiler and a rebuilt persisted graph always agree
/// regardless of task declaration order.
/// </summary>
public static class NodeGraphLinks
{
/// <summary>
/// Returns the edges with <see cref="NodeGraphEdge.IsLoopBack"/> 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.
/// </summary>
public static (List<NodeGraphEdge> Edges, string? Error) MarkLoopBackEdges(
IReadOnlyList<NodeGraphNode> nodes, IReadOnlyList<NodeGraphEdge> edges)
{
var byId = nodes.ToDictionary(n => n.Id, StringComparer.Ordinal);
var adjacency = new Dictionary<string, List<int>>(StringComparer.Ordinal);
for (var i = 0; i < edges.Count; i++)
var adjacency = new Dictionary<string, List<string>>(StringComparer.Ordinal);
foreach (var edge in edges)
{
if (!adjacency.TryGetValue(edges[i].FromNodeId, out var outgoing))
adjacency[edges[i].FromNodeId] = outgoing = new List<int>();
outgoing.Add(i);
}
var color = new Dictionary<string, int>(StringComparer.Ordinal); // 0 = white, 1 = on stack, 2 = done
var backEdges = new HashSet<int>();
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<string>();
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<string, HashSet<string>>(StringComparer.Ordinal);
foreach (var node in nodes)
{
if (color.GetValueOrDefault(node.Id, 0) == 0)
Visit(node.Id);
var seen = new HashSet<string>(StringComparer.Ordinal);
var stack = new Stack<string>();
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<NodeGraphEdge>(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<NodeGraphNode> nodes,
IReadOnlyDictionary<string, List<string>> adjacency,
IReadOnlyDictionary<string, HashSet<string>> reach,
IReadOnlyDictionary<string, NodeGraphNode> byId)
{
var ids = nodes.Select(n => n.Id).ToList();
var assigned = new HashSet<string>(StringComparer.Ordinal);
foreach (var id in ids)
{
if (!assigned.Add(id))
continue;
var members = new List<string> { 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;
}
}

13
Services/Nodes/NodeGraphRunner.cs
View file

@ -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<string>(StringComparer.Ordinal);
var queued = new HashSet<string>(StringComparer.Ordinal);
inputs[graph.EntryNodeId][0].AddRange(seed);
var ready = new Queue<string>();
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<string, int> arrived,
IReadOnlyDictionary<string, int> incoming,
Queue<string> ready,
ISet<string> queued,
ISet<string> 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);
}
}

88
Services/Nodes/NodeWorkflowRunner.cs
View file

@ -13,12 +13,17 @@ namespace w4c_workflows.Services.Nodes;
public sealed record NodeWorkflowRunOutcome(bool Succeeded, string? Error, string? Output);
/// <summary>
/// Runs a persisted node-mode workflow through <see cref="NodeGraphRunner"/> in
/// the control plane, replacing the linear <c>NextId</c> subprocess chain for
/// node definitions (script workflows keep the subprocess path). Reconstructs
/// the executable graph from the stored tasks + <see cref="WorkflowTaskEdge"/>
/// rows, writes a <see cref="TaskRun"/> row per executed node for history, and
/// records the run's terminal output/status.
/// Runs a persisted node-mode workflow through <see cref="NodeGraphRunner"/>,
/// replacing the linear <c>NextId</c> subprocess chain for node definitions
/// (script workflows keep the legacy subprocess path in
/// <see cref="Runs.RunLifecycleEngine"/>). Reconstructs the executable graph from
/// the stored tasks + <see cref="WorkflowTaskEdge"/> rows, writes a
/// <see cref="TaskRun"/> row per executed node for history, and records the run's
/// terminal output/status.
///
/// Invoked by the worker (<see cref="Execution.GraphJobExecutor"/>) when it
/// claims a <c>graph.run</c> job, and recursively in-process by
/// <see cref="SubWorkflowInvoker"/> for <c>core.executeWorkflow</c> nodes.
/// </summary>
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<NodeWorkflowRunOutcome> FailAsync(
WorkflowsDbContext db, WorkflowRun run, string error, CancellationToken ct)
WorkflowsDbContext db,
WorkflowRun run,
string error,
CancellationToken ct,
IReadOnlyCollection<string>? 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);
}
/// <summary>
/// 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.
/// </summary>
private static List<string> CollectSecretValues(IReadOnlyDictionary<string, CredentialData> credentials)
{
var values = new List<string>();
foreach (var credential in credentials.Values)
{
foreach (var (_, node) in credential.Data)
CollectJsonStrings(node, values);
}
return values;
}
private static void CollectJsonStrings(JsonNode? node, List<string> values)
{
switch (node)
{
case JsonValue value when value.TryGetValue<string>(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
/// <summary>
@ -263,11 +314,12 @@ public sealed class NodeWorkflowRunner
WorkflowsDbContext db,
WorkflowRun run,
IReadOnlyDictionary<string, Guid> taskIds,
bool recordTaskRuns)
bool recordTaskRuns,
IReadOnlyCollection<string>? secretValues = null)
{
var listeners = new List<INodeRunListener>(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<string, Guid> _taskIds;
private readonly IReadOnlyCollection<string>? _secretValues;
private readonly Dictionary<string, TaskRun> _rows = new(StringComparer.Ordinal);
public TaskRunHistory(WorkflowsDbContext db, WorkflowRun run, IReadOnlyDictionary<string, Guid> taskIds)
public TaskRunHistory(
WorkflowsDbContext db,
WorkflowRun run,
IReadOnlyDictionary<string, Guid> taskIds,
IReadOnlyCollection<string>? 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);

87
Services/PerTenantWorkflowSource.cs
View file

@ -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<string, CachedSource> _cache = new(StringComparer.Ordinal);
private readonly ConcurrentDictionary<string, SemaphoreSlim> _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 <see cref="Create"/>.
/// </summary>
public async Task<IWorkflowSource> CreateAsync(string tenantId, string? repoName = null, CancellationToken ct = default)
=> (await ResolveAsync(tenantId, repoName, ct)).Source;
/// <summary>
/// Resolves the per-tenant source and its Forgejo login, caching the result
/// for <c>WorkflowSource:PullIntervalSeconds</c> (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.
/// </summary>
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<WorkflowSourceFactory>()
.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<WorkflowSourceFactory>()
.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<PerTenantWorkflowSource>();
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);
/// <summary>Drops the cached source for a tenant+repo, forcing a fresh clone/pull next time.</summary>
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<PerTenantWorkflowSource>();
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;
}
/// <summary>

16
Services/RealtimeEventHub.cs
View file

@ -5,19 +5,29 @@ namespace w4c_workflows.Services;
/// <summary>
/// Server-push hub for tenant-scoped realtime events (SSE), mirroring w4c-webapi's hub. A connected
/// client gets its own unbounded channel; <see cref="Publish"/> 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; <see cref="Publish"/> 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.
/// </summary>
public sealed class RealtimeEventHub
{
/// <summary>Per-client notification backlog ceiling.</summary>
private const int ClientChannelCapacity = 256;
private readonly ConcurrentDictionary<string, ConcurrentDictionary<Guid, Channel<RealtimeEvent>>> _clients = new();
public (Guid Id, ChannelReader<RealtimeEvent> Reader) AddClient(string tenantId)
{
var channel = Channel.CreateUnbounded<RealtimeEvent>();
var channel = Channel.CreateBounded<RealtimeEvent>(new BoundedChannelOptions(ClientChannelCapacity)
{
SingleReader = true,
SingleWriter = false,
FullMode = BoundedChannelFullMode.DropOldest,
});
var id = Guid.NewGuid();
var clients = _clients.GetOrAdd(tenantId, _ => new ConcurrentDictionary<Guid, Channel<RealtimeEvent>>());
clients[id] = channel;

60
Services/Runs/RunLifecycleEngine.cs
View file

@ -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<RunLifecycleEngine> _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<RunLifecycleEngine> logger,
NodeWorkflowRunner? nodeRunner = null)
ILogger<RunLifecycleEngine> 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
}
/// <summary>
/// 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 <see cref="TaskRun"/> per node.
/// Marks a node-mode run <c>running</c> and enqueues one <c>graph.run</c>
/// 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 <c>NodeGraphRunner</c> and owns
/// the terminal run status. The dispatch lease is released by the caller.
/// </summary>
private async Task<bool> DispatchNodeRunAsync(
private async Task<bool> 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;
}

22
Services/Runs/TaskDispatcher.cs
View file

@ -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;
}
/// <summary>
/// Enqueues a <c>graph.run</c> job for a node-mode run. Unlike
/// <see cref="DispatchAsync"/> this writes no <see cref="TaskRun"/> 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).
/// </summary>
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);
}
/// <summary>
/// Dead-letters a failed task (retries exhausted, or a failed compensation)
/// by reconstructing its <c>task.run</c> fields and publishing them to the

6
Services/Security/EgressGuard.cs
View file

@ -7,7 +7,11 @@ namespace w4c_workflows.Services.Security;
/// Authorises an outbound <see cref="Uri"/> 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
/// <see cref="EgressPinning"/>: egress clients re-resolve and re-validate the
/// address in their <c>ConnectCallback</c>, so a name that answers differently on
/// the second lookup (DNS rebinding) cannot reach a blocked address.
/// Literal IPs skip DNS entirely.
/// </summary>
public sealed class EgressGuard
{

83
Services/Security/EgressPinning.cs Normal file
View file

@ -0,0 +1,83 @@
using System.Net;
using System.Net.Sockets;
namespace w4c_workflows.Services.Security;
/// <summary>
/// Builds the <see cref="SocketsHttpHandler"/> used for workflow egress with a
/// <see cref="SocketsHttpHandler.ConnectCallback"/> that resolves the target and
/// re-validates every address against the <see cref="EgressPolicy"/> at the moment
/// of connecting.
///
/// This closes the DNS-rebinding (TOCTOU) hole: the pre-flight
/// <see cref="EgressGuard.AuthorizeAsync"/> 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 <c>Host</c>/SNI, and a host that resolves to any
/// blocked address is refused outright — matching
/// <see cref="EgressPolicy.CheckAddresses"/> semantics.
/// </summary>
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<IPAddress> 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;
}
}

9
Services/Security/EgressPolicy.cs
View file

@ -20,6 +20,15 @@ public sealed class EgressPolicy
/// <summary>Configured ceiling on redirects a node invocation may follow.</summary>
public int MaxRedirects => Math.Max(0, _options.MaxRedirects);
/// <summary>
/// True when reserved-range address checks must be skipped for
/// <paramref name="host"/>: 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.
/// </summary>
public bool SkipsAddressChecks(string host)
=> _options.AllowPrivateNetworks || MatchesAny(host, _options.AllowedHosts);
/// <summary>
/// Stage one: validate the scheme and the host allow/deny lists. Returns a
/// final decision, or <c>null</c> when the caller must resolve the host and

36
Services/SyncGate.cs Normal file
View file

@ -0,0 +1,36 @@
using System.Collections.Concurrent;
namespace w4c_workflows.Services;
/// <summary>
/// Serializes concurrent workflow syncs for the same <c>(tenant, repo)</c> within
/// one process. Two overlapping <c>POST /sync</c> calls would otherwise both
/// insert the same deterministic workflow/task ids and collide on the primary key
/// (<c>DbUpdateException</c>). 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
/// <see cref="ILeaseService"/> lease in <see cref="WorkflowSyncService"/>.
/// </summary>
public sealed class SyncGate
{
private readonly ConcurrentDictionary<string, SemaphoreSlim> _gates = new(StringComparer.Ordinal);
public async Task<IDisposable> 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();
}
}
}

71
Services/WorkflowCompiler.cs
View file

@ -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:
/// - <b>node mode</b>: every step declares a catalog <c>node</c>; steps are
/// validated by <see cref="NodeGraphCompiler"/> and compiled with their port
/// edges.
/// - <b>script mode</b> (legacy): <c>language</c> + <c>entry</c> tasks driven
/// by the linear <c>next</c>/<c>onError</c> chain.
/// Two execution models exist during the S1 convergence:
/// - <b>node mode</b> (target): every step declares a catalog <c>node</c>; steps
/// are validated by <see cref="NodeGraphCompiler"/> and compiled with their
/// port edges. This is the only model the worker kernel executes.
/// - <b>script mode</b> (legacy compatibility): <c>language</c> + <c>entry</c>
/// tasks driven by the linear <c>next</c>/<c>onError</c> chain. It is kept
/// for already-persisted workflows and for <c>durable</c>/<c>handler</c>
/// checkpoint/resume, which the node kernel does not yet cover.
///
/// Setting <c>Workflows:LowerLegacyScripts=true</c> lowers a <c>function</c>-mode
/// script definition to the node kernel at compile time via
/// <see cref="LegacyWorkflowLowerer"/> (synthetic <c>core.code</c> 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.
/// </summary>
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;
}
/// <summary>
/// True for a legacy script definition the node kernel can replace without
/// losing semantics. Only <c>function</c> mode qualifies: <c>durable</c> and
/// <c>handler</c> need the linear engine's checkpoint/resume.
/// </summary>
private static bool IsLowerableLegacyScript(WorkflowDefinition def)
=> string.Equals(def.Mode, WorkflowMode.Function, StringComparison.OrdinalIgnoreCase);
/// <summary>
/// The denormalized webhook path stored on <see cref="Workflow.WebhookPath"/>:
/// only a webhook trigger contributes, so the public receiver can filter by
/// the indexed column instead of parsing every trigger.
/// </summary>
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,

47
Services/WorkflowSyncService.cs
View file

@ -32,22 +32,67 @@ public class WorkflowSyncService
private readonly IWorkflowSource _source;
private readonly WorkflowSourceFactory? _sourceFactory;
private readonly ILogger<WorkflowSyncService> _logger;
private readonly SyncGate? _gate;
private readonly ILeaseService? _leases;
public WorkflowSyncService(
WorkflowsDbContext db,
WorkflowCompiler compiler,
IWorkflowSource source,
ILogger<WorkflowSyncService> 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;
}
/// <summary>
/// Serializes the sync against other syncs of the same <c>(tenant, repo)</c>:
/// 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).
/// </summary>
public async Task<SyncResult> 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<SyncResult> 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

7
appsettings.json
View file

@ -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
}
}

46
w4c-workflows-api.Tests/BoundedChannelTests.cs Normal file
View file

@ -0,0 +1,46 @@
using w4c_workflows.Services;
using w4c_workflows.Services.Messaging;
using Xunit;
namespace w4c_workflows.Tests;
/// <summary>
/// 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.
/// </summary>
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<string, string> { ["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}");
}
}

94
w4c-workflows-api.Tests/CSharpScriptExecutorTests.cs
View file

@ -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<string, string?> { ["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<object> 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<object> Main(string input)
{
await Task.Delay(10000);
return new { ok = true };
}
}
""");
using var cts = new CancellationTokenSource();
cts.Cancel();
await Assert.ThrowsAnyAsync<OperationCanceledException>(
() => _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 })!;
}
}

36
w4c-workflows-api.Tests/EgressPinningTests.cs Normal file
View file

@ -0,0 +1,36 @@
using System.Net;
using w4c_workflows.Services.Security;
using Xunit;
namespace w4c_workflows.Tests;
/// <summary>
/// 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.
/// </summary>
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<HttpRequestException>(
() => 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"));
}
}

46
w4c-workflows-api.Tests/EntryPathContainmentTests.cs Normal file
View file

@ -0,0 +1,46 @@
using Microsoft.Extensions.Configuration;
using w4c_workflows.Services.Execution;
using Xunit;
namespace w4c_workflows.Tests;
/// <summary>
/// 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
/// <c>..</c> escape.
/// </summary>
public class EntryPathContainmentTests
{
private static TaskInvocation Invocation(string entryFile, string workingDir) => new(
TaskInvocation.TypeValue, "run-1", "task-1", "key", "shell", entryFile, null,
new Dictionary<string, string>(), 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);
}
}
}

12
w4c-workflows-api.Tests/Fakes.cs
View file

@ -8,11 +8,17 @@ internal sealed class FakeJobQueue : IJobQueue
{
public List<(string TenantId, IReadOnlyDictionary<string, string> Fields)> Enqueued { get; } = new();
public List<(string TenantId, IReadOnlyDictionary<string, string> Fields, string Reason)> DeadLettered { get; } = new();
public List<string> Acked { get; } = new();
/// <summary>When set, <see cref="EnqueueAsync"/> throws — exercises failure paths.</summary>
public bool FailEnqueue { get; set; }
public Task EnsureGroupAsync(string tenantId, CancellationToken ct) => Task.CompletedTask;
public Task<string> EnqueueAsync(string tenantId, IReadOnlyDictionary<string, string> 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<IReadOnlyList<StreamMessage>> ClaimPendingAsync(string tenantId, string consumer, TimeSpan minIdle, int count, CancellationToken ct)
=> Task.FromResult<IReadOnlyList<StreamMessage>>(Array.Empty<StreamMessage>());
public Task<long> AckAsync(string tenantId, string messageId, CancellationToken ct) => Task.FromResult(0L);
public Task<long> AckAsync(string tenantId, string messageId, CancellationToken ct)
{
Acked.Add(messageId);
return Task.FromResult(0L);
}
public Task DeadLetterAsync(string tenantId, string messageId, IReadOnlyDictionary<string, string> fields, string reason, CancellationToken ct)
{

341
w4c-workflows-api.Tests/GraphJobExecutorTests.cs Normal file
View file

@ -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;
/// <summary>
/// S2 worker half: a <c>graph.run</c> job is claimed by the worker and drives the
/// whole node graph through <see cref="NodeWorkflowRunner"/>, recording one
/// <see cref="TaskRun"/> per node and owning the terminal run status.
/// </summary>
[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<INodeExecutor>? extraExecutors = null,
CredentialVault? vault = null)
{
var catalog = new NodeBlueprintCatalog(NodeBlueprintCatalog.LoadEmbedded());
var executors = new List<INodeExecutor>
{
new NoOpNodeExecutor(),
new SetNodeExecutor(),
new IfNodeExecutor(),
new SplitInBatchesNodeExecutor(),
new ExecuteWorkflowNodeExecutor(),
new CodeNodeExecutor(new RuntimeRegistry(new LanguageRegistry(), Array.Empty<IScriptExecutor>())),
};
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<IServiceScopeFactory>(),
jobs,
nodeRunner,
NullLogger<GraphJobExecutor>.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<string>());
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<string, string>
{
["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<OperationCanceledException>(
() => executor.ExecuteAsync(tenantId, GraphJob(run), cts.Token));
Assert.Empty(jobs.Acked);
}
/// <summary>Stands in for the HTTP executor and records the credentials it was handed.</summary>
private sealed class CapturingHttpExecutor : INodeExecutor
{
public string Type => "core.httpRequest";
public IReadOnlyDictionary<string, CredentialData>? Seen { get; private set; }
public Task<NodeExecutionOutcome> RunAsync(NodeExecutionContext context, CancellationToken ct)
{
Seen = context.Credentials;
return Task.FromResult(NodeExecutionOutcome.Single(context.Input(0).ToList()));
}
}
/// <summary>Fails with the resolved secret embedded in the URL (P1-10 regression).</summary>
private sealed class LeakyFailExecutor : INodeExecutor
{
public string Type => "core.httpRequest";
public Task<NodeExecutionOutcome> RunAsync(NodeExecutionContext context, CancellationToken ct)
{
var secret = context.Credentials["httpAuth"].Data["value"]!.GetValue<string>();
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);
}
}

31
w4c-workflows-api.Tests/NodeGraphCompilerTests.cs
View file

@ -186,6 +186,37 @@ public class NodeGraphCompilerTests
Assert.Equal("loop", result.Graph!.EntryNodeId);
}
/// <summary>
/// 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.
/// </summary>
[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()
{

34
w4c-workflows-api.Tests/NodeGraphRunnerTests.cs
View file

@ -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
}
/// <summary>
/// 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.
/// </summary>
[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);
}
}

178
w4c-workflows-api.Tests/RunLifecycleEngineTests.cs
View file

@ -41,38 +41,6 @@ public class RunLifecycleEngineTests
private static TaskDispatcher NewDispatcher(WorkflowsDbContext db, FakeJobQueue jobs)
=> new(db, jobs, new ConfigurationBuilder().Build(), NullLogger<TaskDispatcher>.Instance);
/// <summary>An engine wired with the node kernel so node-mode workflows run in-process.</summary>
private static RunLifecycleEngine NewNodeEngine(
FakeEventBus events,
FakeLeaseService leases,
IEnumerable<INodeExecutor>? extraExecutors = null,
CredentialVault? vault = null)
{
var catalog = new NodeBlueprintCatalog(NodeBlueprintCatalog.LoadEmbedded());
var executors = new List<INodeExecutor>
{
new NoOpNodeExecutor(),
new SetNodeExecutor(),
new IfNodeExecutor(),
new SplitInBatchesNodeExecutor(),
new ExecuteWorkflowNodeExecutor(),
new CodeNodeExecutor(new RuntimeRegistry(new LanguageRegistry(), Array.Empty<IScriptExecutor>())),
};
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<RunLifecycleEngine>.Instance,
nodeRunner);
}
private static IReadOnlyDictionary<string, string> ResultFields(
WorkflowRun run, WorkflowTask task, int attempt, bool success, string? output = null, string? error = null)
{
@ -483,8 +451,15 @@ public class RunLifecycleEngineTests
// ------------------------------------------------------------------ node mode
/// <summary>
/// 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
/// <see cref="GraphJobExecutorTests"/>.
/// </summary>
[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)));
}
/// <summary>
/// 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).
/// </summary>
[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
}
/// <summary>Stands in for the HTTP executor and records the credentials it was handed.</summary>
private sealed class CapturingHttpExecutor : INodeExecutor
{
public string Type => "core.httpRequest";
public IReadOnlyDictionary<string, CredentialData>? Seen { get; private set; }
public Task<NodeExecutionOutcome> 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<string>());
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)));
}
}

21
w4c-workflows-api.Tests/SecretRedactorTests.cs
View file

@ -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" }));
}
}

50
w4c-workflows-api.Tests/WebhooksControllerTests.cs Normal file
View file

@ -0,0 +1,50 @@
using w4c_workflows.Controllers;
using w4c_workflows.Models;
using Xunit;
namespace w4c_workflows.Tests;
/// <summary>
/// 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.
/// </summary>
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"));
}
}

129
w4c-workflows-api.Tests/WorkflowCompilerTests.cs
View file

@ -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);
}
}