fix(plugin): Fixed batch not finishing once completed

Signed-off-by: Ruben Tordjman <[email protected]>

Author: DrRebus

engine/engine_test.go

@@ -1463,8 +1463,8 @@ func TestBatch(t *testing.T) {
 		}
 	}
 
-	// checking if the parent task is picked up after that the subtask is resolved.
-	// need to sleep a bit because the parent task is resumed asynchronously
+	// checking if the parent task is picked up after the subtask is resolved.
+	// We need to sleep a bit because the parent task is resumed asynchronously
 	ti := time.Second
 	i := time.Duration(0)
 	for i < ti {
@@ -1489,7 +1489,6 @@ func TestBatch(t *testing.T) {
 
 		time.Sleep(time.Millisecond * 10)
 		i += time.Millisecond * 10
-
 	}
 	assert.Equal(t, resolution.StateDone, res.State)
 }

pkg/plugins/builtin/batch/batch.go

@@ -236,8 +236,9 @@ func populateBatch(
 	// Computing how many tasks to start
 	remaining := int64(len(conf.Inputs)) - tasksStarted
 	toStart := int64(conf.SubBatchSize) - running // How many tasks can be started
-	if remaining < toStart {
-		toStart = remaining // There's less tasks to start remaining than the amount of available running slots
+	if remaining < toStart || conf.SubBatchSize == 0 {
+		// There's less tasks remaining to start than the amount of available running slots or slots are unlimited
+		toStart = remaining
 	}
 
 	args := batch.TaskArgs{
@@ -310,7 +311,7 @@ func increaseRunMax(dbp zesty.DBProvider, parentTaskID string, batchStepName str
 		return err
 	}
 
-	if t.Resolution != nil {
+	if t.Resolution == nil {
 		return fmt.Errorf("resolution not found for step '%s' of task '%s'", batchStepName, parentTaskID)
 	}