From ca6b870ee6e355c8b0be13c8fcb8bad4f3942d56 Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Fri, 11 Sep 2026 22:52:21 +0700 Subject: [PATCH 01/11] fix: anchor absolute SLA deadlines to the execution date CalculateAbsoluteDeadline rolled an explicit execution date forward by a day (or an hour) whenever the deadline had already passed. That made sla-monitor cancel publish SLA_MET for runs that finished late, made the reconcile breach branch unreachable, and made the watchdog schedule breach alerts a day late. Roll-forward now only applies when no execution date was supplied, and uses AddDate so the wall clock survives DST. Tests that depended on the wall clock now inject NowFunc. --- internal/lambda/sla_monitor_test.go | 53 ++++++++++++++++++--------- internal/lambda/watchdog_sla.go | 5 +-- internal/lambda/watchdog_test.go | 29 ++++++++++++--- pkg/sla/deadline.go | 19 +++++++--- pkg/sla/deadline_test.go | 56 +++++++++++++++++++++++++++++ 5 files changed, 136 insertions(+), 26 deletions(-) diff --git a/internal/lambda/sla_monitor_test.go b/internal/lambda/sla_monitor_test.go index 10c969c..5e27a95 100644 --- a/internal/lambda/sla_monitor_test.go +++ b/internal/lambda/sla_monitor_test.go @@ -68,7 +68,8 @@ func TestSLAMonitor_Calculate_Midnight(t *testing.T) { } func TestSLAMonitor_Calculate_ReturnsRFC3339(t *testing.T) { - d := &lambda.Deps{Logger: slog.Default()} + now := time.Date(2026, 6, 15, 6, 0, 0, 0, time.UTC) + d := &lambda.Deps{Logger: slog.Default(), NowFunc: func() time.Time { return now }} out, err := lambda.HandleSLAMonitor(context.Background(), d, lambda.SLAMonitorInput{ Mode: "calculate", PipelineID: "gold-orders", @@ -113,31 +114,49 @@ func TestSLAMonitor_Calculate_RelativeDeadline(t *testing.T) { } } -func TestSLAMonitor_Calculate_DailyDeadlineRollsForward(t *testing.T) { - // Daily pipeline with execution date in the past and a small-hour deadline. - // The SLA deadline "02:00" for date 2026-03-04 means 2026-03-05T02:00:00Z - // (next day) because 2026-03-04T02:00 is already past. - d := &lambda.Deps{Logger: slog.Default()} +func TestSLAMonitor_Calculate_ExplicitDateDoesNotRollForward(t *testing.T) { + // A daily pipeline's SLA deadline is anchored to its execution date. Even + // when that deadline is already past, the breach time must stay on the + // execution date, otherwise a late run is scored against tomorrow's deadline. + now := time.Date(2026, 3, 5, 9, 0, 0, 0, time.UTC) + d := &lambda.Deps{Logger: slog.Default(), NowFunc: func() time.Time { return now }} out, err := lambda.HandleSLAMonitor(context.Background(), d, lambda.SLAMonitorInput{ Mode: "calculate", PipelineID: "silver-cdr-day", - Date: "2024-01-15", + Date: "2026-03-04", Deadline: "02:00", ExpectedDuration: "30m", }) if err != nil { t.Fatalf("unexpected error: %v", err) } - // breachAt should NOT be 2024-01-15T02:00:00Z (in the past). - // It should roll forward by 24h to 2024-01-16T02:00:00Z. - if out.BreachAt == "2024-01-15T02:00:00Z" { - t.Errorf("breachAt = %q, should have rolled forward past now", out.BreachAt) + if out.BreachAt != "2026-03-04T02:00:00Z" { + t.Errorf("breachAt = %q, want %q", out.BreachAt, "2026-03-04T02:00:00Z") } - if out.BreachAt != "2024-01-16T02:00:00Z" { - t.Errorf("breachAt = %q, want %q", out.BreachAt, "2024-01-16T02:00:00Z") + if out.WarningAt != "2026-03-04T01:30:00Z" { + t.Errorf("warningAt = %q, want %q", out.WarningAt, "2026-03-04T01:30:00Z") } - if out.WarningAt != "2024-01-16T01:30:00Z" { - t.Errorf("warningAt = %q, want %q", out.WarningAt, "2024-01-16T01:30:00Z") +} + +func TestSLAMonitor_Calculate_NoDateRollsForward(t *testing.T) { + // With no execution date the deadline is anchored to "now", so a deadline + // that already passed today rolls forward to tomorrow. + now := time.Date(2026, 3, 5, 9, 0, 0, 0, time.UTC) + d := &lambda.Deps{Logger: slog.Default(), NowFunc: func() time.Time { return now }} + out, err := lambda.HandleSLAMonitor(context.Background(), d, lambda.SLAMonitorInput{ + Mode: "calculate", + PipelineID: "silver-cdr-day", + Deadline: "02:00", + ExpectedDuration: "30m", + }) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if out.BreachAt != "2026-03-06T02:00:00Z" { + t.Errorf("breachAt = %q, want %q", out.BreachAt, "2026-03-06T02:00:00Z") + } + if out.WarningAt != "2026-03-06T01:30:00Z" { + t.Errorf("warningAt = %q, want %q", out.WarningAt, "2026-03-06T01:30:00Z") } } @@ -492,6 +511,7 @@ func TestSLAMonitor_Cancel_RecalculatesWhenTimesNotProvided(t *testing.T) { EventBridge: eb, EventBusName: "test-bus", Logger: slog.Default(), + NowFunc: func() time.Time { return time.Date(2026, 12, 31, 23, 45, 0, 0, time.UTC) }, } // Cancel with deadline/expectedDuration instead of warningAt/breachAt. @@ -998,7 +1018,8 @@ func TestSLAMonitor_Reconcile_Breach(t *testing.T) { } func TestSLAMonitor_Reconcile_ReturnsDeadlines(t *testing.T) { - d := &lambda.Deps{Logger: slog.Default()} + now := time.Date(2026, 6, 15, 6, 0, 0, 0, time.UTC) + d := &lambda.Deps{Logger: slog.Default(), NowFunc: func() time.Time { return now }} out, err := lambda.HandleSLAMonitor(context.Background(), d, lambda.SLAMonitorInput{ Mode: "reconcile", diff --git a/internal/lambda/watchdog_sla.go b/internal/lambda/watchdog_sla.go index 38b5d86..be89897 100644 --- a/internal/lambda/watchdog_sla.go +++ b/internal/lambda/watchdog_sla.go @@ -159,8 +159,9 @@ func checkTriggerDeadlines(ctx context.Context, d *Deps) error { // resolveWatchdogSLADate determines the execution date for SLA scheduling. // - Hourly pipelines (relative deadline like ":30"): previous hour composite // date, e.g. "2026-03-05T13" when the clock is 14:xx. -// - Daily pipelines (absolute deadline like "02:00"): today's date, -// so handleSLACalculate rolls the deadline forward to the next occurrence. +// - Daily pipelines (absolute deadline like "02:00"): today's date. The +// deadline is anchored to that date; sensor-triggered daily pipelines get +// slaDate shifted to T+1 below because their data completes the next day. func resolveWatchdogSLADate(cfg *types.PipelineConfig, now time.Time) string { if strings.HasPrefix(cfg.SLA.Deadline, ":") { prev := now.Add(-time.Hour) diff --git a/internal/lambda/watchdog_test.go b/internal/lambda/watchdog_test.go index d687b13..43bc572 100644 --- a/internal/lambda/watchdog_test.go +++ b/internal/lambda/watchdog_test.go @@ -620,10 +620,15 @@ func TestWatchdog_ScheduleSLAAlerts_CreatesSchedules(t *testing.T) { d.SchedulerRoleARN = "arn:aws:iam::123:role/scheduler-role" d.SchedulerGroupName = "interlock-sla" - // Use a daily absolute deadline ("02:00") — handleSLACalculate rolls it - // forward when past, so breach is always in the future regardless of - // when this test runs. Hourly ":MM" deadlines are time-dependent and - // would fail if the previous hour's breach is already past. + // Deterministic clock: 00:30 UTC is before the 02:00 SLA deadline, so the + // breach is in the future and proactive schedules are created. Without this + // the test depends on the wall clock (deadlines are no longer rolled + // forward a day when they have already passed). + fixedNow := time.Date(2026, 3, 10, 0, 30, 0, 0, time.UTC) + d.NowFunc = func() time.Time { return fixedNow } + d.StartedAt = fixedNow.Add(-24 * time.Hour) + + // Daily absolute deadline ("02:00") with a fixed clock before the deadline. cfg := types.PipelineConfig{ Pipeline: types.PipelineIdentity{ID: "silver-cdr-day"}, Schedule: types.ScheduleConfig{ @@ -688,6 +693,14 @@ func TestWatchdog_ScheduleSLAAlerts_ConflictSkips(t *testing.T) { d.SchedulerRoleARN = "arn:aws:iam::123:role/scheduler-role" d.SchedulerGroupName = "interlock-sla" + // Deterministic clock: 00:30 UTC is before the 02:00 SLA deadline, so the + // breach is in the future and proactive schedules are created. Without this + // the test depends on the wall clock (deadlines are no longer rolled + // forward a day when they have already passed). + fixedNow := time.Date(2026, 3, 10, 0, 30, 0, 0, time.UTC) + d.NowFunc = func() time.Time { return fixedNow } + d.StartedAt = fixedNow.Add(-24 * time.Hour) + // Use daily deadline to avoid time-dependent breach-past skip. cfg := types.PipelineConfig{ Pipeline: types.PipelineIdentity{ID: "silver-cdr-day"}, @@ -717,6 +730,14 @@ func TestWatchdog_ScheduleSLAAlerts_DailyPipeline(t *testing.T) { d.SchedulerRoleARN = "arn:aws:iam::123:role/scheduler-role" d.SchedulerGroupName = "interlock-sla" + // Deterministic clock: 00:30 UTC is before the 02:00 SLA deadline, so the + // breach is in the future and proactive schedules are created. Without this + // the test depends on the wall clock (deadlines are no longer rolled + // forward a day when they have already passed). + fixedNow := time.Date(2026, 3, 10, 0, 30, 0, 0, time.UTC) + d.NowFunc = func() time.Time { return fixedNow } + d.StartedAt = fixedNow.Add(-24 * time.Hour) + cfg := types.PipelineConfig{ Pipeline: types.PipelineIdentity{ID: "silver-cdr-day"}, Schedule: types.ScheduleConfig{ diff --git a/pkg/sla/deadline.go b/pkg/sla/deadline.go index 346b246..c50f56e 100644 --- a/pkg/sla/deadline.go +++ b/pkg/sla/deadline.go @@ -35,13 +35,19 @@ func CalculateAbsoluteDeadline(date, deadline, expectedDuration, timezone string now = now.In(loc) - // Parse the execution date. + // Parse the execution date. hasExplicitDate records whether the caller + // supplied a parseable execution date. When it did, the deadline is + // anchored to that date and must NOT roll forward: rolling forward makes a + // run that finished after its deadline look like it met the SLA, makes the + // reconcile breach branch unreachable, and schedules breach alerts a day late. baseDate := now baseHour := -1 + hasExplicitDate := false if date != "" { datePart, hourPart := parseExecutionDate(date) parsed, parseErr := time.Parse("2006-01-02", datePart) if parseErr == nil { + hasExplicitDate = true if hourPart != "" { h := 0 if hVal, atoiErr := strconv.Atoi(hourPart); atoiErr == nil { @@ -67,8 +73,10 @@ func CalculateAbsoluteDeadline(date, deadline, expectedDuration, timezone string breach = time.Date(baseDate.Year(), baseDate.Month(), baseDate.Day(), hour, dl.Minute(), 0, 0, loc) if baseHour >= 0 { + // Composite date "YYYY-MM-DDThh": data for hour hh completes in hh+1. breach = breach.Add(time.Hour) - } else if breach.Before(now) { + } else if !hasExplicitDate && breach.Before(now) { + // No execution date: anchor to the next occurrence after now. breach = breach.Add(time.Hour) } } else { @@ -78,8 +86,11 @@ func CalculateAbsoluteDeadline(date, deadline, expectedDuration, timezone string } breach = time.Date(baseDate.Year(), baseDate.Month(), baseDate.Day(), dl.Hour(), dl.Minute(), 0, 0, loc) - if breach.Before(now) { - breach = breach.Add(24 * time.Hour) + if !hasExplicitDate && breach.Before(now) { + // No execution date: anchor to the next occurrence after now. + // AddDate keeps the wall-clock time across DST transitions; + // Add(24*time.Hour) would shift it by an hour. + breach = breach.AddDate(0, 0, 1) } } diff --git a/pkg/sla/deadline_test.go b/pkg/sla/deadline_test.go index a7de3ad..aded8ba 100644 --- a/pkg/sla/deadline_test.go +++ b/pkg/sla/deadline_test.go @@ -94,6 +94,62 @@ func TestCalculateAbsoluteDeadline(t *testing.T) { now: time.Date(2026, 3, 28, 6, 0, 0, 0, time.UTC), wantErr: true, }, + { + name: "explicit date does not roll forward when deadline already passed", + date: "2026-06-15", + deadline: "14:00", + expectedDuration: "15m", + timezone: "UTC", + now: time.Date(2026, 9, 11, 12, 0, 0, 0, time.UTC), + wantBreach: time.Date(2026, 6, 15, 14, 0, 0, 0, time.UTC), + wantWarning: time.Date(2026, 6, 15, 13, 45, 0, 0, time.UTC), + }, + { + name: "explicit hourly date does not roll forward when deadline already passed", + date: "2026-06-15T10", + deadline: ":30", + expectedDuration: "10m", + timezone: "UTC", + now: time.Date(2026, 9, 11, 12, 0, 0, 0, time.UTC), + wantBreach: time.Date(2026, 6, 15, 11, 30, 0, 0, time.UTC), + wantWarning: time.Date(2026, 6, 15, 11, 20, 0, 0, time.UTC), + }, + { + name: "empty date rolls forward one day when deadline already passed", + date: "", + deadline: "08:00", + expectedDuration: "30m", + timezone: "UTC", + now: time.Date(2026, 3, 28, 9, 0, 0, 0, time.UTC), + wantBreach: time.Date(2026, 3, 29, 8, 0, 0, 0, time.UTC), + wantWarning: time.Date(2026, 3, 29, 7, 30, 0, 0, time.UTC), + }, + { + name: "empty date with minute deadline rolls forward one hour when passed", + date: "", + deadline: ":15", + expectedDuration: "5m", + timezone: "UTC", + now: time.Date(2026, 3, 28, 9, 30, 0, 0, time.UTC), + wantBreach: time.Date(2026, 3, 28, 10, 15, 0, 0, time.UTC), + wantWarning: time.Date(2026, 3, 28, 10, 10, 0, 0, time.UTC), + }, + { + name: "empty date roll-forward preserves wall clock across DST spring forward", + date: "", + deadline: "08:00", + expectedDuration: "30m", + timezone: "America/New_York", + now: time.Date(2026, 3, 7, 14, 0, 0, 0, time.UTC), // 09:00 EST + wantBreach: func() time.Time { + loc, _ := time.LoadLocation("America/New_York") + return time.Date(2026, 3, 8, 8, 0, 0, 0, loc) + }(), + wantWarning: func() time.Time { + loc, _ := time.LoadLocation("America/New_York") + return time.Date(2026, 3, 8, 7, 30, 0, 0, loc) + }(), + }, } for _, tt := range tests { From c421bad38e7d47bda3f625d7bc389b8f02a1dc5d Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Fri, 11 Sep 2026 22:58:16 +0700 Subject: [PATCH 02/11] fix: recognise every trigger run-ID metadata key ExtractRunID only matched runId/jobRunId/glue_job_run_id/executionArn/ stepId/dagRunId, but the sfn, emr, emr-serverless, airflow and databricks executors emit sfn_execution_arn, emr_step_id, emr_sl_job_run_id, airflow_dag_run_id and databricks_run_id. The resulting empty runId was omitted from the Lambda result, so the CheckJob state raised States.Runtime after the external job had already been launched. OrchestratorOutput.RunID is now always marshaled and an empty metadata map takes the same sync-sentinel path as nil. --- internal/lambda/orchestrator.go | 42 +++++-- internal/lambda/orchestrator/trigger.go | 20 +--- internal/lambda/orchestrator/trigger_test.go | 116 +++++++++++++++++++ internal/lambda/types.go | 2 +- 4 files changed, 153 insertions(+), 27 deletions(-) create mode 100644 internal/lambda/orchestrator/trigger_test.go diff --git a/internal/lambda/orchestrator.go b/internal/lambda/orchestrator.go index 5832c28..32974b4 100644 --- a/internal/lambda/orchestrator.go +++ b/internal/lambda/orchestrator.go @@ -101,13 +101,13 @@ func handleTrigger(ctx context.Context, d *Deps, input OrchestratorInput) (Orche return OrchestratorOutput{}, fmt.Errorf("%s", errMsg) } - runID := extractRunID(metadata) + runID := ExtractRunID(metadata) if err := PublishEvent(ctx, d, string(types.EventJobTriggered), input.PipelineID, input.ScheduleID, input.Date, fmt.Sprintf("triggered %s job", cfg.Job.Type)); err != nil { d.Logger.WarnContext(ctx, "failed to publish event", "type", types.EventJobTriggered, "error", err) } - if metadata == nil { + if len(metadata) == 0 { if err := d.Store.WriteJobEvent(ctx, input.PipelineID, input.ScheduleID, input.Date, types.JobEventSuccess, "sync", 0, fmt.Sprintf("%s trigger completed synchronously", cfg.Job.Type)); err != nil { d.Logger.Warn("failed to write sync job success joblog", "error", err, "pipeline", input.PipelineID, "schedule", input.ScheduleID, "date", input.Date) @@ -362,16 +362,34 @@ func buildTriggerConfig(job types.JobConfig) (types.TriggerConfig, error) { return tc, nil } -// extractRunID searches trigger metadata for a recognisable run identifier. -func extractRunID(metadata map[string]interface{}) string { - if metadata == nil { - return "" - } - for _, key := range []string{"runId", "jobRunId", "glue_job_run_id", "executionArn", "stepId", "dagRunId"} { - if v, ok := metadata[key]; ok { - if s, ok := v.(string); ok && s != "" { - return s - } +// runIDMetadataKeys lists the trigger-metadata keys that carry a remote run +// identifier, in priority order. The first six are the keys emitted by the +// executors in internal/trigger; the remaining generic keys are retained for +// backward compatibility with records written by older versions. +var runIDMetadataKeys = []string{ + "glue_job_run_id", + "emr_step_id", + "emr_sl_job_run_id", + "sfn_execution_arn", + "airflow_dag_run_id", + "databricks_run_id", + "runId", + "jobRunId", + "executionArn", + "stepId", + "dagRunId", +} + +// ExtractRunID searches trigger metadata for a recognisable run identifier. +// A nil map is safe: indexing a nil map returns the zero value. +func ExtractRunID(metadata map[string]interface{}) string { + for _, key := range runIDMetadataKeys { + v, ok := metadata[key] + if !ok { + continue + } + if s, ok := v.(string); ok && s != "" { + return s } } return "" diff --git a/internal/lambda/orchestrator/trigger.go b/internal/lambda/orchestrator/trigger.go index b330541..4022117 100644 --- a/internal/lambda/orchestrator/trigger.go +++ b/internal/lambda/orchestrator/trigger.go @@ -46,8 +46,9 @@ func handleTrigger(ctx context.Context, d *lambda.Deps, input lambda.Orchestrato // Non-polling triggers (http, command, lambda) complete synchronously // during Execute. Write success to joblog immediately and set a sentinel - // runId so the Step Functions CheckJob JSONPath resolves. - if metadata == nil { + // runId so the Step Functions CheckJob JSONPath resolves. An empty + // (non-nil) map carries no run identity either, so it takes the same path. + if len(metadata) == 0 { if err := d.Store.WriteJobEvent(ctx, input.PipelineID, input.ScheduleID, input.Date, types.JobEventSuccess, "sync", 0, fmt.Sprintf("%s trigger completed synchronously", cfg.Job.Type)); err != nil { d.Logger.Warn("failed to write sync job success joblog", "error", err, "pipeline", input.PipelineID, "schedule", input.ScheduleID, "date", input.Date) @@ -92,19 +93,10 @@ func BuildTriggerConfig(job types.JobConfig) (types.TriggerConfig, error) { } // ExtractRunID searches trigger metadata for a recognisable run identifier. +// The key list is owned by the parent lambda package so the two orchestrator +// implementations cannot drift apart. func ExtractRunID(metadata map[string]interface{}) string { - if metadata == nil { - return "" - } - // Priority order of common identifier keys across trigger types. - for _, key := range []string{"runId", "jobRunId", "glue_job_run_id", "executionArn", "stepId", "dagRunId"} { - if v, ok := metadata[key]; ok { - if s, ok := v.(string); ok && s != "" { - return s - } - } - } - return "" + return lambda.ExtractRunID(metadata) } // InjectDateArgs parses the execution date and injects --par_day (and --par_hour diff --git a/internal/lambda/orchestrator/trigger_test.go b/internal/lambda/orchestrator/trigger_test.go new file mode 100644 index 0000000..0dde3aa --- /dev/null +++ b/internal/lambda/orchestrator/trigger_test.go @@ -0,0 +1,116 @@ +package orchestrator_test + +import ( + "encoding/json" + "testing" + + "github.com/dwsmith1983/interlock/internal/lambda" + "github.com/dwsmith1983/interlock/internal/lambda/orchestrator" +) + +// TestExtractRunID covers the metadata shape emitted by every trigger type in +// internal/trigger. The Step Functions CheckJob state dereferences +// $.triggerResult.runId unconditionally, so a shape that yields no run ID +// must still be visible here rather than silently producing States.Runtime. +func TestExtractRunID(t *testing.T) { + tests := []struct { + name string + metadata map[string]interface{} + want string + }{ + { + name: "glue", + metadata: map[string]interface{}{"glue_job_name": "my-etl", "glue_job_run_id": "jr_abc"}, + want: "jr_abc", + }, + { + name: "emr", + metadata: map[string]interface{}{"emr_cluster_id": "j-123", "emr_step_id": "s-456"}, + want: "s-456", + }, + { + name: "emr-serverless", + metadata: map[string]interface{}{"emr_sl_application_id": "app-1", "emr_sl_job_run_id": "run-9"}, + want: "run-9", + }, + { + name: "step-function", + metadata: map[string]interface{}{"sfn_execution_arn": "arn:aws:states:us-east-1:1:execution:sm:e1"}, + want: "arn:aws:states:us-east-1:1:execution:sm:e1", + }, + { + name: "airflow", + metadata: map[string]interface{}{ + "airflow_dag_run_id": "manual__2026-03-01", + "airflow_dag_id": "dag-1", + "airflow_url": "https://airflow.example.com", + }, + want: "manual__2026-03-01", + }, + { + name: "databricks", + metadata: map[string]interface{}{"databricks_workspace_url": "https://dbc", "databricks_run_id": "778899"}, + want: "778899", + }, + { + name: "command returns nil metadata", + metadata: nil, + want: "", + }, + { + name: "http returns nil metadata", + metadata: nil, + want: "", + }, + { + name: "lambda returns nil metadata", + metadata: nil, + want: "", + }, + { + name: "legacy generic runId key", + metadata: map[string]interface{}{"runId": "abc-123"}, + want: "abc-123", + }, + { + name: "empty value falls through to the next key", + metadata: map[string]interface{}{"runId": "", "glue_job_run_id": "jr-fallback"}, + want: "jr-fallback", + }, + { + name: "non-string value is ignored", + metadata: map[string]interface{}{"glue_job_run_id": 42}, + want: "", + }, + { + name: "unrecognised shape yields empty run id", + metadata: map[string]interface{}{"statusCode": float64(200), "responseBody": "OK"}, + want: "", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := orchestrator.ExtractRunID(tt.metadata); got != tt.want { + t.Errorf("ExtractRunID() = %q, want %q", got, tt.want) + } + }) + } +} + +// TestOrchestratorOutput_AlwaysMarshalsRunID guards the CheckJob and +// JobPollExhausted Parameters, which dereference $.triggerResult.runId. +// An omitted key raises States.Runtime after the external job already started. +func TestOrchestratorOutput_AlwaysMarshalsRunID(t *testing.T) { + data, err := json.Marshal(lambda.OrchestratorOutput{Mode: "trigger"}) + if err != nil { + t.Fatalf("marshal: %v", err) + } + var decoded map[string]interface{} + if err := json.Unmarshal(data, &decoded); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if _, ok := decoded["runId"]; !ok { + t.Errorf("marshaled OrchestratorOutput = %s, want a runId key even when empty", data) + } +} diff --git a/internal/lambda/types.go b/internal/lambda/types.go index 503661e..79ef744 100644 --- a/internal/lambda/types.go +++ b/internal/lambda/types.go @@ -70,7 +70,7 @@ type OrchestratorOutput struct { Mode string `json:"mode"` Status string `json:"status,omitempty"` // "passed" or "not_ready" Results interface{} `json:"results,omitempty"` - RunID string `json:"runId,omitempty"` + RunID string `json:"runId"` // always emitted: SFN CheckJob dereferences $.triggerResult.runId JobType string `json:"jobType,omitempty"` Event string `json:"event,omitempty"` // success, fail, timeout Error string `json:"error,omitempty"` From 5bf2b6842618c5ba34185e61691746e28ab0a450 Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Fri, 11 Sep 2026 23:02:11 +0700 Subject: [PATCH 03/11] test: add exported DynamoDB fake for cross-package tests internal/lambda/orchestrator and deploy both need a store.DynamoAPI double, but every existing mock lives in a package-private _test.go file. storetest exposes a hook-based fake plus a *store.Store constructor and is imported only from test files. --- internal/store/storetest/fake.go | 85 +++++++++++++++++++++++++++ internal/store/storetest/fake_test.go | 69 ++++++++++++++++++++++ 2 files changed, 154 insertions(+) create mode 100644 internal/store/storetest/fake.go create mode 100644 internal/store/storetest/fake_test.go diff --git a/internal/store/storetest/fake.go b/internal/store/storetest/fake.go new file mode 100644 index 0000000..b987672 --- /dev/null +++ b/internal/store/storetest/fake.go @@ -0,0 +1,85 @@ +// Package storetest provides test doubles for the store package. It is used +// only from _test.go files (internal/lambda/orchestrator, deploy); no +// production code imports it. +package storetest + +import ( + "context" + + "github.com/aws/aws-sdk-go-v2/service/dynamodb" + + "github.com/dwsmith1983/interlock/internal/store" +) + +// FakeDynamo is a store.DynamoAPI whose behaviour is supplied by optional +// function hooks. Unset hooks return empty successful responses. +type FakeDynamo struct { + GetItemFn func(ctx context.Context, in *dynamodb.GetItemInput) (*dynamodb.GetItemOutput, error) + PutItemFn func(ctx context.Context, in *dynamodb.PutItemInput) (*dynamodb.PutItemOutput, error) + UpdateItemFn func(ctx context.Context, in *dynamodb.UpdateItemInput) (*dynamodb.UpdateItemOutput, error) + DeleteItemFn func(ctx context.Context, in *dynamodb.DeleteItemInput) (*dynamodb.DeleteItemOutput, error) + QueryFn func(ctx context.Context, in *dynamodb.QueryInput) (*dynamodb.QueryOutput, error) + ScanFn func(ctx context.Context, in *dynamodb.ScanInput) (*dynamodb.ScanOutput, error) +} + +// Compile-time check. +var _ store.DynamoAPI = (*FakeDynamo)(nil) + +// GetItem dispatches to GetItemFn or returns an empty response. +func (f *FakeDynamo) GetItem(ctx context.Context, in *dynamodb.GetItemInput, _ ...func(*dynamodb.Options)) (*dynamodb.GetItemOutput, error) { + if f.GetItemFn != nil { + return f.GetItemFn(ctx, in) + } + return &dynamodb.GetItemOutput{}, nil +} + +// PutItem dispatches to PutItemFn or returns an empty response. +func (f *FakeDynamo) PutItem(ctx context.Context, in *dynamodb.PutItemInput, _ ...func(*dynamodb.Options)) (*dynamodb.PutItemOutput, error) { + if f.PutItemFn != nil { + return f.PutItemFn(ctx, in) + } + return &dynamodb.PutItemOutput{}, nil +} + +// UpdateItem dispatches to UpdateItemFn or returns an empty response. +func (f *FakeDynamo) UpdateItem(ctx context.Context, in *dynamodb.UpdateItemInput, _ ...func(*dynamodb.Options)) (*dynamodb.UpdateItemOutput, error) { + if f.UpdateItemFn != nil { + return f.UpdateItemFn(ctx, in) + } + return &dynamodb.UpdateItemOutput{}, nil +} + +// DeleteItem dispatches to DeleteItemFn or returns an empty response. +func (f *FakeDynamo) DeleteItem(ctx context.Context, in *dynamodb.DeleteItemInput, _ ...func(*dynamodb.Options)) (*dynamodb.DeleteItemOutput, error) { + if f.DeleteItemFn != nil { + return f.DeleteItemFn(ctx, in) + } + return &dynamodb.DeleteItemOutput{}, nil +} + +// Query dispatches to QueryFn or returns an empty response. +func (f *FakeDynamo) Query(ctx context.Context, in *dynamodb.QueryInput, _ ...func(*dynamodb.Options)) (*dynamodb.QueryOutput, error) { + if f.QueryFn != nil { + return f.QueryFn(ctx, in) + } + return &dynamodb.QueryOutput{}, nil +} + +// Scan dispatches to ScanFn or returns an empty response. +func (f *FakeDynamo) Scan(ctx context.Context, in *dynamodb.ScanInput, _ ...func(*dynamodb.Options)) (*dynamodb.ScanOutput, error) { + if f.ScanFn != nil { + return f.ScanFn(ctx, in) + } + return &dynamodb.ScanOutput{}, nil +} + +// NewStore returns a *store.Store backed by the given client with fixed table names. +func NewStore(client store.DynamoAPI) *store.Store { + return &store.Store{ + Client: client, + ControlTable: "control", + JobLogTable: "joblog", + RerunTable: "rerun", + EventsTable: "events", + } +} diff --git a/internal/store/storetest/fake_test.go b/internal/store/storetest/fake_test.go new file mode 100644 index 0000000..41911db --- /dev/null +++ b/internal/store/storetest/fake_test.go @@ -0,0 +1,69 @@ +package storetest_test + +import ( + "context" + "errors" + "testing" + + "github.com/aws/aws-sdk-go-v2/service/dynamodb" + ddbtypes "github.com/aws/aws-sdk-go-v2/service/dynamodb/types" + + "github.com/dwsmith1983/interlock/internal/store/storetest" +) + +func TestFakeDynamo_DefaultsReturnEmptyResponses(t *testing.T) { + f := &storetest.FakeDynamo{} + ctx := context.Background() + + out, err := f.GetItem(ctx, &dynamodb.GetItemInput{}) + if err != nil || out.Item != nil { + t.Errorf("GetItem = (%v, %v), want (empty, nil)", out, err) + } + q, err := f.Query(ctx, &dynamodb.QueryInput{}) + if err != nil || len(q.Items) != 0 { + t.Errorf("Query = (%v, %v), want (empty, nil)", q, err) + } + if _, err := f.PutItem(ctx, &dynamodb.PutItemInput{}); err != nil { + t.Errorf("PutItem err = %v, want nil", err) + } + if _, err := f.UpdateItem(ctx, &dynamodb.UpdateItemInput{}); err != nil { + t.Errorf("UpdateItem err = %v, want nil", err) + } + if _, err := f.DeleteItem(ctx, &dynamodb.DeleteItemInput{}); err != nil { + t.Errorf("DeleteItem err = %v, want nil", err) + } + if _, err := f.Scan(ctx, &dynamodb.ScanInput{}); err != nil { + t.Errorf("Scan err = %v, want nil", err) + } +} + +func TestFakeDynamo_HooksAreInvoked(t *testing.T) { + want := errors.New("dynamodb: internal error") + f := &storetest.FakeDynamo{ + GetItemFn: func(context.Context, *dynamodb.GetItemInput) (*dynamodb.GetItemOutput, error) { + return nil, want + }, + QueryFn: func(context.Context, *dynamodb.QueryInput) (*dynamodb.QueryOutput, error) { + return &dynamodb.QueryOutput{ + Items: []map[string]ddbtypes.AttributeValue{ + {"SK": &ddbtypes.AttributeValueMemberS{Value: "JOB#daily#2026-03-01#1"}}, + }, + }, nil + }, + } + + if _, err := f.GetItem(context.Background(), &dynamodb.GetItemInput{}); !errors.Is(err, want) { + t.Errorf("GetItem err = %v, want %v", err, want) + } + q, err := f.Query(context.Background(), &dynamodb.QueryInput{}) + if err != nil || len(q.Items) != 1 { + t.Errorf("Query = (%v, %v), want 1 item", q, err) + } +} + +func TestNewStore_SetsTableNames(t *testing.T) { + s := storetest.NewStore(&storetest.FakeDynamo{}) + if s.ControlTable != "control" || s.JobLogTable != "joblog" || s.RerunTable != "rerun" || s.EventsTable != "events" { + t.Errorf("NewStore tables = %q/%q/%q/%q", s.ControlTable, s.JobLogTable, s.RerunTable, s.EventsTable) + } +} From 86344279a0c4a4f86ff39aa3630bb3c2ad7c4650 Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Fri, 11 Sep 2026 23:07:03 +0700 Subject: [PATCH 04/11] fix: make evaluate and trigger results match the ASL contract handleEvaluate returned an output without status on storage failures, so the IsReady Choice dereferenced a missing $.evaluateResult.status and raised States.Runtime. It now always emits a status ('error' routes back into the evaluation loop). handleTrigger returned a partial result with a nil error on config failures. HasTriggerResult saw IsPresent=true, CheckJob dereferenced the missing runId, and the execution died with the TRIGGER# lock stuck RUNNING until TTL. Those paths now return a Lambda error so the Trigger Retry/Catch routes to TriggerRetryExhausted, which releases the lock. The deprecated root copy keeps the old contract; its tests assert it and porting them is a separate batch. --- internal/lambda/orchestrator/evaluate.go | 13 +- .../orchestrator/handler_contract_test.go | 272 ++++++++++++++++++ internal/lambda/orchestrator/trigger.go | 13 +- 3 files changed, 292 insertions(+), 6 deletions(-) create mode 100644 internal/lambda/orchestrator/handler_contract_test.go diff --git a/internal/lambda/orchestrator/evaluate.go b/internal/lambda/orchestrator/evaluate.go index 284cb5e..d4f3ad0 100644 --- a/internal/lambda/orchestrator/evaluate.go +++ b/internal/lambda/orchestrator/evaluate.go @@ -9,20 +9,27 @@ import ( "github.com/dwsmith1983/interlock/pkg/validation" ) +// statusError is the evaluate-mode status emitted when evaluation could not be +// performed (storage failure, missing config). The Step Functions IsReady +// Choice state dereferences $.evaluateResult.status unconditionally, so the +// field must always be populated; "error" routes back into the wait/retry loop +// until the evaluation window closes. +const statusError = "error" + // handleEvaluate fetches config and sensors, evaluates validation rules, and // optionally publishes a VALIDATION_PASSED event. func handleEvaluate(ctx context.Context, d *lambda.Deps, input lambda.OrchestratorInput) (lambda.OrchestratorOutput, error) { cfg, err := d.Store.GetConfig(ctx, input.PipelineID) if err != nil { - return lambda.OrchestratorOutput{Mode: "evaluate", Error: err.Error()}, nil + return lambda.OrchestratorOutput{Mode: "evaluate", Status: statusError, Error: err.Error()}, nil } if cfg == nil { - return lambda.OrchestratorOutput{Mode: "evaluate", Error: fmt.Sprintf("config not found for pipeline %q", input.PipelineID)}, nil + return lambda.OrchestratorOutput{Mode: "evaluate", Status: statusError, Error: fmt.Sprintf("config not found for pipeline %q", input.PipelineID)}, nil } sensors, err := d.Store.GetAllSensors(ctx, input.PipelineID) if err != nil { - return lambda.OrchestratorOutput{Mode: "evaluate", Error: err.Error()}, nil + return lambda.OrchestratorOutput{Mode: "evaluate", Status: statusError, Error: err.Error()}, nil } lambda.RemapPerPeriodSensors(sensors, input.Date) diff --git a/internal/lambda/orchestrator/handler_contract_test.go b/internal/lambda/orchestrator/handler_contract_test.go new file mode 100644 index 0000000..cd88e7d --- /dev/null +++ b/internal/lambda/orchestrator/handler_contract_test.go @@ -0,0 +1,272 @@ +package orchestrator_test + +import ( + "context" + "encoding/json" + "errors" + "log/slog" + "strings" + "testing" + + "github.com/aws/aws-sdk-go-v2/service/dynamodb" + ddbtypes "github.com/aws/aws-sdk-go-v2/service/dynamodb/types" + + "github.com/dwsmith1983/interlock/internal/lambda" + "github.com/dwsmith1983/interlock/internal/lambda/orchestrator" + "github.com/dwsmith1983/interlock/internal/store/storetest" + "github.com/dwsmith1983/interlock/pkg/types" +) + +// configItem builds the control-table row that store.GetConfig expects. +// pipelineID mirrors store.GetConfig(ctx, pipelineID) for readability at +// each call site; every test in this file uses pipeline "p". +// +//nolint:unparam // always called with "p" +func configItem(pipelineID string, cfg types.PipelineConfig) map[string]ddbtypes.AttributeValue { + data, _ := json.Marshal(cfg) + return map[string]ddbtypes.AttributeValue{ + "PK": &ddbtypes.AttributeValueMemberS{Value: types.PipelinePK(pipelineID)}, + "SK": &ddbtypes.AttributeValueMemberS{Value: types.ConfigSK}, + "config": &ddbtypes.AttributeValueMemberS{Value: string(data)}, + } +} + +func testDeps(fake *storetest.FakeDynamo) *lambda.Deps { + return &lambda.Deps{ + Store: storetest.NewStore(fake), + Logger: slog.Default(), + } +} + +// fakeExecutor is a lambda.TriggerExecutor double. +type fakeExecutor struct { + meta map[string]interface{} + err error +} + +func (f *fakeExecutor) Execute(context.Context, *types.TriggerConfig) (map[string]interface{}, error) { + return f.meta, f.err +} + +// TestEvaluate_AlwaysEmitsStatus guards the IsReady Choice state, which reads +// $.evaluateResult.status with no IsPresent guard. An absent status raises +// States.Runtime, which Catch: States.ALL cannot intercept. +func TestEvaluate_AlwaysEmitsStatus(t *testing.T) { + goodCfg := types.PipelineConfig{ + Pipeline: types.PipelineIdentity{ID: "p"}, + Validation: types.ValidationConfig{ + Trigger: "ALL", + Rules: []types.ValidationRule{{Key: "upstream", Check: types.CheckExists}}, + }, + } + + tests := []struct { + name string + fake *storetest.FakeDynamo + wantStatus string + wantErrSub string + }{ + { + name: "GetConfig failure", + fake: &storetest.FakeDynamo{ + GetItemFn: func(context.Context, *dynamodb.GetItemInput) (*dynamodb.GetItemOutput, error) { + return nil, errors.New("dynamodb: internal error") + }, + }, + wantStatus: "error", + wantErrSub: "dynamodb: internal error", + }, + { + name: "config not found", + fake: &storetest.FakeDynamo{}, + wantStatus: "error", + wantErrSub: "config not found", + }, + { + name: "GetAllSensors failure", + fake: &storetest.FakeDynamo{ + GetItemFn: func(context.Context, *dynamodb.GetItemInput) (*dynamodb.GetItemOutput, error) { + return &dynamodb.GetItemOutput{Item: configItem("p", goodCfg)}, nil + }, + QueryFn: func(context.Context, *dynamodb.QueryInput) (*dynamodb.QueryOutput, error) { + return nil, errors.New("dynamodb: request limit exceeded") + }, + }, + wantStatus: "error", + wantErrSub: "request limit exceeded", + }, + { + name: "rules not satisfied", + fake: &storetest.FakeDynamo{ + GetItemFn: func(context.Context, *dynamodb.GetItemInput) (*dynamodb.GetItemOutput, error) { + return &dynamodb.GetItemOutput{Item: configItem("p", goodCfg)}, nil + }, + }, + wantStatus: "not_ready", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + out, err := orchestrator.HandleOrchestrator(context.Background(), testDeps(tt.fake), lambda.OrchestratorInput{ + Mode: "evaluate", PipelineID: "p", ScheduleID: "s", Date: "2026-03-01", + }) + if err != nil { + t.Fatalf("evaluate must not return a Lambda error, got %v", err) + } + if out.Status != tt.wantStatus { + t.Errorf("status = %q, want %q", out.Status, tt.wantStatus) + } + if tt.wantErrSub != "" && !strings.Contains(out.Error, tt.wantErrSub) { + t.Errorf("error = %q, want it to contain %q", out.Error, tt.wantErrSub) + } + + data, mErr := json.Marshal(out) + if mErr != nil { + t.Fatalf("marshal: %v", mErr) + } + var decoded map[string]interface{} + if err := json.Unmarshal(data, &decoded); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if _, ok := decoded["status"]; !ok { + t.Errorf("marshaled output = %s, want a status key for $.evaluateResult.status", data) + } + }) + } +} + +// TestTrigger_FailuresReturnLambdaError guards HasTriggerResult/CheckJob. +// A nil error with a partial triggerResult makes IsPresent true and CheckJob +// dereference a missing runId; returning a Go error routes to +// TriggerRetryExhausted instead, which releases the trigger lock. +func TestTrigger_FailuresReturnLambdaError(t *testing.T) { + badTypeCfg := types.PipelineConfig{ + Pipeline: types.PipelineIdentity{ID: "p"}, + Job: types.JobConfig{ + Type: types.TriggerType("bogus"), + Config: map[string]interface{}{"jobName": "x"}, + }, + } + + tests := []struct { + name string + fake *storetest.FakeDynamo + wantErrSub string + }{ + { + name: "GetConfig failure", + fake: &storetest.FakeDynamo{ + GetItemFn: func(context.Context, *dynamodb.GetItemInput) (*dynamodb.GetItemOutput, error) { + return nil, errors.New("dynamodb: internal error") + }, + }, + wantErrSub: "dynamodb: internal error", + }, + { + name: "config not found", + fake: &storetest.FakeDynamo{}, + wantErrSub: "config not found", + }, + { + name: "unsupported trigger type", + fake: &storetest.FakeDynamo{ + GetItemFn: func(context.Context, *dynamodb.GetItemInput) (*dynamodb.GetItemOutput, error) { + return &dynamodb.GetItemOutput{Item: configItem("p", badTypeCfg)}, nil + }, + }, + wantErrSub: "unsupported trigger type", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + d := testDeps(tt.fake) + d.TriggerRunner = &fakeExecutor{} + out, err := orchestrator.HandleOrchestrator(context.Background(), d, lambda.OrchestratorInput{ + Mode: "trigger", PipelineID: "p", ScheduleID: "s", Date: "2026-03-01", + }) + if err == nil { + t.Fatalf("trigger must return a Lambda error, got output %+v", out) + } + if !strings.Contains(err.Error(), tt.wantErrSub) { + t.Errorf("err = %v, want it to contain %q", err, tt.wantErrSub) + } + if out.Mode != "" || out.RunID != "" || out.Metadata != nil { + t.Errorf("output = %+v, want the zero value so ResultPath is never written", out) + } + }) + } +} + +// TestTrigger_SuccessAlwaysCarriesRunIDAndMetadata guards the CheckJob +// Parameters, which dereference both $.triggerResult.runId and +// $.triggerResult.metadata. +func TestTrigger_SuccessAlwaysCarriesRunIDAndMetadata(t *testing.T) { + glueCfg := types.PipelineConfig{ + Pipeline: types.PipelineIdentity{ID: "p"}, + Job: types.JobConfig{ + Type: types.TriggerGlue, + Config: map[string]interface{}{"jobName": "my-etl"}, + }, + } + httpCfg := types.PipelineConfig{ + Pipeline: types.PipelineIdentity{ID: "p"}, + Job: types.JobConfig{ + Type: types.TriggerHTTP, + Config: map[string]interface{}{"url": "https://example.com/trigger", "method": "POST"}, + }, + } + + tests := []struct { + name string + cfg types.PipelineConfig + meta map[string]interface{} + wantRunID string + }{ + { + name: "polling trigger", + cfg: glueCfg, + meta: map[string]interface{}{"glue_job_name": "my-etl", "glue_job_run_id": "jr_1"}, + wantRunID: "jr_1", + }, + { + name: "sync trigger with nil metadata", + cfg: httpCfg, + meta: nil, + wantRunID: "sync", + }, + { + name: "sync trigger with empty metadata", + cfg: httpCfg, + meta: map[string]interface{}{}, + wantRunID: "sync", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := tt.cfg + fake := &storetest.FakeDynamo{ + GetItemFn: func(context.Context, *dynamodb.GetItemInput) (*dynamodb.GetItemOutput, error) { + return &dynamodb.GetItemOutput{Item: configItem("p", cfg)}, nil + }, + } + d := testDeps(fake) + d.TriggerRunner = &fakeExecutor{meta: tt.meta} + + out, err := orchestrator.HandleOrchestrator(context.Background(), d, lambda.OrchestratorInput{ + Mode: "trigger", PipelineID: "p", ScheduleID: "s", Date: "2026-03-01", + }) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if out.RunID != tt.wantRunID { + t.Errorf("runID = %q, want %q", out.RunID, tt.wantRunID) + } + if len(out.Metadata) == 0 { + t.Errorf("metadata = %v, want a non-empty map for $.triggerResult.metadata", out.Metadata) + } + }) + } +} diff --git a/internal/lambda/orchestrator/trigger.go b/internal/lambda/orchestrator/trigger.go index 4022117..aff6c4b 100644 --- a/internal/lambda/orchestrator/trigger.go +++ b/internal/lambda/orchestrator/trigger.go @@ -13,17 +13,24 @@ import ( // handleTrigger builds a TriggerConfig from the JobConfig, executes it, // publishes JOB_TRIGGERED, and returns the run ID. func handleTrigger(ctx context.Context, d *lambda.Deps, input lambda.OrchestratorInput) (lambda.OrchestratorOutput, error) { + // Configuration failures return a Lambda error rather than a partial + // result: the HasTriggerResult Choice tests IsPresent on $.triggerResult, + // so a partial result would send the execution to CheckJob, which + // dereferences $.triggerResult.runId and raises an uncatchable + // States.Runtime while the trigger lock stays RUNNING. A Lambda error is + // retried by the Trigger state and then caught by TriggerRetryExhausted, + // which releases the lock. cfg, err := d.Store.GetConfig(ctx, input.PipelineID) if err != nil { - return lambda.OrchestratorOutput{Mode: "trigger", Error: err.Error()}, nil + return lambda.OrchestratorOutput{}, fmt.Errorf("trigger get config: %w", err) } if cfg == nil { - return lambda.OrchestratorOutput{Mode: "trigger", Error: fmt.Sprintf("config not found for pipeline %q", input.PipelineID)}, nil + return lambda.OrchestratorOutput{}, fmt.Errorf("trigger: config not found for pipeline %q", input.PipelineID) } triggerCfg, err := BuildTriggerConfig(cfg.Job) if err != nil { - return lambda.OrchestratorOutput{Mode: "trigger", Error: fmt.Sprintf("build trigger config: %v", err)}, nil + return lambda.OrchestratorOutput{}, fmt.Errorf("trigger build config: %w", err) } InjectDateArgs(&triggerCfg, input.Date) From ca36584638b14eeb8f48efcdd301f41b27734c65 Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Fri, 11 Sep 2026 23:10:41 +0700 Subject: [PATCH 05/11] test: check marshal error and simplify configItem helper --- .../orchestrator/handler_contract_test.go | 23 ++++++++++--------- 1 file changed, 12 insertions(+), 11 deletions(-) diff --git a/internal/lambda/orchestrator/handler_contract_test.go b/internal/lambda/orchestrator/handler_contract_test.go index cd88e7d..4a84b45 100644 --- a/internal/lambda/orchestrator/handler_contract_test.go +++ b/internal/lambda/orchestrator/handler_contract_test.go @@ -18,14 +18,15 @@ import ( ) // configItem builds the control-table row that store.GetConfig expects. -// pipelineID mirrors store.GetConfig(ctx, pipelineID) for readability at -// each call site; every test in this file uses pipeline "p". -// -//nolint:unparam // always called with "p" -func configItem(pipelineID string, cfg types.PipelineConfig) map[string]ddbtypes.AttributeValue { - data, _ := json.Marshal(cfg) +// Every test in this file uses pipeline "p". +func configItem(t *testing.T, cfg types.PipelineConfig) map[string]ddbtypes.AttributeValue { + t.Helper() + data, err := json.Marshal(cfg) + if err != nil { + t.Fatalf("marshal config: %v", err) + } return map[string]ddbtypes.AttributeValue{ - "PK": &ddbtypes.AttributeValueMemberS{Value: types.PipelinePK(pipelineID)}, + "PK": &ddbtypes.AttributeValueMemberS{Value: types.PipelinePK("p")}, "SK": &ddbtypes.AttributeValueMemberS{Value: types.ConfigSK}, "config": &ddbtypes.AttributeValueMemberS{Value: string(data)}, } @@ -86,7 +87,7 @@ func TestEvaluate_AlwaysEmitsStatus(t *testing.T) { name: "GetAllSensors failure", fake: &storetest.FakeDynamo{ GetItemFn: func(context.Context, *dynamodb.GetItemInput) (*dynamodb.GetItemOutput, error) { - return &dynamodb.GetItemOutput{Item: configItem("p", goodCfg)}, nil + return &dynamodb.GetItemOutput{Item: configItem(t, goodCfg)}, nil }, QueryFn: func(context.Context, *dynamodb.QueryInput) (*dynamodb.QueryOutput, error) { return nil, errors.New("dynamodb: request limit exceeded") @@ -99,7 +100,7 @@ func TestEvaluate_AlwaysEmitsStatus(t *testing.T) { name: "rules not satisfied", fake: &storetest.FakeDynamo{ GetItemFn: func(context.Context, *dynamodb.GetItemInput) (*dynamodb.GetItemOutput, error) { - return &dynamodb.GetItemOutput{Item: configItem("p", goodCfg)}, nil + return &dynamodb.GetItemOutput{Item: configItem(t, goodCfg)}, nil }, }, wantStatus: "not_ready", @@ -172,7 +173,7 @@ func TestTrigger_FailuresReturnLambdaError(t *testing.T) { name: "unsupported trigger type", fake: &storetest.FakeDynamo{ GetItemFn: func(context.Context, *dynamodb.GetItemInput) (*dynamodb.GetItemOutput, error) { - return &dynamodb.GetItemOutput{Item: configItem("p", badTypeCfg)}, nil + return &dynamodb.GetItemOutput{Item: configItem(t, badTypeCfg)}, nil }, }, wantErrSub: "unsupported trigger type", @@ -249,7 +250,7 @@ func TestTrigger_SuccessAlwaysCarriesRunIDAndMetadata(t *testing.T) { cfg := tt.cfg fake := &storetest.FakeDynamo{ GetItemFn: func(context.Context, *dynamodb.GetItemInput) (*dynamodb.GetItemOutput, error) { - return &dynamodb.GetItemOutput{Item: configItem("p", cfg)}, nil + return &dynamodb.GetItemOutput{Item: configItem(t, cfg)}, nil }, } d := testDeps(fake) From 3d0804ffe24c1d9dd9369094700f3015cb968b5b Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Fri, 11 Sep 2026 23:16:17 +0700 Subject: [PATCH 06/11] fix: always emit the SLA keys the cancel states dereference types.SLAConfig marks every field omitempty and sensorArrivalAt was also omitempty, so CancelSLASchedules and CancelSLAOnCompleteTriggerFailure referenced JSONPaths that were absent for every real config: an absolute-only SLA has no maxDuration, a relative-only SLA has no deadline, and sensorArrivalAt was only set for relative SLAs with a recorded arrival. The resulting States.Runtime is not catchable by Catch: States.ALL, so every SLA-configured execution failed after CompleteTrigger. The SFN input now uses dedicated SFNInput/SFNConfig/SFNSLA types that emit every dereferenced key unconditionally; config.sla itself stays omitempty because two Choice states guard it with IsPresent. Both cancel states now also forward timezone, which handleSLACancel needs when it recomputes deadlines. --- deploy/statemachine.asl.json | 2 + internal/lambda/sfn.go | 89 ++++++++++++++++++---------- internal/lambda/sfn_test.go | 110 +++++++++++++++++++++++++++++++++++ 3 files changed, 171 insertions(+), 30 deletions(-) create mode 100644 internal/lambda/sfn_test.go diff --git a/deploy/statemachine.asl.json b/deploy/statemachine.asl.json index caf0867..45f8a9b 100644 --- a/deploy/statemachine.asl.json +++ b/deploy/statemachine.asl.json @@ -397,6 +397,7 @@ "deadline.$": "$.config.sla.deadline", "expectedDuration.$": "$.config.sla.expectedDuration", "maxDuration.$": "$.config.sla.maxDuration", + "timezone.$": "$.config.sla.timezone", "sensorArrivalAt.$": "$.sensorArrivalAt" }, "ResultPath": "$.slaResult", @@ -451,6 +452,7 @@ "deadline.$": "$.config.sla.deadline", "expectedDuration.$": "$.config.sla.expectedDuration", "maxDuration.$": "$.config.sla.maxDuration", + "timezone.$": "$.config.sla.timezone", "sensorArrivalAt.$": "$.sensorArrivalAt" }, "ResultPath": "$.slaResult", diff --git a/internal/lambda/sfn.go b/internal/lambda/sfn.go index 4901f4b..91dd837 100644 --- a/internal/lambda/sfn.go +++ b/internal/lambda/sfn.go @@ -10,28 +10,47 @@ import ( "github.com/dwsmith1983/interlock/pkg/types" ) -// sfnInput is the top-level input for the Step Function state machine. -// It includes pipeline identity fields and a config block used by Wait states. -type sfnInput struct { +// SFNInput is the top-level input for the Step Function state machine. +// +// The state machine runs in JSONPath mode: a Parameters reference to a missing +// path raises States.Runtime, which is neither retriable nor catchable by +// Catch: ["States.ALL"]. Every field the ASL dereferences is therefore emitted +// unconditionally (no omitempty), even when empty. Only Config.SLA keeps +// omitempty, because the CheckCancelSLA and CheckSLAForCompleteTriggerFailure +// Choice states use IsPresent on $.config.sla to decide whether to run the +// SLA branch at all. +type SFNInput struct { PipelineID string `json:"pipelineId"` ScheduleID string `json:"scheduleId"` Date string `json:"date"` - SensorArrivalAt string `json:"sensorArrivalAt,omitempty"` // RFC3339; first sensor arrival for relative SLA - Config sfnConfig `json:"config"` + SensorArrivalAt string `json:"sensorArrivalAt"` // RFC3339; empty when unknown + Config SFNConfig `json:"config"` } -// sfnConfig holds timing parameters for the SFN evaluation loop and SLA branch. -type sfnConfig struct { - EvaluationIntervalSeconds int `json:"evaluationIntervalSeconds"` - EvaluationWindowSeconds int `json:"evaluationWindowSeconds"` - JobCheckIntervalSeconds int `json:"jobCheckIntervalSeconds"` - JobPollWindowSeconds int `json:"jobPollWindowSeconds"` - SLA *types.SLAConfig `json:"sla,omitempty"` +// SFNConfig holds timing parameters for the SFN evaluation loop and SLA branch. +type SFNConfig struct { + EvaluationIntervalSeconds int `json:"evaluationIntervalSeconds"` + EvaluationWindowSeconds int `json:"evaluationWindowSeconds"` + JobCheckIntervalSeconds int `json:"jobCheckIntervalSeconds"` + JobPollWindowSeconds int `json:"jobPollWindowSeconds"` + SLA *SFNSLA `json:"sla,omitempty"` +} + +// SFNSLA mirrors types.SLAConfig without omitempty. types.SLAConfig omits every +// empty field, which is what made CancelSLASchedules and +// CancelSLAOnCompleteTriggerFailure fail with States.Runtime for every +// SLA-configured execution. +type SFNSLA struct { + Deadline string `json:"deadline"` + ExpectedDuration string `json:"expectedDuration"` + MaxDuration string `json:"maxDuration"` + Timezone string `json:"timezone"` + Critical bool `json:"critical"` } // BuildSFNConfig converts a PipelineConfig into the config block for the SFN input. -func BuildSFNConfig(cfg *types.PipelineConfig) sfnConfig { - sc := sfnConfig{ +func BuildSFNConfig(cfg *types.PipelineConfig) SFNConfig { + sc := SFNConfig{ EvaluationIntervalSeconds: DefaultEvalIntervalSec, EvaluationWindowSeconds: DefaultEvalWindowSec, JobCheckIntervalSeconds: DefaultJobCheckIntervalSec, @@ -50,16 +69,33 @@ func BuildSFNConfig(cfg *types.PipelineConfig) sfnConfig { } if cfg.SLA != nil { - sla := *cfg.SLA - if sla.Timezone == "" { - sla.Timezone = "UTC" + tz := cfg.SLA.Timezone + if tz == "" { + tz = "UTC" + } + sc.SLA = &SFNSLA{ + Deadline: cfg.SLA.Deadline, + ExpectedDuration: cfg.SLA.ExpectedDuration, + MaxDuration: cfg.SLA.MaxDuration, + Timezone: tz, + Critical: cfg.SLA.Critical, } - sc.SLA = &sla } return sc } +// BuildSFNInput assembles the full Step Functions execution input. +func BuildSFNInput(cfg *types.PipelineConfig, pipelineID, scheduleID, date, sensorArrivalAt string) SFNInput { + return SFNInput{ + PipelineID: pipelineID, + ScheduleID: scheduleID, + Date: date, + SensorArrivalAt: sensorArrivalAt, + Config: BuildSFNConfig(cfg), + } +} + // TruncateExecName ensures an SFN execution name does not exceed the 80-character // AWS limit. When truncation is needed the suffix (date + timestamp) is preserved // by trimming characters from the beginning of the name. @@ -87,30 +123,23 @@ func StartSFNWithName(ctx context.Context, d *Deps, cfg *types.PipelineConfig, p return nil } - sc := BuildSFNConfig(cfg) + input := BuildSFNInput(cfg, pipelineID, scheduleID, date, "") // Warn if the sum of evaluation + poll windows exceeds the SFN timeout. - totalWindowSec := sc.EvaluationWindowSeconds + sc.JobPollWindowSeconds + totalWindowSec := input.Config.EvaluationWindowSeconds + input.Config.JobPollWindowSeconds sfnTimeout := ResolveTriggerLockTTL() - TriggerLockBuffer // strip the buffer to get raw SFN timeout if sfnTimeout > 0 && time.Duration(totalWindowSec)*time.Second > sfnTimeout { d.Logger.Warn("combined pipeline windows exceed SFN timeout", "pipelineId", pipelineID, - "evalWindowSec", sc.EvaluationWindowSeconds, - "jobPollWindowSec", sc.JobPollWindowSeconds, + "evalWindowSec", input.Config.EvaluationWindowSeconds, + "jobPollWindowSec", input.Config.JobPollWindowSeconds, "totalWindowSec", totalWindowSec, "sfnTimeoutSec", int(sfnTimeout.Seconds()), ) } - input := sfnInput{ - PipelineID: pipelineID, - ScheduleID: scheduleID, - Date: date, - Config: sc, - } - // Populate sensorArrivalAt for relative SLA passthrough. - if sc.SLA != nil && sc.SLA.MaxDuration != "" && d.Store != nil { + if input.Config.SLA != nil && input.Config.SLA.MaxDuration != "" && d.Store != nil { arrivalKey := "first-sensor-arrival#" + date arrivalData, readErr := d.Store.GetSensorData(ctx, pipelineID, arrivalKey) if readErr != nil { diff --git a/internal/lambda/sfn_test.go b/internal/lambda/sfn_test.go new file mode 100644 index 0000000..767d3f6 --- /dev/null +++ b/internal/lambda/sfn_test.go @@ -0,0 +1,110 @@ +package lambda_test + +import ( + "encoding/json" + "testing" + + "github.com/dwsmith1983/interlock/internal/lambda" + "github.com/dwsmith1983/interlock/pkg/types" +) + +// aslDereferencedTopLevelKeys and aslDereferencedSLAKeys mirror the JSONPath +// references in deploy/statemachine.asl.json. A missing key raises +// States.Runtime, which Catch: States.ALL cannot intercept. +var ( + aslDereferencedTopLevelKeys = []string{"pipelineId", "scheduleId", "date", "sensorArrivalAt", "config"} + aslDereferencedConfigKeys = []string{ + "evaluationIntervalSeconds", "evaluationWindowSeconds", + "jobCheckIntervalSeconds", "jobPollWindowSeconds", + } + aslDereferencedSLAKeys = []string{"deadline", "expectedDuration", "maxDuration", "timezone"} +) + +func baseConfig() *types.PipelineConfig { + return &types.PipelineConfig{ + Pipeline: types.PipelineIdentity{ID: "gold-orders"}, + Schedule: types.ScheduleConfig{ + Evaluation: types.EvaluationWindow{Window: "1h", Interval: "5m"}, + }, + Job: types.JobConfig{Type: types.TriggerGlue, Config: map[string]interface{}{"jobName": "etl"}}, + } +} + +func TestBuildSFNInput_AlwaysEmitsDereferencedKeys(t *testing.T) { + tests := []struct { + name string + sla *types.SLAConfig + sensorArrivalAt string + wantSLA bool + }{ + { + name: "absolute SLA only", + sla: &types.SLAConfig{Deadline: "08:00", ExpectedDuration: "30m"}, + wantSLA: true, + }, + { + name: "relative SLA with sensor arrival", + sla: &types.SLAConfig{MaxDuration: "2h"}, + sensorArrivalAt: "2026-03-01T06:00:00Z", + wantSLA: true, + }, + { + name: "relative SLA without sensor arrival", + sla: &types.SLAConfig{MaxDuration: "2h"}, + wantSLA: true, + }, + { + name: "no SLA", + sla: nil, + wantSLA: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := baseConfig() + cfg.SLA = tt.sla + + data, err := json.Marshal(lambda.BuildSFNInput(cfg, "gold-orders", "daily", "2026-03-01", tt.sensorArrivalAt)) + if err != nil { + t.Fatalf("marshal: %v", err) + } + var doc map[string]interface{} + if err := json.Unmarshal(data, &doc); err != nil { + t.Fatalf("unmarshal: %v", err) + } + + for _, key := range aslDereferencedTopLevelKeys { + if _, ok := doc[key]; !ok { + t.Errorf("SFN input is missing %q: %s", key, data) + } + } + + cfgMap, ok := doc["config"].(map[string]interface{}) + if !ok { + t.Fatalf("config is not an object: %s", data) + } + for _, key := range aslDereferencedConfigKeys { + if _, ok := cfgMap[key]; !ok { + t.Errorf("SFN config is missing %q: %s", key, data) + } + } + + slaMap, hasSLA := cfgMap["sla"].(map[string]interface{}) + if hasSLA != tt.wantSLA { + t.Fatalf("config.sla present = %v, want %v (CheckCancelSLA uses IsPresent): %s", hasSLA, tt.wantSLA, data) + } + if !tt.wantSLA { + return + } + for _, key := range aslDereferencedSLAKeys { + if _, ok := slaMap[key]; !ok { + t.Errorf("SFN config.sla is missing %q: %s", key, data) + } + } + if slaMap["timezone"] != "UTC" { + t.Errorf("config.sla.timezone = %v, want UTC default", slaMap["timezone"]) + } + }) + } +} From 75c7cd47df352b3e28c437bd1c4de30731fb80f8 Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Fri, 11 Sep 2026 23:23:09 +0700 Subject: [PATCH 07/11] test: assert every ASL reference path resolves against real Go payloads Collects every Parameters reference (including intrinsic arguments), Wait SecondsPath, Choice comparator path and unguarded Choice Variable from the rendered state machine and resolves each one against the document Step Functions actually holds: the marshaled SFNInput from the real builder plus the marshaled outputs of the real orchestrator handlers. Paths behind an IsPresent guard may be absent; Parameters paths may not. Verified red against the pre-fix sfn.go and evaluate.go. --- deploy/statemachine_contract_test.go | 362 +++++++++++++++++++++++++++ 1 file changed, 362 insertions(+) create mode 100644 deploy/statemachine_contract_test.go diff --git a/deploy/statemachine_contract_test.go b/deploy/statemachine_contract_test.go new file mode 100644 index 0000000..d78eafb --- /dev/null +++ b/deploy/statemachine_contract_test.go @@ -0,0 +1,362 @@ +package deploy_test + +import ( + "context" + "encoding/json" + "errors" + "log/slog" + "regexp" + "strconv" + "strings" + "testing" + + "github.com/aws/aws-sdk-go-v2/service/dynamodb" + ddbtypes "github.com/aws/aws-sdk-go-v2/service/dynamodb/types" + "github.com/stretchr/testify/require" + + "github.com/dwsmith1983/interlock/internal/lambda" + "github.com/dwsmith1983/interlock/internal/lambda/orchestrator" + "github.com/dwsmith1983/interlock/internal/store/storetest" + "github.com/dwsmith1983/interlock/pkg/types" +) + +// --- JSONPath resolution --------------------------------------------------- + +// pathRefRE matches an ASL reference path such as $.a.b[0].c. Used to pull the +// arguments out of intrinsic calls like States.MathAdd($.a, $.b). +var pathRefRE = regexp.MustCompile(`\$(?:\.[A-Za-z0-9_]+(?:\[\d+\])*)+`) + +// resolvePath walks a reference path against a decoded JSON document and +// reports whether it exists. Supports "$", "$.a.b" and "$.a[0].b". +func resolvePath(doc interface{}, path string) bool { + if path == "$" { + return true + } + if !strings.HasPrefix(path, "$.") { + return false + } + cur := doc + for _, seg := range strings.Split(strings.TrimPrefix(path, "$."), ".") { + name, indices := splitSegment(seg) + if name != "" { + m, ok := cur.(map[string]interface{}) + if !ok { + return false + } + v, ok := m[name] + if !ok { + return false + } + cur = v + } + for _, idx := range indices { + arr, ok := cur.([]interface{}) + if !ok || idx < 0 || idx >= len(arr) { + return false + } + cur = arr[idx] + } + } + return true +} + +// splitSegment splits "items[0][1]" into ("items", []int{0, 1}). +func splitSegment(seg string) (name string, indices []int) { + name = seg + if i := strings.Index(seg, "["); i >= 0 { + name = seg[:i] + for _, raw := range strings.Split(strings.Trim(seg[i:], "[]"), "][") { + n, err := strconv.Atoi(raw) + if err != nil { + continue + } + indices = append(indices, n) + } + } + return name, indices +} + +// --- ASL reference collection ---------------------------------------------- + +// aslState is the subset of a state definition that carries reference paths. +type aslState struct { + Type string `json:"Type"` + Parameters map[string]interface{} `json:"Parameters"` + SecondsPath string `json:"SecondsPath"` + Choices []map[string]interface{} `json:"Choices"` +} + +// requiredPaths returns every reference path the state dereferences +// unconditionally. A missing path raises States.Runtime at runtime. +func requiredPaths(st aslState) []string { + out := collectParameterPaths(st.Parameters) + if strings.HasPrefix(st.SecondsPath, "$") { + out = append(out, st.SecondsPath) + } + for _, rule := range st.Choices { + out = append(out, requiredChoicePaths(rule)...) + } + return out +} + +// collectParameterPaths walks a Parameters object and returns every reference +// used by a ".$" entry, including paths inside intrinsic calls. +func collectParameterPaths(params map[string]interface{}) []string { + var out []string + for k, v := range params { + if !strings.HasSuffix(k, ".$") { + if nested, ok := v.(map[string]interface{}); ok { + out = append(out, collectParameterPaths(nested)...) + } + continue + } + s, ok := v.(string) + if !ok { + continue + } + if strings.HasPrefix(s, "States.") { + out = append(out, pathRefRE.FindAllString(s, -1)...) + continue + } + if strings.HasPrefix(s, "$") { + out = append(out, s) + } + } + return out +} + +// requiredChoicePaths returns the reference paths a Choice rule dereferences +// unconditionally. A Variable in a rule that also carries IsPresent is +// explicitly allowed to be absent — that is what IsPresent is for. +func requiredChoicePaths(rule map[string]interface{}) []string { + if _, guarded := rule["IsPresent"]; guarded { + return nil + } + var out []string + for k, v := range rule { + switch k { + case "Variable": + if s, ok := v.(string); ok && strings.HasPrefix(s, "$") { + out = append(out, s) + } + case "Not": + if m, ok := v.(map[string]interface{}); ok { + out = append(out, requiredChoicePaths(m)...) + } + case "And", "Or": + if arr, ok := v.([]interface{}); ok { + for _, e := range arr { + if m, ok := e.(map[string]interface{}); ok { + out = append(out, requiredChoicePaths(m)...) + } + } + } + default: + // Comparator paths such as NumericGreaterThanEqualsPath. + if strings.HasSuffix(k, "Path") { + if s, ok := v.(string); ok && strings.HasPrefix(s, "$") { + out = append(out, s) + } + } + } + } + return out +} + +// --- real handler outputs --------------------------------------------------- + +// contractConfigItem builds the control-table row that store.GetConfig +// expects. Every test in this file uses pipeline "gold-orders". +func contractConfigItem(t *testing.T, cfg types.PipelineConfig) map[string]ddbtypes.AttributeValue { + t.Helper() + data, err := json.Marshal(cfg) + require.NoError(t, err) + return map[string]ddbtypes.AttributeValue{ + "PK": &ddbtypes.AttributeValueMemberS{Value: types.PipelinePK("gold-orders")}, + "SK": &ddbtypes.AttributeValueMemberS{Value: types.ConfigSK}, + "config": &ddbtypes.AttributeValueMemberS{Value: string(data)}, + } +} + +type contractExecutor struct{ meta map[string]interface{} } + +func (c *contractExecutor) Execute(context.Context, *types.TriggerConfig) (map[string]interface{}, error) { + return c.meta, nil +} + +// evaluateErrorResult returns the output the live evaluate handler produces +// when the control table is unavailable — the worst case for IsReady. +func evaluateErrorResult(t *testing.T) lambda.OrchestratorOutput { + t.Helper() + fake := &storetest.FakeDynamo{ + GetItemFn: func(context.Context, *dynamodb.GetItemInput) (*dynamodb.GetItemOutput, error) { + return nil, errors.New("dynamodb: internal error") + }, + } + d := &lambda.Deps{Store: storetest.NewStore(fake), Logger: slog.Default()} + out, err := orchestrator.HandleOrchestrator(context.Background(), d, lambda.OrchestratorInput{ + Mode: "evaluate", PipelineID: "gold-orders", ScheduleID: "daily", Date: "2026-03-01", + }) + require.NoError(t, err, "evaluate must not return a Lambda error for a storage failure") + return out +} + +// triggerResult returns the output the live trigger handler produces for a +// polling (Glue) trigger — the state CheckJob reads. +func triggerResult(t *testing.T) lambda.OrchestratorOutput { + t.Helper() + cfg := types.PipelineConfig{ + Pipeline: types.PipelineIdentity{ID: "gold-orders"}, + Job: types.JobConfig{Type: types.TriggerGlue, Config: map[string]interface{}{"jobName": "etl"}}, + } + fake := &storetest.FakeDynamo{ + GetItemFn: func(context.Context, *dynamodb.GetItemInput) (*dynamodb.GetItemOutput, error) { + return &dynamodb.GetItemOutput{Item: contractConfigItem(t, cfg)}, nil + }, + } + d := &lambda.Deps{Store: storetest.NewStore(fake), Logger: slog.Default()} + d.TriggerRunner = &contractExecutor{meta: map[string]interface{}{ + "glue_job_name": "etl", "glue_job_run_id": "jr_1", + }} + out, err := orchestrator.HandleOrchestrator(context.Background(), d, lambda.OrchestratorInput{ + Mode: "trigger", PipelineID: "gold-orders", ScheduleID: "daily", Date: "2026-03-01", + }) + require.NoError(t, err) + return out +} + +// checkJobTerminalResult returns the terminal check-job output. CompleteTrigger +// is only reachable once IsJobDone has seen a terminal event, so this is the +// document CompleteTrigger's Parameters are resolved against. +func checkJobTerminalResult(t *testing.T) lambda.OrchestratorOutput { + t.Helper() + fake := &storetest.FakeDynamo{ + QueryFn: func(context.Context, *dynamodb.QueryInput) (*dynamodb.QueryOutput, error) { + return &dynamodb.QueryOutput{Items: []map[string]ddbtypes.AttributeValue{{ + "PK": &ddbtypes.AttributeValueMemberS{Value: types.PipelinePK("gold-orders")}, + "SK": &ddbtypes.AttributeValueMemberS{Value: types.JobSK("daily", "2026-03-01", "1709280000000")}, + "event": &ddbtypes.AttributeValueMemberS{Value: types.JobEventSuccess}, + }}}, nil + }, + } + d := &lambda.Deps{Store: storetest.NewStore(fake), Logger: slog.Default()} + out, err := orchestrator.HandleOrchestrator(context.Background(), d, lambda.OrchestratorInput{ + Mode: "check-job", PipelineID: "gold-orders", ScheduleID: "daily", Date: "2026-03-01", + }) + require.NoError(t, err) + return out +} + +// stateDocument builds the JSON document Step Functions holds when each state +// runs: the execution input plus every ResultPath written earlier on the path +// that reaches the state. evalLoop/jobPollLoop mirror the Result blocks of the +// InitEvalLoop and InitJobPollLoop Pass states; errorInfo mirrors the object +// Step Functions writes for a Catch. +func stateDocument(t *testing.T, input lambda.SFNInput) map[string]interface{} { + t.Helper() + data, err := json.Marshal(input) + require.NoError(t, err) + + var doc map[string]interface{} + require.NoError(t, json.Unmarshal(data, &doc)) + + doc["evalLoop"] = map[string]interface{}{"elapsedSeconds": float64(0)} + doc["jobPollLoop"] = map[string]interface{}{"elapsedSeconds": float64(0)} + doc["evaluateResult"] = toDoc(t, evaluateErrorResult(t)) + doc["triggerResult"] = toDoc(t, triggerResult(t)) + doc["checkJobResult"] = toDoc(t, checkJobTerminalResult(t)) + doc["errorInfo"] = map[string]interface{}{ + "Error": "States.TaskFailed", + "Cause": "trigger execute: glue trigger: StartJobRun failed", + } + return doc +} + +func toDoc(t *testing.T, v interface{}) map[string]interface{} { + t.Helper() + data, err := json.Marshal(v) + require.NoError(t, err) + var m map[string]interface{} + require.NoError(t, json.Unmarshal(data, &m)) + return m +} + +// --- the contract ----------------------------------------------------------- + +// slaOnlyStates are reachable only when the CheckCancelSLA / +// CheckSLAForCompleteTriggerFailure IsPresent guard on $.config.sla passes. +var slaOnlyStates = map[string]bool{ + "CancelSLASchedules": true, + "CancelSLAOnCompleteTriggerFailure": true, +} + +func contractPipelineConfig(sla *types.SLAConfig) *types.PipelineConfig { + return &types.PipelineConfig{ + Pipeline: types.PipelineIdentity{ID: "gold-orders"}, + Schedule: types.ScheduleConfig{Evaluation: types.EvaluationWindow{Window: "1h", Interval: "5m"}}, + SLA: sla, + Job: types.JobConfig{Type: types.TriggerGlue, Config: map[string]interface{}{"jobName": "etl"}}, + } +} + +// TestASL_EveryDereferencedPathResolves is the Go<->ASL contract guard. Every +// Parameters reference, Wait SecondsPath, Choice comparator path and +// unguarded Choice Variable in the rendered state machine must resolve against +// the real marshaled Go payloads. A path that does not resolve raises +// States.Runtime at runtime, which Retry cannot retry and Catch: States.ALL +// cannot intercept. +func TestASL_EveryDereferencedPathResolves(t *testing.T) { + asl := loadASL(t) + + scenarios := []struct { + name string + sla *types.SLAConfig + sensorArrivalAt string + }{ + { + name: "absolute SLA only", + sla: &types.SLAConfig{Deadline: "08:00", ExpectedDuration: "30m"}, + }, + { + name: "relative SLA with sensor arrival", + sla: &types.SLAConfig{MaxDuration: "2h"}, + sensorArrivalAt: "2026-03-01T06:00:00Z", + }, + { + name: "relative SLA without sensor arrival", + sla: &types.SLAConfig{MaxDuration: "2h"}, + }, + { + name: "no SLA", + sla: nil, + }, + } + + for _, sc := range scenarios { + t.Run(sc.name, func(t *testing.T) { + input := lambda.BuildSFNInput( + contractPipelineConfig(sc.sla), "gold-orders", "daily", "2026-03-01", sc.sensorArrivalAt) + doc := stateDocument(t, input) + + // The SLA branch is selected by IsPresent on $.config.sla. + slaPresent := resolvePath(doc, "$.config.sla") + require.Equal(t, sc.sla != nil, slaPresent, + "$.config.sla presence must match whether the pipeline has an SLA") + + for name, raw := range asl.States { + if sc.sla == nil && slaOnlyStates[name] { + continue // unreachable without an SLA + } + var st aslState + require.NoError(t, json.Unmarshal(raw, &st), "parsing state %q", name) + + for _, path := range requiredPaths(st) { + ok := resolvePath(doc, path) + require.Truef(t, ok, + "state %q dereferences %q, which is absent from the Step Functions state document; "+ + "this raises States.Runtime and cannot be caught", name, path) + } + } + }) + } +} From 5e8eb16f76d0167a0285b3fbbd6452e4db9540b5 Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Fri, 11 Sep 2026 23:25:55 +0700 Subject: [PATCH 08/11] docs: record state machine contract fixes in the changelog --- CHANGELOG.md | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 23bf10b..b0c2459 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,6 +34,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Stream router correlation IDs** — Per-record correlation IDs injected into context for structured log tracing across services. - **Telemetry flush per invocation** — OTel providers flush (not shutdown) per Lambda invocation to survive environment reuse across warm starts. +### Fixed + +- **SFN SLA cancel states no longer fail with `States.Runtime`** — `CancelSLASchedules` and `CancelSLAOnCompleteTriggerFailure` referenced `$.config.sla.deadline`, `$.config.sla.expectedDuration`, `$.config.sla.maxDuration` and `$.sensorArrivalAt`, but `types.SLAConfig` marks every field `omitempty`, so an absolute-only SLA never emitted `maxDuration`, a relative-only SLA never emitted `deadline`, and `sensorArrivalAt` was usually absent. The resulting `States.Runtime` is not retriable and is not caught by `Catch: ["States.ALL"]`, so every SLA-configured execution failed after `CompleteTrigger`. The SFN input now uses dedicated `SFNInput`/`SFNConfig`/`SFNSLA` types that always emit those keys, and both cancel states now forward `timezone` so `handleSLACancel` recomputes deadlines in the configured zone. +- **Trigger run IDs are extracted for every trigger type** — `ExtractRunID` only matched `runId`, `jobRunId`, `glue_job_run_id`, `executionArn`, `stepId` and `dagRunId`, while the step-function, EMR, EMR Serverless, Airflow and Databricks executors emit `sfn_execution_arn`, `emr_step_id`, `emr_sl_job_run_id`, `airflow_dag_run_id` and `databricks_run_id`. The empty run ID was then omitted from the Lambda result and the `CheckJob` state raised `States.Runtime` after the external job had already been launched. `OrchestratorOutput.RunID` is now always marshaled and an empty metadata map takes the sync-sentinel path. +- **Absolute SLA deadlines are anchored to the execution date** — `CalculateAbsoluteDeadline` rolled an explicit execution date forward by 24h (or 1h for `:MM` deadlines) whenever the deadline had already passed. `sla-monitor` `cancel` therefore published `SLA_MET` for runs that finished late, the `reconcile` breach branch was unreachable, and the watchdog scheduled breach alerts a day late. Roll-forward now applies only when no execution date is supplied, and uses `AddDate` so the wall-clock time survives DST transitions. +- **Orchestrator evaluate/trigger results satisfy the state machine contract** — `handleEvaluate` returned a result without `status` on storage failures, so the `IsReady` Choice dereferenced a missing `$.evaluateResult.status`; it now always emits a status. `handleTrigger` returned a partial result with a nil error on configuration failures, so `HasTriggerResult` saw `IsPresent=true` and `CheckJob` dereferenced a missing `runId`, killing the execution with the `TRIGGER#` lock stuck in `RUNNING` until TTL; those paths now return a Lambda error so `Trigger`'s Retry/Catch routes to `TriggerRetryExhausted`, which releases the lock. +- **Time-dependent tests** — `TestSLAMonitor_Calculate_ReturnsRFC3339` (failing since 2026-06-16), `TestSLAMonitor_Reconcile_ReturnsDeadlines`, `TestSLAMonitor_Cancel_RecalculatesWhenTimesNotProvided` and three watchdog proactive-SLA tests now inject `Deps.NowFunc` instead of relying on the wall clock. + ### Dependencies - `go.opentelemetry.io/otel` v1.43.0 (traces + metrics) From aadcb6b0420f1d4d2d7f39ac6b65dfdf1876e4b2 Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Fri, 11 Sep 2026 23:45:04 +0700 Subject: [PATCH 09/11] fix: apply the T+1 SLA date on the cancel and dry-run paths Sensor-triggered daily pipelines run T+1 (data for date D arrives on D+1), a rule the watchdog already applied when proactively scheduling SLA alerts. handleSLACancel and the dry-run SLA projection instead recomputed the deadline from the data date D. Now that an explicit date no longer rolls forward past a missed deadline, this made cancel publish a false SLA_BREACH for pipelines that finished on D+1 before their deadline. Extract the watchdog's rule into lambda.ResolveSLADate and share it across the watchdog, sla.handleSLACancel, and stream.publishDryRunSLAProjection. Only the recalculation input's date is shifted; the schedule name, GetTrigger lookup, and published event still use the caller's original date. Also guard against a fabricated SLA_MET verdict: if neither an absolute nor a relative deadline can be determined (e.g. a relative SLA with no recorded sensor arrival), cancel now logs and skips publishing a verdict instead of defaulting to SLA_MET. --- internal/lambda/schedule.go | 23 ++++ internal/lambda/sla/cancel.go | 23 +++- internal/lambda/sla/cancel_test.go | 213 +++++++++++++++++++++++++++++ internal/lambda/sla_date_test.go | 47 +++++++ internal/lambda/stream/dryrun.go | 2 +- internal/lambda/watchdog/sla.go | 8 +- 6 files changed, 306 insertions(+), 10 deletions(-) create mode 100644 internal/lambda/sla/cancel_test.go create mode 100644 internal/lambda/sla_date_test.go diff --git a/internal/lambda/schedule.go b/internal/lambda/schedule.go index ff2ec42..e388a54 100644 --- a/internal/lambda/schedule.go +++ b/internal/lambda/schedule.go @@ -3,6 +3,7 @@ package lambda import ( "os" "strconv" + "strings" "time" "github.com/dwsmith1983/interlock/pkg/types" @@ -17,6 +18,28 @@ func ResolveScheduleID(cfg *types.PipelineConfig) string { return "stream" } +// ResolveSLADate returns the calendar date an absolute "HH:MM" SLA deadline +// applies to. Sensor-triggered daily pipelines (no cron) run T+1: data for +// date D arrives on D+1, so the deadline is on D+1. Cron pipelines and hourly +// ":MM" deadlines keep their own date. Composite hourly dates +// ("2006-01-02T15") and unparseable dates are returned unchanged. +func ResolveSLADate(cfg *types.PipelineConfig, date string) string { + if cfg == nil || cfg.SLA == nil { + return date + } + if cfg.Schedule.Cron != "" { + return date + } + if strings.HasPrefix(cfg.SLA.Deadline, ":") { + return date + } + t, err := time.Parse("2006-01-02", date) + if err != nil { + return date + } + return t.AddDate(0, 0, 1).Format("2006-01-02") +} + // ResolveTriggerLockTTL returns the trigger lock TTL based on the // SFN_TIMEOUT_SECONDS env var plus a 30-minute buffer. Defaults to // 4h30m if the env var is not set or invalid. diff --git a/internal/lambda/sla/cancel.go b/internal/lambda/sla/cancel.go index 0cfe9ea..4f8b0a7 100644 --- a/internal/lambda/sla/cancel.go +++ b/internal/lambda/sla/cancel.go @@ -25,7 +25,21 @@ func handleSLACancel(ctx context.Context, d *lambda.Deps, input lambda.SLAMonito input.WarningAt = calc.WarningAt input.BreachAt = calc.BreachAt } else if input.Deadline != "" { - calc, err := handleSLACalculate(input, d.Now()) + calcInput := input + if d.Store != nil { + cfg, cfgErr := d.Store.GetConfig(ctx, input.PipelineID) + switch { + case cfgErr != nil: + d.Logger.WarnContext(ctx, "config lookup failed in cancel, using unshifted SLA date", + "pipeline", input.PipelineID, "error", cfgErr) + case cfg == nil: + d.Logger.WarnContext(ctx, "config not found in cancel, using unshifted SLA date", + "pipeline", input.PipelineID) + default: + calcInput.Date = lambda.ResolveSLADate(cfg, input.Date) + } + } + calc, err := handleSLACalculate(calcInput, d.Now()) if err != nil { return lambda.SLAMonitorOutput{}, fmt.Errorf("cancel recalculate: %w", err) } @@ -63,7 +77,12 @@ func handleSLACancel(ctx context.Context, d *lambda.Deps, input lambda.SLAMonito } publish := true - if d.Store != nil { + if input.WarningAt == "" && input.BreachAt == "" { + d.Logger.WarnContext(ctx, "no SLA deadline could be determined, skipping verdict", + "pipeline", input.PipelineID, "date", input.Date) + publish = false + } + if publish && d.Store != nil { tr, err := d.Store.GetTrigger(ctx, input.PipelineID, input.ScheduleID, input.Date) if err != nil { d.Logger.WarnContext(ctx, "trigger lookup failed in cancel, proceeding with verdict", diff --git a/internal/lambda/sla/cancel_test.go b/internal/lambda/sla/cancel_test.go new file mode 100644 index 0000000..64d72d3 --- /dev/null +++ b/internal/lambda/sla/cancel_test.go @@ -0,0 +1,213 @@ +package sla_test + +import ( + "context" + "encoding/json" + "errors" + "log/slog" + "strings" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/service/dynamodb" + ddbtypes "github.com/aws/aws-sdk-go-v2/service/dynamodb/types" + "github.com/aws/aws-sdk-go-v2/service/eventbridge" + + "github.com/dwsmith1983/interlock/internal/lambda" + "github.com/dwsmith1983/interlock/internal/lambda/sla" + "github.com/dwsmith1983/interlock/internal/store/storetest" + "github.com/dwsmith1983/interlock/pkg/types" +) + +// countingEventBridge is a lambda.EventBridgeAPI double that records how many +// times PutEvents was called, so a test can prove no verdict was published. +type countingEventBridge struct { + calls int +} + +func (c *countingEventBridge) PutEvents(context.Context, *eventbridge.PutEventsInput, ...func(*eventbridge.Options)) (*eventbridge.PutEventsOutput, error) { + c.calls++ + return &eventbridge.PutEventsOutput{}, nil +} + +// configItem builds the control-table CONFIG row that store.GetConfig expects. +func configItem(t *testing.T, cfg types.PipelineConfig) map[string]ddbtypes.AttributeValue { + t.Helper() + data, err := json.Marshal(cfg) + if err != nil { + t.Fatalf("marshal config: %v", err) + } + return map[string]ddbtypes.AttributeValue{ + "PK": &ddbtypes.AttributeValueMemberS{Value: types.PipelinePK("p")}, + "SK": &ddbtypes.AttributeValueMemberS{Value: types.ConfigSK}, + "config": &ddbtypes.AttributeValueMemberS{Value: string(data)}, + } +} + +// triggerItem builds a minimal TRIGGER row so store.GetTrigger returns non-nil. +func triggerItem() map[string]ddbtypes.AttributeValue { + return map[string]ddbtypes.AttributeValue{ + "PK": &ddbtypes.AttributeValueMemberS{Value: types.PipelinePK("p")}, + "SK": &ddbtypes.AttributeValueMemberS{Value: types.TriggerSK("stream", "2026-03-10")}, + "status": &ddbtypes.AttributeValueMemberS{Value: types.TriggerStatusRunning}, + } +} + +// newFakeDynamo returns a FakeDynamo that serves cfg for CONFIG-key reads +// (or cfgErr, if set) and a minimal TRIGGER row for every TRIGGER-key read. +func newFakeDynamo(t *testing.T, cfg types.PipelineConfig, cfgErr error) *storetest.FakeDynamo { + t.Helper() + return &storetest.FakeDynamo{ + GetItemFn: func(_ context.Context, in *dynamodb.GetItemInput) (*dynamodb.GetItemOutput, error) { + skAttr, ok := in.Key["SK"].(*ddbtypes.AttributeValueMemberS) + if !ok { + return &dynamodb.GetItemOutput{}, nil + } + switch { + case skAttr.Value == types.ConfigSK: + if cfgErr != nil { + return nil, cfgErr + } + return &dynamodb.GetItemOutput{Item: configItem(t, cfg)}, nil + case strings.HasPrefix(skAttr.Value, "TRIGGER#"): + return &dynamodb.GetItemOutput{Item: triggerItem()}, nil + default: + return &dynamodb.GetItemOutput{}, nil + } + }, + } +} + +func testDeps(fake *storetest.FakeDynamo, now time.Time) *lambda.Deps { + return &lambda.Deps{ + Store: storetest.NewStore(fake), + Logger: slog.Default(), + NowFunc: func() time.Time { return now }, + } +} + +// TestSLACancel_AppliesT1SLADate pins the T+1 SLA-date rule on the cancel +// path: sensor-triggered daily pipelines shift the deadline date to D+1 (the +// same rule the watchdog applies), while cron pipelines keep their own date. +func TestSLACancel_AppliesT1SLADate(t *testing.T) { + sensorDailyCfg := types.PipelineConfig{ + Pipeline: types.PipelineIdentity{ID: "p"}, + Schedule: types.ScheduleConfig{}, + SLA: &types.SLAConfig{Deadline: "10:00", ExpectedDuration: "30m"}, + } + cronCfg := types.PipelineConfig{ + Pipeline: types.PipelineIdentity{ID: "p"}, + Schedule: types.ScheduleConfig{Cron: "0 8 * * *"}, + SLA: &types.SLAConfig{Deadline: "10:00", ExpectedDuration: "30m"}, + } + + tests := []struct { + name string + cfg types.PipelineConfig + cfgErr error + scheduleID string + now time.Time + wantAlertType string + wantBreachAt string + }{ + { + name: "sensor-daily met before T+1 deadline", + cfg: sensorDailyCfg, + scheduleID: "stream", + now: time.Date(2026, 3, 11, 4, 0, 0, 0, time.UTC), + wantAlertType: string(types.EventSLAMet), + wantBreachAt: "2026-03-11T10:00:00Z", + }, + { + name: "sensor-daily breached after T+1 deadline", + cfg: sensorDailyCfg, + scheduleID: "stream", + now: time.Date(2026, 3, 11, 11, 0, 0, 0, time.UTC), + wantAlertType: string(types.EventSLABreach), + wantBreachAt: "2026-03-11T10:00:00Z", + }, + { + name: "cron pipeline breached, no roll-forward", + cfg: cronCfg, + scheduleID: "cron", + now: time.Date(2026, 3, 10, 11, 0, 0, 0, time.UTC), + wantAlertType: string(types.EventSLABreach), + wantBreachAt: "2026-03-10T10:00:00Z", + }, + { + name: "cron pipeline met, no roll-forward", + cfg: cronCfg, + scheduleID: "cron", + now: time.Date(2026, 3, 10, 9, 0, 0, 0, time.UTC), + wantAlertType: string(types.EventSLAMet), + wantBreachAt: "2026-03-10T10:00:00Z", + }, + { + name: "config lookup failure falls back to unshifted date", + cfg: sensorDailyCfg, + cfgErr: errors.New("dynamodb: internal error"), + scheduleID: "stream", + now: time.Date(2026, 3, 10, 11, 0, 0, 0, time.UTC), + wantAlertType: string(types.EventSLABreach), + wantBreachAt: "2026-03-10T10:00:00Z", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + fake := newFakeDynamo(t, tt.cfg, tt.cfgErr) + d := testDeps(fake, tt.now) + + out, err := sla.HandleSLAMonitor(context.Background(), d, lambda.SLAMonitorInput{ + Mode: "cancel", + PipelineID: "p", + ScheduleID: tt.scheduleID, + Date: "2026-03-10", + Deadline: "10:00", + ExpectedDuration: "30m", + }) + if err != nil { + t.Fatalf("HandleSLAMonitor returned error: %v", err) + } + if out.AlertType != tt.wantAlertType { + t.Errorf("AlertType = %q, want %q", out.AlertType, tt.wantAlertType) + } + if out.BreachAt != tt.wantBreachAt { + t.Errorf("BreachAt = %q, want %q", out.BreachAt, tt.wantBreachAt) + } + }) + } +} + +// TestSLACancel_NoDeadlineSkipsVerdict pins fix 3: when neither an absolute +// deadline nor a relative (maxDuration + sensorArrivalAt) deadline can be +// determined, cancel must not fabricate an SLA_MET verdict. +func TestSLACancel_NoDeadlineSkipsVerdict(t *testing.T) { + fake := newFakeDynamo(t, types.PipelineConfig{}, nil) + d := testDeps(fake, time.Date(2026, 3, 10, 11, 0, 0, 0, time.UTC)) + eb := &countingEventBridge{} + d.EventBridge = eb + d.EventBusName = "test-bus" + + out, err := sla.HandleSLAMonitor(context.Background(), d, lambda.SLAMonitorInput{ + Mode: "cancel", + PipelineID: "p", + ScheduleID: "stream", + Date: "2026-03-10", + MaxDuration: "2h", + SensorArrivalAt: "", + Deadline: "", + }) + if err != nil { + t.Fatalf("HandleSLAMonitor returned error: %v", err) + } + if out.BreachAt != "" { + t.Errorf("BreachAt = %q, want empty (no deadline could be determined)", out.BreachAt) + } + if out.WarningAt != "" { + t.Errorf("WarningAt = %q, want empty (no deadline could be determined)", out.WarningAt) + } + if eb.calls != 0 { + t.Errorf("PutEvents called %d times, want 0 — no verdict should be published without a deadline", eb.calls) + } +} diff --git a/internal/lambda/sla_date_test.go b/internal/lambda/sla_date_test.go new file mode 100644 index 0000000..516b1f6 --- /dev/null +++ b/internal/lambda/sla_date_test.go @@ -0,0 +1,47 @@ +package lambda_test + +import ( + "testing" + + lambda "github.com/dwsmith1983/interlock/internal/lambda" + "github.com/dwsmith1983/interlock/pkg/types" +) + +func TestResolveSLADate(t *testing.T) { + sensorDaily := &types.PipelineConfig{ + SLA: &types.SLAConfig{Deadline: "10:00"}, + } + cron := &types.PipelineConfig{ + Schedule: types.ScheduleConfig{Cron: "0 8 * * *"}, + SLA: &types.SLAConfig{Deadline: "10:00"}, + } + hourly := &types.PipelineConfig{ + SLA: &types.SLAConfig{Deadline: ":30"}, + } + nilSLA := &types.PipelineConfig{} + + tests := []struct { + name string + cfg *types.PipelineConfig + date string + want string + }{ + {"cron pipeline unchanged", cron, "2026-03-10", "2026-03-10"}, + {"sensor daily shifted", sensorDaily, "2026-03-10", "2026-03-11"}, + {"month end", sensorDaily, "2026-03-31", "2026-04-01"}, + {"leap day", sensorDaily, "2028-02-28", "2028-02-29"}, + {"hourly deadline unchanged", hourly, "2026-03-10", "2026-03-10"}, + {"composite hourly date unchanged", sensorDaily, "2026-03-10T13", "2026-03-10T13"}, + {"nil SLA unchanged", nilSLA, "2026-03-10", "2026-03-10"}, + {"unparseable date unchanged", sensorDaily, "bad", "bad"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := lambda.ResolveSLADate(tt.cfg, tt.date) + if got != tt.want { + t.Errorf("ResolveSLADate(cfg, %q) = %q, want %q", tt.date, got, tt.want) + } + }) + } +} diff --git a/internal/lambda/stream/dryrun.go b/internal/lambda/stream/dryrun.go index facaa06..224c996 100644 --- a/internal/lambda/stream/dryrun.go +++ b/internal/lambda/stream/dryrun.go @@ -147,7 +147,7 @@ func publishDryRunSLAProjection(ctx context.Context, d *lambda.Deps, cfg *types. Mode: "calculate", PipelineID: pipelineID, ScheduleID: scheduleID, - Date: date, + Date: lambda.ResolveSLADate(cfg, date), Deadline: cfg.SLA.Deadline, ExpectedDuration: cfg.SLA.ExpectedDuration, Timezone: cfg.SLA.Timezone, diff --git a/internal/lambda/watchdog/sla.go b/internal/lambda/watchdog/sla.go index c19d65a..91daaa3 100644 --- a/internal/lambda/watchdog/sla.go +++ b/internal/lambda/watchdog/sla.go @@ -45,13 +45,7 @@ func scheduleSLAAlerts(ctx context.Context, d *lambda.Deps) error { scheduleID := lambda.ResolveScheduleID(cfg) date := resolveWatchdogSLADate(cfg, now) - slaDate := date - if cfg.Schedule.Cron == "" && !strings.HasPrefix(cfg.SLA.Deadline, ":") { - t, err := time.Parse("2006-01-02", date) - if err == nil { - slaDate = t.AddDate(0, 0, 1).Format("2006-01-02") - } - } + slaDate := lambda.ResolveSLADate(cfg, date) tr, err := d.Store.GetTrigger(ctx, id, scheduleID, date) switch { From 1ea7b154e86a62e7568a07486d73a15c9bce5cc6 Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Fri, 11 Sep 2026 23:45:06 +0700 Subject: [PATCH 10/11] fix: log orchestrator evaluate failures handleEvaluate returned status: "error" on GetConfig failure, missing config, and GetAllSensors failure with no server-side log line, so the only record of the failure was the Step Functions execution history. Log each at error level with the pipeline ID and underlying error before returning. --- internal/lambda/orchestrator/evaluate.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/internal/lambda/orchestrator/evaluate.go b/internal/lambda/orchestrator/evaluate.go index d4f3ad0..136cc22 100644 --- a/internal/lambda/orchestrator/evaluate.go +++ b/internal/lambda/orchestrator/evaluate.go @@ -21,14 +21,18 @@ const statusError = "error" func handleEvaluate(ctx context.Context, d *lambda.Deps, input lambda.OrchestratorInput) (lambda.OrchestratorOutput, error) { cfg, err := d.Store.GetConfig(ctx, input.PipelineID) if err != nil { + d.Logger.ErrorContext(ctx, "evaluate failed", "pipelineId", input.PipelineID, "error", err) return lambda.OrchestratorOutput{Mode: "evaluate", Status: statusError, Error: err.Error()}, nil } if cfg == nil { - return lambda.OrchestratorOutput{Mode: "evaluate", Status: statusError, Error: fmt.Sprintf("config not found for pipeline %q", input.PipelineID)}, nil + notFoundErr := fmt.Sprintf("config not found for pipeline %q", input.PipelineID) + d.Logger.ErrorContext(ctx, "evaluate failed", "pipelineId", input.PipelineID, "error", notFoundErr) + return lambda.OrchestratorOutput{Mode: "evaluate", Status: statusError, Error: notFoundErr}, nil } sensors, err := d.Store.GetAllSensors(ctx, input.PipelineID) if err != nil { + d.Logger.ErrorContext(ctx, "evaluate failed", "pipelineId", input.PipelineID, "error", err) return lambda.OrchestratorOutput{Mode: "evaluate", Status: statusError, Error: err.Error()}, nil } From 4a5fca6769c76e3cdc636c71e23b2cabb9987327 Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Fri, 11 Sep 2026 23:45:09 +0700 Subject: [PATCH 11/11] docs: changelog for T+1 cancel fix and release notes --- CHANGELOG.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index b0c2459..7394d9c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -41,6 +41,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Absolute SLA deadlines are anchored to the execution date** — `CalculateAbsoluteDeadline` rolled an explicit execution date forward by 24h (or 1h for `:MM` deadlines) whenever the deadline had already passed. `sla-monitor` `cancel` therefore published `SLA_MET` for runs that finished late, the `reconcile` breach branch was unreachable, and the watchdog scheduled breach alerts a day late. Roll-forward now applies only when no execution date is supplied, and uses `AddDate` so the wall-clock time survives DST transitions. - **Orchestrator evaluate/trigger results satisfy the state machine contract** — `handleEvaluate` returned a result without `status` on storage failures, so the `IsReady` Choice dereferenced a missing `$.evaluateResult.status`; it now always emits a status. `handleTrigger` returned a partial result with a nil error on configuration failures, so `HasTriggerResult` saw `IsPresent=true` and `CheckJob` dereferenced a missing `runId`, killing the execution with the `TRIGGER#` lock stuck in `RUNNING` until TTL; those paths now return a Lambda error so `Trigger`'s Retry/Catch routes to `TriggerRetryExhausted`, which releases the lock. - **Time-dependent tests** — `TestSLAMonitor_Calculate_ReturnsRFC3339` (failing since 2026-06-16), `TestSLAMonitor_Reconcile_ReturnsDeadlines`, `TestSLAMonitor_Cancel_RecalculatesWhenTimesNotProvided` and three watchdog proactive-SLA tests now inject `Deps.NowFunc` instead of relying on the wall clock. +- **SLA cancel verdict uses the T+1 date for sensor-triggered daily pipelines** — `handleSLACancel` and the dry-run SLA projection recomputed the absolute deadline from `input.Date`/`date` (the data date D), but sensor-triggered daily pipelines run T+1 — data for date D arrives on D+1, and the watchdog's proactive SLA scheduling already anchors the deadline to D+1. Now that an explicit date no longer rolls forward when its deadline has passed, `cancel` published a false `SLA_BREACH` for pipelines that finished on D+1 before their deadline. The new `internal/lambda.ResolveSLADate` centralizes the watchdog's T+1 rule (cron pipelines and hourly `:MM` deadlines are unaffected) and is now shared by the watchdog, `sla.handleSLACancel`, and `stream.publishDryRunSLAProjection`. `orchestrator.handleEvaluate` storage failures (`GetConfig` error, config not found, `GetAllSensors` error) are now also logged at error level instead of surfacing only in the returned `status: "error"` payload. + +**Release notes**: verdicts for pipelines with a non-UTC `sla.timezone` are now computed in that zone; absolute SLA deadlines no longer roll forward to the next day when past; in-flight executions started before deploy are unaffected by the state-machine change; the sla-monitor `reconcile` mode has no production invoker today and is not a safety net for deadlines missed while the watchdog was down. ### Dependencies