From e805d6b054c0b11e4900206e75aa314fde0cd522 Mon Sep 17 00:00:00 2001 From: Peter Amiri Date: Wed, 7 Oct 2026 21:39:42 -0700 Subject: [PATCH 1/5] feat(cli): wheels jobs install and jobsAutoCreateTables wheels.JobSchema defines the job schema once. Auto-create builds its CREATE TABLE, ALTERs and indexes from it, and answers table/column/index checks from the database catalog (never a failing probe, so they are safe in a migration's transaction). `wheels jobs install` writes app/migrator/migrations/_CreateWheelsJobTables.cfc from the running app's schema: migrator DSL that creates the tables or tops up auto-created ones, idempotently. set(jobsAutoCreateTables = false) stops every runtime CREATE and ALTER of the job tables. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Peter Amiri --- changelog.d/4484-jobs-install.added.md | 1 + cli/lucli/Module.cfc | 67 ++- .../tests/specs/commands/JobsCommandSpec.cfc | 45 ++ vendor/wheels/Job.cfc | 141 +++--- vendor/wheels/JobSchema.cfc | 434 ++++++++++++++++++ vendor/wheels/public/CliBridge.cfc | 21 +- .../wheels/tests/specs/cli/CliBridgeSpec.cfc | 2 +- .../wheels/tests/specs/jobs/JobSchemaSpec.cfc | 211 +++++++++ 8 files changed, 826 insertions(+), 96 deletions(-) create mode 100644 changelog.d/4484-jobs-install.added.md create mode 100644 vendor/wheels/JobSchema.cfc create mode 100644 vendor/wheels/tests/specs/jobs/JobSchemaSpec.cfc diff --git a/changelog.d/4484-jobs-install.added.md b/changelog.d/4484-jobs-install.added.md new file mode 100644 index 0000000000..8d225a3baf --- /dev/null +++ b/changelog.d/4484-jobs-install.added.md @@ -0,0 +1 @@ +- `wheels jobs install` writes a migration that creates the background-job tables, so an app can apply the jobs schema once, reviewed and under migration control, instead of having every server create and alter it at startup. The migration adopts tables Wheels already created automatically, adding only what they lack, and running it again changes nothing. Apps that use it set `set(jobsAutoCreateTables = false)`, which stops the framework from creating or altering the job tables at runtime; a missing table or column is then reported with a pointer to `wheels jobs install`. Auto-create stays the default. The job schema is now defined once (`wheels.JobSchema`), so the migration and auto-create build the same tables. ([#4484](https://github.com/wheels-dev/wheels/issues/4484)) diff --git a/cli/lucli/Module.cfc b/cli/lucli/Module.cfc index 5fb99ac37b..c08826a8e8 100644 --- a/cli/lucli/Module.cfc +++ b/cli/lucli/Module.cfc @@ -586,7 +586,7 @@ component extends="modules.BaseModule" { private any function jobsArgSpec() { return new services.ArgSpec() - .positional(name = "action", default = "status", description = "work (long-lived worker loop), status (queue snapshot) or enqueue (add a job). Defaults to status") + .positional(name = "action", default = "status", description = "work (long-lived worker loop), status (queue snapshot), enqueue (add a job) or install (write a migration that creates the job tables). Defaults to status") .positional(name = "job", default = "", description = "enqueue only: the job class under app/jobs/, e.g. SendWelcomeEmailJob or billing.InvoiceJob") .option(name = "queue", default = "", description = "work: comma-delimited queue names to process in order. status: single queue to filter by. enqueue: the queue to put the job on (default: the job's own). Empty = all queues") .option(name = "data", default = "", description = "enqueue only: the job's data, a JSON object passed to perform()") @@ -597,7 +597,8 @@ component extends="modules.BaseModule" { .option(name = "max-jobs", default = 0, type = "numeric", description = "work only: stop after this many jobs (successes + failures count). 0 = run until stopped") .flag(name = "stop-when-empty", default = false, description = "work only: exit when a poll finds no job ready to run, instead of waiting for more. For one-shot batches from cron or CI; combines with --max-jobs") .flag(name = "quiet", default = false, description = "work only: suppress per-job completion output, only print failures") - .option(name = "format", default = "table", description = "status and enqueue: output format, table or json"); + .option(name = "format", default = "table", description = "status and enqueue: output format, table or json") + .flag(name = "force", default = false, description = "install only: overwrite the jobs migration this command wrote before"); } private any function dbArgSpec() { @@ -5831,7 +5832,8 @@ component extends="modules.BaseModule" { data = trim(parsed.data), priority = trim(parsed.priority), "in" = trim(parsed["in"]), - at = trim(parsed.at) + at = trim(parsed.at), + force = parsed.force }; if (!len(opts.action)) { opts.action = "status"; @@ -5858,7 +5860,7 @@ component extends="modules.BaseModule" { } /** - * hint: Background job queue — `work` runs a long-lived worker loop, `status` prints per-queue counts (--format=json for machines), `enqueue ` adds a job now or after a delay. retry/purge/monitor are tracked follow-ups (issue 3090). + * hint: Background job queue — `work` runs a long-lived worker loop, `status` prints per-queue counts (--format=json for machines), `enqueue ` adds a job now or after a delay, `install` writes a migration that creates the job tables. retry/purge/monitor are tracked follow-ups (issue 3090). */ public string function jobs() { var opts = $parseJobsArgs(structuredArgs(arguments)); @@ -5870,6 +5872,8 @@ component extends="modules.BaseModule" { return runJobsStatus(opts); case "enqueue": return runJobsEnqueue(opts); + case "install": + return runJobsInstall(opts); // The framework bridge (vendor/wheels/public/views/cli.cfm) already // implements jobsRetry/jobsPurge/jobsMonitor — the CLI verbs are // deliberate follow-ups tracked in ##3090. Fail loudly with the @@ -5888,7 +5892,7 @@ component extends="modules.BaseModule" { ); default: out("Unknown jobs action: #opts.action#", "red"); - out("Usage: wheels jobs [work|status|enqueue ] [--queue=] [--interval=] [--max-jobs=] [--stop-when-empty] [--quiet] [--format=table|json]"); + out("Usage: wheels jobs [work|status|enqueue |install [--force]] [--queue=] [--interval=] [--max-jobs=] [--stop-when-empty] [--quiet] [--format=table|json]"); throw(type = "Wheels.InvalidArguments", message = "Unknown jobs action: #opts.action#"); } } @@ -5985,6 +5989,59 @@ component extends="modules.BaseModule" { } } + /** + * `wheels jobs install`: write app/migrator/migrations/_CreateWheelsJobTables.cfc, + * the migration that creates the job tables (or tops up the ones Wheels already created), so a + * multi-server app applies the jobs schema once, under migration control. The source comes from + * the running app (wheels.JobSchema) so it matches what auto-create builds; nothing touches the + * database until `wheels migrate latest`. + */ + private string function runJobsInstall(required struct opts) { + var serverPort = $requireRunningServer( + hints = ["Start one with: wheels start (the migration is generated from the app's own framework version)"] + ); + var httpResult = ""; + try { + httpResult = makeHttpRequest("#$serverUrlBase(serverPort)#/wheels/cli?command=jobsInstallSource&format=json"); + } catch (any httpErr) { + throw( + type = "Wheels.Cli.CommandFailed", + message = "Jobs install failed (connection error): #httpErr.message#", + detail = httpErr.detail ?: "" + ); + } + var result = parseCliResponse(httpResult, "Jobs install"); + if (!len(result.source ?: "")) { + throw(type = "Wheels.Cli.CommandFailed", message = "Jobs install: the server returned no migration source."); + } + + var migrationDir = variables.projectRoot & "/app/migrator/migrations"; + $ensureProjectDirectory(migrationDir); + var existing = directoryList(migrationDir, false, "name", "*_#result.migrationName#.cfc"); + var fileName = ""; + if (arrayLen(existing)) { + if (!arguments.opts.force) { + out("The jobs migration already exists: app/migrator/migrations/#existing[1]#", "yellow"); + out("Run it with: wheels migrate latest. To rewrite it from this framework version, pass --force."); + return ""; + } + // Same file name, so a migration that already ran keeps its version. + fileName = existing[1]; + } else { + fileName = getService("helpers").generateMigrationTimestamp() & "_" & result.migrationName & ".cfc"; + } + fileWrite(migrationDir & "/" & fileName, result.source); + + out("Wrote app/migrator/migrations/#fileName#", "green"); + out(""); + out("Next:"); + out(" 1. Review it, then apply it: wheels migrate latest"); + out(" It creates the job tables, or adds what existing ones are missing."); + out(" 2. In config/settings.cfm: set(jobsAutoCreateTables = false)"); + out(" so servers never change the job tables at startup."); + return ""; + } + /** * Print why the worker loop is ending, plus its counts, and return "" for * runJobsWork() to return. diff --git a/cli/lucli/tests/specs/commands/JobsCommandSpec.cfc b/cli/lucli/tests/specs/commands/JobsCommandSpec.cfc index fc3819d735..e4a9c04b81 100644 --- a/cli/lucli/tests/specs/commands/JobsCommandSpec.cfc +++ b/cli/lucli/tests/specs/commands/JobsCommandSpec.cfc @@ -60,6 +60,16 @@ component extends="wheels.wheelstest.system.BaseSpec" { return serializeJSON({success: true, jobResult: {skipped: true}}); } + // A Module whose `jobs install` gets `source` back from a canned jobsInstallSource response. + private any function installModule(required string source) { + var m = new cli.lucli.Module(cwd = variables.tempRoot); + prepareMock(m); + m.$("out"); + m.$(method = "$requireRunningServer", returns = 61999); + m.$("makeHttpRequest", serializeJSON({success: true, migrationName: "CreateWheelsJobTables", source: arguments.source})); + return m; + } + private string function printed(required any m) { var said = ""; for (var call in arguments.m.$callLog().out) { @@ -222,6 +232,41 @@ component extends="wheels.wheelstest.system.BaseSpec" { }); + describe("install", () => { + + it("parses install and --force", () => { + expect(mod.$parseJobsArgs({arg1 = "install"}).action).toBe("install"); + expect(mod.$parseJobsArgs({arg1 = "install"}).force).toBeFalse(); + expect(mod.$parseJobsArgs({arg1 = "install", force = "true"}).force).toBeTrue(); + }); + + it("writes the migration the server generates, once, and rewrites it in place with --force", () => { + var dir = variables.tempRoot & "/app/migrator/migrations"; + if (directoryExists(dir)) { + for (var old in directoryList(dir, false, "path", "*_CreateWheelsJobTables.cfc")) { + fileDelete(old); + } + } + var m = installModule("// first"); + m.jobs(arg1 = "install"); + var written = directoryList(dir, false, "name", "*_CreateWheelsJobTables.cfc"); + expect(arrayLen(written)).toBe(1); + expect(fileRead(dir & "/" & written[1])).toBe("// first"); + expect(printed(m)).toInclude("jobsAutoCreateTables = false"); + + var again = installModule("// second"); + again.jobs(arg1 = "install"); + expect(fileRead(dir & "/" & written[1])).toBe("// first", "without --force an existing migration is left alone"); + expect(printed(again)).toInclude("already exists"); + + var forced = installModule("// third"); + forced.jobs(arg1 = "install", force = "true"); + expect(directoryList(dir, false, "name", "*_CreateWheelsJobTables.cfc")).toBe(written, "rewritten under the same name and version"); + expect(fileRead(dir & "/" & written[1])).toBe("// third"); + }); + + }); + describe("MCP surface", () => { it("hides the jobs command from MCP tools/list", () => { diff --git a/vendor/wheels/Job.cfc b/vendor/wheels/Job.cfc index e9ea446474..1848cd5662 100644 --- a/vendor/wheels/Job.cfc +++ b/vendor/wheels/Job.cfc @@ -365,7 +365,7 @@ component { Throw( type = "Wheels.Job.UniqueKeyUnavailable", message = "Job '#arguments.jobClass#' was enqueued with a uniqueKey, but wheels_jobs can't enforce one yet, so it was not enqueued.", - extendedInfo = "wheels_jobs needs a uniqueKey column (VARCHAR(255), nullable) with a unique index. The framework adds them on a worker poll or an enqueue(uniqueKey=...) made outside a transaction, unless the database user can't ALTER the table or another instance is upgrading it right now. Add them manually: ALTER TABLE wheels_jobs ADD uniqueKey VARCHAR(255); UPDATE wheels_jobs SET uniqueKey = id; CREATE UNIQUE INDEX idx_wjobs_unique_key ON wheels_jobs (uniqueKey) (on SQL Server add WHERE uniqueKey IS NOT NULL)." + extendedInfo = ($jobSchema().autoCreateEnabled() ? "" : $jobSchema().missingSchemaMessage("The wheels_jobs uniqueKey column or index") & " ") & "wheels_jobs needs a uniqueKey column (VARCHAR(255), nullable) with a unique index. The framework adds them on a worker poll or an enqueue(uniqueKey=...) made outside a transaction, unless the database user can't ALTER the table or another instance is upgrading it right now. Add them manually: ALTER TABLE wheels_jobs ADD uniqueKey VARCHAR(255); UPDATE wheels_jobs SET uniqueKey = id; CREATE UNIQUE INDEX idx_wjobs_unique_key ON wheels_jobs (uniqueKey) (on SQL Server add WHERE uniqueKey IS NOT NULL)." ); } @@ -1063,6 +1063,11 @@ component { try { // Check if table already exists by querying it queryExecute("SELECT COUNT(*) AS cnt FROM wheels_jobs WHERE 1=0", {}, {datasource = variables.$datasource}); + // With auto-create off the schema is the app's (its `wheels jobs install` migration): + // no column upgrades run here. + if (!$jobSchema().autoCreateEnabled()) { + return true; + } // Table exists — make sure the claimTimeout column exists too (#3989). Probed on // every call and never cached: a shared or rebuilt dev DB can lose it (see #2780). $ensureClaimTimeoutColumn(); @@ -1073,70 +1078,34 @@ component { // Table doesn't exist — create it } + if (!$jobSchema().autoCreateEnabled()) { + writeLog(text = $jobSchema().missingSchemaMessage("The wheels_jobs table"), type = "error", file = "wheels_jobs"); + return false; + } + try { // Detect actual database type from the datasource via JDBC metadata. // We query the datasource directly rather than using application.wheels.adapterName // because the adapter may have been detected from a different datasource. local.dbType = $detectDatabaseType(); - - // Use database-appropriate types - if (local.dbType == "oracle") { - local.varcharType = "VARCHAR2"; - local.textType = "CLOB"; - local.datetimeType = "TIMESTAMP"; - } else if (local.dbType == "postgresql") { - local.varcharType = "VARCHAR"; - local.textType = "TEXT"; - local.datetimeType = "TIMESTAMP"; - } else if (local.dbType == "h2") { - local.varcharType = "VARCHAR"; - local.textType = "CLOB"; - local.datetimeType = "TIMESTAMP"; - } else { - local.varcharType = "VARCHAR"; - local.textType = "TEXT"; - local.datetimeType = "DATETIME"; - } - - queryExecute(" - CREATE TABLE wheels_jobs ( - id #local.varcharType#(36) NOT NULL PRIMARY KEY, - jobClass #local.varcharType#(255) NOT NULL, - queue #local.varcharType#(100) DEFAULT 'default' NOT NULL, - data #local.textType#, - priority INT DEFAULT 0 NOT NULL, - status #local.varcharType#(20) DEFAULT 'pending' NOT NULL, - attempts INT DEFAULT 0 NOT NULL, - maxRetries INT DEFAULT 3 NOT NULL, - claimTimeout INT, - claimToken #local.varcharType#(36), - claimedBy #local.varcharType#(128), - uniqueKey #local.varcharType#(255), - lastError #local.textType#, - runAt #local.datetimeType#, - completedAt #local.datetimeType#, - failedAt #local.datetimeType#, - createdAt #local.datetimeType#, - updatedAt #local.datetimeType# - ) - ", {}, {datasource = variables.$datasource}); - - // Add indexes for efficient queue processing - try { - queryExecute("CREATE INDEX idx_wjobs_processing ON wheels_jobs (status, runAt, priority)", {}, {datasource = variables.$datasource}); - queryExecute("CREATE INDEX idx_wjobs_queue ON wheels_jobs (queue, status)", {}, {datasource = variables.$datasource}); - queryExecute("CREATE INDEX idx_wjobs_cleanup ON wheels_jobs (status, completedAt)", {}, {datasource = variables.$datasource}); - } catch (any indexError) { - // Indexes are optional — don't fail if they can't be created - } - // The uniqueKey index is not optional: it is what de-duplicates enqueue(uniqueKey=). - // If it can't be built now, the next $ensureJobTable() retries it (the column is there, - // the index isn't), and enqueue(uniqueKey=) refuses to run without it. - try { - queryExecute($uniqueKeyIndexSql(), {}, {datasource = variables.$datasource}); - $recordUniqueKeyIndexVerified(); - } catch (any uniqueIndexError) { - writeLog(text = "Could not create the wheels_jobs uniqueKey index: #uniqueIndexError.message#", type = "error", file = "wheels_jobs"); + local.schema = $jobSchema(); + queryExecute(local.schema.createTableSql(tableName = "wheels_jobs", dbType = local.dbType), {}, {datasource = variables.$datasource}); + + // Indexes for efficient queue processing are optional: don't fail if one can't be + // created. The uniqueKey index is not optional: it is what de-duplicates + // enqueue(uniqueKey=). If it can't be built now, the next $ensureJobTable() retries it + // (the column is there, the index isn't), and enqueue(uniqueKey=) refuses to run without it. + for (local.index in local.schema.tableDef("wheels_jobs").indexes) { + try { + queryExecute(local.schema.indexSql(tableName = "wheels_jobs", indexName = local.index.name, dbType = local.dbType), {}, {datasource = variables.$datasource}); + if (local.index.name == "idx_wjobs_unique_key") { + $recordUniqueKeyIndexVerified(); + } + } catch (any indexError) { + if (!local.index.optional) { + writeLog(text = "Could not create the wheels_jobs index #local.index.name#: #indexError.message#", type = "error", file = "wheels_jobs"); + } + } } writeLog(text = "Auto-created wheels_jobs table", type = "information", file = "wheels_jobs"); @@ -1157,6 +1126,9 @@ component { // DDL inside an open transaction makes MySQL and Oracle commit the caller's work, so the // column is never added there (an INSERT failure's table-ensure runs inside the caller's // transaction). A worker poll or a call outside a transaction adds it. + if (!$jobSchema().autoCreateEnabled()) { + return; + } if (Len($outermostWheelsTransaction())) { return; } @@ -1242,14 +1214,7 @@ component { * Oracle takes a parenthesised column list; everything else accepts ADD COLUMN. */ public string function $claimTimeoutAlterSql() { - local.dbType = $detectDatabaseType(); - if (local.dbType == "oracle") { - return "ALTER TABLE wheels_jobs ADD (claimTimeout NUMBER(10))"; - } - if (local.dbType == "sqlserver") { - return "ALTER TABLE wheels_jobs ADD claimTimeout INT"; - } - return "ALTER TABLE wheels_jobs ADD COLUMN claimTimeout INT"; + return $jobSchema().addColumnSql(tableName = "wheels_jobs", columnName = "claimTimeout", dbType = $detectDatabaseType()); } /** @@ -1281,6 +1246,9 @@ component { // DDL inside an open transaction makes MySQL and Oracle commit the caller's work, so the // columns are never added there (an INSERT failure's table-ensure runs inside the caller's // transaction). A worker poll or a call outside a transaction adds them. + if (!$jobSchema().autoCreateEnabled()) { + return; + } if (Len($outermostWheelsTransaction())) { return; } @@ -1329,14 +1297,7 @@ component { * The per-database "ADD VARCHAR(n)" DDL for a claim-fencing column. */ public string function $claimTokenAlterSql(required string columnName, required numeric size) { - local.dbType = $detectDatabaseType(); - if (local.dbType == "oracle") { - return "ALTER TABLE wheels_jobs ADD (#arguments.columnName# VARCHAR2(#arguments.size#))"; - } - if (local.dbType == "sqlserver") { - return "ALTER TABLE wheels_jobs ADD #arguments.columnName# VARCHAR(#arguments.size#)"; - } - return "ALTER TABLE wheels_jobs ADD COLUMN #arguments.columnName# VARCHAR(#arguments.size#)"; + return $jobSchema().addColumnSql(tableName = "wheels_jobs", columnName = arguments.columnName, dbType = $detectDatabaseType()); } /** @@ -1373,6 +1334,16 @@ component { return variables.$claimTokenColumnsPresent; } + /** + * The job schema (wheels.JobSchema) for this instance's datasource, memoised per instance. + */ + public any function $jobSchema() { + if (!StructKeyExists(variables, "$jobSchemaInstance")) { + variables.$jobSchemaInstance = new wheels.JobSchema(datasource = variables.$datasource); + } + return variables.$jobSchemaInstance; + } + /** * A fresh, unique token for one claim of one job. */ @@ -1442,6 +1413,9 @@ component { // DDL inside an open transaction makes MySQL and Oracle commit the caller's work, so the // upgrade never runs there, whoever calls this (a keyed enqueue, or the table-ensure an // INSERT failure triggers). A worker poll or an enqueue outside a transaction does it. + if (!$jobSchema().autoCreateEnabled()) { + return; + } if (Len($outermostWheelsTransaction())) { return; } @@ -1622,14 +1596,7 @@ component { * The per-database "ADD uniqueKey" DDL (nullable). */ public string function $uniqueKeyAlterSql() { - local.dbType = $detectDatabaseType(); - if (local.dbType == "oracle") { - return "ALTER TABLE wheels_jobs ADD (uniqueKey VARCHAR2(255))"; - } - if (local.dbType == "sqlserver") { - return "ALTER TABLE wheels_jobs ADD uniqueKey VARCHAR(255)"; - } - return "ALTER TABLE wheels_jobs ADD COLUMN uniqueKey VARCHAR(255)"; + return $jobSchema().addColumnSql(tableName = "wheels_jobs", columnName = "uniqueKey", dbType = $detectDatabaseType()); } /** @@ -1637,11 +1604,7 @@ component { * single NULL, so it is filtered to the non-NULL keys there. */ public string function $uniqueKeyIndexSql() { - local.sql = "CREATE UNIQUE INDEX idx_wjobs_unique_key ON wheels_jobs (uniqueKey)"; - if ($detectDatabaseType() == "sqlserver") { - local.sql &= " WHERE uniqueKey IS NOT NULL"; - } - return local.sql; + return $jobSchema().indexSql(tableName = "wheels_jobs", indexName = "idx_wjobs_unique_key", dbType = $detectDatabaseType()); } /** diff --git a/vendor/wheels/JobSchema.cfc b/vendor/wheels/JobSchema.cfc new file mode 100644 index 0000000000..5f5b47d224 --- /dev/null +++ b/vendor/wheels/JobSchema.cfc @@ -0,0 +1,434 @@ +/** + * The background-job schema, defined once. Auto-create (Job.$ensureJobTable() and the column + * upgrades) builds its DDL from it, and `wheels jobs install` writes a migration from it, so the + * two build the same tables, columns and indexes. + * + * Each table lists its columns in creation order, as {name, type, limit, nullable, default, + * primaryKey}, with type one of string, integer, text or datetime, and its indexes as + * {name, columns, unique, optional}. An optional index is a performance index that auto-create + * tolerates failing to build; a non-optional one is relied on (the uniqueKey index de-duplicates + * enqueue). On SQL Server the uniqueKey index is filtered to non-NULL keys, because a SQL Server + * unique index otherwise admits a single NULL. + */ +component { + + /** + * @datasource The datasource the catalog checks run against. Defaults to the app's. + */ + public any function init(string datasource = "") { + variables.datasource = arguments.datasource; + if (!Len(variables.datasource) && StructKeyExists(application, "wheels") && StructKeyExists(application.wheels, "dataSourceName")) { + variables.datasource = application.wheels.dataSourceName; + } + return this; + } + + /** + * The job tables, in creation order. + */ + public array function tables() { + return [ + { + name = "wheels_jobs", + columns = [ + {name = "id", type = "string", limit = 36, nullable = false, primaryKey = true}, + {name = "jobClass", type = "string", limit = 255, nullable = false}, + {name = "queue", type = "string", limit = 100, nullable = false, default = "default"}, + {name = "data", type = "text", nullable = true}, + {name = "priority", type = "integer", nullable = false, default = 0}, + {name = "status", type = "string", limit = 20, nullable = false, default = "pending"}, + {name = "attempts", type = "integer", nullable = false, default = 0}, + {name = "maxRetries", type = "integer", nullable = false, default = 3}, + {name = "claimTimeout", type = "integer", nullable = true}, + {name = "claimToken", type = "string", limit = 36, nullable = true}, + {name = "claimedBy", type = "string", limit = 128, nullable = true}, + {name = "uniqueKey", type = "string", limit = 255, nullable = true}, + {name = "lastError", type = "text", nullable = true}, + {name = "runAt", type = "datetime", nullable = true}, + {name = "completedAt", type = "datetime", nullable = true}, + {name = "failedAt", type = "datetime", nullable = true}, + {name = "createdAt", type = "datetime", nullable = true}, + {name = "updatedAt", type = "datetime", nullable = true} + ], + indexes = [ + {name = "idx_wjobs_processing", columns = "status,runAt,priority", unique = false, optional = true}, + {name = "idx_wjobs_queue", columns = "queue,status", unique = false, optional = true}, + {name = "idx_wjobs_cleanup", columns = "status,completedAt", unique = false, optional = true}, + {name = "idx_wjobs_unique_key", columns = "uniqueKey", unique = true, optional = false} + ] + } + ]; + } + + /** + * One table's definition. + */ + public struct function tableDef(required string name) { + for (local.t in tables()) { + if (CompareNoCase(local.t.name, arguments.name) == 0) { + return local.t; + } + } + Throw(type = "Wheels.JobSchema.UnknownTable", message = "#arguments.name# is not a job table."); + } + + /** + * One column's definition. + */ + public struct function columnDef(required string tableName, required string columnName) { + for (local.c in tableDef(arguments.tableName).columns) { + if (CompareNoCase(local.c.name, arguments.columnName) == 0) { + return local.c; + } + } + Throw(type = "Wheels.JobSchema.UnknownColumn", message = "#arguments.tableName#.#arguments.columnName# is not a job column."); + } + + /** + * One index's definition. + */ + public struct function indexDef(required string tableName, required string indexName) { + for (local.i in tableDef(arguments.tableName).indexes) { + if (CompareNoCase(local.i.name, arguments.indexName) == 0) { + return local.i; + } + } + Throw(type = "Wheels.JobSchema.UnknownIndex", message = "#arguments.indexName# is not an index on #arguments.tableName#."); + } + + /** + * A column's SQL type on a database: VARCHAR2/CLOB/TIMESTAMP on Oracle, CLOB/TIMESTAMP on H2, + * TIMESTAMP on PostgreSQL, and VARCHAR/TEXT/DATETIME elsewhere. + */ + public string function columnType(required struct column, required string dbType) { + local.varcharType = arguments.dbType == "oracle" ? "VARCHAR2" : "VARCHAR"; + local.textType = ListFindNoCase("oracle,h2", arguments.dbType) ? "CLOB" : "TEXT"; + local.datetimeType = ListFindNoCase("oracle,postgresql,h2", arguments.dbType) ? "TIMESTAMP" : "DATETIME"; + switch (arguments.column.type) { + case "string": + return "#local.varcharType#(#arguments.column.limit#)"; + case "text": + return local.textType; + case "datetime": + return local.datetimeType; + default: + return "INT"; + } + } + + /** + * A column's definition inside CREATE TABLE. + */ + public string function columnDefinition(required struct column, required string dbType) { + local.sql = "#arguments.column.name# #columnType(arguments.column, arguments.dbType)#"; + if (StructKeyExists(arguments.column, "default")) { + local.sql &= " DEFAULT " & (arguments.column.type == "integer" ? arguments.column.default : "'#arguments.column.default#'"); + } + if (!arguments.column.nullable) { + local.sql &= " NOT NULL"; + } + if (StructKeyExists(arguments.column, "primaryKey") && arguments.column.primaryKey) { + local.sql &= " PRIMARY KEY"; + } + return local.sql; + } + + /** + * CREATE TABLE for a job table on a database. + */ + public string function createTableSql(required string tableName, required string dbType) { + local.definitions = []; + for (local.c in tableDef(arguments.tableName).columns) { + ArrayAppend(local.definitions, columnDefinition(local.c, arguments.dbType)); + } + return "CREATE TABLE #arguments.tableName# (" & ArrayToList(local.definitions, ", ") & ")"; + } + + /** + * The DDL that adds a (nullable) column to an existing job table. SQL Server has no COLUMN + * keyword and Oracle takes a parenthesised column list; everything else accepts ADD COLUMN. + */ + public string function addColumnSql(required string tableName, required string columnName, required string dbType) { + local.c = columnDef(arguments.tableName, arguments.columnName); + local.definition = "#local.c.name# #columnType(local.c, arguments.dbType)#"; + if (arguments.dbType == "oracle") { + return "ALTER TABLE #arguments.tableName# ADD (#local.definition#)"; + } + if (arguments.dbType == "sqlserver") { + return "ALTER TABLE #arguments.tableName# ADD #local.definition#"; + } + return "ALTER TABLE #arguments.tableName# ADD COLUMN #local.definition#"; + } + + /** + * CREATE INDEX for one of a job table's indexes on a database. + */ + public string function indexSql(required string tableName, required string indexName, required string dbType) { + local.i = indexDef(arguments.tableName, arguments.indexName); + local.sql = "CREATE " & (local.i.unique ? "UNIQUE " : "") & "INDEX #local.i.name# ON #arguments.tableName# (#Replace(local.i.columns, ",", ", ", "all")#)"; + if (local.i.unique && arguments.dbType == "sqlserver") { + local.sql &= " WHERE #local.i.columns# IS NOT NULL"; + } + return local.sql; + } + + /** + * Whether the job tables may be created and upgraded automatically: the jobsAutoCreateTables + * setting, true unless an app that installs the schema through a migration turns it off. + */ + public boolean function autoCreateEnabled() { + if (StructKeyExists(application, "wheels") && StructKeyExists(application.wheels, "jobsAutoCreateTables")) { + return IsBoolean(application.wheels.jobsAutoCreateTables) ? application.wheels.jobsAutoCreateTables : true; + } + return true; + } + + /** + * The message for a job table or column that is missing while auto-create is off. + */ + public string function missingSchemaMessage(required string what) { + return "#arguments.what# is missing and jobsAutoCreateTables is false, so it isn't created automatically. Generate the jobs migration with `wheels jobs install` and run it with `wheels migrate latest`."; + } + + /** + * The database type, as Job.$detectDatabaseType() reports it. + */ + public string function databaseType() { + if (!StructKeyExists(variables, "dbTypeCached")) { + local.job = new wheels.Job(); + variables.dbTypeCached = local.job.$detectDatabaseType(); + } + return variables.dbTypeCached; + } + + /** + * Whether a table exists, from the database's catalog. Never a failing probe query, so it is + * safe inside a migration's transaction (a failed statement aborts a PostgreSQL transaction). + */ + public boolean function hasTable(required string tableName) { + local.name = LCase(arguments.tableName); + switch (databaseType()) { + case "sqlite": + return $catalogHasRow("SELECT 1 FROM sqlite_master WHERE type = 'table' AND LOWER(name) = :name", {name = local.name}); + case "oracle": + return $catalogHasRow("SELECT 1 FROM user_tables WHERE LOWER(table_name) = :name", {name = local.name}); + case "sqlserver": + return $catalogHasRow("SELECT 1 FROM INFORMATION_SCHEMA.TABLES WHERE LOWER(TABLE_NAME) = :name AND TABLE_SCHEMA = SCHEMA_NAME()", {name = local.name}); + case "mysql": + return $catalogHasRow("SELECT 1 FROM information_schema.tables WHERE table_schema = DATABASE() AND LOWER(table_name) = :name", {name = local.name}); + case "postgresql": + return $catalogHasRow("SELECT 1 FROM information_schema.tables WHERE table_schema = current_schema() AND LOWER(table_name) = :name", {name = local.name}); + default: + return $catalogHasRow("SELECT 1 FROM INFORMATION_SCHEMA.TABLES WHERE LOWER(TABLE_NAME) = :name", {name = local.name}); + } + } + + /** + * Whether a table has a column, from the database's catalog (see hasTable()). + */ + public boolean function hasColumn(required string tableName, required string columnName) { + local.params = {name = LCase(arguments.tableName), col = LCase(arguments.columnName)}; + switch (databaseType()) { + case "sqlite": + return $catalogHasRow("SELECT 1 FROM pragma_table_info('#$safeName(arguments.tableName)#') WHERE LOWER(name) = :col", {col = local.params.col}); + case "oracle": + return $catalogHasRow("SELECT 1 FROM user_tab_columns WHERE LOWER(table_name) = :name AND LOWER(column_name) = :col", local.params); + case "sqlserver": + return $catalogHasRow("SELECT 1 FROM INFORMATION_SCHEMA.COLUMNS WHERE LOWER(TABLE_NAME) = :name AND LOWER(COLUMN_NAME) = :col AND TABLE_SCHEMA = SCHEMA_NAME()", local.params); + case "mysql": + return $catalogHasRow("SELECT 1 FROM information_schema.columns WHERE table_schema = DATABASE() AND LOWER(table_name) = :name AND LOWER(column_name) = :col", local.params); + case "postgresql": + return $catalogHasRow("SELECT 1 FROM information_schema.columns WHERE table_schema = current_schema() AND LOWER(table_name) = :name AND LOWER(column_name) = :col", local.params); + default: + return $catalogHasRow("SELECT 1 FROM INFORMATION_SCHEMA.COLUMNS WHERE LOWER(TABLE_NAME) = :name AND LOWER(COLUMN_NAME) = :col", local.params); + } + } + + /** + * Whether a table has a named index, from the database's catalog (see hasTable()). + */ + public boolean function hasIndex(required string tableName, required string indexName) { + local.params = {name = LCase(arguments.tableName), idx = LCase(arguments.indexName)}; + switch (databaseType()) { + case "sqlite": + return $catalogHasRow("SELECT 1 FROM sqlite_master WHERE type = 'index' AND LOWER(tbl_name) = :name AND LOWER(name) = :idx", local.params); + case "oracle": + return $catalogHasRow("SELECT 1 FROM user_indexes WHERE LOWER(table_name) = :name AND LOWER(index_name) = :idx", local.params); + case "sqlserver": + return $catalogHasRow("SELECT 1 FROM sys.indexes WHERE object_id = OBJECT_ID(:name) AND LOWER(name) = :idx", local.params); + case "mysql": + return $catalogHasRow("SELECT 1 FROM information_schema.statistics WHERE table_schema = DATABASE() AND LOWER(table_name) = :name AND LOWER(index_name) = :idx", local.params); + case "postgresql": + return $catalogHasRow("SELECT 1 FROM pg_indexes WHERE schemaname = current_schema() AND LOWER(tablename) = :name AND LOWER(indexname) = :idx", local.params); + default: + return $catalogHasRow("SELECT 1 FROM INFORMATION_SCHEMA.INDEXES WHERE LOWER(TABLE_NAME) = :name AND LOWER(INDEX_NAME) = :idx", local.params); + } + } + + /** + * A table's columns as the database's catalog reports them: lower-cased column name -> + * nullable (boolean). Used to check that the install migration and auto-create agree. + */ + public struct function catalogColumns(required string tableName) { + local.name = LCase(arguments.tableName); + local.params = {name = {value = local.name, cfsqltype = "cf_sql_varchar"}}; + switch (databaseType()) { + case "sqlite": + local.sql = "SELECT name AS colname, CASE WHEN ""notnull"" = 1 OR pk = 1 THEN 'NO' ELSE 'YES' END AS nullable FROM pragma_table_info('#$safeName(arguments.tableName)#')"; + local.params = {}; + break; + case "oracle": + local.sql = "SELECT column_name AS colname, CASE WHEN nullable = 'Y' THEN 'YES' ELSE 'NO' END AS nullable FROM user_tab_columns WHERE LOWER(table_name) = :name"; + break; + case "sqlserver": + local.sql = "SELECT COLUMN_NAME AS colname, IS_NULLABLE AS nullable FROM INFORMATION_SCHEMA.COLUMNS WHERE LOWER(TABLE_NAME) = :name AND TABLE_SCHEMA = SCHEMA_NAME()"; + break; + case "mysql": + local.sql = "SELECT column_name AS colname, is_nullable AS nullable FROM information_schema.columns WHERE table_schema = DATABASE() AND LOWER(table_name) = :name"; + break; + case "postgresql": + local.sql = "SELECT column_name AS colname, is_nullable AS nullable FROM information_schema.columns WHERE table_schema = current_schema() AND LOWER(table_name) = :name"; + break; + default: + local.sql = "SELECT COLUMN_NAME AS colname, IS_NULLABLE AS nullable FROM INFORMATION_SCHEMA.COLUMNS WHERE LOWER(TABLE_NAME) = :name"; + } + local.rows = QueryExecute(local.sql, local.params, {datasource = variables.datasource}); + local.rv = {}; + for (local.i = 1; local.i <= local.rows.recordCount; local.i++) { + local.rv[LCase(local.rows.colname[local.i])] = CompareNoCase(Trim(local.rows.nullable[local.i]), "YES") == 0; + } + return local.rv; + } + + /** + * The source of the migration `wheels jobs install` writes: migrator DSL that creates each job + * table, or tops up one that auto-create already built (adding only its missing columns and + * indexes), so running it on any existing install is safe and running it twice changes nothing. + */ + public string function migrationSource() { + local.nl = Chr(10); + local.lines = []; + ArrayAppend(local.lines, "/**"); + ArrayAppend(local.lines, " * The Wheels background-job tables (generated by `wheels jobs install`)."); + ArrayAppend(local.lines, " *"); + ArrayAppend(local.lines, " * Creates each job table, or adds what an existing one is missing: tables that Wheels created"); + ArrayAppend(local.lines, " * automatically are adopted, and running it again changes nothing. With this migration applied,"); + ArrayAppend(local.lines, " * set(jobsAutoCreateTables = false) so servers never change the job tables at startup."); + ArrayAppend(local.lines, " */"); + ArrayAppend(local.lines, "component extends=""wheels.migrator.Migration"" {"); + ArrayAppend(local.lines, ""); + ArrayAppend(local.lines, " function up() {"); + ArrayAppend(local.lines, " var schema = new wheels.JobSchema();"); + for (local.t in tables()) { + ArrayAppend(local.lines, ""); + ArrayAppend(local.lines, " if (!schema.hasTable(""#local.t.name#"")) {"); + ArrayAppend(local.lines, " var t = createTable(name = ""#local.t.name#"", id = false);"); + for (local.c in local.t.columns) { + ArrayAppend(local.lines, " " & $dslColumn(local.c)); + } + ArrayAppend(local.lines, " t.create();"); + ArrayAppend(local.lines, " } else {"); + for (local.c in local.t.columns) { + if (local.c.nullable) { + ArrayAppend(local.lines, " if (!schema.hasColumn(""#local.t.name#"", ""#local.c.name#"")) {"); + ArrayAppend(local.lines, " " & $dslAddColumn(local.t.name, local.c)); + if (local.c.name == "uniqueKey") { + ArrayAppend(local.lines, " execute(""UPDATE #local.t.name# SET uniqueKey = id WHERE uniqueKey IS NULL"");"); + } + ArrayAppend(local.lines, " }"); + } + } + ArrayAppend(local.lines, " }"); + for (local.i in local.t.indexes) { + ArrayAppend(local.lines, " if (!schema.hasIndex(""#local.t.name#"", ""#local.i.name#"")) {"); + if (local.i.unique) { + ArrayAppend(local.lines, " // SQL Server's unique indexes admit a single NULL, so the index is filtered there."); + ArrayAppend(local.lines, " if (schema.databaseType() == ""sqlserver"") {"); + ArrayAppend(local.lines, " execute(""CREATE UNIQUE INDEX #local.i.name# ON #local.t.name# (#local.i.columns#) WHERE #local.i.columns# IS NOT NULL"");"); + ArrayAppend(local.lines, " } else {"); + ArrayAppend(local.lines, " addIndex(table = ""#local.t.name#"", columnNames = ""#local.i.columns#"", unique = true, indexName = ""#local.i.name#"");"); + ArrayAppend(local.lines, " }"); + } else { + ArrayAppend(local.lines, " addIndex(table = ""#local.t.name#"", columnNames = ""#local.i.columns#"", indexName = ""#local.i.name#"");"); + } + ArrayAppend(local.lines, " }"); + } + } + ArrayAppend(local.lines, " }"); + ArrayAppend(local.lines, ""); + ArrayAppend(local.lines, " function down() {"); + ArrayAppend(local.lines, " var schema = new wheels.JobSchema();"); + local.reversed = tables(); + for (local.n = ArrayLen(local.reversed); local.n >= 1; local.n--) { + ArrayAppend(local.lines, " if (schema.hasTable(""#local.reversed[local.n].name#"")) {"); + ArrayAppend(local.lines, " dropTable(""#local.reversed[local.n].name#"");"); + ArrayAppend(local.lines, " }"); + } + ArrayAppend(local.lines, " }"); + ArrayAppend(local.lines, ""); + ArrayAppend(local.lines, "}"); + return ArrayToList(local.lines, local.nl) & local.nl; + } + + /** + * Internal: a column as a TableDefinition call inside createTable(). + */ + public string function $dslColumn(required struct column) { + if (StructKeyExists(arguments.column, "primaryKey") && arguments.column.primaryKey) { + return "t.primaryKey(name = ""#arguments.column.name#"", type = ""#arguments.column.type#"", limit = #arguments.column.limit#);"; + } + return "t.#$dslType(arguments.column)#(#$dslOptions(arguments.column)#);"; + } + + /** + * Internal: a missing nullable column as an addColumn() call. + */ + public string function $dslAddColumn(required string tableName, required struct column) { + local.call = "addColumn(table = ""#arguments.tableName#"", columnType = ""#$dslType(arguments.column)#"", columnName = ""#arguments.column.name#"""; + if (StructKeyExists(arguments.column, "limit")) { + local.call &= ", limit = #arguments.column.limit#"; + } + return local.call & ", allowNull = true);"; + } + + /** + * Internal: the TableDefinition helper for a column type. + */ + public string function $dslType(required struct column) { + return arguments.column.type; + } + + /** + * Internal: a column's named arguments for its TableDefinition helper. + */ + public string function $dslOptions(required struct column) { + local.options = ["columnNames = ""#arguments.column.name#"""]; + if (StructKeyExists(arguments.column, "limit")) { + ArrayAppend(local.options, "limit = #arguments.column.limit#"); + } + if (StructKeyExists(arguments.column, "default")) { + ArrayAppend(local.options, "default = " & (IsNumeric(arguments.column.default) ? arguments.column.default : """#arguments.column.default#""")); + } + ArrayAppend(local.options, "allowNull = " & (arguments.column.nullable ? "true" : "false")); + return ArrayToList(local.options, ", "); + } + + /** + * Internal: true when a catalog query returns a row. + */ + public boolean function $catalogHasRow(required string sql, required struct values) { + local.params = {}; + for (local.key in arguments.values) { + local.params[local.key] = {value = arguments.values[local.key], cfsqltype = "cf_sql_varchar"}; + } + return QueryExecute(arguments.sql, local.params, {datasource = variables.datasource}).recordCount > 0; + } + + /** + * Internal: a table name safe to inline (letters, digits and underscores only). + */ + public string function $safeName(required string name) { + return REReplace(arguments.name, "[^A-Za-z0-9_]", "", "all"); + } + +} diff --git a/vendor/wheels/public/CliBridge.cfc b/vendor/wheels/public/CliBridge.cfc index 05fea42358..69776ad943 100644 --- a/vendor/wheels/public/CliBridge.cfc +++ b/vendor/wheels/public/CliBridge.cfc @@ -74,7 +74,8 @@ component output="false" displayName="CLI Bridge" { "jobsRetry" = "jobsRetry", "jobsPurge" = "jobsPurge", "jobsMonitor" = "jobsMonitor", - "jobsEnqueue" = "jobsEnqueue" + "jobsEnqueue" = "jobsEnqueue", + "jobsInstallSource" = "jobsInstallSource" }; return this; } @@ -825,6 +826,24 @@ component output="false" displayName="CLI Bridge" { return local.rv; } + /** + * `wheels jobs install`: the source of the migration that creates the job tables, built from + * wheels.JobSchema. Read-only: the CLI writes the file, and nothing touches the database. + */ + public struct function jobsInstallSource(required struct context, required struct params) { + local.rv = {}; + try { + local.rv.success = true; + local.rv.migrationName = "CreateWheelsJobTables"; + local.rv.source = new wheels.JobSchema().migrationSource(); + local.rv.message = "Jobs migration source generated"; + } catch (any e) { + local.rv.success = false; + local.rv.message = "Error generating the jobs migration: " & e.message; + } + return local.rv; + } + public struct function jobsStatus(required struct context, required struct params) { local.rv = {}; try { diff --git a/vendor/wheels/tests/specs/cli/CliBridgeSpec.cfc b/vendor/wheels/tests/specs/cli/CliBridgeSpec.cfc index 52f7f2ef1d..2e5e186148 100644 --- a/vendor/wheels/tests/specs/cli/CliBridgeSpec.cfc +++ b/vendor/wheels/tests/specs/cli/CliBridgeSpec.cfc @@ -31,7 +31,7 @@ component extends="wheels.WheelsTest" { & "migrationLockStatus,migrationUnlock," & "dbStatus,dbVersion,dbRollback,dbSchema,introspect,dbSeed,routes,dbCreate,dbDrop," & "dbReset,dbSetup,dbDump,dbRestore,dbShell,jobsProcessNext,jobsStatus,jobsRetry," - & "jobsPurge,jobsMonitor,jobsEnqueue"; + & "jobsPurge,jobsMonitor,jobsEnqueue,jobsInstallSource"; for (var cmd in ListToArray(declared)) { expect(bridge.handles(cmd)).toBeTrue("CliBridge should handle '" & cmd & "'"); } diff --git a/vendor/wheels/tests/specs/jobs/JobSchemaSpec.cfc b/vendor/wheels/tests/specs/jobs/JobSchemaSpec.cfc new file mode 100644 index 0000000000..6f18504712 --- /dev/null +++ b/vendor/wheels/tests/specs/jobs/JobSchemaSpec.cfc @@ -0,0 +1,211 @@ +/** + * The job schema is defined once (wheels.JobSchema). Auto-create builds wheels_jobs from it, and + * `wheels jobs install` writes a migration from it that builds the same table, or tops up one + * auto-create already built. With jobsAutoCreateTables = false nothing is created or altered at + * runtime. These specs drop and rebuild wheels_jobs, so each ends by restoring the auto-created + * layout (a warm PostgreSQL connection rejects plans prepared against another layout). + */ +component extends="wheels.WheelsTest" { + + function afterAll() { + $removeGeneratedMigration(); + } + + function run() { + + describe("wheels.JobSchema", function() { + + afterEach(function() { + StructDelete(application.wheels, "jobsAutoCreateTables"); + $rebuildAutoCreated(); + }); + + it("is what auto-create builds: every column and index", function() { + $rebuildAutoCreated(); + var schema = new wheels.JobSchema(); + for (var column in schema.tableDef("wheels_jobs").columns) { + expect(schema.hasColumn("wheels_jobs", column.name)).toBeTrue(column.name); + } + for (var index in schema.tableDef("wheels_jobs").indexes) { + expect(schema.hasIndex("wheels_jobs", index.name)).toBeTrue(index.name); + } + expect(schema.hasTable("wheels_jobs")).toBeTrue(); + expect(schema.hasTable("wheels_no_such_table")).toBeFalse(); + expect(schema.hasColumn("wheels_jobs", "noSuchColumn")).toBeFalse(); + }); + + it("writes an install migration that builds the same table as auto-create", function() { + $rebuildAutoCreated(); + var schema = new wheels.JobSchema(); + var autoCreated = schema.catalogColumns("wheels_jobs"); + + $dropJobTable(); + var migration = $generatedMigration(); + migration.up(); + + expect(schema.catalogColumns("wheels_jobs")).toBe(autoCreated, "same columns, same nullability"); + for (var index in schema.tableDef("wheels_jobs").indexes) { + expect(schema.hasIndex("wheels_jobs", index.name)).toBeTrue(index.name); + } + // It works as a jobs table: a keyed enqueue de-duplicates. + $clearUniqueKeyMemos(); + var job = new wheels.tests._assets.jobs.ProcessOrdersJob(); + var first = job.enqueue(queue = "test_schema_install", uniqueKey = "schema:1"); + var second = job.enqueue(queue = "test_schema_install", uniqueKey = "schema:1"); + expect(second.duplicate).toBeTrue(); + expect(second.id).toBe(first.id); + }); + + it("adopts a table auto-create built earlier, adding only what it lacks, and runs again as a no-op", function() { + $createLegacyJobTable(); + var legacyId = CreateUUID(); + $insertLegacyRow(legacyId); + var schema = new wheels.JobSchema(); + expect(schema.hasColumn("wheels_jobs", "uniqueKey")).toBeFalse(); + + var migration = $generatedMigration(); + migration.up(); + for (var column in schema.tableDef("wheels_jobs").columns) { + expect(schema.hasColumn("wheels_jobs", column.name)).toBeTrue(column.name); + } + expect(schema.hasIndex("wheels_jobs", "idx_wjobs_unique_key")).toBeTrue(); + var row = queryExecute( + "SELECT uniqueKey FROM wheels_jobs WHERE id = :id", + {id = {value = legacyId, cfsqltype = "cf_sql_varchar"}}, + {datasource = application.wheels.dataSourceName} + ); + expect(row.uniqueKey).toBe(legacyId, "existing rows get their id as their key, as the automatic upgrade does"); + + migration.up(); + expect(schema.hasColumn("wheels_jobs", "uniqueKey")).toBeTrue(); + }); + + it("creates and alters nothing with jobsAutoCreateTables = false", function() { + $dropJobTable(); + application.wheels.jobsAutoCreateTables = false; + var job = new wheels.Job(); + expect(job.$ensureJobTable()).toBeFalse(); + expect(new wheels.JobSchema().hasTable("wheels_jobs")).toBeFalse(); + + StructDelete(application.wheels, "jobsAutoCreateTables"); + $createLegacyJobTable(); + application.wheels.jobsAutoCreateTables = false; + expect(new wheels.Job().$ensureJobTable()).toBeTrue(); + expect(new wheels.JobSchema().hasColumn("wheels_jobs", "uniqueKey")).toBeFalse("no column upgrade at runtime"); + }); + + it("builds each database's DDL", function() { + var schema = new wheels.JobSchema(); + expect(schema.indexSql(tableName = "wheels_jobs", indexName = "idx_wjobs_unique_key", dbType = "sqlserver")).toBe( + "CREATE UNIQUE INDEX idx_wjobs_unique_key ON wheels_jobs (uniqueKey) WHERE uniqueKey IS NOT NULL" + ); + expect(schema.indexSql(tableName = "wheels_jobs", indexName = "idx_wjobs_unique_key", dbType = "postgresql")).toBe( + "CREATE UNIQUE INDEX idx_wjobs_unique_key ON wheels_jobs (uniqueKey)" + ); + expect(schema.addColumnSql(tableName = "wheels_jobs", columnName = "claimedBy", dbType = "oracle")).toBe("ALTER TABLE wheels_jobs ADD (claimedBy VARCHAR2(128))"); + expect(schema.addColumnSql(tableName = "wheels_jobs", columnName = "claimTimeout", dbType = "sqlserver")).toBe("ALTER TABLE wheels_jobs ADD claimTimeout INT"); + expect(schema.addColumnSql(tableName = "wheels_jobs", columnName = "uniqueKey", dbType = "mysql")).toBe("ALTER TABLE wheels_jobs ADD COLUMN uniqueKey VARCHAR(255)"); + var create = schema.createTableSql(tableName = "wheels_jobs", dbType = "h2"); + expect(create).toInclude("id VARCHAR(36) NOT NULL PRIMARY KEY"); + expect(create).toInclude("queue VARCHAR(100) DEFAULT 'default' NOT NULL"); + expect(create).toInclude("data CLOB"); + expect(create).toInclude("runAt TIMESTAMP"); + }); + + }); + } + + /** + * Writes the install migration's source to a component under tests/_assets and returns an + * instance of it, so the spec runs exactly what `wheels jobs install` would write. + */ + private any function $generatedMigration() { + var dir = ExpandPath("/wheels/tests/_assets/jobs_install"); + if (!DirectoryExists(dir)) { + DirectoryCreate(dir); + } + FileWrite(dir & "/CreateWheelsJobTablesProbe.cfc", new wheels.JobSchema().migrationSource()); + return CreateObject("component", "wheels.tests._assets.jobs_install.CreateWheelsJobTablesProbe").init(); + } + + private void function $removeGeneratedMigration() { + var dir = ExpandPath("/wheels/tests/_assets/jobs_install"); + if (DirectoryExists(dir)) { + DirectoryDelete(dir, true); + } + } + + private void function $dropJobTable() { + try { + queryExecute("DROP TABLE wheels_jobs", {}, {datasource = application.wheels.dataSourceName}); + } catch (any e) { + } + $clearUniqueKeyMemos(); + } + + private void function $rebuildAutoCreated() { + $dropJobTable(); + new wheels.Job().$ensureJobTable(); + $clearUniqueKeyMemos(); + } + + private void function $clearUniqueKeyMemos() { + if (StructKeyExists(application.wheels, "$jobsUniqueKeyIndexVerified")) { + StructDelete(application.wheels, "$jobsUniqueKeyIndexVerified"); + } + StructDelete(application.wheels, "$uniqueKeyAlterFailedAt"); + } + + /** + * wheels_jobs as an older version auto-created it: no claim-fencing or uniqueKey columns, and + * none of the indexes. + */ + private void function $createLegacyJobTable() { + var dbType = new wheels.Job().$detectDatabaseType(); + var types = {varchar = "VARCHAR", text = "TEXT", stamp = "DATETIME"}; + if (dbType == "oracle") { + types = {varchar = "VARCHAR2", text = "CLOB", stamp = "TIMESTAMP"}; + } else if (dbType == "postgresql") { + types = {varchar = "VARCHAR", text = "TEXT", stamp = "TIMESTAMP"}; + } else if (dbType == "h2") { + types = {varchar = "VARCHAR", text = "CLOB", stamp = "TIMESTAMP"}; + } + $dropJobTable(); + queryExecute( + "CREATE TABLE wheels_jobs ( + id #types.varchar#(36) NOT NULL PRIMARY KEY, + jobClass #types.varchar#(255) NOT NULL, + queue #types.varchar#(100) DEFAULT 'default' NOT NULL, + data #types.text#, + priority INT DEFAULT 0 NOT NULL, + status #types.varchar#(20) DEFAULT 'pending' NOT NULL, + attempts INT DEFAULT 0 NOT NULL, + maxRetries INT DEFAULT 3 NOT NULL, + claimTimeout INT, + lastError #types.text#, + runAt #types.stamp#, + completedAt #types.stamp#, + failedAt #types.stamp#, + createdAt #types.stamp#, + updatedAt #types.stamp# + )", + {}, + {datasource = application.wheels.dataSourceName} + ); + } + + private void function $insertLegacyRow(required string id) { + queryExecute( + "INSERT INTO wheels_jobs (id, jobClass, queue, data, priority, status, attempts, maxRetries, runAt, createdAt, updatedAt) + VALUES (:id, 'wheels.tests._assets.jobs.ProcessOrdersJob', 'test_schema_legacy', '{}', 0, 'completed', 1, 3, :stamp, :stamp2, :stamp3)", + { + id = {value = arguments.id, cfsqltype = "cf_sql_varchar"}, + stamp = {value = Now(), cfsqltype = "cf_sql_timestamp"}, + stamp2 = {value = Now(), cfsqltype = "cf_sql_timestamp"}, + stamp3 = {value = Now(), cfsqltype = "cf_sql_timestamp"} + }, + {datasource = application.wheels.dataSourceName} + ); + } + +} From 10ddf9b65730ef3c80cefb14e547b0c98b0440ca Mon Sep 17 00:00:00 2001 From: Peter Amiri Date: Wed, 7 Oct 2026 22:49:47 -0700 Subject: [PATCH 2/5] fix(job): SchemaMissing error, migrator-datasource schema checks, round-trip spec - With jobsAutoCreateTables = false and no table, enqueue and a worker's poll throw Wheels.Job.SchemaMissing with the install guidance (logged once), instead of a generic enqueue failure or a poll that looks like an empty queue. - The generated migration's catalog checks use $migratorDataSource() and its credentials, so a TenantMigrator checks each tenant's database. - jobsAutoCreateTables defaults to true with the other settings. - down() dropping adopted tables is documented in the generated file. - New spec runs the migration up/down/up through the Migrator and compares column type families and nullability with auto-create. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Peter Amiri --- changelog.d/4484-jobs-install.added.md | 2 +- vendor/wheels/Job.cfc | 27 +++- vendor/wheels/JobSchema.cfc | 116 ++++++++++++++++-- vendor/wheels/JobWorker.cfc | 5 + vendor/wheels/events/onapplicationstart.cfc | 3 + .../wheels/tests/specs/jobs/JobSchemaSpec.cfc | 97 ++++++++++++++- 6 files changed, 233 insertions(+), 17 deletions(-) diff --git a/changelog.d/4484-jobs-install.added.md b/changelog.d/4484-jobs-install.added.md index 8d225a3baf..08121ca23d 100644 --- a/changelog.d/4484-jobs-install.added.md +++ b/changelog.d/4484-jobs-install.added.md @@ -1 +1 @@ -- `wheels jobs install` writes a migration that creates the background-job tables, so an app can apply the jobs schema once, reviewed and under migration control, instead of having every server create and alter it at startup. The migration adopts tables Wheels already created automatically, adding only what they lack, and running it again changes nothing. Apps that use it set `set(jobsAutoCreateTables = false)`, which stops the framework from creating or altering the job tables at runtime; a missing table or column is then reported with a pointer to `wheels jobs install`. Auto-create stays the default. The job schema is now defined once (`wheels.JobSchema`), so the migration and auto-create build the same tables. ([#4484](https://github.com/wheels-dev/wheels/issues/4484)) +- `wheels jobs install` writes a migration that creates the background-job tables, so an app can apply the jobs schema once, reviewed and under migration control, instead of having every server create and alter it at startup. The migration adopts tables Wheels already created automatically, adding only what they lack, and running it again changes nothing. Apps that use it set `set(jobsAutoCreateTables = false)`, which stops the framework from creating or altering the job tables at runtime; a missing table then raises `Wheels.Job.SchemaMissing` from enqueue and from a worker's poll, with a pointer to `wheels jobs install`. Auto-create stays the default (`jobsAutoCreateTables = true`). The migration's `down()` drops the job tables, including ones it adopted, and every job in them. The job schema is now defined once (`wheels.JobSchema`), so the migration and auto-create build the same tables. ([#4484](https://github.com/wheels-dev/wheels/issues/4484)) diff --git a/vendor/wheels/Job.cfc b/vendor/wheels/Job.cfc index 1848cd5662..003b14b7fc 100644 --- a/vendor/wheels/Job.cfc +++ b/vendor/wheels/Job.cfc @@ -1079,8 +1079,7 @@ component { } if (!$jobSchema().autoCreateEnabled()) { - writeLog(text = $jobSchema().missingSchemaMessage("The wheels_jobs table"), type = "error", file = "wheels_jobs"); - return false; + $throwJobSchemaMissing("The wheels_jobs table"); } try { @@ -1334,6 +1333,20 @@ component { return variables.$claimTokenColumnsPresent; } + /** + * Throws Wheels.Job.SchemaMissing: a job table or column is missing while jobsAutoCreateTables + * is false, so the framework won't create it. Logged once per application as well, because a + * worker hits this on every poll until the migration runs. + */ + public void function $throwJobSchemaMissing(required string what) { + local.message = $jobSchema().missingSchemaMessage(arguments.what); + if (StructKeyExists(application, "wheels") && !StructKeyExists(application.wheels, "$jobsSchemaMissingLogged")) { + application.wheels.$jobsSchemaMissingLogged = true; + writeLog(text = local.message, type = "error", file = "wheels_jobs"); + } + Throw(type = "Wheels.Job.SchemaMissing", message = local.message); + } + /** * The job schema (wheels.JobSchema) for this instance's datasource, memoised per instance. */ @@ -1823,8 +1836,16 @@ component { * Public with $ prefix so JobWorker can pick database-appropriate SQL syntax. */ public string function $detectDatabaseType() { + return $databaseTypeOf(variables.$datasource); + } + + /** + * The database type behind a datasource, as $detectDatabaseType() reports it. Separate so + * wheels.JobSchema can ask about the migrator's datasource, which may not be the app's. + */ + public string function $databaseTypeOf(required string datasourceName) { try { - cfdbinfo(type = "version", datasource = "#variables.$datasource#", name = "local.info"); + cfdbinfo(type = "version", datasource = "#arguments.datasourceName#", name = "local.info"); local.product = local.info.database_productname; if (FindNoCase("oracle", local.product)) return "oracle"; if (FindNoCase("postgre", local.product)) return "postgresql"; diff --git a/vendor/wheels/JobSchema.cfc b/vendor/wheels/JobSchema.cfc index 5f5b47d224..1ed97afa7f 100644 --- a/vendor/wheels/JobSchema.cfc +++ b/vendor/wheels/JobSchema.cfc @@ -13,10 +13,14 @@ component { /** - * @datasource The datasource the catalog checks run against. Defaults to the app's. + * @datasource The datasource the catalog checks run against. Defaults to the app's. A + * migration passes the migrator's ($migratorDataSource()), which a TenantMigrator points at + * each tenant's database. + * @credentials Optional {username, password} for that datasource, as the migrator uses them. */ - public any function init(string datasource = "") { + public any function init(string datasource = "", struct credentials = {}) { variables.datasource = arguments.datasource; + variables.credentials = arguments.credentials; if (!Len(variables.datasource) && StructKeyExists(application, "wheels") && StructKeyExists(application.wheels, "dataSourceName")) { variables.datasource = application.wheels.dataSourceName; } @@ -98,7 +102,9 @@ component { /** * A column's SQL type on a database: VARCHAR2/CLOB/TIMESTAMP on Oracle, CLOB/TIMESTAMP on H2, - * TIMESTAMP on PostgreSQL, and VARCHAR/TEXT/DATETIME elsewhere. + * TIMESTAMP on PostgreSQL, and VARCHAR/TEXT/DATETIME elsewhere. Integers are INT everywhere, + * which Oracle stores as NUMBER(38); the claimTimeout column added to an older table on Oracle + * used to be NUMBER(10), and both hold any timeout. */ public string function columnType(required struct column, required string dbType) { local.varcharType = arguments.dbType == "oracle" ? "VARCHAR2" : "VARCHAR"; @@ -196,7 +202,7 @@ component { public string function databaseType() { if (!StructKeyExists(variables, "dbTypeCached")) { local.job = new wheels.Job(); - variables.dbTypeCached = local.job.$detectDatabaseType(); + variables.dbTypeCached = local.job.$databaseTypeOf(variables.datasource); } return variables.dbTypeCached; } @@ -292,7 +298,7 @@ component { default: local.sql = "SELECT COLUMN_NAME AS colname, IS_NULLABLE AS nullable FROM INFORMATION_SCHEMA.COLUMNS WHERE LOWER(TABLE_NAME) = :name"; } - local.rows = QueryExecute(local.sql, local.params, {datasource = variables.datasource}); + local.rows = QueryExecute(local.sql, local.params, $queryOptions()); local.rv = {}; for (local.i = 1; local.i <= local.rows.recordCount; local.i++) { local.rv[LCase(local.rows.colname[local.i])] = CompareNoCase(Trim(local.rows.nullable[local.i]), "YES") == 0; @@ -314,11 +320,14 @@ component { ArrayAppend(local.lines, " * Creates each job table, or adds what an existing one is missing: tables that Wheels created"); ArrayAppend(local.lines, " * automatically are adopted, and running it again changes nothing. With this migration applied,"); ArrayAppend(local.lines, " * set(jobsAutoCreateTables = false) so servers never change the job tables at startup."); + ArrayAppend(local.lines, " *"); + ArrayAppend(local.lines, " * down() drops the job tables, including ones this migration adopted rather than created, and"); + ArrayAppend(local.lines, " * every job in them."); ArrayAppend(local.lines, " */"); ArrayAppend(local.lines, "component extends=""wheels.migrator.Migration"" {"); ArrayAppend(local.lines, ""); ArrayAppend(local.lines, " function up() {"); - ArrayAppend(local.lines, " var schema = new wheels.JobSchema();"); + ArrayAppend(local.lines, " var schema = new wheels.JobSchema(datasource = $migratorDataSource(), credentials = $migratorDataSourceCredentials());"); for (local.t in tables()) { ArrayAppend(local.lines, ""); ArrayAppend(local.lines, " if (!schema.hasTable(""#local.t.name#"")) {"); @@ -357,7 +366,7 @@ component { ArrayAppend(local.lines, " }"); ArrayAppend(local.lines, ""); ArrayAppend(local.lines, " function down() {"); - ArrayAppend(local.lines, " var schema = new wheels.JobSchema();"); + ArrayAppend(local.lines, " var schema = new wheels.JobSchema(datasource = $migratorDataSource(), credentials = $migratorDataSourceCredentials());"); local.reversed = tables(); for (local.n = ArrayLen(local.reversed); local.n >= 1; local.n--) { ArrayAppend(local.lines, " if (schema.hasTable(""#local.reversed[local.n].name#"")) {"); @@ -413,6 +422,97 @@ component { return ArrayToList(local.options, ", "); } + /** + * A table's columns by type family as the catalog reports them: lower-cased column name -> + * string, text, integer or datetime. Families, not exact types: a migration's column helpers + * and auto-create may pick different members of a family (NUMBER(10) and NUMBER(38), TEXT and + * NVARCHAR(MAX)). SQLite stores both string and text as TEXT, so it reports them as text. + */ + public struct function catalogColumnFamilies(required string tableName) { + local.name = LCase(arguments.tableName); + local.params = {name = {value = local.name, cfsqltype = "cf_sql_varchar"}}; + local.dbType = databaseType(); + switch (local.dbType) { + case "sqlite": + local.sql = "SELECT name AS colname, type AS coltype, 0 AS collength FROM pragma_table_info('#$safeName(arguments.tableName)#')"; + local.params = {}; + break; + case "oracle": + local.sql = "SELECT column_name AS colname, data_type AS coltype, char_length AS collength FROM user_tab_columns WHERE LOWER(table_name) = :name"; + break; + case "sqlserver": + local.sql = "SELECT COLUMN_NAME AS colname, DATA_TYPE AS coltype, CHARACTER_MAXIMUM_LENGTH AS collength FROM INFORMATION_SCHEMA.COLUMNS WHERE LOWER(TABLE_NAME) = :name AND TABLE_SCHEMA = SCHEMA_NAME()"; + break; + case "mysql": + local.sql = "SELECT column_name AS colname, data_type AS coltype, character_maximum_length AS collength FROM information_schema.columns WHERE table_schema = DATABASE() AND LOWER(table_name) = :name"; + break; + case "postgresql": + local.sql = "SELECT column_name AS colname, data_type AS coltype, character_maximum_length AS collength FROM information_schema.columns WHERE table_schema = current_schema() AND LOWER(table_name) = :name"; + break; + default: + // H2 2.x names the type in DATA_TYPE; H2 1.4 has a JDBC type code there and the + // name in TYPE_NAME. + local.sql = "SELECT COLUMN_NAME AS colname, DATA_TYPE AS coltype, CHARACTER_MAXIMUM_LENGTH AS collength FROM INFORMATION_SCHEMA.COLUMNS WHERE LOWER(TABLE_NAME) = :name"; + try { + local.rows = QueryExecute( + "SELECT COLUMN_NAME AS colname, TYPE_NAME AS coltype, CHARACTER_MAXIMUM_LENGTH AS collength FROM INFORMATION_SCHEMA.COLUMNS WHERE LOWER(TABLE_NAME) = :name", + local.params, + $queryOptions() + ); + } catch (any e) { + // H2 2.x: no TYPE_NAME column. + } + } + if (!StructKeyExists(local, "rows")) { + local.rows = QueryExecute(local.sql, local.params, $queryOptions()); + } + local.rv = {}; + for (local.i = 1; local.i <= local.rows.recordCount; local.i++) { + local.length = IsNumeric(local.rows.collength[local.i]) ? Val(local.rows.collength[local.i]) : 0; + local.family = $typeFamily(type = local.rows.coltype[local.i], length = local.length); + if (local.dbType == "sqlite" && local.family == "string") { + local.family = "text"; + } + local.rv[LCase(local.rows.colname[local.i])] = local.family; + } + return local.rv; + } + + /** + * Internal: a catalog type name's family (string, text, integer, datetime), or the lower-cased + * name itself when it belongs to none. A character type with no length limit (-1) is text. + */ + public string function $typeFamily(required string type, numeric length = 0) { + local.t = LCase(Trim(arguments.type)); + if (FindNoCase("lob", local.t) || FindNoCase("large object", local.t) || FindNoCase("text", local.t)) { + return "text"; + } + if (FindNoCase("char", local.t)) { + return arguments.length == -1 ? "text" : "string"; + } + if (FindNoCase("time", local.t) || FindNoCase("date", local.t)) { + return "datetime"; + } + if (FindNoCase("int", local.t) || FindNoCase("number", local.t) || FindNoCase("numeric", local.t) || FindNoCase("decimal", local.t)) { + return "integer"; + } + return local.t; + } + + /** + * Internal: query options for this schema's datasource, with its credentials when given. + */ + public struct function $queryOptions() { + local.options = {datasource = variables.datasource}; + if (StructKeyExists(variables.credentials, "username") && Len(variables.credentials.username)) { + local.options.username = variables.credentials.username; + } + if (StructKeyExists(variables.credentials, "password") && Len(variables.credentials.password)) { + local.options.password = variables.credentials.password; + } + return local.options; + } + /** * Internal: true when a catalog query returns a row. */ @@ -421,7 +521,7 @@ component { for (local.key in arguments.values) { local.params[local.key] = {value = arguments.values[local.key], cfsqltype = "cf_sql_varchar"}; } - return QueryExecute(arguments.sql, local.params, {datasource = variables.datasource}).recordCount > 0; + return QueryExecute(arguments.sql, local.params, $queryOptions()).recordCount > 0; } /** diff --git a/vendor/wheels/JobWorker.cfc b/vendor/wheels/JobWorker.cfc index 04743adc1c..e70c9cbe46 100644 --- a/vendor/wheels/JobWorker.cfc +++ b/vendor/wheels/JobWorker.cfc @@ -941,6 +941,11 @@ component { } return false; } catch (any e) { + // The table is missing and may not be created (jobsAutoCreateTables = false): say so, + // with the way to create it, instead of skipping every poll as if the queue were empty. + if (e.type == "Wheels.Job.SchemaMissing") { + rethrow; + } return false; } } diff --git a/vendor/wheels/events/onapplicationstart.cfc b/vendor/wheels/events/onapplicationstart.cfc index 9b89af7ce4..48433d8cbb 100644 --- a/vendor/wheels/events/onapplicationstart.cfc +++ b/vendor/wheels/events/onapplicationstart.cfc @@ -348,6 +348,9 @@ component { // one at a time. A waiting instance gives up after migrationLockTimeout seconds; a holder // that died frees the lock once its lease (renewed before each step) runs out. application.$wheels.migratorLockTableName = "wheels_migrator_locks"; + // Background-job tables: true creates and upgrades them automatically at runtime; false + // leaves them to the app's own migration (`wheels jobs install`), for multi-server apps. + application.$wheels.jobsAutoCreateTables = true; application.$wheels.migrationLockTimeout = 300; application.$wheels.migrationLockLease = 3600; application.$wheels.createMigratorTable = true; diff --git a/vendor/wheels/tests/specs/jobs/JobSchemaSpec.cfc b/vendor/wheels/tests/specs/jobs/JobSchemaSpec.cfc index 6f18504712..d2550a804b 100644 --- a/vendor/wheels/tests/specs/jobs/JobSchemaSpec.cfc +++ b/vendor/wheels/tests/specs/jobs/JobSchemaSpec.cfc @@ -9,6 +9,14 @@ component extends="wheels.WheelsTest" { function afterAll() { $removeGeneratedMigration(); + try { + queryExecute( + "DELETE FROM #application.wheels.migratorTableName# WHERE version = :version", + {version = {value = $installVersion(), cfsqltype = "cf_sql_varchar"}}, + {datasource = application.wheels.dataSourceName} + ); + } catch (any e) { + } } function run() { @@ -16,7 +24,8 @@ component extends="wheels.WheelsTest" { describe("wheels.JobSchema", function() { afterEach(function() { - StructDelete(application.wheels, "jobsAutoCreateTables"); + application.wheels.jobsAutoCreateTables = true; + StructDelete(application.wheels, "$jobsSchemaMissingLogged"); $rebuildAutoCreated(); }); @@ -80,14 +89,67 @@ component extends="wheels.WheelsTest" { expect(schema.hasColumn("wheels_jobs", "uniqueKey")).toBeTrue(); }); - it("creates and alters nothing with jobsAutoCreateTables = false", function() { + it("runs up, down and up again through the migrator, building auto-create's column types", function() { + $rebuildAutoCreated(); + var schema = new wheels.JobSchema(); + var autoFamilies = schema.catalogColumnFamilies("wheels_jobs"); + var autoColumns = schema.catalogColumns("wheels_jobs"); + expect(autoFamilies.id).toBe(schema.databaseType() == "sqlite" ? "text" : "string"); + expect(autoFamilies.priority).toBe("integer"); + expect(autoFamilies.runat).toBe("datetime"); + + $dropJobTable(); + var migrator = $installMigrator(); + var output = migrator.migrateTo($installVersion()); + expect(schema.hasTable("wheels_jobs")).toBeTrue(output); + expect(schema.catalogColumnFamilies("wheels_jobs")).toBe(autoFamilies, "same type family for every column"); + expect(schema.catalogColumns("wheels_jobs")).toBe(autoColumns, "same nullability"); + + output = migrator.migrateTo("0"); + expect(schema.hasTable("wheels_jobs")).toBeFalse("down() drops the table: " & output); + + output = migrator.migrateTo($installVersion()); + expect(schema.hasTable("wheels_jobs")).toBeTrue(output); + expect(schema.catalogColumnFamilies("wheels_jobs")).toBe(autoFamilies); + for (var index in schema.tableDef("wheels_jobs").indexes) { + expect(schema.hasIndex("wheels_jobs", index.name)).toBeTrue(index.name); + } + migrator.migrateTo("0"); + }); + + it("with jobsAutoCreateTables = false, creates nothing and says how to install the schema", function() { $dropJobTable(); application.wheels.jobsAutoCreateTables = false; - var job = new wheels.Job(); - expect(job.$ensureJobTable()).toBeFalse(); + var ensured = {type = "", message = ""}; + try { + new wheels.Job().$ensureJobTable(); + } catch (any e) { + ensured.type = e.type; + ensured.message = e.message; + } + expect(ensured.type).toBe("Wheels.Job.SchemaMissing"); + expect(ensured.message).toInclude("wheels jobs install"); + expect(new wheels.JobSchema().hasTable("wheels_jobs")).toBeFalse(); + + var enqueued = {type = ""}; + try { + new wheels.tests._assets.jobs.ProcessOrdersJob().enqueue(queue = "test_schema_off"); + } catch (any e) { + enqueued.type = e.type; + } + expect(enqueued.type).toBe("Wheels.Job.SchemaMissing", "enqueue reports the missing schema, not a generic failure"); + + var polled = {type = ""}; + try { + new wheels.JobWorker().processNext(queues = "test_schema_off"); + } catch (any e) { + polled.type = e.type; + } + expect(polled.type).toBe("Wheels.Job.SchemaMissing", "a worker poll doesn't pass for an empty queue"); expect(new wheels.JobSchema().hasTable("wheels_jobs")).toBeFalse(); + }); - StructDelete(application.wheels, "jobsAutoCreateTables"); + it("with jobsAutoCreateTables = false, doesn't alter an existing table", function() { $createLegacyJobTable(); application.wheels.jobsAutoCreateTables = false; expect(new wheels.Job().$ensureJobTable()).toBeTrue(); @@ -128,6 +190,31 @@ component extends="wheels.WheelsTest" { return CreateObject("component", "wheels.tests._assets.jobs_install.CreateWheelsJobTablesProbe").init(); } + /** + * A Migrator over a directory holding just the install migration, under a version later than + * any other spec's, so migrateTo() runs it through the migrator's own transaction and lock. + */ + private any function $installMigrator() { + // One level at a time: Adobe's DirectoryCreate() takes only the path. + var parent = ExpandPath("/wheels/tests/_assets/jobs_install"); + if (!DirectoryExists(parent)) { + DirectoryCreate(parent); + } + var dir = parent & "/migrations"; + if (!DirectoryExists(dir)) { + DirectoryCreate(dir); + } + FileWrite(dir & "/#$installVersion()#_CreateWheelsJobTables.cfc", new wheels.JobSchema().migrationSource()); + return CreateObject("component", "wheels.Migrator").init( + migratePath = "/wheels/tests/_assets/jobs_install/migrations/", + sqlPath = "/wheels/tests/_assets/jobs_install/sql/" + ); + } + + private string function $installVersion() { + return "20991231000200"; + } + private void function $removeGeneratedMigration() { var dir = ExpandPath("/wheels/tests/_assets/jobs_install"); if (DirectoryExists(dir)) { From a74786d2d929ed9d8edd4b2e2c9eb6e0a73d7e8a Mon Sep 17 00:00:00 2001 From: Peter Amiri Date: Wed, 7 Oct 2026 23:03:37 -0700 Subject: [PATCH 3/5] fix(job): report a failing jobs-table probe as itself when the table exists With jobsAutoCreateTables = false, any failure of the existence probe (a lost connection, a permission) read as a missing table. The catalog now confirms the table is missing before SchemaMissing is thrown; otherwise the probe runs again so its own error reaches the caller. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Peter Amiri --- vendor/wheels/Job.cfc | 9 +++++++- .../wheels/tests/specs/jobs/JobSchemaSpec.cfc | 23 +++++++++++++++++++ 2 files changed, 31 insertions(+), 1 deletion(-) diff --git a/vendor/wheels/Job.cfc b/vendor/wheels/Job.cfc index 003b14b7fc..11587badf7 100644 --- a/vendor/wheels/Job.cfc +++ b/vendor/wheels/Job.cfc @@ -1079,7 +1079,14 @@ component { } if (!$jobSchema().autoCreateEnabled()) { - $throwJobSchemaMissing("The wheels_jobs table"); + // The probe can fail for reasons other than a missing table (a lost connection, a + // permission). Only the catalog's answer makes it a missing schema; otherwise run the + // probe again here, so its own error is what the caller sees. + if (!$jobSchema().hasTable("wheels_jobs")) { + $throwJobSchemaMissing("The wheels_jobs table"); + } + queryExecute("SELECT COUNT(*) AS cnt FROM wheels_jobs WHERE 1=0", {}, {datasource = variables.$datasource}); + return true; } try { diff --git a/vendor/wheels/tests/specs/jobs/JobSchemaSpec.cfc b/vendor/wheels/tests/specs/jobs/JobSchemaSpec.cfc index d2550a804b..89743f44af 100644 --- a/vendor/wheels/tests/specs/jobs/JobSchemaSpec.cfc +++ b/vendor/wheels/tests/specs/jobs/JobSchemaSpec.cfc @@ -149,6 +149,29 @@ component extends="wheels.WheelsTest" { expect(new wheels.JobSchema().hasTable("wheels_jobs")).toBeFalse(); }); + it("with jobsAutoCreateTables = false, reports a probe failure as itself when the table exists", function() { + // The catalog says the table is there but the probe fails (a lost connection, a + // permission): that error surfaces, not a misleading "missing table". + $dropJobTable(); + application.wheels.jobsAutoCreateTables = false; + var schema = new wheels.JobSchema(); + prepareMock(schema); + schema.$("hasTable", true); + var job = new wheels.Job(); + prepareMock(job); + job.$("$jobSchema", schema); + var probed = {type = "", message = ""}; + try { + job.$ensureJobTable(); + } catch (any e) { + probed.type = e.type; + probed.message = e.message; + } + expect(Len(probed.message)).toBeGT(0, "the probe's own error is rethrown"); + expect(probed.type).notToBe("Wheels.Job.SchemaMissing"); + expect(probed.message).notToInclude("jobsAutoCreateTables"); + }); + it("with jobsAutoCreateTables = false, doesn't alter an existing table", function() { $createLegacyJobTable(); application.wheels.jobsAutoCreateTables = false; From 17b100dba1ab121300df1114edc60d1aa750efdd Mon Sep 17 00:00:00 2001 From: Peter Amiri Date: Thu, 8 Oct 2026 11:28:24 -0700 Subject: [PATCH 4/5] docs(changelog): name every job table the install migration covers Co-Authored-By: Claude Opus 5.5 Signed-off-by: Peter Amiri --- changelog.d/4484-jobs-install.added.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/changelog.d/4484-jobs-install.added.md b/changelog.d/4484-jobs-install.added.md index 08121ca23d..b244243143 100644 --- a/changelog.d/4484-jobs-install.added.md +++ b/changelog.d/4484-jobs-install.added.md @@ -1 +1 @@ -- `wheels jobs install` writes a migration that creates the background-job tables, so an app can apply the jobs schema once, reviewed and under migration control, instead of having every server create and alter it at startup. The migration adopts tables Wheels already created automatically, adding only what they lack, and running it again changes nothing. Apps that use it set `set(jobsAutoCreateTables = false)`, which stops the framework from creating or altering the job tables at runtime; a missing table then raises `Wheels.Job.SchemaMissing` from enqueue and from a worker's poll, with a pointer to `wheels jobs install`. Auto-create stays the default (`jobsAutoCreateTables = true`). The migration's `down()` drops the job tables, including ones it adopted, and every job in them. The job schema is now defined once (`wheels.JobSchema`), so the migration and auto-create build the same tables. ([#4484](https://github.com/wheels-dev/wheels/issues/4484)) +- `wheels jobs install` writes a migration that creates the background-job tables (`wheels_jobs`, `wheels_job_hosts`, `wheels_job_schedules`, `wheels_job_locks`), so an app can apply the jobs schema once, reviewed and under migration control, instead of having every server create and alter it at startup. The migration adopts tables Wheels already created automatically, adding only what they lack, and running it again changes nothing. Apps that use it set `set(jobsAutoCreateTables = false)`, which stops the framework from creating or altering the job tables at runtime; a missing table then raises `Wheels.Job.SchemaMissing` from enqueue and from a worker's poll, with a pointer to `wheels jobs install`. Auto-create stays the default (`jobsAutoCreateTables = true`). The migration's `down()` drops the job tables, including ones it adopted, and every job in them. The job schema is now defined once (`wheels.JobSchema`), so the migration and auto-create build the same tables. ([#4484](https://github.com/wheels-dev/wheels/issues/4484)) From 4a605532eaafea1d2824d25f31fe9be0ac2ce9e4 Mon Sep 17 00:00:00 2001 From: Peter Amiri Date: Thu, 8 Oct 2026 12:11:29 -0700 Subject: [PATCH 5/5] test(job): make the jobs-table probe spec independent of DROP visibility The probe-failure spec dropped wheels_jobs and expected the next probe to fail. In the full BoxLang + SQLite suite the DROP was visible to the catalog check yet not to the probe's pooled connection (consistent with a connection left inside an uncommitted transaction by an earlier spec), so the probe passed. The spec now fails the probe with a datasource that doesn't exist, the lost-connection case, with no DROP. The schema specs' drop helpers no longer swallow a failed DROP: they confirm through the catalog and fail with the DROP's own error. The wheels_job_hosts warning no longer says the per-host cap is unavailable: the cap counts claimedBy in wheels_jobs and still works. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Peter Amiri --- vendor/wheels/Job.cfc | 2 +- .../wheels/tests/specs/jobs/JobSchemaSpec.cfc | 33 ++++++++++++++----- 2 files changed, 25 insertions(+), 10 deletions(-) diff --git a/vendor/wheels/Job.cfc b/vendor/wheels/Job.cfc index 89396a1fa2..8be6fd865c 100644 --- a/vendor/wheels/Job.cfc +++ b/vendor/wheels/Job.cfc @@ -2166,7 +2166,7 @@ component { return true; } if (!$jobSchema().autoCreateEnabled()) { - $warnAuxTableMissingOnce("wheels_job_hosts", "the per-host cap and drain are unavailable"); + $warnAuxTableMissingOnce("wheels_job_hosts", "this server's drain/resume and its host record in jobs status are unavailable (the per-host cap still works)"); return false; } if (Len($outermostWheelsTransaction())) { diff --git a/vendor/wheels/tests/specs/jobs/JobSchemaSpec.cfc b/vendor/wheels/tests/specs/jobs/JobSchemaSpec.cfc index 04566b804d..f532d3bc22 100644 --- a/vendor/wheels/tests/specs/jobs/JobSchemaSpec.cfc +++ b/vendor/wheels/tests/specs/jobs/JobSchemaSpec.cfc @@ -167,9 +167,9 @@ component extends="wheels.WheelsTest" { }); it("with jobsAutoCreateTables = false, reports a probe failure as itself when the table exists", function() { - // The catalog says the table is there but the probe fails (a lost connection, a - // permission): that error surfaces, not a misleading "missing table". - $dropJobTable(); + // The catalog says the table is there but the probe fails (here: a datasource that + // doesn't exist, as for a lost connection): that error surfaces, not a misleading + // "missing table". Independent of the table's state, so no DROP is involved. application.wheels.jobsAutoCreateTables = false; var schema = new wheels.JobSchema(); prepareMock(schema); @@ -177,6 +177,7 @@ component extends="wheels.WheelsTest" { var job = new wheels.Job(); prepareMock(job); job.$("$jobSchema", schema); + job.$property(propertyName = "$datasource", propertyScope = "variables", mock = "wheels_spec_no_such_datasource"); var probed = {type = "", message = ""}; try { job.$ensureJobTable(); @@ -262,12 +263,29 @@ component extends="wheels.WheelsTest" { } } + /** + * Drops wheels_jobs and makes sure it's gone: a DROP that fails (a lock) must fail the spec + * with its own error, not leave the table in place for the spec to misread. + */ private void function $dropJobTable() { + $dropTableOrFail("wheels_jobs"); + $clearUniqueKeyMemos(); + } + + private void function $dropTableOrFail(required string tableName) { + var outcome = {error = ""}; + var schema = new wheels.JobSchema(); + if (!schema.hasTable(arguments.tableName)) { + return; + } try { - queryExecute("DROP TABLE wheels_jobs", {}, {datasource = application.wheels.dataSourceName}); + queryExecute("DROP TABLE #arguments.tableName#", {}, {datasource = application.wheels.dataSourceName}); } catch (any e) { + outcome.error = e.message & " " & e.detail; + } + if (schema.hasTable(arguments.tableName)) { + Throw(type = "Spec.DropFailed", message = "Could not drop #arguments.tableName# for the spec: #outcome.error#"); } - $clearUniqueKeyMemos(); } /** @@ -285,10 +303,7 @@ component extends="wheels.WheelsTest" { private void function $dropAllJobTables() { for (var tableDef in new wheels.JobSchema().tables()) { - try { - queryExecute("DROP TABLE #tableDef.name#", {}, {datasource = application.wheels.dataSourceName}); - } catch (any e) { - } + $dropTableOrFail(tableDef.name); } $clearUniqueKeyMemos(); }