Merge pull request distribworks#383 from victorcoder/fix_349

Fix concurrency check

dkron/agent.go

@@ -445,17 +445,21 @@ func (a *Agent) eventLoop() {
 				}
 
 				if query.Name == QueryExecutionDone {
+					group := string(query.Payload)
+
 					log.WithFields(logrus.Fields{
 						"query":   query.Name,
-						"payload": string(query.Payload),
+						"payload": group,
 						"at":      query.LTime,
 					}).Debug("agent: Execution done requested")
 
 					// Find if the indicated execution is done processing
 					var err error
-					if _, ok := runningExecutions.Load(string(query.Payload)); ok {
+					if _, ok := runningExecutions.Load(group); ok {
+						log.WithField("group", group).Debug("agent: Execution is still running")
 						err = query.Respond([]byte("false"))
 					} else {
+						log.WithField("group", group).Debug("agent: Execution is not running")
 						err = query.Respond([]byte("true"))
 					}
 					if err != nil {

dkron/job.go

@@ -251,21 +251,15 @@ func (j *Job) isRunnable() bool {
 	if j.Concurrency == ConcurrencyForbid {
 		j.Agent.RefreshJobStatus(j.Name)
 	}
-	if j.Status == StatusNotSet {
-		j.Status = j.GetStatus()
-	}
-
-	if j.Status == StatusRunning {
-		if j.Concurrency == ConcurrencyAllow {
-			return true
-		} else if j.Concurrency == ConcurrencyForbid {
-			log.WithFields(logrus.Fields{
-				"job":         j.Name,
-				"concurrency": j.Concurrency,
-				"job_status":  j.Status,
-			}).Debug("scheduler: Skipping execution")
-			return false
-		}
+	j.Status = j.GetStatus()
+
+	if j.Status == StatusRunning && j.Concurrency == ConcurrencyForbid {
+		log.WithFields(logrus.Fields{
+			"job":         j.Name,
+			"concurrency": j.Concurrency,
+			"job_status":  j.Status,
+		}).Debug("scheduler: Skipping execution")
+		return false
 	}
 
 	return true