Background job, scheduler আর workflow engine — দীর্ঘ কাজ নির্ভরযোগ্যভাবে চালানো

১ · কেন request path-এ নয় — background job-এর প্রয়োজন

একটি কাল্পনিক অনলাইন দোকানে বিক্রেতা পণ্যের একটি ছবি আপলোড করলেন। সার্ভারকে এখন অনেক কাজ করতে হবে: ছবিটি object storage-এ রাখা, তিন মাপের ছোট সংস্করণ (thumbnail) বানানো, ছবির ধরন যাচাই, search index হালনাগাদ, আর বিক্রেতাকে একটি ইমেইল। সব কাজ যদি অনুরোধের ভেতরেই হয়, বিক্রেতা ঘুরতে থাকা চাকা দেখবেন হয়তো ১০ সেকেন্ড। আর ইমেইল সেবা সেই মুহূর্তে ধীর থাকলে তাঁর আপলোডটিই “ব্যর্থ” দেখাবে — যদিও ছবিটি ঠিকমতো রাখা হয়েছিল।

সমাধান সহজ চিন্তা থেকে আসে: ব্যবহারকারীর এখনই যা জানা দরকার, শুধু সেটুকু অনুরোধের ভেতরে করুন; বাকিটা পরে। অনুরোধটি ছবি রাখে, একটি background job সারিতে রাখে (“photo 7731-এর thumbnail বানাও”), আর সঙ্গে সঙ্গে উত্তর দেয় — HTTP-তে প্রায়ই 202 Accepted, সাথে job-এর id, যা দিয়ে পরে অবস্থা জানা যায় (পর্ব ৮ — REST API নকশা)। আলাদা worker process সারি থেকে কাজ তুলে নিজের গতিতে করে।

কোন কাজ সরাবেন — চারটি লক্ষণ

• ধীর: ছবি বা ভিডিও প্রক্রিয়া, বড় PDF, মাসের রিপোর্ট (ভিডিওর পুরো নকশা পর্ব ৫৯-এ)।
• বাইরের সেবার ওপর নির্ভর: ইমেইল, SMS, push — এদের বিভ্রাট আপনার নিয়ন্ত্রণে নেই (notification-এর নকশা পর্ব ৫৭-এ)।
• এখনই ফল লাগে না: ইমেইল ৫ সেকেন্ড পরে গেলে কেউ টের পান না।
• ভিড়ে জমা রাখা চলে: সারিতে জমুক, worker-রা নিজেদের সামর্থ্যে শেষ করুক (পর্ব ১৮ — Overload সামলানো)।

লাভ তিনটি: latency কমে, কারণ ব্যবহারকারী শুধু দ্রুত অংশের অপেক্ষা করেন; ব্যর্থতা আলাদা থাকে, কারণ ইমেইল সেবার বিভ্রাট এখন একটি সারির দৈর্ঘ্য, ব্যবহারকারীর ত্রুটি নয়; আর ভিড় শোষণ হয়, কারণ worker-এর সংখ্যা চাপ অনুযায়ী আলাদা করে বাড়ানো-কমানো যায়। দাম: কাজটি এখন asynchronous, তাই ব্যবহারকারীকে পরে ফল জানাতে হয় (polling, ইমেইল বা পর্ব ১০ — Real-time যোগাযোগ-এর WebSocket), আর “ব্যর্থ হলে কী হবে” প্রশ্নটি এখন আপনার নকশার দায়।

আরেকটি সূক্ষ্ম সমস্যা: অর্ডার commit হলো, job রাখার আগেই সার্ভার মরল — ইমেইল আর যাবে না। প্রচলিত সমাধান outbox pattern: job-কে অর্ডারের সাথে একই transaction-এ লেখা, পরে একটি relay সারিতে পাঠায় (পর্ব ২৫ — Distributed transaction).

২ · Job queue আর worker — একটি job-এর জীবনচক্র

সারি কীভাবে বার্তা রাখে, ack পেলে মোছে আর ack না পেলে আবার দেয় — সেই যন্ত্রের পূর্ণ বিবরণ পর্ব ৩১ — Message queue-এ। এখানে প্রশ্ন একটু ওপরের স্তরের: সেই সারির ওপর কাজ চালানোর নিয়ম কী? একটি job queue সাধারণ message queue-এর ওপর কয়েকটি ধারণা যোগ করে: প্রতিটি job-এর একটি অবস্থা, কতবার চেষ্টা হলো তার গণনা, কখন আবার চালানো যাবে (run_at), আর ফল বা শেষ ত্রুটি।

{
  "id": "job_7731_thumb",
  "type": "make_thumbnails",
  "payload": { "photo_id": 7731, "sizes": [128, 512, 1024] },
  "state": "queued",
  "attempt": 0,
  "max_attempts": 4,
  "run_at": "2026-10-28T04:00:00Z",
  "timeout_s": 120,
  "last_error": null
}

Payload-এ ছবিটি নিজে নেই — শুধু তার id; বড় ডেটা storage-এ, job-এ শুধু উল্লেখ। আর run_at সময়টি UTC-তে — কেন, তা অংশ ৫-এ দেখবেন।

ছয়টি অবস্থা

Library ভেদে নাম আলাদা — কোথাও “scheduled”, কোথাও “discarded” বা “archived” — কিন্তু প্রায় সব job ব্যবস্থার ভেতরে এই একই state machine.

Worker কীভাবে কাজ তোলে

