jobs and queues

introduction

dframework includes a native background job queue that runs jobs in isolated worker threads. jobs are dispatched from your application code and executed asynchronously without blocking the main http process. this is ideal for sending emails, processing images, syncing external apis, or any task that should not delay the user's response.

creating jobs

the job class

a job is a class stored in the jobs directory at the root of your project. every job must export a default class with a handle(payload) method. the handle method receives the payload that was passed when the job was dispatched and contains the actual work to be performed.

1export default class SendWelcomeEmailJob {
2  async handle(payload) {
3    const user = await User.find(payload.userId);
4    await Mailer.send(user.email, 'welcome', { name: user.name });
5  }
6}

the handle method can be asynchronous. the worker waits for the returned promise to resolve before reporting completion.

generating jobs

you can scaffold a new job file using the cli.

1dstrn make:job SendWelcomeEmailJob

this creates a jobs/SendWelcomeEmailJob.js file with the correct boilerplate structure.

dispatching jobs

from controllers

the most common way to dispatch a job is from a controller after handling the user's request.

1export default class RegistrationController {
2  async register(req, res) {
3    const user = await User.create(req.body);
4    Job.dispatch('SendWelcomeEmailJob', { userId: user.id });
5    return redirect('/dashboard');
6  }
7}

Job.dispatch() is non blocking. it immediately returns control to the controller while the job executes in a background worker thread.

from anywhere

the Job facade is available globally. you can dispatch jobs from controllers, middleware, socket event handlers, other jobs, scheduled commands, or import it directly from the framework root:

1import { Job } from 'dframework';
2 
3Job.dispatch('SyncInventoryJob', { warehouseId: 3 });

or simply use the global Job facade that is automatically available in all runtime contexts:

1Job.dispatch('GenerateReportJob', { month: 'june', year: 2026 });

from the command line

you can also dispatch a job from the command line using the dstrn dispatch command. this is useful for testing jobs during development or running one off tasks manually.

1dstrn dispatch SendWelcomeEmailJob --payload '{"userId": 1}'

the --payload flag accepts a json string that is parsed and passed to the job's handle method. if omitted, the job receives an empty object. the command bootstraps the full application context including database connectivity and all framework globals, exactly as a worker thread does. execution time is reported on completion.

1dstrn dispatch CleanupExpiredSessionsJob

payload data

the second argument to Job.dispatch() is a plain javascript object that will be serialized to json and passed to the job's handle method. keep payloads small and serializable. pass identifiers rather than full objects.

1Job.dispatch('ProcessImageJob', {
2  imageId: upload.id,
3  operations: ['resize', 'compress'],
4  quality: 85
5});

the job facade

the Job global is a proxy facade that delegates to the application's Queue instance. it is automatically registered when the application boots and is available in the global scope.

1Job.dispatch('JobName', { key: 'value' });

Job.dispatch() accepts an optional third argument for per job execution options:

1Job.dispatch('ProcessVideoJob', { videoId: 42 }, {
2  timeout: 30000,
3  tries: 3,
4  backoff: 1000
5});

if you attempt to use the Job facade before the application has initialized, it will throw a descriptive error.

how the worker pool works

worker isolation

each dispatched job runs in its own dedicated Worker thread using node worker_threads. the worker thread initializes a minimal application instance that includes database connectivity, configuration loading, and all framework globals without starting an http server. this means your job code can use the exact same apis as your controllers including the DB facade, Config, model classes, and all helper functions.

concurrency limits

the queue enforces a maximum number of concurrent workers. by default, up to four jobs can run simultaneously. if all worker slots are occupied when a new job is dispatched, the job is placed in an in memory queue and will be executed as soon as a worker becomes available.

the maximum worker count is configurable through the queue.maxWorkers configuration key.

automatic queuing and persistent worker pool

when the worker pool is at capacity, additional dispatched jobs are automatically buffered in an internal queue. workers in dframework operate within a persistent worker thread pool:

1. on demand execution: worker threads are spawned lazily when jobs are dispatched via Job.dispatch(). applications that never dispatch background jobs incur zero worker overhead.
2. persistent execution: when jobs exist, worker threads boot the application environment and database pool once and stay alive in the idle pool. subsequent jobs are picked up immediately by warm workers without thread creation or isolate boot delay.
3. automatic dequeuing: as each worker finishes its job, it picks up the next buffered job from the internal queue, or returns to the idle pool ready for future dispatches.
4. fault recovery: if an unhandled fatal error or process crash occurs inside a worker thread, the worker cleans up and exits, and a replacement worker is spawned to continue processing pending tasks.

job lifecycle

bootstrapping

when a persistent worker thread boots, it performs the following steps:

1. loads environment variables and configuration files from the project
2. initializes an application instance with database connection and global facades
3. signals readiness to the main process and waits for dispatched jobs
4. upon receiving a job, imports the job class from the jobs directory
5. instantiates the job class and validates the handle method
6. invokes handle(payload) with the serialized payload
7. reports success or failure with execution duration back to the main thread
8. returns to the idle worker pool to immediately accept subsequent jobs

database access

jobs have full access to the database through the same DB facade and model classes available in controllers. each worker thread establishes its database connection pool during initialization. the pool stays open across job executions for the worker lifetime and tears down cleanly when the worker exits or the queue drains.

1export default class CleanupExpiredSessionsJob {
2  async handle(payload) {
3    const cutoff = new Date(Date.now() - 7 * 24 * 60 * 60 * 1000);
4    await DB.table('sessions').where('updated_at', '<', cutoff).delete();
5  }
6}

broadcasting from jobs

jobs can dispatch real time websocket events and update component state on connected clients using the global Socket facade. because jobs run inside dedicated worker threads, Socket.broadcast() and Socket.setState() automatically relay messages over the internal worker channel to the main process, which broadcasts them to connected browsers.

1export default class GenerateReportJob {
2  async handle(payload) {
3    Socket.broadcast('report:progress', { percent: 25 });
4 
5    const report = await this.generate(payload.reportId);
6    Socket.broadcast('report:progress', { percent: 100 });
7 
8    Socket.broadcast('report:ready', {
9      reportId: report.id,
10      downloadUrl: `/reports/${report.id}/download`
11    });
12 
13    Socket.setState('#report-banner', {
14      status: 'complete',
15      reportId: report.id
16    });
17  }
18}

no external message brokers (such as redis) are required. the worker thread communicates directly with the primary server instance.

error handling

if a job's handle method throws an error, a structured error message containing the error name, message, stack trace, and execution duration is sent back to the main thread and logged by the application logger. the worker thread runs its cleanup function to close database connections before exiting.

1[JOB] [SendWelcomeEmailJob] failed after 1847ms:
2Error: SMTP connection refused
3    at Mailer.send (mailer/Mailer.js:42)
4    at SendWelcomeEmailJob.handle (jobs/SendWelcomeEmailJob.js:12)

if the worker thread itself crashes (for example due to an out of memory condition), the main thread receives an error event and logs the crash. the worker slot is freed and the next queued job is processed normally. database connections are automatically released when the worker process terminates.

configuration

the queue system reads its configuration from the app.queue config namespace.

1// config/app.js
2export default {
3  queue: {
4    maxWorkers: Env.value('QUEUE_MAX_WORKERS', 4),
5    timeout: Env.value('QUEUE_TIMEOUT', 60000),
6    shutdownTimeout: Env.value('QUEUE_SHUTDOWN_TIMEOUT', 30000),
7  },
8};