Background job ও queue: retry, backoff ও dead job কোডসহ

কিছু কাজ web request-এর ভেতরে হওয়া উচিত নয়: email পাঠানো, ছবি ছোট করা, রিপোর্ট বানানো, ধীর তৃতীয় পক্ষের API ডাকা। Inline করলে ব্যবহারকারী অপেক্ষা করেন, timeout হলে কাজ হারায়, আর একটি ব্যর্থতায় পুরো পাতা ভেঙে যায়। প্রচলিত সমাধান হলো background job queue: request একটি job লিখে রেখে দ্রুত সাড়া দেয়, আর আলাদা worker পরে কাজটি করে। নিচের কোড ও ফল প্রকাশের আগে Python 3-এ চালানো হয়েছে।

অংশগুলো

• Producer: আপনার web app, যা job যোগ করে (enqueue)।
• Queue: job-এর টেকসই সংরক্ষণ। এটি database table, Redis, বা RabbitMQ বা ক্লাউড queue-র মতো broker হতে পারে।
• Worker: যে process job নেয়, চালায় ও ফল লেখে।
• Retry: job ব্যর্থ হলে পরে আবার চেষ্টা, কিন্তু সঙ্গে সঙ্গে নয়, আর চিরকালও নয়।
• Dead job: শেষ চেষ্টার পর job মানুষের দেখার জন্য আলাদা করে রাখা।

একটি চালানো যায় এমন উদাহরণ

Queue-কে অদ্ভুত হতে হবে না। এটি একটি SQLite table। jobs.py নামে সংরক্ষণ করুন:

import json
import sqlite3

SCHEMA = """
CREATE TABLE IF NOT EXISTS jobs (
    id INTEGER PRIMARY KEY,
    kind TEXT NOT NULL,
    payload TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'queued',   -- queued | running | done | dead
    attempts INTEGER NOT NULL DEFAULT 0,
    max_attempts INTEGER NOT NULL DEFAULT 3,
    run_at REAL NOT NULL,
    last_error TEXT
);
CREATE INDEX IF NOT EXISTS jobs_ready ON jobs (status, run_at);
"""


def connect(path=":memory:"):
    db = sqlite3.connect(path, isolation_level=None)  # we manage transactions
    db.row_factory = sqlite3.Row
    db.executescript(SCHEMA)
    return db


def enqueue(db, kind, payload, now, max_attempts=3):
    cur = db.execute(
        "INSERT INTO jobs (kind, payload, run_at, max_attempts) VALUES (?, ?, ?, ?)",
        (kind, json.dumps(payload), now, max_attempts),
    )
    return cur.lastrowid


def claim(db, now):
    """Atomically take one ready job. Two workers can never get the same row."""
    db.execute("BEGIN IMMEDIATE")
    row = db.execute(
        "SELECT * FROM jobs WHERE status = 'queued' AND run_at <= ? ORDER BY run_at, id LIMIT 1",
        (now,),
    ).fetchone()
    if row:
        db.execute(
            "UPDATE jobs SET status = 'running', attempts = attempts + 1 WHERE id = ?",
            (row["id"],),
        )
    db.execute("COMMIT")
    return row


def run_one(db, handlers, now):
    job = claim(db, now)
    if job is None:
        return None
    attempt = job["attempts"] + 1
    try:
        handlers[job["kind"]](json.loads(job["payload"]))
    except Exception as exc:  # a job must never crash the worker
        if attempt >= job["max_attempts"]:
            status, delay = "dead", 0
        else:
            status, delay = "queued", 2 ** attempt  # 2s, 4s, 8s ... backoff
        db.execute(
            "UPDATE jobs SET status = ?, run_at = ?, last_error = ? WHERE id = ?",
            (status, now + delay, repr(exc), job["id"]),
        )
        return job["id"], status
    db.execute("UPDATE jobs SET status = 'done' WHERE id = ?", (job["id"],))
    return job["id"], "done"


