feat: make executor event handling timeout configurable

internal/db/profiles_test.go

@@ -497,8 +497,8 @@ func TestProfileLabels(t *testing.T) {
 	}
 }
 
+//nolint:paralleltest,tparallel // top-level parallel causes interference with other tests (gomock/shared state)
 func TestCreateProfileStatusSingleRuleTransitions(t *testing.T) {
-	t.Parallel()
 
 	tests := []struct {
 		name                      string

internal/engine/handler.go

@@ -25,39 +25,53 @@ const (
 	// DefaultExecutionTimeout is the timeout for execution of a set
 	// of profiles on an entity.
 	DefaultExecutionTimeout = 5 * time.Minute
+
 	// ArtifactSignatureWaitPeriod is the waiting period for potential artifact signature to be available
 	// before proceeding with evaluation.
 	ArtifactSignatureWaitPeriod = 10 * time.Second
 )
 
-// ExecutorEventHandler is responsible for consuming entity events, passing
-// entities to the executor, and then publishing the results.
+// ExecutorEventHandler handles entity events, executes evaluations,
+// and publishes the results.
 type ExecutorEventHandler struct {
 	evt                    interfaces.Publisher
 	handlerMiddleware      []message.HandlerMiddleware
 	wgEntityEventExecution *sync.WaitGroup
 	executor               Executor
-	// cancels are a set of cancel functions for current entity events in flight.
-	// This allows us to cancel rule evaluation directly when terminationContext
-	// is cancelled.
+
+	executionTimeout time.Duration
+
 	cancels []*context.CancelFunc
 	lock    sync.Mutex
 	closed  bool
 }
 
-// NewExecutorEventHandler creates the event handler for the executor
+// NewExecutorEventHandler creates a new ExecutorEventHandler with
+// configurable execution timeout.
 func NewExecutorEventHandler(
 	ctx context.Context,
 	evt interfaces.Publisher,
 	handlerMiddleware []message.HandlerMiddleware,
 	executor Executor,
+	executionTimeout time.Duration,
 ) *ExecutorEventHandler {
+
+	if executionTimeout <= 0 {
+		executionTimeout = DefaultExecutionTimeout
+	}
+
 	eh := &ExecutorEventHandler{
 		evt:                    evt,
 		wgEntityEventExecution: &sync.WaitGroup{},
 		handlerMiddleware:      handlerMiddleware,
 		executor:               executor,
+		executionTimeout:       executionTimeout,
 	}
+
+	zerolog.Ctx(ctx).Debug().
+		Dur("execution_timeout", executionTimeout).
+		Msg("executor event handler initialized")
+
 	go func() {
 		<-ctx.Done()
 		eh.lock.Lock()
@@ -73,30 +87,31 @@ func NewExecutorEventHandler(
 	return eh
 }
 
-// Register implements the Consumer interface.
+// Register registers the handler for entity evaluation events.
 func (e *ExecutorEventHandler) Register(r interfaces.Registrar) {
 	r.Register(constants.TopicQueueEntityEvaluate, e.HandleEntityEvent, e.handlerMiddleware...)
 }
 
-// Wait waits for all the entity executions to finish.
+// Wait blocks until all entity event executions are complete.
 func (e *ExecutorEventHandler) Wait() {
 	e.wgEntityEventExecution.Wait()
 }
 
-// HandleEntityEvent handles events coming from webhooks/signals
-// as well as the init event.
+// HandleEntityEvent processes incoming entity events and triggers evaluation.
 func (e *ExecutorEventHandler) HandleEntityEvent(msg *message.Message) error {
 
 	// NOTE: we're _deliberately_ "escaping" from the parent context's Cancel/Done
 	// completion, because the default watermill behavior for both Go channels and
 	// SQL is to process messages sequentially, but we need additional parallelism
-	// beyond that.  When we switch to a different message processing system, we
+	// beyond that. When we switch to a different message processing system, we
 	// should aim to remove this goroutine altogether and have the messaging system
 	// provide the parallelism.
 	// We _do_ still want to cancel on shutdown, however.
-	// TODO: Make this timeout configurable
 	msgCtx := context.WithoutCancel(msg.Context())
-	//nolint:gosec // this is called when we iterate over e.cancels
+
+	// This allows us to cancel rule evaluation directly when terminationContext
+	// is cancelled.
+	//nolint:gosec
 	msgCtx, shutdownCancel := context.WithCancel(msgCtx)
 
 	e.lock.Lock()
@@ -108,7 +123,8 @@ func (e *ExecutorEventHandler) HandleEntityEvent(msg *message.Message) error {
 	e.cancels = append(e.cancels, &shutdownCancel)
 	e.lock.Unlock()
 
-	// Let's not share memory with the caller.  Note that this does not copy Context
+	// Copy the message so the async goroutine does not share mutable state with
+	// the original Watermill message.
 	msg = msg.Copy()
 
 	inf, err := entities.ParseEntityEvent(msg)
@@ -117,8 +133,10 @@ func (e *ExecutorEventHandler) HandleEntityEvent(msg *message.Message) error {
 	}
 
 	e.wgEntityEventExecution.Add(1)
+
 	go func() {
 		defer e.wgEntityEventExecution.Done()
+
 		if inf.Type == pb.Entity_ENTITY_ARTIFACTS {
 			// Wait for artifact signatures, but allow early exit on shutdown
 			select {
@@ -128,8 +146,9 @@ func (e *ExecutorEventHandler) HandleEntityEvent(msg *message.Message) error {
 			}
 		}
 
-		ctx, cancel := context.WithTimeout(msgCtx, DefaultExecutionTimeout)
+		ctx, cancel := context.WithTimeout(msgCtx, e.executionTimeout)
 		defer cancel()
+
 		defer func() {
 			e.lock.Lock()
 			e.cancels = slices.DeleteFunc(e.cancels, func(cf *context.CancelFunc) bool {
@@ -140,30 +159,34 @@ func (e *ExecutorEventHandler) HandleEntityEvent(msg *message.Message) error {
 
 		ctx = engcontext.WithEntityContext(ctx, &engcontext.EntityContext{
 			Project: engcontext.Project{ID: inf.ProjectID},
-			// TODO: extract Provider name from ProviderID?
+			Provider: engcontext.Provider{
+				Name: inf.ProviderID.String(),
+			},
 		})
 
 		ts := minderlogger.BusinessRecord(ctx)
 		ctx = ts.WithTelemetry(ctx)
 
 		logger := zerolog.Ctx(ctx)
+
 		if err := inf.WithExecutionIDFromMessage(msg); err != nil {
-			logger.Info().
+			logger.Debug().
 				Str("message_id", msg.UUID).
 				Msg("message does not contain execution ID, skipping")
 			return
 		}
 
 		err := e.executor.EvalEntityEvent(ctx, inf)
 
-		// record telemetry regardless of error. We explicitly record telemetry
-		// here even though we also record it in the middleware because the evaluation
-		// is done in a separate goroutine which usually still runs after the middleware
-		// had already recorded the telemetry.
 		logMsg := logger.Info()
 		if err != nil {
 			logMsg = logger.Error()
 		}
+
+		// record telemetry regardless of error. We explicitly record telemetry
+		// here even though we also record it in the middleware because the evaluation
+		// is done in a separate goroutine which usually still runs after the middleware
+		// had already recorded the telemetry.
 		ts.Record(logMsg).Send()
 
 		if err != nil {
@@ -172,18 +195,16 @@ func (e *ExecutorEventHandler) HandleEntityEvent(msg *message.Message) error {
 				Str("provider_id", inf.ProviderID.String()).
 				Str("entity", inf.Type.String()).
 				Str("entity_id", inf.EntityID.String()).
-				Err(err).Msg("got error while evaluating entity event")
+				Err(err).
+				Msg("got error while evaluating entity event")
 		}
 
-		// We don't need to unset the execution ID because the event is going to be
-		// deleted from the database anyway. The aggregator will take care of that.
 		msg, err := inf.BuildMessage()
 		if err != nil {
 			logger.Err(err).Msg("error building message")
 			return
 		}
 
-		// Publish the result of the entity evaluation
 		if err := e.evt.Publish(constants.TopicQueueEntityFlush, msg); err != nil {
 			logger.Err(err).Msg("error publishing flush event")
 		}