Merge pull request #38 from tass500/feat/iter-32-step-timeout-cancel

Feat/iter 32 step timeout cancel

Author: tass500

backend/LowCodePlatform.Backend.Tests/WorkflowRunEndpointsTests.cs

@@ -98,6 +98,108 @@ private static async Task AuthenticateAsync(HttpClient client, string tenantSlug
         client.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", token);
     }
 
+    [Fact]
+    public async Task Workflow_step_timeoutMs_should_fail_step_and_run_with_timeout_error()
+    {
+        var mgmtDbPath = Path.Combine(Path.GetTempPath(), $"lcp-test-mgmt-{Guid.NewGuid():N}.db");
+        var tenantDbPath = Path.Combine(Path.GetTempPath(), $"lcp-test-tenant-t1-{Guid.NewGuid():N}.db");
+
+        var managementCs = $"Data Source={mgmtDbPath}";
+        var tenantCs = $"Data Source={tenantDbPath}";
+
+        await InitializeDatabasesAsync(managementCs, "t1", tenantCs, CancellationToken.None);
+
+        await using var factory = new TestAppFactory("t1", mgmtDbPath, tenantDbPath);
+        using var client = CreateTenantClient(factory, "t1");
+
+        await AuthenticateAsync(client, "t1");
+
+        var createReq = new
+        {
+            name = "wf-timeout",
+            definitionJson = "{\"steps\":[{\"type\":\"delay\",\"ms\":200,\"timeoutMs\":50}]}"
+        };
+        using var createResp = await client.PostAsJsonAsync("/api/workflows", createReq);
+        Assert.Equal(HttpStatusCode.OK, createResp.StatusCode);
+
+        var created = await createResp.Content.ReadFromJsonAsync<Dictionary<string, object?>>();
+        Assert.NotNull(created);
+        var wfId = Guid.Parse(created!["workflowDefinitionId"]!.ToString()!);
+
+        using var startResp = await client.PostAsync($"/api/workflows/{wfId}/runs", content: null);
+        Assert.Equal(HttpStatusCode.OK, startResp.StatusCode);
+
+        var startPayload = await startResp.Content.ReadFromJsonAsync<Dictionary<string, object?>>();
+        Assert.NotNull(startPayload);
+        var runId = Guid.Parse(startPayload!["workflowRunId"]!.ToString()!);
+
+        using var getResp = await client.GetAsync($"/api/workflows/runs/{runId}");
+        Assert.Equal(HttpStatusCode.OK, getResp.StatusCode);
+
+        var details = await getResp.Content.ReadFromJsonAsync<Dictionary<string, object?>>();
+        Assert.NotNull(details);
+
+        Assert.Equal("failed", details!["state"]?.ToString());
+        Assert.Equal("workflow_step_timed_out", details!["errorCode"]?.ToString());
+    }
+
+    [Fact]
+    public async Task Workflow_run_start_cancellation_should_mark_run_canceled_in_db()
+    {
+        var mgmtDbPath = Path.Combine(Path.GetTempPath(), $"lcp-test-mgmt-{Guid.NewGuid():N}.db");
+        var tenantDbPath = Path.Combine(Path.GetTempPath(), $"lcp-test-tenant-t1-{Guid.NewGuid():N}.db");
+
+        var managementCs = $"Data Source={mgmtDbPath}";
+        var tenantCs = $"Data Source={tenantDbPath}";
+
+        await InitializeDatabasesAsync(managementCs, "t1", tenantCs, CancellationToken.None);
+
+        await using var factory = new TestAppFactory("t1", mgmtDbPath, tenantDbPath);
+        using var client = CreateTenantClient(factory, "t1");
+
+        await AuthenticateAsync(client, "t1");
+
+        var createReq = new
+        {
+            name = "wf-cancel",
+            definitionJson = "{\"steps\":[{\"type\":\"delay\",\"ms\":200}]}"
+        };
+        using var createResp = await client.PostAsJsonAsync("/api/workflows", createReq);
+        Assert.Equal(HttpStatusCode.OK, createResp.StatusCode);
+
+        var created = await createResp.Content.ReadFromJsonAsync<Dictionary<string, object?>>();
+        Assert.NotNull(created);
+        var wfId = Guid.Parse(created!["workflowDefinitionId"]!.ToString()!);
+
+        using var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(50));
+        try
+        {
+            using var _ = await client.PostAsync($"/api/workflows/{wfId}/runs", content: null, cancellationToken: cts.Token);
+        }
+        catch (OperationCanceledException)
+        {
+            // expected: client canceled the request
+        }
+
+        await Task.Delay(150);
+
+        using var listResp = await client.GetAsync($"/api/workflows/{wfId}/runs");
+        Assert.Equal(HttpStatusCode.OK, listResp.StatusCode);
+
+        var list = await listResp.Content.ReadFromJsonAsync<Dictionary<string, object?>>();
+        Assert.NotNull(list);
+
+        var itemsJson = JsonSerializer.Serialize(list!["items"]);
+        using var doc = JsonDocument.Parse(itemsJson);
+        var itemsArr = doc.RootElement;
+        Assert.Equal(JsonValueKind.Array, itemsArr.ValueKind);
+        Assert.True(itemsArr.GetArrayLength() >= 1);
+
+        var latest = itemsArr[0];
+        Assert.Equal("canceled", latest.GetProperty("state").GetString());
+        Assert.Equal("canceled", latest.GetProperty("errorCode").GetString());
+    }
+
     [Fact]
     public async Task Workflow_run_start_and_status_should_work()
     {

backend/Services/WorkflowRunnerService.cs

@@ -13,6 +13,8 @@ public sealed class WorkflowRunnerService
 
     private sealed record RetryPolicy(int MaxAttempts, int DelayMs, double BackoffFactor, int? MaxDelayMs);
 
+    private sealed record StepPolicy(RetryPolicy Retry, int? TimeoutMs);
+
     private readonly PlatformDbContext _db;
 
     public WorkflowRunnerService(PlatformDbContext db)
@@ -76,6 +78,33 @@ private static void ExecuteRequireAsync(WorkflowStepRun step, JsonObject context
         }
     }
 
