Development

.github/workflows/cron.yml

@@ -25,9 +25,9 @@ jobs:
             experimental: true
     services:
       mysql:
-        image: mysql:5.7
+        image: mysql:8.0
         env:
-          MYSQL_RANDOM_ROOT_PASSWORD: yes
+          MYSQL_ROOT_PASSWORD: root
           MYSQL_USER: cbq
           MYSQL_PASSWORD: cbq
           MYSQL_DATABASE: cbq
@@ -56,6 +56,10 @@ jobs:
           box install ${{ matrix.coldbox }} --noSave
           box package list
 
+      - name: Configure MySQL user auth plugin
+        run: |
+          mysql -h 127.0.0.1 -P ${{ job.services.mysql.ports[3306] }} -uroot -proot -e "ALTER USER 'cbq'@'%' IDENTIFIED WITH mysql_native_password BY 'cbq'; FLUSH PRIVILEGES;"
+
       - name: Start server
         env:
           DB_HOST: localhost
@@ -78,4 +82,4 @@ jobs:
           DB_PASSWORD: cbq
           DB_DATABASE: cbq
           DB_SCHEMA: cbq
-        run: box testbox run
+        run: box testbox run

.github/workflows/pr.yml

@@ -22,9 +22,9 @@ jobs:
         cfengine: ["lucee@5", "lucee@6", "adobe@2021", "adobe@2023", "adobe@2025"]
     services:
       mysql:
-        image: mysql:5.7
+        image: mysql:8.0
         env:
-          MYSQL_RANDOM_ROOT_PASSWORD: yes
+          MYSQL_ROOT_PASSWORD: root
           MYSQL_USER: cbq
           MYSQL_PASSWORD: cbq
           MYSQL_DATABASE: cbq
@@ -52,6 +52,10 @@ jobs:
           box config set modules.commandbox-dotenv.checkEnvPreServerStart=false
           box package list
 
+      - name: Configure MySQL user auth plugin
+        run: |
+          mysql -h 127.0.0.1 -P ${{ job.services.mysql.ports[3306] }} -uroot -proot -e "ALTER USER 'cbq'@'%' IDENTIFIED WITH mysql_native_password BY 'cbq'; FLUSH PRIVILEGES;"
+
       - name: Start server
         env:
           DB_HOST: localhost
@@ -101,4 +105,4 @@ jobs:
       - name: Commit Format Changes
         uses: stefanzweifel/git-auto-commit-action@v5.2.0
         with:
-          commit_message: Apply cfformat changes
+          commit_message: Apply cfformat changes

.github/workflows/release.yml

@@ -17,9 +17,9 @@ jobs:
         cfengine: ["lucee@5", "lucee@6", "adobe@2021", "adobe@2023", "adobe@2025"]
     services:
       mysql:
-        image: mysql:5.7
+        image: mysql:8.0
         env:
-          MYSQL_RANDOM_ROOT_PASSWORD: yes
+          MYSQL_ROOT_PASSWORD: root
           MYSQL_USER: cbq
           MYSQL_PASSWORD: cbq
           MYSQL_DATABASE: cbq
@@ -47,6 +47,10 @@ jobs:
           box config set modules.commandbox-dotenv.checkEnvPreServerStart=false
           box package list
 
+      - name: Configure MySQL user auth plugin
+        run: |
+          mysql -h 127.0.0.1 -P ${{ job.services.mysql.ports[3306] }} -uroot -proot -e "ALTER USER 'cbq'@'%' IDENTIFIED WITH mysql_native_password BY 'cbq'; FLUSH PRIVILEGES;"
+
       - name: Start server
         env:
           DB_HOST: localhost

box.json

