reverting to parallelism increases only on non-terminal phase updates (…

…#624)

Signed-off-by: Daniel Rammer <daniel@union.ai>

pkg/controller/nodes/task/handler.go

@@ -536,13 +536,6 @@ func (t Handler) Handle(ctx context.Context, nCtx interfaces.NodeExecutionContex
 	ts := nCtx.NodeStateReader().GetTaskNodeState()
 
 	pluginTrns := &pluginRequestedTransition{}
-	defer func() {
-		// increment parallelism if the final pluginTrns is not in a terminal state
-		if pluginTrns != nil && !pluginTrns.pInfo.Phase().IsTerminal() {
-			eCtx := nCtx.ExecutionContext()
-			logger.Infof(ctx, "Parallelism now set to [%d].", eCtx.IncrementParallelism())
-		}
-	}()
 
 	// We will start with the assumption that catalog is disabled
 	pluginTrns.PopulateCacheInfo(catalog.NewFailedCatalogEntry(catalog.NewStatus(core.CatalogCacheStatus_CACHE_DISABLED, nil)))
@@ -667,6 +660,10 @@ func (t Handler) Handle(ctx context.Context, nCtx interfaces.NodeExecutionContex
 		return handler.UnknownTransition, err
 	}
 
+	// increment parallelism if the final pluginTrns is not in a terminal state
+	if pluginTrns != nil && !pluginTrns.pInfo.Phase().IsTerminal() {
+		logger.Infof(ctx, "Parallelism now set to [%d].", nCtx.ExecutionContext().IncrementParallelism())
+	}
 	return pluginTrns.FinalTransition(ctx)
 }