Batched Jobs

cbq allows for tracking jobs as part of a batch. The main use case for batched jobs is to dispatch additional jobs when a batch has completed successfully, completed with failures, or completed in any fashion. (Think try, catch, finally.)

Setup

To utilize batches, you first need to set up a Batch repository. Batches are tracked separate from jobs. cbq utilizes a database repository to track batches regardless of what provider your Job's connection uses.

Copy the batch-table migrations from resources/database/migrations to your own project and run them in filename order with cfmigrations. The initial 2000_01_01_000001_create_cbq_batches_table.cfc migration needs the later batch migrations described in Result Accounting. Alternatively, create the final structure using the migration or SQL scripts below.

You can customize the batch repository using the batchRepositoryProperties of cbq's moduleSettings. This is a struct with the following properties:

tableName

This defaults to cbq_batches. You can set this to any unique table name in your datasource.

datasource

The datasource used for the batch table. This overrides a datasource supplied in queryOptions.

queryOptions

This is a struct of options that will be passed to queryExecute. The most common use case for this property is to specify a specific datasource to use for the batches table. This defaults to an empty struct ({}).

Defining Batches

To create a batch, you can get an instance of PendingBatch@cbq from WireBox or use the cbq.batch() helper method.

Once you have a PendingBatch instance, you can add jobs to the batch using the add method.

add

Adds a single Job or an array of Jobs to a PendingBatch.

Arguments	Type	Required	Default	Description
job	string | Job | array<Job>	true		The Job WireBox id, Job instance, or array of Job instances to add to the PendingBatch.
properties	Struct	false	{}	A struct of properties for the Job instance. (Only used when providing a Job WireBox id.)
chain	array<Job>	false	[]	An array of Jobs to chain after this Job. (Only used when providing a Job WireBox id.)
queue	string	false		The queue the Job belongs to. (Only used when providing a Job WireBox id.)
connection	string	false		The Connection to dispatch the Job on. (Only used when providing a Job WireBox id.)
backoff	numeric	false		The amount of time, in seconds, to wait between attempting Jobs. (Only used when providing a Job WireBox id.)
timeout	numeric	false		The amount of time, in seconds, to wait before treating a Job as erroring. (Only used when providing a Job WireBox id.)
maxAttempts	numeric	false		The maximum amount of attempts of a Job before treating the Job as failed. (Only used when providing a Job WireBox id.)

Return: The PendingBatch to be dispatched.

batch
    .add( cbq.job( "ImportCsvJob", { "start": 1, "end": 100 } ) )
    .add( "ImportCsvJob", { "start": 101, "end": 200 } )
    .add( [
        cbq.job( "ImportCsvJob", { "start": 201, "end": 300 } ),
	cbq.job( "ImportCsvJob", { "start": 301, "end": 400 } ),
	cbq.job( "ImportCsvJob", { "start": 401, "end": 500 } )
    ] );

The primary use case of batches is to dispatch lifecycle jobs when the batch has completed and if the batch completed successfully or completed with failures. These jobs can be configured using the then, onFailure, and onComplete methods. Use the latter names for Adobe ColdFusion compatibility; catch and finally are deprecated aliases on Lucee and BoxLang.

then

Defines a Job to be dispatched when all the jobs in the batch finishes successfully.

Arguments	Type	Required	Default	Description
job	string | Job	true		The Job WireBox id or Job instance to execute if all the jobs in the Batch complete successfully.
properties	Struct	false	{}	A struct of properties for the Job instance. (Only used when providing a Job WireBox id.)
chain	array<Job>	false	[]	An array of Jobs to chain after this Job. (Only used when providing a Job WireBox id.)
queue	string	false		The queue the Job belongs to. (Only used when providing a Job WireBox id.)
connection	string	false		The Connection to dispatch the Job on. (Only used when providing a Job WireBox id.)
backoff	numeric	false		The amount of time, in seconds, to wait between attempting Jobs. (Only used when providing a Job WireBox id.)
timeout	numeric	false		The amount of time, in seconds, to wait before treating a Job as erroring. (Only used when providing a Job WireBox id.)
maxAttempts	numeric	false		The maximum amount of attempts of a Job before treating the Job as failed. (Only used when providing a Job WireBox id.)

Return: The PendingBatch instance.

cbq.batch( [
    cbq.job( "ImportCsvJob", { "start": 1, "end": 100 } ),
    cbq.job( "ImportCsvJob", { "start": 101, "end": 200 } ),
    cbq.job( "ImportCsvJob", { "start": 201, "end": 300 } ),
    cbq.job( "ImportCsvJob", { "start": 301, "end": 400 } ),
    cbq.job( "ImportCsvJob", { "start": 401, "end": 500 } )
] )
.then( cbq.job( "ImportCsvSuccessfulJob" ) );

onFailure

Defines a Job to be dispatched the first time a job in the Batch fails.

