Allow for increased steps based off of identifiers (#1304)

* Allow for increased steps based off of identifiers This allows customisation of step limits dependent on various factors in the identifier. Note that defaults do not change * lints
inngest · Apr 25, 2024 · 312da2e · 312da2e
1 parent 2f11095
commit 312da2e
Show file tree

Hide file tree

Showing 3 changed files with 19 additions and 7 deletions.
diff --git a/pkg/devserver/devserver.go b/pkg/devserver/devserver.go
@@ -218,7 +218,7 @@ func start(ctx context.Context, opts StartOpts) error {
 				eventTopic: opts.Config.EventStream.Service.Concrete.TopicName(),
 			},
 		),
-		executor.WithStepLimits(consts.DefaultMaxStepLimit),
+		executor.WithStepLimits(func(id state.Identifier) int { return consts.DefaultMaxStepLimit }),
 		executor.WithInvokeNotFoundHandler(getInvokeNotFoundHandler(ctx, pb, opts.Config.EventStream.Service.Concrete.TopicName())),
 		executor.WithSendingEventHandler(getSendingEventHandler(ctx, pb, opts.Config.EventStream.Service.Concrete.TopicName())),
 		executor.WithDebouncer(debouncer),

diff --git a/pkg/execution/executor/executor.go b/pkg/execution/executor/executor.go
@@ -168,11 +168,8 @@ func WithLifecycleListeners(l ...execution.LifecycleListener) ExecutorOpt {
 	}
 }
 
-func WithStepLimits(limit uint) ExecutorOpt {
+func WithStepLimits(limit func(id state.Identifier) int) ExecutorOpt {
 	return func(e execution.Executor) error {
-		if limit > consts.AbsoluteMaxStepLimit {
-			return fmt.Errorf("%d is greater than the absolute step limit of %d", limit, consts.AbsoluteMaxStepLimit)
-		}
 		e.(*executor).steplimit = limit
 		return nil
 	}
@@ -241,7 +238,7 @@ type executor struct {
 
 	lifecycles []execution.LifecycleListener
 
-	steplimit uint
+	steplimit func(id state.Identifier) int
 }
 
 func (e *executor) SetFinishHandler(f execution.FinishHandler) {

diff --git a/pkg/execution/executor/validate.go b/pkg/execution/executor/validate.go
@@ -2,8 +2,10 @@ package executor
 
 import (
 	"context"
+	"fmt"
 	"time"
 
+	"github.com/inngest/inngest/pkg/consts"
 	"github.com/inngest/inngest/pkg/enums"
 	"github.com/inngest/inngest/pkg/execution"
 	"github.com/inngest/inngest/pkg/execution/queue"
@@ -63,7 +65,20 @@ func (r *runValidator) checkCancelled(ctx context.Context) error {
 }
 
 func (r *runValidator) checkStepLimit(ctx context.Context) error {
-	if r.e.steplimit != 0 && len(r.s.Actions()) >= int(r.e.steplimit) {
+	var limit int
+
+	if r.e.steplimit != nil {
+		limit = r.e.steplimit(r.item.Identifier)
+	}
+
+	if limit == 0 {
+		limit = consts.DefaultMaxStepLimit
+	}
+	if limit > consts.AbsoluteMaxStepLimit {
+		return fmt.Errorf("%d is greater than the absolute step limit of %d", limit, consts.AbsoluteMaxStepLimit)
+	}
+
+	if limit > 0 && len(r.s.Actions()) >= limit {
 		// Update this function's state to overflowed, if running.
 		if r.md.Status == enums.RunStatusRunning {
 			// XXX: Update error to failed, set error message