@rasenganjs/queue

Background job queues for Rasengan Server: declare a queue like a controller, inject it anywhere DI reaches, and process jobs in the background with retries, backoff, and dead-lettering.

Installation

pnpm add @rasenganjs/queue

@rasenganjs/server is a peer dependency — install it if your project doesn't already have it:

pnpm add @rasenganjs/server

1. Register the plugin

createQueuePlugin() claims the queues key on defineModule(). Register it once, in bootstrap(), before registering your app module:

// main.ts
import { bootstrap } from '@rasenganjs/server';
import { createQueuePlugin } from '@rasenganjs/queue';
import appModule from './app.module.js';

bootstrap((app) => {
  app.registerPlugin(createQueuePlugin());
  app.registerModule(appModule);
});

2. Define a queue

A Queue subclass declares its name and its job handlers. Constructor injection works exactly like a Controller:

// email.queue.ts
import { Queue, JobRouter, type JobHandler } from '@rasenganjs/queue';
import { MailerService } from './mailer.service.js';

export class EmailQueue extends Queue {
  name = 'emails';

  constructor(private mailer: MailerService) {
    super();
  }

  jobs(router: JobRouter) {
    router.process('welcome', this.sendWelcome, {
      attempts: 3, // total tries before dead-letter
      backoff: 5_000, // base delay in ms — doubles per retry
      concurrency: 5, // parallel jobs of this name, per worker
    });
  }

  sendWelcome: JobHandler<{ userId: string }> = async (job) => {
    await this.mailer.sendWelcome(job.data.userId);
    // Resolve → job completes. Throw → retried per the options above,
    // then dead-lettered once attempts are exhausted.
  };
}

3. Register the queue in your module

// email.module.ts
import { defineModule } from '@rasenganjs/server';
import { EmailQueue } from './email.queue.js';
import { MailerService } from './mailer.service.js';

export default defineModule({
  queues: [EmailQueue],
  providers: [MailerService, EmailQueue],
  exports: [EmailQueue], // so other modules can inject it too
});

4. Enqueue jobs

Inject the queue anywhere the DI container reaches — a controller, a gateway, another provider — and call .add():

// signup.controller.ts
import { Controller, type RouteHandler, type Router } from '@rasenganjs/server';
import { EmailQueue } from './email.queue.js';

export class SignupController extends Controller {
  constructor(private emailQueue: EmailQueue) {
    super();
  }

  routes(router: Router) {
    router.post('/signup', this.register);
  }

  register: RouteHandler = async (ctx) => {
    const user = await createUser(ctx.body);
    await this.emailQueue.add('welcome', { userId: user.id });
    return ctx.res.json({ ok: true });
  };
}

5. Delayed and repeatable jobs

Pass a third argument to .add():

// Runs 24 hours from now instead of immediately.
await this.emailQueue.add(
  'followUp',
  { userId: user.id },
  { delay: 86_400_000 }
);

// Registered once at startup — safe to call every time your app boots.
await this.statsQueue.add('digest', {}, { repeat: { every: 3_600_000 } });

Repeat registration is idempotent by jobKey: calling .add() again with the same job name and data (or the same explicit key) does not create a second recurring schedule — so it's safe to call at every boot rather than needing separate first-run logic. jobKey defaults to a value derived from the job name and data; supply your own if two repeat jobs would otherwise share both and need to stay distinct (e.g. a per-tenant digest):

await this.statsQueue.add(
  'digest',
  { tenantId },
  { repeat: { every: 3_600_000, key: `digest:${tenantId}` } }
);

For a { repeat } registration, .add() resolves with the jobKey (not a random id) — that's the identity you'd use to reason about or remove that schedule going forward. delay and repeat can't be combined in the same call.

Plugin options

createQueuePlugin({
  adapter?: QueueAdapter;  // default: MemoryQueueAdapter (dev only)
  worker?: boolean;        // default: true
  stallTimeout?: number;   // default: 30_000
  sweepInterval?: number;  // default: 5_000
});

• stallTimeout — how long a reserved job may go unacknowledged (no complete()/fail()) before it's presumed abandoned by a dead worker and returned to the queue, with its attempt count incremented.

• sweepInterval — how often the plugin checks for delayed/repeat jobs that have become due and reservations that have stalled, across every queue it registers.

• adapter — job storage. Defaults to MemoryQueueAdapter, which is in-process and loses all jobs on restart — fine for local development, not for production. Pass a RedisQueueAdapter (below) in production.

• worker — whether this process consumes jobs. Set to false for a produce-only process (e.g. your web servers enqueue jobs; a separate deployment with worker: true and the same adapter actually processes them):

  // Web process — enqueues only.
  app.registerPlugin(createQueuePlugin({ adapter, worker: false }));
  
  // Worker process — same adapter, actually processes jobs.
  app.registerPlugin(createQueuePlugin({ adapter, worker: true }));

Production storage: RedisQueueAdapter

MemoryQueueAdapter is dev-only. For production, use RedisQueueAdapter — persists across restarts and is safe with multiple processes sharing one queue (e.g. several worker instances, or a produce-only web tier alongside a separate worker deployment).

pnpm add ioredis

import { createQueuePlugin, RedisQueueAdapter } from '@rasenganjs/queue';
import Redis from 'ioredis';

const client = new Redis(process.env.REDIS_URL);

const adapter = new RedisQueueAdapter({
  client,
  // reserve()'s BLMOVE needs a connection of its own — Redis puts a
  // connection into a different mode while blocking, the same reason
  // @rasenganjs/ws's Redis adapter needs a separate SUBSCRIBE connection.
  blockingClient: client.duplicate(),
});

app.registerPlugin(createQueuePlugin({ adapter }));

ioredis isn't required as a hard dependency — RedisQueueAdapter takes a small structural interface (eval, blmove, lrange, hmget), so any client implementing those commands works, including Bun's built-in Bun.redis.

interface RedisQueueAdapterOptions {
  client: RedisLike;
  blockingClient: RedisLike;
  blockTimeoutSeconds?: number; // default: 0.02
  sweepBatchSize?: number; // default: 1000 — caps work done per sweep tick
  keyPrefix?: string; // default: 'queue:'
}

router.process() options

router.process(jobName, handler, {
  attempts?: number;   // default: 1 (no retry)
  backoff?: number;    // default: 0 (ms — doubles per retry attempt)
  concurrency?: number; // default: 1 (in-flight calls per job name)
});

Inspecting and retrying failed jobs

const dead = await this.emailQueue.getDead();
await this.emailQueue.retryDead(dead[0].id);

Current limitations

@rasenganjs/queue has shipped Phases 1 (Core), 2 (Time), and 3 (Redis) of its RFC:

• A handler that consistently outlives stallTimeout can be reclaimed and reprocessed indefinitely — stalled-job reclaim doesn't consult attempts/dead-letter (matches the underlying job lifecycle model; see ARCHITECTURE.md for why).
• RedisQueueAdapter's live-Redis behavior is untested against a real server today (tests run against a faked client only — same bar @rasenganjs/ws's Redis adapter shipped at).
• reserve()'s BLMOVE is capped to a short timeout to fit @rasenganjs/queue's existing fixed-interval poll loop, so it doesn't yet deliver BLMOVE's usual near-zero-latency wakeup — throughput is still bounded by that poll interval.

See proposals/RFC-0004-Background-Job-Queues.md in the monorepo for the full roadmap, and ARCHITECTURE.md for how the package works internally.

License

MIT