feat(job): run job schedules from the worker poll and JobRunner.tick()

changelog.d/schedule-runner-wiring.added.md

@@ -0,0 +1 @@
+- Job schedules (`config/schedules.cfm`, `schedule()`) now run from the normal job processing: every worker poll (`wheels jobs work`), `processQueue()` and `JobRunner.tick()` enqueue due schedule slots, at most once every `jobsScheduleCheckSeconds` (default 15; `0` turns it off), so no separate scheduler process is needed. Any number of servers can do it at once: each slot is still enqueued exactly once. An app without schedules is unaffected: the check only runs when the schedules table exists. `JobRunner.tick()` reports the slots it enqueued as `scheduled`.

vendor/wheels/Job.cfc

@@ -1739,6 +1739,60 @@ component {
 		return {type = Len(arguments.type) ? arguments.type : "Wheels.JobFailed", message = arguments.message, detail = arguments.detail};
 	}
 
+	/**
+	 * Internal: the JobScheduler $enqueueDueSchedules() runs, a seam for specs.
+	 */
+	public any function $newScheduler() {
+		return new wheels.JobScheduler();
+	}
+
+	/**
+	 * Enqueues due schedules (JobScheduler.enqueueDue()) at most once every
+	 * jobsScheduleCheckSeconds (default 15; 0 = never) per application, so every worker poll,
+	 * processQueue() and JobRunner.tick() keeps schedules running without a separate process.
+	 * Any number of servers may do it: each slot is enqueued once (its uniqueKey). Only when the
+	 * schedules table exists, so an app without schedules never gets one. Never throws: a failed
+	 * check is logged and the poll carries on. Returns {checked, enqueued} (both 0 when skipped).
+	 */
+	public struct function $enqueueDueSchedules() {
+		var rv = {checked = 0, enqueued = 0};
+		if (!StructKeyExists(application, "wheels")) {
+			return rv;
+		}
+		local.interval = StructKeyExists(application.wheels, "jobsScheduleCheckSeconds") && IsNumeric(application.wheels.jobsScheduleCheckSeconds) ? Val(application.wheels.jobsScheduleCheckSeconds) : 15;
+		if (local.interval <= 0 || Len($outermostWheelsTransaction())) {
+			return rv;
+		}
+		local.now = GetTickCount();
+		if (StructKeyExists(application.wheels, "$jobsScheduleCheckedAt") && local.now - application.wheels["$jobsScheduleCheckedAt"] < local.interval * 1000) {
+			return rv;
+		}
+		// One check at a time per application; a poll that finds one running skips.
+		var gate = {due = false};
+		lock name="wheels.jobs.scheduleCheck" type="exclusive" timeout="1" throwOnTimeout="false" {
+			gate.due = !StructKeyExists(application.wheels, "$jobsScheduleCheckedAt") || GetTickCount() - application.wheels["$jobsScheduleCheckedAt"] >= local.interval * 1000;
+			if (gate.due) {
+				application.wheels["$jobsScheduleCheckedAt"] = GetTickCount();
+			}
+		}
+		if (gate.due) {
+			try {
+				local.scheduler = $newScheduler();
+				if (local.scheduler.$schedulesTableExists()) {
+					local.outcome = local.scheduler.enqueueDue();
+					rv.checked = local.outcome.checked;
+					rv.enqueued = local.outcome.enqueued;
+					for (local.error in local.outcome.errors) {
+						writeLog(text = "Schedule check: #local.error#", type = "error", file = "wheels_jobs");
+					}
+				}
+			} catch (any e) {
+				writeLog(text = "Could not check job schedules: #e.message#", type = "error", file = "wheels_jobs");
+			}
+		}
+		return rv;
+	}
+
 	/**
 	 * Throws Wheels.Job.SchemaMissing: a job table or column is missing while jobsAutoCreateTables
 	 * is false, so the framework won't create it. Logged once per application as well, because a

vendor/wheels/JobRunner.cfc

@@ -34,11 +34,13 @@ component {
 		variables.$job.$ensureHostsTable();
 		local.host = variables.$job.$jobHostName();
 		local.cap = arguments.maxConcurrent >= 0 ? Int(arguments.maxConcurrent) : variables.$job.$jobsMaxConcurrentPerHost();
-		local.rv = {host = local.host, processed = 0, failed = 0, fenced = 0, reaped = 0, capped = false, draining = false};
+		local.rv = {host = local.host, processed = 0, failed = 0, fenced = 0, reaped = 0, scheduled = 0, capped = false, draining = false};
 
 		local.worker = new wheels.JobWorker();
 		local.worker.maxConcurrentPerHost = local.cap;
 		local.rv.reaped = local.worker.checkTimeouts(timeout = arguments.timeout, queues = arguments.queues);
+		// Due schedules first, so a slot that is due now can run in this same tick.
+		local.rv.scheduled = variables.$job.$enqueueDueSchedules().enqueued;
 
 		local.limit = Max(1, Int(Val(arguments.maxJobs)));
 		for (local.i = 1; local.i <= local.limit; local.i++) {

vendor/wheels/JobWorker.cfc

@@ -62,6 +62,9 @@ component {
 		// nothing is stuck.
 		this.checkTimeouts(timeout = arguments.timeout, queues = arguments.queues);
 
+		// Keep schedules running from the same poll (throttled to jobsScheduleCheckSeconds).
+		$jobBridge().$enqueueDueSchedules();
+
 		// A draining host starts nothing new (in-flight jobs finish; the reap above still runs).
 		local.host = $jobBridge().$jobHostName();
 		if ($jobBridge().$hostDraining(local.host)) {

vendor/wheels/events/onapplicationstart.cfc

@@ -348,6 +348,9 @@ component {
 		// one at a time. A waiting instance gives up after migrationLockTimeout seconds; a holder
 		// that died frees the lock once its lease (renewed before each step) runs out.
 		application.$wheels.migratorLockTableName = "wheels_migrator_locks";
+		// Job schedules (config/schedules.cfm) are checked and their due slots enqueued at most
+		// this often, from every worker poll, processQueue() and JobRunner.tick(). 0 = never.
+		application.$wheels.jobsScheduleCheckSeconds = 15;
 		// Background-job tables: true creates and upgrades them automatically at runtime; false
 		// leaves them to the app's own migration (`wheels jobs install`), for multi-server apps.
 		application.$wheels.jobsAutoCreateTables = true;

vendor/wheels/tests/specs/jobs/JobScheduleRunnerSpec.cfc

@@ -0,0 +1,132 @@
+/**
+ * Schedules run from the normal job processing: every worker poll (wheels jobs work,
+ * processQueue()) and JobRunner.tick() enqueue due schedule slots, at most once every
+ * jobsScheduleCheckSeconds per application. No separate scheduler process is needed.
+ */
+component extends="wheels.WheelsTest" {
+
+	function run() {
+
+		describe("schedules from the job runner", function() {
+
+			beforeEach(function() {
+				new wheels.Job().$ensureJobTable();
+				new wheels.JobScheduler().$ensureSchedulesTable();
+				$cleanup();
+				application.wheels.jobsScheduleCheckSeconds = 15;
+				StructDelete(application.wheels, "$jobsScheduleCheckedAt");
+			});
+
+			afterEach(function() {
+				$cleanup();
+				application.wheels.jobsScheduleCheckSeconds = 15;
+				StructDelete(application.wheels, "$jobsScheduleCheckedAt");
+			});
+
+			it("enqueues a due schedule from a worker poll", function() {
+				$insertDueSchedule("spec_runner_poll");
+				new wheels.JobWorker().processNext(queues = "test_srun_poll", timeout = 300);
+				expect($jobsFor("spec_runner_poll")).toBe(1, "the poll enqueued the due slot");
+			});
+
+			it("reports the slots it enqueued in a tick", function() {
+				$insertDueSchedule("spec_runner_tick");
+				var result = new wheels.JobRunner().tick(queues = "test_srun_poll");
+				expect(result.scheduled).toBe(1);
+				expect($jobsFor("spec_runner_tick")).toBe(1);
+			});
+
+			it("checks at most once per jobsScheduleCheckSeconds", function() {
+				application.wheels.jobsScheduleCheckSeconds = 3600;
+				$insertDueSchedule("spec_runner_throttle");
+				var job = new wheels.Job();
+				expect(job.$enqueueDueSchedules().enqueued).toBe(1);
+				$insertDueSchedule("spec_runner_throttle_2");
+				expect(job.$enqueueDueSchedules().checked).toBe(0, "the next check waits for the interval");
+				expect($jobsFor("spec_runner_throttle_2")).toBe(0);
+			});
+
+			it("carries on when the schedule check throws", function() {
+				var job = new wheels.Job();
+				prepareMock(job);
+				var scheduler = createStub();
+				scheduler.$("$schedulesTableExists", true);
+				scheduler.$(method = "enqueueDue", throwException = true, throwType = "Spec.ScheduleBoom", throwMessage = "check failed");
+				job.$("$newScheduler", scheduler);
+				var result = job.$enqueueDueSchedules();
+				expect(result.checked).toBe(0);
+				expect(result.enqueued).toBe(0);
+			});
+
+			it("doesn't check, or create the table, when there are no schedules", function() {
+				var job = new wheels.Job();
+				prepareMock(job);
+				var scheduler = createStub();
+				scheduler.$("$schedulesTableExists", false);
+				scheduler.$("enqueueDue", {checked = 9, enqueued = 9, errors = []});
+				job.$("$newScheduler", scheduler);
+				expect(job.$enqueueDueSchedules().enqueued).toBe(0);
+				expect(scheduler.$count("enqueueDue")).toBe(0);
+			});
+
+			it("still enqueues due slots from a draining server's tick", function() {
+				$insertDueSchedule("spec_runner_drain");
+				var runner = new wheels.JobRunner();
+				runner.drain();
+				try {
+					var result = runner.tick(queues = "test_srun_poll");
+					expect(result.draining).toBeTrue();
+					expect(result.scheduled).toBe(1, "a slot enqueued by a draining server is run by another one");
+				} finally {
+					runner.resume();
+				}
+				expect($jobsFor("spec_runner_drain")).toBe(1);
+			});
+
+			it("never checks with jobsScheduleCheckSeconds = 0", function() {
+				application.wheels.jobsScheduleCheckSeconds = 0;
+				$insertDueSchedule("spec_runner_off");
+				new wheels.JobWorker().processNext(queues = "test_srun_poll", timeout = 300);
+				expect($jobsFor("spec_runner_off")).toBe(0);
+			});
+
+		});
+	}
+
+	private void function $cleanup() {
+		try {
+			queryExecute("DELETE FROM wheels_job_schedules WHERE name LIKE 'spec_runner_%'", {}, {datasource = application.wheels.dataSourceName});
+		} catch (any e) {
+		}
+		try {
+			queryExecute("DELETE FROM wheels_jobs WHERE queue = 'test_srun_jobs' OR queue = 'test_srun_poll'", {}, {datasource = application.wheels.dataSourceName});
+		} catch (any e) {
+		}
+	}
+
+	/**
+	 * An every-minute schedule whose last slot was five minutes ago, so a slot is due now. Its
+	 * jobs go to their own queue, not the one the polls process, so they stay to be counted.
+	 */
+	private void function $insertDueSchedule(required string name) {
+		var now = new wheels.JobScheduler().$nowMs();
+		queryExecute(
+			"INSERT INTO wheels_job_schedules (name, jobClass, data, queue, kind, spec, timezone, catchUp, catchUpWindowSeconds, enabled, source, lastEnqueuedFor)
+			VALUES (:name, 'wheels.tests._assets.jobs.ProcessOrdersJob', '{}', 'test_srun_jobs', 'interval', '60', 'UTC', 'latest', 3600, 1, 'db', :last)",
+			{
+				name = {value = arguments.name, cfsqltype = "cf_sql_varchar"},
+				last = {value = now - 300000, cfsqltype = "cf_sql_bigint"}
+			},
+			{datasource = application.wheels.dataSourceName}
+		);
+	}
+
+	private numeric function $jobsFor(required string scheduleName) {
+		return queryExecute(
+			"SELECT id FROM wheels_jobs WHERE queue = 'test_srun_jobs' AND uniqueKey LIKE :prefix",
+			{prefix = {value = arguments.scheduleName & ":%", cfsqltype = "cf_sql_varchar"}},
+			{datasource = application.wheels.dataSourceName}
+		).recordCount;
+	}
+
+}