প্রতিটি worker একটি loop: সারি থেকে একটি job চায়, পায় একটি সময়সীমাসহ (lease বা visibility timeout), কাজ করে, তারপর সফল হলে ack দেয় — job মুছে যায় বা succeeded চিহ্নিত হয়। Deploy-এর সময় worker-কে graceful shutdown করতে হয়: বন্ধের সংকেত (যেমন SIGTERM — OS কোর্সের Signal পর্ব) পেলে নতুন job নেওয়া থামানো, হাতের কাজ শেষ করা বা সারিতে ফেরত দেওয়া।

Broker না রেখে সাধারণ database টেবিলকেও job queue বানানো যায় — ছোট-মাঝারি সিস্টেমে এটি প্রচলিত আর সরল, কারণ job রাখা আর ব্যবসার লেখা একই transaction-এ হয় (outbox-এর দরকারই পড়ে না)। কৌশলটি FOR UPDATE SKIP LOCKED: অনেক worker একসাথে পড়লেও প্রত্যেকে আলাদা সারি পায়, কেউ কারও lock-এর জন্য অপেক্ষা করে না।

-- একজন worker একটি job তোলে; অন্যরা একই সারি এড়িয়ে পরেরটি পায়
UPDATE jobs SET state = 'running', attempt = attempt + 1,
       lease_until = now() + interval '60 seconds', lease_token = lease_token + 1
WHERE id = (
  SELECT id FROM jobs
  WHERE state IN ('queued', 'retrying') AND run_at <= now()
  ORDER BY run_at
  LIMIT 1
  FOR UPDATE SKIP LOCKED
)
RETURNING id, type, payload, attempt, lease_token;

এই query-তে run_at <= now() retry-র অপেক্ষা আর delayed job সামলায়; lease_until worker মরলে job ফেরত আনে; আর lease_token একটি ক্রমিক নম্বর, যা দিয়ে পুরোনো worker-এর দেরিতে আসা লেখা চেনা যায় (অংশ ৪)। খুব বেশি job/s-এ টেবিলের চাপ বাড়ে; তখন আলাদা broker বা log-ভিত্তিক stream (পর্ব ৩৩ — Log-ভিত্তিক streaming).

নিচের ছবিটি এই পর্বের মানচিত্র। ওপরে job আসার দুটি উৎস — API (ব্যবহারকারীর কাজ থেকে) আর scheduler (ঘড়ি থেকে); মাঝে সারি আর worker; নিচে দীর্ঘ, বহু ধাপের কাজের জন্য workflow engine আর তার history store.

৩ · ব্যর্থ হলে — retry, exponential backoff, jitter, max attempts আর DLQ

Background job ব্যর্থ হবেই — ইমেইল সেবা ৫০৩ দেয়, storage-এ timeout, network-এ ক্ষণিকের বিচ্ছিন্নতা। বেশিরভাগই ক্ষণস্থায়ী (transient), কিন্তু ভুল নীতির retry নিজেই বিপদ। চারটি প্রশ্নের উত্তর লাগে:

1. কোন ত্রুটিতে retry? Timeout, ৫০৩, ৪২৯, connection ভাঙা — হ্যাঁ। ভুল ইনপুট (৪০০), অনুমতি নেই (৪০৩), ছবির ফাইল নষ্ট — না। এগুলো non-retryable error.
2. কতক্ষণ পরে? সঙ্গে সঙ্গে নয়। যে সেবা এইমাত্র ডুবল, তাকে সঙ্গে সঙ্গে আবার ডাকলে সে আরও ডোবে — retry storm. তাই exponential backoff: প্রতিটি ব্যর্থতায় অপেক্ষা দ্বিগুণ — ২, ৪, ৮, ১৬ সেকেন্ড — একটি ঊর্ধ্বসীমা (cap) পর্যন্ত।
3. সবাই একসাথে নয়: ১০,০০০ job একই মুহূর্তে ব্যর্থ হলে, শুধু backoff-এ তারা ঠিক ২ সেকেন্ড পরে আবার একসাথে ফেরে। তাই jitter: অপেক্ষার সময় এলোমেলো করা। প্রচলিত “full jitter” — অপেক্ষা = ০ থেকে min(cap, base × 2ⁿ)-এর মধ্যে একটি এলোমেলো মান।
4. কতবার পর্যন্ত? Max attempts (বা মোট সময়ের সীমা)। সীমা পেরোলে job dead — dead letter queue-এ। যে job প্রতিবার ব্যর্থ হয়, সেই poison job সীমা না থাকলে চিরকাল সারিতে ঘুরত।

প্রতিটি চেষ্টা ৩০% ব্যর্থ হলে ৫ বারে শেষ পর্যন্ত ব্যর্থতার সম্ভাবনা ০.৩^৫ ≈ ০.২৪% — কিন্তু শুধু ব্যর্থতাগুলো স্বাধীন হলে। সেবাটি পুরো এক ঘণ্টা বন্ধ থাকলে সব চেষ্টা একসাথে ব্যর্থ; base ২ সেকেন্ডে চার বার অপেক্ষার সর্বোচ্চ যোগফল মাত্র ৪ + ৮ + ১৬ + ৩২ = ৬০ সেকেন্ড। ইমেইলের মতো কাজে তাই অনেক library ঘণ্টা বা দিন জুড়ে retry ছড়ায়; যেমন Ruby-র Sidekiq-এর documentation অনুযায়ী ডিফল্টে প্রায় ২৫ বার, মোটামুটি তিন সপ্তাহ ধরে (লেখার সময়ে; সংস্করণভেদে আলাদা)।

