fix(job): guard processQueue job claim with status=pending and affected-row check (#2899)

* fix(job): guard processQueue job claim with status=pending and affected-row check

processQueue's $processJob marked a job as processing with an unguarded
UPDATE (WHERE id = :id only), so two concurrent workers could both claim
and execute the same job. Mirror JobWorker.cfc::$claimJob: add
AND status = 'pending' to the claim UPDATE and check the affected-row
count via the queryExecute result option on the same statement (a
separate verification SELECT breaks on BoxLang + PostgreSQL when the
pool hands out a different connection).

A lost claim now returns {skipped = true} from $processJob, and
processQueue counts it under a new additive 'skipped' key instead of
recording a failure. attempts increments only on a successful claim,
keeping increment-and-claim atomic.

Spec: new "Job Claim Guard" describe in JobQueueSpec proves an
already-processing row is not re-executed (status stays 'processing',
attempts stays 0), plus a 'skipped' key assertion on the processQueue
result shape. Verified red (22 pass / 2 fail pre-fix) then green
(24 pass / 0 fail) on Lucee 7 + SQLite.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Signed-off-by: Peter Amiri <peter@alurium.com>

* fix(job): address Reviewer A/B consensus findings (round 1)

- vendor/wheels/Job.cfc:238 — drop the redundant StructKeyExists guard
  on local.jobResult.skipped. $processJob initialises
  local.result = {success = false, skipped = false, error = ""}
  unconditionally and every return path returns that struct, so the
  key is always present.
- vendor/wheels/tests/specs/jobs/JobQueueSpec.cfc — move the
  "Job Claim Guard" describe block's DELETE cleanup from inline at
  the end of the it block into afterEach (try/catch-wrapped, matching
  beforeEach), so a throw before the inline DELETE no longer leaks
  the seeded row. beforeEach still cleans up too (defense in depth).
- CHANGELOG.md — add the [Unreleased] block (Keep a Changelog
  convention) with a ### Fixed entry for the processQueue claim guard.
  Last [Unreleased] was promoted to 4.0.3 in 08dd480 on 2026-06-09,
  so this PR is the first change targeting the 4.0.4 snapshot.

Signed-off-by: claude[bot] <41898282+claude[bot]@users.noreply.github.com>

---------

Signed-off-by: Peter Amiri <peter@alurium.com>
Signed-off-by: claude[bot] <41898282+claude[bot]@users.noreply.github.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Co-authored-by: claude[bot] <41898282+claude[bot]@users.noreply.github.com>

Author: 3 people

CHANGELOG.md

@@ -20,6 +20,10 @@ All historical references to "CFWheels" in this changelog have been preserved fo
 
 ## [Unreleased]
 
+### Fixed
+
+- `Job.processQueue()`'s private `$processJob` now guards the claim `UPDATE` with `AND status = 'pending'` and verifies the affected-row count via the `queryExecute` `result` option on the same statement, mirroring the matrix-proven `JobWorker.cfc::$claimJob` idiom (a separate verification `SELECT` breaks on BoxLang + PostgreSQL when the connection pool hands out a different connection that cannot see the uncommitted UPDATE). Pre-fix, two concurrent claimers — overlapping `processQueue()` callers, or `processQueue` racing the CLI worker — could both claim the same job and both run `perform()` (duplicate emails/charges, with `attempts` double-incremented). A lost claim now early-returns `{success = false, skipped = true}` before job instantiation, tenant-context setup, and `perform()`; `processQueue()` counts lost claims under a new additive `skipped` result key (#2899)
+
 ### Security
 
 - `$isSafeRedirectUrl()` rejects backslash-containing URLs (`/\evil.com`, `\/evil.com`, `\\evil.com`) and schemeless-authority URLs (`https:/evil.com`, `javascript:alert(1)`) instead of returning them as safe. Browsers normalize backslashes to forward slashes and single-slash schemes to authority form, so any of those vectors smuggled past the previous check would navigate off-site after `redirectTo()`. Scheme detection now uses the RFC 3986 grammar (`ReFindNoCase("^[a-z][a-z0-9+.-]*:")`) and runs before the relative-URL fast path; backslashes are rejected up front, so the same-domain `ListFirst` no longer needs to treat `\` as a delimiter. Same-origin absolute URLs and genuine relative paths remain allowed (#2898)

vendor/wheels/Job.cfc

@@ -209,7 +209,7 @@ component {
 	 * @limit Maximum number of jobs to process in this batch.
 	 */
 	public struct function processQueue(string queue = "", numeric limit = 10) {
-		local.result = {processed = 0, failed = 0, errors = []};
+		local.result = {processed = 0, failed = 0, skipped = 0, errors = []};
 		local.params = {
 			runAt = {value = $now(), cfsqltype = "cf_sql_timestamp"}
 		};
@@ -235,6 +235,11 @@ component {
 
 		for (local.row in local.jobs) {
 			local.jobResult = $processJob(local.row);
+			if (local.jobResult.skipped) {
+				// Another worker claimed the job between our SELECT and the claim UPDATE
+				local.result.skipped++;
+				continue;
+			}
 			if (local.jobResult.success) {
 				local.result.processed++;
 			} else {
@@ -250,20 +255,30 @@ component {
 	 * Internal: Process a single job row.
 	 */
 	private struct function $processJob(required struct jobRow) {
-		local.result = {success = false, error = ""};
-
-		// Mark as processing
+		local.result = {success = false, skipped = false, error = ""};
+
+		// Mark as processing using optimistic locking: the status guard ensures only
+		// one concurrent worker can claim the job. Use the result option to get the
+		// affected-row count from the same connection that executed the UPDATE. A
+		// separate verification SELECT can fail on BoxLang + PostgreSQL when the
+		// connection pool hands out a different connection that cannot see the
+		// uncommitted UPDATE.
 		try {
 			queryExecute(
 				"UPDATE wheels_jobs
 				SET status = 'processing', attempts = attempts + 1, updatedAt = :updatedAt
-				WHERE id = :id",
+				WHERE id = :id AND status = 'pending'",
 				{
 					updatedAt = {value = $now(), cfsqltype = "cf_sql_timestamp"},
 					id = {value = arguments.jobRow.id, cfsqltype = "cf_sql_varchar"}
 				},
-				{datasource = variables.$datasource}
+				{datasource = variables.$datasource, result = "local.updateResult"}
 			);
+			if ((local.updateResult.recordCount ?: 0) == 0) {
+				// Another worker already claimed this job — skip without executing
+				local.result.skipped = true;
+				return local.result;
+			}
 		} catch (any e) {
 			local.result.error = "Failed to lock job #arguments.jobRow.id#: #e.message#";
 			return local.result;

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

@@ -147,6 +147,7 @@ component extends="wheels.WheelsTest" {
 				expect(local.result).toBeStruct();
 				expect(local.result).toHaveKey("processed");
 				expect(local.result).toHaveKey("failed");
+				expect(local.result).toHaveKey("skipped");
 				expect(local.result).toHaveKey("errors");
 			});
 
@@ -167,6 +168,64 @@ component extends="wheels.WheelsTest" {
 			});
 		});
 
+		describe("Job Claim Guard", function() {
+
+			beforeEach(function() {
+				// Clean up any leftover test jobs from prior runs
+				try { queryExecute("DELETE FROM wheels_jobs WHERE queue = 'test_claim_guard'", {}, {datasource = application.wheels.dataSourceName}); }
+				catch (any e) { /* table may not exist */ }
+			});
+
+			afterEach(function() {
+				try { queryExecute("DELETE FROM wheels_jobs WHERE queue = 'test_claim_guard'", {}, {datasource = application.wheels.dataSourceName}); }
+				catch (any e) { /* table may not exist */ }
+			});
+
+			it("does not execute a job already claimed by another worker", function() {
+				// Enqueue using a concrete subclass so jobClass resolves correctly
+				local.testJob = new app.jobs.ProcessOrdersJob();
+				local.enqueued = local.testJob.enqueue(data = {}, queue = "test_claim_guard");
+				expect(local.enqueued).toHaveKey("persisted");
+				expect(local.enqueued.persisted).toBeTrue();
+
+				// Simulate a concurrent worker having already claimed the job
+				queryExecute(
+					"UPDATE wheels_jobs SET status = 'processing' WHERE id = :id",
+					{id = {value = local.enqueued.id, cfsqltype = "cf_sql_varchar"}},
+					{datasource = application.wheels.dataSourceName}
+				);
+
+				// Build the job row as processQueue's SELECT would have seen it pre-claim
+				local.jobRow = {
+					id = local.enqueued.id,
+					jobClass = "app.jobs.ProcessOrdersJob",
+					queue = "test_claim_guard",
+					data = "{}",
+					attempts = 0,
+					maxRetries = 3
+				};
+
+				local.job = new wheels.Job();
+				prepareMock(local.job);
+				makePublic(local.job, "$processJob");
+				local.jobResult = local.job.$processJob(jobRow = local.jobRow);
+
+				// The lost claim must be reported as skipped, not executed
+				expect(local.jobResult).toHaveKey("skipped");
+				expect(local.jobResult.skipped).toBeTrue();
+				expect(local.jobResult.success).toBeFalse();
+
+				// The already-claimed row must be untouched: still processing, attempts not incremented
+				local.row = queryExecute(
+					"SELECT status, attempts FROM wheels_jobs WHERE id = :id",
+					{id = {value = local.enqueued.id, cfsqltype = "cf_sql_varchar"}},
+					{datasource = application.wheels.dataSourceName}
+				);
+				expect(local.row.status).toBe("processing");
+				expect(local.row.attempts).toBe(0);
+			});
+		});
+
 		describe("Job Data Serialization", function() {
 
 			it("enqueue handles complex data structures", function() {