17d4515961
The C# worker failed with 'entry file not found: Main.cs' because the dispatcher resolved the working directory against the DEFAULT workflow repo instead of the tenant's currently-selected one, and never ensured the selected repo was cloned on the executing host. This left the worker working in a directory whose code files did not exist. - WorkflowSourceFactory.ResolveCopiesRoot now anchors a relative WorkflowSource:CopiesRoot to the content root (not the process CWD), so a worker started from a container/different CWD still lands on the same wizard root as the control plane. - ForgejoWorkflowRepoService takes IWebHostEnvironment and uses ResolveCopiesRoot the same way. - TaskDispatcher.ResolveWorkerRootForTenantAsync resolves the tenant's SELECTED repo from WorkflowRepoStore, and calls EnsureCloneAsync to materialize that repo on the executing host before resolving the working directory. EnsureClone is a no-op once the clone exists, so repeated dispatches stay cheap; failures are logged and swallowed so the resolved path still surfaces a deterministic error if the repo is genuinely missing. Adds regression tests: TaskDispatcherWorkingDirTests.
321 lines
14 KiB
C#
321 lines
14 KiB
C#
using w4c_workflows.Data;
|
|
using w4c_workflows.Models;
|
|
using w4c_workflows.Services.Execution;
|
|
using w4c_workflows.Services.Messaging;
|
|
|
|
namespace w4c_workflows.Services.Runs;
|
|
|
|
/// <summary>
|
|
/// Creates the <see cref="TaskRun"/> row for an attempt and enqueues the
|
|
/// corresponding <c>task.run</c> job on the tenant's stream. The row is
|
|
/// persisted <em>before</em> the job is enqueued so a fast worker result can
|
|
/// never race a missing TaskRun. The working directory is derived from the
|
|
/// workflow's path under the tenant's code root (either the monorepo checkout
|
|
/// or a per-tenant Forgejo clone, depending on the configured mode).
|
|
/// </summary>
|
|
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 WorkflowRepoStore? _repoStore;
|
|
private readonly ILogger<TaskDispatcher> _logger;
|
|
|
|
public TaskDispatcher(
|
|
WorkflowsDbContext db,
|
|
IJobQueue jobs,
|
|
IConfiguration config,
|
|
ILogger<TaskDispatcher> logger,
|
|
WorkflowSourceFactory? sourceFactory = null,
|
|
WorkflowRepoStore? repoStore = null)
|
|
{
|
|
_db = db;
|
|
_jobs = jobs;
|
|
_logger = logger;
|
|
_sourceFactory = sourceFactory;
|
|
_repoStore = repoStore;
|
|
_defaultWorkerRoot = ResolveWorkerRoot(config);
|
|
_sharedDir = config["WorkflowSource:SharedDir"] ?? "workflows";
|
|
}
|
|
|
|
public async Task<Guid> DispatchAsync(
|
|
WorkflowRun run,
|
|
WorkflowTask task,
|
|
string workflowPath,
|
|
string? input,
|
|
int attempt,
|
|
CancellationToken ct,
|
|
bool isCompensation = false)
|
|
{
|
|
var taskRun = new TaskRun
|
|
{
|
|
Id = Guid.NewGuid(),
|
|
RunId = run.Id,
|
|
TaskId = task.Id,
|
|
Attempt = attempt,
|
|
Status = TaskRunStatus.Running,
|
|
InputJson = input,
|
|
RetryCount = attempt - 1,
|
|
IsCompensation = isCompensation,
|
|
StartedAt = DateTime.UtcNow,
|
|
};
|
|
|
|
_db.TaskRuns.Add(taskRun);
|
|
await _db.SaveChangesAsync(ct);
|
|
|
|
var workingDir = await ResolveWorkingDirAsync(run.TenantId, workflowPath, ct);
|
|
var fields = TaskRunMessage.ToFields(run, task, input, workingDir, attempt);
|
|
await _jobs.EnqueueAsync(run.TenantId, fields, ct);
|
|
|
|
_logger.LogDebug(
|
|
"Dispatched task {TaskKey} (run {RunId}, attempt {Attempt}, compensation {IsCompensation}) for tenant {TenantId}",
|
|
task.Key, run.Id, attempt, isCompensation, run.TenantId);
|
|
|
|
return taskRun.Id;
|
|
}
|
|
|
|
/// <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
|
|
/// tenant DLQ. The control-plane failure path has no live jobs-stream entry
|
|
/// left to acknowledge, so this uses a publish (not copy+ack) write.
|
|
/// </summary>
|
|
public async Task DeadLetterTaskAsync(
|
|
WorkflowRun run, WorkflowTask task, TaskRun taskRun, CancellationToken ct)
|
|
{
|
|
var workingDir = await ResolveWorkingDirAsync(run.TenantId, run.Workflow.Path, ct);
|
|
var fields = TaskRunMessage.ToFields(
|
|
run, task, taskRun.InputJson, workingDir, taskRun.Attempt);
|
|
|
|
var dlqFields = new Dictionary<string, string>(fields)
|
|
{
|
|
["dlq.error"] = taskRun.Error ?? string.Empty,
|
|
};
|
|
|
|
await _jobs.PublishDeadLetterAsync(run.TenantId, dlqFields, "max_retries_exceeded", ct);
|
|
|
|
_logger.LogWarning(
|
|
"Dead-lettered task {TaskKey} (run {RunId}, attempt {Attempt})",
|
|
task.Key, run.Id, taskRun.Attempt);
|
|
}
|
|
|
|
/// <summary>
|
|
/// 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.
|
|
///
|
|
/// <para>
|
|
/// The stored <c>Workflow.Path</c> may or may not carry the shared-dir prefix
|
|
/// (e.g. <c>workflows/ops-heartbeat/workflow.yaml</c> vs
|
|
/// <c>ops-heartbeat/workflow.yaml</c>) depending on how it was compiled. To be
|
|
/// robust against both generations we prefer the direct
|
|
/// <c>{root}/{dir}</c> resolution and fall back to
|
|
/// <c>{root}/{sharedDir}/{dir}</c> 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.
|
|
/// </para>
|
|
/// </summary>
|
|
public string ResolveWorkingDir(string tenantId, string workflowPath)
|
|
{
|
|
var workerRoot = ResolveWorkerRootForTenant(tenantId);
|
|
var dir = Path.GetDirectoryName(workflowPath) ?? string.Empty;
|
|
if (string.IsNullOrWhiteSpace(dir))
|
|
return workerRoot;
|
|
return ResolveExistingDir(workerRoot, dir);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Async variant used by the live dispatch path. In Forgejo-backed mode the
|
|
/// working root is the tenant's shared local clone, whose path depends on the
|
|
/// user's Forgejo login — resolved here (async, DB-backed) so the worker runs
|
|
/// the exact files being edited in the source-code explorer.
|
|
/// </summary>
|
|
public async Task<string> ResolveWorkingDirAsync(string tenantId, string workflowPath, CancellationToken ct = default)
|
|
{
|
|
var workerRoot = await ResolveWorkerRootForTenantAsync(tenantId, ct);
|
|
var dir = Path.GetDirectoryName(workflowPath) ?? string.Empty;
|
|
if (string.IsNullOrWhiteSpace(dir))
|
|
return workerRoot;
|
|
return ResolveExistingDir(workerRoot, dir);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Legacy overload for backward compatibility. Resolves against the default worker root.
|
|
/// </summary>
|
|
public string ResolveWorkingDir(string workflowPath)
|
|
{
|
|
var dir = Path.GetDirectoryName(workflowPath) ?? string.Empty;
|
|
if (string.IsNullOrWhiteSpace(dir))
|
|
return _defaultWorkerRoot;
|
|
return ResolveExistingDir(_defaultWorkerRoot, dir);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Returns the first of <c>{root}/{dir}</c> or <c>{root}/{sharedDir}/{dir}</c>
|
|
/// 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
|
|
/// <paramref name="dir"/> — returns the primary candidate unchanged so the
|
|
/// caller still gets a deterministic path to surface in the error.
|
|
/// </summary>
|
|
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)
|
|
{
|
|
// Resolve the tenant's OWN source directory — the exact directory the
|
|
// workflow-file writer (WorkflowFilesController ->
|
|
// WorkflowSourceFactory.ResolveTenantSourceDir) stores files in. Falling
|
|
// back to the global/git root never matched {CopiesRoot}/{tenantId}, so
|
|
// every task ran against a working dir that did not contain its entry
|
|
// file (surfacing as "entry file not found: Main.cs" in the C# executor).
|
|
if (_sourceFactory != null)
|
|
return _sourceFactory.ResolveTenantSourceDir(tenantId);
|
|
return _defaultWorkerRoot;
|
|
}
|
|
|
|
private async Task<string> ResolveWorkerRootForTenantAsync(string tenantId, CancellationToken ct)
|
|
{
|
|
// Same contract as the sync variant: the run's code root must be the one
|
|
// the writer used. In Forgejo-backed mode this is the tenant's shared
|
|
// local clone (which depends on the user's Forgejo login); otherwise it
|
|
// is the per-tenant filesystem dir. Both are exactly where the writer
|
|
// lands, so the entry file is always found relative to this root.
|
|
if (_sourceFactory == null)
|
|
return _defaultWorkerRoot;
|
|
|
|
// Use the tenant's SELECTED workflow repo (the clone the file-writer and
|
|
// the source explorer land on), falling back to the configured default
|
|
// (repoName=null) when the store is unavailable. Before this, the worker
|
|
// always resolved the DEFAULT repo, so a tenant that had switched to a
|
|
// different repo got a working directory that did not contain its entry
|
|
// file (surfacing as "entry file not found: Main.cs" in the C# executor).
|
|
var repoName = await ResolveRepoNameAsync(tenantId, ct);
|
|
|
|
if (_sourceFactory.IsForgejoBacked)
|
|
{
|
|
var login = await _sourceFactory.ResolveForgejoLoginAsync(tenantId, ct);
|
|
if (!string.IsNullOrWhiteSpace(login))
|
|
{
|
|
// Materialize the tenant's repo clone if it is not already present on
|
|
// THIS instance before resolving the working directory. The writer may
|
|
// have cloned the repo on a different host/instance; without this the
|
|
// worker resolves a working directory whose code files do not exist
|
|
// ("entry file not found: Main.cs"). EnsureClone is a no-op once the
|
|
// clone exists, so repeated dispatches stay cheap.
|
|
await EnsureCloneAsync(tenantId, login, repoName, ct);
|
|
return _sourceFactory.ResolveTenantSourceDir(tenantId, login, repoName);
|
|
}
|
|
}
|
|
|
|
return _sourceFactory.ResolveTenantSourceDir(tenantId, null, repoName);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Best-effort clone/pull of the tenant's workflow repo so the working directory
|
|
/// the worker will execute in actually contains the code files. Failures are
|
|
/// logged and swallowed: the resolved path is still returned, so a genuinely
|
|
/// missing repo surfaces its normal "entry file not found" at execution time
|
|
/// rather than silently dispatching to a directory we cannot reason about.
|
|
/// </summary>
|
|
private async Task EnsureCloneAsync(string tenantId, string login, string? repoName, CancellationToken ct)
|
|
{
|
|
var forgejo = _sourceFactory?.Forgejo;
|
|
if (forgejo == null)
|
|
return;
|
|
|
|
try
|
|
{
|
|
await forgejo.EnsureCloneAsync(tenantId, login, repoName, ct);
|
|
// No per-dispatch PullAsync: a full git pull is too slow to run inside
|
|
// the control-plane dispatch loop on every tick. EnsureClone guarantees
|
|
// the working directory exists; freshness is handled by the compile/edit
|
|
// paths that already pull the repo on write/sync.
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex,
|
|
"Could not ensure workflow repo clone for tenant {TenantId} (login {Login}); dispatching with the resolved path as-is",
|
|
tenantId, login);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Resolves the tenant's currently-selected workflow repo name (as persisted in
|
|
/// <see cref="WorkflowRepoStore"/>). Returns <c>null</c> when the store is not
|
|
/// configured or the lookup fails, letting the caller fall back to the factory's
|
|
/// configured default repo.
|
|
/// </summary>
|
|
private async Task<string?> ResolveRepoNameAsync(string tenantId, CancellationToken ct)
|
|
{
|
|
if (_repoStore == null)
|
|
return null;
|
|
|
|
try
|
|
{
|
|
return await _repoStore.GetNameAsync(tenantId, ct);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogDebug(ex,
|
|
"Could not resolve selected workflow repo for tenant {TenantId}; using default", tenantId);
|
|
return null;
|
|
}
|
|
}
|
|
|
|
private static string ResolveWorkerRoot(IConfiguration config)
|
|
{
|
|
// Explicit override wins: a dedicated worker mount (e.g. a shared checkout
|
|
// volume in deployment) or an explicit repo root.
|
|
var workerRoot = config["Workflows:WorkerRoot"];
|
|
if (!string.IsNullOrWhiteSpace(workerRoot))
|
|
return Path.GetFullPath(workerRoot);
|
|
|
|
var repoRoot = config["SourceCode:RepoRoot"];
|
|
if (!string.IsNullOrWhiteSpace(repoRoot))
|
|
return Path.GetFullPath(repoRoot);
|
|
|
|
// Dev fallback: resolve the monorepo/git root the same way the control
|
|
// plane's WorkflowSource does, so the worker and the source reader agree
|
|
// on where workflow code files live. Without this, a worker started from
|
|
// the service directory would resolve files under <cwd>/workflows/...,
|
|
// which is wrong (the code files sit at the repo root).
|
|
var start = Directory.GetCurrentDirectory();
|
|
var walk = Path.GetFullPath(start);
|
|
while (true)
|
|
{
|
|
if (Directory.Exists(Path.Combine(walk, ".git")))
|
|
return Path.GetFullPath(walk);
|
|
var parent = Directory.GetParent(walk)?.FullName;
|
|
if (string.IsNullOrEmpty(parent) || parent == walk)
|
|
break;
|
|
walk = parent;
|
|
}
|
|
|
|
return Path.GetFullPath(start);
|
|
}
|
|
}
|