নিচের simulator-এ ১,০০০টি job: বেশিরভাগ “normal” (প্রতি চেষ্টায় ৩০% ব্যর্থ), কিছু “flaky” (৭০% — যেমন একটি অস্থির বাইরের সেবা), কিছু “poison” (সবসময় ব্যর্থ), আর কিছু “invalid” (ভুল ইনপুট, retry চলে না)। Max attempts ৫, base ২ সেকেন্ড, cap ৬০ সেকেন্ড, full jitter; seed স্থির, তাই প্রতিবার একই ফল।

// Seeded job runner: flaky jobs, retries with exponential backoff + full jitter, max attempts, DLQ.
function mulberry32(a) {                     // small seeded PRNG: same run every time
  return () => { a = (a + 0x6D2B79F5) | 0; let t = Math.imul(a ^ (a >>> 15), 1 | a);
    t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t; return ((t ^ (t >>> 14)) >>> 0) / 4294967296; };
}
const rnd = mulberry32(37);
const MAX = 5, BASE = 2, CAP = 60;           // max attempts; backoff base and cap (seconds)
const FAIL = { normal: 0.3, flaky: 0.7, poison: 1 };     // chance that one attempt fails
const wait = n => rnd() * Math.min(CAP, BASE * 2 ** n);  // full jitter after failure n

const jobs = [];
for (let i = 1; i <= 1000; i++) {
  const r = rnd();
  jobs.push({ id: i, kind: r < 0.8 ? "normal" : r < 0.95 ? "flaky" : r < 0.98 ? "poison" : "invalid" });
}
const okAt = [0, 0, 0, 0, 0, 0], dlq = { normal: 0, flaky: 0, poison: 0, invalid: 0 };
let attempts = 0, shown = 0;
for (const job of jobs) {
  let t = 0, log = [];
  for (let n = 1; ; n++) {
    attempts++;
    if (job.kind === "invalid") { log.push("t=" + t.toFixed(1) + " 400 bad input -> DLQ"); dlq.invalid++; break; }
    const failed = rnd() < FAIL[job.kind];
    log.push("t=" + t.toFixed(1) + (failed ? " fail" : " ok"));
    if (!failed) { okAt[n]++; break; }
    if (n === MAX) { log.push("-> DLQ after " + MAX); dlq[job.kind]++; break; }
    t += wait(n);                            // retrying: next run_at = now + jittered backoff
  }
  if (shown < 3 && log.length > 2) { console.log("job " + job.id + " (" + job.kind + "): " + log.join(" | ")); shown++; }
}
const ok = okAt.reduce((a, b) => a + b, 0);
console.log("\nsucceeded " + ok + "/1000 (" + (ok / 10).toFixed(1) + "%), total attempts " + attempts);
console.log("succeeded on attempt 1..5: " + okAt.slice(1).join(", "));
console.log("DLQ after " + MAX + " tries: normal " + dlq.normal + ", flaky " + dlq.flaky + ", poison " +
  dlq.poison + "; non-retryable at once: " + dlq.invalid);

job 8 (poison): t=0.0 fail | t=2.3 fail | t=8.6 fail | t=22.5 fail | t=39.8 fail | -> DLQ after 5
job 15 (flaky): t=0.0 fail | t=1.9 fail | t=6.2 fail | t=17.1 ok
job 16 (flaky): t=0.0 fail | t=1.9 fail | t=6.4 fail | t=19.6 fail | t=36.7 fail | -> DLQ after 5

succeeded 915/1000 (91.5%), total attempts 1716
succeeded on attempt 1..5: 615, 180, 79, 26, 15
DLQ after 5 tries: normal 3, flaky 26, poison 31; non-retryable at once: 25

নিচের stepper-এ একটি job-এর পুরো জীবন, আর শেষে একটি poison job-এর DLQ-যাত্রা। নীতি একই: max attempts ৪, base ২ সেকেন্ড, full jitter.

৪ · অন্তত একবার, তাই idempotent — timeout, heartbeat আর lease

একটি অস্বস্তিকর সত্য দিয়ে শুরু করি: job ব্যবস্থা প্রায় সবসময় at-least-once. Worker ইমেইল পাঠাল, তারপর ack-এর ঠিক আগে crash; সারি শুধু জানে ack আসেনি, তাই আবার দেয় — একই ইমেইল দুবার। উল্টো পথ — আগে ack, পরে কাজ — at-most-once: crash হলে কাজটি হারায়। “ঠিক একবার” পাঠানো সাধারণভাবে সম্ভব নয়; যা সম্ভব তা হলো একাধিকবার চালালেও ফল একবারের মতো। Delivery semantics-এর পুরো তত্ত্ব পর্ব ৩৪ — Delivery semantics-এ।

Idempotent job লেখার কৌশল

