From 6f10e330692afb95d2288afea9135372885003ce Mon Sep 17 00:00:00 2001 From: Vitali sharp8n Date: Wed, 2 Sep 2026 18:10:08 +0300 Subject: [PATCH] Fix working-dir resolution + add trigger backpressure ResolveWorkingDir now prefers the on-disk layout (root/dir, else root/sharedDir/dir) so workflows compiled with or without the shared-dir prefix stop spawning failing jobs against a missing cwd. TriggerScheduler defers an auto-trigger once a tenant has >= Workflow:TriggerMaxInFlight (default 5) in-flight runs, so a slow worker no longer stacks a never-draining backlog (the root of the run-timeout overload). --- Services/Runs/TaskDispatcher.cs | 49 +++++++++++++++++++++++++-- Services/Triggers/TriggerScheduler.cs | 44 ++++++++++++++++++++++-- 2 files changed, 89 insertions(+), 4 deletions(-) diff --git a/Services/Runs/TaskDispatcher.cs b/Services/Runs/TaskDispatcher.cs index 19a8f0e..c880845 100644 --- a/Services/Runs/TaskDispatcher.cs +++ b/Services/Runs/TaskDispatcher.cs @@ -18,6 +18,7 @@ public class TaskDispatcher private readonly WorkflowsDbContext _db; private readonly IJobQueue _jobs; private readonly string _defaultWorkerRoot; + private readonly string _sharedDir; private readonly WorkflowSourceFactory? _sourceFactory; private readonly ILogger _logger; @@ -33,6 +34,7 @@ public class TaskDispatcher _logger = logger; _sourceFactory = sourceFactory; _defaultWorkerRoot = ResolveWorkerRoot(config); + _sharedDir = config["WorkflowSource:SharedDir"] ?? "workflows"; } public async Task DispatchAsync( @@ -100,6 +102,18 @@ public class TaskDispatcher /// Resolves the absolute working directory for a workflow's code files. /// In Forgejo-backed mode, resolves relative to the tenant's local clone. /// In filesystem-only mode, resolves relative to the global worker root. + /// + /// + /// The stored Workflow.Path may or may not carry the shared-dir prefix + /// (e.g. workflows/ops-heartbeat/workflow.yaml vs + /// ops-heartbeat/workflow.yaml) depending on how it was compiled. To be + /// robust against both generations we prefer the direct + /// {root}/{dir} resolution and fall back to + /// {root}/{sharedDir}/{dir} only when the direct path does not exist but + /// the shared-dir layout does. This prevents dispatching a job whose working + /// directory does not exist — which is exactly what made every such task + /// spawn-fail and cascade into the timeout/retry overload. + /// /// public string ResolveWorkingDir(string tenantId, string workflowPath) { @@ -107,7 +121,7 @@ public class TaskDispatcher var dir = Path.GetDirectoryName(workflowPath) ?? string.Empty; if (string.IsNullOrWhiteSpace(dir)) return workerRoot; - return Path.Combine(workerRoot, dir.Replace('/', Path.DirectorySeparatorChar)); + return ResolveExistingDir(workerRoot, dir); } /// @@ -118,7 +132,38 @@ public class TaskDispatcher var dir = Path.GetDirectoryName(workflowPath) ?? string.Empty; if (string.IsNullOrWhiteSpace(dir)) return _defaultWorkerRoot; - return Path.Combine(_defaultWorkerRoot, dir.Replace('/', Path.DirectorySeparatorChar)); + return ResolveExistingDir(_defaultWorkerRoot, dir); + } + + /// + /// Returns the first of {root}/{dir} or {root}/{sharedDir}/{dir} + /// that exists on disk (in that order, so the newer, already-prefixed path + /// wins). If neither exists — or the shared dir is already part of + /// — returns the primary candidate unchanged so the + /// caller still gets a deterministic path to surface in the error. + /// + private string ResolveExistingDir(string root, string dir) + { + var primary = Path.Combine(root, dir.Replace('/', Path.DirectorySeparatorChar)); + + if (string.IsNullOrWhiteSpace(_sharedDir)) + return primary; + + // If dir is already under the shared dir (e.g. "workflows/..."), a + // fallback would double the prefix — don't. + var sharedPrefix = _sharedDir.Replace('/', Path.DirectorySeparatorChar); + var dirPrefix = dir.Replace('/', Path.DirectorySeparatorChar); + if (dirPrefix == sharedPrefix + || dirPrefix.StartsWith(sharedPrefix + Path.DirectorySeparatorChar, StringComparison.Ordinal)) + return primary; + + var fallback = Path.Combine(root, sharedPrefix, dirPrefix.Replace('/', Path.DirectorySeparatorChar)); + + // Prefer the layout that actually contains the code files. + if (Directory.Exists(fallback) && !Directory.Exists(primary)) + return fallback; + + return primary; } private string ResolveWorkerRootForTenant(string tenantId) diff --git a/Services/Triggers/TriggerScheduler.cs b/Services/Triggers/TriggerScheduler.cs index 3677e1d..152ec78 100644 --- a/Services/Triggers/TriggerScheduler.cs +++ b/Services/Triggers/TriggerScheduler.cs @@ -19,6 +19,7 @@ public class TriggerScheduler : BackgroundService private readonly TimeProvider _time; private readonly ILogger _logger; private readonly int _pollSeconds; + private readonly int _maxInFlight; public TriggerScheduler( IServiceScopeFactory scopeFactory, @@ -32,6 +33,12 @@ public class TriggerScheduler : BackgroundService _time = time; _logger = logger; _pollSeconds = config.GetValue("Workflows:SchedulerPollSeconds", 30); + // Backpressure: never let an auto-trigger pile up more in-flight runs + // than the worker can plausibly clear. When the cap is exceeded the + // trigger is deferred (lastFire is NOT advanced) so it fires again once + // the backlog drains — a saturated control plane keeps creating work + // otherwise, which is how run queues stack into overload. + _maxInFlight = config.GetValue("Workflows:TriggerMaxInFlight", 5); } protected override async Task ExecuteAsync(CancellationToken stoppingToken) @@ -80,11 +87,11 @@ public class TriggerScheduler : BackgroundService return; if (!workflow.TriggerEnabled) continue; // auto-trigger toggled off from the UI; manual runs still work - await EvaluateWorkflowAsync(workflow, launcher, now, ct); + await EvaluateWorkflowAsync(workflow, db, launcher, now, ct); } } - private async Task EvaluateWorkflowAsync(Workflow workflow, IRunLauncher launcher, DateTimeOffset now, CancellationToken ct) + private async Task EvaluateWorkflowAsync(Workflow workflow, WorkflowsDbContext db, IRunLauncher launcher, DateTimeOffset now, CancellationToken ct) { var spec = TriggerSpec.Parse(workflow.TriggerJson, out var error); if (spec == null) @@ -101,6 +108,18 @@ public class TriggerScheduler : BackgroundService if (!TriggerDue.IsDue(spec, lastFire, now)) return; + // Backpressure guard: if the tenant already has a full queue of + // in-flight runs for this workflow, defer the trigger. This is what + // stops an interval/cron trigger from stacking a never-draining + // backlog when the worker cannot keep up (single-core / slow tenants). + if (await IsOverloadedAsync(workflow, db, ct)) + { + _logger.LogDebug( + "Deferring {Type} trigger for workflow {WorkflowId} (tenant {TenantId}): max {Max} in-flight runs reached", + spec.Type, workflow.Id, workflow.TenantId, _maxInFlight); + return; + } + var correlation = $"trigger:{workflow.Id}:{now.ToUnixTimeSeconds()}"; await launcher.LaunchAsync( new LaunchRequest(workflow.TenantId, workflow.Id, workflow.TriggerJson, null, correlation), ct); @@ -116,4 +135,25 @@ public class TriggerScheduler : BackgroundService workflow.Id, workflow.TenantId); } } + + /// + /// Returns true when the tenant already has or more + /// in-flight runs (pending/running/compensating) for the given workflow. A + /// deferred trigger leaves lastFire untouched so the same schedule is + /// re-evaluated next tick once the backlog clears. + /// + private async Task IsOverloadedAsync(Workflow workflow, WorkflowsDbContext db, CancellationToken ct) + { + if (_maxInFlight <= 0) + return false; + + var inFlight = await db.WorkflowRuns.CountAsync(r => + r.WorkflowId == workflow.Id + && r.TenantId == workflow.TenantId + && (r.Status == RunStatus.Pending + || r.Status == RunStatus.Running + || r.Status == RunStatus.Compensating), ct); + + return inFlight >= _maxInFlight; + } }