@@ -1,6 +1,6 @@
 {
     "name":"cbq",
-    "version":"5.0.7",
+    "version":"6.0.0-beta.3",
     "author":"Eric Peterson <eric@elpete.com>",
     "location":"forgeboxStorage",
     "homepage":"https://github.com/coldbox-modules/cbq",
@@ -31,9 +31,9 @@
         "cfmigrations":"modules/cfmigrations/"
     },
     "scripts":{
-        "format":"cfformat run ModuleConfig.cfc,models/**/*.cfc,tests/specs/**/*.cfc,tests/resources/app/handlers/**/*.cfc,tests/resources/app/config/**/*.cfc --overwrite",
-        "format:check":"cfformat check ModuleConfig.cfc,models/**/*.cfc,tests/specs/**/*.cfc,tests/resources/app/handlers/**/*.cfc,tests/resources/app/config/**/*.cfc --verbose",
-        "format:watch":"cfformat watch ModuleConfig.cfc,models/**/*.cfc,tests/specs/**/*.cfc,tests/resources/app/handlers/**/*.cfc,tests/resources/app/config/**/*.cfc",
+        "format":"cfformat run ModuleConfig.cfc,models/**/*.cfc,interceptors/**/*.cfc,tests/specs/**/*.cfc,tests/resources/app/handlers/**/*.cfc,tests/resources/app/config/**/*.cfc --overwrite",
+        "format:check":"cfformat check ModuleConfig.cfc,models/**/*.cfc,interceptors/**/*.cfc,tests/specs/**/*.cfc,tests/resources/app/handlers/**/*.cfc,tests/resources/app/config/**/*.cfc --verbose",
+        "format:watch":"cfformat watch ModuleConfig.cfc,models/**/*.cfc,interceptors/**/*.cfc,tests/specs/**/*.cfc,tests/resources/app/handlers/**/*.cfc,tests/resources/app/config/**/*.cfc",
         "install:2021":"cfpm install document,feed,zip"
     },
     "ignore":[

interceptors/LogFailedJobsInterceptor.cfc

@@ -44,7 +44,12 @@ component {
 				"null" : ( arguments.data.exception.type ?: "" ) == "",
 				"nulls" : ( arguments.data.exception.type ?: "" ) == ""
 			},
-			"exceptionMessage" : arguments.data.exception.message,
+			"exceptionMessage" : {
+				"value" : arguments.data.exception.message ?: "",
+				"cfsqltype" : "CF_SQL_VARCHAR",
+				"null" : ( arguments.data.exception.message ?: "" ) == "",
+				"nulls" : ( arguments.data.exception.message ?: "" ) == ""
+			},
 			"exceptionDetail" : {
 				"value" : arguments.data.exception.detail ?: "",
 				"cfsqltype" : "CF_SQL_VARCHAR",
@@ -57,23 +62,38 @@ component {
 				"null" : ( arguments.data.exception.extendedInfo ?: "" ) == "",
 				"nulls" : ( arguments.data.exception.extendedInfo ?: "" ) == ""
 			},
-			"exceptionStackTrace" : arguments.data.exception.stackTrace,
-			"exception" : serializeJSON( arguments.data.exception ),
-			"failedDate" : { "value": getCurrentUnixTimestamp(), "cfsqltype": "CF_SQL_BIGINT" },
-			"originalId" : { "value": arguments.data.job.getId(), "cfsqltype": "CF_SQL_VARCHAR" }
+			"exceptionStackTrace" : {
+				"value" : isNull( arguments.data.exception.stackTrace ) ? "" : (
+					isSimpleValue( arguments.data.exception.stackTrace ) ? arguments.data.exception.stackTrace : serializeJSON(
+						arguments.data.exception.stackTrace
+					)
+				),
+				"cfsqltype" : "CF_SQL_VARCHAR",
+				"null" : isNull( arguments.data.exception.stackTrace ) || (
+					isSimpleValue( arguments.data.exception.stackTrace ) && arguments.data.exception.stackTrace == ""
+				),
+				"nulls" : isNull( arguments.data.exception.stackTrace ) || (
+					isSimpleValue( arguments.data.exception.stackTrace ) && arguments.data.exception.stackTrace == ""
+				)
+			},
+			"exception" : isNull( arguments.data.exception ) ? javacast( "null", "" ) : serializeJSON(
+				arguments.data.exception
+			),
+			"failedDate" : {
+				"value" : getCurrentUnixTimestamp(),
+				"cfsqltype" : "CF_SQL_BIGINT"
+			},
+			"originalId" : {
+				"value" : arguments.data.job.getId(),
+				"cfsqltype" : "CF_SQL_VARCHAR"
+			}
 		};
 
 		try {
 			qb.table( variables.settings.logFailedJobsProperties.tableName )
-				.insert(
-					values = logData,
-					options = options
-				);
+				.insert( values = logData, options = options );
 		} catch ( any e ) {
-			log.error( "Failed to log failed job: #e.message#", {
-				"log": logData,
-				"exception": e
-			} );
+			log.error( "Failed to log failed job: #e.message#", { "log" : logData, "exception" : e } );
 			rethrow;
 		}
 	}

models/Jobs/Batch.cfc

