fix: eliminate RetrieveFindings activity to avoid Temporal 4MB payload limit

The RetrieveFindings activity returned all findings as a single Temporal
activity result, which exceeded the 4MB gRPC message limit for large
inventories (12K+ Aurora clusters = ~10MB serialized).

Instead of passing findings through Temporal payloads, CreateSnapshot now
reads directly from the in-memory store. Findings stay within the worker
process and never transit Temporal's gRPC layer.

- Remove RetrieveFindings activity and its registration
- CreateSnapshot accepts ResourceTypes and reads from store itself
- Orchestrator passes successful resource types instead of findings
- All detection workflows still run in parallel via child workflows

Amp-Thread-ID: https://ampcode.com/threads/T-019d92b6-b80d-731a-8a83-64e6442ae52c
Co-authored-by: Amp <amp@ampcode.com>

Author: bakayolo

cmd/server/main.go

@@ -329,13 +329,9 @@ func (s *ServerCLI) Run(_ *kong.Context) error {
 	// Orchestrator workflow activities
 	if snapshotStore != nil {
 		orchestratorActivities := orchestrator.NewActivities(st, snapshotStore)
-		w.RegisterActivityWithOptions(orchestratorActivities.RetrieveFindings, activity.RegisterOptions{Name: orchestrator.RetrieveFindingsActivityName})
 		w.RegisterActivityWithOptions(orchestratorActivities.CreateSnapshot, activity.RegisterOptions{Name: orchestrator.CreateSnapshotActivityName})
 		fmt.Println("✓ Orchestrator activities registered (with S3)")
 	} else {
-		// Without S3, we can still retrieve findings but can't create snapshots
-		orchestratorActivities := orchestrator.NewActivities(st, nil)
-		w.RegisterActivityWithOptions(orchestratorActivities.RetrieveFindings, activity.RegisterOptions{Name: orchestrator.RetrieveFindingsActivityName})
 		fmt.Println("⚠️  Orchestrator snapshot activity not registered (no S3 store)")
 	}
 

pkg/workflow/orchestrator/activities.go

@@ -13,21 +13,16 @@ import (
 
 // Activity names
 const (
-	RetrieveFindingsActivityName = "version-guard.RetrieveFindings"
-	CreateSnapshotActivityName   = "version-guard.CreateSnapshot"
+	CreateSnapshotActivityName = "version-guard.CreateSnapshot"
 )
 
 // Activity input/output types
 
-type RetrieveFindingsInput struct {
-	ResourceType types.ResourceType
-}
-
 type CreateSnapshotInput struct {
-	ScanID         string
-	FindingsByType map[types.ResourceType][]*types.Finding
-	ScanStartTime  time.Time
-	ScanEndTime    time.Time
+	ScanID        string
+	ResourceTypes []types.ResourceType
+	ScanStartTime time.Time
+	ScanEndTime   time.Time
 }
 
 type SnapshotResult struct {
@@ -57,34 +52,27 @@ func NewActivities(
 	}
 }
 
-// RetrieveFindings retrieves all findings for a given resource type from the store
-func (a *Activities) RetrieveFindings(ctx context.Context, input RetrieveFindingsInput) ([]*types.Finding, error) {
-	logger := activity.GetLogger(ctx)
-	logger.Info("Retrieving findings from store", "resourceType", input.ResourceType)
-
-	filters := store.FindingFilters{
-		ResourceType: &input.ResourceType,
-	}
-
-	findings, err := a.Store.ListFindings(ctx, filters)
-	if err != nil {
-		return nil, err
-	}
-
-	logger.Info("Findings retrieved", "count", len(findings))
-	return findings, nil
-}
-
-// CreateSnapshot creates a snapshot from findings and persists it to S3
+// CreateSnapshot reads findings directly from the store and persists a snapshot to S3.
+// This avoids passing large finding payloads through Temporal activity results,
+// which would exceed the 4MB gRPC message limit for large inventories (12K+ resources).
 func (a *Activities) CreateSnapshot(ctx context.Context, input CreateSnapshotInput) (*SnapshotResult, error) {
 	logger := activity.GetLogger(ctx)
-	logger.Info("Creating snapshot", "scanID", input.ScanID, "resourceTypeCount", len(input.FindingsByType))
+	logger.Info("Creating snapshot", "scanID", input.ScanID, "resourceTypeCount", len(input.ResourceTypes))
 
-	// Build snapshot
+	// Build snapshot by reading findings directly from the store per resource type
 	builder := snapshot.NewBuilder()
 	builder.WithScanTiming(input.ScanStartTime, input.ScanEndTime)
 
-	for resourceType, findings := range input.FindingsByType {
+	for _, resourceType := range input.ResourceTypes {
+		rt := resourceType
+		findings, err := a.Store.ListFindings(ctx, store.FindingFilters{
+			ResourceType: &rt,
+		})
+		if err != nil {
+			logger.Warn("Failed to retrieve findings for snapshot", "resourceType", resourceType, "error", err)
+			continue
+		}
+		logger.Info("Retrieved findings for snapshot", "resourceType", resourceType, "count", len(findings))
 		builder.AddFindings(resourceType, findings)
 	}
 
@@ -113,9 +101,6 @@ func (a *Activities) CreateSnapshot(ctx context.Context, input CreateSnapshotInp
 func RegisterActivities(worker interface {
 	RegisterActivityWithOptions(interface{}, activity.RegisterOptions)
 }, activities *Activities) {
-	worker.RegisterActivityWithOptions(activities.RetrieveFindings, activity.RegisterOptions{
-		Name: RetrieveFindingsActivityName,
-	})
 	worker.RegisterActivityWithOptions(activities.CreateSnapshot, activity.RegisterOptions{
 		Name: CreateSnapshotActivityName,
 	})

pkg/workflow/orchestrator/workflow.go

@@ -108,7 +108,7 @@ func OrchestratorWorkflow(ctx workflow.Context, input WorkflowInput) (*WorkflowO
 
 	// Wait for all child workflows to complete and collect results
 	resourceTypeResults := make(map[types.ResourceType]*ResourceTypeResult)
-	allFindings := make(map[types.ResourceType][]*types.Finding)
+	var successfulTypes []types.ResourceType
 
 	for resourceType, future := range futures {
 		var output detectionWorkflow.WorkflowOutput
@@ -136,27 +136,10 @@ func OrchestratorWorkflow(ctx workflow.Context, input WorkflowInput) (*WorkflowO
 		}
 
 		resourceTypeResults[resourceType] = result
-
-		// Retrieve findings from store for snapshot creation
-		var retrievedFindings []*types.Finding
-		err = workflow.ExecuteActivity(
-			workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
-				StartToCloseTimeout: 2 * time.Minute,
-				RetryPolicy:         retryPolicy,
-			}),
-			RetrieveFindingsActivityName,
-			RetrieveFindingsInput{ResourceType: resourceType},
-		).Get(ctx, &retrievedFindings)
-
-		if err != nil {
-			logger.Warn("Failed to retrieve findings for snapshot", "resourceType", resourceType, "error", err)
-			continue
-		}
-
-		allFindings[resourceType] = retrievedFindings
+		successfulTypes = append(successfulTypes, resourceType)
 	}
 
-	logger.Info("Stage 1: Detect - All detection workflows completed", "successCount", len(allFindings))
+	logger.Info("Stage 1: Detect - All detection workflows completed", "successCount", len(successfulTypes))
 
 	// Stage 2: STORE - Create and persist snapshot to S3
 	logger.Info("Stage 2: Store - Creating snapshot")
@@ -169,10 +152,10 @@ func OrchestratorWorkflow(ctx workflow.Context, input WorkflowInput) (*WorkflowO
 		}),
 		CreateSnapshotActivityName,
 		CreateSnapshotInput{
-			ScanID:         input.ScanID,
-			FindingsByType: allFindings,
-			ScanStartTime:  startTime,
-			ScanEndTime:    workflow.Now(ctx),
+			ScanID:        input.ScanID,
+			ResourceTypes: successfulTypes,
+			ScanStartTime: startTime,
+			ScanEndTime:   workflow.Now(ctx),
 		},
 	).Get(ctx, &snapshotResult)