fix(job): reclaim a job when its heartbeats stop, not only after 2x its timeout

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))

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");

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.

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;

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 = {}) {
+	}
+
+}

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 = {}) {
+	}
+
+}

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 : "";
+	}
+
+}