@@ -3,13 +3,15 @@ component accessors="true" {
 	property name="dispatcher" inject="provider:Dispatcher@cbq";
 	property name="cbq" inject="provider:@cbq";
 	property name="wirebox" inject="wirebox";
+	property name="log" inject="logbox:logger:{this}";
 	property name="repository";
 
 	property name="id" type="string";
 	property name="name" type="string";
 	property name="totalJobs" type="numeric";
 	property name="pendingJobs" type="numeric";
 	property name="failedJobs" type="numeric";
+	property name="successfulJobs" type="numeric";
 	property name="failedJobIds" type="array";
 	property name="options" type="struct";
 	property name="createdDate" type="numeric";
@@ -49,7 +51,12 @@ component accessors="true" {
 		}
 
 		getRepository().markAsFinished( variables.id );
-		dispatchThenJobIfNeeded();
+
+		try {
+			dispatchThenJobIfNeeded();
+		} catch ( any e ) {
+			log.error( "Failed to dispatch then job for batch [#variables.id#]: #e.message#", e );
+		}
 
 		if ( counts.allJobsHaveRanExactlyOnce ) {
 			dispatchFinallyJobIfNeeded();
@@ -64,7 +71,11 @@ component accessors="true" {
 				cancel();
 			}
 
-			dispatchCatchJobIfNeeded( arguments.error );
+			try {
+				dispatchCatchJobIfNeeded( arguments.error );
+			} catch ( any e ) {
+				log.error( "Failed to dispatch catch job for batch [#variables.id#]: #e.message#", e );
+			}
 		}
 
 		if ( counts.allJobsHaveRanExactlyOnce ) {

models/Jobs/DBBatchRepository.cfc

@@ -52,14 +52,19 @@ component singleton accessors="true" {
 
 	public Batch function store( required PendingBatch batch ) {
 		var id = variables.timeBasedUUIDGenerator.generate().toString();
+		var batchName = arguments.batch.getName();
+		if ( isNull( batchName ) || !isSimpleValue( batchName ) ) {
+			batchName = "";
+		}
 
 		qb.table( variables.batchTableName )
 			.insert(
 				values = {
 					"id" : id,
-					"name" : arguments.batch.getName(),
+					"name" : batchName,
 					"totalJobs" : 0,
 					"pendingJobs" : 0,
+					"successfulJobs" : 0,
 					"failedJobs" : 0,
 					"failedJobIds" : "[]",
 					"options" : serializeJSON( arguments.batch.getOptions() ),
@@ -102,23 +107,20 @@ component singleton accessors="true" {
 				throw( type = "cbq.BatchNotFound", message = "No batch found for id [#arguments.batchId#]" );
 			}
 
+			var updatedValues = {
+				"pendingJobs" : data.pendingJobs - 1,
+				"successfulJobs" : data.successfulJobs + 1,
+				"failedJobs" : data.failedJobs
+			};
+
 			qb.table( variables.batchTableName )
 				.where( "id", arguments.batchId )
-				.update(
-					values = {
-						"pendingJobs" : data.pendingJobs - 1,
-						"failedJobs" : data.failedJobs,
-						"failedJobIds" : serializeJSON(
-							deserializeJSON( data.failedJobIds ).filter( ( failedJobId ) => failedJobId != jobId )
-						)
-					},
-					options = variables.defaultQueryOptions
-				);
+				.update( values = updatedValues, options = variables.defaultQueryOptions );
 
 			return {
 				"pendingJobs" : data.pendingJobs - 1,
 				"failedJobs" : data.failedJobs,
-				"allJobsHaveRanExactlyOnce" : ( data.pendingJobs - 1 ) - data.failedJobs == 0
+				"allJobsHaveRanExactlyOnce" : ( data.pendingJobs - 1 ) == 0
 			};
 		}
 	}
@@ -135,21 +137,20 @@ component singleton accessors="true" {
 				throw( type = "cbq.BatchNotFound", message = "No batch found for id [#arguments.batchId#]" );
 			}
 
+			var updatedValues = {
+				"pendingJobs" : data.pendingJobs - 1,
+				"failedJobs" : data.failedJobs + 1,
+				"failedJobIds" : serializeJSON( deserializeJSON( data.failedJobIds ).append( arguments.jobId ) )
+			};
+
 			qb.table( variables.batchTableName )
 				.where( "id", arguments.batchId )
-				.update(
-					values = {
-						"pendingJobs" : data.pendingJobs,
-						"failedJobs" : data.failedJobs + 1,
-						"failedJobIds" : serializeJSON( deserializeJSON( data.failedJobIds ).append( arguments.jobId ) )
-					},
-					options = variables.defaultQueryOptions
-				);
+				.update( values = updatedValues, options = variables.defaultQueryOptions );
 
 			return {
-				"pendingJobs" : data.pendingJobs,
+				"pendingJobs" : data.pendingJobs - 1,
 				"failedJobs" : data.failedJobs + 1,
-				"allJobsHaveRanExactlyOnce" : data.pendingJobs - ( data.failedJobs + 1 ) == 0
+				"allJobsHaveRanExactlyOnce" : ( data.pendingJobs - 1 ) == 0
 			};
 		}
 	}
@@ -191,6 +192,7 @@ component singleton accessors="true" {
 		batch.setTotalJobs( data.totalJobs );
 		batch.setPendingJobs( data.pendingJobs );
 		batch.setFailedJobs( data.failedJobs );
+		batch.setSuccessfulJobs( data.successfulJobs );
 		batch.setFailedJobIds( deserializeJSON( data.failedJobIds ) );
 		batch.setOptions( deserializeJSON( data.options ) );
 		batch.setCreatedDate( data.createdDate );

models/Providers/ColdBoxAsyncProvider.cfc

@@ -23,9 +23,9 @@ component accessors="true" extends="AbstractQueueProvider" {
 				sleep( delay * 1000 );
 				return true;
 			}, workerPool.getExecutor() )
-			.then( function() {
+			.thenCompose( function() {
 				job.setId( createUUID() );
-				if ( !isNull( arguments.currentAttempt ) ) {
+				if ( attempts > 0 ) {
 					job.setCurrentAttempt( attempts );
 				}
 				return marshalJob( job, workerPool );

models/Providers/DBProvider.cfc

@@ -394,12 +394,15 @@ component accessors="true" extends="AbstractQueueProvider" {
 							variables.getCurrentUnixTimestamp()
 						);
 				} );
-				// past the reserved date
-				q1.orWhere(
-					"reservedDate",
-					"<=",
-					variables.getCurrentUnixTimestamp() - pool.getTimeout()
-				);
+				// past the job's own timeout (availableDate was set to now + jobTimeout at reservation time)
+				q1.orWhere( ( q2 ) => {
+					q2.whereNotNull( "reservedDate" )
+						.where(
+							"availableDate",
+							"<=",
+							variables.getCurrentUnixTimestamp()
+						);
+				} );
 				// reserved by a worker but never released
 				q1.orWhere( ( q3 ) => {
 					q3.whereNull( "reservedDate" )
@@ -448,11 +451,14 @@ component accessors="true" extends="AbstractQueueProvider" {
 						q2.whereNull( "reservedBy" );
 						q2.whereNull( "reservedDate" );
 					} )
-					.orWhere(
-						"reservedDate",
-						"<=",
-						variables.getCurrentUnixTimestamp() - pool.getTimeout()
-					);
+					.orWhere( ( q2 ) => {
+						q2.whereNotNull( "reservedDate" )
+							.where(
+								"availableDate",
+								"<=",
+								variables.getCurrentUnixTimestamp()
+							);
+					} );
 			} )
 			.update(
 				values = {

models/Providers/SyncProvider.cfc

@@ -139,7 +139,7 @@ component accessors="true" extends="AbstractQueueProvider" {
 					invoke(
 						job,
 						"onFailure",
-						{ "excpetion" : e }
+						{ "exception" : e }
 					);
 				}
 

resources/database/migrations/2000_01_01_000008_add_successfulJobs_to_cbq_batches_table.cfc

@@ -0,0 +1,15 @@
+component {
+
+	function up( schema ) {
+		schema.alter( "cbq_batches", ( t ) => {
+			t.addColumn( t.unsignedInteger( "successfulJobs" ).default( 0 ) );
+		} );
+	}
+
+	function down( schema ) {
+		schema.alter( "cbq_batches", ( t ) => {
+			t.dropColumn( "successfulJobs" );
+		} );
+	}
+
+}

resources/database/migrations/2000_01_01_000009_make_cbq_batches_name_nullable.cfc

@@ -0,0 +1,15 @@
+component {
+
+	function up( schema, qb ) {
+		schema.alter( "cbq_batches", ( t ) => {
+			t.modifyColumn( "name", t.string( "name" ).nullable() );
+		} );
+	}
+
+	function down( schema, qb ) {
+		schema.alter( "cbq_batches", ( t ) => {
+			t.modifyColumn( "name", t.string( "name" ) );
+		} );
+	}
+
+}