Workflow Behavior Change: Sub-job queuing now considers each workflow job to be its own queue "namespace". Fixes #218.

Author: jhuckaby

lib/job.js

@@ -264,8 +264,12 @@ class Jobs {
 			self.monitorJob(job);
 
 			if (job.state == 'queued') {
-				var queue_id = job.event;
-				if (job.workflow && job.workflow.node && (job.type == 'adhoc')) queue_id += '-' + job.workflow.node;
+				// build up a job queue id using same logic as findSimilarJobs
+				var parts = [ job.event ];
+				if (job.workflow && job.workflow.node) parts = [ job.workflow.node ]; // replace event
+				if (job.workflow && job.workflow.job) parts.push( job.workflow.job );
+				if (job.type) parts.push( job.type );
+				var queue_id = parts.join('-');
 
 				if (!queues[queue_id]) queues[queue_id] = [];
 				queues[queue_id].push(job);
@@ -275,25 +279,14 @@ class Jobs {
 		// check queues for available slots
 		for (var queue_id in queues) {
 			var queued = queues[queue_id];
-			// var event = Tools.findObject( this.events, { id: event_id } );
-			// if (!event) continue; // sanity
 
-			var criteria = { state: 'active' };
-			if (queue_id.match(/^(\w+)\-(\w+)$/)) {
-				// support adhoc workflow jobs (where the event is the WF event)
-				var event_id = RegExp.$1;
-				var node_id = RegExp.$2;
-				criteria['event'] = event_id;
-				criteria['workflow.node'] = node_id;
-			}
-			else {
-				criteria['event'] = queue_id;
-			}
-			
-			// use limit definition from first queued job (honor system)
+			// use limit definition from first queued job
+			// (it SHOULD match the rest of the jobs in the queue group)
 			var job = queued[0];
 			var job_limit = Tools.findObject( job.limits || [], { type: 'job', enabled: true } );
-			var jobs = this.findActiveJobsDeep(criteria);
+			var jobs = this.findSimilarJobs(job, { state: 'active' }).concat(
+				this.findSimilarJobs(job, { state: 'starting' })
+			);
 
 			if (job_limit && (jobs.length < job_limit.amount)) {
 				// we have room!  launch queued job waiting the longest
@@ -318,7 +311,8 @@ class Jobs {
 					this.masterSync();
 				}
 
-				// FUTURE NOTE: This will only launch a max of 1 queued job per event per second (per tick)
+				// FUTURE NOTE: This will only launch a max of 1 queued job per group per second (per tick)
+				// However, monitorJobs is also called when jobs complete, so that compensates
 			}
 		} // foreach queue
 	}
@@ -444,8 +438,19 @@ class Jobs {
 		// find jobs with criteria and same event, OR same wf+node if adhoc
 		criteria['event'] = job.event;
 
-		if (job.workflow && job.workflow.node && (job.type == 'adhoc')) {
+		// include job type if applicable
+		if (job.type) criteria['type'] = job.type;
+		
+		// for wf nodes use the node id as criteria (replace event)
+		if (job.workflow && job.workflow.node) {
 			criteria['workflow.node'] = job.workflow.node;
+			delete criteria['event'];
+		}
+		
+		// if running inside a workflow, include the parent workflow job id itself as criteria,
+		// so that limits are applied in a "single workflow execution run scope" (see gh issue # 218)
+		if (job.workflow && job.workflow.job) {
+			criteria['workflow.job'] = job.workflow.job;
 		}
 
 		return Tools.findObjectsDeep( Object.values(this.activeJobs), criteria );
@@ -890,7 +895,7 @@ class Jobs {
 			var queue_limit = Tools.findObject( job.limits, { type: 'queue', enabled: true } );
 
 			if (queue_limit && queue_limit.amount) {
-				var queued = this.findActiveJobs({ event: job.event, state: 'queued' });
+				var queued = this.findSimilarJobs(job, { state: 'queued' });
 
 				if (queued.length < queue_limit.amount) {
 					// room in queue, yay!