fix: update Requeue and Reassign logic to fix performance degradation when many events are queued (#310)

Logic for requeueing and reassigning did not limit the number of step runs to requeue, so when events accumulate with no worker present it causes memory to spike along with a very high query latency on the database. This commit limits the number of step runs returned in the requeue and reassign queries, and also properly locks step run rows for these queries so only a step run in a PENDING or PENDING_ASSIGNMENT state can be requeued.

It also improves performance of the `AssignStepRunToWorker` query and ensures that `maxRuns` on workers are always respected through the introduction of a `WorkerSemaphore` model. The value gets decremented when a step run is assigned and incremented when a step run is in a final state. 

Co-authored-by: Luca Steeb <[email protected]>

* Update controller.go

---------

Co-authored-by: steebchen <[email protected]>

Author: abelanger5

examples/loadtest/cli/cli_e2e_test.go

@@ -11,6 +11,8 @@ import (
 
 	"go.uber.org/goleak"
 
+	"github.com/hatchet-dev/hatchet/internal/config/shared"
+	"github.com/hatchet-dev/hatchet/internal/logger"
 	"github.com/hatchet-dev/hatchet/internal/testutils"
 )
 
@@ -22,8 +24,18 @@ func TestLoadCLI(t *testing.T) {
 		eventsPerSecond int
 		delay           time.Duration
 		wait            time.Duration
+		workerDelay     time.Duration
 		concurrency     int
 	}
+
+	l = logger.NewStdErr(
+		&shared.LoggerConfigFile{
+			Level:  "warn",
+			Format: "console",
+		},
+		"loadtest",
+	)
+
 	tests := []struct {
 		name    string
 		args    args
@@ -46,6 +58,16 @@ func TestLoadCLI(t *testing.T) {
 			wait:            30 * time.Second,
 			concurrency:     0,
 		},
+	}, {
+		name: "test for many queued events and little worker throughput",
+		args: args{
+			duration:        60 * time.Second,
+			eventsPerSecond: 100,
+			delay:           0 * time.Second,
+			workerDelay:     60 * time.Second,
+			wait:            240 * time.Second,
+			concurrency:     0,
+		},
 	}}
 
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
@@ -65,8 +87,7 @@ func TestLoadCLI(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-
-			if err := do(tt.args.duration, tt.args.eventsPerSecond, tt.args.delay, tt.args.wait, tt.args.concurrency); (err != nil) != tt.wantErr {
+			if err := do(tt.args.duration, tt.args.eventsPerSecond, tt.args.delay, tt.args.wait, tt.args.concurrency, tt.args.workerDelay); (err != nil) != tt.wantErr {
 				t.Errorf("do() error = %v, wantErr %v", err, tt.wantErr)
 			}
 		})

examples/loadtest/cli/do.go

@@ -7,8 +7,8 @@ import (
 	"time"
 )
 
