Fix DATETIME fractional-second rounding and detect DB type from datasource

bpamiri · claude · bpamiri · commit 12aa8fea11d0 · 2026-03-10T19:03:33.000-07:00
MySQL/H2 DATETIME columns round fractional seconds &gt;= 0.5 to the next
second. When enqueue() stores runAt=Now() and processNext() checks
runAt &lt;= Now() milliseconds later, the rounded-up value appears in the
future, making the job invisible. Fix: $now() helper truncates to whole
seconds.

Also replace application.wheels.adapterName with $detectDatabaseType()
using cfdbinfo on the actual datasource — the adapter may have been
detected from a different datasource in CI. Add H2-specific TIMESTAMP
type in $ensureJobTable().

Test now asserts enqueue persisted before checking processNext result.

Co-Authored-By: Claude Opus 4.6 &lt;noreply@anthropic.com&gt;
diff --git a/vendor/wheels/Job.cfc b/vendor/wheels/Job.cfc
@@ -72,7 +72,7 @@ component {
 			data = arguments.data,
 			queue = arguments.queue,
 			priority = arguments.priority,
-			runAt = Now()
+			runAt = $now()
 		);
 	}
 
@@ -94,7 +94,7 @@ component {
 			data = arguments.data,
 			queue = arguments.queue,
 			priority = arguments.priority,
-			runAt = DateAdd("s", arguments.seconds, Now())
+			runAt = DateAdd("s", arguments.seconds, $now())
 		);
 	}
 
