diff --git a/arazzo/engine.go b/arazzo/engine.go index 995d5412..ada2e6e4 100644 --- a/arazzo/engine.go +++ b/arazzo/engine.go @@ -44,6 +44,19 @@ type ExecutionResponse struct { // EngineConfig configures engine behavior. type EngineConfig struct { RetainResponseBodies bool // If false, nil out response bodies after extracting outputs + + // SleepFunc, when non-nil, replaces the engine's default wait between + // retry attempts (a failure action's retryAfter). The engine calls it + // with the workflow context and the retryAfter duration; returning an + // error aborts the workflow, exactly as a cancelled context does with + // the default sleeper. + // + // Injecting it lets a host substitute its own timer: a durable workflow + // runtime (Temporal, DBOS, Restate) passes its replay-safe durable + // sleep — a real time.Timer inside a replayed workflow body would block + // the scheduler or break determinism — and tests pass a recording fake + // to avoid real waits. Nil uses the default context-aware timer. + SleepFunc func(ctx context.Context, d time.Duration) error } // Engine orchestrates the execution of Arazzo workflows. @@ -292,7 +305,7 @@ func (e *Engine) runWorkflow(ctx context.Context, workflowId string, inputs map[ } if actionResult.retryCurrent { retryCounts[step.StepId]++ - if err := sleepWithContext(ctx, actionResult.retryAfter); err != nil { + if err := e.sleep(ctx, actionResult.retryAfter); err != nil { result.Success = false result.Error = err break @@ -428,6 +441,16 @@ func stepFailureOrDefault(stepID string, stepErr error) error { return &StepFailureError{StepId: stepID, CriterionIndex: -1} } +// sleep waits between retry attempts, through the configured SleepFunc when +// one is set (see EngineConfig.SleepFunc) and the default context-aware +// timer otherwise. +func (e *Engine) sleep(ctx context.Context, d time.Duration) error { + if e.config != nil && e.config.SleepFunc != nil { + return e.config.SleepFunc(ctx, d) + } + return sleepWithContext(ctx, d) +} + // parseExpression parses and caches an expression. func (e *Engine) parseExpression(input string) (expression.Expression, error) { if cached, ok := e.exprCache[input]; ok { diff --git a/arazzo/engine_test.go b/arazzo/engine_test.go index 71fcee9a..4dfb3c32 100644 --- a/arazzo/engine_test.go +++ b/arazzo/engine_test.go @@ -7,6 +7,7 @@ import ( "context" "errors" "testing" + "time" "github.com/pb33f/libopenapi/arazzo/expression" high "github.com/pb33f/libopenapi/datamodel/high/arazzo" @@ -674,3 +675,62 @@ func TestEngine_RunWorkflow_RetainResponseBodiesHonorsConfig(t *testing.T) { assert.NotNil(t, exec.response.Body) }) } + +func TestEngine_RunWorkflow_CustomSleepFunc(t *testing.T) { + retryAfter := 0.05 + retryLimit := int64(3) + doc := &high.Arazzo{ + Workflows: []*high.Workflow{ + { + WorkflowId: "wf1", + Steps: []*high.Step{ + { + StepId: "s1", + OperationId: "op1", + SuccessCriteria: []*high.Criterion{ + {Condition: "$statusCode == 200"}, + }, + OnFailure: []*high.FailureAction{ + {Name: "retry", Type: "retry", RetryAfter: &retryAfter, RetryLimit: &retryLimit}, + }, + }, + }, + }, + }, + } + executor := &sequenceExecutor{ + statuses: map[string][]int{ + "op1": {500, 200}, + }, + } + + t.Run("custom sleeper replaces the default wait", func(t *testing.T) { + var slept []time.Duration + engine := NewEngineWithConfig(doc, executor, nil, &EngineConfig{ + SleepFunc: func(_ context.Context, d time.Duration) error { + slept = append(slept, d) + return nil + }, + }) + result, err := engine.RunWorkflow(context.Background(), "wf1", nil) + require.NoError(t, err) + require.True(t, result.Success) + require.Len(t, slept, 1) + assert.Equal(t, 50*time.Millisecond, slept[0]) + assert.Equal(t, []string{"op1", "op1"}, executor.operationIDs) + }) + + t.Run("sleeper error aborts the workflow", func(t *testing.T) { + executor := &sequenceExecutor{ + statuses: map[string][]int{"op1": {500, 200}}, + } + sleepErr := errors.New("timer torn down") + engine := NewEngineWithConfig(doc, executor, nil, &EngineConfig{ + SleepFunc: func(context.Context, time.Duration) error { return sleepErr }, + }) + result, err := engine.RunWorkflow(context.Background(), "wf1", nil) + require.NoError(t, err) + require.False(t, result.Success) + assert.ErrorIs(t, result.Error, sleepErr) + }) +}