• নির্দিষ্ট নামে লেখা: thumbnail 7731_512.jpg নামে — দুবার লিখলেও একটিই ফাইল। “নতুন একটি ফাইল যোগ করো” নয়, “এই নামে ফাইলটি এমন হোক”।
• Idempotency key: বাইরের সেবাকে ডাকার সময় job-এর id থেকে বানানো একটি idempotency key পাঠানো (যেমন email:order-4217:confirm); যে সেবা key মনে রাখে, সে দ্বিতীয় অনুরোধে আগের ফলটাই ফেরত দেয়। অনেক পেমেন্ট API প্রকাশ্যে এই সুবিধা দেয়।
• আগে দেখে তারপর করা: sent_emails টেবিলে key-র unique constraint; সারি ঢোকানো না গেলে “আগেই গেছে” — idempotent consumer-এর ধাঁচ। পাঠানো আর চিহ্ন দুটি আলাদা সিস্টেমে বলে ছোট ফাঁক থাকে; বাইরের সেবার নিজের idempotency সবচেয়ে মজবুত।
• অবস্থা-যাচাই: “অর্ডার বাতিল করো” job প্রথমে দেখে অর্ডারটি এখনো pending কি না; ইতিমধ্যে পরিশোধিত হলে কিছুই না করে সফল।

Timeout, heartbeat আর lease — আটকে থাকা worker

Worker মরে গেলে job কে ফেরত আনবে? Worker নিজে তো জানাতে পারবে না। তাই job তোলার সময়েই একটি lease — “এই job ৬০ সেকেন্ডের জন্য এই worker-এর”। সময়ের মধ্যে ack না এলে job আবার দৃশ্যমান হয়, অন্য worker তোলে। Amazon SQS-এ এর নাম visibility timeout (documentation অনুযায়ী লেখার সময়ে ডিফল্ট ৩০ সেকেন্ড, সর্বোচ্চ ১২ ঘণ্টা); Google Pub/Sub-এ ack deadline.

দীর্ঘ কাজে (৪ মিনিটের রিপোর্ট) lease-ই ৪ মিনিটের বেশি রাখলে মরা worker-এর job ফিরতে ৪ মিনিট। সমাধান: ছোট lease, আর চলাকালে নিয়মিত heartbeat দিয়ে lease বাড়ানো (SQS-এ ChangeMessageVisibility). একটি প্রচলিত নিয়ম: heartbeat-এর বিরতি lease-এর মোটামুটি এক-তৃতীয়াংশ, যাতে একটি-দুটি heartbeat হারালেও lease না ফুরায়। একটি আলাদা timeout-ও লাগে — job-এর সর্বোচ্চ মোট সময় — কারণ heartbeat দিতে থাকা কিন্তু অসীম loop-এ ঘোরা worker-কেও থামাতে হয়।

সবচেয়ে কঠিন ক্ষেত্রটি মরা worker নয়, থেমে থাকা worker: দীর্ঘ garbage collection-এর বিরতি, disk-এ আটকে যাওয়া, বা VM সাময়িক থেমে যাওয়া। তার lease ফুরিয়েছে, job গেছে আরেকজনের কাছে — কিন্তু সে নিজে জানে না, জেগে উঠে কাজ শেষ করে ফল লিখতে চায়। নিচের stepper সেই দৌড়।

৫ · সময় ধরে কাজ — cron আর বড় আকারে তার ফাঁদ

এ পর্যন্ত job এসেছে কোনো ঘটনা থেকে — আপলোড, অর্ডার। আরেক ধরনের কাজ আসে ঘড়ি থেকে: প্রতি রাতে বিক্রির রিপোর্ট, প্রতি ঘণ্টায় মেয়াদোত্তীর্ণ session মোছা, প্রতি মাসের ১ তারিখে subscription-এর বিল। এই দায়িত্ব scheduler-এর, আর তার সবচেয়ে পুরোনো, সবচেয়ে পরিচিত রূপ cron.

# min hour day-of-month month day-of-week   command
0     0    *            *     *             /opt/app/nightly_report     # every day at 00:00
*/15  *    *            *     *             /opt/app/expire_sessions    # every 15 minutes
0     3    1            *     *             /opt/app/monthly_billing    # 03:00 on the 1st

ফাঁদ ১ · সবাই রাত ১২টায় — thundering herd

একটি SaaS অ্যাপে ১০,০০০ tenant-এর প্রত্যেকের রাতের রিপোর্ট 0 0 * * *-এ রাখলে ঠিক ০০:০০:০০-এ ১০,০০০ job একসাথে — একটি thundering herd. নিচের simulator একই ১০,০০০ job তিনভাবে চালায়: সবাই ০০:০০-এ, অথবা tenant id-এর hash দিয়ে এক ঘণ্টা বা ১০ মিনিটে ছড়ানো। Worker-রা মিলে সেকেন্ডে ৪০টি job শেষ করে (ধরা যাক)।

// Cron herd: 10,000 tenants' nightly report. All at 00:00:00, or spread by a hash of the tenant id?
function fnv1a(str) {                        // FNV-1a, 32-bit
  let h = 0x811c9dc5;
  for (let i = 0; i < str.length; i++) h = Math.imul(h ^ str.charCodeAt(i), 0x01000193) >>> 0;
  return h;
}
const TENANTS = 10000, WINDOW = 3600;        // spread window: 00:00 to 01:00 (seconds)
const CAP = 40;                              // workers finish 40 jobs per second (assumed)