+    private static StepPolicy ParseStepPolicy(string? stepConfigJson)
+    {
+        var retry = ParseRetryPolicy(stepConfigJson);
+        if (string.IsNullOrWhiteSpace(stepConfigJson))
+            return new StepPolicy(Retry: retry, TimeoutMs: null);
+
+        try
+        {
+            using var doc = JsonDocument.Parse(stepConfigJson);
+            var root = doc.RootElement;
+
+            int? timeoutMs = null;
+            if (root.TryGetProperty("timeoutMs", out var timeoutEl)
+                && timeoutEl.TryGetInt32(out var parsed)
+                && parsed >= 1)
+            {
+                timeoutMs = parsed;
+            }
+
+            return new StepPolicy(Retry: retry, TimeoutMs: timeoutMs);
+        }
+        catch
+        {
+            return new StepPolicy(Retry: retry, TimeoutMs: null);
+        }
+    }
+
     private static void ExecuteSetAsync(WorkflowStepRun step)
     {
         if (string.IsNullOrWhiteSpace(step.StepConfigJson))
@@ -673,32 +702,72 @@ public async Task<WorkflowRun> StartAsync(WorkflowDefinition wf, string traceId,
         _db.WorkflowRuns.Add(run);
         await _db.SaveChangesAsync(ct);
 
-        foreach (var step in run.Steps.OrderBy(x => x.StepKey))
+        try
         {
-            await ExecuteStepAsync(run, step, context, ct);
-            await _db.SaveChangesAsync(ct);
-
-            if (step.State == WorkflowStepRunStates.Succeeded && !string.IsNullOrWhiteSpace(step.OutputJson))
+            foreach (var step in run.Steps.OrderBy(x => x.StepKey))
             {
-                try
+                await ExecuteStepAsync(run, step, context, ct);
+                await _db.SaveChangesAsync(ct);
+
+                if (step.State == WorkflowStepRunStates.Succeeded && !string.IsNullOrWhiteSpace(step.OutputJson))
+                {
+                    try
+                    {
+                        context[step.StepKey] = JsonNode.Parse(step.OutputJson);
+                    }
+                    catch
+                    {
+                        // ignore invalid output json; step succeeded but context won't have it
+                    }
+                }
+
+                if (step.State == WorkflowStepRunStates.Failed)
                 {
-                    context[step.StepKey] = JsonNode.Parse(step.OutputJson);
+                    run.State = WorkflowRunStates.Failed;
+                    run.ErrorCode = step.LastErrorCode;
+                    run.ErrorMessage = step.LastErrorMessage;
+                    run.FinishedAtUtc = DateTime.UtcNow;
+                    await _db.SaveChangesAsync(ct);
+                    return run;
                 }
-                catch
+
+                if (step.State == WorkflowStepRunStates.Canceled)
                 {
-                    // ignore invalid output json; step succeeded but context won't have it
+                    run.State = WorkflowRunStates.Canceled;
+                    run.ErrorCode = step.LastErrorCode;
+                    run.ErrorMessage = step.LastErrorMessage;
+                    run.FinishedAtUtc = DateTime.UtcNow;
+                    await _db.SaveChangesAsync(ct);
+                    return run;
                 }
             }
+        }
+        catch (OperationCanceledException) when (ct.IsCancellationRequested)
+        {
+            var now = DateTime.UtcNow;
+            run.State = WorkflowRunStates.Canceled;
+            run.ErrorCode = "canceled";
+            run.ErrorMessage = "Canceled.";
+            run.FinishedAtUtc = now;
 
-            if (step.State == WorkflowStepRunStates.Failed)
+            foreach (var step in run.Steps.Where(x => x.State is WorkflowStepRunStates.Pending or WorkflowStepRunStates.Running))
             {
-                run.State = WorkflowRunStates.Failed;
-                run.ErrorCode = step.LastErrorCode;
-                run.ErrorMessage = step.LastErrorMessage;
-                run.FinishedAtUtc = DateTime.UtcNow;
-                await _db.SaveChangesAsync(ct);
-                return run;
+                step.State = WorkflowStepRunStates.Canceled;
+                step.FinishedAtUtc ??= now;
+                step.LastErrorCode ??= "canceled";
+                step.LastErrorMessage ??= "Canceled.";
+            }
+
+            try
+            {
+                await _db.SaveChangesAsync(CancellationToken.None);
             }
+            catch
+            {
+                // ignore
+            }
+
+            throw;
         }
 
         run.State = WorkflowRunStates.Succeeded;
@@ -759,17 +828,17 @@ private static List<WorkflowStepRun> BuildSteps(WorkflowDefinition wf)
 
     private async Task ExecuteStepAsync(WorkflowRun run, WorkflowStepRun step, JsonObject context, CancellationToken ct)
     {
-        var policy = ParseRetryPolicy(step.StepConfigJson);
+        var policy = ParseStepPolicy(step.StepConfigJson);
 
         var startedAt = DateTime.UtcNow;
         step.StartedAtUtc = startedAt;
 
         Exception? lastEx = null;
-        for (var attemptNumber = 1; attemptNumber <= policy.MaxAttempts; attemptNumber += 1)
+        for (var attemptNumber = 1; attemptNumber <= policy.Retry.MaxAttempts; attemptNumber += 1)
         {
             ct.ThrowIfCancellationRequested();
 
-            var delayMs = GetRetryDelayMs(policy, attemptNumber);
+            var delayMs = GetRetryDelayMs(policy.Retry, attemptNumber);
             if (delayMs > 0)
                 await Task.Delay(delayMs, ct);
 
@@ -779,21 +848,59 @@ private async Task ExecuteStepAsync(WorkflowRun run, WorkflowStepRun step, JsonO
             step.LastErrorMessage = null;
             step.OutputJson = null;
 
+            CancellationToken attemptCt = ct;
+            CancellationTokenSource? timeoutCts = null;
+            if (policy.TimeoutMs is not null)
+            {
+                timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
+                timeoutCts.CancelAfter(policy.TimeoutMs.Value);
+                attemptCt = timeoutCts.Token;
+            }
+
             try
             {
                 InterpolateStepConfigJson(step, context);
 
-                await ExecuteStepBodyAsync(step, context, ct);
+                await ExecuteStepBodyAsync(step, context, attemptCt);
 
                 step.State = WorkflowStepRunStates.Succeeded;
                 step.FinishedAtUtc = DateTime.UtcNow;
                 return;
             }
+            catch (OperationCanceledException oce)
+            {
+                lastEx = oce;
+
+                if (ct.IsCancellationRequested)
+                {
+                    step.State = WorkflowStepRunStates.Canceled;
+                    step.FinishedAtUtc = DateTime.UtcNow;
+                    step.LastErrorCode ??= "canceled";
+                    step.LastErrorMessage ??= "Canceled.";
+
+                    run.ErrorCode = step.LastErrorCode;
+                    run.ErrorMessage = step.LastErrorMessage;
+                    return;
+                }
+
+                // timeout (linked token canceled)
+                step.LastErrorCode ??= "workflow_step_timed_out";
+                step.LastErrorMessage ??= $"Step timed out after {policy.TimeoutMs} ms.";
+
+                if (attemptNumber < policy.Retry.MaxAttempts)
+                    continue;
+
+                step.State = WorkflowStepRunStates.Failed;
+                step.FinishedAtUtc = DateTime.UtcNow;
+                run.ErrorCode = step.LastErrorCode;
+                run.ErrorMessage = step.LastErrorMessage;
+                return;
+            }
             catch (Exception ex)
             {
                 lastEx = ex;
 
-                if (attemptNumber < policy.MaxAttempts)
+                if (attemptNumber < policy.Retry.MaxAttempts)
                     continue;
 
                 step.State = WorkflowStepRunStates.Failed;
@@ -805,6 +912,10 @@ private async Task ExecuteStepAsync(WorkflowRun run, WorkflowStepRun step, JsonO
                 run.ErrorMessage = step.LastErrorMessage;
                 return;
             }
+            finally
+            {
+                timeoutCts?.Dispose();
+            }
         }
 
         step.State = WorkflowStepRunStates.Failed;

docs/live/02_allapot.md

@@ -21,7 +21,7 @@ Drift-proof observability egy greenfield lowcode platformban.
     - `require`
     - `domainCommand`
     - `unstable`
-  - Jelenlegi fókusz (roadmap): Iteráció 31 — step-level retry/backoff
+  - Jelenlegi fókusz (roadmap): Iteráció 32 — step timeout / cancellation hardening
 
 - **Frontend (Angular)**
   - Drift-proof “now”: kliens oldali **`serverNowOffsetMs`** kalibráció `serverTimeUtc` alapján.