Queue Backends

গল্পে বুঝি

আল-খোয়ারিজমির কাঠের কারখানায় সারাদিন অর্ডার আসে — এই টেবিলটা বানাও, ওই দরজাটা মেরামত করো। প্রতিটা অর্ডার এক টুকরো কাগজে লেখা স্লিপ। সমস্যা একটাই: কাজ যতক্ষণ শুরু না হচ্ছে, ততক্ষণ এই জমতে থাকা স্লিপগুলো কোথায় রাখা হবে? আল-খোয়ারিজমি প্রথমে কাউন্টারের ওপর একটা লোহার শিক গেঁথে রাখল — নতুন স্লিপ এলেই টুক করে শিকে গেঁথে দাও, কারিগর এসে ওপরেরটা টেনে নিয়ে কাজে লেগে যায়। ঝামেলা নেই, বিদ্যুৎগতির কাজ। কিন্তু একদিন জানালা দিয়ে দমকা বাতাস এসে অর্ধেক স্লিপ উড়িয়ে নিয়ে গেল — কোন অর্ডার হারালো, কেউ আর বলতে পারে না।

তখন কারখানার হিসাবরক্ষক ইবনে সিনা বলল, “শিক দিয়ে হবে না, প্রতিটা অর্ডার বাঁধাই খাতায় কালি দিয়ে লিখি — দোকানের হিসাবের পাশেই।” এতে একটা অর্ডারও আর হারায় না, চাইলে ছয় মাস আগের অর্ডারও খুঁজে বের করা যায়। কিন্তু প্রতিবার খাতা খুলে, লাইন টেনে, কালিতে লেখা — শিকের চেয়ে ধীর, আর অর্ডার বেশি বেড়ে গেলে খাতা সামলানো কঠিন। শেষে যখন কাজের চাপ আকাশছোঁয়া হলো, ফাতিমা আল-ফিহরি আলাদা একটা ডিসপ্যাচ অফিস খুলে বসল — একদল কর্মী শুধু অর্ডার লগ করে, ঠিক কারিগরের কাছে রুট করে, একটা অর্ডারও কখনো হারায় না। নিখুঁত, কিন্তু আলাদা অফিস চালানোর খরচ আর লোকবল আছে।

গল্পের শিকটাই হলো Redis-backed queue (BullMQ) — দ্রুত আর সহজ, কিন্তু durability কম, মেমরি উড়ে গেলে জমে থাকা job হারাতে পারে। বাঁধাই খাতা হলো database-backed queue (pg-boss) — প্রতিটা job একটা row, transactional আর টেকসই, “record লেখা + job enqueue” একসাথে atomic করা যায়, বিনিময়ে একটু ধীর। আর ডিসপ্যাচ অফিস হলো dedicated broker — RabbitMQ বা Kafka-র মতো আলাদা সিস্টেম, বিপুল throughput আর শক্ত delivery guarantee দেয়, কিন্তু চালানোর operational complexity বেশি। তিনটার বাছাই আসলে একই তিন-অক্ষের হিসাব: durability বনাম speed বনাম operational complexity। বাস্তবে বেশিরভাগ টিম Redis বা database দিয়েই শুরু করে, আর স্কেল সত্যিই বিশাল হলে তবেই Kafka/RabbitMQ-র মতো broker-এ যায়।

BullMQ (Redis-backed)

BullMQ হলো সবচেয়ে জনপ্রিয় Node.js queue library। এটি job-এর state transition track করতে Redis sorted set ও list ব্যবহার করে।

BullMQ-তে job state:

waiting → active → completed
                 → failed → (retry) → waiting
                          → dead-letter

Setup:

import { Queue, Worker, QueueEvents } from 'bullmq';
import { Redis } from 'ioredis';

const connection = new Redis({ host: 'localhost', port: 6379, maxRetriesPerRequest: null });

// Producer (your API server)
const emailQueue = new Queue('emails', { connection });

