diff --git a/changelog.d/schedule-runner-wiring.added.md b/changelog.d/schedule-runner-wiring.added.md new file mode 100644 index 0000000000..92235c911c --- /dev/null +++ b/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`. diff --git a/vendor/wheels/Job.cfc b/vendor/wheels/Job.cfc index 801ec4bb83..c2604c7096 100644 --- a/vendor/wheels/Job.cfc +++ b/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 diff --git a/vendor/wheels/JobRunner.cfc b/vendor/wheels/JobRunner.cfc index 44e5b379b0..8ef99e6fe0 100644 --- a/vendor/wheels/JobRunner.cfc +++ b/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++) { diff --git a/vendor/wheels/JobWorker.cfc b/vendor/wheels/JobWorker.cfc index d52109f9dc..a19dffead4 100644 --- a/vendor/wheels/JobWorker.cfc +++ b/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)) { diff --git a/vendor/wheels/events/onapplicationstart.cfc b/vendor/wheels/events/onapplicationstart.cfc index 8039b446aa..1e9561758a 100644 --- a/vendor/wheels/events/onapplicationstart.cfc +++ b/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; diff --git a/vendor/wheels/tests/specs/jobs/JobScheduleRunnerSpec.cfc b/vendor/wheels/tests/specs/jobs/JobScheduleRunnerSpec.cfc new file mode 100644 index 0000000000..9a33500caf --- /dev/null +++ b/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; + } + +}