Arguments	Type	Required	Default	Description
job	string | Job	true		The Job WireBox id or Job instance to execute the first time a job in the Batch fails.
properties	Struct	false	{}	A struct of properties for the Job instance. (Only used when providing a Job WireBox id.)
chain	array<Job>	false	[]	An array of Jobs to chain after this Job. (Only used when providing a Job WireBox id.)
queue	string	false		The queue the Job belongs to. (Only used when providing a Job WireBox id.)
connection	string	false		The Connection to dispatch the Job on. (Only used when providing a Job WireBox id.)
backoff	numeric	false		The amount of time, in seconds, to wait between attempting Jobs. (Only used when providing a Job WireBox id.)
timeout	numeric	false		The amount of time, in seconds, to wait before treating a Job as erroring. (Only used when providing a Job WireBox id.)
maxAttempts	numeric	false		The maximum amount of attempts of a Job before treating the Job as failed. (Only used when providing a Job WireBox id.)

Return: The PendingBatch instance.

cbq.batch( [
    cbq.job( "ImportCsvJob", { "start": 1, "end": 100 } ),
    cbq.job( "ImportCsvJob", { "start": 101, "end": 200 } ),
    cbq.job( "ImportCsvJob", { "start": 201, "end": 300 } ),
    cbq.job( "ImportCsvJob", { "start": 301, "end": 400 } ),
    cbq.job( "ImportCsvJob", { "start": 401, "end": 500 } )
] )
.onFailure( cbq.job( "ImportCsvFailedJob" ) );

onComplete

Defines a Job to be dispatched after all the jobs in the Batch have executed successfully or failed.

Arguments	Type	Required	Default	Description
job	string | Job	true		The Job WireBox id or Job instance to execute after all the jobs in the Batch have executed successfully or failed.
properties	Struct	false	{}	A struct of properties for the Job instance. (Only used when providing a Job WireBox id.)
chain	array<Job>	false	[]	An array of Jobs to chain after this Job. (Only used when providing a Job WireBox id.)
queue	string	false		The queue the Job belongs to. (Only used when providing a Job WireBox id.)
connection	string	false		The Connection to dispatch the Job on. (Only used when providing a Job WireBox id.)
backoff	numeric	false		The amount of time, in seconds, to wait between attempting Jobs. (Only used when providing a Job WireBox id.)
timeout	numeric	false		The amount of time, in seconds, to wait before treating a Job as erroring. (Only used when providing a Job WireBox id.)
maxAttempts	numeric	false		The maximum amount of attempts of a Job before treating the Job as failed. (Only used when providing a Job WireBox id.)

Return: The PendingBatch instance.

cbq.batch( [
    cbq.job( "ImportCsvJob", { "start": 1, "end": 100 } ),
    cbq.job( "ImportCsvJob", { "start": 101, "end": 200 } ),
    cbq.job( "ImportCsvJob", { "start": 201, "end": 300 } ),
    cbq.job( "ImportCsvJob", { "start": 301, "end": 400 } ),
    cbq.job( "ImportCsvJob", { "start": 401, "end": 500 } )
] )
.onComplete( cbq.job( "ImportCsvCompletedJob" ) );

Dispatching Batches

A PendingBatch must be dispatched before any of the jobs contained in it are dispatched. This is done using the dispatch method on the PendingBatch.

dispatch

Dispatches a PendingBatch and all the Jobs it contains.

Arguments	Type	Required	Default	Description
No arguments

Return: A Batch instance created from this PendingBatch.

Result Accounting

cbq 6 requires migration 2000_01_01_000008_add_successfulJobs_to_cbq_batches_table.cfc for successfulJobs, and migration 2000_01_01_000009_make_cbq_batches_name_nullable.cfc makes batch names optional. Apply them to the configured batch table even when your jobs use SyncProvider or ColdBoxAsyncProvider. cbq 7 additionally requires 2000_01_01_000011_track_processed_batch_jobs.cfc for processedJobIds. See the Upgrade Guide.

totalJobs counts jobs added to the batch. Each successful result increments successfulJobs; each terminal failure increments failedJobs and adds its ID to failedJobIds. Both outcomes decrement pendingJobs. A retry is not a terminal result. The onComplete job can run when the final pending job fails, and an exception dispatching a then or onFailure job is logged so the completion path can continue.

Starting in v7, the repository uses a row lock and a processedJobIds JSON array to record one terminal result per job ID. Repeated success or failure callbacks do not consume another pending job or repeat lifecycle dispatch. Existing failed IDs are also protected. Successful IDs from before the migration cannot be reconstructed; drain active batches before upgrading if they might receive duplicate callbacks. This accounting does not make the job's external side effects execute exactly once.

Interacting with Batches

Reload a batch from its repository to read current counts; a previously returned Batch instance is a snapshot.

var currentBatch = getInstance( "DBBatchRepository@cbq" ).find( batch.getId() );
var counts = {
    total: currentBatch.getTotalJobs(),
    pending: currentBatch.getPendingJobs(),
    successful: currentBatch.getSuccessfulJobs(),
    failed: currentBatch.getFailedJobs()
};

