Feat skip states for workflows

.pre-commit-config.yaml

@@ -6,7 +6,7 @@ repos:
       - id: mixed-line-ending
         args: ["--fix=lf"]
       - id: end-of-file-fixer
-        exclude: prisma/migrations/.*\.sql|sql/migrations/.*\.sql
+        exclude: prisma/migrations/.*\.sql|sql/migrations/.*\.sql|sql/schema/schema.sql
       - id: trailing-whitespace
         exclude: prisma/migrations/.*\.sql|sql/migrations/.*\.sql
       - id: check-yaml

Taskfile.yaml

@@ -59,8 +59,8 @@ tasks:
   recreate-db-from-scratch:
     cmds:
       - docker compose down
-      - docker volume rm oss_hatchet_postgres_data
-      - docker volume rm oss_hatchet_rabbitmq_data
+      - docker volume rm oss_hatchet_postgres_data || true
+      - docker volume rm oss_hatchet_rabbitmq_data || true
       - docker compose up -d
       - task: setup
       - task: init-dev-env

api/v1/server/handlers/workflows/trigger.go

@@ -12,6 +12,7 @@ import (
 	"github.com/hatchet-dev/hatchet/api/v1/server/oas/transformers"
 	"github.com/hatchet-dev/hatchet/internal/msgqueue"
 	"github.com/hatchet-dev/hatchet/internal/services/shared/tasktypes"
+	wutils "github.com/hatchet-dev/hatchet/internal/workflowutils"
 	"github.com/hatchet-dev/hatchet/pkg/repository"
 	"github.com/hatchet-dev/hatchet/pkg/repository/metered"
 	"github.com/hatchet-dev/hatchet/pkg/repository/prisma/db"
@@ -95,21 +96,37 @@ func (t *WorkflowService) WorkflowRunCreate(ctx echo.Context, request gen.Workfl
 		return nil, fmt.Errorf("trigger.go could not create workflow run: %w", err)
 	}
 
-	// send to workflow processing queue
-	err = t.config.MessageQueue.AddMessage(
-		ctx.Request().Context(),
-		msgqueue.WORKFLOW_PROCESSING_QUEUE,
-		tasktypes.WorkflowRunQueuedToTask(
-			sqlchelpers.UUIDToStr(createdWorkflowRun.TenantId),
-			sqlchelpers.UUIDToStr(createdWorkflowRun.ID),
-		),
-	)
+	if !wutils.CanShortCircuit(createdWorkflowRun.Row) {
+		// send to workflow processing queue
+		err = t.config.MessageQueue.AddMessage(
+			ctx.Request().Context(),
+			msgqueue.WORKFLOW_PROCESSING_QUEUE,
+			tasktypes.WorkflowRunQueuedToTask(
+				sqlchelpers.UUIDToStr(createdWorkflowRun.Row.WorkflowRun.TenantId),
+				sqlchelpers.UUIDToStr(createdWorkflowRun.Row.WorkflowRun.ID),
+			),
+		)
 
-	if err != nil {
-		return nil, fmt.Errorf("could not add workflow run to queue: %w", err)
+		if err != nil {
+			return nil, fmt.Errorf("could not add workflow run to queue: %w", err)
+		}
 	}
 
-	workflowRun, err := t.config.APIRepository.WorkflowRun().GetWorkflowRunById(ctx.Request().Context(), tenant.ID, sqlchelpers.UUIDToStr(createdWorkflowRun.ID))
+	for _, queueName := range createdWorkflowRun.InitialStepRunQueueNames {
+
+		if schedPartitionId, ok := tenant.SchedulerPartitionID(); ok {
+			err = t.config.MessageQueue.AddMessage(
+				ctx.Request().Context(),
+				msgqueue.QueueTypeFromPartitionIDAndController(schedPartitionId, msgqueue.Scheduler),
+				tasktypes.CheckTenantQueueToTask(tenant.ID, queueName, true, false),
+			)
+
+			if err != nil {
+				t.config.Logger.Err(err).Msg("could not add message to scheduler partition queue")
+			}
+		}
+	}
+	workflowRun, err := t.config.APIRepository.WorkflowRun().GetWorkflowRunById(ctx.Request().Context(), tenant.ID, sqlchelpers.UUIDToStr(createdWorkflowRun.Row.WorkflowRun.ID))
 
 	if err != nil {
 		return nil, fmt.Errorf("could not get workflow run: %w", err)

cmd/hatchet-engine/engine/run.go

@@ -113,7 +113,6 @@ func RunWithConfig(ctx context.Context, sc *server.ServerConfig) ([]Teardown, er
 	if isV1 {
 		return runV1Config(ctx, sc)
 	}
-
 	return runV0Config(ctx, sc)
 }
 
@@ -371,6 +370,7 @@ func runV0Config(ctx context.Context, sc *server.ServerConfig) ([]Teardown, erro
 			admin.WithRepository(sc.EngineRepository),
 			admin.WithMessageQueue(sc.MessageQueue),
 			admin.WithEntitlementsRepository(sc.EntitlementRepository),
+			admin.WithLogger(sc.Logger),
 		)
 		if err != nil {
 			return nil, fmt.Errorf("could not create admin service: %w", err)
@@ -551,7 +551,6 @@ func runV1Config(ctx context.Context, sc *server.ServerConfig) ([]Teardown, erro
 		if err != nil {
 			return nil, fmt.Errorf("could not create events controller: %w", err)
 		}
-
 		cleanup, err := ec.Start()
 
 		if err != nil {
@@ -574,7 +573,6 @@ func runV1Config(ctx context.Context, sc *server.ServerConfig) ([]Teardown, erro
 		if err != nil {
 			return nil, fmt.Errorf("could not create ticker: %w", err)
 		}
-
 		cleanup, err = t.Start()
 
 		if err != nil {
@@ -599,7 +597,6 @@ func runV1Config(ctx context.Context, sc *server.ServerConfig) ([]Teardown, erro
 		if err != nil {
 			return nil, fmt.Errorf("could not create jobs controller: %w", err)
 		}
-
 		cleanupJobs, err := jc.Start()
 
 		if err != nil {
@@ -672,7 +669,6 @@ func runV1Config(ctx context.Context, sc *server.ServerConfig) ([]Teardown, erro
 		})
 
 		wh := webhooks.New(sc, p)
-
 		cleanup2, err := wh.Start()
 
 		if err != nil {
@@ -701,7 +697,6 @@ func runV1Config(ctx context.Context, sc *server.ServerConfig) ([]Teardown, erro
 		if err != nil {
 			return nil, fmt.Errorf("could not create dispatcher: %w", err)
 		}
-
 		dispatcherCleanup, err := d.Start()
 
 		if err != nil {
@@ -732,6 +727,7 @@ func runV1Config(ctx context.Context, sc *server.ServerConfig) ([]Teardown, erro
 			admin.WithRepository(sc.EngineRepository),
 			admin.WithMessageQueue(sc.MessageQueue),
 			admin.WithEntitlementsRepository(sc.EntitlementRepository),
+			admin.WithLogger(sc.Logger),
 		)
 
 		if err != nil {
@@ -761,7 +757,6 @@ func runV1Config(ctx context.Context, sc *server.ServerConfig) ([]Teardown, erro
 		if err != nil {
 			return nil, fmt.Errorf("could not create grpc server: %w", err)
 		}
-
 		grpcServerCleanup, err := s.Start()
 		if err != nil {
 			return nil, fmt.Errorf("could not start grpc server: %w", err)
@@ -852,7 +847,6 @@ func runV1Config(ctx context.Context, sc *server.ServerConfig) ([]Teardown, erro
 	if healthProbes {
 		h.SetReady(true)
 	}
-
 	<-ctx.Done()
 
 	if healthProbes {

examples/bulk_imports/main.go

@@ -83,9 +83,9 @@ func run() (func() error, error) {
 
 	var events []client.EventWithAdditionalMetadata
 
-	// 20000 times to test the bulk push
+	// 999 (max amount) times to test the bulk push
 
-	for i := 0; i < 20000; i++ {
+	for i := 0; i < 999; i++ {
 		testEvent := userCreateEvent{
 			Username: "echo-test",
 			UserID:   "1234 " + fmt.Sprint(i),

examples/concurrency/main.go

@@ -30,27 +30,63 @@ func main() {
 	}
 
 	events := make(chan string, 50)
+	wfrIds := make(chan *client.Workflow, 50)
 	interrupt := cmdutils.InterruptChan()
+	c, err := client.New()
 
-	cleanup, err := run(events)
+	if err != nil {
+		log.Fatalf("error creating client: %v", err)
+	}
+	cleanup, err := run(c, events, wfrIds)
 	if err != nil {
 		panic(err)
 	}
+selectLoop:
+	for {
+		select {
+
+		case <-interrupt:
+			log.Print("Interrupted")
+			break selectLoop
+		case wfrId := <-wfrIds:
+			log.Printf("Workflow run id: %s", wfrId.WorkflowRunId())
+			wfResult, err := wfrId.Result()
+			if err != nil {
 
-	<-interrupt
+				if err.Error() == "step output for step-one not found" {
+					log.Printf("Step output for step-one not found because it was cancelled due to CANCELLED_BY_CONCURRENCY_LIMIT")
+					continue
+				}
+				panic(fmt.Errorf("error getting workflow run result: %w", err))
+			}
+
+			stepOneOutput := &stepOneOutput{}
+
+			err = wfResult.StepOutput("step-one", stepOneOutput)
+
+			if err != nil {
+				if err.Error() == "step run failed: this step run was cancelled due to CANCELLED_BY_CONCURRENCY_LIMIT" {
+					log.Printf("Workflow run was cancelled due to CANCELLED_BY_CONCURRENCY_LIMIT")
+					continue
+				}
+				if err.Error() == "step output for step-one not found" {
+					log.Printf("Step output for step-one not found because it was cancelled due to CANCELLED_BY_CONCURRENCY_LIMIT")
+					continue
+				}
+				panic(fmt.Errorf("error getting workflow run result: %w", err))
+			}
+		case e := <-events:
+			log.Printf("Event: %s", e)
+		}
+	}
 
 	if err := cleanup(); err != nil {
 
 		panic(fmt.Errorf("error cleaning up: %w", err))
 	}
 }
 
-func run(events chan<- string) (func() error, error) {
-	c, err := client.New()
-
-	if err != nil {
-		return nil, fmt.Errorf("error creating client: %w", err)
-	}
+func run(c client.Client, events chan<- string, wfrIds chan<- *client.Workflow) (func() error, error) {
 
 	w, err := worker.NewWorker(
 		worker.WithClient(
@@ -74,7 +110,8 @@ func run(events chan<- string) (func() error, error) {
 					err = ctx.WorkflowInput(input)
 
 					// we sleep to simulate a long running task
-					time.Sleep(10 * time.Second)
+
+					time.Sleep(7 * time.Second)
 
 					if err != nil {
 
@@ -98,7 +135,11 @@ func run(events chan<- string) (func() error, error) {
 					err = ctx.StepOutput("step-one", input)
 
 					if err != nil {
-						return nil, err
+
+						if err.Error() == "step run failed: this step run was cancelled due to CANCELLED_BY_CONCURRENCY_LIMIT" {
+							return nil, nil
+						}
+
 					}
 
 					if ctx.Err() != nil {
@@ -125,18 +166,39 @@ func run(events chan<- string) (func() error, error) {
 			"test": "test",
 		},
 	}
+
+	// I want some to be in Running and some to be in Pending so we cancel both
+
 	go func() {
-		// do this 10 times to test concurrency
-		for i := 0; i < 10; i++ {
+		// do this 7 times to test concurrency
+		for i := 0; i < 7; i++ {
 
 			wfr_id, err := c.Admin().RunWorkflow("simple-concurrency", testEvent)
 
-			log.Println("Starting workflow run id: ", wfr_id)
+			log.Println("Starting workflow run id: ", wfr_id.WorkflowRunId())
 
 			if err != nil {
 				panic(fmt.Errorf("error running workflow: %w", err))
 			}
 
+			wfrIds <- wfr_id
+			time.Sleep(400 * time.Millisecond)
+		}
+	}()
+	go func() {
+		// do this 13 times to test concurrency (20 times total)
+		for i := 0; i < 13; i++ {
+
+			wfr_id, err := c.Admin().RunWorkflow("simple-concurrency", testEvent)
+
+			log.Println("Starting workflow run id: ", wfr_id.WorkflowRunId())
+
+			if err != nil {
+				panic(fmt.Errorf("error running workflow: %w", err))
+			}
+
+			wfrIds <- wfr_id
+
 		}
 	}()