From 8258542d2b517717702f4d59bda22e6472c16728 Mon Sep 17 00:00:00 2001 From: Peter Amiri Date: Wed, 7 Oct 2026 20:31:11 -0700 Subject: [PATCH 1/3] feat(job): exclusive jobs and concurrency keys on a shared lease lock Extract the migrator's cross-server lease into wheels.LeaseLock: statement builders plus execution methods over one lock table, every change after the insert fenced on the owner. The migrator keeps its table and lock methods and builds the same SQL with it through its own query function. A job with this.exclusive = true, a this.concurrencyKey, or a concurrencyKey(data) method now holds a lease in wheels_job_locks while it runs. A busy lease puts the job back to pending (attempt given back, counted as skipped); a lease lost by the end is logged and counted, and the finished job is not failed. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Peter Amiri --- .../4479-exclusive-jobs-lease-lock.added.md | 1 + vendor/wheels/Job.cfc | 254 +++++++++++++++- vendor/wheels/JobWorker.cfc | 41 ++- vendor/wheels/LeaseLock.cfc | 278 ++++++++++++++++++ vendor/wheels/Migrator.cfc | 92 +++--- .../tests/_assets/jobs/ExclusiveProbeJob.cfc | 25 ++ .../tests/_assets/jobs/KeyedProbeJob.cfc | 13 + .../tests/_assets/jobs/LeaseThiefJob.cfc | 20 ++ .../tests/_assets/jobs/StaticKeyJob.cfc | 15 + .../wheels/tests/specs/jobs/JobLeaseSpec.cfc | 251 ++++++++++++++++ 10 files changed, 935 insertions(+), 55 deletions(-) create mode 100644 changelog.d/4479-exclusive-jobs-lease-lock.added.md create mode 100644 vendor/wheels/LeaseLock.cfc create mode 100644 vendor/wheels/tests/_assets/jobs/ExclusiveProbeJob.cfc create mode 100644 vendor/wheels/tests/_assets/jobs/KeyedProbeJob.cfc create mode 100644 vendor/wheels/tests/_assets/jobs/LeaseThiefJob.cfc create mode 100644 vendor/wheels/tests/_assets/jobs/StaticKeyJob.cfc create mode 100644 vendor/wheels/tests/specs/jobs/JobLeaseSpec.cfc diff --git a/changelog.d/4479-exclusive-jobs-lease-lock.added.md b/changelog.d/4479-exclusive-jobs-lease-lock.added.md new file mode 100644 index 0000000000..cf93107ee7 --- /dev/null +++ b/changelog.d/4479-exclusive-jobs-lease-lock.added.md @@ -0,0 +1 @@ +- Background jobs can now be made exclusive: set `this.exclusive = true` and a second run of that job class can't start on any server while one is running. A job can instead set `this.concurrencyKey = "..."`, or define `concurrencyKey(struct data)`, to share one lease among the jobs with the same key (for example one run per account). A job that finds its lease busy waits as `pending` without using up an attempt, and counts as skipped. The lease is held in a new `wheels_job_locks` table, created on first use, through `wheels.LeaseLock`, the cross-server lease the migrator already used, now extracted for reuse. Exclusivity is only as strong as the lease: it lasts the job's claim timeout plus a grace period, and a run that outlives it (for example one stuck in a blocked database call) can overlap a second run; that is logged and counted as a lost lease. Servers sharing the table need synchronised clocks. ([#4479](https://github.com/wheels-dev/wheels/issues/4479)) diff --git a/vendor/wheels/Job.cfc b/vendor/wheels/Job.cfc index e9ea446474..ba2ff3e28a 100644 --- a/vendor/wheels/Job.cfc +++ b/vendor/wheels/Job.cfc @@ -37,6 +37,10 @@ component { // false: write the job after the outermost Wheels transaction resolves, on commit or // rollback alike, so it survives a rollback (failure notices, audit records). this.transactional = true; + // 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 concurrencyKey(struct data), to + // share one lease among the jobs with the same key. See $jobLeaseName(). + this.exclusive = false; /** * Constructor @@ -595,7 +599,7 @@ component { * @limit Maximum number of jobs to process in this batch. `0` (or less) means no limit: every due job is processed. */ public struct function processQueue(string queue = "", numeric limit = 10) { - local.result = {processed = 0, failed = 0, skipped = 0, fenced = 0, errors = []}; + local.result = {processed = 0, failed = 0, skipped = 0, fenced = 0, leasesLost = 0, errors = []}; local.params = { runAt = {value = $now(), cfsqltype = "cf_sql_timestamp"} }; @@ -633,6 +637,9 @@ component { for (local.row in local.jobs) { local.jobResult = $processJob(local.row); + if (local.jobResult.leaseLost) { + local.result.leasesLost++; + } if (local.jobResult.skipped) { // Another worker claimed the job between our SELECT and the claim UPDATE local.result.skipped++; @@ -710,7 +717,7 @@ component { * Internal: Process a single job row. */ private struct function $processJob(required struct jobRow) { - local.result = {success = false, skipped = false, fenced = false, error = ""}; + local.result = {success = false, skipped = false, fenced = false, leaseLost = false, error = ""}; // The retry/fail UPDATEs run inside a catch, where a local. write doesn't survive on // BoxLang, so a fenced outcome there is carried out through this struct. var fence = {lost = false}; @@ -784,11 +791,31 @@ component { // strip the internal $wheelsTenantContext key before passing data to perform() local.hasTenantContext = $restoreTenantContext(local.jobData); local.timeoutSeconds = $takeJobTimeout(local.jobData, local.jobInstance.timeout ?: 300); - local.performOutcome = $runPerformWithTimeout( + local.performOutcome = $runPerformExclusively( jobInstance = local.jobInstance, jobData = local.jobData, - timeoutSeconds = local.timeoutSeconds + timeoutSeconds = local.timeoutSeconds, + jobId = arguments.jobRow.id, + jobClass = arguments.jobRow.jobClass, + claimToken = local.claimToken, + claimTimeout = local.timeoutSeconds ); + if (local.performOutcome.busy) { + // Another run holds the job's lease: wait for it without using up an attempt. + $deferJobForLease( + jobId = arguments.jobRow.id, + jobClass = arguments.jobRow.jobClass, + claimToken = local.claimToken, + delaySeconds = local.performOutcome.retryInSeconds, + leaseName = local.performOutcome.leaseName + ); + local.result.skipped = true; + if (local.hasTenantContext) { + $clearTenantContext(); + } + return local.result; + } + local.result.leaseLost = local.performOutcome.leaseLost; if (local.performOutcome.timedOut) { throw(type = "Wheels.JobTimeout", message = local.performOutcome.error); } @@ -1827,6 +1854,225 @@ component { } } + /** + * Internal: runs perform() like $runPerformWithTimeout(), under the job's lease when it is + * exclusive or has a concurrency key. When another run holds the lease, perform() does not + * run: the outcome is `busy`, with `retryInSeconds` until the job should be tried again. + * The lease is released afterwards, except after a timeout: perform() may still be running + * in its thread then, so the lease is left to expire. `leaseLost` is true when the lease had + * already been taken over by the time the job finished. + */ + public struct function $runPerformExclusively( + required any jobInstance, + required struct jobData, + required numeric timeoutSeconds, + required string jobId, + required string jobClass, + required string claimToken, + required numeric claimTimeout + ) { + local.lease = $acquireJobLease( + jobInstance = arguments.jobInstance, + jobData = arguments.jobData, + jobClass = arguments.jobClass, + claimToken = arguments.claimToken, + claimTimeout = arguments.claimTimeout + ); + if (local.lease.busy) { + return {success = false, error = "", timedOut = false, busy = true, retryInSeconds = local.lease.retryInSeconds, leaseName = local.lease.name, leaseLost = false}; + } + local.outcome = $runPerformWithTimeout( + jobInstance = arguments.jobInstance, + jobData = arguments.jobData, + timeoutSeconds = arguments.timeoutSeconds + ); + local.outcome.busy = false; + local.outcome.retryInSeconds = 0; + local.outcome.leaseName = local.lease.name; + local.outcome.leaseLost = false; + if (local.lease.held && !local.outcome.timedOut) { + local.outcome.leaseLost = !$releaseJobLease(lease = local.lease, jobId = arguments.jobId, jobClass = arguments.jobClass); + } + return local.outcome; + } + + /** + * Internal: the lease a job run must hold, or "" when it needs none. A concurrency key (the + * job's concurrencyKey(data) method, or else its this.concurrencyKey) names a lease shared by + * every job with that key; otherwise this.exclusive = true leases the job class. A name that + * would not fit the lock table is hashed. + */ + public string function $jobLeaseName(required any jobInstance, required struct jobData, required string jobClass) { + local.key = ""; + if (StructKeyExists(arguments.jobInstance, "concurrencyKey")) { + if (IsSimpleValue(arguments.jobInstance.concurrencyKey)) { + local.key = arguments.jobInstance.concurrencyKey; + } else { + local.key = arguments.jobInstance.concurrencyKey(arguments.jobData); + } + } + if (Len(Trim(local.key))) { + return $fitLeaseName(prefix = "key:", value = Trim(local.key)); + } + if (StructKeyExists(arguments.jobInstance, "exclusive") && IsBoolean(arguments.jobInstance.exclusive) && arguments.jobInstance.exclusive) { + return $fitLeaseName(prefix = "job:", value = arguments.jobClass); + } + return ""; + } + + /** + * Internal: prefix & value, or prefix & the value's SHA-256 when that is longer than the lock + * table's 100-character name. + */ + public string function $fitLeaseName(required string prefix, required string value) { + if (Len(arguments.prefix & arguments.value) <= 100) { + return arguments.prefix & arguments.value; + } + return arguments.prefix & LCase(Hash(arguments.value, "SHA-256")); + } + + /** + * Internal: takes the job's lease for this attempt, if it needs one. The owner is the + * attempt's claim token, and the lease lasts as long as the claim does before the row itself + * can be reaped (claimTimeout + Max(60, claimTimeout)), so it never ends inside the claim + * timeout. Returns `held`, `busy` (another run holds it), `retryInSeconds`, `name` and `owner`. + */ + public struct function $acquireJobLease( + required any jobInstance, + required struct jobData, + required string jobClass, + required string claimToken, + required numeric claimTimeout + ) { + local.rv = {held = false, busy = false, retryInSeconds = 0, name = "", owner = ""}; + local.rv.name = $jobLeaseName(jobInstance = arguments.jobInstance, jobData = arguments.jobData, jobClass = arguments.jobClass); + if (!Len(local.rv.name)) { + return local.rv; + } + local.rv.owner = Len(arguments.claimToken) ? arguments.claimToken : CreateUUID(); + local.lock = $jobLeaseLock(); + local.claimSeconds = Val(arguments.claimTimeout) > 0 ? Val(arguments.claimTimeout) : 300; + local.now = local.lock.nowMs(); + local.expiresAt = local.now + (local.claimSeconds + Max(60, local.claimSeconds)) * 1000; + if (local.lock.tryAcquire(name = local.rv.name, owner = local.rv.owner, host = $jobHostName(), now = local.now, expiresAt = local.expiresAt)) { + local.rv.held = true; + return local.rv; + } + // Try again once the holder's lease runs out, but at least every 30 seconds: the holder + // usually finishes and releases it well before then. + local.holder = local.lock.read(local.rv.name); + local.remaining = local.holder.held ? Ceiling((local.holder.expiresAt - local.now) / 1000) : 1; + local.rv.busy = true; + local.rv.retryInSeconds = Min(30, Max(1, local.remaining)); + return local.rv; + } + + /** + * Internal: releases a held job lease. False, with a warning logged, when the lease had + * already expired and been taken over (or removed): exclusivity is only as strong as the + * lease, and a run that outlives it may have overlapped another. + */ + public boolean function $releaseJobLease(required struct lease, required string jobId, required string jobClass) { + var state = {released = false}; + try { + state.released = $jobLeaseLock().release(name = arguments.lease.name, owner = arguments.lease.owner); + } catch (any e) { + WriteLog(type = "error", file = "wheels_jobs", text = "Job '#arguments.jobClass#' [#arguments.jobId#] could not release its lease '#arguments.lease.name#': #e.message#. It expires on its own."); + return true; + } + if (!state.released) { + WriteLog( + type = "warning", + file = "wheels_jobs", + text = "Job '#arguments.jobClass#' [#arguments.jobId#] finished after losing its lease '#arguments.lease.name#': the lease expired while it ran, so another run may have overlapped it." + ); + } + return state.released; + } + + /** + * Internal: puts a job whose lease is busy back to 'pending' to run in `delaySeconds`, and + * gives back the attempt its claim counted, so waiting on a busy lease never uses up retries. + * Fenced on the claim token like the other end-of-attempt UPDATEs. Returns the rows changed. + */ + public numeric function $deferJobForLease( + required string jobId, + required string jobClass, + required string claimToken, + required numeric delaySeconds, + required string leaseName + ) { + local.params = { + runAt = {value = DateAdd("s", arguments.delaySeconds, $now()), cfsqltype = "cf_sql_timestamp"}, + updatedAt = {value = $now(), cfsqltype = "cf_sql_timestamp"}, + id = {value = arguments.jobId, cfsqltype = "cf_sql_varchar"} + }; + local.guard = $claimTokenGuard(claimToken = arguments.claimToken, params = local.params); + local.clearToken = Len(arguments.claimToken) ? ", claimToken = NULL" : ""; + queryExecute( + "UPDATE wheels_jobs + SET status = 'pending', attempts = attempts - 1, runAt = :runAt, updatedAt = :updatedAt" & local.clearToken & " + WHERE id = :id AND status = 'processing' AND attempts > 0" & local.guard, + local.params, + {datasource = variables.$datasource, result = "local.deferResult"} + ); + WriteLog( + type = "information", + file = "wheels_jobs", + text = "Job '#arguments.jobClass#' [#arguments.jobId#] waits #arguments.delaySeconds#s: another run holds its lease '#arguments.leaseName#'" + ); + return Val(local.deferResult.recordCount ?: 0); + } + + /** + * Internal: the lease lock over wheels_job_locks, creating the table on first use. Not inside + * a Wheels transaction: DDL there commits the open transaction on MySQL and Oracle, so the + * job fails with the reason instead and is retried outside it. + */ + public any function $jobLeaseLock() { + if (!StructKeyExists(variables, "$jobLeaseLockInstance")) { + local.lock = new wheels.LeaseLock(table = "wheels_job_locks", datasource = variables.$datasource); + $ensureJobLockTable(local.lock); + variables.$jobLeaseLockInstance = local.lock; + } + return variables.$jobLeaseLockInstance; + } + + /** + * Internal: creates wheels_job_locks when it is missing. + */ + public void function $ensureJobLockTable(required any lock) { + if ($jobLockTableExists()) { + return; + } + if (Len($outermostWheelsTransaction())) { + Throw( + type = "Wheels.JobLockTableMissing", + message = "wheels_job_locks doesn't exist yet, and it can't be created inside a transaction. Run the job outside a transaction once, or create the table: #arguments.lock.createTableSql()#" + ); + } + try { + queryExecute(arguments.lock.createTableSql(), {}, {datasource = variables.$datasource}); + } catch (any e) { + // Tolerate "already exists" from another server creating it at the same time. + if (!$jobLockTableExists()) { + rethrow; + } + } + } + + /** + * Internal: whether wheels_job_locks exists. + */ + public boolean function $jobLockTableExists() { + try { + queryExecute("SELECT COUNT(*) AS cnt FROM wheels_job_locks WHERE 1=0", {}, {datasource = variables.$datasource}); + return true; + } catch (any e) { + return false; + } + } + /** * Retry delay in seconds. `retryBackoff=exponential` is the live schedule * (`baseDelay * 2^attempts`, capped at `maxDelay`). `linear` is a reserved diff --git a/vendor/wheels/JobWorker.cfc b/vendor/wheels/JobWorker.cfc index 04743adc1c..1b7a16d881 100644 --- a/vendor/wheels/JobWorker.cfc +++ b/vendor/wheels/JobWorker.cfc @@ -16,6 +16,8 @@ component { this.startedAt = Now(); this.jobsProcessed = 0; this.jobsFailed = 0; + // Runs that finished after their exclusive lease had expired and been taken over. + this.leasesLost = 0; variables.$datasource = ""; if (StructKeyExists(application, "wheels") && StructKeyExists(application.wheels, "dataSourceName")) { variables.$datasource = application.wheels.dataSourceName; @@ -31,7 +33,7 @@ component { * @timeout Timeout in seconds for a single job execution. */ public struct function processNext(string queues = "", numeric timeout = 300) { - local.result = {success = false, jobId = "", jobClass = "", error = "", skipped = false, fenced = false}; + local.result = {success = false, jobId = "", jobClass = "", error = "", skipped = false, fenced = false, deferred = false, leaseLost = false}; // Ensure the table (and the claimTimeout column) on the normal path, so an existing // install gets the column once per worker rather than only when a query happens to @@ -131,8 +133,23 @@ component { local.processResult = $executeJob(jobRow = local.jobRow, timeout = arguments.timeout); local.result.jobId = local.row.id; local.result.jobClass = local.row.jobClass; + if (local.processResult.leaseLost) { + this.leasesLost++; + local.result.leaseLost = true; + } - if (local.processResult.fenced) { + if (local.processResult.busy) { + // Another run holds the job's lease: it waits without using up an attempt. + $jobBridge().$deferJobForLease( + jobId = local.row.id, + jobClass = local.row.jobClass, + claimToken = local.jobRow.claimToken, + delaySeconds = local.processResult.retryInSeconds, + leaseName = local.processResult.leaseName + ); + local.result.skipped = true; + local.result.deferred = true; + } else if (local.processResult.fenced) { // Ran, but the claim was reaped and re-issued while it did: the completion // was rejected, and retrying it would requeue the attempt that replaced it. local.result.fenced = true; @@ -698,7 +715,7 @@ component { * Execute a job's perform() method. */ private struct function $executeJob(required struct jobRow, numeric timeout = 300) { - local.result = {success = false, fenced = false, error = ""}; + local.result = {success = false, fenced = false, busy = false, retryInSeconds = 0, leaseName = "", leaseLost = false, error = ""}; local.claimToken = StructKeyExists(arguments.jobRow, "claimToken") ? arguments.jobRow.claimToken : ""; // Initialized before the try so the cleanup below never reads an undefined @@ -725,11 +742,25 @@ component { local.workerCap = 300; } local.timeoutSeconds = Min(local.fromRow, local.workerCap); - local.performOutcome = $jobBridge().$runPerformWithTimeout( + local.performOutcome = $jobBridge().$runPerformExclusively( jobInstance = local.jobInstance, jobData = local.jobData, - timeoutSeconds = local.timeoutSeconds + timeoutSeconds = local.timeoutSeconds, + jobId = arguments.jobRow.id, + jobClass = arguments.jobRow.jobClass, + claimToken = local.claimToken, + claimTimeout = local.workerCap ); + local.result.leaseLost = local.performOutcome.leaseLost; + if (local.performOutcome.busy) { + local.result.busy = true; + local.result.retryInSeconds = local.performOutcome.retryInSeconds; + local.result.leaseName = local.performOutcome.leaseName; + if (local.hasTenantContext) { + $jobBridge().$clearTenantContext(); + } + return local.result; + } if (local.performOutcome.timedOut) { throw(type = "Wheels.JobTimeout", message = local.performOutcome.error); } diff --git a/vendor/wheels/LeaseLock.cfc b/vendor/wheels/LeaseLock.cfc new file mode 100644 index 0000000000..22aa6ce4d7 --- /dev/null +++ b/vendor/wheels/LeaseLock.cfc @@ -0,0 +1,278 @@ +/** + * A named, cross-process lease lock backed by one database table: one row per held lock, + * (lockname PK, lockowner, lockhost, acquiredat, expiresat), with the times as epoch + * milliseconds so every database compares them as plain numbers. + * + * A caller takes a lock by inserting its row, or by taking over a row whose lease has expired; + * it extends the lease while it works, and deletes the row when done. Every change after the + * insert is fenced on the owner id, so an instance whose lease was taken over can neither + * extend nor release the new holder's lock. + * + * The statement builders are used by the Migrator, which runs them through its own query + * function (its datasource credentials and test seams). The execution methods run them + * against `datasource` and are used by background jobs (`this.exclusive`, `concurrencyKey`). + * + * Lease times come from the app host's clock, so the hosts sharing a lock table need + * synchronised clocks (NTP): skew between two hosts shortens or lengthens a lease by as much. + */ +component { + + /** + * @table The lock table's name. + * @datasource The datasource the execution methods use. + */ + public any function init(required string table, string datasource = "") { + variables.table = arguments.table; + variables.datasource = arguments.datasource; + return this; + } + + /** + * The DDL for the lock table, portable across the supported databases. + */ + public string function createTableSql() { + return "CREATE TABLE #variables.table# (lockname VARCHAR(100) NOT NULL PRIMARY KEY, lockowner VARCHAR(64) NOT NULL, lockhost VARCHAR(255), acquiredat DECIMAL(15,0) NOT NULL, expiresat DECIMAL(15,0) NOT NULL)"; + } + + /** + * Reads the owner of a lock. + */ + public struct function ownerStatement(required string name) { + return { + sql = "SELECT lockowner FROM #variables.table# WHERE lockname = :lockName", + params = {lockName = text(arguments.name)} + }; + } + + /** + * Reads a lock's whole row. + */ + public struct function readStatement(required string name) { + return { + sql = "SELECT lockowner, lockhost, acquiredat, expiresat FROM #variables.table# WHERE lockname = :lockName", + params = {lockName = text(arguments.name)} + }; + } + + /** + * Takes a free lock: fails on the primary key when another owner inserted it first. + */ + public struct function insertStatement( + required string name, + required string owner, + required string host, + required numeric acquiredAt, + required numeric expiresAt + ) { + return { + sql = "INSERT INTO #variables.table# (lockname, lockowner, lockhost, acquiredat, expiresat) VALUES (:lockName, :owner, :host, :acquiredAt, :expiresAt)", + params = { + lockName = text(arguments.name), + owner = text(arguments.owner), + host = text(arguments.host), + acquiredAt = ms(arguments.acquiredAt), + expiresAt = ms(arguments.expiresAt) + } + }; + } + + /** + * Takes over a lock whose lease expired before `now`. The WHERE makes it atomic; reading the + * owner back tells whether this caller won. + */ + public struct function takeOverStatement( + required string name, + required string owner, + required string host, + required numeric now, + required numeric expiresAt + ) { + return { + sql = "UPDATE #variables.table# SET lockowner = :owner, lockhost = :host, acquiredat = :now, expiresat = :expiresAt WHERE lockname = :lockName AND expiresat < :now", + params = { + owner = text(arguments.owner), + host = text(arguments.host), + now = ms(arguments.now), + expiresAt = ms(arguments.expiresAt), + lockName = text(arguments.name) + } + }; + } + + /** + * Extends the lease, only while `owner` still holds the lock. + */ + public struct function renewStatement(required string name, required string owner, required numeric expiresAt) { + return { + sql = "UPDATE #variables.table# SET expiresat = :expiresAt WHERE lockname = :lockName AND lockowner = :owner", + params = { + expiresAt = ms(arguments.expiresAt), + lockName = text(arguments.name), + owner = text(arguments.owner) + } + }; + } + + /** + * Deletes the lock while `owner` holds it; with `expiredBefore`, only when its lease expired + * before that time. + */ + public struct function releaseStatement(required string name, required string owner, any expiredBefore = "") { + local.rv = { + sql = "DELETE FROM #variables.table# WHERE lockname = :lockName AND lockowner = :owner", + params = {lockName = text(arguments.name), owner = text(arguments.owner)} + }; + if (IsNumeric(arguments.expiredBefore)) { + local.rv.sql &= " AND expiresat < :now"; + local.rv.params.now = ms(arguments.expiredBefore); + } + return local.rv; + } + + /** + * An epoch-milliseconds value as a typed query parameter: an untyped one binds as text, which + * PostgreSQL won't compare with the numeric columns. Not through Int(), which Lucee truncates + * to 32 bits. + */ + public struct function ms(required numeric value) { + return {value = arguments.value, cfsqltype = "cf_sql_bigint"}; + } + + /** + * A text value as a typed query parameter. + */ + public struct function text(required string value) { + return {value = arguments.value, cfsqltype = "cf_sql_varchar"}; + } + + /** + * The current time in epoch milliseconds, from the JVM's wall clock. GetTickCount() is only a + * fallback for an engine without java.lang.System. + */ + public numeric function nowMs() { + try { + return CreateObject("java", "java.lang.System").currentTimeMillis(); + } catch (any e) { + return GetTickCount(); + } + } + + /** + * Runs a statement against `datasource`. Returns `rows` (a query, empty for DML) and `count`, + * the rows a DML statement changed, read through the result option: Adobe 2023 returns + * nothing at all from QueryExecute for an UPDATE or DELETE. + */ + public struct function executeStatement(required struct statement) { + local.rows = QueryExecute( + arguments.statement.sql, + arguments.statement.params, + {datasource = variables.datasource, result = "local.info"} + ); + return { + rows = (StructKeyExists(local, "rows") && IsQuery(local.rows)) ? local.rows : QueryNew(""), + count = StructKeyExists(local, "info") ? Val(local.info.recordCount ?: 0) : 0 + }; + } + + /** + * The owner id holding the lock, or "" when it is free. + */ + public string function ownerOf(required string name) { + local.rows = executeStatement(ownerStatement(arguments.name)).rows; + return local.rows.recordCount ? local.rows.lockowner : ""; + } + + /** + * The lock's row: `held`, `owner`, `host`, `acquiredAt` and `expiresAt` (epoch ms). + */ + public struct function read(required string name) { + local.rows = executeStatement(readStatement(arguments.name)).rows; + if (!local.rows.recordCount) { + return {held = false, owner = "", host = "", acquiredAt = 0, expiresAt = 0}; + } + return { + held = true, + owner = local.rows.lockowner, + host = local.rows.lockhost, + acquiredAt = Val(local.rows.acquiredat), + expiresAt = Val(local.rows.expiresat) + }; + } + + /** + * One attempt to take the lock for `owner` until `expiresAt`: insert its row, or take it over + * when the lease expired. True when `owner` now holds it. + */ + public boolean function tryAcquire( + required string name, + required string owner, + required string host, + required numeric now, + required numeric expiresAt + ) { + // Insert only when no row exists, so the duplicate-key path (which aborts an enclosing + // PostgreSQL transaction) is left to two callers inserting at the same moment. + if (Len(ownerOf(arguments.name))) { + return $takeOver(argumentCollection = arguments); + } + try { + executeStatement( + insertStatement( + name = arguments.name, + owner = arguments.owner, + host = arguments.host, + acquiredAt = arguments.now, + expiresAt = arguments.expiresAt + ) + ); + return true; + } catch (any e) { + // No row even now: the INSERT failed for another reason (a missing table, a lost + // connection), so report it. + if (!Len(ownerOf(arguments.name))) { + rethrow; + } + } + // Another caller inserted it first: take it over only if its lease has expired. + return $takeOver(argumentCollection = arguments); + } + + /** + * Extends the lease to `expiresAt`. True while `owner` still holds the lock; false once it + * expired and another owner took it over (or it was removed). + */ + public boolean function renew(required string name, required string owner, required numeric expiresAt) { + return executeStatement(renewStatement(name = arguments.name, owner = arguments.owner, expiresAt = arguments.expiresAt)).count > 0; + } + + /** + * Deletes the lock while `owner` holds it. True when it did; false when the lease had already + * been taken over or removed. + */ + public boolean function release(required string name, required string owner) { + return executeStatement(releaseStatement(name = arguments.name, owner = arguments.owner)).count > 0; + } + + /** + * Internal: the owner-fenced takeover of an expired lease, then the read-back. + */ + public boolean function $takeOver( + required string name, + required string owner, + required string host, + required numeric now, + required numeric expiresAt + ) { + executeStatement( + takeOverStatement( + name = arguments.name, + owner = arguments.owner, + host = arguments.host, + now = arguments.now, + expiresAt = arguments.expiresAt + ) + ); + return ownerOf(arguments.name) == arguments.owner; + } + +} diff --git a/vendor/wheels/Migrator.cfc b/vendor/wheels/Migrator.cfc index c255b6c633..06faf5aadb 100644 --- a/vendor/wheels/Migrator.cfc +++ b/vendor/wheels/Migrator.cfc @@ -474,10 +474,7 @@ component output="false" extends="wheels.Global"{ } StructDelete(local.held, arguments.migrationLock.key); try { - $migrationLockQuery( - sql = "DELETE FROM #$migrationLockTable()# WHERE lockname = :lockName AND lockowner = :owner", - params = {lockName = $migrationLockText($migrationLockName()), owner = $migrationLockText(arguments.migrationLock.owner)} - ); + $runMigrationLockStatement($migrationLease().releaseStatement(name = $migrationLockName(), owner = arguments.migrationLock.owner)); } catch (any e) { WriteLog(type = "error", file = "wheels", text = "Migrator: could not release the migration lock: #e.message#"); } @@ -495,13 +492,8 @@ component output="false" extends="wheels.Global"{ return; } local.migrationLock = local.held[local.key]; - $migrationLockQuery( - sql = "UPDATE #$migrationLockTable()# SET expiresat = :expiresAt WHERE lockname = :lockName AND lockowner = :owner", - params = { - expiresAt = $migrationLockMs($migrationLockExpiry()), - lockName = $migrationLockText($migrationLockName()), - owner = $migrationLockText(local.migrationLock.owner) - } + $runMigrationLockStatement( + $migrationLease().renewStatement(name = $migrationLockName(), owner = local.migrationLock.owner, expiresAt = $migrationLockExpiry()) ); if ($migrationLockOwner() != local.migrationLock.owner) { StructDelete(local.held, local.key); @@ -545,15 +537,14 @@ component output="false" extends="wheels.Global"{ return $takeOverExpiredMigrationLock(migrationLock = arguments.migrationLock, now = local.now); } try { - $migrationLockQuery( - sql = "INSERT INTO #$migrationLockTable()# (lockname, lockowner, lockhost, acquiredat, expiresat) VALUES (:lockName, :owner, :host, :acquiredAt, :expiresAt)", - params = { - lockName = $migrationLockText($migrationLockName()), - owner = $migrationLockText(arguments.migrationLock.owner), - host = $migrationLockText($migrationLockHost()), - acquiredAt = $migrationLockMs(local.now), - expiresAt = $migrationLockMs($migrationLockExpiry()) - } + $runMigrationLockStatement( + $migrationLease().insertStatement( + name = $migrationLockName(), + owner = arguments.migrationLock.owner, + host = $migrationLockHost(), + acquiredAt = local.now, + expiresAt = $migrationLockExpiry() + ) ); return true; } catch (any e) { @@ -572,15 +563,14 @@ component output="false" extends="wheels.Global"{ * WHERE makes it atomic; reading the owner back tells whether this migration lock won. */ public boolean function $takeOverExpiredMigrationLock(required struct migrationLock, required numeric now) { - $migrationLockQuery( - sql = "UPDATE #$migrationLockTable()# SET lockowner = :owner, lockhost = :host, acquiredat = :now, expiresat = :expiresAt WHERE lockname = :lockName AND expiresat < :now", - params = { - owner = $migrationLockText(arguments.migrationLock.owner), - host = $migrationLockText($migrationLockHost()), - now = $migrationLockMs(arguments.now), - expiresAt = $migrationLockMs($migrationLockExpiry()), - lockName = $migrationLockText($migrationLockName()) - } + $runMigrationLockStatement( + $migrationLease().takeOverStatement( + name = $migrationLockName(), + owner = arguments.migrationLock.owner, + host = $migrationLockHost(), + now = arguments.now, + expiresAt = $migrationLockExpiry() + ) ); return $migrationLockOwner() == arguments.migrationLock.owner; } @@ -604,7 +594,7 @@ component output="false" extends="wheels.Global"{ try { $query( datasource = arguments.dataSource, - sql = "CREATE TABLE #local.table# (lockname VARCHAR(100) NOT NULL PRIMARY KEY, lockowner VARCHAR(64) NOT NULL, lockhost VARCHAR(255), acquiredat DECIMAL(15,0) NOT NULL, expiresat DECIMAL(15,0) NOT NULL)" + sql = $migrationLease().createTableSql() ); } catch (any e) { // Tolerate "already exists" from another instance creating it at the same time. @@ -639,10 +629,7 @@ component output="false" extends="wheels.Global"{ * Internal function. The owner id holding the migration lock, or "" when it is free. */ public string function $migrationLockOwner() { - local.rows = $migrationLockQuery( - sql = "SELECT lockowner FROM #$migrationLockTable()# WHERE lockname = :lockName", - params = {lockName = $migrationLockText($migrationLockName())} - ); + local.rows = $runMigrationLockStatement($migrationLease().ownerStatement($migrationLockName())); return local.rows.recordCount ? local.rows.lockowner : ""; } @@ -677,10 +664,7 @@ component output="false" extends="wheels.Global"{ if (!$migratorTableExists($migratorDataSource(), $migrationLockTable())) { return local.rv; } - local.rows = $migrationLockQuery( - sql = "SELECT lockowner, lockhost, acquiredat, expiresat FROM #$migrationLockTable()# WHERE lockname = :lockName", - params = {lockName = $migrationLockText($migrationLockName())} - ); + local.rows = $runMigrationLockStatement($migrationLease().readStatement($migrationLockName())); if (!local.rows.recordCount) { return local.rv; } @@ -712,14 +696,15 @@ component output="false" extends="wheels.Global"{ if (!local.rv.lock.held) { return local.rv; } - local.sql = "DELETE FROM #$migrationLockTable()# WHERE lockname = :lockName AND lockowner = :owner"; - local.params = {lockName = $migrationLockText($migrationLockName()), owner = $migrationLockText(local.rv.lock.owner)}; - if (!arguments.force) { - // In the DELETE itself, so a holder renewing its lease meanwhile keeps it. - local.sql &= " AND expiresat < :now"; - local.params.now = $migrationLockMs(GetTickCount()); - } - $migrationLockQuery(sql = local.sql, params = local.params); + // Without force, the expiry test is in the DELETE itself, so a holder renewing its lease + // meanwhile keeps it. + $runMigrationLockStatement( + $migrationLease().releaseStatement( + name = $migrationLockName(), + owner = local.rv.lock.owner, + expiredBefore = arguments.force ? "" : GetTickCount() + ) + ); // Read back, not compared with the old owner: an instance that took the lease between the // read and the DELETE keeps it, and that is not a release. local.rv["heldBy"] = $migrationLockOwner(); @@ -727,6 +712,21 @@ component output="false" extends="wheels.Global"{ return local.rv; } + /** + * Internal function. The lease lock over the migration lock table. Builds the lock's SQL; the + * Migrator runs it through $migrationLockQuery() for its datasource credentials. + */ + public any function $migrationLease() { + return new wheels.LeaseLock(table = $migrationLockTable()); + } + + /** + * Internal function. Runs a LeaseLock statement through $migrationLockQuery(). + */ + public query function $runMigrationLockStatement(required struct statement) { + return $migrationLockQuery(sql = arguments.statement.sql, params = arguments.statement.params); + } + /** * Internal function. Runs a query with named, typed parameters against the migrator's * datasource. Named, not positional: RustCFML binds a `{value, cfsqltype}` struct in a diff --git a/vendor/wheels/tests/_assets/jobs/ExclusiveProbeJob.cfc b/vendor/wheels/tests/_assets/jobs/ExclusiveProbeJob.cfc new file mode 100644 index 0000000000..38a7218c78 --- /dev/null +++ b/vendor/wheels/tests/_assets/jobs/ExclusiveProbeJob.cfc @@ -0,0 +1,25 @@ +/** + * An exclusive job (core-suite fixture). perform() counts its runs and records who held its + * lease while it ran, so the lease specs can see both. + */ +component extends="wheels.Job" { + + function config() { + super.config(); + this.exclusive = true; + } + + public void function perform(struct data = {}) { + if (!StructKeyExists(application, "$leaseSpecRuns")) { + application["$leaseSpecRuns"] = 0; + } + application["$leaseSpecRuns"] = application["$leaseSpecRuns"] + 1; + local.rows = queryExecute( + "SELECT lockowner FROM wheels_job_locks WHERE lockname = :name", + {name = {value = "job:wheels.tests._assets.jobs.ExclusiveProbeJob", cfsqltype = "cf_sql_varchar"}}, + {datasource = application.wheels.dataSourceName} + ); + application["$leaseSpecHolder"] = local.rows.recordCount ? local.rows.lockowner : ""; + } + +} diff --git a/vendor/wheels/tests/_assets/jobs/KeyedProbeJob.cfc b/vendor/wheels/tests/_assets/jobs/KeyedProbeJob.cfc new file mode 100644 index 0000000000..de91e07158 --- /dev/null +++ b/vendor/wheels/tests/_assets/jobs/KeyedProbeJob.cfc @@ -0,0 +1,13 @@ +/** + * A job whose concurrency key comes from its data (core-suite fixture): one run per account. + */ +component extends="wheels.Job" { + + public string function concurrencyKey(struct data = {}) { + return "acct-" & arguments.data.account; + } + + public void function perform(struct data = {}) { + } + +} diff --git a/vendor/wheels/tests/_assets/jobs/LeaseThiefJob.cfc b/vendor/wheels/tests/_assets/jobs/LeaseThiefJob.cfc new file mode 100644 index 0000000000..a5aa3aad43 --- /dev/null +++ b/vendor/wheels/tests/_assets/jobs/LeaseThiefJob.cfc @@ -0,0 +1,20 @@ +/** + * An exclusive job whose lease is taken over while it runs (core-suite fixture), as happens when + * a run outlives its lease and another server takes the lease over. + */ +component extends="wheels.Job" { + + function config() { + super.config(); + this.exclusive = true; + } + + public void function perform(struct data = {}) { + queryExecute( + "UPDATE wheels_job_locks SET lockowner = 'spec-thief' WHERE lockname = :name", + {name = {value = "job:wheels.tests._assets.jobs.LeaseThiefJob", cfsqltype = "cf_sql_varchar"}}, + {datasource = application.wheels.dataSourceName} + ); + } + +} diff --git a/vendor/wheels/tests/_assets/jobs/StaticKeyJob.cfc b/vendor/wheels/tests/_assets/jobs/StaticKeyJob.cfc new file mode 100644 index 0000000000..2a9b2e2725 --- /dev/null +++ b/vendor/wheels/tests/_assets/jobs/StaticKeyJob.cfc @@ -0,0 +1,15 @@ +/** + * A job with a fixed concurrency key, and exclusive left false (core-suite fixture): the key + * alone makes it share one lease. + */ +component extends="wheels.Job" { + + function config() { + super.config(); + this.concurrencyKey = "nightly-report"; + } + + public void function perform(struct data = {}) { + } + +} diff --git a/vendor/wheels/tests/specs/jobs/JobLeaseSpec.cfc b/vendor/wheels/tests/specs/jobs/JobLeaseSpec.cfc new file mode 100644 index 0000000000..683e1d4eef --- /dev/null +++ b/vendor/wheels/tests/specs/jobs/JobLeaseSpec.cfc @@ -0,0 +1,251 @@ +/** + * Lease locks: wheels.LeaseLock (the cross-process lease the Migrator uses, now reusable) and + * the jobs built on it. A job with this.exclusive = true, or a concurrency key, holds a lease + * in wheels_job_locks while it runs, so a second run can't start on any server until it ends. + * A run that finds the lease busy waits as 'pending' without using up an attempt. + */ +component extends="wheels.WheelsTest" { + + function run() { + + describe("wheels.LeaseLock", function() { + + beforeEach(function() { + new wheels.Job().$jobLeaseLock(); + $cleanup(); + }); + + afterEach(function() { + $cleanup(); + }); + + it("lets one owner take a free lease, and no other until it is released", function() { + var lock = $lock(); + expect(lock.tryAcquire(name = "spec:one", owner = "a", host = "h1", now = 1000, expiresAt = 9000000000000)).toBeTrue(); + expect(lock.tryAcquire(name = "spec:one", owner = "b", host = "h2", now = 2000, expiresAt = 9000000000000)).toBeFalse(); + expect(lock.ownerOf("spec:one")).toBe("a"); + expect(lock.release(name = "spec:one", owner = "b")).toBeFalse("only the owner can release it"); + expect(lock.release(name = "spec:one", owner = "a")).toBeTrue(); + expect(lock.tryAcquire(name = "spec:one", owner = "b", host = "h2", now = 3000, expiresAt = 9000000000000)).toBeTrue(); + }); + + it("takes over an expired lease, and the old owner can neither renew nor release it", function() { + var lock = $lock(); + lock.tryAcquire(name = "spec:two", owner = "a", host = "h1", now = 1000, expiresAt = 5000); + expect(lock.tryAcquire(name = "spec:two", owner = "b", host = "h2", now = 4000, expiresAt = 90000)).toBeFalse("not expired yet"); + expect(lock.tryAcquire(name = "spec:two", owner = "b", host = "h2", now = 6000, expiresAt = 90000)).toBeTrue(); + expect(lock.renew(name = "spec:two", owner = "a", expiresAt = 99000)).toBeFalse(); + expect(lock.release(name = "spec:two", owner = "a")).toBeFalse(); + var row = lock.read("spec:two"); + expect(row.owner).toBe("b"); + expect(row.host).toBe("h2"); + expect(row.expiresAt).toBe(90000); + }); + + it("extends the lease while the owner holds it", function() { + var lock = $lock(); + lock.tryAcquire(name = "spec:three", owner = "a", host = "h1", now = 1000, expiresAt = 5000); + expect(lock.renew(name = "spec:three", owner = "a", expiresAt = 1791039632175)).toBeTrue(); + expect(lock.read("spec:three").expiresAt).toBe(1791039632175, "epoch milliseconds survive the round trip"); + expect(lock.tryAcquire(name = "spec:three", owner = "b", host = "h2", now = 6000, expiresAt = 90000)).toBeFalse(); + }); + + it("builds the Migrator's lock statements for the migrator's own table", function() { + var m = CreateObject("component", "wheels.Migrator").init( + migratePath = "/wheels/tests/_assets/migrator/migrations_4134/", + sqlPath = "/wheels/tests/_assets/migrator/sql_4134/" + ); + var statement = m.$migrationLease().releaseStatement(name = "migrate", owner = "x", expiredBefore = 5); + expect(statement.sql).toBe("DELETE FROM #m.$migrationLockTable()# WHERE lockname = :lockName AND lockowner = :owner AND expiresat < :now"); + expect(statement.params.now.cfsqltype).toBe("cf_sql_bigint"); + }); + + }); + + describe("exclusive jobs", function() { + + beforeEach(function() { + new wheels.Job().$ensureJobTable(); + new wheels.Job().$jobLeaseLock(); + $cleanup(); + application["$leaseSpecRuns"] = 0; + application["$leaseSpecHolder"] = ""; + }); + + afterEach(function() { + $cleanup(); + }); + + it("holds the job class's lease while it runs, and releases it", function() { + var id = $insertJob(queue = "test_lease_run", jobClass = "wheels.tests._assets.jobs.ExclusiveProbeJob"); + var result = new wheels.JobWorker().processNext(queues = "test_lease_run", timeout = 300); + expect(result.success).toBeTrue(result.error); + expect(application["$leaseSpecRuns"]).toBe(1); + expect(Len(application["$leaseSpecHolder"])).toBeGT(0, "the lease must be held during perform()"); + expect($leaseRows("job:wheels.tests._assets.jobs.ExclusiveProbeJob")).toBe(0, "released afterwards"); + expect($row(id).status).toBe("completed"); + }); + + it("doesn't start while another server holds the lease, and waits as pending", function() { + $holdLease("job:wheels.tests._assets.jobs.ExclusiveProbeJob"); + var id = $insertJob(queue = "test_lease_busy", jobClass = "wheels.tests._assets.jobs.ExclusiveProbeJob"); + var worker = new wheels.JobWorker(); + var result = worker.processNext(queues = "test_lease_busy", timeout = 300); + expect(application["$leaseSpecRuns"]).toBe(0, "perform() must not run while another run holds the lease"); + expect(result.skipped).toBeTrue(); + expect(result.deferred).toBeTrue(); + expect(result.success).toBeFalse(); + expect(worker.jobsFailed).toBe(0); + var row = $row(id); + expect(row.status).toBe("pending"); + expect(Val(row.attempts)).toBe(0, "waiting for the lease must not use up an attempt"); + expect(row.due).toBeFalse("it waits before it is tried again"); + expect($leaseOwner("job:wheels.tests._assets.jobs.ExclusiveProbeJob")).toBe("spec-other-server"); + }); + + it("never uses up its retries while the lease stays busy", function() { + $holdLease("key:nightly-report"); + var id = $insertJob(queue = "test_lease_forever", jobClass = "wheels.tests._assets.jobs.StaticKeyJob", maxRetries = 1); + for (var i = 1; i <= 4; i++) { + $makeDue(id); + new wheels.JobWorker().processNext(queues = "test_lease_forever", timeout = 300); + } + var row = $row(id); + expect(row.status).toBe("pending", "a busy lease is not a failure"); + expect(Val(row.attempts)).toBe(0); + }); + + it("counts a busy lease as skipped in processQueue()", function() { + $holdLease("job:wheels.tests._assets.jobs.ExclusiveProbeJob"); + var id = $insertJob(queue = "test_lease_pq", jobClass = "wheels.tests._assets.jobs.ExclusiveProbeJob"); + var result = new wheels.Job().processQueue(queue = "test_lease_pq"); + expect(result.skipped).toBe(1); + expect(result.failed).toBe(0); + expect(application["$leaseSpecRuns"]).toBe(0); + expect(Val($row(id).attempts)).toBe(0); + }); + + it("runs once the other server's lease has expired", function() { + $holdLease(name = "job:wheels.tests._assets.jobs.ExclusiveProbeJob", expiresAt = 1000); + $insertJob(queue = "test_lease_expired", jobClass = "wheels.tests._assets.jobs.ExclusiveProbeJob"); + var result = new wheels.JobWorker().processNext(queues = "test_lease_expired", timeout = 300); + expect(result.success).toBeTrue(result.error); + expect(application["$leaseSpecRuns"]).toBe(1); + }); + + it("finishes a job whose lease was taken over while it ran, and reports the lost lease", function() { + var id = $insertJob(queue = "test_lease_lost", jobClass = "wheels.tests._assets.jobs.LeaseThiefJob"); + var worker = new wheels.JobWorker(); + var result = worker.processNext(queues = "test_lease_lost", timeout = 300); + expect(result.success).toBeTrue("an already-finished job is not failed: " & result.error); + expect(result.leaseLost).toBeTrue(); + expect(worker.leasesLost).toBe(1); + expect($row(id).status).toBe("completed"); + expect($leaseOwner("job:wheels.tests._assets.jobs.LeaseThiefJob")).toBe("spec-thief", "another owner's lease is not released"); + }); + + }); + + describe("job lease names", function() { + + it("uses the concurrencyKey(data) method, then this.concurrencyKey, then the job class", function() { + var bridge = new wheels.Job(); + var keyed = new wheels.tests._assets.jobs.KeyedProbeJob(); + expect(bridge.$jobLeaseName(jobInstance = keyed, jobData = {account = "a1"}, jobClass = "x.KeyedProbeJob")).toBe("key:acct-a1"); + var fixed = new wheels.tests._assets.jobs.StaticKeyJob(); + expect(bridge.$jobLeaseName(jobInstance = fixed, jobData = {}, jobClass = "x.StaticKeyJob")).toBe("key:nightly-report"); + var exclusive = new wheels.tests._assets.jobs.ExclusiveProbeJob(); + expect(bridge.$jobLeaseName(jobInstance = exclusive, jobData = {}, jobClass = "x.ExclusiveProbeJob")).toBe("job:x.ExclusiveProbeJob"); + var plain = new wheels.tests._assets.jobs.ProcessOrdersJob(); + expect(bridge.$jobLeaseName(jobInstance = plain, jobData = {}, jobClass = "x.ProcessOrdersJob")).toBe(""); + }); + + it("hashes a name too long for the lock table", function() { + var name = new wheels.Job().$fitLeaseName(prefix = "key:", value = RepeatString("k", 120)); + expect(Len(name)).toBe(68); + expect(Left(name, 4)).toBe("key:"); + }); + + }); + } + + private any function $lock() { + return new wheels.LeaseLock(table = "wheels_job_locks", datasource = application.wheels.dataSourceName); + } + + private void function $cleanup() { + try { + queryExecute("DELETE FROM wheels_jobs WHERE queue LIKE 'test_lease_%'", {}, {datasource = application.wheels.dataSourceName}); + } catch (any e) { + } + try { + queryExecute( + "DELETE FROM wheels_job_locks WHERE lockname LIKE 'spec:%' OR lockname LIKE 'job:wheels.tests.%' OR lockname = 'key:nightly-report'", + {}, + {datasource = application.wheels.dataSourceName} + ); + } catch (any e) { + } + } + + private void function $holdLease(required string name, numeric expiresAt = 9000000000000) { + $lock().tryAcquire(name = arguments.name, owner = "spec-other-server", host = "other", now = 500, expiresAt = arguments.expiresAt); + } + + private string function $leaseOwner(required string name) { + return $lock().ownerOf(arguments.name); + } + + private numeric function $leaseRows(required string name) { + return $lock().read(arguments.name).held ? 1 : 0; + } + + private string function $insertJob(required string queue, required string jobClass, numeric maxRetries = 3) { + var id = CreateUUID(); + var stamp = DateAdd("s", -5, Now()); + queryExecute( + "INSERT INTO wheels_jobs (id, jobClass, queue, data, priority, status, attempts, maxRetries, runAt, createdAt, updatedAt) + VALUES (:id, :jobClass, :queue, '{}', 0, 'pending', 0, :maxRetries, :runAt, :createdAt, :updatedAt)", + { + id = {value = id, cfsqltype = "cf_sql_varchar"}, + jobClass = {value = arguments.jobClass, cfsqltype = "cf_sql_varchar"}, + queue = {value = arguments.queue, cfsqltype = "cf_sql_varchar"}, + maxRetries = {value = arguments.maxRetries, cfsqltype = "cf_sql_integer"}, + runAt = {value = stamp, cfsqltype = "cf_sql_timestamp"}, + createdAt = {value = stamp, cfsqltype = "cf_sql_timestamp"}, + updatedAt = {value = stamp, cfsqltype = "cf_sql_timestamp"} + }, + {datasource = application.wheels.dataSourceName} + ); + return id; + } + + private void function $makeDue(required string id) { + queryExecute( + "UPDATE wheels_jobs SET runAt = :runAt WHERE id = :id", + { + runAt = {value = DateAdd("s", -5, Now()), cfsqltype = "cf_sql_timestamp"}, + id = {value = arguments.id, cfsqltype = "cf_sql_varchar"} + }, + {datasource = application.wheels.dataSourceName} + ); + } + + private struct function $row(required string id) { + var q = queryExecute( + "SELECT status, attempts FROM wheels_jobs WHERE id = :id", + {id = {value = arguments.id, cfsqltype = "cf_sql_varchar"}}, + {datasource = application.wheels.dataSourceName} + ); + var due = queryExecute( + "SELECT id FROM wheels_jobs WHERE id = :id AND runAt <= :now", + { + id = {value = arguments.id, cfsqltype = "cf_sql_varchar"}, + now = {value = Now(), cfsqltype = "cf_sql_timestamp"} + }, + {datasource = application.wheels.dataSourceName} + ); + return {status = q.recordCount ? q.status : "", attempts = q.recordCount ? q.attempts : 0, due = due.recordCount > 0}; + } + +} From e2bf678e89de583c9b3ded827953999d0fa92698 Mon Sep 17 00:00:00 2001 From: Peter Amiri Date: Wed, 7 Oct 2026 21:20:30 -0700 Subject: [PATCH 2/3] fix(job): concurrencyKeyFor() wins over this.concurrencyKey; catalog lock-table check - The data-driven key method is now concurrencyKeyFor(struct data): a method named concurrencyKey was replaced by a this.concurrencyKey set in config(), so the method couldn't win. Both can now be set, and the method wins. - wheels_job_locks existence is asked of the database catalog. A failing probe SELECT aborted an enclosing PostgreSQL transaction; the probe now only runs outside a Wheels transaction, for an unknown database. - The lease spec no longer names a variable `lock`, a keyword on one engine. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Peter Amiri --- .../4479-exclusive-jobs-lease-lock.added.md | 2 +- vendor/wheels/Job.cfc | 75 +++++++++++++------ .../tests/_assets/jobs/KeyedProbeJob.cfc | 8 +- .../wheels/tests/specs/jobs/JobLeaseSpec.cfc | 40 +++++----- 4 files changed, 82 insertions(+), 43 deletions(-) diff --git a/changelog.d/4479-exclusive-jobs-lease-lock.added.md b/changelog.d/4479-exclusive-jobs-lease-lock.added.md index cf93107ee7..58fe2bc9eb 100644 --- a/changelog.d/4479-exclusive-jobs-lease-lock.added.md +++ b/changelog.d/4479-exclusive-jobs-lease-lock.added.md @@ -1 +1 @@ -- Background jobs can now be made exclusive: set `this.exclusive = true` and a second run of that job class can't start on any server while one is running. A job can instead set `this.concurrencyKey = "..."`, or define `concurrencyKey(struct data)`, to share one lease among the jobs with the same key (for example one run per account). A job that finds its lease busy waits as `pending` without using up an attempt, and counts as skipped. The lease is held in a new `wheels_job_locks` table, created on first use, through `wheels.LeaseLock`, the cross-server lease the migrator already used, now extracted for reuse. Exclusivity is only as strong as the lease: it lasts the job's claim timeout plus a grace period, and a run that outlives it (for example one stuck in a blocked database call) can overlap a second run; that is logged and counted as a lost lease. Servers sharing the table need synchronised clocks. ([#4479](https://github.com/wheels-dev/wheels/issues/4479)) +- Background jobs can now be made exclusive: set `this.exclusive = true` and a second run of that job class can't start on any server while one is running. A job can instead set `this.concurrencyKey = "..."`, or define `concurrencyKeyFor(struct data)` (which takes precedence), to share one lease among the jobs with the same key (for example one run per account). A job that finds its lease busy waits as `pending` without using up an attempt, and counts as skipped. The lease is held in a new `wheels_job_locks` table, created on first use, through `wheels.LeaseLock`, the cross-server lease the migrator already used, now extracted for reuse. Exclusivity is only as strong as the lease: it lasts the job's claim timeout plus a grace period, and a run that outlives it (for example one stuck in a blocked database call) can overlap a second run; that is logged and counted as a lost lease. Servers sharing the table need synchronised clocks. ([#4479](https://github.com/wheels-dev/wheels/issues/4479)) diff --git a/vendor/wheels/Job.cfc b/vendor/wheels/Job.cfc index ba2ff3e28a..b6e907b058 100644 --- a/vendor/wheels/Job.cfc +++ b/vendor/wheels/Job.cfc @@ -38,8 +38,8 @@ component { // rollback alike, so it survives a rollback (failure notices, audit records). this.transactional = true; // 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 concurrencyKey(struct data), to - // share one lease among the jobs with the same key. See $jobLeaseName(). + // 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(). this.exclusive = false; /** @@ -1898,18 +1898,16 @@ component { /** * Internal: the lease a job run must hold, or "" when it needs none. A concurrency key (the - * job's concurrencyKey(data) method, or else its this.concurrencyKey) names a lease shared by - * every job with that key; otherwise this.exclusive = true leases the job class. A name that + * job's concurrencyKeyFor(data) method, or else its this.concurrencyKey) names a lease shared + * by every job with that key; otherwise this.exclusive = true leases the job class. A name that * would not fit the lock table is hashed. */ public string function $jobLeaseName(required any jobInstance, required struct jobData, required string jobClass) { local.key = ""; - if (StructKeyExists(arguments.jobInstance, "concurrencyKey")) { - if (IsSimpleValue(arguments.jobInstance.concurrencyKey)) { - local.key = arguments.jobInstance.concurrencyKey; - } else { - local.key = arguments.jobInstance.concurrencyKey(arguments.jobData); - } + if (StructKeyExists(arguments.jobInstance, "concurrencyKeyFor") && !IsSimpleValue(arguments.jobInstance.concurrencyKeyFor)) { + local.key = arguments.jobInstance.concurrencyKeyFor(arguments.jobData); + } else if (StructKeyExists(arguments.jobInstance, "concurrencyKey") && IsSimpleValue(arguments.jobInstance.concurrencyKey)) { + local.key = arguments.jobInstance.concurrencyKey; } if (Len(Trim(local.key))) { return $fitLeaseName(prefix = "key:", value = Trim(local.key)); @@ -1950,17 +1948,17 @@ component { return local.rv; } local.rv.owner = Len(arguments.claimToken) ? arguments.claimToken : CreateUUID(); - local.lock = $jobLeaseLock(); + local.leaseLock = $jobLeaseLock(); local.claimSeconds = Val(arguments.claimTimeout) > 0 ? Val(arguments.claimTimeout) : 300; - local.now = local.lock.nowMs(); + local.now = local.leaseLock.nowMs(); local.expiresAt = local.now + (local.claimSeconds + Max(60, local.claimSeconds)) * 1000; - if (local.lock.tryAcquire(name = local.rv.name, owner = local.rv.owner, host = $jobHostName(), now = local.now, expiresAt = local.expiresAt)) { + if (local.leaseLock.tryAcquire(name = local.rv.name, owner = local.rv.owner, host = $jobHostName(), now = local.now, expiresAt = local.expiresAt)) { local.rv.held = true; return local.rv; } // Try again once the holder's lease runs out, but at least every 30 seconds: the holder // usually finishes and releases it well before then. - local.holder = local.lock.read(local.rv.name); + local.holder = local.leaseLock.read(local.rv.name); local.remaining = local.holder.held ? Ceiling((local.holder.expiresAt - local.now) / 1000) : 1; local.rv.busy = true; local.rv.retryInSeconds = Min(30, Max(1, local.remaining)); @@ -2031,9 +2029,9 @@ component { */ public any function $jobLeaseLock() { if (!StructKeyExists(variables, "$jobLeaseLockInstance")) { - local.lock = new wheels.LeaseLock(table = "wheels_job_locks", datasource = variables.$datasource); - $ensureJobLockTable(local.lock); - variables.$jobLeaseLockInstance = local.lock; + local.leaseLock = new wheels.LeaseLock(table = "wheels_job_locks", datasource = variables.$datasource); + $ensureJobLockTable(local.leaseLock); + variables.$jobLeaseLockInstance = local.leaseLock; } return variables.$jobLeaseLockInstance; } @@ -2041,18 +2039,18 @@ component { /** * Internal: creates wheels_job_locks when it is missing. */ - public void function $ensureJobLockTable(required any lock) { + public void function $ensureJobLockTable(required any leaseLock) { if ($jobLockTableExists()) { return; } if (Len($outermostWheelsTransaction())) { Throw( type = "Wheels.JobLockTableMissing", - message = "wheels_job_locks doesn't exist yet, and it can't be created inside a transaction. Run the job outside a transaction once, or create the table: #arguments.lock.createTableSql()#" + message = "wheels_job_locks doesn't exist yet, and it can't be created inside a transaction. Run the job outside a transaction once, or create the table: #arguments.leaseLock.createTableSql()#" ); } try { - queryExecute(arguments.lock.createTableSql(), {}, {datasource = variables.$datasource}); + queryExecute(arguments.leaseLock.createTableSql(), {}, {datasource = variables.$datasource}); } catch (any e) { // Tolerate "already exists" from another server creating it at the same time. if (!$jobLockTableExists()) { @@ -2062,9 +2060,22 @@ component { } /** - * Internal: whether wheels_job_locks exists. + * Internal: whether wheels_job_locks exists, asked of the database's catalog. A failing probe + * query would abort an enclosing PostgreSQL transaction, so one only runs outside a + * transaction, for a database without a known catalog query. */ public boolean function $jobLockTableExists() { + local.catalogSql = $jobLockTableCatalogSql(); + if (Len(local.catalogSql)) { + try { + return queryExecute(local.catalogSql, {}, {datasource = variables.$datasource}).recordCount > 0; + } catch (any e) { + // Fall back to the probe below. + } + } + if (Len($outermostWheelsTransaction())) { + return false; + } try { queryExecute("SELECT COUNT(*) AS cnt FROM wheels_job_locks WHERE 1=0", {}, {datasource = variables.$datasource}); return true; @@ -2073,6 +2084,28 @@ component { } } + /** + * Internal: the catalog query that finds wheels_job_locks ("" for an unknown database). + */ + public string function $jobLockTableCatalogSql() { + switch ($detectDatabaseType()) { + case "postgresql": + return "SELECT 1 FROM information_schema.tables WHERE table_schema = current_schema() AND LOWER(table_name) = 'wheels_job_locks'"; + case "mysql": + return "SELECT 1 FROM information_schema.tables WHERE table_schema = DATABASE() AND LOWER(table_name) = 'wheels_job_locks'"; + case "sqlserver": + return "SELECT 1 FROM INFORMATION_SCHEMA.TABLES WHERE LOWER(TABLE_NAME) = 'wheels_job_locks' AND TABLE_SCHEMA = SCHEMA_NAME()"; + case "oracle": + return "SELECT 1 FROM user_tables WHERE UPPER(table_name) = 'WHEELS_JOB_LOCKS'"; + case "h2": + return "SELECT 1 FROM INFORMATION_SCHEMA.TABLES WHERE UPPER(TABLE_NAME) = 'WHEELS_JOB_LOCKS'"; + case "sqlite": + return "SELECT 1 FROM sqlite_master WHERE type = 'table' AND LOWER(name) = 'wheels_job_locks'"; + default: + return ""; + } + } + /** * Retry delay in seconds. `retryBackoff=exponential` is the live schedule * (`baseDelay * 2^attempts`, capped at `maxDelay`). `linear` is a reserved diff --git a/vendor/wheels/tests/_assets/jobs/KeyedProbeJob.cfc b/vendor/wheels/tests/_assets/jobs/KeyedProbeJob.cfc index de91e07158..29043c243e 100644 --- a/vendor/wheels/tests/_assets/jobs/KeyedProbeJob.cfc +++ b/vendor/wheels/tests/_assets/jobs/KeyedProbeJob.cfc @@ -1,9 +1,15 @@ /** * A job whose concurrency key comes from its data (core-suite fixture): one run per account. + * It also sets this.concurrencyKey, which concurrencyKeyFor() takes precedence over. */ component extends="wheels.Job" { - public string function concurrencyKey(struct data = {}) { + function config() { + super.config(); + this.concurrencyKey = "ignored-static-key"; + } + + public string function concurrencyKeyFor(struct data = {}) { return "acct-" & arguments.data.account; } diff --git a/vendor/wheels/tests/specs/jobs/JobLeaseSpec.cfc b/vendor/wheels/tests/specs/jobs/JobLeaseSpec.cfc index 683e1d4eef..4f323c8fbb 100644 --- a/vendor/wheels/tests/specs/jobs/JobLeaseSpec.cfc +++ b/vendor/wheels/tests/specs/jobs/JobLeaseSpec.cfc @@ -20,34 +20,34 @@ component extends="wheels.WheelsTest" { }); it("lets one owner take a free lease, and no other until it is released", function() { - var lock = $lock(); - expect(lock.tryAcquire(name = "spec:one", owner = "a", host = "h1", now = 1000, expiresAt = 9000000000000)).toBeTrue(); - expect(lock.tryAcquire(name = "spec:one", owner = "b", host = "h2", now = 2000, expiresAt = 9000000000000)).toBeFalse(); - expect(lock.ownerOf("spec:one")).toBe("a"); - expect(lock.release(name = "spec:one", owner = "b")).toBeFalse("only the owner can release it"); - expect(lock.release(name = "spec:one", owner = "a")).toBeTrue(); - expect(lock.tryAcquire(name = "spec:one", owner = "b", host = "h2", now = 3000, expiresAt = 9000000000000)).toBeTrue(); + var leaseLock = $lock(); + expect(leaseLock.tryAcquire(name = "spec:one", owner = "a", host = "h1", now = 1000, expiresAt = 9000000000000)).toBeTrue(); + expect(leaseLock.tryAcquire(name = "spec:one", owner = "b", host = "h2", now = 2000, expiresAt = 9000000000000)).toBeFalse(); + expect(leaseLock.ownerOf("spec:one")).toBe("a"); + expect(leaseLock.release(name = "spec:one", owner = "b")).toBeFalse("only the owner can release it"); + expect(leaseLock.release(name = "spec:one", owner = "a")).toBeTrue(); + expect(leaseLock.tryAcquire(name = "spec:one", owner = "b", host = "h2", now = 3000, expiresAt = 9000000000000)).toBeTrue(); }); it("takes over an expired lease, and the old owner can neither renew nor release it", function() { - var lock = $lock(); - lock.tryAcquire(name = "spec:two", owner = "a", host = "h1", now = 1000, expiresAt = 5000); - expect(lock.tryAcquire(name = "spec:two", owner = "b", host = "h2", now = 4000, expiresAt = 90000)).toBeFalse("not expired yet"); - expect(lock.tryAcquire(name = "spec:two", owner = "b", host = "h2", now = 6000, expiresAt = 90000)).toBeTrue(); - expect(lock.renew(name = "spec:two", owner = "a", expiresAt = 99000)).toBeFalse(); - expect(lock.release(name = "spec:two", owner = "a")).toBeFalse(); - var row = lock.read("spec:two"); + var leaseLock = $lock(); + leaseLock.tryAcquire(name = "spec:two", owner = "a", host = "h1", now = 1000, expiresAt = 5000); + expect(leaseLock.tryAcquire(name = "spec:two", owner = "b", host = "h2", now = 4000, expiresAt = 90000)).toBeFalse("not expired yet"); + expect(leaseLock.tryAcquire(name = "spec:two", owner = "b", host = "h2", now = 6000, expiresAt = 90000)).toBeTrue(); + expect(leaseLock.renew(name = "spec:two", owner = "a", expiresAt = 99000)).toBeFalse(); + expect(leaseLock.release(name = "spec:two", owner = "a")).toBeFalse(); + var row = leaseLock.read("spec:two"); expect(row.owner).toBe("b"); expect(row.host).toBe("h2"); expect(row.expiresAt).toBe(90000); }); it("extends the lease while the owner holds it", function() { - var lock = $lock(); - lock.tryAcquire(name = "spec:three", owner = "a", host = "h1", now = 1000, expiresAt = 5000); - expect(lock.renew(name = "spec:three", owner = "a", expiresAt = 1791039632175)).toBeTrue(); - expect(lock.read("spec:three").expiresAt).toBe(1791039632175, "epoch milliseconds survive the round trip"); - expect(lock.tryAcquire(name = "spec:three", owner = "b", host = "h2", now = 6000, expiresAt = 90000)).toBeFalse(); + var leaseLock = $lock(); + leaseLock.tryAcquire(name = "spec:three", owner = "a", host = "h1", now = 1000, expiresAt = 5000); + expect(leaseLock.renew(name = "spec:three", owner = "a", expiresAt = 1791039632175)).toBeTrue(); + expect(leaseLock.read("spec:three").expiresAt).toBe(1791039632175, "epoch milliseconds survive the round trip"); + expect(leaseLock.tryAcquire(name = "spec:three", owner = "b", host = "h2", now = 6000, expiresAt = 90000)).toBeFalse(); }); it("builds the Migrator's lock statements for the migrator's own table", function() { @@ -148,7 +148,7 @@ component extends="wheels.WheelsTest" { describe("job lease names", function() { - it("uses the concurrencyKey(data) method, then this.concurrencyKey, then the job class", function() { + it("uses the concurrencyKeyFor(data) method over this.concurrencyKey, then the job class", function() { var bridge = new wheels.Job(); var keyed = new wheels.tests._assets.jobs.KeyedProbeJob(); expect(bridge.$jobLeaseName(jobInstance = keyed, jobData = {account = "a1"}, jobClass = "x.KeyedProbeJob")).toBe("key:acct-a1"); From 37fb206ead604521a69b31bfa25a9c5e10c2674e Mon Sep 17 00:00:00 2001 From: Peter Amiri Date: Thu, 8 Oct 2026 09:24:13 -0700 Subject: [PATCH 3/3] feat(job): heartbeat() renews an exclusive job's lease Now that heartbeat() exists, an exclusive run (this.exclusive or a concurrency key) hands its lease to the job instance, and each heartbeat extends it by the lease's window from now. A lease already taken over is logged once and left alone; the run's end still reports it as lost. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Peter Amiri --- vendor/wheels/Job.cfc | 49 ++++++++++++++++++- .../tests/_assets/jobs/HeartbeatLeaseJob.cfc | 28 +++++++++++ .../wheels/tests/specs/jobs/JobLeaseSpec.cfc | 11 +++++ 3 files changed, 87 insertions(+), 1 deletion(-) create mode 100644 vendor/wheels/tests/_assets/jobs/HeartbeatLeaseJob.cfc diff --git a/vendor/wheels/Job.cfc b/vendor/wheels/Job.cfc index 0478d679db..b2a5344b24 100644 --- a/vendor/wheels/Job.cfc +++ b/vendor/wheels/Job.cfc @@ -78,7 +78,9 @@ component { * heartbeats on time is never reaped and run a second time. 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. + * worker (perform() called directly) it does nothing. For an exclusive job (this.exclusive + * or a concurrency key) it also extends the run's lease, so a run that keeps heartbeating + * keeps its lease. */ public void function heartbeat() { if (!StructKeyExists(variables, "$claim") || !Len(variables.$claim.jobId)) { @@ -104,6 +106,8 @@ component { extendedInfo = "Stop working and return from perform(): another attempt owns this job now, and this attempt's result will be discarded." ); } + // An exclusive run's lease lives as long as its heartbeats do. + $renewJobLease(); } /** @@ -114,6 +118,45 @@ component { variables.$claim = {jobId = arguments.jobId, claimToken = arguments.claimToken}; } + /** + * Internal: the exclusive lease this instance's run holds (this.exclusive or a concurrency + * key), so heartbeat() extends it too. Set by $runPerformExclusively() once the lease is taken. + */ + public void function $setLeaseContext(required string name, required string owner, required numeric windowSeconds) { + variables.$lease = {name = arguments.name, owner = arguments.owner, windowSeconds = arguments.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 + * end of the run reports it as a lost lease. + */ + public void function $renewJobLease() { + if (!StructKeyExists(variables, "$lease")) { + return; + } + var leaseState = {renewed = true}; + try { + local.leaseLock = $jobLeaseLock(); + leaseState.renewed = local.leaseLock.renew( + name = variables.$lease.name, + owner = variables.$lease.owner, + expiresAt = local.leaseLock.nowMs() + variables.$lease.windowSeconds * 1000 + ); + } catch (any e) { + writeLog(text = "Job lease '#variables.$lease.name#' could not be renewed: #e.message#", type = "error", file = "wheels_jobs"); + return; + } + if (!leaseState.renewed && !StructKeyExists(variables.$lease, "lostLogged")) { + variables.$lease.lostLogged = true; + writeLog( + text = "Job lease '#variables.$lease.name#' was lost before a heartbeat could renew it: another run may be running at the same time.", + type = "warning", + file = "wheels_jobs" + ); + } + } + /** * Returns a model class object, so perform() can call model("User") as a controller does. * Delegates to application.wo.model(); a job extends no controller or model base class. @@ -2233,6 +2276,10 @@ component { if (local.lease.busy) { return {success = false, error = "", timedOut = false, busy = true, retryInSeconds = local.lease.retryInSeconds, leaseName = local.lease.name, leaseLost = false}; } + if (local.lease.held) { + // So heartbeat() inside perform() renews the lease as well as the claim. + arguments.jobInstance.$setLeaseContext(name = local.lease.name, owner = local.lease.owner, windowSeconds = local.lease.windowSeconds); + } local.outcome = $runPerformWithTimeout( jobInstance = arguments.jobInstance, jobData = arguments.jobData, diff --git a/vendor/wheels/tests/_assets/jobs/HeartbeatLeaseJob.cfc b/vendor/wheels/tests/_assets/jobs/HeartbeatLeaseJob.cfc new file mode 100644 index 0000000000..8064cf74da --- /dev/null +++ b/vendor/wheels/tests/_assets/jobs/HeartbeatLeaseJob.cfc @@ -0,0 +1,28 @@ +/** + * An exclusive job that heartbeats (core-suite fixture). perform() records its lease's expiry + * before and after a heartbeat, so the lease specs can see heartbeat() extend it. + */ +component extends="wheels.Job" { + + function config() { + super.config(); + this.exclusive = true; + } + + public void function perform(struct data = {}) { + application["$leaseSpecBefore"] = $leaseExpiry(); + Sleep(50); + heartbeat(); + application["$leaseSpecAfter"] = $leaseExpiry(); + } + + private numeric function $leaseExpiry() { + local.rows = queryExecute( + "SELECT expiresat FROM wheels_job_locks WHERE lockname = :name", + {name = {value = "job:wheels.tests._assets.jobs.HeartbeatLeaseJob", cfsqltype = "cf_sql_varchar"}}, + {datasource = application.wheels.dataSourceName} + ); + return local.rows.recordCount ? Val(local.rows.expiresat) : 0; + } + +} diff --git a/vendor/wheels/tests/specs/jobs/JobLeaseSpec.cfc b/vendor/wheels/tests/specs/jobs/JobLeaseSpec.cfc index 4f323c8fbb..4bf66c894a 100644 --- a/vendor/wheels/tests/specs/jobs/JobLeaseSpec.cfc +++ b/vendor/wheels/tests/specs/jobs/JobLeaseSpec.cfc @@ -133,6 +133,17 @@ component extends="wheels.WheelsTest" { expect(application["$leaseSpecRuns"]).toBe(1); }); + it("extends the run's lease when the job heartbeats", function() { + application["$leaseSpecBefore"] = 0; + application["$leaseSpecAfter"] = 0; + var id = $insertJob(queue = "test_lease_heartbeat", jobClass = "wheels.tests._assets.jobs.HeartbeatLeaseJob"); + var result = new wheels.JobWorker().processNext(queues = "test_lease_heartbeat", timeout = 300); + expect(result.success).toBeTrue(result.error); + expect(application["$leaseSpecBefore"]).toBeGT(0, "the lease is held while perform() runs"); + expect(application["$leaseSpecAfter"]).toBeGT(application["$leaseSpecBefore"], "heartbeat() moves the lease's expiry forward"); + expect($leaseRows("job:wheels.tests._assets.jobs.HeartbeatLeaseJob")).toBe(0, "released afterwards"); + }); + it("finishes a job whose lease was taken over while it ran, and reports the lost lease", function() { var id = $insertJob(queue = "test_lease_lost", jobClass = "wheels.tests._assets.jobs.LeaseThiefJob"); var worker = new wheels.JobWorker();