function run(name, startOf) {
  const arrivals = new Array(WINDOW).fill(0);
  for (let i = 0; i < TENANTS; i++) arrivals[startOf("tenant-" + i)]++;
  let backlog = 0, maxBacklog = 0, sec = 0, done = 0;
  while (done < TENANTS) {                   // one loop = one second of the night
    backlog += sec < WINDOW ? arrivals[sec] : 0;
    maxBacklog = Math.max(maxBacklog, backlog);
    const now = Math.min(CAP, backlog);
    backlog -= now; done += now; sec++;
  }
  const busiest = Math.max(...arrivals);
  console.log(name.padEnd(22) + String(busiest).padStart(7) + String(maxBacklog).padStart(9) +
    String(Math.ceil(maxBacklog / CAP)).padStart(10) + " s" + String(sec).padStart(8) + " s");
}
console.log("schedule              max/s  backlog  worst wait  all done");
run("all at 00:00:00", () => 0);
run("hash % 3600 s", id => fnv1a("nightly-report:" + id) % WINDOW);
run("hash % 600 s", id => fnv1a("nightly-report:" + id) % 600);

schedule              max/s  backlog  worst wait  all done
all at 00:00:00         10000    10000       250 s     250 s
hash % 3600 s               9        9         1 s    3600 s
hash % 600 s               28       28         1 s     600 s

ফাঁদ ২ · আগের run শেষ হয়নি — overlap

প্রতি ১৫ মিনিটের job একদিন database-এর ধীরতায় ২০ মিনিট নিল — পরের run শুরু, দুটি মিলে আরও ধীর, তৃতীয়টি এল; জমতে জমতে সিস্টেম ডোবে। প্রতিটি scheduled job-এর একটি overlap policy লাগে: দুটিই চলুক (Allow), নতুনটি বাদ (Forbid), নাকি পুরোনোটি থামিয়ে নতুনটি (Replace). Kubernetes CronJob-এ ঠিক এই তিনটি মানের concurrencyPolicy আছে (লেখার সময়ে ডিফল্ট Allow — অর্থাৎ নিরাপদ বিকল্পটি আপনাকেই বেছে নিতে হয়)।

ফাঁদ ৩ · Time zone আর DST

“রাত ২:৩০-এ” — কোন দেশের রাত? আরও সূক্ষ্ম DST: যেসব দেশে গ্রীষ্মে ঘড়ি এক ঘণ্টা এগোয়, সেই রাতে ০২:০০ থেকে সরাসরি ০৩:০০ — “০২:৩০” সময়টাই নেই, তাই local সময়ের job বাদ পড়তে পারে; আর ঘড়ি পেছানোর রাতে ০১:০০–০২:০০ দুবার আসে, job দুবার চলতে পারে। আচরণ implementation-ভেদে আলাদা — যেমন Debian-এর cron-এর manual অনুযায়ী তিন ঘণ্টার কম ঘড়ি-বদলে বাদ পড়া job বদলের পরপরই চালানো হয়, আর পুনরাবৃত্ত সময়ের job আবার চালানো হয় না।

বাংলাদেশে এখন DST নেই (মান সময় UTC+6); ২০০৯ সালে কয়েক মাসের জন্য চালু হয়ে বছর শেষে আবার তুলে দেওয়া হয়েছিল। কিন্তু ব্যবহারকারী বা সার্ভার অন্য দেশে থাকলে, বা নিয়ম বদলালে, ফাঁদটি আপনারও। নিরাপদ অভ্যাস: scheduler আর সব timestamp UTC-তে; ব্যবহারকারীর “আমার সময়ে সকাল ৯টা” দরকার হলে তাঁর time zone-এর নামসহ (যেমন Asia/Dhaka) রাখা, আর DST-এর রাতে কী হবে তা স্পষ্ট লেখা।

ফাঁদ ৪ · Scheduler বন্ধ ছিল — বাদ পড়া run

রাত ১২টায় deploy চলছিল, বা scheduler-এর মেশিন বন্ধ — রাতের রিপোর্ট চলেনি। এখন কী? একটি misfire policy লাগে: বাদ পড়া সব run পরপর চালানো (বিলিংয়ে প্রতিটি মাস জরুরি), শুধু সর্বশেষটি (session মোছায় একবারই যথেষ্ট), নাকি বাদ। Kubernetes CronJob-এ startingDeadlineSeconds ঠিক করে কত দেরি পর্যন্ত বাদ পড়া run তবুও শুরু হবে। আর নকশার একটি ভালো অভ্যাস: job-কে “গত ২৪ ঘণ্টার কাজ করো” না বলে “২০২৬-১০-২৭ তারিখের কাজ করো” বলা — নির্দিষ্ট সময়-জানালার job পরে চালালেও একই ফল দেয়, আর দুবার চালালেও (অংশ ৬)।

৬ · Distributed cron, delayed job আর timer

একটি মেশিনে cron চালালে সেই মেশিনই single point of failure: সে মরলে বিলিং চলে না। স্বাভাবিক প্রতিক্রিয়া — তিনটি সার্ভারেই একই crontab. ফল: প্রতিটি job তিনবার, প্রতিটি গ্রাহকের বিল তিনবার। Scheduler-এর আসল চাহিদা দুটি, আর দুটি একে অপরের বিপরীতে টানে: একটি মেশিন মরলেও সময়সূচি চলবে, অথচ প্রতিটি নির্ধারিত run একবারই শুরু হবে। এই দুটি মেলানোই distributed cron.

একজন নেতা, মেয়াদসহ

