arazzo: make the retry sleeper injectable via EngineConfig.SleepFunc

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 {

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)
+	})
+}