await emailQueue.add(
	'send-welcome',
	{ userId: 'u_123', email: 'user@example.com' },
	{
		attempts: 3, // retry up to 3 times
		backoff: { type: 'exponential', delay: 2000 }, // 2s, 4s, 8s
		removeOnComplete: { count: 1000 }, // keep last 1000 completed
		removeOnFail: { count: 5000 } // keep last 5000 failed
	}
);

// Consumer (your worker process)
const worker = new Worker(
	'emails',
	async (job) => {
		const { userId, email } = job.data;
		await sendEmail(email, 'Welcome!');
		return { sent: true };
	},
	{
		connection,
		concurrency: 10 // 10 simultaneous jobs
	}
);

worker.on('completed', (job, result) => {
	console.log(`Job ${job.id} done:`, result);
});

worker.on('failed', (job, err) => {
	console.error(`Job ${job?.id} failed:`, err.message);
});

BullMQ যেসব Redis data structure ব্যবহার করে:

• bull:emails:wait — waiting job-দের sorted set (score = priority)
• bull:emails:active — এই মুহূর্তে প্রসেস হচ্ছে এমন job-দের set
• bull:emails:completed — completed job-দের sorted set
• bull:emails:failed — failed job-দের sorted set
• bull:emails:delayed — ভবিষ্যতের job-দের sorted set (score = run timestamp)

pg-boss (PostgreSQL-backed)

Redis-এর দরকার নেই। Job হলো একটি Postgres টেবিলের row। আপনি ACID transaction পান — “যে transaction record তৈরি করে সেই একই transaction-এর অংশ হিসেবে job enqueue করা”-র জন্য নিখুঁত:

import PgBoss from 'pg-boss';

const boss = new PgBoss(process.env.DATABASE_URL!);
await boss.start();

// Producer — can be inside a transaction
await db.transaction(async (trx) => {
	const user = await trx.users.create(userData);

	// Job is created atomically with the user — no chance of user-without-job
	await boss.sendOnce(
		'send-welcome-email',
		{ userId: user.id },
		{ retryLimit: 3, retryDelay: 30, expireInHours: 24 },
		user.id // deduplication key
	);
});

// Consumer
await boss.work('send-welcome-email', { teamSize: 5 }, async (job) => {
	await sendEmail(job.data.userId);
});

pg-boss যে schema তৈরি করে:

CREATE TABLE pgboss.job (
  id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  name        TEXT NOT NULL,
  data        JSONB,
  state       TEXT NOT NULL DEFAULT 'created', -- created/retry/active/completed/expired/cancelled/failed
  retryLimit  INT NOT NULL DEFAULT 0,
  retryCount  INT NOT NULL DEFAULT 0,
  retryDelay  INT NOT NULL DEFAULT 0,
  startAfter  TIMESTAMPTZ NOT NULL DEFAULT now(),
  startedOn   TIMESTAMPTZ,
  singletonKey TEXT, -- deduplication
  expireIn    INTERVAL NOT NULL DEFAULT '15 minutes',
  createdOn   TIMESTAMPTZ NOT NULL DEFAULT now(),
  completedOn TIMESTAMPTZ
);

Worker এই টেবিলটি SELECT ... FOR UPDATE SKIP LOCKED দিয়ে poll করে — এটি একটি Postgres প্যাটার্ন যা একাধিক worker-কে conflict ছাড়াই নিরাপদে job claim করতে দেয়:

-- What pg-boss does internally on each poll
UPDATE pgboss.job
SET state = 'active', startedOn = now()
WHERE id IN (
  SELECT id FROM pgboss.job
  WHERE name = 'send-welcome-email'
    AND state = 'created'
    AND startAfter <= now()
  ORDER BY createdOn
  LIMIT 5
  FOR UPDATE SKIP LOCKED -- skip rows locked by other workers
)
RETURNING *;

