diff --git a/box.json b/box.json index 90165a9..894593d 100644 --- a/box.json +++ b/box.json @@ -1,6 +1,6 @@ { "name":"cbq", - "version":"5.0.8", + "version":"6.0.0-beta.5", "author":"Eric Peterson ", "location":"forgeboxStorage", "homepage":"https://github.com/coldbox-modules/cbq", diff --git a/models/Providers/DBProvider.cfc b/models/Providers/DBProvider.cfc index 654f111..2e51880 100644 --- a/models/Providers/DBProvider.cfc +++ b/models/Providers/DBProvider.cfc @@ -279,10 +279,10 @@ component accessors="true" extends="AbstractQueueProvider" { { "job" : jobCFC.getMemento() } ); } - markJobAsFailedById( arguments.record.id, arguments.pool ); - ensureFailedBatchJobIsRecorded( + markJobFailed( jobCFC, - "Job exceeded maximum attempts (#jobMaxAttempts#) before execution." + arguments.pool, + buildMaxAttemptsReachedException( jobCFC, jobMaxAttempts ) ); return; } @@ -349,22 +349,7 @@ component accessors="true" extends="AbstractQueueProvider" { newQuery() .table( variables.tableName ) .where( "id", arguments.id ) - .update( - values = { - "failedDate" : getCurrentUnixTimestamp(), - "reservedBy" : { - "value" : "", - "null" : true, - "nulls" : true - }, - "reservedDate" : { - "value" : "", - "null" : true, - "nulls" : true - } - }, - options = variables.defaultQueryOptions - ); + .update( values = { "failedDate" : getCurrentUnixTimestamp() }, options = variables.defaultQueryOptions ); } private void function markJobAsFailedById( required numeric id, WorkerPool pool ) { @@ -381,6 +366,17 @@ component accessors="true" extends="AbstractQueueProvider" { .update( values = { "failedDate" : getCurrentUnixTimestamp() }, options = variables.defaultQueryOptions ); } + private any function buildMaxAttemptsReachedException( required AbstractJob job, required numeric maxAttempts ) { + try { + throw( + type = "cbq.MaxAttemptsReached", + message = "Job ###arguments.job.getId()# exceeded maximum attempts (#arguments.maxAttempts#) before execution." + ); + } catch ( any e ) { + return e; + } + } + public void function releaseJob( required AbstractJob job, required WorkerPool pool ) { arguments.job.setCurrentAttempt( arguments.job.getCurrentAttempt() + 1 ); newQuery() @@ -502,6 +498,15 @@ component accessors="true" extends="AbstractQueueProvider" { "<=", variables.getCurrentUnixTimestamp() ); + } ) + .orWhere( ( q2 ) => { + q2.whereNotNull( "reservedBy" ) + .whereNull( "reservedDate" ) + .where( + "availableDate", + "<=", + variables.getCurrentUnixTimestamp() + ); } ); } ) .update( diff --git a/tests/specs/integration/Providers/DBProviderMaxAttemptsSpec.cfc b/tests/specs/integration/Providers/DBProviderMaxAttemptsSpec.cfc index 3729861..fa07b61 100644 --- a/tests/specs/integration/Providers/DBProviderMaxAttemptsSpec.cfc +++ b/tests/specs/integration/Providers/DBProviderMaxAttemptsSpec.cfc @@ -3,23 +3,44 @@ component extends="tests.resources.ModuleIntegrationSpec" appMapping="/app" { function run() { describe( "DBProvider maxAttempts safeguards", function() { beforeEach( function() { - variables.provider = getWireBox().getInstance( "DBProvider@cbq" ).setProperties( {} ); + variables.workerPools = []; + variables.provider = getWireBox() + .buildInstance( getWireBox().getBinder().getMapping( "DBProvider@cbq" ) ) + .setProperties( {} ); + getWireBox().autowire( + target = variables.provider, + mapping = getWireBox().getBinder().getMapping( "DBProvider@cbq" ) + ); makePublic( variables.provider, "processLockedRecord" ); variables.pool = makeWorkerPool( variables.provider ); + variables.cbqSettings = getController().getModuleSettings( "cbq" ); + variables.originalLogFailedJobs = variables.cbqSettings.logFailedJobs; variables.provider .newQuery() .table( "cbq_jobs" ) .delete(); + variables.provider + .newQuery() + .table( "cbq_failed_jobs" ) + .delete(); } ); afterEach( function() { + for ( var pool in variables.workerPools ) { + pool.shutdown( force = true, timeout = 1 ); + } + variables.cbqSettings.logFailedJobs = variables.originalLogFailedJobs; variables.provider .newQuery() .table( "cbq_jobs" ) .delete(); + variables.provider + .newQuery() + .table( "cbq_failed_jobs" ) + .delete(); } ); - it( "forceFailJob sets failedDate and clears the reservation", function() { + it( "forceFailJob sets failedDate and preserves the reservation", function() { var job = getWireBox().getInstance( "SendWelcomeEmailJob" ).setMaxAttempts( 3 ); variables.provider.push( "default", job ); @@ -49,8 +70,8 @@ component extends="tests.resources.ModuleIntegrationSpec" appMapping="/app" { expect( row.failedDate ).notToBeNull( "failedDate should be set" ); expect( row.failedDate ).toBeGT( 0, "failedDate should be a unix timestamp" ); - expect( row.reservedBy ?: "" ).toBe( "", "reservedBy should be cleared" ); - expect( row.reservedDate ?: "" ).toBe( "", "reservedDate should be cleared" ); + expect( row.reservedBy ?: "" ).toBe( variables.pool.getUniqueId(), "reservedBy should be preserved" ); + expect( row.reservedDate ).toBe( now, "reservedDate should be preserved" ); } ); it( "skips dispatch and marks the job failed when attempts already meets maxAttempts", function() { @@ -98,6 +119,93 @@ component extends="tests.resources.ModuleIntegrationSpec" appMapping="/app" { .where( "id", record.id ) .first(); expect( row.failedDate ).notToBeNull( "the runaway job should be marked failed" ); + expect( row.reservedBy ?: "" ).toBe( + variables.pool.getUniqueId(), + "reservedBy should be preserved after terminal failure" + ); + expect( row.reservedDate ?: "" ).toBe( + "", + "reservedDate should remain unchanged after terminal failure" + ); + } ); + + it( "logs a failed job when a timeout retry already meets maxAttempts before dispatch", function() { + variables.cbqSettings.logFailedJobs = true; + var job = getWireBox().getInstance( "AlwaysErrorJob" ).setMaxAttempts( 3 ); + variables.provider.push( "default", job ); + + var now = javacast( "long", getTickCount() / 1000 ); + variables.provider + .newQuery() + .table( "cbq_jobs" ) + .update( { + "reservedBy" : variables.pool.getUniqueId(), + "reservedDate" : { + "value" : "", + "null" : true, + "nulls" : true + }, + "availableDate" : now - 1, + "attempts" : 3 + } ); + + var record = variables.provider + .newQuery() + .from( "cbq_jobs" ) + .first(); + + variables.provider.processLockedRecord( record, variables.pool ); + + var failedLog = variables.provider + .newQuery() + .from( "cbq_failed_jobs" ) + .first(); + + expect( failedLog ).notToBeNull( + "terminal maxAttempts failures discovered by the timeout watcher should be visible in the failed jobs log" + ); + expect( failedLog.originalId ).toBe( record.id ); + expect( failedLog.exceptionType ).toBe( "cbq.MaxAttemptsReached" ); + expect( failedLog.exceptionMessage ).toInclude( "exceeded maximum attempts" ); + } ); + + it( "persists the terminal attempt before failing a job that reaches maxAttempts during execution", function() { + var job = getWireBox().getInstance( "AlwaysErrorJob" ).setMaxAttempts( 3 ); + variables.provider.push( "default", job ); + + var now = javacast( "long", getTickCount() / 1000 ); + variables.provider + .newQuery() + .table( "cbq_jobs" ) + .update( { + "reservedBy" : variables.pool.getUniqueId(), + "reservedDate" : { + "value" : "", + "null" : true, + "nulls" : true + }, + "availableDate" : now - 1, + "attempts" : 2 + } ); + + var record = variables.provider + .newQuery() + .from( "cbq_jobs" ) + .first(); + + variables.provider.processLockedRecord( record, variables.pool ); + var row = waitForFailedJobRow( record.id ); + + expect( row.failedDate ).notToBeNull( "the third failed run should mark the row failed" ); + expect( row.attempts ).toBe( 3, "the terminal third run should be reflected in the attempts column" ); + expect( row.reservedBy ?: "" ).toBe( + variables.pool.getUniqueId(), + "reservedBy should be preserved after terminal failure" + ); + expect( row.reservedDate ?: "" ).notToBe( + "", + "reservedDate should be preserved after terminal failure" + ); } ); it( "still proceeds normally when attempts is below maxAttempts", function() { @@ -151,7 +259,6 @@ component extends="tests.resources.ModuleIntegrationSpec" appMapping="/app" { .newQuery() .table( "cbq_jobs" ) .delete(); - var job = getWireBox() .getInstance( "AlwaysErrorJob" ) .setMaxAttempts( 5 ) @@ -253,16 +360,39 @@ component extends="tests.resources.ModuleIntegrationSpec" appMapping="/app" { .value( "id" ); } + private struct function waitForFailedJobRow( required numeric id ) { + for ( var i = 1; i <= 20; i++ ) { + var row = variables.provider + .newQuery() + .from( "cbq_jobs" ) + .where( "id", arguments.id ) + .first(); + if ( !isNull( row.failedDate ) && row.attempts == 3 ) { + return row; + } + sleep( 100 ); + } + + return variables.provider + .newQuery() + .from( "cbq_jobs" ) + .where( "id", arguments.id ) + .first(); + } + private any function makeWorkerPool( required any provider ) { + var uniqueName = createUUID(); var connection = getInstance( "QueueConnection@cbq" ) - .setName( "TestMaxAttemptsConnection" ) + .setName( "TestMaxAttemptsConnection-#uniqueName#" ) .setProvider( arguments.provider ); - return getInstance( "WorkerPool@cbq" ) - .setName( "TestMaxAttemptsPool" ) + var pool = getInstance( "WorkerPool@cbq" ) + .setName( "TestMaxAttemptsPool-#uniqueName#" ) .setConnection( connection ) .setConnectionName( connection.getName() ) .startWorkers(); + variables.workerPools.append( pool ); + return pool; } } diff --git a/tests/specs/integration/Providers/DBProviderTimeoutWatcherSpec.cfc b/tests/specs/integration/Providers/DBProviderTimeoutWatcherSpec.cfc index 84a7d8a..cf41e88 100644 --- a/tests/specs/integration/Providers/DBProviderTimeoutWatcherSpec.cfc +++ b/tests/specs/integration/Providers/DBProviderTimeoutWatcherSpec.cfc @@ -5,6 +5,7 @@ component extends="tests.resources.ModuleIntegrationSpec" appMapping="/app" { beforeEach( function() { variables.provider = getWireBox().getInstance( "DBProvider@cbq" ).setProperties( {} ); makePublic( variables.provider, "fetchPotentiallyOpenRecords" ); + makePublic( variables.provider, "tryToLockRecords" ); variables.pool = makeWorkerPool( variables.provider ); // clean up any leftover test records variables.provider @@ -89,6 +90,41 @@ component extends="tests.resources.ModuleIntegrationSpec" appMapping="/app" { "A job past the pool timeout but still within its job-specific timeout should not be re-grabbed" ); } ); + + it( "locks an orphaned reservation that was claimed without a reserved date", function() { + var job = getWireBox().getInstance( "SendWelcomeEmailJob" ); + variables.provider.push( "default", job ); + + var deadWorkerUUID = createUUID(); + var now = javacast( "long", getTickCount() / 1000 ); + variables.provider + .newQuery() + .table( "cbq_jobs" ) + .update( { + "reservedBy" : deadWorkerUUID, + "reservedDate" : { + "value" : "", + "null" : true, + "nulls" : true + }, + "availableDate" : now - 1 + } ); + + var ids = variables.provider.fetchPotentiallyOpenRecords( capacity = 10, pool = variables.pool ); + variables.provider.tryToLockRecords( ids, variables.pool ); + + var row = variables.provider + .newQuery() + .from( "cbq_jobs" ) + .first(); + + expect( ids ).toHaveLength( 1, "The orphaned reservation should be selected for reclaiming" ); + expect( row.reservedBy ).toBe( + variables.pool.getUniqueId(), + "The live worker pool should be able to claim the orphaned reservation" + ); + expect( row.reservedDate ?: "" ).toBe( "", "The job should remain pending reservation processing" ); + } ); } ); }