Migrations and SQL Scripts

The examples below show the batch table after the v7 migrations. For a fresh installation, create this final structure or apply the supplied migrations in filename order. For an existing table, use the upgrade migrations instead of recreating it. Dates are Unix timestamps, matching the repository's stored values.

cfmigrations

schema.create( "cbq_batches", function ( t ) {
    t.string( "id" ).primaryKey();
    t.string( "name" ).nullable();
    t.unsignedInteger( "totalJobs" );
    t.unsignedInteger( "pendingJobs" );
    t.unsignedInteger( "successfulJobs" ).default( 0 );
    t.unsignedInteger( "failedJobs" );
    t.text( "failedJobIds" ); // JSON column
    t.longText( "processedJobIds" ).nullable(); // JSON column
    t.text( "options" ).nullable(); // JSON column
    t.unsignedInteger( "createdDate" );
    t.unsignedInteger( "cancelledDate" ).nullable();
    t.unsignedInteger( "completedDate" ).nullable();
} );

MySQL

CREATE TABLE `cbq_batches` (
    `id` VARCHAR(255) NOT NULL,
    `name` VARCHAR(255),
    `totalJobs` INTEGER UNSIGNED NOT NULL,
    `pendingJobs` INTEGER UNSIGNED NOT NULL,
    `successfulJobs` INTEGER UNSIGNED NOT NULL DEFAULT 0,
    `failedJobs` INTEGER UNSIGNED NOT NULL,
    `failedJobIds` TEXT NOT NULL,
    `processedJobIds` LONGTEXT,
    `options` TEXT,
    `createdDate` INTEGER UNSIGNED NOT NULL,
    `cancelledDate` INTEGER UNSIGNED,
    `completedDate` INTEGER UNSIGNED,
    CONSTRAINT `pk_cbq_batches_id` PRIMARY KEY (`id`)
)

SQL Server

CREATE TABLE [cbq_batches] (
    [id] VARCHAR(255) NOT NULL,
    [name] VARCHAR(255),
    [totalJobs] INTEGER NOT NULL,
    [pendingJobs] INTEGER NOT NULL,
    [successfulJobs] INTEGER NOT NULL DEFAULT 0,
    [failedJobs] INTEGER NOT NULL,
    [failedJobIds] VARCHAR(MAX) NOT NULL,
    [processedJobIds] VARCHAR(MAX),
    [options] VARCHAR(MAX),
    [createdDate] INTEGER NOT NULL,
    [cancelledDate] INTEGER,
    [completedDate] INTEGER,
    CONSTRAINT [pk_cbq_batches_id] PRIMARY KEY ([id])
)

Postgres

CREATE TABLE "cbq_batches" (
    "id" VARCHAR(255) NOT NULL,
    "name" VARCHAR(255),
    "totalJobs" INTEGER NOT NULL,
    "pendingJobs" INTEGER NOT NULL,
    "successfulJobs" INTEGER NOT NULL DEFAULT 0,
    "failedJobs" INTEGER NOT NULL,
    "failedJobIds" TEXT NOT NULL,
    "processedJobIds" TEXT,
    "options" TEXT,
    "createdDate" INTEGER NOT NULL,
    "cancelledDate" INTEGER,
    "completedDate" INTEGER,
    CONSTRAINT "pk_cbq_batches_id" PRIMARY KEY ("id")
)

Oracle

CREATE TABLE "CBQ_BATCHES" (
    "ID" VARCHAR2(255) NOT NULL,
    "NAME" VARCHAR2(255),
    "TOTALJOBS" NUMBER(10, 0) NOT NULL,
    "PENDINGJOBS" NUMBER(10, 0) NOT NULL,
    "SUCCESSFULJOBS" NUMBER(10, 0) DEFAULT 0 NOT NULL,
    "FAILEDJOBS" NUMBER(10, 0) NOT NULL,
    "FAILEDJOBIDS" CLOB NOT NULL,
    "PROCESSEDJOBIDS" CLOB,
    "OPTIONS" CLOB,
    "CREATEDDATE" NUMBER(10, 0) NOT NULL,
    "CANCELLEDDATE" NUMBER(10, 0),
    "COMPLETEDDATE" NUMBER(10, 0),
    CONSTRAINT "PK_CBQ_BATCHES_ID" PRIMARY KEY ("ID")
)

SQLite

CREATE TABLE "cbq_batches" (
    "id" TEXT NOT NULL,
    "name" TEXT,
    "totalJobs" INTEGER NOT NULL,
    "pendingJobs" INTEGER NOT NULL,
    "successfulJobs" INTEGER NOT NULL DEFAULT 0,
    "failedJobs" INTEGER NOT NULL,
    "failedJobIds" TEXT NOT NULL,
    "processedJobIds" TEXT,
    "options" TEXT,
    "createdDate" INTEGER NOT NULL,
    "cancelledDate" INTEGER,
    "completedDate" INTEGER,
    PRIMARY KEY ("id")
)