প্রচলিত পথ: কয়েকটি scheduler instance চলে, কিন্তু শুধু একজন — leader — job ছাড়ে। কে leader, তা ঠিক হয় leader election দিয়ে, প্রায়ই etcd, ZooKeeper বা একটি database-এর একটি lease-এর মাধ্যমে: যে lease ধরে আছে সে leader, নিয়মিত নবায়ন করে; মরলে lease ফুরায়, অন্য instance leader হয়। থেমে থাকা পুরোনো leader-এর ঝুঁকি আর fencing token-এর পুরো আলোচনা পর্ব ৪৩ — Coordination — distributed lock, lease আর fencing token-এ, আর consensus-এর ভিত্তি পর্ব ৪২-এ।

দ্বিতীয় স্তরের পাহারা — run-এর অনন্য key

Leader election কখনো নিখুঁত নয়, তাই প্রতিটি run-কে একটি অনন্য পরিচয় দিন: job_name + নির্ধারিত সময়, যেমন monthly_billing@2026-11-01T03:00Z. Run শুরুর আগে এই key একটি unique constraint-ওয়ালা টেবিলে লেখা হয়; দুজন leader একসাথে চেষ্টা করলেও একজনই সফল। আর job নিজে যদি নির্দিষ্ট সময়-জানালার কাজ করে (অংশ ৫-এর শেষ অনুচ্ছেদ) ও idempotent হয়, তাহলে বাদ পড়া run পরে চালানো বা ভুলে দুবার চালানো — দুটোই নিরাপদ। মূল চিন্তা: scheduler “একবার” নিশ্চিত করার চেষ্টা করে, কিন্তু নিরাপত্তা আসে job-এর idempotency থেকে।

Delayed job আর timer

সব সময়ভিত্তিক কাজ বারবার ঘটে না। “অর্ডারের ৩০ মিনিট পরেও পরিশোধ না হলে বাতিল করুন”, “ট্রায়াল শেষের ৩ দিন আগে মনে করিয়ে দিন” — এগুলো একবারের delayed job. সরল বাস্তবায়ন অংশ ২-এর query-তেই আছে: job রাখার সময় run_at = এখন + ৩০ মিনিট, আর worker শুধু run_at <= now() তোলে (run_at-এ index থাকলে এটি সস্তা)।

“Scheduling”-এর আরেক অর্থ — কোন process এখন CPU পাবে — OS কোর্সের CPU scheduling-এর ভিত্তি পর্বে।

৭ · Workflow engine — durable execution, event history আর replay

একটি কাল্পনিক ঋণ-আবেদন: আবেদন জমা → নথি যাচাই (বাইরের সেবা) → ঝুঁকির হিসাব → ৫০,০০০ টাকার বেশি হলে একজন ম্যানেজারের অনুমোদন (হয়তো দুই দিন) → চুক্তির PDF → টাকা পাঠানো → SMS. সাতটি ধাপ, কয়েকটি সেবা, কয়েক দিন। কোনো ধাপ ব্যর্থ হলে retry; টাকা পাঠানো ব্যর্থ হলে আগের ধাপগুলোর ক্ষতিপূরণ (saga, পর্ব ২৫)। Job queue দিয়ে বানালে (প্রতিটি ধাপ একটি job, যা পরেরটি রাখে) সব যুক্তি — কোন ধাপে, কতবার চেষ্টা, ম্যানেজার উত্তর না দিলে কী — দশটি handler আর একটি অবস্থার টেবিলে ছড়িয়ে পড়ে। Event দিয়ে সেবাগুলোর নিজে নিজে সমন্বয় (choreography) বনাম কেন্দ্রীয় পরিচালনা (orchestration)-র তুলনা পর্ব ৩২ — Pub/sub আর event-driven architecture-এ।

Workflow engine উল্টো প্রস্তাব দেয়: পুরো প্রক্রিয়াটি একটি সাধারণ function হিসেবে লিখুন — ধাপের পর ধাপ, if-else, loop, এমনকি “দুই দিন অপেক্ষা” — আর engine নিশ্চিত করবে যে function-টি মেশিন মরলেও শেষ পর্যন্ত চলবে। এর নাম durable execution. কৌশলটি তিনটি অংশে:

• Activity: বাইরের জগতে আসল কাজ করে এমন প্রতিটি ধাপ — API ডাকা, টাকা পাঠানো, ইমেইল — একটি activity. Engine এর timeout, heartbeat আর retry (backoff সহ) সামলায়, অংশ ৩ আর ৪-এর মতোই। Activity অন্তত একবার চলে, তাই idempotent হতে হয়।
• Event history: প্রতিটি activity ডাকা, তার ফল, প্রতিটি timer, প্রতিটি বাইরের বার্তা — সব একটি শুধু-যোগ-করা event history-তে টেকসইভাবে লেখা। State আলাদা করে রাখা হয় না; history-ই সত্য — event sourcing-এর চিন্তা (পর্ব ৩৫ — Event sourcing আর CQRS).
• Replay: worker মরলে নতুন worker workflow function শুরু থেকে আবার চালায়। প্রতিটি activity-ডাকার মুহূর্তে engine history দেখে: ফল আগেই আছে? তাহলে কাজটি আবার না করে লেখা ফলটি ফেরত। এই replay-এ function-এর ভেতরের সব variable আবার ঠিক আগের অবস্থায় পৌঁছায়, তারপর বাকি অংশ নতুন করে চলে।

নিচের প্রোগ্রামে এই যন্ত্রের একটি ছোট খেলনা-রূপ: চার ধাপের একটি অর্ডার workflow, প্রথম worker তৃতীয় ধাপের মাঝে crash করে, দ্বিতীয় worker history থেকে replay করে শেষ করে। আসল engine-এর history অনেক বিস্তারিত।

