diff --git a/CHANGELOG.md b/CHANGELOG.md index 23bf10b..7394d9c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,6 +34,17 @@ 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. +- **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 - `go.opentelemetry.io/otel` v1.43.0 (traces + metrics) 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/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) + } + } + }) + } +} 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/evaluate.go b/internal/lambda/orchestrator/evaluate.go index 284cb5e..136cc22 100644 --- a/internal/lambda/orchestrator/evaluate.go +++ b/internal/lambda/orchestrator/evaluate.go @@ -9,20 +9,31 @@ 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 + 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", 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 { - return lambda.OrchestratorOutput{Mode: "evaluate", Error: err.Error()}, nil + d.Logger.ErrorContext(ctx, "evaluate failed", "pipelineId", input.PipelineID, "error", err) + 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..4a84b45 --- /dev/null +++ b/internal/lambda/orchestrator/handler_contract_test.go @@ -0,0 +1,273 @@ +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. +// 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("p")}, + "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(t, 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(t, 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(t, 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(t, 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 b330541..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) @@ -46,8 +53,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 +100,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/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/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"]) + } + }) + } +} 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/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/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/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"` 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 { 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/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) + } +} 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 {