-func do(duration time.Duration, eventsPerSecond int, delay time.Duration, wait time.Duration, concurrency int) error {
-	log.Printf("testing with duration=%s, eventsPerSecond=%d, delay=%s, wait=%s, concurrency=%d", duration, eventsPerSecond, delay, wait, concurrency)
+func do(duration time.Duration, eventsPerSecond int, delay time.Duration, wait time.Duration, concurrency int, workerDelay time.Duration) error {
+	l.Info().Msgf("testing with duration=%s, eventsPerSecond=%d, delay=%s, wait=%s, concurrency=%d", duration, eventsPerSecond, delay, wait, concurrency)
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -23,6 +23,11 @@ func do(duration time.Duration, eventsPerSecond int, delay time.Duration, wait t
 	ch := make(chan int64, 2)
 	durations := make(chan time.Duration, eventsPerSecond*int(duration.Seconds())*3)
 	go func() {
+		if workerDelay.Seconds() > 0 {
+			l.Info().Msgf("wait %s before starting the worker", workerDelay)
+			time.Sleep(workerDelay)
+		}
+		l.Info().Msg("starting worker now")
 		count, uniques := run(ctx, delay, durations, concurrency)
 		ch <- count
 		ch <- uniques
@@ -35,7 +40,7 @@ func do(duration time.Duration, eventsPerSecond int, delay time.Duration, wait t
 	executed := <-ch
 	uniques := <-ch
 
-	log.Printf("ℹ️ emitted %d, executed %d, uniques %d, using %d events/s", emitted, executed, uniques, eventsPerSecond)
+	l.Info().Msgf("emitted %d, executed %d, uniques %d, using %d events/s", emitted, executed, uniques, eventsPerSecond)
 
 	if executed == 0 {
 		return fmt.Errorf("❌ no events executed")
@@ -53,6 +58,7 @@ func do(duration time.Duration, eventsPerSecond int, delay time.Duration, wait t
 		totalDurationScheduled += <-scheduled
 	}
 	scheduleTimePerEvent := totalDurationScheduled / time.Duration(emitted)
+
 	log.Printf("ℹ️ average scheduling time per event: %s", scheduleTimePerEvent)
 
 	if emitted != executed {

examples/loadtest/cli/emit.go

@@ -3,7 +3,6 @@ package main
 import (
 	"context"
 	"fmt"
-	"log"
 	"sync"
 	"time"
 
@@ -34,25 +33,27 @@ func emit(ctx context.Context, amountPerSecond int, duration time.Duration, sche
 			select {
 			case <-ticker.C:
 				mx.Lock()
-				id += 1
-				mx.Unlock()
+				id++
 
 				go func(id int64) {
+					var err error
 					ev := Event{CreatedAt: time.Now(), ID: id}
-					fmt.Println("pushed event", ev.ID)
+					l.Info().Msgf("pushed event %d", ev.ID)
 					err = c.Event().Push(context.Background(), "load-test:event", ev)
 					if err != nil {
 						panic(fmt.Errorf("error pushing event: %w", err))
 					}
 					took := time.Since(ev.CreatedAt)
-					fmt.Println("pushed event", ev.ID, "took", took)
+					l.Info().Msgf("pushed event %d took %s", ev.ID, took)
 					scheduled <- took
 				}(id)
+
+				mx.Unlock()
 			case <-timer:
-				log.Println("done emitting events due to timer at", id)
+				l.Info().Msg("done emitting events due to timer")
 				return
 			case <-ctx.Done():
-				log.Println("done emitting events due to interruption at", id)
+				l.Info().Msgf("done emitting events due to interruption at %d", id)
 				return
 			}
 		}

examples/loadtest/cli/main.go

@@ -5,15 +5,23 @@ import (
 	"time"
 
 	"github.com/joho/godotenv"
+	"github.com/rs/zerolog"
 	"github.com/spf13/cobra"
+
+	"github.com/hatchet-dev/hatchet/internal/config/shared"
+	"github.com/hatchet-dev/hatchet/internal/logger"
 )
 
+var l zerolog.Logger
+
 func main() {
 	var events int
 	var concurrency int
 	var duration time.Duration
 	var wait time.Duration
 	var delay time.Duration
+	var workerDelay time.Duration
+	var logLevel string
 
 	var loadtest = &cobra.Command{
 		Use: "loadtest",
@@ -23,7 +31,15 @@ func main() {
 				panic(err)
 			}
 
-			if err := do(duration, events, delay, wait, concurrency); err != nil {
+			l = logger.NewStdErr(
+				&shared.LoggerConfigFile{
+					Level:  logLevel,
+					Format: "console",
+				},
+				"loadtest",
+			)
+
+			if err := do(duration, events, delay, wait, concurrency, workerDelay); err != nil {
 				log.Println(err)
 				panic("load test failed")
 			}
@@ -35,6 +51,8 @@ func main() {
 	loadtest.Flags().DurationVarP(&duration, "duration", "d", 10*time.Second, "duration specifies the total time to run the load test")
 	loadtest.Flags().DurationVarP(&delay, "delay", "D", 0, "delay specifies the time to wait in each event to simulate slow tasks")
 	loadtest.Flags().DurationVarP(&wait, "wait", "w", 10*time.Second, "wait specifies the total time to wait until events complete")
+	loadtest.Flags().DurationVarP(&workerDelay, "workerDelay", "p", 0*time.Second, "workerDelay specifies the time to wait before starting the worker")
+	loadtest.Flags().StringVarP(&logLevel, "level", "l", "info", "logLevel specifies the log level (debug, info, warn, error)")
 
 	cmd := &cobra.Command{Use: "app"}
 	cmd.AddCommand(loadtest)

examples/loadtest/cli/run.go

@@ -19,7 +19,9 @@ func getConcurrencyKey(ctx worker.HatchetContext) (string, error) {
 }
 
 func run(ctx context.Context, delay time.Duration, executions chan<- time.Duration, concurrency int) (int64, int64) {
-	c, err := client.New()
+	c, err := client.New(
+		client.WithLogLevel("warn"),
+	)
 
 	if err != nil {
 		panic(err)
@@ -29,6 +31,8 @@ func run(ctx context.Context, delay time.Duration, executions chan<- time.Durati
 		worker.WithClient(
 			c,
 		),
+		worker.WithLogLevel("warn"),
+		worker.WithMaxRuns(200),
 	)
 
 	if err != nil {
@@ -60,7 +64,7 @@ func run(ctx context.Context, delay time.Duration, executions chan<- time.Durati
 					}
 
 					took := time.Since(input.CreatedAt)
-					fmt.Println("executing", input.ID, "took", took)
+					l.Info().Msgf("executing %d took %s", input.ID, took)
 
 					mx.Lock()
 					executions <- took
@@ -69,13 +73,16 @@ func run(ctx context.Context, delay time.Duration, executions chan<- time.Durati
 					for i := 0; i < len(executed)-1; i++ {
 						if executed[i] == input.ID {
 							duplicate = true
-							fmt.Println("DUPLICATE:", input.ID)
+							break
 						}
 					}
+					if duplicate {
+						l.Warn().Str("step-run-id", ctx.StepRunId()).Msgf("duplicate %d", input.ID)
+					}
 					if !duplicate {
-						uniques += 1
+						uniques++
 					}
-					count += 1
+					count++
 					executed = append(executed, input.ID)
 					mx.Unlock()
 

examples/loadtest/rampup/do.go

@@ -7,10 +7,14 @@ import (
 	"slices"
 	"sync"
 	"time"
+
+	"github.com/rs/zerolog"
 )
 
+var l zerolog.Logger
+
 func do(duration time.Duration, startEventsPerSecond, amount int, increase, delay, wait, maxAcceptableDuration, maxAcceptableSchedule time.Duration, includeDroppedEvents bool, concurrency int) error {
-	log.Printf("testing with duration=%s, amount=%d, increase=%d, delay=%s, wait=%s, concurrency=%d", duration, amount, increase, delay, wait, concurrency)
+	l.Debug().Msgf("testing with duration=%s, amount=%d, increase=%d, delay=%s, wait=%s, concurrency=%d", duration, amount, increase, delay, wait, concurrency)
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -32,7 +36,7 @@ func do(duration time.Duration, startEventsPerSecond, amount int, increase, dela
 
 	go func() {
 		for s := range scheduled {
-			log.Printf("scheduled %d", s)
+			l.Debug().Msgf("scheduled %d", s)
 			idLock.Lock()
 			ids = append(ids, s)
 			idLock.Unlock()
@@ -46,7 +50,7 @@ func do(duration time.Duration, startEventsPerSecond, amount int, increase, dela
 						if includeDroppedEvents {
 							panic(fmt.Errorf("event %d did not execute in time", s))
 						} else {
-							log.Printf("warning: event %d did not execute in time", s)
+							l.Warn().Msgf("event %d did not execute in time", s)
 						}
 					}
 				}
@@ -56,7 +60,7 @@ func do(duration time.Duration, startEventsPerSecond, amount int, increase, dela
 
 	go func() {
 		for e := range executed {
-			log.Printf("executed %d", e)
+			l.Debug().Msgf("executed %d", e)
 			idLock.Lock()
 			ids = slices.DeleteFunc(ids, func(s int64) bool {
 				return s == e

examples/loadtest/rampup/emit.go

@@ -3,7 +3,6 @@ package rampup
 import (
 	"context"
 	"fmt"
-	"log"
 	"sync"
 	"time"
 
@@ -40,34 +39,36 @@ func emit(ctx context.Context, startEventsPerSecond, amount int, increase, durat
 			if eventsPerSecond < 1 {
 				eventsPerSecond = 1
 			}
-			log.Printf("emitting %d events per second", eventsPerSecond)
+			l.Debug().Msgf("emitting %d events per second", eventsPerSecond)
 			select {
 			case <-time.After(time.Second / time.Duration(eventsPerSecond)):
 				mx.Lock()
 				id += 1
-				mx.Unlock()
 
 				go func(id int64) {
+					var err error
 					ev := Event{CreatedAt: time.Now(), ID: id}
-					fmt.Println("pushed event", ev.ID)
+					l.Debug().Msgf("pushed event %d", ev.ID)
 					err = c.Event().Push(context.Background(), "load-test:event", ev)
 					if err != nil {
 						panic(fmt.Errorf("error pushing event: %w", err))
 					}
 					took := time.Since(ev.CreatedAt)
-					fmt.Println("pushed event", ev.ID, "took", took)
+					l.Debug().Msgf("pushed event %d took %s", ev.ID, took)
 
 					if took > maxAcceptableSchedule {
 						panic(fmt.Errorf("event took too long to schedule: %s at %d events/s", took, eventsPerSecond))
 					}
 
 					scheduled <- id
 				}(id)
+
+				mx.Unlock()
 			case <-timer:
-				log.Println("done emitting events due to timer at", id)
+				l.Debug().Msgf("done emitting events due to timer at %d", id)
 				return
 			case <-ctx.Done():
-				log.Println("done emitting events due to interruption at", id)
+				l.Debug().Msgf("done emitting events due to interruption at %d", id)
 				return
 			}
 		}

examples/loadtest/rampup/ramp_up_e2e_test.go

@@ -9,6 +9,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/hatchet-dev/hatchet/internal/config/shared"
+	"github.com/hatchet-dev/hatchet/internal/logger"
 	"github.com/hatchet-dev/hatchet/internal/testutils"
 )
 
@@ -30,6 +32,15 @@ func TestRampUp(t *testing.T) {
 		concurrency           int
 		startEventsPerSecond  int
 	}
+
+	l = logger.NewStdErr(
+		&shared.LoggerConfigFile{
+			Level:  "warn",
+			Format: "console",
+		},
+		"loadtest",
+	)
+
 	tests := []struct {
 		name    string
 		args    args

examples/loadtest/rampup/run.go

@@ -19,7 +19,9 @@ func getConcurrencyKey(ctx worker.HatchetContext) (string, error) {
 }
 
 func run(ctx context.Context, delay time.Duration, concurrency int, maxAcceptableDuration time.Duration, hook chan<- time.Duration, executedCh chan<- int64) (int64, int64) {
-	c, err := client.New()
+	c, err := client.New(
+		client.WithLogLevel("warn"),
+	)
 
 	if err != nil {
 		panic(err)
@@ -29,6 +31,8 @@ func run(ctx context.Context, delay time.Duration, concurrency int, maxAcceptabl
 		worker.WithClient(
 			c,
 		),
+		worker.WithLogLevel("warn"),
+		worker.WithMaxRuns(200),
 	)
 
 	if err != nil {
@@ -60,7 +64,8 @@ func run(ctx context.Context, delay time.Duration, concurrency int, maxAcceptabl
 					}
 
 					took := time.Since(input.CreatedAt)
-					fmt.Println("executing", input.ID, "took", took)
+
+					l.Debug().Msgf("executing %d took %s", input.ID, took)
 
 					if took > maxAcceptableDuration {
 						hook <- took
@@ -78,7 +83,7 @@ func run(ctx context.Context, delay time.Duration, concurrency int, maxAcceptabl
 						}
 					}
 					if duplicate {
-						fmt.Println("DUPLICATE:", input.ID)
+						l.Warn().Str("step-run-id", ctx.StepRunId()).Msgf("duplicate %d", input.ID)
 					} else {
 						uniques += 1
 					}