if __name__ == "__main__":
    db = connect()
    calls = {"flaky": 0}

    def send_email(p):
        print(f"    sent email to {p['to']}")

    def flaky(p):
        calls["flaky"] += 1
        if calls["flaky"] < 3:
            raise ConnectionError("provider timeout")
        print("    flaky job finally worked")

    def broken(p):
        raise ValueError("bad data")

    handlers = {"email": send_email, "flaky": flaky, "broken": broken}
    enqueue(db, "email", {"to": "a@example.com"}, now=0)
    enqueue(db, "flaky", {}, now=0)
    enqueue(db, "broken", {}, now=0)

    for t in range(0, 12):
        while (r := run_one(db, handlers, now=t)) is not None:
            print(f"t={t:>2}s job {r[0]} -> {r[1]}")

    print()
    for r in db.execute("SELECT id, kind, status, attempts, last_error FROM jobs"):
        print(dict(r))

python3 jobs.py-এর ফল (সময় সেকেন্ডে অনুকরণ করা, তাই মুহূর্তে চলে):

    sent email to a@example.com
t= 0s job 1 -> done
t= 0s job 2 -> queued
t= 0s job 3 -> queued
t= 2s job 2 -> queued
t= 2s job 3 -> queued
    flaky job finally worked
t= 6s job 2 -> done
t= 6s job 3 -> dead

{'id': 1, 'kind': 'email', 'status': 'done', 'attempts': 1, 'last_error': None}
{'id': 2, 'kind': 'flaky', 'status': 'done', 'attempts': 3, 'last_error': "ConnectionError('provider timeout')"}
{'id': 3, 'kind': 'broken', 'status': 'dead', 'attempts': 3, 'last_error': "ValueError('bad data')"}

ফল পড়া

• Job 1 (email) প্রথমবারেই সফল।
• Job 2 (flaky) timeout দিয়ে দুবার ব্যর্থ, তারপর তৃতীয় চেষ্টায় সফল। সাময়িক সমস্যা নিজে সেরে গেছে।
• Job 3 (broken) প্রতিবারই ব্যর্থ। ৩ চেষ্টার পর এটি dead হয়ে last_error-সহ table-এ থেকে যায়। বাগ চিরকাল retry করে লাভ নেই।
• চেষ্টার মাঝে job অপেক্ষা করে: ২ সেকেন্ড, তারপর ৪। এটিই exponential backoff: প্রতিবার অপেক্ষা বাড়ে, ফলে সমস্যায় থাকা সেবা ঘা খেতে না থেকে সেরে ওঠার সুযোগ পায়।

যে খুঁটিনাটি গুরুত্বপূর্ণ

Atomic ভাবে claim করুন। claim একটি BEGIN IMMEDIATE transaction-এর ভেতরে job বেছে running চিহ্নিত করে, তাই দুই worker একই row নিতে পারে না। test_claim_is_exclusive তা যাচাই করে। PostgreSQL-এ সাধারণ উপায় SELECT ... FOR UPDATE SKIP LOCKED, যাতে বহু worker একে অপরকে আটকে না রেখে ভিন্ন row নেয়।

যা সফল হতে পারে তাই retry করুন। Timeout কেটে যেতে পারে; খারাপ data থেকে আসা ValueError যাবে না। আসল সিস্টেমে প্রায়ই সাময়িক error (retry) ও স্থায়ী error (সরাসরি dead) আলাদা করা হয়।

Jitter যোগ করুন। ১,০০০ job একসঙ্গে ব্যর্থ হয়ে ঠিক ৪ সেকেন্ড পরে সবাই retry করলে আবার একসঙ্গে ব্যর্থ হবে। প্রতিটি অপেক্ষায় সামান্য এলোমেলো সময় যোগ করলে ছড়িয়ে যায়। ফল অনুমানযোগ্য রাখতে উদাহরণে তা বাদ দেওয়া হয়েছে।

Job-কে idempotent বানান। Queue সিস্টেমে delivery সাধারণত অন্তত একবার (at least once): worker কাজ শেষ করে "done" লেখার আগে crash করলে job আবার চলে। Handler এমন লিখুন যাতে দুবার চালানো নিরাপদ, যেমন idempotency key বা "এই order-এর email আগেই গেছে" পরীক্ষা দিয়ে।

Payload ছোট রাখুন। পুরো order নয়, id (order_id) রাখুন। Worker নতুন data লোড করে, আর queue-তে পুরোনো বা সংবেদনশীল কপি থাকে না।

Test

test_jobs.py নামে সংরক্ষণ করুন:

import unittest

