-
Notifications
You must be signed in to change notification settings - Fork 0
feat(27-02): map loop failures to typed FailureState records #21
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,67 @@ | ||
| using GsdOrchestrator.Loop; | ||
| using GsdOrchestrator.Scheduling; | ||
| using Xunit; | ||
|
|
||
| namespace GsdOrchestrator.Tests; | ||
|
|
||
| public sealed class FailureStateTests | ||
| { | ||
| [Fact] | ||
| public void Timeout_Is_Classified_As_Transient_And_Retryable() | ||
| { | ||
| var result = FailureClassifier.Classify(new TimeoutException("timed out"), "component"); | ||
|
|
||
| Assert.Equal(FailureClass.Transient, result.FailureClass); | ||
| Assert.True(result.Retryable); | ||
| Assert.Equal(5d, result.RetryAfterSeconds); | ||
| } | ||
|
|
||
| [Fact] | ||
| public void OperationCanceled_Is_Classified_As_Cancellation() | ||
| { | ||
| var result = FailureClassifier.Classify(new OperationCanceledException("cancelled"), "component"); | ||
|
|
||
| Assert.Equal(FailureClass.Cancellation, result.FailureClass); | ||
| Assert.False(result.Retryable); | ||
| Assert.Null(result.RetryAfterSeconds); | ||
| } | ||
|
|
||
| [Fact] | ||
| public void InvalidOperation_Is_Classified_As_Deterministic() | ||
| { | ||
| var result = FailureClassifier.Classify(new InvalidOperationException("Goal 'x' was not found."), "component"); | ||
|
|
||
| Assert.Equal(FailureClass.Deterministic, result.FailureClass); | ||
| Assert.False(result.Retryable); | ||
| } | ||
|
|
||
| [Fact] | ||
| public void UnauthorizedAccess_Is_Classified_As_Policy() | ||
| { | ||
| var result = FailureClassifier.Classify(new UnauthorizedAccessException("denied"), "component"); | ||
|
|
||
| Assert.Equal(FailureClass.Policy, result.FailureClass); | ||
| Assert.False(result.Retryable); | ||
| } | ||
|
|
||
| [Fact] | ||
| public void Unknown_Exception_Falls_Back_To_Unrecoverable() | ||
| { | ||
| var result = FailureClassifier.Classify(new UnknownFailureException("boom"), "component"); | ||
|
|
||
| Assert.Equal(FailureClass.Unrecoverable, result.FailureClass); | ||
| Assert.False(result.Retryable); | ||
| } | ||
|
|
||
| [Fact] | ||
| public void Cause_Chain_Is_Collected_Outermost_First() | ||
| { | ||
| var exception = new InvalidOperationException("outer", new TimeoutException("inner")); | ||
|
|
||
| var result = FailureClassifier.Classify(exception, "component"); | ||
|
|
||
| Assert.Equal(["outer", "inner"], result.CauseChain); | ||
| } | ||
|
|
||
| private sealed class UnknownFailureException(string message) : Exception(message); | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,61 @@ | ||
| using System.Net.Http; | ||
| using System.Net.Sockets; | ||
| using GsdOrchestrator.Scheduling; | ||
|
|
||
| namespace GsdOrchestrator.Loop; | ||
|
|
||
| public sealed record FailureState( | ||
| FailureClass FailureClass, | ||
| string Component, | ||
| string Message, | ||
| bool Retryable, | ||
| string? ExceptionType = null, | ||
| IReadOnlyList<string>? CauseChain = null, | ||
| double? RetryAfterSeconds = null); | ||
|
|
||
| public static class FailureClassifier | ||
| { | ||
| private const int MaxMessageLength = 1024; | ||
| private const int MaxCauseChainEntries = 10; | ||
|
|
||
| public static FailureState Classify(Exception exception, string component) | ||
| { | ||
| var (failureClass, retryable, retryAfterSeconds) = exception switch | ||
| { | ||
| OperationCanceledException => (FailureClass.Cancellation, false, (double?)null), | ||
| TimeoutException => (FailureClass.Transient, true, 5d), | ||
| HttpRequestException => (FailureClass.Transient, true, 5d), | ||
| SocketException => (FailureClass.Transient, true, 5d), | ||
| UnauthorizedAccessException => (FailureClass.Policy, false, (double?)null), | ||
| ArgumentOutOfRangeException => (FailureClass.Policy, false, (double?)null), | ||
| InvalidOperationException => (FailureClass.Deterministic, false, (double?)null), | ||
| ArgumentException => (FailureClass.Deterministic, false, (double?)null), | ||
| _ => (FailureClass.Unrecoverable, false, (double?)null), | ||
| }; | ||
|
|
||
| return new FailureState( | ||
| failureClass, | ||
| component, | ||
| Truncate(exception.Message), | ||
| retryable, | ||
| exception.GetType().FullName, | ||
| BuildCauseChain(exception), | ||
| retryAfterSeconds); | ||
| } | ||
|
|
||
| private static string Truncate(string message) => | ||
| message.Length <= MaxMessageLength ? message : message[..MaxMessageLength]; | ||
|
|
||
| private static IReadOnlyList<string>? BuildCauseChain(Exception exception) | ||
| { | ||
| var chain = new List<string>(capacity: 4); | ||
| var current = exception; | ||
| while (current is not null && chain.Count < MaxCauseChainEntries) | ||
| { | ||
| chain.Add(Truncate(current.Message)); | ||
| current = current.InnerException; | ||
| } | ||
|
|
||
| return chain.Count == 0 ? null : chain; | ||
| } | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -4,6 +4,7 @@ | |
| using GsdOrchestrator.Mcp; | ||
| using GsdOrchestrator.Scheduling; | ||
| using GsdOrchestrator.Verification; | ||
| using Microsoft.Extensions.Logging; | ||
|
|
||
| namespace GsdOrchestrator.Loop; | ||
|
|
||
|
|
@@ -87,7 +88,7 @@ public async Task PublishAsync(TerminalLoopOutcome outcome, CancellationToken ca | |
| } | ||
| } | ||
|
|
||
| public sealed class LoopCoordinator(IGoalStore store, ILoopWorker worker, ILoopVerifier verifier, ITerminalOutcomePublisher learning) | ||
| public sealed class LoopCoordinator(IGoalStore store, ILoopWorker worker, ILoopVerifier verifier, ITerminalOutcomePublisher learning, ILogger<LoopCoordinator> logger) | ||
| { | ||
| public async Task<LoopRunResult> RunAsync( | ||
| string goalId, | ||
|
|
@@ -107,55 +108,93 @@ public async Task<LoopRunResult> RunAsync( | |
| ?? throw new InvalidOperationException($"Work item '{workItem.Id}' could not be leased."); | ||
|
|
||
| aggregate = (await store.LoadAsync(goalId, cancellationToken))!; | ||
| var first = await ExecuteAttemptAsync(aggregate, workItem, 1, false, cancellationToken); | ||
| aggregate = first.Aggregate; | ||
| var workerAttempts = 1; | ||
| var peakConcurrency = first.Work.PeakConcurrency; | ||
| var workerAttempts = 0; | ||
| var peakConcurrency = 0; | ||
| var verificationRuns = 0; | ||
| var repairCreated = false; | ||
|
|
||
| if (first.Work.Succeeded) | ||
| try | ||
| { | ||
| var verification = await verifier.VerifyAsync(goalId, first.Work, cancellationToken); | ||
| verificationRuns++; | ||
| aggregate = AddVerification(aggregate, workItem.Id, verification); | ||
| if (verification.Outcome == VerificationOutcome.Failed) | ||
| var first = await ExecuteAttemptAsync(aggregate, workItem, 1, false, cancellationToken); | ||
| aggregate = first.Aggregate; | ||
| workerAttempts = 1; | ||
| peakConcurrency = first.Work.PeakConcurrency; | ||
|
|
||
| if (first.Work.Succeeded) | ||
| { | ||
| var consumedBudget = repairBudget with | ||
| { | ||
| ConsumedAttempts = Math.Max(repairBudget.ConsumedAttempts, 1), | ||
| ConsumedIterations = Math.Max(repairBudget.ConsumedIterations, 1), | ||
| }; | ||
| var repair = RepairPolicy.CreateRepair(verification, consumedBudget, workItem.Id); | ||
| if (repair is not null) | ||
| var verification = await verifier.VerifyAsync(goalId, first.Work, cancellationToken); | ||
| verificationRuns++; | ||
| aggregate = AddVerification(aggregate, workItem.Id, verification); | ||
| if (verification.Outcome == VerificationOutcome.Failed) | ||
| { | ||
| repairCreated = true; | ||
| aggregate = AddEvent(aggregate, "repair.created", $"attempt-{repair.AttemptNumber}"); | ||
| await store.SaveAsync(aggregate, cancellationToken); | ||
| var repaired = await ExecuteAttemptAsync(aggregate, workItem, repair.AttemptNumber, true, cancellationToken); | ||
| aggregate = repaired.Aggregate; | ||
| workerAttempts++; | ||
| peakConcurrency = Math.Max(peakConcurrency, repaired.Work.PeakConcurrency); | ||
| if (repaired.Work.Succeeded) | ||
| var consumedBudget = repairBudget with | ||
| { | ||
| ConsumedAttempts = Math.Max(repairBudget.ConsumedAttempts, 1), | ||
| ConsumedIterations = Math.Max(repairBudget.ConsumedIterations, 1), | ||
| }; | ||
| var repair = RepairPolicy.CreateRepair(verification, consumedBudget, workItem.Id); | ||
| if (repair is not null) | ||
| { | ||
| verification = await verifier.VerifyAsync(goalId, repaired.Work, cancellationToken); | ||
| verificationRuns++; | ||
| aggregate = AddVerification(aggregate, workItem.Id, verification); | ||
| repairCreated = true; | ||
| aggregate = AddEvent(aggregate, "repair.created", $"attempt-{repair.AttemptNumber}"); | ||
| await store.SaveAsync(aggregate, cancellationToken); | ||
| var repaired = await ExecuteAttemptAsync(aggregate, workItem, repair.AttemptNumber, true, cancellationToken); | ||
| aggregate = repaired.Aggregate; | ||
| workerAttempts++; | ||
| peakConcurrency = Math.Max(peakConcurrency, repaired.Work.PeakConcurrency); | ||
| if (repaired.Work.Succeeded) | ||
| { | ||
| verification = await verifier.VerifyAsync(goalId, repaired.Work, cancellationToken); | ||
| verificationRuns++; | ||
| aggregate = AddVerification(aggregate, workItem.Id, verification); | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| var outcome = aggregate.Evidence.Any(item => item.Kind == "verification-failed") | ||
| && !aggregate.Evidence.Any(item => item.Kind == "verification-passed" && item.Id.EndsWith($"-{verificationRuns}", StringComparison.Ordinal)) | ||
| ? GoalStatus.Failed | ||
| : aggregate.Evidence.Any(item => item.Kind == "verification-passed") ? GoalStatus.Completed : GoalStatus.Failed; | ||
| aggregate = Complete(aggregate, workItem.Id, outcome, lease.Id); | ||
| await store.SaveAsync(aggregate, cancellationToken); | ||
|
|
||
| var evidence = aggregate.Evidence.Select(item => item.Uri).Distinct(StringComparer.Ordinal).ToArray(); | ||
| await learning.PublishAsync(new(goalId, aggregate.Goal.CorrelationId, outcome, $"Goal {outcome} after {workerAttempts} worker attempt(s).", evidence), cancellationToken); | ||
| return new(aggregate, workerAttempts, verificationRuns, repairCreated, peakConcurrency); | ||
| } | ||
| catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) | ||
| { | ||
| var failure = FailureClassifier.Classify(new OperationCanceledException("Loop execution cancelled by caller."), "gsd-orchestrator.LoopCoordinator"); | ||
| logger.LogError( | ||
| "Loop execution failed for goal {GoalId} with failure class {FailureClass}: {Message}", | ||
| goalId, | ||
| failure.FailureClass, | ||
| failure.Message); | ||
| throw; | ||
| } | ||
| catch (Exception ex) | ||
| { | ||
| var failure = FailureClassifier.Classify(ex, "gsd-orchestrator.LoopCoordinator"); | ||
| logger.LogError( | ||
| ex, | ||
| "Loop execution failed for goal {GoalId} with failure class {FailureClass}: {Message}", | ||
| goalId, | ||
| failure.FailureClass, | ||
| failure.Message); | ||
|
|
||
| var outcome = aggregate.Evidence.Any(item => item.Kind == "verification-failed") | ||
| && !aggregate.Evidence.Any(item => item.Kind == "verification-passed" && item.Id.EndsWith($"-{verificationRuns}", StringComparison.Ordinal)) | ||
| ? GoalStatus.Failed | ||
| : aggregate.Evidence.Any(item => item.Kind == "verification-passed") ? GoalStatus.Completed : GoalStatus.Failed; | ||
| aggregate = Complete(aggregate, workItem.Id, outcome, lease.Id); | ||
| await store.SaveAsync(aggregate, cancellationToken); | ||
| aggregate = await store.LoadAsync(goalId, cancellationToken) ?? aggregate; | ||
| var status = failure.FailureClass == FailureClass.Cancellation ? GoalStatus.Cancelled : GoalStatus.Failed; | ||
| var evidenceUri = $"cas://evidence/failure/{failure.FailureClass.ToString().ToLowerInvariant()}/{Guid.NewGuid():N}"; | ||
| aggregate = CompleteWithFailure(aggregate, workItem.Id, lease.Id, status, failure, evidenceUri); | ||
| await store.SaveAsync(aggregate, cancellationToken); | ||
|
|
||
| var evidence = aggregate.Evidence.Select(item => item.Uri).Distinct(StringComparer.Ordinal).ToArray(); | ||
| await learning.PublishAsync(new(goalId, aggregate.Goal.CorrelationId, outcome, $"Goal {outcome} after {workerAttempts} worker attempt(s).", evidence), cancellationToken); | ||
| return new(aggregate, workerAttempts, verificationRuns, repairCreated, peakConcurrency); | ||
| var evidence = aggregate.Evidence.Select(item => item.Uri).Distinct(StringComparer.Ordinal).ToArray(); | ||
| await learning.PublishAsync( | ||
| new(goalId, aggregate.Goal.CorrelationId, status, failure.Message, evidence), | ||
| cancellationToken); | ||
| return new(aggregate, workerAttempts, verificationRuns, repairCreated, peakConcurrency); | ||
| } | ||
| } | ||
|
|
||
| private async Task<(GoalAggregate Aggregate, LoopWorkResult Work)> ExecuteAttemptAsync( | ||
|
|
@@ -215,6 +254,46 @@ private static GoalAggregate Complete(GoalAggregate aggregate, string workItemId | |
| }; | ||
| } | ||
|
|
||
| private static GoalAggregate CompleteWithFailure( | ||
| GoalAggregate aggregate, | ||
| string workItemId, | ||
| string leaseId, | ||
| GoalStatus status, | ||
| FailureState failure, | ||
| string evidenceUri) | ||
| { | ||
| var now = DateTimeOffset.UtcNow; | ||
| var eventPayload = JsonSerializer.Serialize(new | ||
| { | ||
| failureClass = failure.FailureClass.ToString(), | ||
| component = failure.Component, | ||
| message = failure.Message, | ||
| retryable = failure.Retryable, | ||
| exceptionType = failure.ExceptionType, | ||
| causeChain = failure.CauseChain, | ||
| retryAfterSeconds = failure.RetryAfterSeconds, | ||
| }); | ||
|
|
||
| return aggregate with | ||
| { | ||
| Goal = aggregate.Goal with { Status = status }, | ||
| WorkItems = aggregate.WorkItems.Select(item => item.Id == workItemId | ||
| ? item with { Status = status == GoalStatus.Cancelled ? WorkItemStatus.Cancelled : WorkItemStatus.Failed } | ||
| : item).ToArray(), | ||
| Leases = aggregate.Leases.Where(item => item.Id != leaseId).ToArray(), | ||
| BudgetReservations = [], | ||
| Evidence = [.. aggregate.Evidence, new EvidenceRecord(Guid.NewGuid().ToString("N"), aggregate.Goal.Id, workItemId, "failure-state", evidenceUri)], | ||
| Transitions = [.. aggregate.Transitions, new(Guid.NewGuid().ToString("N"), aggregate.Goal.Id, aggregate.Goal.Status.ToString(), status.ToString(), status == GoalStatus.Cancelled ? GoalStopReason.Cancellation.ToString() : GoalStopReason.UnrecoverableFault.ToString(), now)], | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
For any non-cancellation typed failure, including Useful? React with 👍 / 👎. |
||
| Events = | ||
| [ | ||
| .. aggregate.Events, | ||
| new GoalEventRecord(Guid.NewGuid().ToString("N"), aggregate.Goal.Id, | ||
| aggregate.Events.Count == 0 ? 1 : aggregate.Events.Max(item => item.Sequence) + 1, | ||
| "goal.failed", eventPayload, now) | ||
| ] | ||
| }; | ||
| } | ||
|
|
||
| private static GoalEventRecord NewEvent(GoalAggregate aggregate, string type, string payload) => | ||
| new(Guid.NewGuid().ToString("N"), aggregate.Goal.Id, | ||
| aggregate.Events.Count == 0 ? 1 : aggregate.Events.Max(item => item.Sequence) + 1, | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
When
learning.PublishAsyncthrows afterstore.SaveAsynchas already persisted a completed outcome, this call is still inside the broadtry; the catch reloads that terminal aggregate and callsCompleteWithFailure, so a telemetry/MCP publication outage rewrites a successful goal/work item to Failed with a synthetic failure-state. Keep terminal publication failures outside the failure-mapping path or avoid overwriting already-terminal outcomes.Useful? React with 👍 / 👎.