Add graceful termination to executor

This makes sure that the executor cancels any profile runs based on
a per execution timeout or the server itself shutting down.

cmd/server/app/serve.go

@@ -109,7 +109,7 @@ var serveCmd = &cobra.Command{
 			return fmt.Errorf("unable to create server: %w", err)
 		}
 
-		exec, err := engine.NewExecutor(store, &cfg.Auth, engine.WithProviderMetrics(providerMetrics))
+		exec, err := engine.NewExecutor(ctx, store, &cfg.Auth, engine.WithProviderMetrics(providerMetrics))
 		if err != nil {
 			return fmt.Errorf("unable to create executor: %w", err)
 		}

internal/engine/executor.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/ThreeDotsLabs/watermill/message"
 	"github.com/google/uuid"
@@ -40,12 +41,21 @@ const (
 	InternalEntityEventTopic = "internal.entity.event"
 )
 
+const (
+	// DefaultExecutionTimeout is the timeout for execution of a set
+	// of profiles on an entity.
+	DefaultExecutionTimeout = 5 * time.Minute
+)
+
 // Executor is the engine that executes the rules for a given event
 type Executor struct {
 	querier    db.Store
 	crypteng   *crypto.Engine
 	provMt     providertelemetry.ProviderMetrics
 	executions *sync.WaitGroup
+	// terminationcontext is used to terminate the executor
+	// when the server is shutting down.
+	terminationcontext context.Context
 }
 
 // ExecutorOption is a function that modifies an executor
@@ -60,6 +70,7 @@ func WithProviderMetrics(mt providertelemetry.ProviderMetrics) ExecutorOption {
 
 // NewExecutor creates a new executor
 func NewExecutor(
+	ctx context.Context,
 	querier db.Store,
 	authCfg *config.AuthConfig,
 	opts ...ExecutorOption,
@@ -70,10 +81,11 @@ func NewExecutor(
 	}
 
 	e := &Executor{
-		querier:    querier,
-		crypteng:   crypteng,
-		provMt:     providertelemetry.NewNoopMetrics(),
-		executions: &sync.WaitGroup{},
+		querier:            querier,
+		crypteng:           crypteng,
+		provMt:             providertelemetry.NewNoopMetrics(),
+		executions:         &sync.WaitGroup{},
+		terminationcontext: ctx,
 	}
 
 	for _, opt := range opts {
@@ -107,9 +119,10 @@ func (e *Executor) HandleEntityEvent(msg *message.Message) error {
 	e.executions.Add(1)
 	go func() {
 		defer e.executions.Done()
-		// TODO: use a timed context. We should use metrics to
-		//determine an appropriate timeout.
-		ctx := context.Background()
+		// TODO: Make this timeout configurable
+		ctx, cancel := context.WithTimeout(e.terminationcontext, DefaultExecutionTimeout)
+		defer cancel()
+
 		if err := e.prepAndEvalEntityEvent(ctx, inf); err != nil {
 			zerolog.Ctx(ctx).Info().
 				Str("project", inf.ProjectID.String()).

internal/engine/executor_test.go

@@ -15,6 +15,7 @@
 package engine_test
 
 import (
+	"context"
 	"encoding/base64"
 	"encoding/json"
 	"os"
@@ -235,7 +236,11 @@ default allow = true`,
 	err = os.WriteFile(tokenKeyPath, []byte(fakeTokenKey), 0600)
 	require.NoError(t, err, "expected no error")
 
-	e, err := engine.NewExecutor(mockStore, &config.AuthConfig{
+	testTimeout := 5 * time.Second
+	ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
+	defer cancel()
+
+	e, err := engine.NewExecutor(ctx, mockStore, &config.AuthConfig{
 		TokenKey: tokenKeyPath,
 	})
 	require.NoError(t, err, "expected no error")