From db35dcfb1f16bec82b3b297701c146190e40b544 Mon Sep 17 00:00:00 2001 From: Peter Amiri Date: Thu, 8 Oct 2026 12:37:00 -0700 Subject: [PATCH 1/2] feat(job): run job schedules from the worker poll and JobRunner.tick() Due schedule slots are now enqueued by every worker poll (wheels jobs work, processQueue()) and JobRunner.tick(), at most once every jobsScheduleCheckSeconds (default 15, 0 = off) per application, so schedules need no separate process. Each slot is still enqueued once however many servers check. The check only runs when the schedules table exists, and never throws. tick() reports `scheduled`. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Peter Amiri --- changelog.d/schedule-runner-wiring.added.md | 1 + vendor/wheels/Job.cfc | 47 +++++++++ vendor/wheels/JobRunner.cfc | 4 +- vendor/wheels/JobWorker.cfc | 3 + vendor/wheels/events/onapplicationstart.cfc | 3 + .../specs/jobs/JobScheduleRunnerSpec.cfc | 95 +++++++++++++++++++ 6 files changed, 152 insertions(+), 1 deletion(-) create mode 100644 changelog.d/schedule-runner-wiring.added.md create mode 100644 vendor/wheels/tests/specs/jobs/JobScheduleRunnerSpec.cfc 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 aa3e6db254..a43ae16761 100644 --- a/vendor/wheels/Job.cfc +++ b/vendor/wheels/Job.cfc @@ -1743,6 +1743,53 @@ component { return {type = Len(arguments.type) ? arguments.type : "Wheels.JobFailed", message = arguments.message, detail = arguments.detail}; } + /** + * 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 = new wheels.JobScheduler(); + 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; + } + /** * A fresh, unique token for one claim of one job. */ 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 eb9a7a223e..be860f6ea7 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 d8562baa4f..49ea29723c 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; application.$wheels.migrationLockTimeout = 300; application.$wheels.migrationLockLease = 3600; application.$wheels.createMigratorTable = 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..e4012551fb --- /dev/null +++ b/vendor/wheels/tests/specs/jobs/JobScheduleRunnerSpec.cfc @@ -0,0 +1,95 @@ +/** + * 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("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; + } + +} From b4366e7e88331c6bb67c833cb3e114d5394a4178 Mon Sep 17 00:00:00 2001 From: Peter Amiri Date: Thu, 8 Oct 2026 15:36:44 -0700 Subject: [PATCH 2/2] test(job): schedule check that throws, with no table, and on a draining server A failing enqueueDue() leaves the poll running; with no schedules table nothing is checked (or created); a draining server's tick still enqueues due slots, which another server runs. Adds a $newScheduler() seam for the first two. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Peter Amiri --- vendor/wheels/Job.cfc | 9 ++++- .../specs/jobs/JobScheduleRunnerSpec.cfc | 37 +++++++++++++++++++ 2 files changed, 45 insertions(+), 1 deletion(-) diff --git a/vendor/wheels/Job.cfc b/vendor/wheels/Job.cfc index a43ae16761..b0a92df826 100644 --- a/vendor/wheels/Job.cfc +++ b/vendor/wheels/Job.cfc @@ -1743,6 +1743,13 @@ 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, @@ -1774,7 +1781,7 @@ component { } if (gate.due) { try { - local.scheduler = new wheels.JobScheduler(); + local.scheduler = $newScheduler(); if (local.scheduler.$schedulesTableExists()) { local.outcome = local.scheduler.enqueueDue(); rv.checked = local.outcome.checked; diff --git a/vendor/wheels/tests/specs/jobs/JobScheduleRunnerSpec.cfc b/vendor/wheels/tests/specs/jobs/JobScheduleRunnerSpec.cfc index e4012551fb..9a33500caf 100644 --- a/vendor/wheels/tests/specs/jobs/JobScheduleRunnerSpec.cfc +++ b/vendor/wheels/tests/specs/jobs/JobScheduleRunnerSpec.cfc @@ -46,6 +46,43 @@ component extends="wheels.WheelsTest" { 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");