// Tiny durable execution: the event history survives a worker crash; replay skips finished steps.
const history = [{ type: "WorkflowStarted" }];   // the engine's durable, append-only store
const ran = {};                              // how many times each activity REALLY executed
const activities = {                         // side effects live here (other services, email ...)
  reserveStock: o => "hold-" + o, chargePayment: o => "txn-" + o * 3,
  makeInvoice: t => "INV-" + t.slice(4), sendEmail: inv => "mail-for-" + inv,
};
function workflow(call, order) {             // deterministic: same history, same calls, same order
  call("reserveStock", order);
  const txn = call("chargePayment", order);
  const inv = call("makeInvoice", txn);
  call("sendEmail", inv);
}
function worker(name, crashInside) {
  let seq = 0;
  const call = (act, arg) => {
    const n = seq++, done = history.find(e => e.n === n && e.type === "Completed");
    if (done) { console.log(name + " replay  #" + n + " " + act + " = " + done.result); return done.result; }
    if (!history.some(e => e.n === n)) history.push({ n, type: "Scheduled", act });
    ran[act] = (ran[act] || 0) + 1;
    const result = activities[act](arg);     // the real work happens ...
    if (n === crashInside) throw new Error(name + " crashed inside #" + n + " before recording it");
    history.push({ n, type: "Completed", act, result });   // ... and is recorded durably
    console.log(name + " execute #" + n + " " + act + " = " + result);
    return result;
  };
  try { workflow(call, 4217); history.push({ type: "WorkflowCompleted" }); }
  catch (e) { console.log("!! " + e.message); }
}
worker("A", 2);                              // first worker dies while making the invoice
worker("B", -1);                             // engine hands the workflow to another worker
console.log("\nhistory:");
for (const e of history)
  console.log(("  " + e.type.padEnd(18) + (e.n ?? "") + " " + (e.act || "") + " " + (e.result || "")).trimEnd());
console.log("real executions: " + JSON.stringify(ran));

A execute #0 reserveStock = hold-4217
A execute #1 chargePayment = txn-12651
!! A crashed inside #2 before recording it
B replay  #0 reserveStock = hold-4217
B replay  #1 chargePayment = txn-12651
B execute #2 makeInvoice = INV-12651
B execute #3 sendEmail = mail-for-INV-12651

history:
  WorkflowStarted
  Scheduled         0 reserveStock
  Completed         0 reserveStock hold-4217
  Scheduled         1 chargePayment
  Completed         1 chargePayment txn-12651
  Scheduled         2 makeInvoice
  Completed         2 makeInvoice INV-12651
  Scheduled         3 sendEmail
  Completed         3 sendEmail mail-for-INV-12651
  WorkflowCompleted
real executions: {"reserveStock":1,"chargePayment":1,"makeInvoice":2,"sendEmail":1}

কেন workflow কোড deterministic হতে হবে

ধরুন workflow-এর ভেতরে লেখা: if (Math.random() < 0.5) call("A") else call("B"). প্রথম run-এ A ডাকা হয়েছিল, history-তে A-র ফল। Replay-তে random এবার B বলল — কোড এখন B-এর ফল চায়, history-তে আছে A-র। Engine এই অমিল ধরে ত্রুটি দেয় (non-determinism). একই বিপদ সরাসরি ঘড়ি পড়ায় (new Date()-এর মান replay-তে আলাদা), network বা ফাইল পড়ায়, আর map-এর ক্রমহীন iteration-এ। তাই নিয়ম: workflow কোড deterministic; বাইরের জগতের সবকিছু activity-তে, আর সময়, random বা ঘুম engine-এর দেওয়া API দিয়ে, যা প্রথমবারের মান history-তে লিখে replay-তে সেটাই ফেরত দেয়।

কোড বদলালেও একই সমস্যা: তিন দিন ধরে চলা একটি workflow-এর মাঝে আপনি একটি নতুন ধাপ যোগ করে deploy করলেন; পুরোনো history-র replay নতুন কোডে মেলে না। তাই workflow versioning — বদলের জায়গায় একটি version চিহ্ন রেখে পুরোনো চলমান workflow-গুলোকে পুরোনো পথে চলতে দেওয়া।

Timer, signal আর মানুষের অপেক্ষা

দুই দিন অপেক্ষা workflow-এ একটি লাইন: await sleep("2 days"). এটি thread আটকে রাখে না — engine history-তে একটি durable timer লেখে, worker অন্য কাজে যায়; সময় হলে engine workflow-কে আবার জাগায়, replay করে, আর কোড ঘুমের পরের লাইন থেকে চলে। ম্যানেজারের অনুমোদন আসে একটি signal হিসেবে — বাইরে থেকে পাঠানো একটি টেকসই বার্তা, যা history-তে লেখা হয়। “অনুমোদনের signal অথবা ৪৮ ঘণ্টা, যেটি আগে” — এই human-in-the-loop প্রক্রিয়া কয়েক লাইনের কোড।

৮ · Data pipeline-এর DAG, আর কোনটি কখন — queue, cron না workflow engine

