diff --git a/cmd/server/main.go b/cmd/server/main.go index b44a0e5..261551b 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -479,7 +479,7 @@ func (s *ServerCLI) Run(_ *kong.Context) error { fmt.Printf(" Scans will run automatically (schedule: %s)\n", s.ScheduleCron) } fmt.Println("\nšŸ“– To trigger a scan manually, use the Temporal UI or CLI:") - fmt.Printf(" temporal workflow start --task-queue %s --type %s --input '{}'\n", s.TemporalTaskQueue, orchestrator.OrchestratorWorkflowType) + fmt.Printf(" temporal workflow start --workflow-id %s --task-queue %s --type %s --input '{}'\n", orchestrator.ActiveScanWorkflowID, s.TemporalTaskQueue, orchestrator.OrchestratorWorkflowType) fmt.Println("\nšŸ“– For more information, see the README.md") fmt.Println("\nPress Ctrl+C to stop...") diff --git a/pkg/scan/handler.go b/pkg/scan/handler.go index 9504d99..76ddd69 100644 --- a/pkg/scan/handler.go +++ b/pkg/scan/handler.go @@ -2,9 +2,12 @@ package scan import ( "encoding/json" + "errors" "fmt" "net/http" + "go.temporal.io/api/serviceerror" + "github.com/block/Version-Guard/pkg/telemetry" "github.com/block/Version-Guard/pkg/types" ) @@ -65,6 +68,10 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { Source: telemetry.ScanSourceHTTP, }) if err != nil { + if isScanAlreadyRunning(err) { + writeError(w, http.StatusConflict, err.Error()) + return + } writeError(w, http.StatusInternalServerError, err.Error()) return } @@ -83,3 +90,8 @@ func writeJSON(w http.ResponseWriter, status int, body interface{}) { func writeError(w http.ResponseWriter, status int, msg string) { writeJSON(w, status, map[string]string{"error": msg}) } + +func isScanAlreadyRunning(err error) bool { + var alreadyStarted *serviceerror.WorkflowExecutionAlreadyStarted + return errors.As(err, &alreadyStarted) +} diff --git a/pkg/scan/handler_test.go b/pkg/scan/handler_test.go index 83c4d17..2c1ffb1 100644 --- a/pkg/scan/handler_test.go +++ b/pkg/scan/handler_test.go @@ -10,6 +10,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "go.temporal.io/api/serviceerror" "github.com/block/Version-Guard/pkg/types" "github.com/block/Version-Guard/pkg/workflow/orchestrator" @@ -122,3 +123,15 @@ func TestHandler_TriggerError_Returns500(t *testing.T) { assert.Equal(t, http.StatusInternalServerError, rec.Code) assert.Contains(t, rec.Body.String(), "temporal unavailable") } + +func TestHandler_ScanAlreadyRunning_Returns409(t *testing.T) { + mock := &mockStarter{err: serviceerror.NewWorkflowExecutionAlreadyStarted("already running", "request", "run")} + h := newTestHandler(t, mock) + + req := httptest.NewRequest(http.MethodPost, "/scan", http.NoBody) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + + assert.Equal(t, http.StatusConflict, rec.Code) + assert.Contains(t, rec.Body.String(), "already running") +} diff --git a/pkg/scan/scan.go b/pkg/scan/scan.go index 8376c16..f983fe1 100644 --- a/pkg/scan/scan.go +++ b/pkg/scan/scan.go @@ -11,6 +11,7 @@ import ( "time" "github.com/google/uuid" + enumspb "go.temporal.io/api/enums/v1" "go.temporal.io/sdk/client" "github.com/block/Version-Guard/pkg/telemetry" @@ -147,9 +148,11 @@ func (t *Trigger) Run(ctx context.Context, in Input) (res Result, err error) { workflowID = buildWorkflowID(scanID) opts := client.StartWorkflowOptions{ - ID: workflowID, - TaskQueue: t.taskQueue, - WorkflowExecutionTimeout: defaultExecutionTimeout, + ID: workflowID, + TaskQueue: t.taskQueue, + WorkflowExecutionTimeout: defaultExecutionTimeout, + WorkflowIDConflictPolicy: enumspb.WORKFLOW_ID_CONFLICT_POLICY_FAIL, + WorkflowExecutionErrorWhenAlreadyStarted: true, } run, err := t.starter.ExecuteWorkflow(ctx, opts, orchestrator.OrchestratorWorkflow, orchestrator.WorkflowInput{ @@ -171,9 +174,8 @@ func (t *Trigger) Run(ctx context.Context, in Input) (res Result, err error) { }, nil } -// buildWorkflowID produces a workflow ID that is distinguishable from -// scheduled executions. Scheduled runs use the schedule's generated IDs; -// manual runs are prefixed so they are easy to find in Temporal UI/CLI. -func buildWorkflowID(scanID string) string { - return fmt.Sprintf("version-guard-scan-%s", scanID) +// buildWorkflowID returns the singleton orchestrator workflow ID. Temporal +// rejects a new run with this ID while the previous scan is still open. +func buildWorkflowID(_ string) string { + return orchestrator.ActiveScanWorkflowID } diff --git a/pkg/scan/scan_test.go b/pkg/scan/scan_test.go index 1040827..bc2c93b 100644 --- a/pkg/scan/scan_test.go +++ b/pkg/scan/scan_test.go @@ -3,11 +3,11 @@ package scan import ( "context" "errors" - "strings" "testing" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + enumspb "go.temporal.io/api/enums/v1" "go.temporal.io/sdk/client" "github.com/block/Version-Guard/pkg/types" @@ -50,7 +50,7 @@ func (m *mockStarter) ExecuteWorkflow(_ context.Context, options client.StartWor func TestTrigger_Run_FullScan(t *testing.T) { mock := &mockStarter{ - run: &mockWorkflowRun{id: "version-guard-scan-abc", runID: "run-1"}, + run: &mockWorkflowRun{id: orchestrator.ActiveScanWorkflowID, runID: "run-1"}, } defaults := []types.ResourceType{"aurora-mysql", "eks"} tr := NewTriggerWithStarter(mock, "version-guard-orchestrator", defaults) @@ -58,13 +58,15 @@ func TestTrigger_Run_FullScan(t *testing.T) { res, err := tr.Run(context.Background(), Input{ScanID: "abc"}) require.NoError(t, err) - assert.Equal(t, "version-guard-scan-abc", res.WorkflowID) + assert.Equal(t, orchestrator.ActiveScanWorkflowID, res.WorkflowID) assert.Equal(t, "run-1", res.RunID) assert.Equal(t, "abc", res.ScanID) require.True(t, mock.called) - assert.Equal(t, "version-guard-scan-abc", mock.calledOpts.ID) + assert.Equal(t, orchestrator.ActiveScanWorkflowID, mock.calledOpts.ID) assert.Equal(t, "version-guard-orchestrator", mock.calledOpts.TaskQueue) + assert.Equal(t, enumspb.WORKFLOW_ID_CONFLICT_POLICY_FAIL, mock.calledOpts.WorkflowIDConflictPolicy) + assert.True(t, mock.calledOpts.WorkflowExecutionErrorWhenAlreadyStarted) require.Len(t, mock.calledArgs, 1) in, ok := mock.calledArgs[0].(orchestrator.WorkflowInput) @@ -115,8 +117,7 @@ func TestTrigger_Run_GeneratesScanIDWhenEmpty(t *testing.T) { require.NoError(t, err) assert.NotEmpty(t, res.ScanID, "ScanID should be generated when not provided") - assert.True(t, strings.HasPrefix(mock.calledOpts.ID, "version-guard-scan-"), - "workflow ID should be prefixed") + assert.Equal(t, orchestrator.ActiveScanWorkflowID, mock.calledOpts.ID) in := mock.calledArgs[0].(orchestrator.WorkflowInput) assert.Equal(t, res.ScanID, in.ScanID, "generated ScanID should be passed to workflow") } diff --git a/pkg/schedule/schedule.go b/pkg/schedule/schedule.go index 4f6f4c6..ead7e34 100644 --- a/pkg/schedule/schedule.go +++ b/pkg/schedule/schedule.go @@ -6,6 +6,7 @@ import ( "fmt" "time" + enumspb "go.temporal.io/api/enums/v1" "go.temporal.io/sdk/client" "go.temporal.io/sdk/temporal" @@ -77,6 +78,7 @@ func (m *Manager) EnsureSchedule(ctx context.Context, cfg Config) error { Jitter: cfg.Jitter, }, Action: &client.ScheduleWorkflowAction{ + ID: orchestrator.ActiveScanWorkflowID, Workflow: orchestrator.OrchestratorWorkflow, Args: []interface{}{orchestrator.WorkflowInput{ ResourceTypes: cfg.ResourceTypes, @@ -86,7 +88,8 @@ func (m *Manager) EnsureSchedule(ctx context.Context, cfg Config) error { TaskQueue: cfg.TaskQueue, WorkflowExecutionTimeout: 2 * time.Hour, }, - Paused: cfg.Paused, + Overlap: enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, + Paused: cfg.Paused, } _, err := m.scheduleClient.Create(ctx, opts) @@ -117,8 +120,9 @@ func (m *Manager) EnsureSchedule(ctx context.Context, cfg Config) error { } existingCrons := existingSpec.CronExpressions specMatches := len(existingCrons) == 1 && existingCrons[0] == cfg.CronExpression && existingSpec.Jitter == cfg.Jitter + overlapMatches := desc.Schedule.Policy != nil && desc.Schedule.Policy.Overlap == enumspb.SCHEDULE_OVERLAP_POLICY_SKIP actionMatches := scheduleActionMatches(desc.Schedule.Action, &cfg) - if specMatches && actionMatches { + if specMatches && overlapMatches && actionMatches { fmt.Printf(" Schedule %q already configured (cron: %s)\n", cfg.ScheduleID, cfg.CronExpression) return nil } @@ -140,7 +144,12 @@ func (m *Manager) EnsureSchedule(ctx context.Context, cfg Config) error { CronExpressions: []string{cfg.CronExpression}, Jitter: cfg.Jitter, } + if input.Description.Schedule.Policy == nil { + input.Description.Schedule.Policy = &client.SchedulePolicies{} + } + input.Description.Schedule.Policy.Overlap = enumspb.SCHEDULE_OVERLAP_POLICY_SKIP if action, ok := input.Description.Schedule.Action.(*client.ScheduleWorkflowAction); ok { + action.ID = orchestrator.ActiveScanWorkflowID action.TaskQueue = cfg.TaskQueue action.Args = []interface{}{orchestrator.WorkflowInput{ ResourceTypes: cfg.ResourceTypes, @@ -175,6 +184,9 @@ func scheduleActionMatches(action client.ScheduleAction, cfg *Config) bool { if !ok { return false } + if wfAction.ID != orchestrator.ActiveScanWorkflowID { + return false + } if wfAction.TaskQueue != cfg.TaskQueue { return false } diff --git a/pkg/schedule/schedule_test.go b/pkg/schedule/schedule_test.go index c20fa34..ae66069 100644 --- a/pkg/schedule/schedule_test.go +++ b/pkg/schedule/schedule_test.go @@ -8,6 +8,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + enumspb "go.temporal.io/api/enums/v1" "go.temporal.io/sdk/client" "go.temporal.io/sdk/temporal" @@ -129,7 +130,9 @@ func TestEnsureSchedule_CreatesNew(t *testing.T) { assert.Equal(t, "test-schedule", mock.createOpts.ID) assert.Equal(t, []string{"0 */6 * * *"}, mock.createOpts.Spec.CronExpressions) assert.Equal(t, 5*time.Minute, mock.createOpts.Spec.Jitter) + assert.Equal(t, enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, mock.createOpts.Overlap) action := mock.createOpts.Action.(*client.ScheduleWorkflowAction) + assert.Equal(t, orchestrator.ActiveScanWorkflowID, action.ID) assert.Equal(t, "test-queue", action.TaskQueue) assert.Equal(t, 2*time.Hour, action.WorkflowExecutionTimeout) require.Len(t, action.Args, 1) @@ -143,11 +146,13 @@ func TestEnsureSchedule_AlreadyExists_SameCron(t *testing.T) { id: "test-schedule", describeOut: &client.ScheduleDescription{ Schedule: client.Schedule{ + Policy: &client.SchedulePolicies{Overlap: enumspb.SCHEDULE_OVERLAP_POLICY_SKIP}, Spec: &client.ScheduleSpec{ CronExpressions: []string{"0 */6 * * *"}, Jitter: 5 * time.Minute, }, Action: &client.ScheduleWorkflowAction{ + ID: orchestrator.ActiveScanWorkflowID, TaskQueue: "test-queue", Args: []interface{}{orchestrator.WorkflowInput{ ResourceTypes: testResourceTypes, @@ -227,8 +232,11 @@ func TestEnsureSchedule_AlreadyExists_NewWebhookURL(t *testing.T) { require.NoError(t, err) assert.True(t, handle.updateCalled, "Update must be called when EmitterWebhookURL changes") require.NotNil(t, captured) + require.NotNil(t, captured.Schedule.Policy) + assert.Equal(t, enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, captured.Schedule.Policy.Overlap) action, ok := captured.Schedule.Action.(*client.ScheduleWorkflowAction) require.True(t, ok, "action should be a ScheduleWorkflowAction") + assert.Equal(t, orchestrator.ActiveScanWorkflowID, action.ID) require.Len(t, action.Args, 1) in, ok := action.Args[0].(orchestrator.WorkflowInput) require.True(t, ok) diff --git a/pkg/store/memory/store.go b/pkg/store/memory/store.go index aa1d7f7..0152240 100644 --- a/pkg/store/memory/store.go +++ b/pkg/store/memory/store.go @@ -114,17 +114,27 @@ func (s *Store) GetSummary(ctx context.Context, filters store.FindingFilters) (* return summary, nil } -// DeleteStaleFindings removes findings older than the specified time -func (s *Store) DeleteStaleFindings(ctx context.Context, resourceType types.ResourceType, olderThan time.Time) error { +// ReplaceFindings atomically replaces all findings of resourceType with +// the given set, evicting entries absent from it +func (s *Store) ReplaceFindings(ctx context.Context, resourceType types.ResourceType, findings []*types.Finding) error { s.mu.Lock() defer s.mu.Unlock() for id, finding := range s.findings { - if finding.ResourceType == resourceType && finding.UpdatedAt.Before(olderThan) { + if finding.ResourceType == resourceType { delete(s.findings, id) } } + now := time.Now() + for _, finding := range findings { + finding.UpdatedAt = now + if finding.DetectedAt.IsZero() { + finding.DetectedAt = now + } + s.findings[finding.ResourceID] = finding + } + return nil } diff --git a/pkg/store/memory/store_test.go b/pkg/store/memory/store_test.go index 676fd96..3397226 100644 --- a/pkg/store/memory/store_test.go +++ b/pkg/store/memory/store_test.go @@ -203,45 +203,45 @@ func TestStore_GetSummary(t *testing.T) { assert.False(t, summary.LastScanTime.IsZero()) } -func TestStore_DeleteStaleFindings(t *testing.T) { +func TestStore_ReplaceFindings(t *testing.T) { ctx := context.Background() s := NewStore() - oldTime := time.Now().Add(-48 * time.Hour) - recentTime := time.Now().Add(-1 * time.Hour) - - findings := []*types.Finding{ - {ResourceID: "1", ResourceType: types.ResourceTypeAurora, UpdatedAt: oldTime}, - {ResourceID: "2", ResourceType: types.ResourceTypeAurora, UpdatedAt: recentTime}, - {ResourceID: "3", ResourceType: types.ResourceTypeElastiCache, UpdatedAt: oldTime}, - } - - // Manually set UpdatedAt (SaveFindings would overwrite it) - s.mu.Lock() - for _, f := range findings { - s.findings[f.ResourceID] = f - } - s.mu.Unlock() - - // Delete Aurora findings older than 24 hours ago - cutoff := time.Now().Add(-24 * time.Hour) - err := s.DeleteStaleFindings(ctx, types.ResourceTypeAurora, cutoff) + err := s.SaveFindings(ctx, []*types.Finding{ + {ResourceID: "1", ResourceType: types.ResourceTypeAurora}, + {ResourceID: "2", ResourceType: types.ResourceTypeAurora}, + {ResourceID: "3", ResourceType: types.ResourceTypeElastiCache}, + }) require.NoError(t, err) - // Should have 2 findings left (1 recent Aurora + 1 old ElastiCache) - assert.Len(t, s.findings, 2) + // Replace Aurora's set: "2" survives (re-listed), "1" is evicted, + // "4" is new, and the ElastiCache finding is untouched. + err = s.ReplaceFindings(ctx, types.ResourceTypeAurora, []*types.Finding{ + {ResourceID: "2", ResourceType: types.ResourceTypeAurora}, + {ResourceID: "4", ResourceType: types.ResourceTypeAurora}, + }) + require.NoError(t, err) - // Old Aurora finding should be deleted + assert.Len(t, s.findings, 3) _, exists := s.findings["1"] assert.False(t, exists) - - // Recent Aurora finding should remain _, exists = s.findings["2"] assert.True(t, exists) - - // Old ElastiCache finding should remain (different resource type) + _, exists = s.findings["4"] + assert.True(t, exists) _, exists = s.findings["3"] assert.True(t, exists) + + // UpdatedAt is stamped on the replacement set + assert.False(t, s.findings["4"].UpdatedAt.IsZero()) + + // An empty replacement set empties the type + err = s.ReplaceFindings(ctx, types.ResourceTypeAurora, nil) + require.NoError(t, err) + aurora := types.ResourceTypeAurora + remaining, err := s.ListFindings(ctx, store.FindingFilters{ResourceType: &aurora}) + require.NoError(t, err) + assert.Empty(t, remaining) } func TestStore_UpdateExistingFinding(t *testing.T) { diff --git a/pkg/store/store.go b/pkg/store/store.go index 51cd6e2..c6d5c1e 100644 --- a/pkg/store/store.go +++ b/pkg/store/store.go @@ -21,8 +21,9 @@ type Store interface { // GetSummary calculates aggregate statistics for findings GetSummary(ctx context.Context, filters FindingFilters) (*ScanSummary, error) - // DeleteStaleFindings removes findings older than the specified time - DeleteStaleFindings(ctx context.Context, resourceType types.ResourceType, olderThan time.Time) error + // ReplaceFindings atomically replaces all findings of resourceType + // with the given set, evicting entries absent from it + ReplaceFindings(ctx context.Context, resourceType types.ResourceType, findings []*types.Finding) error } // FindingFilters defines optional filters for querying findings. diff --git a/pkg/workflow/detection/activities.go b/pkg/workflow/detection/activities.go index d5c6ba9..d3c6a7f 100644 --- a/pkg/workflow/detection/activities.go +++ b/pkg/workflow/detection/activities.go @@ -60,9 +60,11 @@ type DetectResult struct { Findings []*types.Finding } +//nolint:govet // field alignment sacrificed for logical grouping type StoreInput struct { FindingsBatchID string Findings []*types.Finding + ResourceType types.ResourceType } //nolint:govet // field alignment sacrificed for logical grouping @@ -288,7 +290,12 @@ func (a *Activities) DetectDrift(ctx context.Context, input DetectInput) (*Detec }, nil } -// StoreFindings persists findings to the store +// StoreFindings replaces the resource type's findings with this scan's +// output, evicting entries that dropped out of the inventory. Without +// eviction, deleted resources persist in the worker-local store (and in +// every snapshot) until the pod restarts. Inputs recorded by workflows +// started before ResourceType existed carry the zero value and fall +// back to plain upserts. func (a *Activities) StoreFindings(ctx context.Context, input StoreInput) error { findings, err := a.loadFindings(input.FindingsBatchID, input.Findings) if err != nil { @@ -298,7 +305,19 @@ func (a *Activities) StoreFindings(ctx context.Context, input StoreInput) error logger := activity.GetLogger(ctx) logger.Info("Storing findings", "count", len(findings)) - if err := a.Store.SaveFindings(ctx, findings); err != nil { + if input.ResourceType == "" { + if err := a.Store.SaveFindings(ctx, findings); err != nil { + return err + } + logger.Info("Findings stored successfully") + return nil + } + + if len(findings) == 0 { + logger.Warn("Scan returned no findings; evicting all findings for type", + "resourceType", input.ResourceType) + } + if err := a.Store.ReplaceFindings(ctx, input.ResourceType, findings); err != nil { return err } diff --git a/pkg/workflow/detection/activities_test.go b/pkg/workflow/detection/activities_test.go index 4428fce..d85b4db 100644 --- a/pkg/workflow/detection/activities_test.go +++ b/pkg/workflow/detection/activities_test.go @@ -1,6 +1,7 @@ package detection import ( + "context" "testing" "time" @@ -13,6 +14,7 @@ import ( "github.com/block/Version-Guard/pkg/inventory" invmock "github.com/block/Version-Guard/pkg/inventory/mock" "github.com/block/Version-Guard/pkg/policy" + "github.com/block/Version-Guard/pkg/store" "github.com/block/Version-Guard/pkg/store/memory" "github.com/block/Version-Guard/pkg/types" ) @@ -457,6 +459,140 @@ func TestStoreFindings_CacheMiss(t *testing.T) { assert.Contains(t, err.Error(), "findings batch \"nonexistent\" not found") } +// seedFindings stores findings as if written by a previous scan. +func seedFindings(t *testing.T, act *Activities, findings ...*types.Finding) { + t.Helper() + require.NoError(t, act.Store.SaveFindings(context.Background(), findings)) +} + +func listByType(t *testing.T, act *Activities, rt types.ResourceType) []*types.Finding { + t.Helper() + findings, err := act.Store.ListFindings(context.Background(), store.FindingFilters{ResourceType: &rt}) + require.NoError(t, err) + return findings +} + +func TestStoreFindings_EvictsSameTypeFindingsMissingFromScan(t *testing.T) { + act := newTestActivities(nil, nil) + + // Previous scan: a lambda that has since been deleted from the + // inventory, plus a finding of another type this scan must not touch. + seedFindings(t, act, + &types.Finding{ResourceID: "arn:aws:lambda:us-west-2:1:function:deleted", ResourceType: types.ResourceTypeLambda, Status: types.StatusRed}, + &types.Finding{ResourceID: "arn:aws:eks:us-west-2:1:cluster/other", ResourceType: types.ResourceTypeEKS, Status: types.StatusGreen}, + ) + + env := newActivityEnv() + env.RegisterActivity(act.StoreFindings) + + _, err := env.ExecuteActivity(act.StoreFindings, StoreInput{ + ResourceType: types.ResourceTypeLambda, + Findings: []*types.Finding{ + {ResourceID: "arn:aws:lambda:us-west-2:1:function:current", ResourceType: types.ResourceTypeLambda, Status: types.StatusRed}, + }, + }) + require.NoError(t, err) + + lambdas := listByType(t, act, types.ResourceTypeLambda) + require.Len(t, lambdas, 1) + assert.Equal(t, "arn:aws:lambda:us-west-2:1:function:current", lambdas[0].ResourceID) + + require.Len(t, listByType(t, act, types.ResourceTypeEKS), 1) +} + +func TestStoreFindings_FromCache_Evicts(t *testing.T) { + act := newTestActivities(nil, nil) + + seedFindings(t, act, + &types.Finding{ResourceID: "arn:aws:lambda:us-west-2:1:function:deleted", ResourceType: types.ResourceTypeLambda, Status: types.StatusRed}, + ) + + // Production always routes findings through the worker-local batch + // cache (ScanID is always set), so eviction must work on this path. + act.resourceCache.Store("scan-evict-findings", []*types.Finding{ + {ResourceID: "arn:aws:lambda:us-west-2:1:function:current", ResourceType: types.ResourceTypeLambda, Status: types.StatusRed}, + }) + + env := newActivityEnv() + env.RegisterActivity(act.StoreFindings) + + _, err := env.ExecuteActivity(act.StoreFindings, StoreInput{ + FindingsBatchID: "scan-evict-findings", + ResourceType: types.ResourceTypeLambda, + }) + require.NoError(t, err) + + lambdas := listByType(t, act, types.ResourceTypeLambda) + require.Len(t, lambdas, 1) + assert.Equal(t, "arn:aws:lambda:us-west-2:1:function:current", lambdas[0].ResourceID) +} + +func TestStoreFindings_FromCache_NilBatchEvictsTypeToZero(t *testing.T) { + act := newTestActivities(nil, nil) + + seedFindings(t, act, + &types.Finding{ResourceID: "arn:aws:lambda:us-west-2:1:function:existing", ResourceType: types.ResourceTypeLambda, Status: types.StatusRed}, + ) + + // DetectDrift caches a nil slice when inventory comes back empty; the + // type must converge to zero on the cache path too. + act.resourceCache.Store("scan-empty-findings", []*types.Finding(nil)) + + env := newActivityEnv() + env.RegisterActivity(act.StoreFindings) + + _, err := env.ExecuteActivity(act.StoreFindings, StoreInput{ + FindingsBatchID: "scan-empty-findings", + ResourceType: types.ResourceTypeLambda, + }) + require.NoError(t, err) + + require.Empty(t, listByType(t, act, types.ResourceTypeLambda)) +} + +func TestStoreFindings_EmptyBatchEvictsTypeToZero(t *testing.T) { + act := newTestActivities(nil, nil) + + seedFindings(t, act, + &types.Finding{ResourceID: "arn:aws:lambda:us-west-2:1:function:existing", ResourceType: types.ResourceTypeLambda, Status: types.StatusRed}, + &types.Finding{ResourceID: "arn:aws:eks:us-west-2:1:cluster/other", ResourceType: types.ResourceTypeEKS, Status: types.StatusGreen}, + ) + + env := newActivityEnv() + env.RegisterActivity(act.StoreFindings) + + _, err := env.ExecuteActivity(act.StoreFindings, StoreInput{ + ResourceType: types.ResourceTypeLambda, + Findings: []*types.Finding{}, + }) + require.NoError(t, err) + + require.Empty(t, listByType(t, act, types.ResourceTypeLambda)) + require.Len(t, listByType(t, act, types.ResourceTypeEKS), 1) +} + +func TestStoreFindings_NoResourceTypeDoesNotEvict(t *testing.T) { + act := newTestActivities(nil, nil) + + seedFindings(t, act, + &types.Finding{ResourceID: "arn:aws:lambda:us-west-2:1:function:existing", ResourceType: types.ResourceTypeLambda, Status: types.StatusRed}, + ) + + env := newActivityEnv() + env.RegisterActivity(act.StoreFindings) + + // Inputs recorded by workflows started before ResourceType existed on + // StoreInput carry the zero value and must behave as before. + _, err := env.ExecuteActivity(act.StoreFindings, StoreInput{ + Findings: []*types.Finding{ + {ResourceID: "arn:aws:lambda:us-west-2:1:function:current", ResourceType: types.ResourceTypeLambda, Status: types.StatusRed}, + }, + }) + require.NoError(t, err) + + require.Len(t, listByType(t, act, types.ResourceTypeLambda), 2) +} + // --- EmitMetrics tests --- func TestEmitMetrics_FromCache_CleansUp(t *testing.T) { diff --git a/pkg/workflow/detection/workflow.go b/pkg/workflow/detection/workflow.go index 65c84bd..6282b70 100644 --- a/pkg/workflow/detection/workflow.go +++ b/pkg/workflow/detection/workflow.go @@ -135,6 +135,7 @@ func DetectionWorkflow(ctx workflow.Context, input WorkflowInput) (*WorkflowOutp err = workflow.ExecuteActivity(shortCtx, StoreFindingsActivityName, StoreInput{ FindingsBatchID: detectResult.FindingsBatchID, Findings: detectResult.Findings, + ResourceType: input.ResourceType, }).Get(shortCtx, nil) if err != nil { logger.Error("Failed to store findings", "error", err) diff --git a/pkg/workflow/detection/workflow_test.go b/pkg/workflow/detection/workflow_test.go index 899ce88..12af9b5 100644 --- a/pkg/workflow/detection/workflow_test.go +++ b/pkg/workflow/detection/workflow_test.go @@ -98,8 +98,10 @@ func TestDetectionWorkflow_Success(t *testing.T) { activity.RegisterOptions{Name: DetectDriftActivityName}, ) + var storeInput StoreInput env.RegisterActivityWithOptions( func(ctx context.Context, input StoreInput) error { + storeInput = input return nil }, activity.RegisterOptions{Name: StoreFindingsActivityName}, @@ -129,6 +131,7 @@ func TestDetectionWorkflow_Success(t *testing.T) { assert.Equal(t, "test-scan-001", output.ScanID) assert.Equal(t, types.ResourceTypeAurora, output.ResourceType) + assert.Equal(t, types.ResourceTypeAurora, storeInput.ResourceType) assert.Equal(t, 2, output.FindingsCount) assert.NotNil(t, output.Summary) assert.Equal(t, 50.0, output.Summary.CompliancePercentage) diff --git a/pkg/workflow/orchestrator/orchestrator_test.go b/pkg/workflow/orchestrator/orchestrator_test.go index 154bc47..06f62b3 100644 --- a/pkg/workflow/orchestrator/orchestrator_test.go +++ b/pkg/workflow/orchestrator/orchestrator_test.go @@ -9,6 +9,7 @@ import ( "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" "go.temporal.io/sdk/activity" + "go.temporal.io/sdk/client" "go.temporal.io/sdk/testsuite" "go.temporal.io/sdk/workflow" @@ -24,6 +25,7 @@ func newOrchestratorEnv(t *testing.T) *testsuite.TestWorkflowEnvironment { t.Helper() suite := &testsuite.WorkflowTestSuite{} env := suite.NewTestWorkflowEnvironment() + env.SetStartWorkflowOptions(client.StartWorkflowOptions{ID: ActiveScanWorkflowID}) env.RegisterWorkflow(OrchestratorWorkflow) env.RegisterActivityWithOptions( func(_ context.Context, _ RecordResourceScanResultInput) error { @@ -114,6 +116,23 @@ func TestOrchestratorWorkflow_EmptyResourceTypes_ReturnsError(t *testing.T) { assert.Contains(t, err.Error(), "ResourceTypes is empty") } +func TestOrchestratorWorkflow_RejectsNonSingletonWorkflowID(t *testing.T) { + suite := &testsuite.WorkflowTestSuite{} + env := suite.NewTestWorkflowEnvironment() + env.SetStartWorkflowOptions(client.StartWorkflowOptions{ID: "not-the-active-scan"}) + env.RegisterWorkflow(OrchestratorWorkflow) + + env.ExecuteWorkflow(OrchestratorWorkflow, WorkflowInput{ + ScanID: "scan-wrong-id", + ResourceTypes: []types.ResourceType{types.ResourceTypeAurora}, + }) + + require.True(t, env.IsWorkflowCompleted()) + err := env.GetWorkflowError() + require.Error(t, err) + assert.Contains(t, err.Error(), ActiveScanWorkflowID) +} + func TestOrchestratorWorkflow_ScanIDDefaultsToWorkflowID(t *testing.T) { env := newOrchestratorEnv(t) diff --git a/pkg/workflow/orchestrator/workflow.go b/pkg/workflow/orchestrator/workflow.go index 15197e2..553ee7f 100644 --- a/pkg/workflow/orchestrator/workflow.go +++ b/pkg/workflow/orchestrator/workflow.go @@ -23,6 +23,7 @@ var ErrNoResourceTypes = fmt.Errorf("orchestrator: WorkflowInput.ResourceTypes i const ( OrchestratorWorkflowType = "VersionGuardOrchestratorWorkflow" TaskQueueName = "version-guard-orchestrator" + ActiveScanWorkflowID = "version-guard-active-scan" ScanScopeFull = "full" ScanScopeTargeted = "targeted" @@ -87,6 +88,15 @@ type ResourceTypeResult struct { func OrchestratorWorkflow(ctx workflow.Context, input WorkflowInput) (*WorkflowOutput, error) { logger := workflow.GetLogger(ctx) info := workflow.GetInfo(ctx) + if info.WorkflowExecution.ID != ActiveScanWorkflowID { + err := fmt.Errorf("orchestrator: workflow ID must be %q, got %q", ActiveScanWorkflowID, info.WorkflowExecution.ID) + logger.Error("Orchestrator workflow failed", + "event", "scan_workflow_failed", + "workflowID", info.WorkflowExecution.ID, + "runID", info.WorkflowExecution.RunID, + "error", err) + return nil, err + } // Ensure ScanID is set for correlation across child workflows and snapshots // (scheduled executions pass empty ScanID)