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

@@ -107,7 +107,7 @@ var serveCmd = &cobra.Command{
 			return fmt.Errorf("unable to create server: %w", err)
 		}
 
-		exec, err := engine.NewExecutor(store, &cfg.Auth)
+		exec, err := engine.NewExecutor(ctx, store, &cfg.Auth)
 		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"
@@ -39,24 +40,34 @@ 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
 	executions *sync.WaitGroup
+	// terminationcontext is used to terminate the executor
+	// when the server is shutting down.
+	terminationcontext context.Context
 }
 
 // NewExecutor creates a new executor
-func NewExecutor(querier db.Store, authCfg *config.AuthConfig) (*Executor, error) {
+func NewExecutor(ctx context.Context, querier db.Store, authCfg *config.AuthConfig) (*Executor, error) {
 	crypteng, err := crypto.EngineFromAuthConfig(authCfg)
 	if err != nil {
 		return nil, err
 	}
 
 	return &Executor{
-		querier:    querier,
-		crypteng:   crypteng,
-		executions: &sync.WaitGroup{},
+		querier:            querier,
+		crypteng:           crypteng,
+		executions:         &sync.WaitGroup{},
+		terminationcontext: ctx,
 	}, nil
 }
 
@@ -84,9 +95,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")