Respect max_concurrency in pipeline and input steps. #815.

CHANGELOG.md

@@ -5,6 +5,7 @@
 _Bug fixes_
 
 * Misleading error message for invalid step dependencies. ([#816](https://github.com/turbot/flowpipe/issues/816)).
+* Respect `max_concurrency` in `pipeline` and `input` steps. ([#815](https://github.com/turbot/flowpipe/issues/815)).
 
 ## v0.4.2 [2023-03-26]
 

internal/es/command/step_pipeline_finish.go

@@ -260,6 +260,11 @@ func (h StepPipelineFinishHandler) Handle(ctx context.Context, c interface{}) er
 		}
 	}
 
+	err = execution.ReleasePipelineExecutionStepSemaphore(cmd.PipelineExecutionID, stepDefn)
+	if err != nil {
+		return h.EventBus.Publish(ctx, event.NewPipelineFailedFromStepPipelineFinish(cmd, err))
+	}
+
 	e, err := event.NewStepFinished(event.ForPipelineStepFinish(cmd))
 	e.StepLoop = stepLoop
 

internal/es/command/step_start.go

@@ -76,6 +76,9 @@ func (h StepStartHandler) Handle(ctx context.Context, c interface{}) error {
 			if stepDefn.GetType() == schema.BlockTypePipelineStepInput {
 				slog.Debug("Step execution is an input step, not releasing semaphore", "step_name", cmd.StepName, "pipeline_execution_id", cmd.PipelineExecutionID)
 				return
+			} else if stepDefn.GetType() == schema.BlockTypePipelineStepPipeline {
+				slog.Debug("Step execution is a pipeline step, not releasing semaphore", "step_name", cmd.StepName, "pipeline_execution_id", cmd.PipelineExecutionID)
+				return
 			}
 
 			err := execution.ReleasePipelineExecutionStepSemaphore(cmd.PipelineExecutionID, stepDefn)

internal/es/estest/test_suite_mod/pipelines/lots_of_sleep.fp

@@ -16,6 +16,21 @@ pipeline "lots_of_sleep_bound" {
     }
 }
 
+pipeline "lots_of_nested_pipeline" {
+    step "pipeline" "pipeline" {
+        for_each = [1, 2, 3, 4, 5, 6, 7]
+        max_concurrency = 2
+        pipeline = pipeline.nested
+    }
+}
+
+pipeline "nested" {
+    step "transform" "transform" {
+        value = "foo"
+    }
+}
+
+
 pipeline "lots_of_sleep_bound_with_param" {
     param "concurrency" {
         default = 1

internal/service/api/form.go

@@ -349,6 +349,10 @@ func (api *APIService) finishInputStepFromForm(ex *execution.ExecutionInMemory,
 	}
 
 	err := command.EndStepFromApi(ex, stepExecution, pipelineDefn, stepDefn, &out, api.EsService.EventBus)
+	if err != nil {
+		return err
+	}
 
+	err = execution.ReleasePipelineExecutionStepSemaphore(stepExecution.PipelineExecutionID, stepDefn)
 	return err
 }

internal/service/api/integration_slack.go

@@ -390,5 +390,10 @@ func (api *APIService) finishInputStep(execId string, pExecId string, sExecId st
 		return false, nil, perr.InternalWithMessage(fmt.Sprintf("error raising step finished event: %s", err.Error()))
 	}
 
+	err = execution.ReleasePipelineExecutionStepSemaphore(stepExecution.PipelineExecutionID, stepDefn)
+	if err != nil {
+		return false, nil, perr.InternalWithMessage(fmt.Sprintf("error releasing step semaphore: %s", err.Error()))
+	}
+
 	return true, stepExecution, nil
 }