diff --git a/changelog.d/4502-heartbeat-grace.fixed.md b/changelog.d/4502-heartbeat-grace.fixed.md new file mode 100644 index 0000000000..c2ba7f87e7 --- /dev/null +++ b/changelog.d/4502-heartbeat-grace.fixed.md @@ -0,0 +1 @@ +- `heartbeat()` now shortens how long a dead worker's job goes unnoticed. Before, a job was only reclaimed after twice its timeout whether it heartbeated or not, so heartbeats changed nothing. Now a job that has heartbeated is reclaimed once its heartbeats stop for its heartbeat grace, `set(jobsHeartbeatGraceSeconds = n)` (default 300; `0` turns it off; values below 30 are raised to 30) or the job's own `this.heartbeatGrace`, even inside its timeout. Jobs that never heartbeat behave exactly as before. Heartbeat at least every grace / 3. A stopped heartbeat isn't proof the worker died: a `perform()` stuck in a long call that can't heartbeat is reclaimed too. `heartbeat()` still never extends a job's timeout, which stays the hard cap on one attempt. An exclusive job's heartbeat now renews its lease by the grace window, so a job that stops heartbeating loses its lease on the same clock it's reclaimed on. ([#4502](https://github.com/wheels-dev/wheels/issues/4502)) diff --git a/vendor/wheels/Job.cfc b/vendor/wheels/Job.cfc index aa3e6db254..cc4d37ce5b 100644 --- a/vendor/wheels/Job.cfc +++ b/vendor/wheels/Job.cfc @@ -41,6 +41,9 @@ component { // at-least-once, so perform() must be idempotent. false: it is not re-run; the reaped // attempt ends 'interrupted' (at-most-once). this.idempotent = true; + // A job may set this.heartbeatGrace = n (seconds): once it heartbeats, it is reclaimed when + // its heartbeats stop for that long, even inside its timeout. Unset, the app's + // set(jobsHeartbeatGraceSeconds = n) applies (default 300); 0 turns it off. See heartbeat(). // true: never two runs of this job class at once, on any server (a lease in wheels_job_locks). // A job can instead set this.concurrencyKey = "...", or define concurrencyKeyFor(struct data) // (which wins), to share one lease among the jobs with the same key. See $jobLeaseName(). @@ -73,9 +76,13 @@ component { } /** - * Tell the queue this job is still running. Call it from a long perform() more often than - * the job's timeout: the stale-job reaper measures from the latest heartbeat, so a job that - * heartbeats on time is never reaped and run a second time. Throws Wheels.Job.Fenced when + * Tell the queue this job is still running. Once a job has heartbeated, a stop in its + * heartbeats is how its worker's death is noticed: the job is reclaimed when it hasn't + * heartbeated for its heartbeat grace (this.heartbeatGrace, or jobsHeartbeatGraceSeconds, + * default 300 seconds), even before its timeout. Heartbeat at least every grace / 3. A stopped + * heartbeat isn't proof the worker died: perform() stuck in a long call that can't heartbeat + * is reclaimed too, so either heartbeat around such calls or raise the grace. heartbeat() + * never extends the job's timeout, which stays the hard cap on one attempt. Throws Wheels.Job.Fenced when * the job's claim is gone (it was reaped and claimed again): stop working and return, since * another attempt now owns the job and this one's result will be discarded. Outside a * worker (perform() called directly) it does nothing. For an exclusive job (this.exclusive @@ -126,6 +133,38 @@ component { variables.$lease = {name = arguments.name, owner = arguments.owner, windowSeconds = arguments.windowSeconds}; } + /** + * The heartbeat grace for this job, in seconds: how long after its last heartbeat a running + * job is taken for dead and reclaimed, even inside its timeout. this.heartbeatGrace when set, + * else set(jobsHeartbeatGraceSeconds = n) (default 300). 0 = off. Values below 30 are raised + * to 30, so an occasional slow heartbeat can't get a live job reclaimed. + */ + public numeric function $heartbeatGraceSeconds() { + local.grace = 300; + if (StructKeyExists(this, "heartbeatGrace") && IsNumeric(this.heartbeatGrace)) { + local.grace = Val(this.heartbeatGrace); + } else if (StructKeyExists(application, "wheels") && StructKeyExists(application.wheels, "jobsHeartbeatGraceSeconds") && IsNumeric(application.wheels.jobsHeartbeatGraceSeconds)) { + local.grace = Val(application.wheels.jobsHeartbeatGraceSeconds); + } + if (local.grace <= 0) { + return 0; + } + return Max(30, local.grace); + } + + /** + * Internal: how far a heartbeat extends the run's exclusive lease. With a heartbeat grace, + * grace + Max(60, grace), so a job that stops heartbeating loses its lease on the same clock + * it is reclaimed on; without one, the lease's original window. + */ + public numeric function $leaseRenewalSeconds() { + local.grace = $heartbeatGraceSeconds(); + if (local.grace > 0) { + return local.grace + Max(60, local.grace); + } + return variables.$lease.windowSeconds; + } + /** * Internal: extends this run's exclusive lease by its window from now. A lease that is no * longer this run's (it expired and another run took it) is logged once and left alone; the @@ -141,7 +180,7 @@ component { leaseState.renewed = local.leaseLock.renew( name = variables.$lease.name, owner = variables.$lease.owner, - expiresAt = local.leaseLock.nowMs() + variables.$lease.windowSeconds * 1000 + expiresAt = local.leaseLock.nowMs() + $leaseRenewalSeconds() * 1000 ); } catch (any e) { writeLog(text = "Job lease '#variables.$lease.name#' could not be renewed: #e.message#", type = "error", file = "wheels_jobs"); diff --git a/vendor/wheels/JobWorker.cfc b/vendor/wheels/JobWorker.cfc index eb9a7a223e..86a8c82148 100644 --- a/vendor/wheels/JobWorker.cfc +++ b/vendor/wheels/JobWorker.cfc @@ -283,7 +283,9 @@ component { // a floor cutoff of 60s, since max(60s, …) makes the smallest possible grace ~60s so // a row idle 60s or less can never be reapable. Rows with no claimTimeout (pre-#3989 // or an ALTER-blocked table) fall back to this poller's own timeout. - local.floorCutoff = DateAdd("s", -60, $now()); + // The floor is 30s, not 60s: a heartbeating row can be reapable from its heartbeat grace, + // whose minimum is 30s. + local.floorCutoff = DateAdd("s", -30, $now()); local.params = {cutoff = {value = local.floorCutoff, cfsqltype = "cf_sql_timestamp"}}; @@ -325,6 +327,15 @@ component { local.rowTimeout = Val(local.row.claimTimeout); } local.grace = local.rowTimeout + Max(60, local.rowTimeout); + // A job that has heartbeated is also reapable once its heartbeats stop for its + // heartbeat grace (#4502), whichever comes first. Rows that never heartbeat keep the + // claim window alone. + if (StructKeyExists(local.row, "hasBeat") && Val(local.row.hasBeat) == 1) { + local.beatGrace = $jobHeartbeatGrace(local.row.jobClass); + if (local.beatGrace > 0) { + local.grace = Min(local.grace, local.beatGrace); + } + } local.rowCutoff = DateAdd("s", -local.grace, local.now); local.currentAttempts = Val(local.row.attempts); @@ -379,8 +390,11 @@ component { // a failing SELECT every poll; the catch is a safety net if the memo is stale. if ($claimTimeoutColumnAvailable()) { try { + // hasBeat: whether the job has heartbeated (#4502), as a flag so no timestamp is + // read in CFML. + local.beatColumn = $heartbeatColumnAvailable() ? ", CASE WHEN heartbeatAt IS NULL THEN 0 ELSE 1 END AS hasBeat" : ""; return queryExecute( - "SELECT id, jobClass, queue, attempts, maxRetries, updatedAt, claimTimeout + "SELECT id, jobClass, queue, attempts, maxRetries, updatedAt, claimTimeout" & local.beatColumn & " FROM wheels_jobs " & arguments.whereClause, arguments.params, {datasource = variables.$datasource} @@ -1272,6 +1286,26 @@ component { return $heartbeatColumnAvailable() ? "COALESCE(heartbeatAt, updatedAt)" : "updatedAt"; } + /** + * A job class's heartbeat grace in seconds (its $heartbeatGraceSeconds()), memoised per + * worker and class. A class that can't be loaded gets the app's setting. + */ + private numeric function $jobHeartbeatGrace(required string jobClass) { + if (!StructKeyExists(variables, "$heartbeatGraceByClass")) { + variables.$heartbeatGraceByClass = {}; + } + if (!StructKeyExists(variables.$heartbeatGraceByClass, arguments.jobClass)) { + var resolved = {seconds = 0}; + try { + resolved.seconds = $jobBridge().$instantiateJobClass(jobClass = arguments.jobClass).$heartbeatGraceSeconds(); + } catch (any e) { + resolved.seconds = $jobBridge().$heartbeatGraceSeconds(); + } + variables.$heartbeatGraceByClass[arguments.jobClass] = resolved.seconds; + } + return variables.$heartbeatGraceByClass[arguments.jobClass]; + } + /** * Whether a reaped job may be retried: its class's this.idempotent, true by default. A class * that can't be loaded counts as the default; its retry then fails the usual way. diff --git a/vendor/wheels/events/onapplicationstart.cfc b/vendor/wheels/events/onapplicationstart.cfc index d8562baa4f..4260f4cfdf 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"; + // A background job that has heartbeated is reclaimed once its heartbeats stop for this many + // seconds, even inside its timeout (0 = off; minimum 30). A job's this.heartbeatGrace wins. + application.$wheels.jobsHeartbeatGraceSeconds = 300; application.$wheels.migrationLockTimeout = 300; application.$wheels.migrationLockLease = 3600; application.$wheels.createMigratorTable = true; diff --git a/vendor/wheels/tests/_assets/jobs/LongGraceJob.cfc b/vendor/wheels/tests/_assets/jobs/LongGraceJob.cfc new file mode 100644 index 0000000000..b31c3478e5 --- /dev/null +++ b/vendor/wheels/tests/_assets/jobs/LongGraceJob.cfc @@ -0,0 +1,14 @@ +/** + * A job with its own, longer heartbeat grace (core-suite fixture). + */ +component extends="wheels.Job" { + + function config() { + super.config(); + this.heartbeatGrace = 1200; + } + + public void function perform(struct data = {}) { + } + +} diff --git a/vendor/wheels/tests/_assets/jobs/NoGraceJob.cfc b/vendor/wheels/tests/_assets/jobs/NoGraceJob.cfc new file mode 100644 index 0000000000..8912b97212 --- /dev/null +++ b/vendor/wheels/tests/_assets/jobs/NoGraceJob.cfc @@ -0,0 +1,14 @@ +/** + * A job that turns the heartbeat grace off for itself (core-suite fixture). + */ +component extends="wheels.Job" { + + function config() { + super.config(); + this.heartbeatGrace = 0; + } + + public void function perform(struct data = {}) { + } + +} diff --git a/vendor/wheels/tests/specs/jobs/JobHeartbeatGraceSpec.cfc b/vendor/wheels/tests/specs/jobs/JobHeartbeatGraceSpec.cfc new file mode 100644 index 0000000000..428249abda --- /dev/null +++ b/vendor/wheels/tests/specs/jobs/JobHeartbeatGraceSpec.cfc @@ -0,0 +1,126 @@ +/** + * Heartbeat grace: a running job that has heartbeated is reclaimed once its heartbeats stop + * for its grace (jobsHeartbeatGraceSeconds, default 300, or the job's this.heartbeatGrace), + * even while its long timeout hasn't passed. Jobs that never heartbeat keep the claim window + * alone. Each row here claims a one-hour timeout, so only the heartbeat grace can reap it. + */ +component extends="wheels.WheelsTest" { + + function run() { + + describe("heartbeat grace", function() { + + beforeEach(function() { + new wheels.Job().$ensureJobTable(); + $cleanup(); + application.wheels.jobsHeartbeatGraceSeconds = 300; + }); + + afterEach(function() { + $cleanup(); + application.wheels.jobsHeartbeatGraceSeconds = 300; + }); + + it("reclaims a job whose heartbeats stopped longer ago than the grace", function() { + var id = $insertRunning(queue = "test_grace_stale", heartbeatMinutesAgo = 10); + expect(new wheels.JobWorker().checkTimeouts(timeout = 300, queues = "test_grace_stale")).toBe(1); + expect($status(id)).toBe("pending", "retried: its worker stopped heartbeating"); + }); + + it("leaves a job that heartbeated within the grace", function() { + var id = $insertRunning(queue = "test_grace_fresh", heartbeatMinutesAgo = 1); + new wheels.JobWorker().checkTimeouts(timeout = 300, queues = "test_grace_fresh"); + expect($status(id)).toBe("processing"); + }); + + it("leaves a job that never heartbeated to its claim window, as before", function() { + var id = $insertRunning(queue = "test_grace_never", claimedMinutesAgo = 10); + new wheels.JobWorker().checkTimeouts(timeout = 300, queues = "test_grace_never"); + expect($status(id)).toBe("processing", "without heartbeats only the one-hour claim window applies"); + }); + + it("turns off with jobsHeartbeatGraceSeconds = 0", function() { + application.wheels.jobsHeartbeatGraceSeconds = 0; + var id = $insertRunning(queue = "test_grace_off", heartbeatMinutesAgo = 10); + new wheels.JobWorker().checkTimeouts(timeout = 300, queues = "test_grace_off"); + expect($status(id)).toBe("processing"); + }); + + it("uses the job's own this.heartbeatGrace over the setting", function() { + var longGrace = $insertRunning(queue = "test_grace_class", heartbeatMinutesAgo = 10, jobClass = "wheels.tests._assets.jobs.LongGraceJob"); + var noGrace = $insertRunning(queue = "test_grace_class", heartbeatMinutesAgo = 10, jobClass = "wheels.tests._assets.jobs.NoGraceJob"); + new wheels.JobWorker().checkTimeouts(timeout = 300, queues = "test_grace_class"); + expect($status(longGrace)).toBe("processing", "its own grace is 20 minutes"); + expect($status(noGrace)).toBe("processing", "it turned the grace off"); + }); + + it("raises a grace below 30 seconds to 30", function() { + application.wheels.jobsHeartbeatGraceSeconds = 5; + expect(new wheels.Job().$heartbeatGraceSeconds()).toBe(30); + var id = $insertRunning(queue = "test_grace_floor", heartbeatSecondsAgo = 15); + new wheels.JobWorker().checkTimeouts(timeout = 300, queues = "test_grace_floor"); + expect($status(id)).toBe("processing", "15 seconds is inside the 30-second minimum"); + }); + + it("renews an exclusive job's lease by the heartbeat grace window", function() { + application.wheels.jobsHeartbeatGraceSeconds = 100; + var job = new wheels.Job(); + expect(job.$heartbeatGraceSeconds()).toBe(100); + job.$setLeaseContext(name = "spec:grace", owner = "o", windowSeconds = 7200); + expect(job.$leaseRenewalSeconds()).toBe(200, "grace + Max(60, grace), not the 2-hour claim window"); + application.wheels.jobsHeartbeatGraceSeconds = 0; + expect(job.$leaseRenewalSeconds()).toBe(7200, "without a grace the lease keeps its window"); + }); + + }); + } + + private void function $cleanup() { + try { + queryExecute("DELETE FROM wheels_jobs WHERE queue LIKE 'test_grace_%'", {}, {datasource = application.wheels.dataSourceName}); + } catch (any e) { + } + } + + /** + * A row claimed with a one-hour timeout: claimed claimedMinutesAgo, and last heartbeat + * heartbeatMinutesAgo / heartbeatSecondsAgo (none when both are 0). + */ + private string function $insertRunning( + required string queue, + numeric claimedMinutesAgo = 2, + numeric heartbeatMinutesAgo = 0, + numeric heartbeatSecondsAgo = 0, + string jobClass = "wheels.tests._assets.jobs.ProcessOrdersJob" + ) { + var id = CreateUUID(); + var claimedAt = DateAdd("n", -Max(arguments.claimedMinutesAgo, arguments.heartbeatMinutesAgo + 1), Now()); + var beatSeconds = arguments.heartbeatMinutesAgo * 60 + arguments.heartbeatSecondsAgo; + var params = { + id = {value = id, cfsqltype = "cf_sql_varchar"}, + jobClass = {value = arguments.jobClass, cfsqltype = "cf_sql_varchar"}, + queue = {value = arguments.queue, cfsqltype = "cf_sql_varchar"}, + runAt = {value = claimedAt, cfsqltype = "cf_sql_timestamp"}, + createdAt = {value = claimedAt, cfsqltype = "cf_sql_timestamp"}, + updatedAt = {value = claimedAt, cfsqltype = "cf_sql_timestamp"}, + heartbeatAt = {value = DateAdd("s", -beatSeconds, Now()), cfsqltype = "cf_sql_timestamp", null = beatSeconds == 0} + }; + queryExecute( + "INSERT INTO wheels_jobs (id, jobClass, queue, data, priority, status, attempts, maxRetries, runAt, createdAt, updatedAt, claimTimeout, heartbeatAt) + VALUES (:id, :jobClass, :queue, '{}', 0, 'processing', 1, 3, :runAt, :createdAt, :updatedAt, 3600, :heartbeatAt)", + params, + {datasource = application.wheels.dataSourceName} + ); + return id; + } + + private string function $status(required string id) { + var q = queryExecute( + "SELECT status FROM wheels_jobs WHERE id = :id", + {id = {value = arguments.id, cfsqltype = "cf_sql_varchar"}}, + {datasource = application.wheels.dataSourceName} + ); + return q.recordCount ? q.status : ""; + } + +}