from jobs import claim, connect, enqueue, run_one


class JobTests(unittest.TestCase):
    def setUp(self):
        self.db = connect()

    def status(self, job_id):
        return self.db.execute("SELECT status, attempts FROM jobs WHERE id=?", (job_id,)).fetchone()

    def test_success(self):
        j = enqueue(self.db, "ok", {}, now=0)
        run_one(self.db, {"ok": lambda p: None}, now=0)
        self.assertEqual(tuple(self.status(j)), ("done", 1))

    def test_retry_then_dead(self):
        def fail(p):
            raise RuntimeError("x")
        j = enqueue(self.db, "f", {}, now=0, max_attempts=2)
        run_one(self.db, {"f": fail}, now=0)
        self.assertEqual(self.status(j)["status"], "queued")
        self.assertIsNone(run_one(self.db, {"f": fail}, now=1))  # backoff: not ready yet
        run_one(self.db, {"f": fail}, now=2)
        self.assertEqual(tuple(self.status(j)), ("dead", 2))

    def test_claim_is_exclusive(self):
        enqueue(self.db, "ok", {}, now=0)
        self.assertIsNotNone(claim(self.db, now=0))
        self.assertIsNone(claim(self.db, now=0))  # already running

    def test_future_job_waits(self):
        enqueue(self.db, "ok", {}, now=100)
        self.assertIsNone(claim(self.db, now=50))
        self.assertIsNotNone(claim(self.db, now=100))


if __name__ == "__main__":
    unittest.main()

$ python3 -m unittest -v
Ran 4 tests in 0.009s

OK

এই উদাহরণ যা সামলায় না

কপি করার আগে ফাঁকগুলো জানুন:

• Crash হওয়া worker। Job running থাকা অবস্থায় worker মরলে সেই row চিরকাল running থাকে। আসল queue visibility timeout বা heartbeat ব্যবহার করে: চলমান job সময়মতো শেষ না হলে আবার পাওয়া যায়।
• Concurrency। SQLite-এ একবারে একজন writer। শেখা ও ছোট অ্যাপে ঠিক আছে, ভারী সমান্তরাল চাপে নয়।
• Schedule, অগ্রাধিকার, rate limit, monitoring। পরিণত লাইব্রেরি বা সেবা এগুলো দেয়।

নিজে বানাবেন, না লাইব্রেরি?

প্রোডাকশনে সাধারণত নিজে না লিখে যাচাই করা টুল ব্যবহার করুন: Celery বা RQ (Python), BullMQ (Node.js), Sidekiq (Ruby), বা managed ক্লাউড queue। উপরের মতো ছোট queue বানানো এগুলো বোঝার জন্য চমৎকার। যা আগে থেকে চালান তার ওপর ভিত্তি করে বাছুন: Redis থাকলে Redis-ভিত্তিক queue স্বাভাবিক; শুধু PostgreSQL থাকলে মাঝারি মাপে Postgres-ভিত্তিক queue প্রায়ই যথেষ্ট।

সাধারণ ভুল

• Request-এর ভেতরে ধীর কাজ করা।
• Backoff ছাড়া সঙ্গে সঙ্গে retry।
• Dead অবস্থা বা alert ছাড়া চিরকাল retry।
• At-least-once delivery-তে non-idempotent handler।
• Queue-র দৃশ্যমানতা না থাকা: ৫,০০০ job জমলে বা ৫০টি dead হলে কেউ টের পায় না। Queue-র দৈর্ঘ্য, সবচেয়ে পুরোনো job-এর বয়স ও ব্যর্থতার সংখ্যা ট্র্যাক করুন।
• Payload-এ বড় বা গোপন data রাখা।

মূল কথা: request গ্রহণ করুন, টেকসই job লিখুন, দ্রুত সাড়া দিন। Worker সীমিত retry, বাড়তি বিরতি ও dead অবস্থা সহ কাজ চালাক, আর প্রতিটি job এমনভাবে লিখুন যাতে দুবার চালালেও ক্ষতি না হয়।

---

Backend সিরিজ, পর্ব ৩ (নির্ভরযোগ্য API): 3 নম্বর (মোট 5)। আগের লেখা: Webhook সঠিকভাবে: HMAC signature যাচাই ও replay ঠেকান.