Workflow engine-এর একটি আলাদা পরিবার data pipeline-এর জন্য। প্রতি রাতে: বিক্রির টেবিল থেকে ডেটা তোলা → পরিষ্কার করা → গ্রাহক আর পণ্যের টেবিলের সাথে জোড়া → দৈনিক সারাংশ → dashboard হালনাগাদ, আর পাশাপাশি ঝুঁকির একটি মডেলের জন্য আলাদা হিসাব। এখানে কাজগুলোর মধ্যে নির্ভরতা একটি DAG — দিকযুক্ত, চক্রহীন graph: তীর মানে “এটি শেষ হলে তবে ওটি”, আর চক্র নেই বলে topological ক্রমে চালানো যায় (DSA কোর্সের Topological sort).

Apache Airflow এই ধাঁচের পরিচিত open-source উদাহরণ — ২০১৪ সালে Airbnb-তে শুরু, পরে Apache Software Foundation-এর প্রকল্প। এখানে সবচেয়ে গুরুত্বপূর্ণ নকশা-নীতি অংশ ৫-এর শেষ অভ্যাসটিই: প্রতিটি task একটি নির্দিষ্ট তারিখের (logical date) কাজ করে, আর সেই তারিখের output পুরোটা নতুন করে লেখে (“২০২৬-১০-২৭-এর partition মুছে আবার লেখা” ধাঁচ)। তাহলে task দুবার চললেও একই ফল, আর পুরোনো তারিখগুলোর জন্য আবার চালানো — backfill — নিরাপদ। Batch আর stream processing-এর গভীর আলোচনা পর্ব ৩৬ — Batch আর stream processing-এ।

# Illustrative only: a nightly sales pipeline as a DAG of idempotent, date-partitioned tasks
with DAG("nightly_sales", schedule="30 0 * * *", catchup=True, max_active_runs=1) as dag:
    extract   = extract_sales(ds="{{ ds }}")       # rewrite partition for the logical date
    clean     = clean_sales(ds="{{ ds }}")
    customers = load_customers(ds="{{ ds }}")
    summary   = build_daily_summary(ds="{{ ds }}")
    dashboard = refresh_dashboard(ds="{{ ds }}")
    extract >> clean >> summary >> dashboard
    customers >> summary                           # summary waits for both branches

schedule একটি cron expression; catchup হলো misfire policy; max_active_runs=1 হলো overlap policy; আর ds প্রতিটি task-কে তার তারিখ দেয় — idempotency-র ভিত্তি।

DAG engine বনাম durable execution engine

দুটোকেই “workflow” বলা হয়, তাই গুলিয়ে যায়। DAG engine (Airflow-ধাঁচ) সাধারণত আগে থেকে জানা, স্থির আকারের graph চালায় — প্রতিটি task মিনিট বা ঘণ্টার, প্রায়ই সময়সূচি ধরে, ডেটা-কেন্দ্রিক। Durable execution engine (Temporal- বা Cadence-ধাঁচ) চালায় কোডে লেখা যেকোনো যুক্তি — loop, শর্ত, signal, মাসব্যাপী timer — প্রতিটি ব্যবসায়িক ঘটনার (অর্ডার, আবেদন) জন্য আলাদা একটি workflow, একসাথে হয়তো লাখো। AWS Step Functions-এর মতো managed সেবা এর মাঝামাঝি: workflow একটি JSON-ভিত্তিক state machine হিসেবে লেখা হয় (সীমা ও দাম লেখার সময়ের documentation দেখে নিন)।

বেছে নেওয়া — সবচেয়ে সরলটি দিয়ে শুরু

একটি ব্যবহারিক নিয়ম: স্বাধীন একক কাজে job queue; ঘড়ি-ধরা কাজে scheduler; আর যখন দেখবেন job-গুলো একে অপরকে ডাকছে, অবস্থার টেবিলে “step” কলাম জন্মেছে, আর “তিন দিন পরে যদি…” ধরনের যুক্তি পাঁচ জায়গায় ছড়িয়ে আছে — তখন workflow engine-এর কথা ভাবুন।

৯ · AI-এর যুগে — job আর workflow-এর নকশা যাচাই

“ইমেইল পাঠানো ধীর, কী করব?” — AI প্রায় সবসময় বলবে, “Kafka বা RabbitMQ বসান, worker বানান, retry দিন।” দিক ঠিক, কিন্তু আসল কঠিন প্রশ্নগুলোতে উত্তর অস্পষ্ট। এই বিষয়ে AI-এর সাধারণ ভুলগুলো:

AI-এর job নকশা হাতে পেলে — যাচাইয়ের চেকলিস্ট

• Worker-এর হিসাব L = λW দিয়ে নিজে আবার করেছি; retry-র বাড়তি চেষ্টা আর utilisation-এর headroom ধরেছি
• প্রতিটি ত্রুটিকে retryable বা non-retryable ভাগ করেছি; backoff-এ jitter, max attempts আর মোট retry-সময় বিভ্রাটের দৈর্ঘ্যের সাথে মিলিয়েছি
• প্রতিটি job-এর জন্য “দুবার চললে কী হবে” লিখেছি, আর idempotency-র কৌশল বেছেছি
• Lease, heartbeat আর job timeout-এর মান কাজের দৈর্ঘ্যের সাথে মিলিয়েছি; আটকে থাকা worker-এর লেখা আটকানোর ব্যবস্থা আছে
• Cron-এর চারটি ফাঁদ (herd, overlap, time zone/DST, বাদ পড়া run) আর একাধিক scheduler-এ দ্বিগুণ run-এর পরীক্ষা করেছি
• একটি ছোট simulation বা load test-এ ঢল আর ব্যর্থতা নিজে তৈরি করে দেখেছি; product-এর সীমা documentation-এ মিলিয়েছি