@@ -132,7 +132,7 @@ component {
 	) {
 		local.id = CreateUUID();
 		local.serializedData = SerializeJSON(arguments.data);
-		local.now = Now();
+		local.now = $now();
 
 		try {
 			queryExecute(
@@ -198,7 +198,7 @@ component {
 	public struct function processQueue(string queue = "", numeric limit = 10) {
 		local.result = {processed = 0, failed = 0, errors = []};
 		local.params = {
-			runAt = {value = Now(), cfsqltype = "cf_sql_timestamp"}
+			runAt = {value = $now(), cfsqltype = "cf_sql_timestamp"}
 		};
 
 		local.sql = "SELECT id, jobClass, queue, data, attempts, maxRetries
@@ -246,7 +246,7 @@ component {
 				SET status = 'processing', attempts = attempts + 1, updatedAt = :updatedAt
 				WHERE id = :id",
 				{
-					updatedAt = {value = Now(), cfsqltype = "cf_sql_timestamp"},
+					updatedAt = {value = $now(), cfsqltype = "cf_sql_timestamp"},
 					id = {value = arguments.jobRow.id, cfsqltype = "cf_sql_varchar"}
 				},
 				{datasource = variables.$datasource}
@@ -268,8 +268,8 @@ component {
 				SET status = 'completed', completedAt = :completedAt, updatedAt = :updatedAt
 				WHERE id = :id",
 				{
-					completedAt = {value = Now(), cfsqltype = "cf_sql_timestamp"},
-					updatedAt = {value = Now(), cfsqltype = "cf_sql_timestamp"},
+					completedAt = {value = $now(), cfsqltype = "cf_sql_timestamp"},
+					updatedAt = {value = $now(), cfsqltype = "cf_sql_timestamp"},
 					id = {value = arguments.jobRow.id, cfsqltype = "cf_sql_varchar"}
 				},
 				{datasource = variables.$datasource}
@@ -291,7 +291,7 @@ component {
 			if (local.currentAttempts < local.maxRetries) {
 				// Schedule retry with configurable exponential backoff, capped at maxDelay
 				local.backoffSeconds = Min(this.baseDelay * (2 ^ local.currentAttempts), this.maxDelay);
-				local.nextRunAt = DateAdd("s", local.backoffSeconds, Now());
+				local.nextRunAt = DateAdd("s", local.backoffSeconds, $now());
 
 				queryExecute(
 					"UPDATE _wheels_jobs
@@ -303,7 +303,7 @@ component {
 					{
 						lastError = {value = Left(e.message, 1000), cfsqltype = "cf_sql_longvarchar"},
 						runAt = {value = local.nextRunAt, cfsqltype = "cf_sql_timestamp"},
-						updatedAt = {value = Now(), cfsqltype = "cf_sql_timestamp"},
+						updatedAt = {value = $now(), cfsqltype = "cf_sql_timestamp"},
 						id = {value = arguments.jobRow.id, cfsqltype = "cf_sql_varchar"}
 					},
 					{datasource = variables.$datasource}
@@ -324,9 +324,9 @@ component {
 						updatedAt = :updatedAt
 					WHERE id = :id",
 					{
-						failedAt = {value = Now(), cfsqltype = "cf_sql_timestamp"},
+						failedAt = {value = $now(), cfsqltype = "cf_sql_timestamp"},
 						lastError = {value = Left(e.message, 1000), cfsqltype = "cf_sql_longvarchar"},
-						updatedAt = {value = Now(), cfsqltype = "cf_sql_timestamp"},
+						updatedAt = {value = $now(), cfsqltype = "cf_sql_timestamp"},
 						id = {value = arguments.jobRow.id, cfsqltype = "cf_sql_varchar"}
 					},
 					{datasource = variables.$datasource}
@@ -388,8 +388,8 @@ component {
 				runAt = :runAt, updatedAt = :updatedAt
 			WHERE status = 'failed'";
 		local.params = {
-			runAt = {value = Now(), cfsqltype = "cf_sql_timestamp"},
-			updatedAt = {value = Now(), cfsqltype = "cf_sql_timestamp"}
+			runAt = {value = $now(), cfsqltype = "cf_sql_timestamp"},
+			updatedAt = {value = $now(), cfsqltype = "cf_sql_timestamp"}
 		};
 
 		if (Len(arguments.queue)) {
@@ -412,7 +412,7 @@ component {
 	 * @queue Optional queue name to filter by.
 	 */
 	public numeric function purgeCompleted(numeric days = 7, string queue = "") {
-		local.cutoff = DateAdd("d", -arguments.days, Now());
+		local.cutoff = DateAdd("d", -arguments.days, $now());
 		local.sql = "DELETE FROM _wheels_jobs WHERE status = 'completed' AND completedAt < :cutoff";
 		local.params = {
 			cutoff = {value = local.cutoff, cfsqltype = "cf_sql_timestamp"}
@@ -447,21 +447,24 @@ component {
 		}
 
 		try {
-			// Detect database adapter for type compatibility
-			local.adapterName = "";
-			if (StructKeyExists(application, "wheels") && StructKeyExists(application.wheels, "adapterName")) {
-				local.adapterName = application.wheels.adapterName;
-			}
+			// 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.adapterName == "OracleModel") {
+			if (local.dbType == "oracle") {
 				local.varcharType = "VARCHAR2";
 				local.textType = "CLOB";
 				local.datetimeType = "TIMESTAMP";
-			} else if (local.adapterName == "PostgreSQLModel" || local.adapterName == "CockroachDBModel") {
+			} 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";
@@ -503,4 +506,34 @@ component {
 			return false;
 		}
 	}
+
+	/**
+	 * Returns Now() truncated to whole seconds.
+	 * Prevents MySQL/H2 DATETIME rounding: when fractional seconds >= 0.5,
+	 * these databases round UP to the next second, making runAt appear in the future.
+	 */
+	private date function $now() {
+		local.n = Now();
+		return CreateDateTime(Year(local.n), Month(local.n), Day(local.n), Hour(local.n), Minute(local.n), Second(local.n));
+	}
+
+	/**
+	 * Detect the database type from the actual datasource via JDBC metadata.
+	 * Returns: "oracle", "postgresql", "h2", "mysql", "sqlserver", "sqlite", or "default".
+	 */
+	private string function $detectDatabaseType() {
+		try {
+			cfdbinfo(type = "version", datasource = "#variables.$datasource#", name = "local.info");
+			local.product = local.info.database_productname;
+			if (FindNoCase("oracle", local.product)) return "oracle";
+			if (FindNoCase("postgre", local.product)) return "postgresql";
+			if (FindNoCase("h2", local.product)) return "h2";
+			if (FindNoCase("mysql", local.product) || FindNoCase("mariadb", local.product)) return "mysql";
+			if (FindNoCase("sql server", local.product)) return "sqlserver";
+			if (FindNoCase("sqlite", local.product)) return "sqlite";
+		} catch (any e) {
+			// cfdbinfo not available — fall through to default
+		}
+		return "default";
+	}
 }
diff --git a/vendor/wheels/JobWorker.cfc b/vendor/wheels/JobWorker.cfc
@@ -35,7 +35,7 @@ component {
 
 		// Find the next candidate job
 		local.params = {
-			runAt = {value = Now(), cfsqltype = "cf_sql_timestamp"}
+			runAt = {value = $now(), cfsqltype = "cf_sql_timestamp"}
 		};
 
 		local.sql = "SELECT id, jobClass, queue, data, attempts, maxRetries
@@ -109,7 +109,7 @@ component {
 	 * @timeout Seconds after which a processing job is considered timed out. Default 300.
 	 */
 	public numeric function checkTimeouts(numeric timeout = 300) {
-		local.cutoff = DateAdd("s", -arguments.timeout, Now());
+		local.cutoff = DateAdd("s", -arguments.timeout, $now());
 
 		// Find timed-out jobs
 		try {
@@ -202,7 +202,7 @@ component {
 			worker = {id = this.workerId, startedAt = this.startedAt, processed = this.jobsProcessed, failed = this.jobsFailed}
 		};
 
-		local.lookback = DateAdd("n", -arguments.minutes, Now());
+		local.lookback = DateAdd("n", -arguments.minutes, $now());
 		local.params = {lookback = {value = local.lookback, cfsqltype = "cf_sql_timestamp"}};
 
 		// Throughput — completed and failed in the window
@@ -270,7 +270,7 @@ component {
 	 * @limit Maximum number of jobs to retry. 0 = unlimited.
 	 */
 	public numeric function retryFailed(string queue = "", numeric limit = 0) {
-		local.now = Now();
+		local.now = $now();
 
 		// If limit specified, get the IDs first
 		if (arguments.limit > 0) {
@@ -356,7 +356,7 @@ component {
 			throw(type = "Wheels.InvalidArgument", message = "Purge status must be 'completed' or 'failed'.");
 		}
 
-		local.cutoff = DateAdd("d", -arguments.days, Now());
+		local.cutoff = DateAdd("d", -arguments.days, $now());
 		local.dateColumn = (arguments.status == "completed") ? "completedAt" : "failedAt";
 
 		local.sql = "DELETE FROM _wheels_jobs WHERE status = :status AND #local.dateColumn# < :cutoff";
@@ -395,7 +395,7 @@ component {
 					SET status = 'processing', attempts = attempts + 1, updatedAt = :updatedAt
 					WHERE id = :id AND status = 'pending'",
 					{
-						updatedAt = {value = Now(), cfsqltype = "cf_sql_timestamp"},
+						updatedAt = {value = $now(), cfsqltype = "cf_sql_timestamp"},
 						id = {value = arguments.jobId, cfsqltype = "cf_sql_varchar"}
 					},
 					{datasource = variables.$datasource}
@@ -429,8 +429,8 @@ component {
 				SET status = 'completed', completedAt = :completedAt, updatedAt = :updatedAt
 				WHERE id = :id",
 				{
-					completedAt = {value = Now(), cfsqltype = "cf_sql_timestamp"},
-					updatedAt = {value = Now(), cfsqltype = "cf_sql_timestamp"},
+					completedAt = {value = $now(), cfsqltype = "cf_sql_timestamp"},
+					updatedAt = {value = $now(), cfsqltype = "cf_sql_timestamp"},
 					id = {value = arguments.jobRow.id, cfsqltype = "cf_sql_varchar"}
 				},
 				{datasource = variables.$datasource}
@@ -473,7 +473,7 @@ component {
 		}
 
 		local.backoffSeconds = Min(local.baseDelay * (2 ^ arguments.currentAttempts), local.maxDelay);
-		local.nextRunAt = DateAdd("s", local.backoffSeconds, Now());
+		local.nextRunAt = DateAdd("s", local.backoffSeconds, $now());
 
 		queryExecute(
 			"UPDATE _wheels_jobs
@@ -485,7 +485,7 @@ component {
 			{
 				lastError = {value = Left(arguments.errorMessage, 1000), cfsqltype = "cf_sql_longvarchar"},
 				runAt = {value = local.nextRunAt, cfsqltype = "cf_sql_timestamp"},
-				updatedAt = {value = Now(), cfsqltype = "cf_sql_timestamp"},
+				updatedAt = {value = $now(), cfsqltype = "cf_sql_timestamp"},
 				id = {value = arguments.jobId, cfsqltype = "cf_sql_varchar"}
 			},
 			{datasource = variables.$datasource}
@@ -515,9 +515,9 @@ component {
 				updatedAt = :updatedAt
 			WHERE id = :id",
 			{
-				failedAt = {value = Now(), cfsqltype = "cf_sql_timestamp"},
+				failedAt = {value = $now(), cfsqltype = "cf_sql_timestamp"},
 				lastError = {value = Left(arguments.errorMessage, 1000), cfsqltype = "cf_sql_longvarchar"},
-				updatedAt = {value = Now(), cfsqltype = "cf_sql_timestamp"},
+				updatedAt = {value = $now(), cfsqltype = "cf_sql_timestamp"},
 				id = {value = arguments.jobId, cfsqltype = "cf_sql_varchar"}
 			},
 			{datasource = variables.$datasource}
@@ -530,6 +530,15 @@ component {
 		);
 	}
 
+	/**
+	 * Returns Now() truncated to whole seconds.
+	 * Prevents MySQL/H2 DATETIME rounding: fractional seconds >= 0.5 round UP.
+	 */
+	private date function $now() {
+		local.n = Now();
+		return CreateDateTime(Year(local.n), Month(local.n), Day(local.n), Hour(local.n), Minute(local.n), Second(local.n));
+	}
+
 	/**
 	 * Ensure the _wheels_jobs table exists. Delegates to Job.cfc's implementation.
 	 */
diff --git a/vendor/wheels/tests/specs/jobs/JobWorkerSpec.cfc b/vendor/wheels/tests/specs/jobs/JobWorkerSpec.cfc
@@ -69,6 +69,10 @@ component extends="wheels.WheelsTest" {
 				local.testJob = new app.jobs.ProcessOrdersJob();
 				local.enqueued = local.testJob.enqueue(data = {test: true}, queue = "test_claim");
 
+				// Verify the job was persisted (catches silent enqueue failures)
+				expect(local.enqueued).toHaveKey("persisted");
+				expect(local.enqueued.persisted).toBeTrue();
+
 				// Process it
 				local.worker = new wheels.JobWorker();
 				local.result = local.worker.processNext(queues = "test_claim");
