Expose publish outcomes and explicit worker attempt boundaries

.github/workflows/pr.yml

@@ -98,12 +98,10 @@ jobs:
         uses: Ortus-Solutions/setup-commandbox@v2.0.1
 
       - name: Install CFFormat
-        run: box install commandbox-cfformat
+        # 0.22.1 adds whitespace before unquoted struct-key colons on every pass.
+        run: box install commandbox-cfformat@0.21.0
 
-      - name: Run CFFormat
-        run: box run-script format
-
-      - name: Commit Format Changes
-        uses: stefanzweifel/git-auto-commit-action@v5.2.0
-        with:
-          commit_message: Apply cfformat changes
+      - name: Check CFFormat
+        # Report changes for the author to apply locally; never create another
+        # workflow-triggering commit from the formatter job.
+        run: box run-script format:check

ModuleConfig.cfc

@@ -5,7 +5,7 @@ component {
 	this.webUrl = "https://github.com/coldbox-modules/cbq";
 	this.cfmapping = "cbq";
 
-	variables.customInterceptionPointMethodNames = "onCBQJobAdded,onCBQJobMarshalled,onCBQJobComplete,onCBQJobException,onCBQJobFailed";
+	variables.customInterceptionPointMethodNames = "onCBQJobAdded,onCBQJobMarshalled,onCBQJobComplete,onCBQJobException,onCBQJobFailed,onCBQJobAttemptScheduled,onCBQJobAttemptFinished,onCBQJobExecutionStarted,onCBQJobExecutionExited,onCBQJobPublished,onCBQJobPublishException";
 
 	function configure() {
 		settings = {

README.md

@@ -301,3 +301,11 @@ With a larger batch size, each job is still announced separately and becomes its
 The DB provider writes at most 100 rows (500 bindings) per insert. Other providers continue through their ordinary `push` implementation unless they implement `pushMany(entries)`. Entries contain the ordinary `push` arguments: `queueName`, `job`, and optionally `delay` and `attempts`. `DBProvider.pushMany` also bounds direct calls to 100 rows per insert. It uses the configured table, query options and datasource.
 
 The caller owns transaction boundaries. Wrap the dispatch with related database changes when all chunks must commit or roll back together. Without a surrounding transaction, a later event, serialization or insert failure can leave earlier chunks persisted.
+
+## Observational execution hooks
+
+Publishing announces onCBQJobPublished or onCBQJobPublishException after the provider boundary, including bulk dispatch. Added observational hooks isolate observer failures and preserve the original enqueue error. onCBQJobAdded retains its existing contract.
+
+Every attempt announces onCBQJobAttemptScheduled, onCBQJobExecutionStarted, onCBQJobExecutionExited, and onCBQJobAttemptFinished. Data contains job, immutable executionId, and attempt. Finished additionally contains status: ok, released, internal_error, cancelled, or deadline_exceeded. Scheduled precedes dispatch, Started/Exited bracket the actual worker body in try/finally, while Finished can execute on another thread. Consumers must key handles by executionId, restore context in Exited, and finish atomically. A timed-out worker can exit later. Existing completion/error hooks retain their behavior.
+
+Focused coverage: synchronous observer failures, bulk enqueue failures, and a real asynchronous timeout. Comprehensive retry/chain/batch/provider and engine matrices remain separate checks.

box.json

@@ -9,34 +9,34 @@
         "type":"git",
         "URL":"https://github.com/coldbox-modules/cbq"
     },
-    "bugs":"https://github.com/coldbox-modules/cbq/issues",
-    "slug":"cbq",
-    "shortDescription":"A protocol-based queueing system for ColdBox",
-    "description":"A protocol-based queueing system for ColdBox",
-    "type":"modules",
-    "dependencies":{
-        "qb":"^13.0.0",
-        "str":"^4.0.0"
+    "bugs": "https://github.com/coldbox-modules/cbq/issues",
+    "slug": "cbq",
+    "shortDescription": "A protocol-based queueing system for ColdBox",
+    "description": "A protocol-based queueing system for ColdBox",
+    "type": "modules",
+    "dependencies": {
+        "qb": "^13.0.0",
+        "str": "^4.0.0"
     },
-    "devDependencies":{
-        "testbox":"^6.0.0",
-        "coldbox":"^7.0.0",
-        "cfmigrations":"^5.0.0"
+    "devDependencies": {
+        "testbox": "^6.0.0",
+        "coldbox": "^7.0.0",
+        "cfmigrations": "^5.0.0"
     },
-    "installPaths":{
-        "testbox":"testbox/",
-        "coldbox":"tests/resources/app/coldbox/",
-        "qb":"modules/qb/",
-        "str":"modules/str/",
-        "cfmigrations":"modules/cfmigrations/"
+    "installPaths": {
+        "testbox": "testbox/",
+        "coldbox": "tests/resources/app/coldbox/",
+        "qb": "modules/qb/",
+        "str": "modules/str/",
+        "cfmigrations": "modules/cfmigrations/"
     },
-    "scripts":{
-        "format":"cfformat run ModuleConfig.cfc,models/**/*.cfc,interceptors/**/*.cfc,tests/specs/**/*.cfc,tests/resources/app/handlers/**/*.cfc,tests/resources/app/config/**/*.cfc,tests/resources/app/models/**/*.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,tests/resources/app/models/**/*.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,tests/resources/app/models/**/*.cfc",
-        "install:2021":"cfpm install document,feed,zip"
+    "scripts": {
+        "format": "cfformat run ModuleConfig.cfc,models/**/*.cfc,interceptors/**/*.cfc,tests/specs/**/*.cfc,tests/resources/app/handlers/**/*.cfc,tests/resources/app/config/**/*.cfc,tests/resources/app/models/**/*.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,tests/resources/app/models/**/*.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,tests/resources/app/models/**/*.cfc",
+        "install:2021": "cfpm install document,feed,zip"
     },
-    "ignore":[
+    "ignore": [
         "**/.*",
         "test",
         "tests"

models/Jobs/Dispatcher.cfc

@@ -2,6 +2,7 @@ component singleton accessors="true" {
 
 	property name="interceptorService" inject="box:interceptorService";
 	property name="config" inject="Config@cbq";
+	property name="log" inject="logbox:logger:{this}";
 
 	public Dispatcher function dispatch( required any job ) {
 		var connectionName = arguments.job.getConnection();
@@ -22,11 +23,26 @@ component singleton accessors="true" {
 		);
 
 		arguments.job.setCurrentAttempt( 0 );
-		connection.push(
-			queueName = queueName,
-			job = arguments.job,
-			delay = delay,
-			attempts = 0
+		try {
+			connection.push(
+				queueName = queueName,
+				job = arguments.job,
+				delay = delay,
+				attempts = 0
+			);
+		} catch ( any failure ) {
+			announcePublishResult(
+				"onCBQJobPublishException",
+				arguments.job,
+				connection,
+				failure
+			);
+			rethrow;
+		}
+		announcePublishResult(
+			"onCBQJobPublished",
+			arguments.job,
+			connection
 		);
 
 		return this;
@@ -61,11 +77,11 @@ component singleton accessors="true" {
 				"attempts" : 0
 			};
 			if ( arguments.batchSize == 1 ) {
-				connection.push( argumentCollection = entry );
+				publishMany( connection, [ entry ] );
 			} else {
 				pending.append( entry );
 				if ( pending.len() == arguments.batchSize ) {
-					connection.pushMany( pending );
+					publishMany( connection, pending );
 					pending = [];
 				}
 			}
@@ -78,4 +94,55 @@ component singleton accessors="true" {
 		return this;
 	}
 
+	private function publishMany( required any connection, required array entries ) {
+		try {
+			if ( arguments.entries.len() == 1 ) {
+				arguments.connection.push( argumentCollection = arguments.entries[ 1 ] );
+			} else {
+				arguments.connection.pushMany( arguments.entries );
+			}
+		} catch ( any failure ) {
+			for ( var entry in arguments.entries ) {
+				announcePublishResult(
+					"onCBQJobPublishException",
+					entry.job,
+					arguments.connection,
+					failure
+				);
+			}
+			rethrow;
+		}
+		for ( var entry in arguments.entries ) {
+			announcePublishResult(
+				"onCBQJobPublished",
+				entry.job,
+				arguments.connection
+			);
+		}
+	}
+
+	private function announcePublishResult(
+		required string state,
+		required any job,
+		required any connection,
+		any exception
+	) {
+		try {
+			variables.interceptorService.announce(
+				arguments.state,
+				{
+					"job" : arguments.job,
+					"connection" : arguments.connection,
+					"exception" : arguments.exception ?: javacast( "null", "" )
+				}
+			);
+		} catch ( any observerFailure ) {
+			new cbq.models.Support.FailureDiagnostics().report(
+				arguments.state,
+				observerFailure,
+				variables.log ?: javacast( "null", "" )
+			);
+		}
+	}
+
 }

models/Providers/AbstractQueueProvider.cfc

@@ -105,27 +105,43 @@ component accessors="true" {
 		function afterJobHook
 	) {
 		arguments.job.setCurrentAttempt( ( arguments.job.getCurrentAttempt() ?: 0 ) + 1 );
+		var observation = {
+			"job" : arguments.job,
+			"executionId" : createUUID(),
+			"attempt" : arguments.job.getCurrentAttempt()
+		};
+		announceExecutionObservation( "onCBQJobAttemptScheduled", observation );
 		return variables.async
 			.newFuture( function() {
-				if ( variables.log.canDebug() ) {
-					variables.log.debug( "Marshaling job ###job.getId()#", job.getMemento() );
-				}
+				announceExecutionObservation( "onCBQJobExecutionStarted", observation );
+				try {
+					if ( variables.log.canDebug() ) {
+						variables.log.debug( "Marshaling job ###job.getId()#", job.getMemento() );
+					}
 
-				beforeJobRun( job );
-				if ( structKeyExists( job, "before" ) ) {
-					job.before();
-				}
+					beforeJobRun( job );
+					if ( structKeyExists( job, "before" ) ) {
+						job.before();
+					}
 
-				variables.interceptorService.announce( "onCBQJobMarshalled", { "job" : job } );
+					variables.interceptorService.announce( "onCBQJobMarshalled", { "job" : job } );
 
-				if ( variables.log.canDebug() ) {
-					variables.log.debug( "Running job ###job.getId()#", job.getMemento() );
-				}
+					if ( variables.log.canDebug() ) {
+						variables.log.debug( "Running job ###job.getId()#", job.getMemento() );
+					}
 
-				return job.handle();
+					// These paired hooks bracket the actual worker execution, including failures.
+					// Completion callbacks may run on a different thread.
+					return job.handle();
+				} finally {
+					announceExecutionObservation( "onCBQJobExecutionExited", observation );
+				}
 			}, arguments.pool.getExecutor() )
 			.orTimeout( getTimeoutForJob( arguments.job, arguments.pool ), "seconds" )
 			.then( function( result ) {
+				var completed = structCopy( observation );
+				completed.status = job.getIsReleased() ? "released" : "ok";
+				announceExecutionObservation( "onCBQJobAttemptFinished", completed );
 				if ( job.getIsReleased() ) {
 					variables.log.debug( "Job [#job.getId()#] requested manual release." );
 
@@ -180,6 +196,11 @@ component accessors="true" {
 				}
 			} )
 			.onException( function( e ) {
+				var failed = structCopy( observation );
+				failed.status = job.getIsCancelled() ? "cancelled" : (
+					!isNull( e ) && findNoCase( "Timeout", e.toString() ) ? "deadline_exceeded" : "internal_error"
+				);
+				announceExecutionObservation( "onCBQJobAttemptFinished", failed );
 				try {
 					// log failed job
 					if ( !isNull( e ) && "java.util.concurrent.CompletionException" == e.getClass().getName() ) {
@@ -264,7 +285,12 @@ component accessors="true" {
 										"releaseException" : releaseException
 									}
 								);
-							} catch ( any ignored ) {
+							} catch ( any diagnosticFailure ) {
+								new cbq.models.Support.FailureDiagnostics().report(
+									"releaseJob diagnostic",
+									diagnosticFailure,
+									variables.log ?: javacast( "null", "" )
+								);
 							}
 							markJobFailed(
 								job,
@@ -293,11 +319,21 @@ component accessors="true" {
 								"handlerException" : outerException
 							}
 						);
-					} catch ( any ignored ) {
+					} catch ( any diagnosticFailure ) {
+						new cbq.models.Support.FailureDiagnostics().report(
+							"onException handler diagnostic",
+							diagnosticFailure,
+							variables.log ?: javacast( "null", "" )
+						);
 					}
 					try {
 						forceFailJob( job.getId(), pool, job );
-					} catch ( any ignored ) {
+					} catch ( any diagnosticFailure ) {
+						new cbq.models.Support.FailureDiagnostics().report(
+							"forceFailJob fallback",
+							diagnosticFailure,
+							variables.log ?: javacast( "null", "" )
+						);
 					}
 				}
 
@@ -409,7 +445,12 @@ component accessors="true" {
 					arguments.pool,
 					arguments.job
 				);
-			} catch ( any ignored ) {
+			} catch ( any diagnosticFailure ) {
+				new cbq.models.Support.FailureDiagnostics().report(
+					"afterJobFailed forceFailJob fallback",
+					diagnosticFailure,
+					variables.log ?: javacast( "null", "" )
+				);
 			}
 		}
 
@@ -439,7 +480,12 @@ component accessors="true" {
 					{ "exception" : arguments.exception }
 				);
 			}
-		} catch ( any ignored ) {
+		} catch ( any diagnosticFailure ) {
+			new cbq.models.Support.FailureDiagnostics().report(
+				"side effect diagnostic",
+				diagnosticFailure,
+				variables.log ?: javacast( "null", "" )
+			);
 		}
 	}
 
@@ -595,4 +641,16 @@ component accessors="true" {
 		return;
 	}
 
+	function announceExecutionObservation( required string state, required struct data ) {
+		try {
+			variables.interceptorService.announce( arguments.state, arguments.data );
+		} catch ( any observerFailure ) {
+			new cbq.models.Support.FailureDiagnostics().report(
+				arguments.state,
+				observerFailure,
+				variables.log ?: javacast( "null", "" )
+			);
+		}
+	}
+
 }