FOR UPDATE SKIP LOCKED হলো মূল ব্যাপার — একাধিক worker একসাথে poll করতে পারে একে অপরকে block না করে বা একই job claim না করে।

দুটোর মধ্যে বাছাই

BullMQ (Redis) ব্যবহার করুন যখন:

• real-time job pickup দরকার (sub-second)
• High throughput (সেকেন্ডে হাজার হাজার job)
• আপনার built-in job progress tracking, প্রতি queue-তে rate limiting, বা priority queue দরকার
• আপনি আগে থেকেই Redis চালাচ্ছেন

pg-boss (Postgres) ব্যবহার করুন যখন:

• আপনি একটি DB write-এর সাথে atomically enqueue করতে চান (enqueue fail করলে job হারানোর ঝুঁকি নেই)
• আপনি Redis চালাতে চান না
• আপনার job history-র পূর্ণ auditability দরকার
• আপনার throughput মাঝারি (<100 job/সেকেন্ড)
• আপনি সহজতর ops চান (একটা কম infra component)

Transactional enqueueing-এর সুবিধা:

// BullMQ — NOT transactional
await db.users.create(user);
// If this crashes, user exists but no welcome email is ever sent
await emailQueue.add('send-welcome', { userId: user.id });

// pg-boss — transactional
await db.transaction(async (trx) => {
	await trx.users.create(user);
	await boss.send('send-welcome', { userId: user.id }); // same transaction
	// If either fails, both are rolled back — consistent state
});

যেসব অপারেশনে “record লেখা + job enqueue করা” atomic হতেই হবে, তাদের জন্য এটি একটি বড় সুবিধা।

Delayed ও Scheduled Job

Delayed (একবার চলবে, ভবিষ্যতে):

// BullMQ
await queue.add(
	'send-trial-expiry-email',
	{ userId },
	{
		delay: 14 * 24 * 60 * 60 * 1000 // 14 days from now
	}
);

// pg-boss
await boss.send(
	'send-trial-expiry-email',
	{ userId },
	{
		startAfter: new Date(Date.now() + 14 * 24 * 60 * 60 * 1000)
	}
);

Recurring (cron-এর মতো):

// BullMQ — repeatable jobs
await queue.add(
	'cleanup-expired-sessions',
	{},
	{
		repeat: { cron: '0 3 * * *' } // 3am every day
	}
);

// pg-boss — schedules
await boss.schedule('cleanup-expired-sessions', '0 3 * * *', {});
await boss.work('cleanup-expired-sessions', async () => {
	await db.sessions.deleteExpired();
});

Queue Health মনিটর করা

track করার মূল metric:

// BullMQ counts
const [waiting, active, completed, failed] = await Promise.all([
	queue.getWaitingCount(),
	queue.getActiveCount(),
	queue.getCompletedCount(),
	queue.getFailedCount()
]);

console.log({ waiting, active, completed, failed });

Alert threshold:

• waiting > 1000: queue জমছে, worker যোগ করুন
• failed > 0 and growing: job type-এ bug আছে অথবা dependency down
• active == workerCount and waiting > 0: worker capacity-তে পৌঁছেছে, scale out করুন
• oldest waiting job > 5 minutes: job pickup SLA ভেঙে গেছে

Bull Board — BullMQ-র জন্য visual UI:

import { createBullBoard } from '@bull-board/api';
import { BullMQAdapter } from '@bull-board/api/bullMQAdapter';
import { ExpressAdapter } from '@bull-board/express';

const serverAdapter = new ExpressAdapter();
serverAdapter.setBasePath('/admin/queues');

createBullBoard({
	queues: [new BullMQAdapter(emailQueue), new BullMQAdapter(pdfQueue)],
	serverAdapter
});

app.use('/admin/queues', serverAdapter.getRouter());

auth-এর পেছনে mount করুন — এটি job payload দেখায় যাতে sensitive data থাকতে পারে।