Add failure category plumbing for pipeline retry loops (#14)

* [self-healing]: add failure category plumbing for pipeline retry loops

Run-checker now classifies failures from all trigger types (Glue, SFN,
EMR, EMR Serverless, Databricks) into FailureCategory (TRANSIENT,
TIMEOUT, PERMANENT) and passes them through to the orchestrator.

IsRetryable defaults empty/unknown categories to retryable so that
missing classification never silently disables retries. The orchestrator
logResult action defaults empty failureCategory to TRANSIENT for
backward compatibility with ASL templates that don't pass it yet.

checkReadiness returns "not_ready" with poll metadata (pollAdvised,
failedTraits) instead of bare "skip", enabling ASL readiness polling.

* fix diagram alignment

Author: dwsmith1983

cmd/lambda/orchestrator/main.go

@@ -355,10 +355,11 @@ func checkReadiness(_ context.Context, _ *intlambda.Deps, req intlambda.Orchestr
 	if !allPass {
 		return intlambda.OrchestratorResponse{
 			Action: req.Action,
-			Result: "skip",
+			Result: "not_ready",
 			Payload: map[string]interface{}{
-				"reason":   "not ready",
-				"blocking": blocking,
+				"pollAdvised":  true,
+				"failedTraits": blocking,
+				"message":      fmt.Sprintf("blocked by %d traits", len(blocking)),
 			},
 		}, nil
 	}
@@ -408,6 +409,9 @@ func logResult(ctx context.Context, d *intlambda.Deps, req intlambda.Orchestrato
 	}
 
 	if status == string(types.RunFailed) {
+		if failureCategory == "" {
+			failureCategory = string(types.FailureTransient)
+		}
 		entry.FailureMessage = message
 		entry.FailureCategory = types.FailureCategory(failureCategory)
 

cmd/lambda/orchestrator/main_test.go

@@ -206,9 +206,10 @@ func TestCheckReadiness_HasBlocking(t *testing.T) {
 		},
 	})
 	require.NoError(t, err)
-	assert.Equal(t, "skip", resp.Result)
-	blocking := resp.Payload["blocking"].([]string)
-	assert.Contains(t, blocking, "schema")
+	assert.Equal(t, "not_ready", resp.Result)
+	assert.Equal(t, true, resp.Payload["pollAdvised"])
+	failedTraits := resp.Payload["failedTraits"].([]string)
+	assert.Contains(t, failedTraits, "schema")
 }
 
 func TestCheckReadiness_OptionalFail(t *testing.T) {
@@ -301,6 +302,57 @@ func TestLogResult(t *testing.T) {
 	assert.Equal(t, "proceed", resp.Result)
 }
 
+func TestLogResult_FailedWithCategory(t *testing.T) {
+	d := testDeps(t)
+	seedPipeline(t, d, types.PipelineConfig{Name: "pipe-a"})
+
+	resp, err := handleOrchestrator(context.Background(), d, intlambda.OrchestratorRequest{
+		Action:     "logResult",
+		PipelineID: "pipe-a",
+		ScheduleID: "daily",
+		Payload: map[string]interface{}{
+			"status":          string(types.RunFailed),
+			"runID":           "run-2",
+			"message":         "glue timeout",
+			"failureCategory": string(types.FailureTimeout),
+		},
+	})
+	require.NoError(t, err)
+	assert.Equal(t, "proceed", resp.Result)
+	assert.Equal(t, true, resp.Payload["retryable"])
+
+	// Verify category was persisted
+	date := time.Now().UTC().Format("2006-01-02")
+	entry, err := d.Provider.GetRunLog(context.Background(), "pipe-a", date, "daily")
+	require.NoError(t, err)
+	assert.Equal(t, types.FailureTimeout, entry.FailureCategory)
+}
+
+func TestLogResult_FailedEmptyCategoryDefaultsTransient(t *testing.T) {
+	d := testDeps(t)
+	seedPipeline(t, d, types.PipelineConfig{Name: "pipe-a"})
+
+	resp, err := handleOrchestrator(context.Background(), d, intlambda.OrchestratorRequest{
+		Action:     "logResult",
+		PipelineID: "pipe-a",
+		ScheduleID: "daily",
+		Payload: map[string]interface{}{
+			"status":  string(types.RunFailed),
+			"runID":   "run-3",
+			"message": "unknown error",
+		},
+	})
+	require.NoError(t, err)
+	assert.Equal(t, "proceed", resp.Result)
+	assert.Equal(t, true, resp.Payload["retryable"])
+
+	// Verify empty category defaulted to transient
+	date := time.Now().UTC().Format("2006-01-02")
+	entry, err := d.Provider.GetRunLog(context.Background(), "pipe-a", date, "daily")
+	require.NoError(t, err)
+	assert.Equal(t, types.FailureTransient, entry.FailureCategory)
+}
+
 // --- releaseLock ---
 
 func TestReleaseLock(t *testing.T) {

cmd/lambda/run-checker/main.go

@@ -42,8 +42,9 @@ func handleRunCheck(ctx context.Context, d *intlambda.Deps, req intlambda.RunChe
 	}
 
 	return intlambda.RunCheckResponse{
-		State:   result.State,
-		Message: result.Message,
+		State:           result.State,
+		Message:         result.Message,
+		FailureCategory: result.FailureCategory,
 	}, nil
 }
 

cmd/lambda/run-checker/main_test.go

@@ -177,18 +177,19 @@ func TestHandleRunCheck_UnknownType(t *testing.T) {
 
 func TestHandleRunCheck_GlueStatusMapping(t *testing.T) {
 	tests := []struct {
-		name     string
-		state    gluetypes.JobRunState
-		expected trigger.RunCheckState
+		name            string
+		state           gluetypes.JobRunState
+		expected        trigger.RunCheckState
+		failureCategory types.FailureCategory
 	}{
-		{"succeeded", gluetypes.JobRunStateSucceeded, trigger.RunCheckSucceeded},
-		{"failed", gluetypes.JobRunStateFailed, trigger.RunCheckFailed},
-		{"timeout", gluetypes.JobRunStateTimeout, trigger.RunCheckFailed},
-		{"stopped", gluetypes.JobRunStateStopped, trigger.RunCheckFailed},
-		{"error", gluetypes.JobRunStateError, trigger.RunCheckFailed},
-		{"running", gluetypes.JobRunStateRunning, trigger.RunCheckRunning},
-		{"starting", gluetypes.JobRunStateStarting, trigger.RunCheckRunning},
-		{"waiting", gluetypes.JobRunStateWaiting, trigger.RunCheckRunning},
+		{"succeeded", gluetypes.JobRunStateSucceeded, trigger.RunCheckSucceeded, ""},
+		{"failed", gluetypes.JobRunStateFailed, trigger.RunCheckFailed, types.FailureTransient},
+		{"timeout", gluetypes.JobRunStateTimeout, trigger.RunCheckFailed, types.FailureTimeout},
+		{"stopped", gluetypes.JobRunStateStopped, trigger.RunCheckFailed, types.FailureTransient},
+		{"error", gluetypes.JobRunStateError, trigger.RunCheckFailed, types.FailureTransient},
+		{"running", gluetypes.JobRunStateRunning, trigger.RunCheckRunning, ""},
+		{"starting", gluetypes.JobRunStateStarting, trigger.RunCheckRunning, ""},
+		{"waiting", gluetypes.JobRunStateWaiting, trigger.RunCheckRunning, ""},
 	}
 
 	for _, tt := range tests {
@@ -215,6 +216,7 @@ func TestHandleRunCheck_GlueStatusMapping(t *testing.T) {
 			require.NoError(t, err)
 			assert.Equal(t, tt.expected, resp.State)
 			assert.Equal(t, string(tt.state), resp.Message)
+			assert.Equal(t, tt.failureCategory, resp.FailureCategory)
 		})
 	}
 }

docs/content/docs/guides/retry-loop-asl.md

@@ -0,0 +1,159 @@
+---
+title: "Retry Loop ASL Pattern"
+weight: 10
+---
+
+# Step Function Retry & Readiness Polling Pattern
+
+This guide shows the recommended ASL patterns for implementing retry loops and readiness polling in your Step Function state machine.
+
+## Prerequisites
+
+These patterns require interlock v0.2.2+ which adds:
+
+- `failureCategory` in run-checker responses (classifies failures as `TRANSIENT`, `TIMEOUT`, or `PERMANENT`)
+- `retryable` and `retryBackoffSeconds` in orchestrator `logResult` responses
+- `not_ready` result with `pollAdvised` from orchestrator `checkReadiness`
+
+## Failure Retry Loop
+
+When a job fails, the orchestrator's `logResult` action returns retry metadata. The ASL can use this to loop back and retry.
+
+```json
+{
+  "LogRunFailed": {
+    "Type": "Task",
+    "Resource": "${OrchestratorArn}",
+    "Parameters": {
+      "action": "logResult",
+      "pipelineID.$": "$.pipelineID",
+      "scheduleID.$": "$.scheduleID",
+      "payload": {
+        "status": "FAILED",
+        "runID.$": "$.runID",
+        "message.$": "$.failureMessage",
+        "failureCategory.$": "$.failureCategory"
+      }
+    },
+    "ResultPath": "$.logResult",
+    "Next": "IsRetryable"
+  },
+
+  "IsRetryable": {
+    "Type": "Choice",
+    "Choices": [
+      {
+        "Variable": "$.logResult.payload.retryable",
+        "BooleanEquals": true,
+        "Next": "WaitRetryBackoff"
+      }
+    ],
+    "Default": "ReleaseLockFailed"
+  },
+
+  "WaitRetryBackoff": {
+    "Type": "Wait",
+    "SecondsPath": "$.logResult.payload.retryBackoffSeconds",
+    "Next": "AcquireLock"
+  }
+}
+```
+
+### How it works
+
+1. `LogRunFailed` calls the orchestrator with `failureCategory` from the run-checker
+2. The orchestrator computes `retryable` (based on category + attempt count + max attempts) and `retryBackoffSeconds`
+3. `IsRetryable` branches: if retryable, wait and loop back to `AcquireLock`; otherwise, proceed to final cleanup
+4. `WaitRetryBackoff` uses `SecondsPath` for dynamic exponential backoff
+
+### Backward compatibility
+
+If the ASL does not pass `failureCategory`, the orchestrator defaults it to `TRANSIENT`, making the failure retryable. This ensures existing deployments get retry behavior without ASL changes.
+
+## Readiness Polling
+
+When traits fail (data not ready), the orchestrator returns `not_ready` with poll metadata. The ASL can use this to wait and re-evaluate.
+
+```json
+{
+  "CheckReadiness": {
+    "Type": "Task",
+    "Resource": "${OrchestratorArn}",
+    "Parameters": {
+      "action": "checkReadiness",
+      "pipelineID.$": "$.pipelineID",
+      "payload": {
+        "traitResults.$": "$.traitResults"
+      }
+    },
+    "ResultPath": "$.readiness",
+    "Next": "IsReady"
+  },
+
+  "IsReady": {
+    "Type": "Choice",
+    "Choices": [
+      {
+        "Variable": "$.readiness.result",
+        "StringEquals": "proceed",
+        "Next": "TriggerJob"
+      },
+      {
+        "Variable": "$.readiness.result",
+        "StringEquals": "not_ready",
+        "Next": "WaitReadiness"
+      }
+    ],
+    "Default": "HandleEvaluatorError"
+  },
+
+  "WaitReadiness": {
+    "Type": "Wait",
+    "Seconds": 60,
+    "Next": "AcquireLock"
+  }
+}
+```
+
+### How it works
+
+1. `CheckReadiness` evaluates trait results and returns `proceed`, `not_ready`, or `error`
+2. `IsReady` branches on the result:
+   - `proceed`: all required traits pass, trigger the job
+   - `not_ready`: data not ready yet, wait and re-evaluate (loops back to `AcquireLock`)
+   - `error`: evaluator infrastructure failure, handle separately
+3. `WaitReadiness` pauses before re-evaluation (use a fixed interval or compute dynamically)
+
+### Backward compatibility
+
+The previous `skip` result is replaced by `not_ready`. Existing ASL templates that check `result == "proceed"` with a default fallback will treat `not_ready` the same as `skip` — both hit the default path. No ASL changes are required for existing deployments to continue working.
+
+## Complete Flow
+
+The recommended state machine flow combining both patterns:
+
+```
+AcquireLock → CheckRunLog → ResolvePipeline → EvaluateTraits → CheckReadiness
+                                                                    │
+                                                         ┌──────────┼──────────┐
+                                                         │          │          │
+                                                     proceed    not_ready    error
+                                                         │          │          │
+                                                    TriggerJob  Wait(60s)  Alert+Skip
+                                                         │          │
+                                                    PollStatus  → AcquireLock
+                                                         │
+                                                   ┌─────┼─────┐
+                                                   │           │
+                                               succeeded    failed
+                                                   │           │
+                                              LogCompleted  LogFailed
+                                                               │
+                                                         ┌─────┼─────┐
+                                                         │           │
+                                                     retryable  non-retryable
+                                                         │           │
+                                                    Wait(backoff)  Cleanup
+                                                         │
+                                                    AcquireLock
+```

internal/lambda/types.go

@@ -94,6 +94,7 @@ type RunCheckRequest struct {
 
 // RunCheckResponse is the output of the run-checker Lambda.
 type RunCheckResponse struct {
-	State   trigger.RunCheckState `json:"state"`
-	Message string                `json:"message"`
+	State           trigger.RunCheckState `json:"state"`
+	Message         string                `json:"message"`
+	FailureCategory types.FailureCategory `json:"failureCategory,omitempty"`
 }

internal/schedule/retry.go

@@ -40,10 +40,15 @@ func CalculateBackoff(policy types.RetryPolicy, attempt int) time.Duration {
 }
 
 // IsRetryable returns whether a failure category should be retried.
+// An empty category defaults to retryable — better to retry once too many
+// than silently drop a run that could have recovered.
 func IsRetryable(policy types.RetryPolicy, category types.FailureCategory) bool {
 	if category == types.FailurePermanent {
 		return false
 	}
+	if category == "" {
+		return true
+	}
 	if len(policy.RetryableFailures) == 0 {
 		// Default: retry transient and timeout
 		return category == types.FailureTransient || category == types.FailureTimeout

internal/schedule/retry_test.go

@@ -69,6 +69,18 @@ func TestIsRetryable(t *testing.T) {
 	}
 }
 
+func TestIsRetryable_EmptyCategory(t *testing.T) {
+	policy := DefaultRetryPolicy()
+	assert.True(t, IsRetryable(policy, ""))
+}
+
+func TestIsRetryable_EmptyCategory_CustomPolicy(t *testing.T) {
+	policy := types.RetryPolicy{
+		RetryableFailures: []types.FailureCategory{types.FailureTransient},
+	}
+	assert.True(t, IsRetryable(policy, ""))
+}
+
 func TestIsRetryable_EmptyPolicyDefaults(t *testing.T) {
 	policy := types.RetryPolicy{}
 

internal/trigger/databricks.go

@@ -134,11 +134,15 @@ func (r *Runner) checkDatabricksStatus(ctx context.Context, metadata map[string]
 		if resultState == "SUCCESS" {
 			return StatusResult{State: RunCheckSucceeded, Message: msg}, nil
 		}
-		return StatusResult{State: RunCheckFailed, Message: msg}, nil
+		return StatusResult{State: RunCheckFailed, Message: msg, FailureCategory: types.FailureTransient}, nil
 	}
 
-	if lifeCycleState == "INTERNAL_ERROR" || lifeCycleState == "SKIPPED" {
-		return StatusResult{State: RunCheckFailed, Message: msg}, nil
+	if lifeCycleState == "SKIPPED" {
+		return StatusResult{State: RunCheckFailed, Message: msg, FailureCategory: types.FailurePermanent}, nil
+	}
+
+	if lifeCycleState == "INTERNAL_ERROR" {
+		return StatusResult{State: RunCheckFailed, Message: msg, FailureCategory: types.FailureTransient}, nil
 	}
 
 	return StatusResult{State: RunCheckRunning, Message: msg}, nil